haematite 0.6.1

//! Unit tests for snapshot pruning. Split out of `prune.rs` via `#[path]` so
//! that file stays within the branch module's 500-line cap now that prune
//! consumes the durable branch-ref denominator (LEDGER A1 §6(c)).

use super::{PruneError, PruneReport, prune};
use crate::branch::{
    BranchRefError, BranchRefRecord, BranchRefStore, BranchRegistry, BranchShardRef, CommitLog,
    SnapshotError, SnapshotRegistry,
};
use crate::store::MemoryStore;
use crate::tree::{Hash, InternalNode, LeafNode, Node, NodeError};

#[derive(Debug)]
enum TestError {
    Node(NodeError),
    Snapshot(SnapshotError),
    Prune(PruneError),
    Refs(BranchRefError),
}

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

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

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

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

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

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

fn tempdir() -> Result<tempfile::TempDir, TestError> {
    tempfile::tempdir().map_err(|error| TestError::Refs(BranchRefError::Io(error)))
}

/// A ref store with no records: prune's durable pin set is empty, so these
/// tests exercise exactly the pre-A1 registry+snapshot semantics.
fn empty_refs(dir: &tempfile::TempDir) -> Result<BranchRefStore, TestError> {
    Ok(BranchRefStore::open(dir.path())?)
}

fn leaf(key: &[u8], value: &[u8]) -> Result<Node, NodeError> {
    LeafNode::new(vec![(key.to_vec(), value.to_vec())]).map(Node::Leaf)
}

fn internal(children: Vec<(&[u8], Hash)>) -> Result<Node, NodeError> {
    InternalNode::new(
        children
            .into_iter()
            .map(|(key, hash)| (key.to_vec(), hash))
            .collect(),
    )
    .map(Node::Internal)
}

#[test]
fn prune_removes_snapshot_registry_entry_without_touching_commit_log() -> Result<(), TestError> {
    let dir = tempdir()?;
    let refs = empty_refs(&dir)?;
    let mut store = MemoryStore::new();
    let root_node = leaf(b"root", b"value")?;
    let root_hash = store.put(&root_node);
    let expected_bytes = root_node.serialise().len();
    let branches = BranchRegistry::new();
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name_at("old", root_hash, 10)?;
    let mut log = CommitLog::new();
    log.append(root_hash, 20)?;

    let report = prune(&store, &branches, &refs, &mut snapshots, "old")?;

    assert_eq!(snapshots.get("old"), None);
    assert!(log.list().iter().any(|entry| entry.root_hash == root_hash));
    assert_eq!(
        report,
        PruneReport {
            node_count: 1,
            bytes_reclaimed: expected_bytes,
        }
    );
    Ok(())
}

#[test]
fn prune_unknown_snapshot_returns_error() -> Result<(), TestError> {
    let dir = tempdir()?;
    let refs = empty_refs(&dir)?;
    let store = MemoryStore::new();
    let branches = BranchRegistry::new();
    let mut snapshots = SnapshotRegistry::new();

    let result = prune(&store, &branches, &refs, &mut snapshots, "missing");

    assert!(matches!(
        result,
        Err(PruneError::UnknownSnapshot { name }) if name == "missing"
    ));
    Ok(())
}

