haematite 0.6.1

//! LEDGER A1: the durable branch-ref substrate (`BranchRefStore`, HBR1).
//!
//! One file per named branch under a caller-supplied directory (no silent
//! default). Each file is a single atomically-installed HBR1 record naming the
//! branch's per-shard fork anchors and committed heads, so "which root is this
//! branch at, and what pins it against prune" is never split across two
//! durable writes: the rename swaps old-head-pin and new-head-pin in one
//! indivisible step (BRANCH-COMMIT-PATH.md §1 Rule 1, §2).
//!
//! ## HBR1 record layout (§2.2, as amended by §16.2)
//!
//! ```text
//! magic            "HBR1"                       (4 bytes)
//! name             len-prefixed bytes
//! created          u64   // creation identity: the create-time timestamp (§16.2)
//! seq              u64   // durable commit sequence; 0 = created, never committed
//! timestamp        u64   // ns since epoch, caller-supplied
//! shard_count      u64
//! per shard (strictly ascending shard id):
//!   shard_id       u64   // ShardId is usize; encoded u64, checked narrow on decode
//!   fork_anchor    32-byte hash   // divergence point — the merge ancestor.
//!                                 // Immutable for branch life (§16.3).
//!   head           32-byte hash   // current committed root for this shard
//! parent_count     u64
//! per parent:      32-byte hash   // tree roots of this head commit's parents
//! optional kind extension (absent in pre-marker records):
//!   magic          "HBK1"         // extension/version stamp
//!   kind           u64            // 0 = Work, 1 = Namespace
//!   lineage_tag    u64            // 0 = none, 1 = namespace name follows
//!   lineage        len-prefixed bytes (only when lineage_tag = 1)
//! (other trailing bytes rejected)
//! ```
//!
//! The tail position makes absence unambiguous: old records decode as `Work`.
//!
//! Torn writes are impossible (atomic rename); corruption therefore means bit
//! rot, and [`BranchRefStore::open`] **fails loud** naming the file rather
//! than skipping it — silently dropping a ref record would silently drop a
//! prune pin, converting bit rot into cascading node reclamation (§2.2).
//!
//! ## The in-memory map never trails the disk
//!
//! `records` mirrors the directory, and [`BranchRefStore::protected_roots`]
//! (prune's durable pin source) reads the MAP, not the disk. An install whose
//! rename landed but whose directory-entry fsync then failed
//! ([`InstallError::InstalledUnfenced`]) has still changed disk truth — a
//! cold reopen or a crash recovery would surface the new record — so
//! [`BranchRefStore::create`] and [`BranchRefStore::advance`] adopt the
//! replacement into the map on that error path before returning it. Adopting
//! over-pins at worst (leak-safe); keeping the old record would under-pin and
//! let a prune reclaim nodes a durable record references. `remove` needs no
//! mirror-image handling: keeping the removed record on its error path is the
//! conservative (over-pinning) direction already.

use std::collections::{BTreeMap, HashSet};
use std::fmt;
use std::path::Path;

use super::durable_record::{
    CreateExclusive, DurableRecordStore, EntryFence, entry_fence_with_source,
};
use super::native_record_store::NativeDurableRecordStore;
use super::operation_error::RecordOperationError;
use super::persist::CodecError;
use super::policy::BranchKind;
pub use super::refrecord::{BranchRefRecord, BranchShardRef};
use super::snapshot::Timestamp;
use crate::ids::ShardId;
use crate::tree::Hash;

#[path = "refstore_codec.rs"]
pub(crate) mod codec;
// Shared with the vacuum's read-only HBR1 reader (STORAGE-VACUUM.md §7
// ⟨r4, M4⟩): the vacuum decodes records through EXACTLY these functions, so
// codec drift between the mutating store and the report-only reader is
// impossible by construction.
#[cfg(test)]
use codec::encode_record;
pub(crate) use codec::{REF_EXTENSION, decode_record, ref_file_name};

/// Temp-file naming pinned in `persist.rs` (`.branch-*.tmp`); the sweep in
/// [`BranchRefStore::open`] matches exactly this pattern (§16.3), and the
/// vacuum's read-only reader counts (never unlinks) the same pattern.
pub(crate) const TEMP_PREFIX: &str = ".branch-";
pub(crate) const TEMP_SUFFIX: &str = ".tmp";

