haematite 0.6.1

//! §6(b) prune-vs-commit interleaving at the orchestration layer, including
//! the §1 dedup-collision case (LEDGER A1 stage 4).
//!
//! The borrow checker serialises every prune entirely-before or
//! entirely-after each `commit_branch` call (`&mut S` vs `&S` on one store),
//! so these tests drive BOTH legal orders around the §1 hazard: a node
//! byte-identical to one reachable ONLY from the snapshot being pruned.
//!
//! * **Prune entirely-before**: prune reclaims the collision node, then a
//!   commit re-derives the identical hash — its own `put` recreates the
//!   node, and the head reads back intact.
//! * **Prune entirely-after**: the commit registered (volatile) or durably
//!   recorded (durable) its head before returning, so a prune of the
//!   snapshot whose reachable set is byte-identical to the fresh head
//!   reclaims NOTHING. These are the falsifiable halves of §14.3(a) as seen
//!   from prune's side: a commit that forgot to register (or record) its
//!   advanced root loses the collision node out from under a live handle.

use std::fmt;

use super::commit::{BranchCommitError, CommitDurability, CommitRequest, commit_branch};
use super::fork::fork_registered;
use super::handle::{BranchHandle, DEFAULT_SHARD_ID};
use super::lifecycle::create_branch;
use super::prune::{PruneError, prune};
use super::refstore::{BranchRefError, BranchRefStore};
use super::registry::BranchRegistry;
use super::snapshot::{SnapshotError, SnapshotRegistry};
use crate::store::MemoryStore;
use crate::tree::{Hash, LeafNode, Node, NodeError, TreeError};

/// Aggregate error for these tests, per the house Result-returning idiom.
#[derive(Debug)]
enum TestError {
    Commit(BranchCommitError),
    Refs(BranchRefError),
    Prune(PruneError),
    Snapshot(SnapshotError),
    Branch(super::handle::BranchError),
    Tree(TreeError),
    Node(NodeError),
    Io(std::io::Error),
}

impl fmt::Display for TestError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Commit(error) => write!(formatter, "commit error: {error}"),
            Self::Refs(error) => write!(formatter, "ref store error: {error}"),
            Self::Prune(error) => write!(formatter, "prune error: {error}"),
            Self::Snapshot(error) => write!(formatter, "snapshot error: {error}"),
            Self::Branch(error) => write!(formatter, "branch error: {error}"),
            Self::Tree(error) => write!(formatter, "tree error: {error}"),
            Self::Node(error) => write!(formatter, "node error: {error}"),
            Self::Io(error) => write!(formatter, "I/O error: {error}"),
        }
    }
}

impl std::error::Error for TestError {}

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

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

impl From<PruneError> for TestError {
    fn from(error: PruneError) -> Self {
        Self::Prune(error)
    }
}

impl From<SnapshotError> for TestError {
    fn from(error: SnapshotError) -> Self {
        Self::Snapshot(error)
    }
}

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

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

impl From<NodeError> for TestError {
    fn from(error: NodeError) -> Self {
        Self::Node(error)
    }
}

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

fn empty_root(store: &mut MemoryStore) -> Result<Hash, TestError> {
    let leaf = LeafNode::new(Vec::new())?;
    Ok(store.put(&Node::Leaf(leaf)))
}

fn volatile() -> CommitRequest<'static> {
    CommitRequest {
        durability: CommitDurability::Volatile,
        extra_parents: &[],
        timestamp: 1,
    }
}

/// Materialise `entries` from the empty root through a scratch branch and
/// return the head. The scratch handle drops before returning, so ONLY a
/// snapshot the caller names afterwards pins the result — the "reachable
/// solely from the pruned snapshot" precondition of the §1 collision.
fn scratch_head(store: &mut MemoryStore, entries: &[(&[u8], &[u8])]) -> Result<Hash, TestError> {
    let registry = BranchRegistry::new();
    let scratch = fork_registered(empty_root(store)?, &registry);
    for (key, value) in entries {
        scratch.put(DEFAULT_SHARD_ID, key, value)?;
    }
    commit_branch(&scratch, store, &registry, volatile())?;
    Ok(scratch.current_root())
}

/// Buffer `entries` on `branch` (shared write half of the collision builds).
fn buffer_entries(branch: &BranchHandle, entries: &[(&[u8], &[u8])]) -> Result<(), TestError> {
    for (key, value) in entries {
        branch.put(DEFAULT_SHARD_ID, key, value)?;
    }
    Ok(())
}

