haematite 0.7.0

//! §11 crash matrix for the sweep, extending the fence crash harness
//! (`PinCase`/`run_pin`, one OS-kill protocol): a writer child sweeps to a
//! phase boundary and is `kill -9`'d (no destructors, no flush); a fresh
//! recovery child reruns the sweep and asserts CONVERGENCE.
//!
//! The invariant every case pins (§5): a killed-mid-sweep store is a strict
//! SUPERSET of live, and because the mark set is recomputed per run and never
//! persisted, a rerun converges to the same minimal store — every root
//! resolves, live events read back, and a second sweep deletes nothing. The
//! unlinks are unfenced by design: a process kill does not roll back a
//! committed unlink, and a power-loss rollback would only reinstate a node the
//! rerun re-deletes — either way the recovery rerun is the whole repair.
//!
//! The four phase boundaries the design names are the four writers below: kill
//! between mark and sweep (nothing deleted yet), mid-sweep within a store (a
//! strict subset unlinked), between stores (one shard fully swept, the next
//! untouched), and during report write (all unlinks done, the report never
//! emitted — the vacuum persists no state, so recovery just confirms
//! quiescence).

use std::error::Error;
use std::path::{Path, PathBuf};

use crate::db::Database;
use crate::fence::crash_harness::{PinCase, TestResult, run_pin};

use super::test_support::{attested, build_store, sweep_attested};
use super::vacuum_stats;

const KEYS: [&[u8]; 3] = [b"alpha", b"beta", b"gamma"];

fn data_dir(world: &Path) -> PathBuf {
    world.join("db")
}

/// The deletion plan for the store, computed by a zero-unlink dry run.
fn plan(dir: &Path) -> Result<Vec<PathBuf>, Box<dyn Error>> {
    let report = sweep_attested(dir, false)?;
    Ok(report
        .sweep
        .ok_or("dry run carries a summary")?
        .deleted_paths)
}

/// Rerun the sweep to convergence and assert the terminal gate: a completed
/// run proves every root resolves (read-only decode, not `BranchRefStore::open`),
/// a second sweep deletes nothing, and every key's events read back.
fn recover_converges(world: &Path) -> TestResult {
    let dir = data_dir(world);
    sweep_attested(&dir, true)?;
    vacuum_stats(&attested(&dir))?;
    let second = sweep_attested(&dir, true)?;
    let deleted = second.sweep.ok_or("summary")?.deleted_nodes;
    if deleted != 0 {
        return Err(format!("second sweep deleted {deleted} nodes — not converged").into());
    }
    let db = Database::open(&dir)?;
    for key in KEYS {
        db.read_events(key)?;
    }
    drop(db);
    Ok(())
}

// --- kill between mark and sweep: nothing deleted, full superset ------------

fn between_mark_and_sweep_write(world: &Path) -> TestResult {
    build_store(&data_dir(world), 1, &KEYS, 4)?;
    // The mark ran (the dry run proves the plan is non-empty); no unlink yet.
    let planned = plan(&data_dir(world))?;
    if planned.is_empty() {
        return Err("the store must carry garbage to sweep".into());
    }
    Ok(())
}

#[test]
fn kill_between_mark_and_sweep_converges() -> TestResult {
    run_pin(&PinCase {
        test_path: "db::vacuum::sweep_crash_tests::kill_between_mark_and_sweep_converges",
        write: between_mark_and_sweep_write,
        recover: recover_converges,
    })
}

// --- kill mid-sweep within a store: a strict subset unlinked ----------------

fn mid_sweep_write(world: &Path) -> TestResult {
    let dir = data_dir(world);
    build_store(&dir, 1, &KEYS, 4)?;
    let planned = plan(&dir)?;
    if planned.len() < 2 {
        return Err("need at least two garbage nodes to unlink a strict subset".into());
    }
    // Unlink HALF the plan directly: the store is now a strict superset of
    // live (some garbage remains). The kill lands here.
    for path in &planned[..planned.len() / 2] {
        std::fs::remove_file(path)?;
    }
    Ok(())
}

#[test]
fn kill_mid_sweep_converges() -> TestResult {
    run_pin(&PinCase {
        test_path: "db::vacuum::sweep_crash_tests::kill_mid_sweep_converges",
        write: mid_sweep_write,
        recover: recover_converges,
    })
}

// --- kill between stores: shard 0 fully swept, shard 1 untouched ------------

fn between_stores_write(world: &Path) -> TestResult {
    let dir = data_dir(world);
    build_store(&dir, 2, &KEYS, 4)?;
    let planned = plan(&dir)?;
    let shard0: Vec<&PathBuf> = planned
        .iter()
        .filter(|path| path.to_string_lossy().contains("/shard-0/"))
        .collect();
    if shard0.is_empty() || shard0.len() == planned.len() {
        return Err("both shards must carry garbage for a between-stores kill".into());
    }
    // Fully sweep shard 0, leave shard 1 entirely — the between-stores window.
    for path in shard0 {
        std::fs::remove_file(path)?;
    }
    Ok(())
}

#[test]
fn kill_between_stores_converges() -> TestResult {
    run_pin(&PinCase {
        test_path: "db::vacuum::sweep_crash_tests::kill_between_stores_converges",
        write: between_stores_write,
        recover: recover_converges,
    })
}

// --- kill during report write: all unlinks done, report never emitted -------

fn during_report_write_write(world: &Path) -> TestResult {
    let dir = data_dir(world);
    build_store(&dir, 1, &KEYS, 4)?;
    // The full sweep completed its unlinks; the kill lands before the report
    // would be returned. The vacuum persists no state, so the store is already
    // minimal and recovery simply confirms quiescence.
    sweep_attested(&dir, true)?;
    Ok(())
}

#[test]
fn kill_during_report_write_converges() -> TestResult {
    run_pin(&PinCase {
        test_path: "db::vacuum::sweep_crash_tests::kill_during_report_write_converges",
        write: during_report_write_write,
        recover: recover_converges,
    })
}