haematite 0.7.0

//! FENCE-CHAIN r8 R3 — the crash pin census (six pins + the Replace falsifier).
//!
//! Each pin: a writer child performs a durable operation whose fences run for
//! real, runs the CUT (which — for a correctly fenced op — finds nothing pending
//! and rewinds nothing), reports READY, and is killed; a fresh recovery child
//! cold-reopens and asserts the EXACT committed state. A missing fence would leave
//! its entry pending, the CUT would rewind it, and recovery would fail — so these
//! are the pins the R4a walls' patched-away polarity fails against.
//!
//! Census: `primary` (L2+L3, refs L2-isolated), `l4`, `l1`, `l3_alone`, `l5`,
//! `replace_positive`; plus `replace_falsifier` — labelled evidence, never a green
//! pin (it observes the OLD record, proving the `ReplaceInstall` polarity can express
//! the loss).

use crate::branch::handle::DEFAULT_SHARD_ID;
use crate::branch::{
    BranchKind, BranchRefRecord, BranchRefStore, BranchRegistry, BranchShardRef, CommitDurability,
    CommitLog, CommitRequest, SnapshotRegistry, commit_branch, create_branch, open_branch,
};
use crate::db::{Database, DatabaseConfig};
use crate::store::{DiskStore, NodeStore};
use crate::tree::{Hash, LeafNode, Node, TreePolicy, insert};

use std::error::Error;
use std::path::Path;

use super::crash_harness::{PinCase, TestResult, run_pin};
use super::journal;

fn db_config(dir: &Path) -> DatabaseConfig {
    DatabaseConfig {
        data_dir: dir.join("db"),
        shard_count: 1,
        distributed: None,
        executor_threads: None,
    }
}

fn record(name: &str, head: u8) -> BranchRefRecord {
    BranchRefRecord {
        name: name.to_owned(),
        created: 1,
        kind: BranchKind::Work,
        namespace_lineage: None,
        seq: 0,
        timestamp: 1,
        shards: vec![BranchShardRef {
            shard_id: 0,
            fork_anchor: Hash::from_bytes([1; 32]),
            head: Hash::from_bytes([head; 32]),
        }],
        parents: Vec::new(),
    }
}

fn boxed<T, E: std::fmt::Display>(result: Result<T, E>) -> Result<T, Box<dyn Error>> {
    result.map_err(|error| -> Box<dyn Error> { error.to_string().into() })
}

// --- L5: database create, cut at the open-time fence ---------------------------

fn l5_write(dir: &Path) -> TestResult {
    let database = Database::create(db_config(dir))?;
    drop(database);
    journal::cut()?;
    Ok(())
}
fn l5_recover(dir: &Path) -> TestResult {
    let database = Database::open(db_config(dir).data_dir)?;
    assert_eq!(
        database.shard_count(),
        1,
        "cold reopen must find the created database"
    );
    Ok(())
}
#[test]
fn pin_l5_database_create() -> TestResult {
    run_pin(&PinCase {
        test_path: "fence::crash_pin_tests::pin_l5_database_create",
        write: l5_write,
        recover: l5_recover,
    })
}

// --- L1: append to a never-materialised shard ----------------------------------

fn l1_write(dir: &Path) -> TestResult {
    let database = Database::create(db_config(dir))?;
    database.append(b"l1".to_vec(), vec![b"payload".to_vec()], 0)?;
    drop(database);
    journal::cut()?;
    Ok(())
}
fn l1_recover(dir: &Path) -> TestResult {
    let database = Database::open(db_config(dir).data_dir)?;
    assert_eq!(
        database.read_events(b"l1")?,
        vec![b"payload".to_vec()],
        "the acknowledged append must survive the cut at the shard-entry fence"
    );
    Ok(())
}
#[test]
fn pin_l1_append_fresh_shard() -> TestResult {
    run_pin(&PinCase {
        test_path: "fence::crash_pin_tests::pin_l1_append_fresh_shard",
        write: l1_write,
        recover: l1_recover,
    })
}

