haematite 0.7.0

//! The v1-anchor fork-crossing witnesses (CHUNKING-POLICY.md §4.2 step 5, F1):
//! the non-mixed witness, idempotence, v1-unchanged, a pure-v2 commit off a
//! crossed base, and prune-safety of the rebuilt working root.

use std::error::Error;

use crate::store::MemoryStore;
use crate::tree::{Cursor, Hash, LeafNode, Node, TreeError, TreePolicy, batch_mutate};

use super::super::commit::{CommitDurability, CommitRequest, commit_branch};
use super::super::handle::DEFAULT_SHARD_ID;
use super::super::lifecycle::fork_at;
use super::super::prune::prune;
use super::super::refstore::BranchRefStore;
use super::super::registry::{BranchRegistry, CrossRegisterError};
use super::super::snapshot::SnapshotRegistry;

type Res = Result<(), Box<dyn Error>>;

fn hash(byte: u8) -> Hash {
    Hash::from_bytes([byte; 32])
}

/// Small byte targets so a 40-byte-value entry is a DETERMINISTIC v2 singleton
/// (§2.1) — the v2 tree shape then differs observably from the v1 packing.
fn v2() -> TreePolicy {
    TreePolicy::v2(48, 48)
}

fn oversized(seed: u8) -> Vec<u8> {
    vec![seed; 40]
}

/// Four oversized entries — enough that the v2 spine differs from the v1 leaf.
fn map() -> Vec<(Vec<u8>, Option<Vec<u8>>)> {
    (0u8..4)
        .map(|n| (format!("k{n:02}").into_bytes(), Some(oversized(n))))
        .collect()
}

/// A DISJOINT map (different keys) — its tree shares NO nodes with [`map`]'s, so
/// a prune of a snapshot at its root performs real reclamation.
fn other_map() -> Vec<(Vec<u8>, Option<Vec<u8>>)> {
    (0u8..3)
        .map(|n| {
            (
                format!("other-{n:02}").into_bytes(),
                Some(oversized(200 + n)),
            )
        })
        .collect()
}

fn build_from(
    store: &mut MemoryStore,
    entries: &[(Vec<u8>, Option<Vec<u8>>)],
    policy: TreePolicy,
) -> Result<Hash, Box<dyn Error>> {
    let empty = store.put(&Node::Leaf(LeafNode::new(Vec::new())?));
    Ok(batch_mutate(store, empty, entries, policy)?)
}

fn build_root(store: &mut MemoryStore, policy: TreePolicy) -> Result<Hash, Box<dyn Error>> {
    build_from(store, &map(), policy)
}

/// THE non-mixed witness (F1 4a): `fork_at` of a genuinely v1-shaped anchor
/// under a v2 policy rebuilds the working tree to the fresh-build canonical v2
/// root, while the fork LINEAGE stays at the original v1 anchor. The `assert_ne`
/// makes it non-vacuous (v1 shape ≠ v2 shape on this fixture).
#[test]
fn fork_at_crosses_v1_anchor_to_canonical_v2() -> Res {
    let mut store = MemoryStore::new();
    let v1_anchor = build_root(&mut store, TreePolicy::V1_DEFAULT)?;
    let v2_canonical = build_root(&mut store, v2())?;
    assert_ne!(
        v1_anchor, v2_canonical,
        "the v2 shape must differ from v1 for the witness to bite"
    );

    let registry = BranchRegistry::new();
    let dir = tempfile::tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name("anchor-pin", v1_anchor)?;

    let handle = fork_at(v1_anchor, &mut store, v2(), &registry, &refs, &snapshots)?;
    assert_eq!(
        handle.fork_point(),
        v1_anchor,
        "the fork lineage must stay at the original v1 anchor (merge ancestor)"
    );
    assert_eq!(
        handle.current_root(),
        v2_canonical,
        "the mutation base must be the fresh-build canonical v2 root, never the v1 anchor"
    );
    Ok(())
}