#[test]
fn prune_deletes_only_nodes_unreachable_from_live_roots() -> Result<(), TestError> {
    let dir = tempdir()?;
    let refs = empty_refs(&dir)?;
    let mut store = MemoryStore::new();
    let shared = leaf(b"shared", b"kept by every root")?;
    let pruned_only = leaf(b"pruned", b"delete me")?;
    let branch_only = leaf(b"branch", b"active branch keeps me")?;
    let snapshot_only = leaf(b"snapshot", b"named snapshot keeps me")?;

    let shared_hash = store.put(&shared);
    let pruned_only_hash = store.put(&pruned_only);
    let branch_only_hash = store.put(&branch_only);
    let snapshot_only_hash = store.put(&snapshot_only);

    let pruned_root = internal(vec![(b"a", shared_hash), (b"b", pruned_only_hash)])?;
    let branch_root = internal(vec![(b"a", shared_hash), (b"c", branch_only_hash)])?;
    let snapshot_root = internal(vec![(b"a", shared_hash), (b"d", snapshot_only_hash)])?;
    let pruned_root_hash = store.put(&pruned_root);
    let branch_root_hash = store.put(&branch_root);
    let snapshot_root_hash = store.put(&snapshot_root);

    let branches = BranchRegistry::new();
    branches.register(branch_root_hash);
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name_at("old", pruned_root_hash, 10)?;
    snapshots.name_at("keep", snapshot_root_hash, 20)?;

    let expected_bytes = pruned_root.serialise().len() + pruned_only.serialise().len();
    let report = prune(&store, &branches, &refs, &mut snapshots, "old")?;

    assert_eq!(
        report,
        PruneReport {
            node_count: 2,
            bytes_reclaimed: expected_bytes,
        }
    );
    assert_eq!(snapshots.get("old"), None);
    assert_eq!(snapshots.get("keep"), Some(snapshot_root_hash));
    assert_eq!(store.get(&pruned_root_hash), None);
    assert_eq!(store.get(&pruned_only_hash), None);
    assert_eq!(store.get(&shared_hash), Some(std::sync::Arc::new(shared)));
    assert_eq!(
        store.get(&branch_root_hash),
        Some(std::sync::Arc::new(branch_root))
    );
    assert_eq!(
        store.get(&branch_only_hash),
        Some(std::sync::Arc::new(branch_only))
    );
    assert_eq!(
        store.get(&snapshot_root_hash),
        Some(std::sync::Arc::new(snapshot_root))
    );
    assert_eq!(
        store.get(&snapshot_only_hash),
        Some(std::sync::Arc::new(snapshot_only))
    );
    Ok(())
}

#[test]
fn prune_retains_snapshot_when_a_referenced_node_is_missing() -> Result<(), TestError> {
    // A graph-walk failure must leave the snapshot named and the operation
    // retryable: the reclaim set is collected BEFORE the registry entry is
    // removed. Falsifiable — removing the snapshot before the walk (the
    // prior ordering) would drop "broken" even though prune returns Err,
    // permanently orphaning its nodes.
    let dir = tempdir()?;
    let refs = empty_refs(&dir)?;
    let mut store = MemoryStore::new();
    let present = leaf(b"present", b"here")?;
    let present_hash = store.put(&present);
    let missing_child = Hash::from_bytes([0xEE; 32]);
    let root_node = internal(vec![(b"a", present_hash), (b"z", missing_child)])?;
    let root_hash = store.put(&root_node);
    let branches = BranchRegistry::new();
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name_at("broken", root_hash, 10)?;

    let result = prune(&store, &branches, &refs, &mut snapshots, "broken");

    assert!(matches!(
        result,
        Err(PruneError::MissingNode { hash }) if hash == missing_child
    ));
    // Still named (retryable) and nothing it referenced was deleted.
    assert_eq!(snapshots.get("broken"), Some(root_hash));
    assert_eq!(store.get(&root_hash), Some(std::sync::Arc::new(root_node)));
    assert_eq!(store.get(&present_hash), Some(std::sync::Arc::new(present)));
    Ok(())
}

#[test]
fn prune_last_snapshot_reclaims_all_its_nodes() -> Result<(), TestError> {
    // Pruning the only snapshot with no active branches: the live set is
    // empty, so every node reachable from the pruned root is reclaimed.
    let dir = tempdir()?;
    let refs = empty_refs(&dir)?;
    let mut store = MemoryStore::new();
    let child_a = leaf(b"a", b"1")?;
    let child_b = leaf(b"b", b"2")?;
    let a_hash = store.put(&child_a);
    let b_hash = store.put(&child_b);
    let root_node = internal(vec![(b"a", a_hash), (b"b", b_hash)])?;
    let root_hash = store.put(&root_node);
    let branches = BranchRegistry::new();
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name_at("solo", root_hash, 10)?;

    let expected_bytes =
        root_node.serialise().len() + child_a.serialise().len() + child_b.serialise().len();
    let report = prune(&store, &branches, &refs, &mut snapshots, "solo")?;

    assert_eq!(
        report,
        PruneReport {
            node_count: 3,
            bytes_reclaimed: expected_bytes,
        }
    );
    assert_eq!(snapshots.get("solo"), None);
    assert_eq!(store.get(&root_hash), None);
    assert_eq!(store.get(&a_hash), None);
    assert_eq!(store.get(&b_hash), None);
    Ok(())
}