/// Errors raised by the branch ref store.
#[derive(Debug)]
pub enum BranchRefError {
    /// A branch with this name already exists (create).
    DuplicateBranch(String),
    /// The requested name's file hash collides with a different existing
    /// branch name — fail loud, never silent reuse (§2.1).
    NameHashCollision { requested: String, existing: String },
    /// CAS failure: the stored commit sequence differs from what the caller's
    /// handle last observed — two handles on one named branch cannot silently
    /// clobber each other.
    StaleSeq {
        name: String,
        expected: u64,
        found: u64,
    },
    /// CAS failure: the stored creation identity differs — the branch was
    /// removed and recreated under the same name since the caller's handle
    /// was bound (§16.2's ABA case).
    BranchGenerationMismatch {
        name: String,
        expected_created: Timestamp,
        found_created: Timestamp,
    },
    /// Advance on a branch whose record no longer exists (§16.2).
    BranchRemoved(String),
    /// A create tried to persist a kind/lineage combination forbidden by HBK1.
    InvalidKindLineage {
        /// Durable branch name from the refused record.
        name: String,
        /// Durable kind from the refused record.
        kind: BranchKind,
    },
    /// A create carried two entries for one shard.
    DuplicateShard { name: String, shard_id: ShardId },
    /// An advance named a shard the branch's record does not carry.
    UnknownShard { name: String, shard_id: ShardId },
    /// A persisted record could not be decoded.
    Corrupt(String),
    /// An I/O error occurred while persisting or loading.
    Io(std::io::Error),
}

impl fmt::Display for BranchRefError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::DuplicateBranch(name) => write!(f, "branch already exists: {name}"),
            Self::NameHashCollision {
                requested,
                existing,
            } => write!(
                f,
                "branch name hash collision: {requested:?} maps to the same ref file as existing branch {existing:?}"
            ),
            Self::StaleSeq {
                name,
                expected,
                found,
            } => write!(
                f,
                "stale branch commit sequence for {name}: handle expected {expected}, record holds {found}"
            ),
            Self::BranchGenerationMismatch {
                name,
                expected_created,
                found_created,
            } => write!(
                f,
                "branch generation mismatch for {name}: handle was bound to creation {expected_created}, record was created at {found_created}"
            ),
            Self::BranchRemoved(name) => write!(f, "branch has been removed: {name}"),
            Self::InvalidKindLineage { name, kind } => write!(
                f,
                "branch {name} of kind {kind} cannot store a namespace lineage"
            ),
            Self::DuplicateShard { name, shard_id } => {
                write!(f, "branch {name} names shard {shard_id} twice")
            }
            Self::UnknownShard { name, shard_id } => {
                write!(f, "branch {name} has no shard {shard_id}")
            }
            Self::Corrupt(reason) => write!(f, "branch ref store corrupted: {reason}"),
            Self::Io(error) => write!(f, "branch ref store I/O error: {error}"),
        }
    }
}

impl std::error::Error for BranchRefError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            _ => None,
        }
    }
}

impl From<std::io::Error> for BranchRefError {
    fn from(error: std::io::Error) -> Self {
        Self::Io(error)
    }
}

impl From<CodecError> for BranchRefError {
    fn from(error: CodecError) -> Self {
        match error {
            CodecError::Corrupt(reason) => Self::Corrupt(reason),
            CodecError::Io(io_error) => Self::Io(io_error),
        }
    }
}

/// Persistent store of one HBR1 record per named branch (§2).
///
/// Prune's durable denominator: [`BranchRefStore::protected_roots`] is the set
/// of roots named branches pin across restarts, with zero startup ceremony.
#[derive(Debug)]
pub struct BranchRefStore {
    backend: Box<dyn DurableRecordStore<Error = BranchRefError>>,
    records: BTreeMap<String, BranchRefRecord>,
}

/// Compile-time wall for the v0.6.0 auto-trait regression: `BranchRefStore`
/// must stay `Send + Sync`. The 0.6.0 durable-record refactor boxed the
/// backend seam without a bound — trait objects are `?Send` by default — so
/// both auto-traits silently vanished from `BranchRefStore` and every type
/// embedding it, surfacing downstream as E0277 in threaded consumers. The
/// `DurableRecordStore` supertrait restores them; these asserts turn any
/// future loss into a local compile error instead of a downstream one.
const _: () = {
    const fn assert_send_sync<T: Send + Sync>() {}
    assert_send_sync::<BranchRefStore>();
    assert_send_sync::<NativeDurableRecordStore>();
};

impl BranchRefStore {
    /// Opens (creating if needed) the ref store rooted at `dir`, sweeping any
    /// stale `.branch-*.tmp` files orphaned by a crash mid-install and loading
    /// every `*.ref` record. A record that fails to decode — or that lives in
    /// a file whose name does not match its branch name's hash — fails the
    /// open loud (§2.2): a silently dropped record is a silently dropped
    /// prune pin.
    pub fn open<P: AsRef<Path>>(dir: P) -> Result<Self, BranchRefError> {
        let backend = NativeDurableRecordStore::open(dir)?;
        Self::from_backend(backend)
    }

