trine-kv 0.5.13

//! Copy-on-write branches and time travel, built over the existing MVCC read
//! API and named buckets so the LSM read/write hot path is untouched — a
//! database that never branches pays nothing (see `docs/branching.md`).
//!
//! A [`Branch`] forks from a parent [`ReadVersion`] (a pinned [`crate::Snapshot`]
//! that also keeps the fork's history retained while the branch lives). It shares
//! all parent history at or below the fork — O(1) to create, no data copied — and
//! keeps its own divergent writes separate; reads consult the branch's writes
//! first and fall through to the pinned parent snapshot. The parent is never
//! affected.
//!
//! Two flavors share one API:
//!
//! * **Ephemeral clone** ([`Db::branch_from_latest`], [`Db::branch_at`]): writes
//!   live in an in-memory overlay and vanish with the handle — a scratch
//!   "what-if" clone or a point-in-time (`AS OF`) read view.
//! * **Durable named branch** ([`Db::create_branch`] + [`Db::open_branch`]): writes
//!   persist in the branch's own buckets, so they survive reopen and are
//!   compacted and recovered like any data — a git-style named branch. Because a
//!   branch's writes live in their **own** buckets (their own layer-set), they
//!   never enter the parent's trees, so branch activity cannot perturb the
//!   parent's compaction or read amplification.
//!
//! A durable branch pins its fork with a durable checkpoint, so the parent keeps
//! the branch's fork history — and the branch stays openable — across restarts
//! and aggressive retention, with no manual retention configuration, until
//! [`Db::delete_branch`] releases the pin.
//!
//! Branches nest: [`Db::create_branch_from`] forks a branch off another branch,
//! and a read walks the whole ancestor chain (branch → parent branch → … → root),
//! each ancestor seen frozen at the version its child forked it. This is the
//! git-style DAG. [`Db::delete_branch`] releases a branch's fork pin, drops its
//! divergent data buckets via [`Db::drop_bucket`] (reclaiming the space; on a
//! backend without bucket-drop it clears them instead), and forgets it; it
//! refuses while the branch still has children (they read through it).
//!
//! [`Branch::range`] is a lazy [`BranchRange`] iterator: the branch level, each
//! ancestor, and the root are streamed from their own sorted scans and k-way
//! merged on the fly (no full copy).

use std::collections::{BTreeSet, HashMap};
use std::ops::Bound;

use crate::bucket::BucketName;
use crate::db::Db;
use crate::error::{Error, Result};
use crate::snapshot::Snapshot;
use crate::types::{KeyRange, KeyValue, ReadVersion, Value};

/// Prefix reserving the buckets branching keeps its own state in. Branch names
/// must not contain the `\u{1}` separator (they are simple identifiers).
const RESERVED: &str = "\u{1}trine-branch\u{1}";
const SEP: char = '\u{1}';

/// The bucket holding the branch registry: branch name → [`RegistryEntry`].
fn registry_bucket() -> String {
    format!("{RESERVED}registry")
}

/// The bucket holding a durable branch's divergent writes for one user bucket.
fn data_bucket(branch: &str, user_bucket: &str) -> String {
    format!("{RESERVED}{branch}{SEP}{user_bucket}")
}

/// A durable branch's lineage: the global version it forked at and its parent
/// branch (`None` = forked from the root lineage). Returned by
/// [`Db::branch_info`] for a higher layer that manages its **own** divergent
/// storage (e.g. a SQL/document engine whose writes must be one atomic
/// multi-bucket batch) and only needs the fork point — to read the parent
/// through [`Db::snapshot_at`] — and the parent link — to walk a nested branch's
/// ancestry — while still relying on the durable fork pin, registry, and nesting
/// this crate maintains.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BranchInfo {
    fork: ReadVersion,
    parent: Option<String>,
}

impl BranchInfo {
    /// The global version this branch forked at. Read the parent's state as of
    /// this version (via [`Db::snapshot_at`]) to resolve a branch read that the
    /// branch's own storage does not hold (the fall-through).
    #[must_use]
    pub const fn fork(&self) -> ReadVersion {
        self.fork
    }

    /// The parent branch's name, or `None` when this branch forked the root
    /// lineage. Walk it to assemble a nested branch's ancestor chain.
    #[must_use]
    pub fn parent(&self) -> Option<&str> {
        self.parent.as_deref()
    }
}

