haematite 0.4.1

//! LEDGER A1: the branch commit path (`commit_branch`).
//!
//! A branch commit drains every dirty shard buffer into the shared
//! content-addressed tree, advances the branch's in-memory heads, and — for
//! [`CommitDurability::Durable`] — installs the branch's replacement HBR1
//! record in one atomic rename. The exact step ordering is
//! BRANCH-COMMIT-PATH.md §5, as amended by §14.1 (refcount-2 anchors, no
//! first-advance special case), §14.2 (record-only commits), §16.1 (MF1:
//! every lock is acquired up front, so nothing fallible follows the commit
//! point) and §16.3 (durable-to-durable parent lineage).
//!
//! # Commit/prune borrow exclusivity (§6(b), §14.4)
//!
//! [`commit_branch`] takes the node store as `&mut S`; [`prune`](super::prune)
//! borrows it as `&S`. Safe Rust therefore cannot overlap the two on one
//! store: every prune runs entirely-before or entirely-after each commit
//! call, which is what closes the materialisation gap — the window in which
//! a commit's freshly written nodes are reachable from no live root (and,
//! under content-address dedup, may be byte-identical to nodes reachable only
//! from the snapshot being pruned). Entirely-before, this commit's nodes do
//! not exist yet and any dedup casualty is rewritten by our own `put`;
//! entirely-after, steps 4-5 registered and durably recorded the new heads
//! before we returned. The argument is carried by the borrow checker, so an
//! interior-mutable store wrapper (one that hands out `&S` while writes
//! proceed) voids it and must itself fence prune against commit — the
//! commit/prune `RwLock` successor recorded in §14.4 for the A2/A6
//! shared-store work.

use std::fmt;
use std::sync::{Mutex, MutexGuard};

use crate::store::NodeStore;
use crate::tree::{Hash, TreeError, batch_mutate_owned};
use crate::wal::{Mutation, WalBuffer};

use super::handle::{BranchError, BranchHandle, RefBinding, ShardId, ShardState};
use super::refstore::{BranchRefError, BranchRefStore};
use super::registry::{BranchRegistry, advance_in};
use super::snapshot::Timestamp;

/// Errors raised by the branch commit path and the named-branch lifecycle.
#[derive(Debug)]
pub enum BranchCommitError {
    /// The handle holds no registry guard: an unregistered handle advancing a
    /// root is precisely the prune registration hazard (prune.rs contract)
    /// and is refused, not documented around.
    Unregistered,
    /// A durable commit was requested on a handle with no name binding.
    /// Anonymous (`fork*`) branches are volatile by contract; use
    /// `create_branch`/`open_branch` for a durable name.
    UnnamedBranch,
    /// `extra_parents` were supplied with [`CommitDurability::Volatile`].
    /// Parent links are durable-record content (§16.3: lineage is
    /// durable-to-durable); a volatile commit has no record to carry them,
    /// and silently dropping converge links is worse than refusing.
    VolatileExtraParents,
    /// A fork-by-hash anchor is protected by no live branch pin, no named
    /// branch record, and no named snapshot (§16.4 Q3 hard refusal).
    UnprotectedAnchor {
        /// The refused anchor root.
        root: Hash,
    },
    /// A branch-handle operation failed (poisoned shard state, unknown or
    /// duplicate shard).
    Branch(BranchError),
    /// Materialising a shard buffer into the shared tree failed.
    Tree(TreeError),
    /// The store's dirty-directory barrier failed before the durable install.
    Barrier(String),
    /// The durable ref store refused or failed the record install
    /// (`StaleSeq`, `BranchGenerationMismatch`, `BranchRemoved`,
    /// `DuplicateBranch`, I/O, ...).
    Ref(BranchRefError),
}

impl fmt::Display for BranchCommitError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Unregistered => write!(
                formatter,
                "branch handle is not registered against a branch registry; \
                 fork it via a registering constructor before committing"
            ),
            Self::UnnamedBranch => write!(
                formatter,
                "durable commit requires a named branch handle from create_branch/open_branch"
            ),
            Self::VolatileExtraParents => write!(
                formatter,
                "extra parents require a durable commit: volatile commits install no record to carry them"
            ),
            Self::UnprotectedAnchor { root } => write!(
                formatter,
                "fork anchor {root} is protected by no live branch, no named branch record, and \
                 no named snapshot, so a prune could reclaim it at any moment; name a snapshot \
                 at this root first (SnapshotRegistry::name), then fork"
            ),
            Self::Branch(error) => write!(formatter, "branch commit handle error: {error}"),
            Self::Tree(error) => write!(formatter, "branch commit tree error: {error}"),
            Self::Barrier(reason) => {
                write!(
                    formatter,
                    "branch commit durability barrier failed: {reason}"
                )
            }
            Self::Ref(error) => write!(formatter, "branch commit ref store error: {error}"),
        }
    }
}

