haematite 0.6.2

//! 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::sync::{Mutex, MutexGuard};

use crate::store::NodeStore;
use crate::tree::Hash;
use crate::tree::mutate::batch_mutate_owned_with_source;
use crate::wal::{Mutation, WalBuffer};

pub use super::commit_error::BranchCommitError;
use super::conflict::ConflictPolicy;
use super::handle::{BranchHandle, RefBinding, ShardId, ShardState};
use super::merge::{MergeReport, merge_with_report};
use super::operation_error::{CommitOperationError, preserve_node_publication_fence};
use super::policy::check_merge;
use super::refstore::{BranchRefError, BranchRefStore};
use super::registry::{BranchRegistry, advance_in};
use super::snapshot::Timestamp;

/// 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> {
    commit_branch_with_source(branch, store, registry, request)
        .map_err(CommitOperationError::into_public)
}

pub(crate) fn commit_branch_with_source<S: NodeStore + ?Sized>(
    branch: &BranchHandle,
    store: &mut S,
    registry: &BranchRegistry,
    request: CommitRequest<'_>,
) -> Result<CommitOutcome, CommitOperationError<S::Error>> {
    let CommitRequest {
        durability,
        extra_parents,
        timestamp,
    } = request;

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

    // 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.into()),
        },
    };
    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_with_source(
            store,
            state.current_root,
            buffered_batch(&state.buffer),
        )
        .map_err(CommitOperationError::Tree)?;
        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.
        preserve_node_publication_fence(store.sync_dirty_dirs())
            .map_err(CommitOperationError::Fence)?;

        // 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)).into());
        };
        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,
    })
}

/// Merge one named branch into another after resolving durable kind policy.
///
/// This is the public content-merge entry point. The root-level three-way
/// walker is crate-private because roots alone carry no branch identity and
/// therefore cannot enforce Namespace boundaries. Both identities are taken
/// from handle bindings, while both kinds and lineages are resolved from the
/// matching ref records; callers cannot relabel either side.
pub fn merge<S: NodeStore + ?Sized>(
    store: &mut S,
    source: &BranchHandle,
    destination: &BranchHandle,
    refs: &BranchRefStore,
    policy: &ConflictPolicy,
) -> Result<MergeReport, BranchCommitError> {
    let source_binding = source
        .binding_snapshot()
        .ok_or(BranchCommitError::UnnamedBranch)?;
    let destination_binding = destination
        .binding_snapshot()
        .ok_or(BranchCommitError::UnnamedBranch)?;

    let source_record = refs.cas_check(
        &source_binding.name,
        source_binding.created,
        source_binding.seq,
    )?;
    let source_name = source_record.name.clone();
    let source_kind = source_record.kind;
    let source_lineage = source_record
        .resolved_namespace_lineage()
        .map(str::to_owned);

    let destination_record = refs.cas_check(
        &destination_binding.name,
        destination_binding.created,
        destination_binding.seq,
    )?;
    let destination_name = destination_record.name.clone();
    let destination_kind = destination_record.kind;
    let destination_lineage = destination_record
        .resolved_namespace_lineage()
        .map(str::to_owned);

    check_merge(
        &source_name,
        source_kind,
        source_lineage.as_deref(),
        &destination_name,
        destination_kind,
        destination_lineage.as_deref(),
    )?;
    if source.shard_count() != 1 || destination.shard_count() != 1 {
        return Err(super::merge::MergeError::Unimplemented {
            feature: "multi-shard branch merge",
        }
        .into());
    }

    merge_with_report(
        store,
        destination.current_root(),
        source.current_root(),
        source.fork_point(),
        policy,
    )
    .map_err(BranchCommitError::from)
}

/// 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;