/// A durable branch's persisted metadata: where it forked, the parent branch it
/// forked from (`None` = the root lineage), and which user buckets it has written
/// (so a read need not touch — or create — a data bucket the branch never wrote).
struct RegistryEntry {
    /// The global version this branch forked its parent at.
    fork: ReadVersion,
    /// The parent branch name, or `None` when forked from the root lineage.
    parent: Option<String>,
    written_buckets: BTreeSet<String>,
}

fn put_str(out: &mut Vec<u8>, value: &str) {
    let len = u32::try_from(value.len()).unwrap_or(u32::MAX);
    out.extend_from_slice(&len.to_le_bytes());
    out.extend_from_slice(value.as_bytes());
}

impl RegistryEntry {
    fn encode(&self) -> Vec<u8> {
        let mut out = Vec::new();
        out.extend_from_slice(&self.fork.as_u64().to_le_bytes());
        let count = u32::try_from(self.written_buckets.len()).unwrap_or(u32::MAX);
        out.extend_from_slice(&count.to_le_bytes());
        for bucket in &self.written_buckets {
            put_str(&mut out, bucket);
        }
        // Parent is a trailing optional field (a flag byte then the name) so an
        // entry written before nesting existed still decodes (parent = None).
        match &self.parent {
            Some(parent) => {
                out.push(1);
                put_str(&mut out, parent);
            }
            None => out.push(0),
        }
        out
    }

    fn decode(bytes: &[u8]) -> Result<Self> {
        let corrupt = || Error::Corruption {
            message: "malformed branch registry entry".to_owned(),
        };
        let mut pos = 0usize;
        let take_u32 = |pos: &mut usize| -> Result<u32> {
            let raw: [u8; 4] = bytes
                .get(*pos..*pos + 4)
                .ok_or_else(corrupt)?
                .try_into()
                .expect("4");
            *pos += 4;
            Ok(u32::from_le_bytes(raw))
        };
        let fork_bytes: [u8; 8] = bytes.get(0..8).ok_or_else(corrupt)?.try_into().expect("8");
        let fork = ReadVersion::from_u64(u64::from_le_bytes(fork_bytes));
        pos += 8;
        let count = take_u32(&mut pos)?;
        let mut written_buckets = BTreeSet::new();
        for _ in 0..count {
            let len = take_u32(&mut pos)? as usize;
            let name = bytes.get(pos..pos + len).ok_or_else(corrupt)?;
            pos += len;
            written_buckets.insert(String::from_utf8(name.to_vec()).map_err(|_| corrupt())?);
        }
        // Trailing optional parent (absent in pre-nesting entries).
        let parent = match bytes.get(pos) {
            None | Some(&0) => None,
            Some(&1) => {
                pos += 1;
                let len = take_u32(&mut pos)? as usize;
                let name = bytes.get(pos..pos + len).ok_or_else(corrupt)?;
                Some(String::from_utf8(name.to_vec()).map_err(|_| corrupt())?)
            }
            Some(_) => return Err(corrupt()),
        };
        Ok(Self {
            fork,
            parent,
            written_buckets,
        })
    }
}

/// Value tag in a durable branch's data bucket: a present value or a tombstone
/// (the branch deleted a key the parent still has). Distinguishes "the branch
/// wrote nothing here, fall through to the parent" (key absent) from "the branch
/// deleted it" (tombstone).
const TAG_PRESENT: u8 = 0;
const TAG_TOMBSTONE: u8 = 1;

fn encode_present(value: &[u8]) -> Vec<u8> {
    let mut out = Vec::with_capacity(value.len() + 1);
    out.push(TAG_PRESENT);
    out.extend_from_slice(value);
    out
}

/// One ephemeral branch-local write held in the in-memory overlay.
enum OverlayWrite {
    Put(Value),
    Delete,
}

/// One level of a durable branch's read chain. The leaf (the opened branch
/// itself) reads its own latest writes (`at = None`); each ancestor is read
/// frozen at the version the child forked it (`at = Some`). `written` is that
/// branch's set of diverged user buckets, so untouched buckets are skipped
/// without opening (or creating) a data bucket.
struct DurableLayer {
    name: String,
    written: BTreeSet<String>,
    at: Option<Snapshot>,
}

