haematite 0.7.0

//! §11 sweep behaviour pins: the dry run, the write-surface identity, the
//! terminal acceptance gate, the empty-root orphan, and the property that
//! sweep preserves exactly the union of resolvable pin sources.

use std::collections::{BTreeMap, HashSet};
use std::error::Error;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::SystemTime;

use crate::db::Database;
use crate::store::DiskStore;
use crate::tree::{Hash, Node, empty_root_hash};

use super::test_support::{
    attested, build_store, committed_root, config_for, node_path, snapshot_tree, store_dir,
    sweep_attested,
};
use super::{VacuumOptions, vacuum_stats, vacuum_sweep};

type Tree = BTreeMap<PathBuf, (Option<Vec<u8>>, SystemTime)>;

/// The write-surface pin, exactly (§5, §11 ⟨r4, M4⟩⟨r5, m⟩): `after` must be
/// `before` minus PRECISELY `deleted`, allowing only the ancestor-directory
/// mtime changes the unlinks force. Every surviving FILE — marked nodes,
/// `.node-*.tmp` debris, all metadata — must be byte+mtime identical, and no
/// entry may appear that was not there before. Removing one extra file (even
/// non-live forensic debris) fails this.
fn assert_only_deleted(
    before: &Tree,
    after: &Tree,
    deleted: &[PathBuf],
    root: &Path,
) -> Result<(), Box<dyn Error>> {
    let mut deleted_rel: HashSet<PathBuf> = HashSet::new();
    for path in deleted {
        let rel = path
            .strip_prefix(root)
            .map_err(|_error| "deleted path is not under the data dir")?
            .to_path_buf();
        deleted_rel.insert(rel);
    }

    for rel in &deleted_rel {
        assert!(
            before.contains_key(rel),
            "reported-deleted path {rel:?} was not in the pre-sweep tree"
        );
        assert!(
            !after.contains_key(rel),
            "reported-deleted path {rel:?} still exists after sweep"
        );
    }
    for (rel, identity) in before {
        if deleted_rel.contains(rel) {
            continue;
        }
        match after.get(rel) {
            None => {
                return Err(
                    format!("sweep removed {rel:?}, which it did NOT report deleting").into(),
                );
            }
            Some(after_identity) => {
                // Files must be byte+mtime identical; directories may have a
                // fresh mtime (they are ancestors of the unlinks).
                if identity.0.is_some() {
                    assert_eq!(
                        identity, after_identity,
                        "surviving file {rel:?} changed under sweep"
                    );
                }
            }
        }
    }
    for rel in after.keys() {
        assert!(
            before.contains_key(rel),
            "sweep created a new entry {rel:?}"
        );
    }
    Ok(())
}

/// Every hash whose canonical node file exists under `store_dir`.
fn stored_hashes(store: &Path) -> Result<HashSet<Hash>, Box<dyn Error>> {
    let mut hashes = HashSet::new();
    for prefix in fs::read_dir(store)? {
        let prefix = prefix?.path();
        if !prefix.is_dir() {
            continue;
        }
        let prefix_hex = prefix
            .file_name()
            .and_then(|name| name.to_str())
            .ok_or("prefix name")?
            .to_owned();
        for entry in fs::read_dir(&prefix)? {
            let path = entry?.path();
            if !path.is_file() {
                continue;
            }
            let name = path
                .file_name()
                .and_then(|name| name.to_str())
                .ok_or("name")?;
            let hex = format!("{prefix_hex}{name}");
            let mut bytes = [0_u8; 32];
            for (index, chunk) in hex.as_bytes().chunks_exact(2).enumerate() {
                bytes[index] = u8::from_str_radix(std::str::from_utf8(chunk)?, 16)?;
            }
            hashes.insert(Hash::from_bytes(bytes));
        }
    }
    Ok(hashes)
}