// --- L3-alone: established store, commit into a new prefix subdir ---------------

fn l3_write(dir: &Path) -> TestResult {
    let database = Database::create(db_config(dir))?;
    database.append(b"established".to_vec(), vec![b"a".to_vec()], 0)?;
    database.append(b"new-prefix".to_vec(), vec![b"b".to_vec()], 0)?;
    drop(database);
    journal::cut()?;
    Ok(())
}
fn l3_recover(dir: &Path) -> TestResult {
    let database = Database::open(db_config(dir).data_dir)?;
    assert_eq!(database.read_events(b"established")?, vec![b"a".to_vec()]);
    assert_eq!(
        database.read_events(b"new-prefix")?,
        vec![b"b".to_vec()],
        "a commit into a new prefix subdir must survive the cut at the store-root entry fence"
    );
    Ok(())
}
#[test]
fn pin_l3_alone_new_prefix() -> TestResult {
    run_pin(&PinCase {
        test_path: "fence::crash_pin_tests::pin_l3_alone_new_prefix",
        write: l3_write,
        recover: l3_recover,
    })
}

// --- L4: fresh containing dir via SnapshotRegistry AND CommitLog ----------------

fn l4_reg_path(dir: &Path) -> std::path::PathBuf {
    dir.join("snap").join("registry.bin")
}
fn l4_log_path(dir: &Path) -> std::path::PathBuf {
    dir.join("log").join("commit.log")
}
fn l4_write(dir: &Path) -> TestResult {
    let mut registry = SnapshotRegistry::open(l4_reg_path(dir))?;
    registry.name_at("snap", Hash::from_bytes([5; 32]), 5)?;
    let mut log = CommitLog::open(l4_log_path(dir))?;
    log.append(Hash::from_bytes([5; 32]), 5)?;
    drop((registry, log));
    journal::cut()?;
    Ok(())
}
fn l4_recover(dir: &Path) -> TestResult {
    let registry = SnapshotRegistry::open(l4_reg_path(dir))?;
    assert_eq!(
        registry.get("snap"),
        Some(Hash::from_bytes([5; 32])),
        "the named snapshot must survive the cut at the containing-dir fence"
    );
    let log = CommitLog::open(l4_log_path(dir))?;
    assert!(
        log.list()
            .iter()
            .any(|entry| entry.root_hash == Hash::from_bytes([5; 32])),
        "the appended commit-log record must survive the cut"
    );
    Ok(())
}
#[test]
fn pin_l4_metadata_open() -> TestResult {
    run_pin(&PinCase {
        test_path: "fence::crash_pin_tests::pin_l4_metadata_open",
        write: l4_write,
        recover: l4_recover,
    })
}

// --- primary: first commit after a fresh branch, L2+L3, refs L2-isolated -------