/// A durable branch's read chain (leaf first, then each ancestor branch), plus
/// the leaf's own registry fields — needed to rewrite its entry when it first
/// writes a user bucket. The root fall-through below the chain is the branch's
/// pinned [`Branch::fork`] snapshot.
struct DurableState {
    chain: Vec<DurableLayer>,
    leaf_fork: ReadVersion,
    leaf_parent: Option<String>,
}

/// How a branch stores its divergent writes.
enum Backing {
    /// In-memory, lost with the handle (ephemeral clone / `AS OF` view).
    Ephemeral(HashMap<(BucketName, Vec<u8>), OverlayWrite>),
    /// Persisted in the branch's own buckets (durable named branch), as a read
    /// chain from the branch up through its ancestor branches.
    Durable(DurableState),
}

/// A copy-on-write branch forked from a parent database at a fixed
/// [`ReadVersion`]. Reads see the parent's state as of the fork with the
/// branch's own writes layered on top; the parent is unaffected.
pub struct Branch<'db> {
    db: &'db Db,
    fork: Snapshot,
    backing: Backing,
}

impl<'db> Branch<'db> {
    fn ephemeral(db: &'db Db, fork: Snapshot) -> Self {
        Self {
            db,
            fork,
            backing: Backing::Ephemeral(HashMap::new()),
        }
    }

    fn durable(db: &'db Db, fork: Snapshot, state: DurableState) -> Self {
        Self {
            db,
            fork,
            backing: Backing::Durable(state),
        }
    }

    /// The parent version this branch forked from. Reads that fall through to the
    /// parent see its state as of exactly this version.
    #[must_use]
    pub const fn fork_version(&self) -> ReadVersion {
        self.fork.read_version()
    }

    /// Whether this branch's writes are persisted (durable named branch) or live
    /// only in memory (ephemeral clone).
    #[must_use]
    pub const fn is_durable(&self) -> bool {
        matches!(self.backing, Backing::Durable(_))
    }

    /// Reads a key on the branch: the branch's own write if it has one, otherwise
    /// the parent's value as of the fork version.
    ///
    /// # Errors
    ///
    /// Returns an error if a bucket cannot be opened or a read fails.
    pub fn get(&self, bucket: impl Into<BucketName>, key: &[u8]) -> Result<Option<Value>> {
        let bucket = bucket.into();
        match &self.backing {
            Backing::Ephemeral(overlay) => match overlay.get(&(bucket.clone(), key.to_vec())) {
                Some(OverlayWrite::Put(value)) => return Ok(Some(value.clone())),
                Some(OverlayWrite::Delete) => return Ok(None),
                None => {}
            },
            Backing::Durable(state) => {
                // Walk the chain leaf-first: the first level that holds the key
                // (a present value or a tombstone) is definitive; otherwise fall
                // through to the next ancestor, and finally to the root snapshot.
                for layer in &state.chain {
                    if !layer.written.contains(bucket.as_str()) {
                        continue;
                    }
                    let data = self
                        .db
                        .bucket_sync(data_bucket(&layer.name, bucket.as_str()))?;
                    let raw = match &layer.at {
                        None => data.get_sync(key)?,
                        Some(at) => data.get_at_sync(at, key)?,
                    };
                    if let Some(raw) = raw {
                        return Ok(decode_branch_value(&raw));
                    }
                }
            }
        }
        self.parent_get(&bucket, key)
    }

    fn parent_get(&self, bucket: &BucketName, key: &[u8]) -> Result<Option<Value>> {
        self.db
            .bucket_sync(bucket.clone())?
            .get_at_sync(&self.fork, key)
    }

    /// Writes a key on the branch. The write is visible to this branch's reads and
    /// never touches the parent. For a durable branch the write is persisted.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting a durable write fails (ephemeral never
    /// fails).
    pub fn put(
        &mut self,
        bucket: impl Into<BucketName>,
        key: impl Into<Vec<u8>>,
        value: impl Into<Value>,
    ) -> Result<()> {
        self.write(bucket.into(), key.into(), OverlayWrite::Put(value.into()))
    }

    /// Deletes a key on the branch (hiding any parent value, via a tombstone for a
    /// durable branch). The parent is unaffected.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting a durable tombstone fails.
    pub fn delete(&mut self, bucket: impl Into<BucketName>, key: impl Into<Vec<u8>>) -> Result<()> {
        self.write(bucket.into(), key.into(), OverlayWrite::Delete)
    }

