haematite 0.6.1

//! LEDGER A1: the branch lifecycle — fork, create, reopen, remove (§5's
//! creation block, §9, §16.4 Q3).
//!
//! Every constructor here returns a registered handle whose pins follow the
//! §14.1 refcount-2 pattern: each shard's fork anchor is pinned once in the
//! anchor role (held until guard drop — `merge` needs the divergence point
//! for the handle's whole life) and once in the current-head role (the pin
//! `commit_branch` retargets on every advance). Named constructors
//! additionally bind the handle to its durable HBR1 record so durable
//! commits are CAS-guarded by creation identity and sequence (§16.2).

use std::collections::HashSet;

use crate::tree::Hash;

use super::commit::BranchCommitError;
use super::handle::{BranchError, BranchHandle, RefBinding, ShardId};
use super::policy::BranchKind;
use super::refstore::{BranchRefError, BranchRefRecord, BranchRefStore, BranchShardRef};
use super::registry::BranchRegistry;
use super::snapshot::{SnapshotRegistry, Timestamp};

/// Anonymous fork from another branch's *committed* heads (all shards).
///
/// The child anchors at the parent's committed roots at call time; the
/// parent's uncommitted buffer is invisible — buffered intent is not a tree
/// and has no hash to anchor to (§9, stated not accidental). The parent MUST
/// itself be registered (typed [`BranchError::UnregisteredParent`] otherwise
/// — an unregistered parent's heads carry no prune pin, the §16.4 Q3 hazard);
/// the child comes back registered.
pub fn fork_from(
    parent: &BranchHandle,
    registry: &BranchRegistry,
) -> Result<BranchHandle, BranchError> {
    // An unregistered parent's heads carry no prune pin, so anchoring a child
    // to them is the §16.4 Q3 unprotected-anchor hazard by another door —
    // refused here exactly as fork_at refuses it (review finding: this was
    // documentation-only).
    if parent.registry_guard().is_none() {
        return Err(BranchError::UnregisteredParent);
    }
    // The parent's shard-state locks are held from reading its committed
    // heads THROUGH registering the child's pins (lock order §16.1: shard
    // states ascending, then the registry lock — binding tier unused).
    // Released between the read and the register, a concurrent parent commit
    // could advance and deregister the very head the child just chose as its
    // anchor, handing prune an unpinned root in the gap.
    let states = parent.lock_states_ascending()?;
    let heads: Vec<(ShardId, Hash)> = states
        .iter()
        .map(|(shard_id, state)| (*shard_id, state.current_root))
        .collect();
    let child = BranchHandle::from_shard_roots(heads)?;
    // Refcount-2 anchor registration (§14.1): anchor role + head role.
    let guard = registry.register_roots(child.fork_points().flat_map(|anchor| [anchor, anchor]));
    drop(states);
    Ok(child.with_registry_guard(guard))
}

/// Fork at an arbitrary committed root hash, refusing unprotected anchors
/// (§16.4 Q3, superseding §9's documentation-only stance).
///
/// The anchor must be protected by at least one of: a live branch pin in
/// `registry`, a named branch record in `refs` (anchor or head), or a named
/// snapshot in `snapshots`. Anything else is inherently racy with prune —
/// documentation-only would convert a caller bug into a delayed
/// `MissingNode` miles from the fork site — so the engine refuses with
/// [`BranchCommitError::UnprotectedAnchor`], whose message names the recipe:
/// name a snapshot at the root first, then fork. The membership check and
/// the child's registration happen under the registry's one counts lock, so
/// an anchor protected only by another live handle cannot lose that pin
/// between check and register.
pub fn fork_at(
    root: Hash,
    registry: &BranchRegistry,
    refs: &BranchRefStore,
    snapshots: &SnapshotRegistry,
) -> Result<BranchHandle, BranchCommitError> {
    // Both durable membership sets are materialised BEFORE taking the
    // registry lock: the predicate below runs with the counts guard held and
    // must stay O(1) and lock-free.
    let durable = refs.protected_roots();
    let named: HashSet<Hash> = snapshots
        .list_snapshots()
        .into_iter()
        .map(|(_name, root_hash, _timestamp)| root_hash)
        .collect();

    let guard = registry
        .register_roots_protected([root, root], |anchor| {
            durable.contains(anchor) || named.contains(anchor)
        })
        .map_err(|root| BranchCommitError::UnprotectedAnchor { root })?;
    Ok(BranchHandle::new(root).with_registry_guard(guard))
}

