haematite 0.4.1

//! §14.3(a): the falsifiable prune-regression tests, end-to-end through
//! `commit_branch` (LEDGER A1 stage 4).
//!
//! `registry_tests.rs` proves the `advance` primitive's one-lock swap in
//! isolation; these tests prove the COMMIT PATH actually uses it. The two
//! regressions the brief names, and how each test fails if it comes back:
//!
//! * **commit forgets to register the advanced root** — the sequential test
//!   asserts the new head is in `live_roots()` the instant `commit_branch`
//!   returns (and the superseded head is out, and the anchor never moves);
//! * **deregister precedes register (or the swap spans two lock takes)** —
//!   the concurrent spy snapshots `live_roots()` in a tight loop during a
//!   storm of commits; any interleaving in which neither the superseded nor
//!   the advanced head is live shrinks the set below `{anchor, head}` and is
//!   counted as a violation. `commit_branch` holds the registry counts guard
//!   from step 1 to the end (§16.1 MF1), so a snapshot can never land inside
//!   the dance; a two-lock regression fails this probabilistically within a
//!   handful of iterations.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

use super::commit::{BranchCommitError, CommitDurability, CommitRequest, commit_branch};
use super::fork::fork_registered;
use super::handle::DEFAULT_SHARD_ID;
use super::registry::BranchRegistry;
use crate::store::MemoryStore;
use crate::tree::{Hash, LeafNode, Node, TreeError, insert};

type TestResult = Result<(), BranchCommitError>;

fn build_tree(store: &mut MemoryStore, entries: &[(&[u8], &[u8])]) -> Result<Hash, TreeError> {
    let leaf = LeafNode::new(Vec::new())?;
    let mut root = store.put(&Node::Leaf(leaf));
    for (key, value) in entries {
        root = insert(store, root, key, value)?;
    }
    Ok(root)
}

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

#[test]
fn every_commit_registers_the_advanced_root_before_returning() -> TestResult {
    // The register half of §14.3(a): a chain of commits, asserting after each
    // one that the pin set is exactly {anchor, current head} — the advanced
    // root registered, the superseded head released, the anchor's refcount-2
    // pin (§14.1) untouched. A commit that "forgot" the register would leave
    // the fresh head unpinned here and hand prune a live root to reclaim.
    let mut store = MemoryStore::new();
    let anchor = build_tree(&mut store, &[(b"base", b"tree")])?;
    let registry = BranchRegistry::new();
    let branch = fork_registered(anchor, &registry);

    let mut previous_head = anchor;
    for round in 0u32..50 {
        branch.put(DEFAULT_SHARD_ID, b"key", round.to_le_bytes())?;
        commit_branch(&branch, &mut store, &registry, volatile())?;
        let head = branch.current_root();
        assert_ne!(head, previous_head, "each round must advance the head");

        let live = registry.live_roots();
        assert!(
            live.contains(&head),
            "round {round}: the advanced root must be registered when commit returns"
        );
        assert!(
            live.contains(&anchor),
            "round {round}: the anchor-role pin must survive every advance"
        );
        if previous_head != anchor {
            assert!(
                !live.contains(&previous_head),
                "round {round}: the superseded head must be released"
            );
        }
        assert_eq!(
            live.len(),
            2,
            "round {round}: the pin set must be exactly {{anchor, head}}"
        );
        previous_head = head;
    }
    Ok(())
}

#[test]
fn concurrent_live_roots_snapshots_never_observe_a_pin_gap() -> TestResult {
    // The ordering half of §14.3(a): while the main thread storms commits,
    // a spy thread snapshots live_roots(). Invariants at every instant once
    // the first commit has landed: the anchor is pinned, and at least one of
    // {superseded head, advanced head} is pinned (set size >= 2, since every
    // head differs from the anchor). A commit path that deregistered the old
    // head before registering the new one — or did the two under separate
    // lock takes — exposes a {anchor}-only window the spy counts.
    //
    // The spy records violations instead of asserting: panicking off the test
    // thread is both denied by house law and invisible to the harness.
    let mut store = MemoryStore::new();
    let anchor = build_tree(&mut store, &[(b"base", b"tree")])?;
    let registry = BranchRegistry::new();
    let branch = fork_registered(anchor, &registry);

    // First advance outside the spy window, so the invariant below is simply
    // "anchor + some head" for the whole observed run.
    branch.put(DEFAULT_SHARD_ID, b"key", 0u32.to_le_bytes())?;
    commit_branch(&branch, &mut store, &registry, volatile())?;

    let stop = Arc::new(AtomicBool::new(false));
    let violations = Arc::new(AtomicUsize::new(0));
    let spy = {
        let registry = registry.clone();
        let stop = Arc::clone(&stop);
        let violations = Arc::clone(&violations);
        std::thread::spawn(move || {
            let mut snapshots = 0u64;
            while !stop.load(Ordering::Relaxed) {
                let live = registry.live_roots();
                if !live.contains(&anchor) || live.len() < 2 {
                    violations.fetch_add(1, Ordering::Relaxed);
                }
                snapshots += 1;
            }
            snapshots
        })
    };

    for round in 1u32..400 {
        branch.put(DEFAULT_SHARD_ID, b"key", round.to_le_bytes())?;
        commit_branch(&branch, &mut store, &registry, volatile())?;
    }
    stop.store(true, Ordering::Relaxed);
    let snapshots = spy.join();
    assert!(
        matches!(snapshots, Ok(count) if count > 0),
        "the spy must have taken at least one snapshot"
    );
    assert_eq!(
        violations.load(Ordering::Relaxed),
        0,
        "a live_roots() snapshot observed a pin gap during the commit storm"
    );
    Ok(())
}