/// Store and refs sit under DIFFERENT durably-anchored parents so no later
/// unconditional refs-open fsync of a shared parent can clear the store's pending
/// entry and mask a patched-away `DiskStore` fence (r6 F1 L2-isolation).
fn primary_paths(dir: &Path) -> (std::path::PathBuf, std::path::PathBuf) {
    (
        dir.join("store-anchor").join("store"),
        dir.join("refs-anchor").join("refs"),
    )
}
fn primary_write(dir: &Path) -> TestResult {
    let (store_dir, refs_dir) = primary_paths(dir);
    std::fs::create_dir(store_dir.parent().ok_or("store parent")?)?;
    std::fs::create_dir(refs_dir.parent().ok_or("refs parent")?)?;

    let mut store = DiskStore::new(&store_dir)?;
    let empty = store.put(&Node::Leaf(LeafNode::new(Vec::new())?))?;
    let anchor = boxed(insert(
        &mut store,
        empty,
        b"base",
        b"history",
        TreePolicy::V1_DEFAULT,
    ))?;
    store.sync_dirty_dirs()?;

    let mut refs = BranchRefStore::open(&refs_dir)?;
    let registry = BranchRegistry::new();
    let branch = boxed(create_branch(
        "main",
        [(DEFAULT_SHARD_ID, anchor)],
        &mut refs,
        &registry,
        10,
    ))?;
    boxed(branch.put(DEFAULT_SHARD_ID, b"key", b"value"))?;
    boxed(commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: &[],
            timestamp: 20,
        },
        TreePolicy::V1_DEFAULT,
    ))?;
    drop((store, refs));
    journal::cut()?;
    Ok(())
}
fn primary_recover(dir: &Path) -> TestResult {
    let (store_dir, refs_dir) = primary_paths(dir);
    let store = DiskStore::new(&store_dir)?;
    let refs = BranchRefStore::open(&refs_dir)?;
    let registry = BranchRegistry::new();
    let reopened = boxed(open_branch("main", &refs, &registry))?;
    assert!(
        refs.get("main").is_some(),
        "the branch record must resolve after the cut"
    );
    assert_eq!(
        boxed(reopened.get(DEFAULT_SHARD_ID, &store, b"key"))?,
        Some(b"value".to_vec()),
        "the first commit's value must read back exactly after the cut (L2+L3)"
    );
    Ok(())
}
#[test]
fn pin_primary_first_branch_commit() -> TestResult {
    run_pin(&PinCase {
        test_path: "fence::crash_pin_tests::pin_primary_first_branch_commit",
        write: primary_write,
        recover: primary_recover,
    })
}

// --- Replace-positive: advance a durable ref again, cut after READY ------------

fn replace_write(dir: &Path) -> TestResult {
    let refs_dir = dir.join("refs");
    let mut refs = BranchRefStore::open(&refs_dir)?;
    refs.create(record("branch", 0x11))?;
    refs.advance(
        "branch",
        1,
        0,
        &[(0, Hash::from_bytes([0x22; 32]))],
        Vec::new(),
        2,
    )?;
    drop(refs);
    journal::cut()?;
    Ok(())
}
fn replace_recover(dir: &Path) -> TestResult {
    let refs = BranchRefStore::open(dir.join("refs"))?;
    let head = refs.get("branch").ok_or("branch record")?.shards[0].head;
    assert_eq!(
        head,
        Hash::from_bytes([0x22; 32]),
        "the advanced (NEW) record must survive the cut exactly (Replace-positive)"
    );
    Ok(())
}
#[test]
fn pin_replace_positive() -> TestResult {
    run_pin(&PinCase {
        test_path: "fence::crash_pin_tests::pin_replace_positive",
        write: replace_write,
        recover: replace_recover,
    })
}

// --- Replace-falsifier: LOSSY advance, cut restores the OLD record (evidence) --

fn replace_falsifier_write(dir: &Path) -> TestResult {
    let refs_dir = dir.join("refs");
    let mut refs = BranchRefStore::open(&refs_dir)?;
    refs.create(record("branch", 0x11))?; // durable head 0x11 (real fence)
    journal::set_lossy(true);
    refs.advance(
        "branch",
        1,
        0,
        &[(0, Hash::from_bytes([0x99; 32]))],
        Vec::new(),
        2,
    )?;
    journal::set_lossy(false);
    drop(refs);
    journal::cut()?; // restores the preimage: the OLD 0x11 record
    Ok(())
}
fn replace_falsifier_recover(dir: &Path) -> TestResult {
    let refs = BranchRefStore::open(dir.join("refs"))?;
    let head = refs.get("branch").ok_or("branch record")?.shards[0].head;
    assert_eq!(
        head,
        Hash::from_bytes([0x11; 32]),
        "FALSIFIER EVIDENCE: the unfenced advance was rewound to the OLD record (loss expressible)"
    );
    Ok(())
}
#[test]
fn pin_replace_falsifier_evidence() -> TestResult {
    run_pin(&PinCase {
        test_path: "fence::crash_pin_tests::pin_replace_falsifier_evidence",
        write: replace_falsifier_write,
        recover: replace_falsifier_recover,
    })
}