/// Independent oracle: the reachable-AND-present hashes from `roots`, walked
/// through a plain `DiskStore` (no hash verification) — a different code path
/// than the vacuum's verified-file walk, so agreement is real evidence.
fn oracle_reachable(store: &DiskStore, roots: &[Hash]) -> Result<HashSet<Hash>, Box<dyn Error>> {
    let mut seen = HashSet::new();
    let mut stack: Vec<Hash> = roots.to_vec();
    while let Some(hash) = stack.pop() {
        if seen.contains(&hash) {
            continue;
        }
        if let Some(node) = store.get(&hash)? {
            seen.insert(hash);
            if let Node::Internal(internal) = &*node {
                stack.extend(internal.children().iter().map(|(_lower, child)| *child));
            }
        }
    }
    Ok(seen)
}

/// `--dry-run`: full pass, refusals evaluated, deletion list computed and
/// reported, ZERO unlinks. The tree is byte+mtime identical afterwards.
#[test]
fn dry_run_reports_plan_and_deletes_nothing() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"key-a", b"key-b"], 4)?;
    let before = snapshot_tree(temp.path())?;

    let report = sweep_attested(temp.path(), false)?;
    let summary = report.sweep.as_ref().ok_or("dry run carries a summary")?;
    assert!(!summary.executed, "dry run executes nothing");
    assert!(summary.deleted_nodes > 0, "there is garbage to plan");
    assert_eq!(summary.deleted_nodes, report.totals.unmarked_nodes);
    assert_eq!(summary.deleted_paths.len(), summary.deleted_nodes);
    assert_eq!(
        before,
        snapshot_tree(temp.path())?,
        "a dry run performs ZERO unlinks"
    );
    Ok(())
}

/// The sweep deletes PRECISELY the complement (write-surface pin), then the
/// terminal gate: reopen, every root resolves, live events read back
/// byte-identical, and a second sweep deletes nothing.
#[test]
fn sweep_deletes_exact_complement_then_terminal_gate() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    let keys: &[&[u8]] = &[b"key-a", b"key-b", b"key-c"];
    build_store(temp.path(), 1, keys, 4)?;

    // Capture the full live history BEFORE sweep for the liveness check.
    let live_before: Vec<Vec<Vec<u8>>> = {
        let db = Database::open(temp.path())?;
        let history = keys
            .iter()
            .map(|key| db.read_events(key))
            .collect::<Result<_, _>>()?;
        drop(db);
        history
    };

    let before = snapshot_tree(temp.path())?;
    let report = sweep_attested(temp.path(), true)?;
    let summary = report.sweep.as_ref().ok_or("summary")?;
    assert!(summary.executed);
    assert!(summary.deleted_nodes > 0, "version garbage existed");

    // Write-surface: after = before minus PRECISELY the reported paths.
    let after = snapshot_tree(temp.path())?;
    assert_only_deleted(&before, &after, &summary.deleted_paths, temp.path())?;

    // Terminal gate — every root resolves and live bytes are identical.
    let db = Database::open(temp.path())?;
    for (key, expected) in keys.iter().zip(&live_before) {
        assert_eq!(&db.read_events(key)?, expected, "live events survived");
    }
    drop(db);

    // A completed second stats run proves every M1/M2/M3 root still fully
    // resolves (any dangling root would refuse UnresolvableRoot), using the
    // read-only decode path — not BranchRefStore::open.
    vacuum_stats(&attested(temp.path()))?;

    // Second sweep = ZERO deletions (idempotence as quiescence, §5).
    let second = sweep_attested(temp.path(), true)?;
    assert_eq!(
        second.sweep.as_ref().ok_or("summary")?.deleted_nodes,
        0,
        "the store is already minimal — a second sweep deletes nothing"
    );
    Ok(())
}