impl std::error::Error for BranchCommitError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Branch(error) => Some(error),
            Self::Tree(error) => Some(error),
            Self::Ref(error) => Some(error),
            Self::Unregistered
            | Self::UnnamedBranch
            | Self::VolatileExtraParents
            | Self::UnprotectedAnchor { .. }
            | Self::Barrier(_) => None,
        }
    }
}

impl From<BranchError> for BranchCommitError {
    fn from(error: BranchError) -> Self {
        Self::Branch(error)
    }
}

impl From<TreeError> for BranchCommitError {
    fn from(error: TreeError) -> Self {
        Self::Tree(error)
    }
}

impl From<BranchRefError> for BranchCommitError {
    fn from(error: BranchRefError) -> Self {
        Self::Ref(error)
    }
}

/// How durable a branch commit must be — explicit, no silent default (§4).
///
/// # Volatile-then-durable soundness (§4)
///
/// A volatile commit on a `DiskStore` still `put`s node files (each file's
/// DATA is fsynced and its parent directory is recorded in the store's dirty
/// set). `DiskStore`'s dirty set is cleared ONLY by a `sync_dirty_dirs`
/// barrier (disk.rs), never by the puts themselves, so a later durable
/// commit's barrier also fences the directory entries left pending by every
/// earlier volatile commit — including nodes a durable commit dedup-reuses
/// from a volatile one via `write_node`'s `path_exists` short-circuit. A
/// durable record therefore never names a root whose nodes' directory
/// entries could vanish on power loss.
#[derive(Debug)]
pub enum CommitDurability<'a> {
    /// Advance in memory only: no dirty-dir barrier, no ref-record install.
    /// The advance (and the branch itself, if anonymous) does not survive the
    /// process. Frame speculative writes; norn's intra-cadence checkpoints.
    Volatile,
    /// Barrier + atomic ref-record install. Requires a named handle
    /// (binding present), else typed [`BranchCommitError::UnnamedBranch`].
    Durable {
        /// The durable ref store holding this branch's HBR1 record.
        refs: &'a mut BranchRefStore,
    },
}

/// One branch commit request.
#[derive(Debug)]
pub struct CommitRequest<'a> {
    /// Volatile or durable — explicit, no silent default.
    pub durability: CommitDurability<'a>,
    /// R4 converge/DAG links recorded beyond the implicit previous-record
    /// parent. Durable only ([`BranchCommitError::VolatileExtraParents`]).
    pub extra_parents: &'a [Hash],
    /// Caller-supplied wall clock (`current_timestamp()`); no hidden clock
    /// reads inside the engine.
    pub timestamp: Timestamp,
}

/// What a branch commit did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommitOutcome {
    /// Post-commit in-memory heads, ascending shard id. For a durable commit
    /// these are the heads the installed record names for every shard that
    /// materialised new content; shards left untouched keep their previous
    /// durable head in the record even if volatile commits advanced them in
    /// memory (volatile advances are volatile, full stop — §4).
    pub heads: Vec<(ShardId, Hash)>,
    /// `Some(new_seq)` for a durable commit that installed a record;
    /// `Some(current_seq)` for a durable no-op; `None` for volatile.
    pub seq: Option<u64>,
    /// `false` exactly when every buffer was empty and `extra_parents` was
    /// empty: the documented no-op — nothing drained, nothing installed.
    pub advanced: bool,
}