    fn write(&mut self, bucket: BucketName, key: Vec<u8>, write: OverlayWrite) -> Result<()> {
        let db = self.db;
        match &mut self.backing {
            Backing::Ephemeral(overlay) => {
                overlay.insert((bucket, key), write);
                Ok(())
            }
            Backing::Durable(state) => {
                // Writes only ever touch the leaf (the opened branch's own data
                // bucket); ancestors are read-only fall-through.
                let leaf_fork = state.leaf_fork;
                let leaf_parent = state.leaf_parent.clone();
                let leaf = &mut state.chain[0];
                let data = db.bucket_sync(data_bucket(&leaf.name, bucket.as_str()))?;
                match write {
                    OverlayWrite::Put(value) => data.put_sync(key, encode_present(&value))?,
                    OverlayWrite::Delete => data.put_sync(key, vec![TAG_TOMBSTONE])?,
                }
                // Record the first write to a user bucket so reads consult it (and
                // so the parent is consulted directly for never-written buckets).
                if leaf.written.insert(bucket.as_str().to_owned()) {
                    persist_registry(
                        db,
                        &leaf.name,
                        &RegistryEntry {
                            fork: leaf_fork,
                            parent: leaf_parent,
                            written_buckets: leaf.written.clone(),
                        },
                    )?;
                }
                Ok(())
            }
        }
    }

    /// Scans a key range on the branch, lazily merging its writes over the
    /// parent's state as of the fork (and over each ancestor branch, for a nested
    /// branch): branch puts replace and branch deletes hide the parent's rows.
    /// Returns a [`BranchRange`] iterator yielding the merged rows in key order
    /// without building a full copy — each branch level and the root are streamed
    /// from their own sorted scans and k-way merged on the fly.
    ///
    /// # Errors
    ///
    /// Returns an error if a bucket cannot be opened or a scan cannot be started;
    /// per-row scan errors surface from the iterator.
    pub fn range(&self, bucket: impl Into<BucketName>, range: &KeyRange) -> Result<BranchRange> {
        let bucket = bucket.into();
        // Sources in precedence order, highest first; the root is lowest.
        let mut sources: Vec<MergeSource> = Vec::new();
        match &self.backing {
            Backing::Ephemeral(overlay) => {
                // The overlay is unsorted in memory, so collect its in-range
                // entries for this bucket and sort them into one source.
                let mut entries: Vec<(Vec<u8>, Option<Value>)> = overlay
                    .iter()
                    .filter(|((overlay_bucket, key), _)| {
                        overlay_bucket == &bucket && range_contains(range, key)
                    })
                    .map(|((_, key), write)| {
                        let value = match write {
                            OverlayWrite::Put(value) => Some(value.clone()),
                            OverlayWrite::Delete => None,
                        };
                        (key.clone(), value)
                    })
                    .collect();
                entries.sort_by(|a, b| a.0.cmp(&b.0));
                sources.push(MergeSource::new(Box::new(entries.into_iter().map(Ok))));
            }
            Backing::Durable(state) => {
                for layer in &state.chain {
                    if !layer.written.contains(bucket.as_str()) {
                        continue;
                    }
                    let data = self
                        .db
                        .bucket_sync(data_bucket(&layer.name, bucket.as_str()))?;
                    let rows = match &layer.at {
                        None => data.range_sync(range)?,
                        Some(at) => data.range_at_sync(at, range)?,
                    };
                    sources.push(MergeSource::new(Box::new(rows.map(|row| {
                        row.map(|kv| {
                            let value = decode_branch_value(&kv.value);
                            (kv.key, value)
                        })
                    }))));
                }
            }
        }
        // The root (lowest precedence): every row is a present value.
        let root = self
            .db
            .bucket_sync(bucket.clone())?
            .range_at_sync(&self.fork, range)?;
        sources.push(MergeSource::new(Box::new(
            root.map(|row| row.map(|kv| (kv.key, Some(kv.value)))),
        )));
        Ok(BranchRange { sources })
    }
}

/// One row a merge source yields: its key and either a value or a tombstone
/// (`None`, meaning the level deletes the key).
type MergeRow = Result<(Vec<u8>, Option<Value>)>;

/// A sorted merge source with one buffered head row, so the merge can compare
/// keys across sources before consuming them.
struct MergeSource {
    iter: Box<dyn Iterator<Item = MergeRow>>,
    head: Option<MergeRow>,
}

impl MergeSource {
    fn new(mut iter: Box<dyn Iterator<Item = MergeRow>>) -> Self {
        let head = iter.next();
        Self { iter, head }
    }

