haematite 0.4.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
//! (trailing bytes rejected)
//! ```
//!
//! 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::fs;
use std::path::{Path, PathBuf};

use super::persist::{
    CodecError, InstallError, NoclobberOutcome, sync_parent_dir, write_atomic,
    write_atomic_noclobber,
};
use super::snapshot::Timestamp;
use crate::ids::ShardId;
use crate::tree::Hash;

#[path = "refstore_codec.rs"]
mod codec;
use codec::{REF_EXTENSION, corrupt_in_file, decode_record, encode_record, ref_file_name};

/// Temp-file naming pinned in `persist.rs` (`.branch-*.tmp`); the sweep in
/// [`BranchRefStore::open`] matches exactly this pattern (§16.3).
const TEMP_PREFIX: &str = ".branch-";
const TEMP_SUFFIX: &str = ".tmp";

/// One shard's entry in a branch ref record: where the branch diverged and
/// where its committed head currently is.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BranchShardRef {
    /// Shard this entry pins roots for.
    pub shard_id: ShardId,
    /// Divergence point — the three-way-merge ancestor. Immutable for the
    /// branch's life; [`BranchRefStore::advance`] preserves it verbatim (§16.3).
    pub fork_anchor: Hash,
    /// Current committed root for this shard.
    pub head: Hash,
}

/// The durable truth for one named branch: creation identity, commit
/// sequence, per-shard anchors and heads, and the head commit's parent roots.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BranchRefRecord {
    /// Full branch name (the on-disk file name is only its hash).
    pub name: String,
    /// Creation identity (§16.2): the timestamp captured at create, immutable
    /// for the record's life. [`BranchRefStore::advance`] compares it so a
    /// stale handle from a removed-and-recreated branch of the same name can
    /// never hijack the new generation.
    pub created: Timestamp,
    /// Durable commit sequence; `0` = created, never committed.
    pub seq: u64,
    /// When this record was installed (ns since epoch, caller-supplied).
    pub timestamp: Timestamp,
    /// Per-shard anchor/head pairs, strictly ascending by shard id.
    pub shards: Vec<BranchShardRef>,
    /// Tree roots of this head commit's parents. By convention `parents[0]`
    /// is the previous durable record's head; extras are converge links.
    pub parents: Vec<Hash>,
}

/// 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 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::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 {
    dir: PathBuf,
    records: BTreeMap<String, BranchRefRecord>,
}

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 dir = dir.as_ref().to_path_buf();
        let created = !dir.exists();
        fs::create_dir_all(&dir)?;
        if created {
            // Fence the ref directory's OWN entry in its parent: every record
            // install fsyncs `dir` itself, but a freshly created `dir` whose
            // parent entry was never synced can vanish wholesale on power
            // loss — records, pins, and all. Immediate parent only, same
            // honest scope as the data-dir fence in db/startup.rs.
            if let Some(parent) = dir.parent() {
                let parent = if parent.as_os_str().is_empty() {
                    Path::new(".")
                } else {
                    parent
                };
                sync_parent_dir(parent)?;
            }
        }

        let mut records = BTreeMap::new();
        for entry in fs::read_dir(&dir)? {
            let entry = entry?;
            let path = entry.path();
            let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
                continue;
            };
            if file_name.starts_with(TEMP_PREFIX) && file_name.ends_with(TEMP_SUFFIX) {
                // Orphaned temp file from a crash mid-write_atomic: never
                // referenced by anything, safe to unlink (§2.1).
                fs::remove_file(&path)?;
                continue;
            }
            if path.extension().and_then(|ext| ext.to_str()) != Some(REF_EXTENSION) {
                continue;
            }
            let record =
                decode_record(&fs::read(&path)?).map_err(|error| corrupt_in_file(&path, &error))?;
            if ref_file_name(&record.name) != file_name {
                return Err(BranchRefError::Corrupt(format!(
                    "ref file {} holds a record for branch {:?}, whose name hashes elsewhere",
                    path.display(),
                    record.name,
                )));
            }
            records.insert(record.name.clone(), record);
        }

        Ok(Self { dir, 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, mut record: BranchRefRecord) -> Result<(), BranchRefError> {
        if self.records.contains_key(&record.name) {
            return Err(BranchRefError::DuplicateBranch(record.name));
        }
        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,
            });
        }

        let path = self.ref_path(&record.name);
        let outcome = match write_atomic_noclobber(&path, &encode_record(&record)) {
            Ok(outcome) => outcome,
            Err(InstallError::NotInstalled(error)) => return Err(error.into()),
            Err(InstallError::InstalledUnfenced(error)) => {
                // The rename landed: the target file already holds this
                // record, so the map must adopt it — a mirror keeping no
                // record would under-pin the new anchors/heads and let prune
                // reclaim nodes a cold reopen would still resolve. The error
                // still surfaces (directory-entry durability unconfirmed).
                self.records.insert(record.name.clone(), record);
                return Err(error.into());
            }
        };
        match outcome {
            NoclobberOutcome::Installed => {
                self.records.insert(record.name.clone(), record);
                Ok(())
            }
            NoclobberOutcome::TargetExists => {
                let existing = decode_record(&fs::read(&path)?)
                    .map_err(|error| corrupt_in_file(&path, &error))?;
                if existing.name == record.name {
                    Err(BranchRefError::DuplicateBranch(record.name))
                } else {
                    Err(BranchRefError::NameHashCollision {
                        requested: record.name,
                        existing: existing.name,
                    })
                }
            }
        }
    }

    /// 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> {
        let record = self.cas_check(name, expected_created, expected_seq)?;
        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,
                });
            };
            // 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).
        match write_atomic(&self.ref_path(name), &encode_record(&replacement)) {
            Ok(()) => {
                let seq = replacement.seq;
                self.records.insert(name.to_owned(), replacement);
                Ok(seq)
            }
            Err(InstallError::NotInstalled(error)) => Err(error.into()),
            Err(InstallError::InstalledUnfenced(error)) => {
                self.records.insert(name.to_owned(), replacement);
                Err(error.into())
            }
        }
    }

    /// 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> {
        if !self.records.contains_key(name) {
            return Ok(None);
        }
        match fs::remove_file(self.ref_path(name)) {
            Ok(()) => {}
            // Already absent on disk: the unlink is idempotent.
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(BranchRefError::Io(error)),
        }
        sync_parent_dir(&self.dir)?;
        Ok(self.records.remove(name))
    }

    fn ref_path(&self, name: &str) -> PathBuf {
        self.dir.join(ref_file_name(name))
    }
}

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