/// Commit a branch: drain every dirty shard buffer into the tree and advance
/// the branch heads, all-shards-atomically (§5, §7).
///
/// `registry` MUST be the registry the handle was registered against — the
/// handle's own guard retargets its pins there, and the step-5 head swap runs
/// under that registry's one counts lock so no prune `live_roots()` snapshot
/// can observe neither the superseded nor the advanced head.
///
/// Failure disposition (§5 step 2): on ANY error nothing observable has
/// changed — no state swapped, no registration moved, buffers intact — and
/// the operation is retryable. Nodes already written by a failed
/// materialisation are orphans: leak-safe, and a retry re-derives the
/// identical hashes (history independence), so the orphans are simply reused.
///
/// See the module docs for the commit/prune borrow-exclusivity contract
/// (§6(b), §14.4): taking the store as `&mut S` here is what serialises every
/// prune entirely-before or entirely-after this call.
pub fn commit_branch<S: NodeStore + ?Sized>(
    branch: &BranchHandle,
    store: &mut S,
    registry: &BranchRegistry,
    request: CommitRequest<'_>,
) -> Result<CommitOutcome, BranchCommitError> {
    let CommitRequest {
        durability,
        extra_parents,
        timestamp,
    } = request;

    let Some(registry_guard) = branch.registry_guard() else {
        return Err(BranchCommitError::Unregistered);
    };
    if matches!(durability, CommitDurability::Volatile) && !extra_parents.is_empty() {
        return Err(BranchCommitError::VolatileExtraParents);
    }

    // Step 1 (§5 as amended by §16.1 MF1): acquire ALL locks up front, in the
    // stated total order — shard-state mutexes ascending by shard id, then
    // the binding mutex, then the registry counts guard — and hold them to
    // the end. Lock acquisition is the last thing here that can fail
    // (poisoned shard state), so it MUST all precede the step-4 durable
    // install: a lock failure after the record renamed in would return `Err`
    // from a commit that durably happened, leaving the in-memory state stale
    // and every retry `StaleSeq`. With every guard already in hand, steps
    // 5-6 are pure writes that cannot fail. Holding the shard guards
    // throughout also makes the commit atomic with respect to every handle
    // clone (§7): no clone can observe some shards advanced and others not.
    let mut states = branch.lock_states_ascending()?;
    let mut binding = branch.binding.as_deref().map(lock_binding);
    let durable = match durability {
        CommitDurability::Volatile => None,
        CommitDurability::Durable { refs } => match binding.as_deref() {
            Some(bound) => Some((refs, bound.name.clone(), bound.created, bound.seq)),
            None => return Err(BranchCommitError::UnnamedBranch),
        },
    };
    let mut counts = registry.lock_counts();

    // The documented no-op (§14.2): nothing buffered and no converge links.
    // Checked under the locks so the answer cannot race a sibling clone's put.
    // Durable handles still run the §16.2 CAS first (review finding): a stale
    // handle on a removed or recreated branch must hear
    // BranchRemoved/BranchGenerationMismatch/StaleSeq here too, never a
    // reassuring Ok with a seq the ref store no longer recognises.
    if extra_parents.is_empty() && states.iter().all(|(_, state)| state.buffer.is_empty()) {
        if let Some((refs, name, created, seq)) = &durable {
            refs.cas_check(name, *created, *seq)?;
        }
        return Ok(CommitOutcome {
            heads: current_heads(&states),
            seq: durable.map(|(_, _, _, seq)| seq),
            advanced: false,
        });
    }

    // Step 2: materialise each dirty shard's buffer into the tree. The
    // buffers are NOT cleared yet: a tree/store failure here returns with
    // every buffer intact and every root unmoved (retryable, mirroring
    // WalBuffer::commit's retain-on-failure discipline). `None` = shard had
    // nothing buffered.
    let mut materialised: Vec<Option<Hash>> = Vec::with_capacity(states.len());
    for (_, state) in &states {
        if state.buffer.is_empty() {
            materialised.push(None);
            continue;
        }
        let new_root =
            batch_mutate_owned(store, state.current_root, buffered_batch(&state.buffer))?;
        materialised.push(Some(new_root));
    }

    // Shards whose head actually moves: a buffer can materialise to the same
    // root (e.g. rewriting existing values), which drains the buffer but
    // advances nothing.
    let advanced: Vec<(ShardId, Hash, Hash)> = states
        .iter()
        .zip(&materialised)
        .filter_map(|((shard_id, state), new_root)| {
            new_root.and_then(|new_root| {
                (new_root != state.current_root).then_some((
                    *shard_id,
                    state.current_root,
                    new_root,
                ))
            })
        })
        .collect();

    let new_seq = if let Some((refs, name, created, expected_seq)) = durable {
        // Step 3: the Tier-0 durability barrier, STRICTLY BEFORE the record
        // install (same load-bearing ordering as shard/actor.rs::commit).
        // Every node this commit wrote — and every node an earlier volatile
        // commit wrote and this record's heads now reach (§4 soundness note
        // on CommitDurability) — has fsynced DATA but possibly no durable
        // directory entry; fencing the dirty directories first means a power
        // loss can never leave an installed HBR1 record naming roots whose
        // nodes' directory entries are gone.
        store
            .sync_dirty_dirs()
            .map_err(|error| BranchCommitError::Barrier(error.to_string()))?;

        // Parents rule (§16.3, uniform): parents[0..] = the previous durable
        // RECORD's heads — durable-to-durable lineage, never the in-memory
        // roots, skipping volatile intermediates — then the caller's converge
        // links. A record-only commit (§14.2) reaches here with `advanced`
        // empty and self-parents its unchanged heads: no DAG special case.
        let Some(record) = refs.get(&name) else {
            return Err(BranchCommitError::Ref(BranchRefError::BranchRemoved(name)));
        };
        let parents: Vec<Hash> = record
            .shards
            .iter()
            .map(|shard| shard.head)
            .chain(extra_parents.iter().copied())
            .collect();
        let new_heads: Vec<(ShardId, Hash)> = advanced
            .iter()
            .map(|&(shard_id, _, new_root)| (shard_id, new_root))
            .collect();

        // Step 4: the commit point. One atomic install (temp → fsync →
        // rename → dir fsync) covering ALL shards: recovery sees every old
        // head or every new head, never a mix (§7). The CAS inside `advance`
        // (creation identity AND seq, §16.2) is the last fallible operation
        // in the whole commit.
        Some(refs.advance(&name, created, expected_seq, &new_heads, parents, timestamp)?)
    } else {
        None
    };

    // Step 5 (pure writes, §14.1): retarget the in-memory pins, one shard at
    // a time, inside the counts guard held since step 1 — no prune
    // `live_roots()` snapshot can ever observe neither the superseded nor
    // the advanced head. Only the head-role pin moves; the anchor-role pin
    // (refcount-2 registration at fork/create/open) keeps the fork anchor
    // live for the guard's whole life, so there is no first-advance special
    // case. The guard's pin list is retargeted in the same breath so its
    // drop releases the CURRENT pins, not stale ones.
    for &(_, old_root, new_root) in &advanced {
        advance_in(&mut counts, old_root, new_root);
        registry_guard.replace(old_root, new_root);
    }

    // Step 6 (pure writes): swap each drained shard's {current_root, buffer}
    // as one unit under its still-held state guard — the §3 clone-coherence
    // invariant — and bump the binding to the sequence the record now holds,
    // so this handle's next durable commit passes the CAS.
    for ((_, state), new_root) in states.iter_mut().zip(&materialised) {
        if let Some(new_root) = new_root {
            state.current_root = *new_root;
            state.buffer = WalBuffer::new();
        }
    }
    if let (Some(bound), Some(seq)) = (binding.as_deref_mut(), new_seq) {
        bound.seq = seq;
    }

    let heads = current_heads(&states);
    // Every write above is complete: release the step-1 locks explicitly, in
    // reverse acquisition order ("locks release; done" — §5 step 6).
    drop(counts);
    drop(binding);
    drop(states);

    Ok(CommitOutcome {
        heads,
        seq: new_seq,
        advanced: true,
    })
}