    /// The buffered head key, or `None` when the source is exhausted or its head
    /// is an error (handled separately).
    fn key(&self) -> Option<&[u8]> {
        match &self.head {
            Some(Ok((key, _))) => Some(key),
            _ => None,
        }
    }

    fn is_err(&self) -> bool {
        matches!(&self.head, Some(Err(_)))
    }

    /// Takes the head row and refills from the underlying iterator.
    fn take(&mut self) -> Option<MergeRow> {
        let row = self.head.take();
        self.head = self.iter.next();
        row
    }
}

/// A lazy k-way merge of a branch's read chain — the branch's own writes, each
/// ancestor branch, and the root — yielding the resolved rows in key order. The
/// nearest level holding a key wins; a tombstone there hides the key entirely.
/// Returned by [`Branch::range`].
pub struct BranchRange {
    /// Sources in precedence order: index 0 is highest (the branch itself), the
    /// last is the root.
    sources: Vec<MergeSource>,
}

impl Iterator for BranchRange {
    type Item = Result<KeyValue>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            // Surface a pending scan error from any source.
            for source in &mut self.sources {
                if source.is_err() {
                    if let Some(Err(error)) = source.take() {
                        return Some(Err(error));
                    }
                    unreachable!("is_err guarantees an error head");
                }
            }
            // The smallest head key across all sources.
            let mut smallest: Option<&[u8]> = None;
            for source in &self.sources {
                if let Some(key) = source.key() {
                    let replace = match smallest {
                        None => true,
                        Some(current) => key < current,
                    };
                    if replace {
                        smallest = Some(key);
                    }
                }
            }
            let key = smallest?.to_vec();
            // Consume that key from every source; the highest-precedence value
            // (first in source order) wins.
            let mut chosen: Option<Option<Value>> = None;
            for source in &mut self.sources {
                if source.key() == Some(key.as_slice()) {
                    if let Some(Ok((_, value))) = source.take() {
                        if chosen.is_none() {
                            chosen = Some(value);
                        }
                    }
                }
            }
            // A present value is emitted; a tombstone (or nothing) skips the key.
            if let Some(Some(value)) = chosen {
                return Some(Ok(KeyValue::new(key, value)));
            }
        }
    }
}

/// Whether `key` falls within `range`.
fn range_contains(range: &KeyRange, key: &[u8]) -> bool {
    let after_start = match &range.start {
        Bound::Unbounded => true,
        Bound::Included(start) => key >= start.as_slice(),
        Bound::Excluded(start) => key > start.as_slice(),
    };
    let before_end = match &range.end {
        Bound::Unbounded => true,
        Bound::Included(end) => key <= end.as_slice(),
        Bound::Excluded(end) => key < end.as_slice(),
    };
    after_start && before_end
}

/// Decodes a durable branch data value: `Some(value)` for a present write,
/// `None` for a tombstone (deleted on the branch) or a malformed/empty record.
fn decode_branch_value(raw: &[u8]) -> Option<Value> {
    match raw.first() {
        Some(&TAG_PRESENT) => Some(raw[1..].to_vec()),
        _ => None,
    }
}

fn persist_registry(db: &Db, name: &str, entry: &RegistryEntry) -> Result<()> {
    db.bucket_sync(registry_bucket())?
        .put_sync(name.as_bytes().to_vec(), entry.encode())
}

async fn persist_registry_async(db: &Db, name: &str, entry: &RegistryEntry) -> Result<()> {
    db.bucket(registry_bucket())
        .await?
        .put(name.as_bytes().to_vec(), entry.encode())
        .await
}

async fn list_branches_async(db: &Db) -> Result<Vec<String>> {
    let registry = db.bucket(registry_bucket()).await?;
    let mut names = Vec::new();
    for row in registry.range(&KeyRange::all()).await? {
        let row = row?;
        names.push(String::from_utf8(row.key).map_err(|_| Error::Corruption {
            message: "branch registry holds a non-utf8 name".to_owned(),
        })?);
    }
    Ok(names)
}

/// The checkpoint name pinning a durable branch's fork. A checkpoint is durable
/// metadata that the retained-history floor and GC respect, so the parent keeps
/// the branch's fork history across restarts.
fn fork_checkpoint(branch: &str) -> String {
    format!("{RESERVED}fork{SEP}{branch}")
}