/// Durable named branch: pins the anchors in `registry`, then installs the
/// HBR1 record with heads = anchors, seq = 0 (§5, norn R6).
///
/// The no-clobber install makes reservation and creation one atomic step
/// (typed [`BranchRefError::DuplicateBranch`] via `Ref`); `timestamp` becomes
/// the branch's creation identity (§16.2), immutable for the record's life.
/// Returns a bound, registered handle.
pub fn create_branch<I>(
    name: &str,
    roots: I,
    refs: &mut BranchRefStore,
    registry: &BranchRegistry,
    timestamp: Timestamp,
) -> Result<BranchHandle, BranchCommitError>
where
    I: IntoIterator<Item = (ShardId, Hash)>,
{
    create_branch_record(
        name,
        roots,
        BranchKind::Work,
        None,
        refs,
        registry,
        timestamp,
    )
}

pub(crate) fn create_branch_record<I>(
    name: &str,
    roots: I,
    kind: BranchKind,
    namespace_lineage: Option<String>,
    refs: &mut BranchRefStore,
    registry: &BranchRegistry,
    timestamp: Timestamp,
) -> Result<BranchHandle, BranchCommitError>
where
    I: IntoIterator<Item = (ShardId, Hash)>,
{
    let roots: Vec<(ShardId, Hash)> = roots.into_iter().collect();
    let handle = BranchHandle::from_shard_roots(roots.iter().copied())?;

    // Pins strictly BEFORE the durable install (parent-first reservation
    // ordering, §0 anchor 2): at every instant the roots are protected by
    // the in-memory pins, the installed record, or both — never neither. If
    // the install fails (duplicate, collision, I/O), the guard drops right
    // here and the pins release: leak-free refusal, nothing durable left.
    let guard = registry.register_roots(handle.fork_points().flat_map(|anchor| [anchor, anchor]));

    refs.create(BranchRefRecord {
        name: name.to_owned(),
        created: timestamp,
        kind,
        namespace_lineage,
        seq: 0,
        timestamp,
        shards: roots
            .iter()
            .map(|&(shard_id, root)| BranchShardRef {
                shard_id,
                fork_anchor: root,
                head: root,
            })
            .collect(),
        parents: Vec::new(),
    })?;

    Ok(handle.with_registry_guard(guard).with_binding(RefBinding {
        name: name.to_owned(),
        created: timestamp,
        seq: 0,
    }))
}

/// Reopen a named branch after restart (or by name) at its recorded state.
///
/// The handle sits at the recorded committed heads, `fork_point` = the
/// recorded fork anchor (the merge ancestor, preserved across any number of
/// advances), and the binding carries the recorded seq so a forgotten stale
/// handle gets `StaleSeq` — never a silent clobber (§9).
///
/// A name with no record yields [`BranchRefError::BranchRemoved`] (via
/// `Ref`). The recorded roots are durably pinned by the record itself, so
/// registering the in-memory pins here has no unprotected window.
pub fn open_branch(
    name: &str,
    refs: &BranchRefStore,
    registry: &BranchRegistry,
) -> Result<BranchHandle, BranchCommitError> {
    let Some(record) = refs.get(name) else {
        return Err(BranchCommitError::Ref(BranchRefError::BranchRemoved(
            name.to_owned(),
        )));
    };

    let handle = BranchHandle::from_anchor_head_pairs(
        record
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.fork_anchor, shard.head)),
    )?;
    // The two-role pin pattern (§14.1) generalised past the first advance:
    // anchor role + current-head role per shard. On a never-committed branch
    // they coincide (refcount 2 on one hash), exactly the create/fork shape.
    let guard = registry.register_roots(
        record
            .shards
            .iter()
            .flat_map(|shard| [shard.fork_anchor, shard.head]),
    );

    Ok(handle.with_registry_guard(guard).with_binding(RefBinding {
        name: record.name.clone(),
        created: record.created,
        seq: record.seq,
    }))
}

/// Delete a named branch's durable record (unlink + parent-dir fsync).
///
/// Returns the removed record, or `None` for an unknown name. In-memory pins
/// held via any open handle release normally on guard drop; node reclamation
/// is prune's job. To keep the branch's final state queryable past removal
/// (tharsis lease expiry), name its head as a snapshot FIRST, then remove —
/// a crash between the two leaves both alive, which is leak-safe (§11.3).
pub fn remove_branch(
    name: &str,
    refs: &mut BranchRefStore,
) -> Result<Option<BranchRefRecord>, BranchRefError> {
    refs.remove(name)
}

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