/// Poison-tolerant binding lock (tier 2 of the §16.1 order). `RefBinding` is
/// only ever mutated as a single `seq` field store under the full commit lock
/// set, so a poisoned mutex still holds a coherent value; recovering keeps
/// binding acquisition infallible, which MF1's "nothing fallible after the
/// commit point" argument leans on.
fn lock_binding(binding: &Mutex<RefBinding>) -> MutexGuard<'_, RefBinding> {
    match binding.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    }
}

/// One final intent per key, in key order: the buffer is a `BTreeMap` with
/// last-wins overwrite, so the batch is a pure function of the buffered SET,
/// not the put order — the first leg of the §10 history-independence proof.
fn buffered_batch(buffer: &WalBuffer) -> Vec<(Vec<u8>, Option<Vec<u8>>)> {
    buffer
        .iter()
        .map(|mutation| match mutation {
            Mutation::Put { key, value } => (key.clone(), Some(value.clone())),
            Mutation::Delete { key } => (key.clone(), None),
        })
        .collect()
}

fn current_heads(states: &[(ShardId, MutexGuard<'_, ShardState>)]) -> Vec<(ShardId, Hash)> {
    states
        .iter()
        .map(|(shard_id, state)| (*shard_id, state.current_root))
        .collect()
}

#[cfg(test)]
#[path = "commit_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "commit_error_tests.rs"]
mod error_tests;