impl Db {
    /// Forks an **ephemeral** copy-on-write [`Branch`] from a past `version` — an
    /// `AS OF` read view with an in-memory write overlay that vanishes with the
    /// handle. O(1) and copies no data; the parent is unaffected.
    ///
    /// The fork pins `version`'s history for the branch's lifetime, so it is
    /// subject to the same retained-history floor as [`Db::snapshot_at`].
    ///
    /// # Errors
    ///
    /// Returns an error if `version` is newer than the latest committed version
    /// or older than the retained-history floor.
    pub fn branch_at(&self, version: ReadVersion) -> Result<Branch<'_>> {
        Ok(Branch::ephemeral(self, self.snapshot_at(version)?))
    }

    /// Forks an ephemeral branch from the latest committed version — an instant
    /// in-memory clone of the current state.
    ///
    /// # Errors
    ///
    /// Returns an error if a snapshot at the latest version cannot be pinned.
    pub fn branch_from_latest(&self) -> Result<Branch<'_>> {
        self.branch_at(self.latest_read_version())
    }

    /// Creates a **durable** named branch forked at `from`. The name is recorded
    /// so the branch can be reopened later with [`Db::open_branch`]; its writes
    /// persist in its own buckets. O(1) and copies no data.
    ///
    /// Creating an existing name with the same fork is idempotent; with a
    /// different fork it is an error.
    ///
    /// The fork is pinned with a durable checkpoint, so the parent keeps the
    /// branch's fork history — and the branch stays openable — across restarts
    /// and aggressive retention, until the branch is deleted (no manual retention
    /// configuration needed).
    ///
    /// # Errors
    ///
    /// Returns an error if `from` is not a readable version, if the name already
    /// exists with a different fork, or if persisting the branch fails.
    pub fn create_branch(&self, name: &str, from: ReadVersion) -> Result<()> {
        if let Some(existing) = self.read_registry(name)? {
            if existing.fork == from {
                return Ok(());
            }
            return Err(Error::invalid_options(
                "branch already exists with a different fork version",
            ));
        }
        // Pin the fork durably (this also validates `from` is readable). The
        // checkpoint lives in the manifest, so the parent's GC cannot reclaim the
        // history the branch reads through, even after a restart.
        match self.create_checkpoint_at_sync(&fork_checkpoint(name), from) {
            Ok(()) | Err(Error::CheckpointAlreadyExists { .. }) => {}
            Err(error) => return Err(error),
        }
        persist_registry(
            self,
            name,
            &RegistryEntry {
                fork: from,
                parent: None,
                written_buckets: BTreeSet::new(),
            },
        )
    }

    /// Async-first form of [`Db::create_branch`]. Required for object-store
    /// backends because both the fork checkpoint and the branch registry are
    /// durable metadata writes.
    ///
    /// # Errors
    ///
    /// Same validation and persistence errors as [`Db::create_branch`].
    pub async fn create_branch_at(&self, name: &str, from: ReadVersion) -> Result<()> {
        if let Some(existing) = self.read_registry_async(name).await? {
            if existing.fork == from {
                return Ok(());
            }
            return Err(Error::invalid_options(
                "branch already exists with a different fork version",
            ));
        }
        match self
            .create_checkpoint_at(&fork_checkpoint(name), from)
            .await
        {
            Ok(()) | Err(Error::CheckpointAlreadyExists { .. }) => {}
            Err(error) => return Err(error),
        }
        persist_registry_async(
            self,
            name,
            &RegistryEntry {
                fork: from,
                parent: None,
                written_buckets: BTreeSet::new(),
            },
        )
        .await
    }

    /// Creates a **durable** named branch forked from another branch `parent` at
    /// its current state — a branch of a branch (the git-style DAG). The new
    /// branch reads `parent`'s state (and `parent`'s own ancestors) with its own
    /// writes on top; `parent` is unaffected. O(1), copies no data.
    ///
    /// The fork is pinned with a checkpoint just like [`Db::create_branch`], so
    /// the chain stays readable. Do not delete `parent` while this branch exists
    /// (see [`Db::delete_branch`]).
    ///
    /// # Errors
    ///
    /// Returns an error if `parent` does not exist, if `name` already exists, or
    /// if persisting the branch fails.
    pub fn create_branch_from(&self, name: &str, parent: &str) -> Result<()> {
        if self.read_registry(parent)?.is_none() {
            return Err(Error::invalid_options("parent branch does not exist"));
        }
        if self.read_registry(name)?.is_some() {
            return Err(Error::invalid_options("branch already exists"));
        }
        // Fork at the current global version: the child sees the parent's state as
        // of now. Pinning it keeps the parent's (and its ancestors') history that
        // the chain reads through retained.
        let from = self.latest_read_version();
        match self.create_checkpoint_at_sync(&fork_checkpoint(name), from) {
            Ok(()) | Err(Error::CheckpointAlreadyExists { .. }) => {}
            Err(error) => return Err(error),
        }
        persist_registry(
            self,
            name,
            &RegistryEntry {
                fork: from,
                parent: Some(parent.to_owned()),
                written_buckets: BTreeSet::new(),
            },
        )
    }

    /// Opens a durable named branch, re-pinning its fork and assembling its read
    /// chain (the branch, then each ancestor branch, then the root). The returned
    /// handle sees that chain with the branch's persisted writes on top.
    ///
    /// # Errors
    ///
    /// Returns an error if the branch (or an ancestor) does not exist, or if a
    /// fork version is no longer retained (see the module docs on retention).
    pub fn open_branch(&self, name: &str) -> Result<Branch<'_>> {
        let leaf = self
            .read_registry(name)?
            .ok_or_else(|| Error::invalid_options("no such branch"))?;
        let leaf_fork = leaf.fork;
        let leaf_parent = leaf.parent.clone();

        // The leaf reads its own latest writes; each ancestor is read frozen at
        // the version the child below it forked it.
        let mut chain = vec![DurableLayer {
            name: name.to_owned(),
            written: leaf.written_buckets,
            at: None,
        }];
        let mut child_fork = leaf.fork;
        let mut parent = leaf.parent;
        while let Some(parent_name) = parent {
            let entry = self
                .read_registry(&parent_name)?
                .ok_or_else(|| Error::Corruption {
                    message: format!("branch {parent_name} is missing (an ancestor of {name})"),
                })?;
            chain.push(DurableLayer {
                name: parent_name,
                written: entry.written_buckets,
                at: Some(self.snapshot_at(child_fork)?),
            });
            child_fork = entry.fork;
            parent = entry.parent;
        }
        // The base ancestor forked the root lineage at `child_fork`.
        let root_fork = self.snapshot_at(child_fork)?;
        Ok(Branch::durable(
            self,
            root_fork,
            DurableState {
                chain,
                leaf_fork,
                leaf_parent,
            },
        ))
    }

    /// Lists the durable branch names, in name order.
    ///
    /// # Errors
    ///
    /// Returns an error if the registry cannot be scanned.
    pub fn list_branches(&self) -> Result<Vec<String>> {
        let registry = self.bucket_sync(registry_bucket())?;
        let mut names = Vec::new();
        for row in registry.range_sync(&KeyRange::all())? {
            let row = row?;
            names.push(String::from_utf8(row.key).map_err(|_| Error::Corruption {
                message: "branch registry holds a non-utf8 name".to_owned(),
            })?);
        }
        Ok(names)
    }

    /// Deletes a durable branch: releases its fork pin (so the parent can again
    /// GC that history) and forgets the branch, so it can no longer be opened.
    ///
    /// The branch's data buckets are left in place for now (a bucket-drop
    /// branches forked from it, is an error (a child depends on this branch's
    /// fork pin staying in place).
    ///
    /// The branch's divergent data is reclaimed: each data bucket it wrote is
    /// cleared, so the space is recovered by compaction and a future branch
    /// reusing the name starts clean. (The now-empty bucket shells themselves are
    /// removed only once the KV gains a bucket-drop primitive — a later slice —
    /// but they are gated by the registry's `written_buckets`, so they are never
    /// read after the branch is gone.)
    ///
    /// # Errors
    ///
    /// Returns an error if the branch does not exist, still has children, or if
    /// releasing its state fails.
    pub fn delete_branch(&self, name: &str) -> Result<()> {
        let entry = self
            .read_registry(name)?
            .ok_or_else(|| Error::invalid_options("no such branch"))?;
        // A child branch reads through this branch's history; refuse to drop the
        // pin out from under it.
        for other in self.list_branches()? {
            if other == name {
                continue;
            }
            if let Some(other_entry) = self.read_registry(&other)? {
                if other_entry.parent.as_deref() == Some(name) {
                    return Err(Error::invalid_options(
                        "cannot delete a branch that still has child branches",
                    ));
                }
            }
        }
        // Release the fork pin (the checkpoint may be absent if a prior delete was
        // interrupted after this step — tolerate that).
        match self.delete_checkpoint_sync(&fork_checkpoint(name)) {
            Ok(()) | Err(Error::CheckpointNotFound { .. }) => {}
            Err(error) => return Err(error),
        }
        // Reclaim the branch's divergent data: drop each data bucket it wrote.
        // On a backend without bucket-drop, fall back to clearing the contents so
        // a same-named branch created later does not inherit stale rows (the empty
        // shell remains there).
        for user_bucket in &entry.written_buckets {
            let data = data_bucket(name, user_bucket);
            match self.drop_bucket_sync(data.clone()) {
                Ok(()) => {}
                Err(Error::UnsupportedBackend { .. }) => {
                    self.bucket_sync(data)?.delete_range_sync(KeyRange::all())?;
                }
                Err(error) => return Err(error),
            }
        }
        self.bucket_sync(registry_bucket())?
            .delete_sync(name.as_bytes().to_vec())
    }

    /// Async-first form of [`Db::delete_branch`]. Required for object-store
    /// backends because branch metadata and bucket deletion publish durable
    /// metadata through async compare-and-swap calls.
    ///
    /// # Errors
    ///
    /// Same validation and persistence errors as [`Db::delete_branch`].
    pub async fn delete_branch_async(&self, name: &str) -> Result<()> {
        let entry = self
            .read_registry_async(name)
            .await?
            .ok_or_else(|| Error::invalid_options("no such branch"))?;
        for other in list_branches_async(self).await? {
            if other == name {
                continue;
            }
            if let Some(other_entry) = self.read_registry_async(&other).await? {
                if other_entry.parent.as_deref() == Some(name) {
                    return Err(Error::invalid_options(
                        "cannot delete a branch that still has child branches",
                    ));
                }
            }
        }
        match self.delete_checkpoint(&fork_checkpoint(name)).await {
            Ok(()) | Err(Error::CheckpointNotFound { .. }) => {}
            Err(error) => return Err(error),
        }
        for user_bucket in &entry.written_buckets {
            let data = data_bucket(name, user_bucket);
            match self.drop_bucket(data.clone()).await {
                Ok(()) => {}
                Err(Error::UnsupportedBackend { .. }) => {
                    self.bucket(data)
                        .await?
                        .delete_range(KeyRange::all())
                        .await?;
                }
                Err(error) => return Err(error),
            }
        }
        self.bucket(registry_bucket())
            .await?
            .delete(name.as_bytes().to_vec())
            .await
    }

    /// Returns a durable branch's lineage (its fork version and parent branch),
    /// or `None` when no such branch exists — without assembling a read chain or
    /// opening any data bucket.
    ///
    /// This lets a higher layer reuse this crate's durable branch lifecycle (the
    /// fork pin that survives restarts and aggressive GC, the registry, and
    /// nesting) while storing its **own** divergent data and doing its own
    /// fall-through reads against [`Db::snapshot_at`] of the returned
    /// [`BranchInfo::fork`]. Combine with [`Db::create_branch`] /
    /// [`Db::create_branch_from`] / [`Db::list_branches`] / [`Db::delete_branch`].
    ///
    /// # Errors
    ///
    /// Returns an error if the registry cannot be read or a stored entry is
    /// malformed.
    pub fn branch_info(&self, name: &str) -> Result<Option<BranchInfo>> {
        Ok(self.read_registry(name)?.map(|entry| BranchInfo {
            fork: entry.fork,
            parent: entry.parent,
        }))
    }

    fn read_registry(&self, name: &str) -> Result<Option<RegistryEntry>> {
        match self
            .bucket_sync(registry_bucket())?
            .get_sync(name.as_bytes())?
        {
            Some(bytes) => Ok(Some(RegistryEntry::decode(&bytes)?)),
            None => Ok(None),
        }
    }

    async fn read_registry_async(&self, name: &str) -> Result<Option<RegistryEntry>> {
        match self
            .bucket(registry_bucket())
            .await?
            .get(name.as_bytes())
            .await?
        {
            Some(bytes) => Ok(Some(RegistryEntry::decode(&bytes)?)),
            None => Ok(None),
        }
    }
}

#[cfg(test)]
mod tests;