    pub(super) fn from_backend(
        backend: impl DurableRecordStore<Error = BranchRefError> + 'static,
    ) -> Result<Self, BranchRefError> {
        let mut backend: Box<dyn DurableRecordStore<Error = BranchRefError>> = Box::new(backend);
        let records = backend
            .list_read_at_open()?
            .into_iter()
            .map(|record| (record.name.clone(), record))
            .collect();
        Ok(Self { backend, records })
    }

    /// Returns the record for `name`, or `None` if no such branch exists.
    pub fn get(&self, name: &str) -> Option<&BranchRefRecord> {
        self.records.get(name)
    }

    /// Iterates every branch record in name order.
    pub fn list(&self) -> impl Iterator<Item = &BranchRefRecord> {
        self.records.values()
    }

    /// Union of every record's fork anchors AND heads — prune's durable pin
    /// set (§2.3). Anchors stay protected for the branch's durable lifetime
    /// because `merge` still needs the divergence point after the head moves.
    pub fn protected_roots(&self) -> HashSet<Hash> {
        self.records
            .values()
            .flat_map(|record| record.shards.iter())
            .flat_map(|shard| [shard.fork_anchor, shard.head])
            .collect()
    }

    /// Reserve-then-create in ONE atomic no-clobber install (§2.3, norn R6).
    ///
    /// Shard entries are normalised to ascending shard-id order. If the target
    /// ref file already exists it is decoded: same name ⇒
    /// [`BranchRefError::DuplicateBranch`]; different name ⇒
    /// [`BranchRefError::NameHashCollision`] (§2.1).
    ///
    /// The blessed caller is `create_branch` (§5), which also registers the
    /// in-memory pins. Crate-private (§2.3): a direct external caller could
    /// record a branch with no pins, exactly the confinement this brief
    /// exists to enforce.
    pub(crate) fn create(&mut self, record: BranchRefRecord) -> Result<(), BranchRefError> {
        self.create_with_source(record)
            .map_err(RecordOperationError::into_public)
    }

    pub(crate) fn create_with_source(
        &mut self,
        mut record: BranchRefRecord,
    ) -> Result<(), RecordOperationError<BranchRefError>> {
        if record.kind == BranchKind::Namespace && record.namespace_lineage.is_some() {
            return Err(BranchRefError::InvalidKindLineage {
                name: record.name,
                kind: record.kind,
            }
            .into());
        }
        if self.records.contains_key(&record.name) {
            return Err(BranchRefError::DuplicateBranch(record.name).into());
        }
        record.shards.sort_by_key(|shard| shard.shard_id);
        if let Some(pair) = record
            .shards
            .windows(2)
            .find(|pair| pair[0].shard_id == pair[1].shard_id)
        {
            return Err(BranchRefError::DuplicateShard {
                name: record.name,
                shard_id: pair[0].shard_id,
            }
            .into());
        }

        match self.backend.create_exclusive(&record)? {
            CreateExclusive::Installed => {
                self.records.insert(record.name.clone(), record.clone());
                entry_fence_with_source(self.backend.as_mut(), EntryFence::Present(&record))
                    .map_err(RecordOperationError::Fence)
            }
            CreateExclusive::TargetExists(existing) if existing.name == record.name => {
                Err(BranchRefError::DuplicateBranch(record.name).into())
            }
            CreateExclusive::TargetExists(existing) => Err(BranchRefError::NameHashCollision {
                requested: record.name,
                existing: existing.name,
            }
            .into()),
        }
    }

    /// CAS-guarded head advance (§2.3 as amended by §16.2): the caller's
    /// handle presents the creation identity it was bound to AND the sequence
    /// it last observed, and the full replacement record is installed
    /// atomically only if both still match.
    ///
    /// Typed failures: no record ⇒ [`BranchRefError::BranchRemoved`]; identity
    /// mismatch (same name, different generation) ⇒
    /// [`BranchRefError::BranchGenerationMismatch`]; sequence mismatch ⇒
    /// [`BranchRefError::StaleSeq`]. `heads` may cover any subset of the
    /// record's shards (unmentioned shards keep their head); every
    /// `fork_anchor` is preserved VERBATIM — the anchor is immutable for the
    /// branch's life (§16.3). Returns the new sequence.
    ///
    /// The §16.2 CAS predicate on its own: does `name` still exist, in the
    /// generation the caller was bound to, at the sequence it last observed?
    /// Returns the record on success. Shared by [`Self::advance`] and by
    /// `commit_branch`'s documented no-op path, so a stale handle is refused
    /// identically whether or not it has anything to commit.
    ///
    /// Identity is checked before sequence: a recreated branch legitimately
    /// restarts at seq 0, so a seq comparison alone would let a stale handle
    /// from the previous generation pass (§16.2's ABA).
    pub(crate) fn cas_check(
        &self,
        name: &str,
        expected_created: Timestamp,
        expected_seq: u64,
    ) -> Result<&BranchRefRecord, BranchRefError> {
        let Some(record) = self.records.get(name) else {
            return Err(BranchRefError::BranchRemoved(name.to_owned()));
        };
        if record.created != expected_created {
            return Err(BranchRefError::BranchGenerationMismatch {
                name: name.to_owned(),
                expected_created,
                found_created: record.created,
            });
        }
        if record.seq != expected_seq {
            return Err(BranchRefError::StaleSeq {
                name: name.to_owned(),
                expected: expected_seq,
                found: record.seq,
            });
        }
        Ok(record)
    }