/// A commit off a crossed base is PURE v2 — never a mixed tree grafted onto v1
/// nodes: the committed root equals the fresh-build of (map + the new entry)
/// under v2.
#[test]
fn crossed_fork_commits_pure_v2() -> Res {
    let mut store = MemoryStore::new();
    let v1_anchor = build_root(&mut store, TreePolicy::V1_DEFAULT)?;

    let registry = BranchRegistry::new();
    let dir = tempfile::tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name("anchor-pin", v1_anchor)?;

    let handle = fork_at(v1_anchor, &mut store, v2(), &registry, &refs, &snapshots)?;
    handle.put(DEFAULT_SHARD_ID, b"k04", oversized(9))?;
    commit_branch(
        &handle,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Volatile,
            extra_parents: &[],
            timestamp: 1,
        },
        v2(),
    )?;

    let empty = store.put(&Node::Leaf(LeafNode::new(Vec::new())?));
    let mut expected_map = map();
    expected_map.push((b"k04".to_vec(), Some(oversized(9))));
    let expected = batch_mutate(&mut store, empty, &expected_map, v2())?;
    assert_eq!(
        handle.current_root(),
        expected,
        "a commit off the crossed base must equal the fresh-build canonical v2 tree"
    );
    Ok(())
}

/// Idempotence (F1 4b): re-crossing an already-v2 anchor rebuilds to the
/// IDENTICAL root (every node dedupes) — a structural no-op.
#[test]
fn fork_at_of_v2_anchor_is_identity() -> Res {
    let mut store = MemoryStore::new();
    let v2_anchor = build_root(&mut store, v2())?;

    let registry = BranchRegistry::new();
    let dir = tempfile::tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name("anchor-pin", v2_anchor)?;

    let handle = fork_at(v2_anchor, &mut store, v2(), &registry, &refs, &snapshots)?;
    assert_eq!(
        handle.current_root(),
        v2_anchor,
        "re-crossing an already-v2 anchor is the identity"
    );
    assert_eq!(handle.fork_point(), v2_anchor);
    Ok(())
}

/// v1-database behaviour unchanged (F1 4c): under a v1 policy the crossing is
/// skipped entirely — the working root IS the anchor, byte-for-byte as today.
#[test]
fn fork_at_under_v1_policy_skips_the_crossing() -> Res {
    let mut store = MemoryStore::new();
    let v1_anchor = build_root(&mut store, TreePolicy::V1_DEFAULT)?;

    let registry = BranchRegistry::new();
    let dir = tempfile::tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name("anchor-pin", v1_anchor)?;

    let handle = fork_at(
        v1_anchor,
        &mut store,
        TreePolicy::V1_DEFAULT,
        &registry,
        &refs,
        &snapshots,
    )?;
    assert_eq!(
        handle.current_root(),
        v1_anchor,
        "v1 keeps today's behaviour: the working root is the anchor, no rebuild"
    );
    assert_eq!(handle.fork_point(), v1_anchor);
    Ok(())
}

/// Prune-safety (F1 4e, variant a — non-vacuous): a prune of a GENUINELY
/// UNRELATED snapshot (a disjoint map) performs REAL reclamation AROUND the
/// pinned working root; the working root stays fully readable, and the pin
/// releases when the handle drops.
#[test]
fn crossed_working_root_survives_prune_of_a_disjoint_snapshot() -> Res {
    let mut store = MemoryStore::new();
    let v1_anchor = build_root(&mut store, TreePolicy::V1_DEFAULT)?;

    let registry = BranchRegistry::new();
    let dir = tempfile::tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name("anchor-pin", v1_anchor)?;

    let handle = fork_at(v1_anchor, &mut store, v2(), &registry, &refs, &snapshots)?;
    let working = handle.current_root();
    assert!(
        registry.live_roots().contains(&working),
        "the rebuilt working root must be pinned in the registry"
    );

    // A snapshot at a DISJOINT tree (no shared nodes): pruning it reclaims real
    // nodes around the pin, proving the witness is non-vacuous.
    let scratch = build_from(&mut store, &other_map(), v2())?;
    snapshots.name("scratch", scratch)?;
    let report = prune(&store, &registry, &refs, &mut snapshots, "scratch")?;
    assert!(
        report.node_count > 0,
        "the disjoint snapshot must actually reclaim nodes (non-vacuous prune)"
    );
    assert!(
        Cursor::new(&store, working).get(b"k00")?.is_some(),
        "the crossed working root must survive the prune"
    );

    drop(handle);
    assert!(
        !registry.live_roots().contains(&working),
        "dropping the handle must release the working-root pin"
    );
    Ok(())
}