const COLLISION: &[(&[u8], &[u8])] = &[(b"dup", b"payload")];

#[test]
fn prune_before_commit_dedup_casualty_is_recreated_by_the_commit() -> Result<(), TestError> {
    // Entirely-before: the snapshot's node is reclaimed FIRST, then a commit
    // producing the byte-identical node runs. `write_node`'s path_exists
    // short-circuit sees nothing (the file is gone), rewrites it, and the
    // head reads back intact — the §6(b) "entirely-before" disposition.
    let mut store = MemoryStore::new();
    let dir = tempfile::tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();
    let registry = BranchRegistry::new();

    let doomed = scratch_head(&mut store, COLLISION)?;
    snapshots.name_at("old", doomed, 5)?;

    let report = prune(&store, &registry, &refs, &mut snapshots, "old")?;
    assert_eq!(report.node_count, 1, "the collision node must be reclaimed");
    assert!(store.get(&doomed).is_none());

    // The interleaved commit: same anchor, same final map ⇒ same head hash
    // (history independence), and its own put recreates the missing node.
    let branch = fork_registered(empty_root(&mut store)?, &registry);
    buffer_entries(&branch, COLLISION)?;
    commit_branch(&branch, &mut store, &registry, volatile())?;

    assert_eq!(
        branch.current_root(),
        doomed,
        "the commit must re-derive the identical (dedup-colliding) head"
    );
    assert!(
        store.get(&doomed).is_some(),
        "the commit's put must have recreated the reclaimed node"
    );
    assert_eq!(
        branch.get(DEFAULT_SHARD_ID, &store, b"dup")?,
        Some(b"payload".to_vec())
    );
    Ok(())
}

#[test]
fn prune_after_volatile_commit_spares_the_identical_live_head() -> Result<(), TestError> {
    // Entirely-after, volatile: the fresh head is byte-identical to the
    // pruned snapshot's root, and its ONLY protection is the registry pin
    // commit_branch took in step 5. A commit that forgot the register (or a
    // guard that retargeted the wrong pin) fails here with the node deleted
    // out from under the live handle.
    let mut store = MemoryStore::new();
    let dir = tempfile::tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();
    let registry = BranchRegistry::new();

    let doomed = scratch_head(&mut store, COLLISION)?;
    snapshots.name_at("old", doomed, 5)?;

    let branch = fork_registered(empty_root(&mut store)?, &registry);
    buffer_entries(&branch, COLLISION)?;
    commit_branch(&branch, &mut store, &registry, volatile())?;
    assert_eq!(branch.current_root(), doomed);

    let report = prune(&store, &registry, &refs, &mut snapshots, "old")?;
    assert_eq!(
        report.node_count, 0,
        "the live registry pin must protect the byte-identical head"
    );
    assert!(store.get(&doomed).is_some());
    assert_eq!(
        branch.get(DEFAULT_SHARD_ID, &store, b"dup")?,
        Some(b"payload".to_vec())
    );
    Ok(())
}

#[test]
fn prune_after_durable_commit_spares_the_identical_recorded_head() -> Result<(), TestError> {
    // Entirely-after, durable, across a simulated restart: after the commit
    // the handle and registry are DROPPED (every in-memory pin gone), so the
    // only thing standing between the prune and the byte-identical head is
    // the HBR1 record commit_branch installed in step 4.
    let mut store = MemoryStore::new();
    let dir = tempfile::tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();

    let doomed = scratch_head(&mut store, COLLISION)?;
    snapshots.name_at("old", doomed, 5)?;

    {
        let registry = BranchRegistry::new();
        let anchor = empty_root(&mut store)?;
        let branch = create_branch(
            "layer",
            [(DEFAULT_SHARD_ID, anchor)],
            &mut refs,
            &registry,
            10,
        )?;
        buffer_entries(&branch, COLLISION)?;
        commit_branch(
            &branch,
            &mut store,
            &registry,
            CommitRequest {
                durability: CommitDurability::Durable { refs: &mut refs },
                extra_parents: &[],
                timestamp: 20,
            },
        )?;
        assert_eq!(branch.current_root(), doomed);
        // Simulated restart: handle and registry drop; the record remains.
    }

    let refs = BranchRefStore::open(dir.path())?;
    let registry = BranchRegistry::new();
    let report = prune(&store, &registry, &refs, &mut snapshots, "old")?;
    assert_eq!(
        report.node_count, 0,
        "the durable record must protect the byte-identical head across restart"
    );
    assert!(store.get(&doomed).is_some());
    Ok(())
}