    /// The blessed caller is `commit_branch` (§5), which orders this install
    /// strictly after the dirty-directory barrier. Crate-private (§2.3): a
    /// direct external caller could install a record naming heads whose
    /// nodes' directory entries were never fenced.
    pub(crate) fn advance(
        &mut self,
        name: &str,
        expected_created: Timestamp,
        expected_seq: u64,
        heads: &[(ShardId, Hash)],
        parents: Vec<Hash>,
        timestamp: Timestamp,
    ) -> Result<u64, BranchRefError> {
        self.advance_with_source(
            name,
            expected_created,
            expected_seq,
            heads,
            parents,
            timestamp,
        )
        .map_err(RecordOperationError::into_public)
    }

    fn advance_with_source(
        &mut self,
        name: &str,
        expected_created: Timestamp,
        expected_seq: u64,
        heads: &[(ShardId, Hash)],
        parents: Vec<Hash>,
        timestamp: Timestamp,
    ) -> Result<u64, RecordOperationError<BranchRefError>> {
        let Some(record) = self.records.get(name) else {
            return Err(BranchRefError::BranchRemoved(name.to_owned()).into());
        };
        let mut replacement = record.clone();
        replacement.seq = expected_seq + 1;
        replacement.timestamp = timestamp;
        replacement.parents = parents;
        for &(shard_id, head) in heads {
            let Some(shard) = replacement
                .shards
                .iter_mut()
                .find(|shard| shard.shard_id == shard_id)
            else {
                return Err(BranchRefError::UnknownShard {
                    name: name.to_owned(),
                    shard_id,
                }
                .into());
            };
            // fork_anchor deliberately untouched (§16.3).
            shard.head = head;
        }

        // One write_atomic install covering all shards: the rename swaps the
        // old pins for the new in a single indivisible step. The install
        // error SPLIT is load-bearing (adversarial-review blocker): after a
        // post-rename fsync failure the target file already holds the
        // replacement — a map keeping the OLD record would make
        // `protected_roots()` under-pin the new heads, and a prune could then
        // reclaim nodes the on-disk record references (resurrected as live by
        // any reopen). Under-pin is corruption; over-pin is a leak — so the
        // map adopts the replacement on BOTH success and unfenced-install,
        // and only stays put when the rename provably never happened. The
        // unfenced path still returns the error; the caller's handle then
        // holds a stale seq, and the documented recovery is a rebind via
        // `open_branch` (the buffered batch re-applies idempotently — history
        // independence makes the re-commit converge on the same root).
        let installed =
            self.backend
                .cas_replace_install(name, expected_created, expected_seq, &replacement)?;
        let seq = installed.seq;
        self.records.insert(name.to_owned(), installed.clone());
        entry_fence_with_source(self.backend.as_mut(), EntryFence::Present(&installed))
            .map_err(RecordOperationError::Fence)?;
        Ok(seq)
    }

    /// Deletes the durable record: unlink + parent-dir fsync (§5's
    /// `remove_branch` substrate). Unknown names return `Ok(None)`. Node
    /// reclamation is prune's job, and any in-memory pins held via open
    /// handles release normally on guard drop.
    pub fn remove(&mut self, name: &str) -> Result<Option<BranchRefRecord>, BranchRefError> {
        self.remove_with_source(name)
            .map_err(RecordOperationError::into_public)
    }

    fn remove_with_source(
        &mut self,
        name: &str,
    ) -> Result<Option<BranchRefRecord>, RecordOperationError<BranchRefError>> {
        if !self.records.contains_key(name) {
            return Ok(None);
        }
        self.backend.unlink(name)?;
        entry_fence_with_source(self.backend.as_mut(), EntryFence::Absent(name))
            .map_err(RecordOperationError::Fence)?;
        Ok(self.records.remove(name))
    }
}

#[cfg(test)]
#[path = "refstore_kind_tests.rs"]
mod kind_tests;
#[cfg(test)]
#[path = "refstore_tests.rs"]
mod tests;