/// ⟨board defect⟩ the empty-root orphan (`store_empty_root`, unreachable after
/// the first real commit) is ordinary garbage — zero special code. A fresh
/// store + one commit leaves exactly one unmarked node, the empty root; sweep
/// deletes it; the terminal gate is green.
#[test]
fn empty_root_orphan_is_swept_as_ordinary_garbage() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"only"], 1)?;

    let empty = empty_root_hash();
    let store = store_dir(temp.path(), 0);
    let empty_path = node_path(&store, empty);
    assert!(
        empty_path.exists(),
        "the first-commit baseline wrote the empty root"
    );

    let report = sweep_attested(temp.path(), false)?;
    assert_eq!(
        report.totals.unmarked_nodes, 1,
        "exactly one unmarked node — the empty root — after one real commit"
    );
    assert!(
        report
            .sweep
            .as_ref()
            .ok_or("summary")?
            .deleted_paths
            .contains(&empty_path),
        "the empty root falls out in the complement"
    );

    sweep_attested(temp.path(), true)?;
    assert!(!empty_path.exists(), "sweep deleted the empty-root orphan");

    // Terminal gate: reopen, the live key reads, a second sweep is zero.
    let db = Database::open(temp.path())?;
    assert_eq!(db.read_events(b"only")?, vec![b"v0".to_vec()]);
    drop(db);
    let second = sweep_attested(temp.path(), true)?;
    assert_eq!(second.sweep.as_ref().ok_or("summary")?.deleted_nodes, 0);
    Ok(())
}

/// Property (§11): a store with commit garbage AND a snapshot pin on a
/// superseded root ⇒ sweep preserves EXACTLY the union of resolvable pin
/// sources (WAL committed root ∪ snapshot root), compared against an
/// independent oracle walk. Nothing live is lost; nothing garbage survives.
#[test]
fn sweep_preserves_exactly_the_resolvable_pin_union() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    // A superseded root captured mid-history, then more commits pile garbage.
    let db = Database::create(config_for(temp.path(), 1))?;
    for key in [b"alpha".as_slice(), b"beta", b"gamma"] {
        db.append(key.to_vec(), vec![b"v0".to_vec()], 0)?;
    }
    drop(db);
    let pinned_root = committed_root(temp.path(), 0).ok_or("v0 root")?;
    let db = Database::open(temp.path())?;
    for version in 1..4_u64 {
        for key in [b"alpha".as_slice(), b"beta", b"gamma"] {
            db.append(
                key.to_vec(),
                vec![format!("v{version}").into_bytes()],
                version,
            )?;
        }
    }
    drop(db);
    let head_root = committed_root(temp.path(), 0).ok_or("head root")?;
    assert_ne!(
        pinned_root, head_root,
        "the pinned root is superseded garbage"
    );

    // Name the superseded root in an external registry (a resolvable pin).
    let registry_dir = tempfile::tempdir()?;
    let registry_file = registry_dir.path().join("snapshots.hsr");
    let mut registry = crate::branch::snapshot::SnapshotRegistry::open(&registry_file)?;
    registry.name_at("keep", pinned_root, 1)?;
    drop(registry);

    // The oracle: reachable-and-present from the pin union, walked through a
    // plain DiskStore.
    let store = store_dir(temp.path(), 0);
    let oracle = {
        let disk = DiskStore::new(&store)?;
        oracle_reachable(&disk, &[head_root, pinned_root])?
    };

    let mut options = attested(temp.path());
    options.snapshot_files.push(registry_file);
    let report = vacuum_sweep(&options, true)?;
    assert!(report.sweep.as_ref().ok_or("summary")?.deleted_nodes > 0);

    // Surviving node files == the oracle live set, exactly.
    let survivors = stored_hashes(&store)?;
    assert_eq!(
        survivors, oracle,
        "sweep must preserve EXACTLY the resolvable pin union — no live loss, no garbage kept"
    );
    // And the pinned superseded root is among the survivors (leak-safe pin).
    assert!(
        survivors.contains(&pinned_root),
        "the snapshot pin was honoured"
    );
    Ok(())
}

/// A completed sweep with no metadata universe established still refuses
/// (guards the happy-path tests: they pass only because they attest).
#[test]
fn sweep_without_universe_refuses_even_with_garbage() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"k"], 3)?;
    let result = vacuum_sweep(&VacuumOptions::new(temp.path().to_path_buf()), true);
    assert!(result.is_err(), "no manifest, no attestation ⇒ refuse");
    Ok(())
}