/// Prune-safety (F1 4e, variant b — the dedup collision `registry.rs`'s
/// `cross_and_register` comment names on purpose): a snapshot whose root is
/// BYTE-IDENTICAL to the pinned working root (same map, same v2 policy → same
/// hash). Pruning that snapshot must NOT reclaim the shared nodes, because the
/// registry pin — taken under the same one-counts-lock as the rebuild — protects
/// them; the working root stays fully readable.
#[test]
fn crossed_working_root_survives_prune_of_a_dedup_colliding_snapshot() -> Res {
    let mut store = MemoryStore::new();
    let v1_anchor = build_root(&mut store, TreePolicy::V1_DEFAULT)?;

    let registry = BranchRegistry::new();
    let dir = tempfile::tempdir()?;
    let refs = BranchRefStore::open(dir.path())?;
    let mut snapshots = SnapshotRegistry::new();
    snapshots.name("anchor-pin", v1_anchor)?;

    let handle = fork_at(v1_anchor, &mut store, v2(), &registry, &refs, &snapshots)?;
    let working = handle.current_root();

    // A snapshot at the SAME v2 map: HI canonicity makes its root byte-identical
    // to the working root (the exact dedup-collision hazard). Pruning it must
    // reclaim NONE of the working root's nodes.
    let collision = build_root(&mut store, v2())?;
    assert_eq!(
        collision, working,
        "the collision snapshot must equal the working root"
    );
    snapshots.name("collision", collision)?;
    prune(&store, &registry, &refs, &mut snapshots, "collision")?;
    assert!(
        Cursor::new(&store, working).get(b"k00")?.is_some(),
        "a dedup-colliding prune must not reclaim the pinned working root's nodes"
    );
    Ok(())
}

/// MUST-FIX regression pin: a `build` closure that violates the
/// one-working-root-per-anchor contract (too FEW, then too MANY) is refused with
/// the typed [`CrossRegisterError::WorkingRootCountMismatch`] BEFORE any
/// registration — so NOTHING is pinned (no partial/silent under-pin).
#[test]
fn cross_and_register_refuses_a_mismatched_working_root_count() {
    let registry = BranchRegistry::new();
    let anchors = [hash(1), hash(2)];

    // Too FEW: one working root for two anchors.
    let short: Result<_, CrossRegisterError<TreeError>> =
        registry.cross_and_register(&anchors, |_| true, |_| Ok(vec![hash(9)]));
    assert!(
        matches!(
            short,
            Err(CrossRegisterError::WorkingRootCountMismatch {
                expected: 2,
                got: 1
            })
        ),
        "a short build must be refused as a count mismatch"
    );
    assert!(
        registry.live_roots().is_empty(),
        "a short-build mismatch must register NOTHING (no partial pins)"
    );

    // Too MANY: three working roots for two anchors.
    let long: Result<_, CrossRegisterError<TreeError>> = registry.cross_and_register(
        &anchors,
        |_| true,
        |_| Ok(vec![hash(9), hash(10), hash(11)]),
    );
    assert!(
        matches!(
            long,
            Err(CrossRegisterError::WorkingRootCountMismatch {
                expected: 2,
                got: 3
            })
        ),
        "a long build must be refused as a count mismatch"
    );
    assert!(
        registry.live_roots().is_empty(),
        "a long-build mismatch must register NOTHING"
    );
}