#[test]
fn prune_protects_ref_store_anchors_and_heads_across_a_restart() -> Result<(), TestError> {
    // §6(c): named branches are protected by the DURABLE pin set, not by
    // in-memory registration discipline. A branch is created and advanced,
    // then the process "dies" (ref store dropped, registry lost). After the
    // simulated restart — reopened ref store, FRESH empty registry — pruning
    // an overlapping snapshot must spare both the branch's fork anchor and
    // its advanced head, with zero startup re-registration ceremony.
    let dir = tempdir()?;
    let mut store = MemoryStore::new();
    let shared = leaf(b"shared", b"reachable from every root")?;
    let anchor_only = leaf(b"anchor", b"merge ancestor keeps me")?;
    let head_only = leaf(b"head", b"advanced head keeps me")?;
    let pruned_only = leaf(b"pruned", b"delete me")?;
    let shared_hash = store.put(&shared);
    let anchor_only_hash = store.put(&anchor_only);
    let head_only_hash = store.put(&head_only);
    let pruned_only_hash = store.put(&pruned_only);

    let anchor_root = internal(vec![(b"a", shared_hash), (b"b", anchor_only_hash)])?;
    let head_root = internal(vec![(b"a", shared_hash), (b"c", head_only_hash)])?;
    let pruned_root = internal(vec![(b"a", shared_hash), (b"d", pruned_only_hash)])?;
    let anchor_root_hash = store.put(&anchor_root);
    let head_root_hash = store.put(&head_root);
    let pruned_root_hash = store.put(&pruned_root);

    {
        let mut refs = BranchRefStore::open(dir.path())?;
        refs.create(BranchRefRecord {
            name: "sessions/alpha".to_owned(),
            created: 1_000,
            kind: crate::branch::BranchKind::Work,
            namespace_lineage: None,
            seq: 0,
            timestamp: 1_000,
            shards: vec![BranchShardRef {
                shard_id: 0,
                fork_anchor: anchor_root_hash,
                head: anchor_root_hash,
            }],
            parents: Vec::new(),
        })?;
        // One durable advance: the head moves off the anchor, and the record
        // now pins BOTH roots (anchors ∪ heads).
        refs.advance(
            "sessions/alpha",
            1_000,
            0,
            &[(0, head_root_hash)],
            vec![anchor_root_hash],
            2_000,
        )?;
    }

    // Simulated restart: records reload from disk; the registry is empty.
    let refs = BranchRefStore::open(dir.path())?;
    let branches = BranchRegistry::new();
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name_at("old", pruned_root_hash, 10)?;

    let expected_bytes = pruned_root.serialise().len() + pruned_only.serialise().len();
    let report = prune(&store, &branches, &refs, &mut snapshots, "old")?;

    assert_eq!(
        report,
        PruneReport {
            node_count: 2,
            bytes_reclaimed: expected_bytes,
        }
    );
    assert_eq!(store.get(&pruned_root_hash), None);
    assert_eq!(store.get(&pruned_only_hash), None);
    // The durable pin set protected the anchor, the head, and everything
    // they reach — including the node shared with the pruned snapshot.
    assert_eq!(
        store.get(&anchor_root_hash),
        Some(std::sync::Arc::new(anchor_root))
    );
    assert_eq!(
        store.get(&anchor_only_hash),
        Some(std::sync::Arc::new(anchor_only))
    );
    assert_eq!(
        store.get(&head_root_hash),
        Some(std::sync::Arc::new(head_root))
    );
    assert_eq!(
        store.get(&head_only_hash),
        Some(std::sync::Arc::new(head_only))
    );
    assert_eq!(store.get(&shared_hash), Some(std::sync::Arc::new(shared)));
    Ok(())
}