haematite 0.6.2

//! Consumer-shaped adversarial battery (LEDGER A1 stage 4): the tharsis
//! lease-expiry pattern (§11.3), the full §16.2 ABA end-to-end, and
//! bit-rot fail-loud on a REAL installed record (§2.2).

use std::fmt;
use std::fs;
use std::path::{Path, PathBuf};

use super::commit::{BranchCommitError, CommitDurability, CommitRequest, commit_branch};
use super::handle::{BranchError, DEFAULT_SHARD_ID};
use super::lifecycle::{create_branch, open_branch, remove_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(BranchError),
    Tree(TreeError),
    Node(NodeError),
    Io(std::io::Error),
    Missing(&'static str),
}

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}"),
            Self::Missing(what) => write!(formatter, "missing: {what}"),
        }
    }
}

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<BranchError> for TestError {
    fn from(error: 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 durable(refs: &mut BranchRefStore, timestamp: u64) -> CommitRequest<'_> {
    CommitRequest {
        durability: CommitDurability::Durable { refs },
        extra_parents: &[],
        timestamp,
    }
}

/// The one `.ref` file in a ref directory (these tests hold one branch each).
fn sole_ref_file(refs_dir: &Path) -> Result<PathBuf, TestError> {
    for entry in fs::read_dir(refs_dir)? {
        let path = entry?.path();
        if path.extension().and_then(|extension| extension.to_str()) == Some("ref") {
            return Ok(path);
        }
    }
    Err(TestError::Missing("a .ref file"))
}

#[test]
fn tharsis_lease_expiry_snapshot_then_remove_survives_a_crash_between() -> Result<(), TestError> {
    // §11.3's lease-expiry recipe: name the final head as a snapshot FIRST,
    // then remove the branch. A crash between the two leaves BOTH durable
    // artefacts alive — leak-safe, the diff still queryable — and after the
    // remove completes, the snapshot alone must carry the head through a
    // prune. The final arm drops the snapshot too and proves the earlier
    // sparing was not vacuous.
    let mut store = MemoryStore::new();
    let dir = tempfile::tempdir()?;
    let refs_dir = dir.path().join("refs");
    let snapshots_path = dir.path().join("snapshots.reg");
    let registry = BranchRegistry::new();

    let head = {
        let mut refs = BranchRefStore::open(&refs_dir)?;
        let mut snapshots = SnapshotRegistry::open(&snapshots_path)?;
        let anchor = empty_root(&mut store)?;
        let layer = create_branch(
            "layer",
            [(DEFAULT_SHARD_ID, anchor)],
            &mut refs,
            &registry,
            10,
        )?;
        layer.put(DEFAULT_SHARD_ID, b"delta", b"lease work")?;
        commit_branch(&layer, &mut store, &registry, durable(&mut refs, 20))?;
        let head = layer.current_root();

        // Lease expiry, step 1 of 2: pin the final head by name.
        snapshots.name_at("layer/final", head, 30)?;
        head
        // CRASH between the two steps: handle, registry pins, ref store and
        // snapshot registry all drop here.
    };

    // Recovery: both durable artefacts survived the crash — leak-safe.
    let mut refs = BranchRefStore::open(&refs_dir)?;
    let mut snapshots = SnapshotRegistry::open(&snapshots_path)?;
    assert!(refs.get("layer").is_some(), "the record survived the crash");
    assert_eq!(snapshots.get("layer/final"), Some(head));

    // Doubly pinned: a prune whose target reaches exactly the head's nodes
    // reclaims nothing.
    let fresh_registry = BranchRegistry::new();
    snapshots.name_at("cull", head, 40)?;
    let report = prune(&store, &fresh_registry, &refs, &mut snapshots, "cull")?;
    assert_eq!(report.node_count, 0);

    // Lease expiry, step 2 of 2: remove the branch. The snapshot alone now
    // protects the final head.
    let removed = remove_branch("layer", &mut refs)?;
    assert!(matches!(removed, Some(record) if record.name == "layer"));
    snapshots.name_at("cull-again", head, 50)?;
    let report = prune(&store, &fresh_registry, &refs, &mut snapshots, "cull-again")?;
    assert_eq!(
        report.node_count, 0,
        "the named snapshot must carry the head alone"
    );
    assert!(store.get(&head).is_some());

    // Falsification arm: with the final-head snapshot gone too, the same
    // prune reclaims the head — the zeros above were the pins working, not
    // the prune being idle.
    snapshots
        .remove("layer/final")?
        .ok_or(TestError::Missing("layer/final snapshot"))?;
    snapshots.name_at("cull-last", head, 60)?;
    let report = prune(&store, &fresh_registry, &refs, &mut snapshots, "cull-last")?;
    assert!(report.node_count > 0);
    assert!(store.get(&head).is_none());
    Ok(())
}

#[test]
fn stale_handle_cannot_hijack_a_recreated_branch_generation() -> Result<(), TestError> {
    // The full §16.2 ABA, end-to-end: create A → open a handle → remove A →
    // recreate A. Both generations sit at seq 0, so a seq-only CAS would let
    // the stale handle's durable commit through; the creation identity must
    // refuse it — and the NEW generation's record must be untouched,
    // byte-for-byte.
    let mut store = MemoryStore::new();
    let dir = tempfile::tempdir()?;
    let registry = BranchRegistry::new();
    let mut refs = BranchRefStore::open(dir.path())?;
    let anchor = empty_root(&mut store)?;

    create_branch("A", [(DEFAULT_SHARD_ID, anchor)], &mut refs, &registry, 100)?;
    let stale = open_branch("A", &refs, &registry)?;
    remove_branch("A", &mut refs)?;
    let fresh = create_branch("A", [(DEFAULT_SHARD_ID, anchor)], &mut refs, &registry, 200)?;

    let ref_file = sole_ref_file(dir.path())?;
    let bytes_before = fs::read(&ref_file)?;

    stale.put(DEFAULT_SHARD_ID, b"key", b"hijack")?;
    let result = commit_branch(&stale, &mut store, &registry, durable(&mut refs, 300));
    assert!(matches!(
        result,
        Err(BranchCommitError::Ref(
            BranchRefError::BranchGenerationMismatch {
                expected_created: 100,
                found_created: 200,
                ..
            }
        ))
    ));

    // The new generation's record is untouched, byte-for-byte.
    assert_eq!(fs::read(&ref_file)?, bytes_before);
    // Refused, not half-applied: the stale handle keeps its buffer and root.
    assert!(stale.shard_is_dirty(DEFAULT_SHARD_ID)?);
    assert_eq!(stale.current_root(), anchor);

    // The new generation commits normally afterwards.
    fresh.put(DEFAULT_SHARD_ID, b"key", b"legitimate")?;
    let outcome = commit_branch(&fresh, &mut store, &registry, durable(&mut refs, 400))?;
    assert_eq!(outcome.seq, Some(1));
    assert_eq!(
        fresh.get(DEFAULT_SHARD_ID, &store, b"key")?,
        Some(b"legitimate".to_vec())
    );
    Ok(())
}

#[test]
fn bit_rot_in_an_installed_record_fails_open_loud_naming_the_file() -> Result<(), TestError> {
    // §2.2: torn writes are impossible (atomic rename), so corruption means
    // bit rot — and open must fail LOUD naming the file, never skip it (a
    // silently dropped record is a silently dropped prune pin). Rot a REAL
    // installed record's header, then a truncation, and assert both refuse.
    let mut store = MemoryStore::new();
    let dir = tempfile::tempdir()?;
    let registry = BranchRegistry::new();
    let head_bytes = {
        let mut refs = BranchRefStore::open(dir.path())?;
        let anchor = empty_root(&mut store)?;
        let branch = create_branch(
            "fragile",
            [(DEFAULT_SHARD_ID, anchor)],
            &mut refs,
            &registry,
            10,
        )?;
        branch.put(DEFAULT_SHARD_ID, b"key", b"value")?;
        commit_branch(&branch, &mut store, &registry, durable(&mut refs, 20))?;
        fs::read(sole_ref_file(dir.path())?)?
    };
    let ref_file = sole_ref_file(dir.path())?;
    let file_name = ref_file
        .file_name()
        .and_then(|name| name.to_str())
        .ok_or(TestError::Missing("ref file name"))?
        .to_owned();

    // One flipped bit in the magic.
    let mut rotten = head_bytes.clone();
    rotten[0] ^= 0x01;
    fs::write(&ref_file, &rotten)?;
    let result = BranchRefStore::open(dir.path());
    let Err(BranchRefError::Corrupt(message)) = result else {
        return Err(TestError::Missing("a Corrupt error for the bit-flip"));
    };
    assert!(
        message.contains(&file_name),
        "the corruption error must name the offending file: {message}"
    );

    // A truncated record (bit rot of the length kind) refuses the same way.
    fs::write(&ref_file, &head_bytes[..head_bytes.len() / 2])?;
    assert!(matches!(
        BranchRefStore::open(dir.path()),
        Err(BranchRefError::Corrupt(_))
    ));

    // Restore the healthy bytes: the store opens again and serves the record
    // — fail-loud is about refusing rot, not bricking the directory.
    fs::write(&ref_file, &head_bytes)?;
    let refs = BranchRefStore::open(dir.path())?;
    assert!(refs.get("fragile").is_some());
    Ok(())
}