Named branches on top of BucketHead's snapshot identity.
A branch is a named R2 snapshot key. Each branch can independently diverge;
BucketHead tracks which branch the in-memory DB is currently paired with
and directs flush/restore against that branch's key.
Branches are tracked by live-listing R2 (no registry) — see
BranchManager.list. Branch names must match [a-zA-Z0-9_-]+;
main is reserved.
See plan/build-plan.md (branching section) and docs/diagrams.md
for the lifecycle.
Bases: Exception
Base class for branching-related failures.
Bases: BranchError, ValueError
Branch name is malformed or uses a reserved name.
Bases: BranchError, KeyError
Operation references a branch that has no snapshot in the bucket.
Bases: BranchError, ValueError
create called with a name that already has a snapshot.
Public bh.branches facade.
Holds a reference back to the owning BucketSQLite so switch/overwrite
can coordinate with the keep-alive connection and the flush machinery.
Source code in src/buckethead/storage/branches.py
| def __init__(self, bh: BucketSQLite, client: BucketClient, base_key: str) -> None:
self._bh = bh
self._client = client
self._base_key = base_key
self._branch_prefix = f"{base_key}.branch."
|
The branch currently paired with the in-memory DB.
Source code in src/buckethead/storage/branches.py
| def current(self) -> str:
"""The branch currently paired with the in-memory DB."""
return self._bh._current_branch
|
Return all known branches — always includes main.
Implementation: live-list R2 under the branch prefix, strip the
prefix, skip .prev copies. One ListObjectsV2 op per call.
Source code in src/buckethead/storage/branches.py
| def list(self) -> list[str]:
"""Return all known branches — always includes ``main``.
Implementation: live-list R2 under the branch prefix, strip the
prefix, skip ``.prev`` copies. One ListObjectsV2 op per call.
"""
names: set[str] = {"main"}
for key in self._client.list_keys(self._branch_prefix):
suffix = key[len(self._branch_prefix) :]
if suffix.endswith(".prev"):
continue
names.add(suffix)
return sorted(names)
|
create(
name: str, *, from_branch: str | None = None
) -> None
Create a new branch as a copy of the current branch (or a named source).
Auto-flushes the source first so in-memory writes are captured.
Raises BranchExistsError if name already has a snapshot.
Source code in src/buckethead/storage/branches.py
| def create(self, name: str, *, from_branch: str | None = None) -> None:
"""Create a new branch as a copy of the current branch (or a named source).
Auto-flushes the source first so in-memory writes are captured.
Raises BranchExistsError if ``name`` already has a snapshot.
"""
_check_name_for_create(name)
source = from_branch if from_branch is not None else self._bh._current_branch
target_key = self._branch_key(name)
if self._client.exists(target_key):
raise BranchExistsError(f"branch {name!r} already exists")
if source == self._bh._current_branch:
# In-memory writes may not have reached R2 yet — capture them.
self._bh.force_flush()
source_key = self._branch_key(source)
if not self._client.exists(source_key):
raise BranchNotFoundError(f"source branch {source!r} has no snapshot")
self._client.copy(source_key, target_key)
|
switch(name: str) -> None
Flush the current branch and rehydrate memory from name.
If name is the current branch, no-op. If the target branch has
no snapshot (and isn't main), raises BranchNotFoundError. When
switching to an empty main that has no snapshot, the in-memory
DB is cleared and BucketHead's own tables (filestore, bh_kv) are
re-initialized.
Source code in src/buckethead/storage/branches.py
| def switch(self, name: str) -> None:
"""Flush the current branch and rehydrate memory from ``name``.
If ``name`` is the current branch, no-op. If the target branch has
no snapshot (and isn't ``main``), raises BranchNotFoundError. When
switching to an empty ``main`` that has no snapshot, the in-memory
DB is cleared and BucketHead's own tables (filestore, bh_kv) are
re-initialized.
"""
try:
_validate_branch_name(name)
except ValueError as e:
raise InvalidBranchNameError(str(e)) from e
if name == self._bh._current_branch:
return
target_key = self._branch_key(name)
target_exists = self._client.exists(target_key)
if name != "main" and not target_exists:
raise BranchNotFoundError(f"no such branch: {name}")
# 1. Flush outgoing branch so its snapshot reflects in-memory writes.
self._bh.force_flush()
# 2. Switch the identity. From here on, flushes go to the new key.
self._bh._current_branch = name
self._bh._last_snapshot_hash = None
# 3. Rehydrate memory. sqlite3.backup() is a full page-level replace,
# so restoring drops any tables that aren't in the target. If the
# target has no snapshot (only possible for main), clear manually
# and re-init BucketHead's own tables.
if target_exists:
restore(self._bh.connection, self._client, target_key)
else:
self._clear_in_memory_db()
# Schema bootstrap: idempotent. Ensures filestore/bh_kv exist whether
# we restored or cleared (restore may have dropped them if they
# weren't in the target snapshot, which happens if the branch
# was created before the Interface tables existed).
self._bh._filestore._init_schema() # pyright: ignore[reportOptionalMemberAccess]
self._bh._kv._init_schema() # pyright: ignore[reportOptionalMemberAccess]
|
overwrite(target: str) -> None
Change the current branch identity to target and force-flush.
The in-memory DB is NOT reloaded — its bytes become target's new
content. The previously-active branch is left untouched; delete it
separately if no longer needed.
Use case: you were experimenting on branch X, it succeeded, now you
want X's state to be main's state going forward.
Source code in src/buckethead/storage/branches.py
| def overwrite(self, target: str) -> None:
"""Change the current branch identity to ``target`` and force-flush.
The in-memory DB is NOT reloaded — its bytes become ``target``'s new
content. The previously-active branch is left untouched; delete it
separately if no longer needed.
Use case: you were experimenting on branch X, it succeeded, now you
want X's state to be main's state going forward.
"""
try:
_validate_branch_name(name := target)
except ValueError as e:
raise InvalidBranchNameError(str(e)) from e
# Identity swap is a local change. Next flush goes to the new key.
self._bh._current_branch = name
# The in-memory content differs from whatever the target had, so the
# dirty-bit tracker must reset — otherwise an identical-hash target
# would wrongly skip the upload.
self._bh._last_snapshot_hash = None
self._bh.force_flush()
|
delete(name: str) -> None
Remove a branch's snapshot (and its .prev if present).
Rejects main and the currently-active branch. Idempotent for
branches that don't exist.
Source code in src/buckethead/storage/branches.py
| def delete(self, name: str) -> None:
"""Remove a branch's snapshot (and its ``.prev`` if present).
Rejects ``main`` and the currently-active branch. Idempotent for
branches that don't exist.
"""
try:
_validate_branch_name(name)
except ValueError as e:
raise InvalidBranchNameError(str(e)) from e
if name == "main":
raise InvalidBranchNameError("'main' cannot be deleted")
if name == self._bh._current_branch:
raise ValueError(
f"cannot delete the currently-active branch {name!r}; "
"switch to another branch first"
)
key = self._branch_key(name)
prev = key + ".prev"
# BucketClient.delete is idempotent — no-op on missing keys.
self._client.delete(key)
self._client.delete(prev)
|