haematite 0.7.0

//! FENCE-CHAIN r8 R4b/R4e/R4f/R4g pins + R1a journal-coverage census.
//!
//! * R4b — fsync-neutrality via the OBSERVER (warm commit + observer open at ZERO
//!   directory fsyncs; the sweep clause is SUPERSEDED — no sweep exists).
//! * R4e — D1 refusal pins (missing parent → D0 remedy; WAL `NotFound` preserved;
//!   vanished metadata dir surfaces the owning error).
//! * R4f — compatibility pins (existing layouts succeed and fence; the store
//!   constructor observes exactly ONE parent sync on an existing store).
//! * R4g — open-mode fence pins (`Database::open` fences once; an injected
//!   open-mode fence failure refuses without removing anything).
//! * R1a — journal coverage naming BOTH native install sites.

use std::error::Error;

use super::journal;
use crate::branch::{
    BranchKind, BranchRefRecord, BranchRefStore, BranchShardRef, CommitLog, SnapshotRegistry,
};
use crate::db::{Database, DatabaseConfig, DatabaseError, ReadOnlyDatabase};
use crate::store::{DiskStore, NodeStore};
use crate::tree::{Hash, LeafNode, Node};
use crate::wal::{DurableWal, FsyncPolicy, WalError};

type TestResult = Result<(), Box<dyn Error>>;

fn config(data_dir: &std::path::Path) -> DatabaseConfig {
    DatabaseConfig {
        data_dir: data_dir.to_path_buf(),
        shard_count: 1,
        distributed: None,
        executor_threads: None,
    }
}

fn leaf(key: &[u8], value: &[u8]) -> Result<Node, Box<dyn Error>> {
    Ok(Node::Leaf(LeafNode::new(vec![(
        key.to_vec(),
        value.to_vec(),
    )])?))
}

fn record_at(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(),
    }
}

// --- R4b: fsync-neutrality (OBSERVER) -----------------------------------------

/// R4b: a warm commit creating NO new directory performs ZERO directory fsyncs.
#[test]
fn r4b_warm_commit_is_fsync_neutral() -> TestResult {
    let _guard = journal::test_guard();
    let shard = tempfile::tempdir()?;
    let mut store = DiskStore::with_cache_capacity(shard.path().join("store"), 64)?;

    // Warm the store: publish a node and fence it.
    store.put(&leaf(b"warm", b"value")?)?;
    store.sync_dirty_dirs()?;

    // A warm "commit" that publishes no NEW node/subdir: re-put the SAME node.
    journal::reset_observer();
    store.put(&leaf(b"warm", b"value")?)?;
    store.sync_dirty_dirs()?;

    assert_eq!(
        journal::total_sync_count(),
        0,
        "a warm commit creating no new directory must perform zero directory fsyncs"
    );
    Ok(())
}

/// R4b: a read-only observer open performs ZERO directory fsyncs (open-existing
/// UNFENCED mode).
#[test]
fn r4b_observer_open_is_fsync_neutral() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let data_dir = temp.path().join("db");
    let database = Database::create(config(&data_dir))?;
    database.append(b"k".to_vec(), vec![b"v".to_vec()], 0)?;
    drop(database);

    journal::reset_observer();
    let observer = ReadOnlyDatabase::open(&data_dir)?;
    let _root = observer.committed_root(0)?;

    assert_eq!(
        journal::total_sync_count(),
        0,
        "an observer open must perform zero directory fsyncs"
    );
    Ok(())
}

// --- R4e: D1 refusal pins ------------------------------------------------------

/// R4e: `Database::create` with a missing parent refuses with `NotFound`
/// preserved and the D0 remedy text present.
#[test]
fn r4e_create_missing_parent_carries_d0_remedy() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    // Parent `absent/` does not exist, so `absent/db` has no pre-existing parent.
    let data_dir = temp.path().join("absent").join("db");

    match Database::create(config(&data_dir)) {
        Err(DatabaseError::DirectoryCreate(error)) => {
            assert_eq!(
                error.kind(),
                std::io::ErrorKind::NotFound,
                "NotFound preserved"
            );
            assert!(
                error.to_string().contains("durably linked"),
                "the D0 remedy text must be present: {error}"
            );
        }
        other => return Err(format!("expected DirectoryCreate refusal, got {other:?}").into()),
    }
    Ok(())
}

/// R4e: `DurableWal::new` with a missing parent refuses via `WalError` with
/// `NotFound` PRESERVED and the D0 remedy text included (the remedy law applies to
/// WAL refusals too — no exception). The rewrite path shares `ensure_wal_parent`.
#[test]
fn r4e_wal_missing_parent_preserves_notfound_and_remedy() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let wal_path = temp.path().join("absent").join("shard.wal");

    match DurableWal::new(&wal_path, FsyncPolicy::CommitOnly) {
        Err(WalError::Io(error)) => {
            assert_eq!(
                error.kind(),
                std::io::ErrorKind::NotFound,
                "NotFound preserved"
            );
            assert!(
                error.to_string().contains("durably linked"),
                "the D0 remedy text must be present in the WAL refusal: {error}"
            );
        }
        other => return Err(format!("expected WalError::Io(NotFound), got {other:?}").into()),
    }
    Ok(())
}

/// R4e: a metadata write with a vanished containing directory surfaces the owning
/// error (no silent recreation).
#[test]
fn r4e_metadata_write_with_vanished_dir_surfaces_error() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let containing = temp.path().join("meta");
    let path = containing.join("registry.bin");
    let mut registry = SnapshotRegistry::open(&path)?;
    // Remove the containing dir behind the registry's back.
    std::fs::remove_dir_all(&containing)?;

    let outcome = registry.name_at("x", Hash::from_bytes([1; 32]), 5);
    assert!(
        outcome.is_err(),
        "a metadata write into a vanished containing dir must surface the owning error"
    );
    assert!(
        !containing.exists(),
        "the write must NOT silently recreate the vanished directory"
    );
    Ok(())
}

// --- R4f: compatibility pins ---------------------------------------------------

/// R4f: existing layouts (pre-made empty dirs) all open successfully and fence.
#[test]
fn r4f_existing_layouts_open_and_fence() -> TestResult {
    let _guard = journal::test_guard();

    // Database::create into a pre-made empty dir.
    let temp = tempfile::tempdir()?;
    let data_dir = temp.path().join("db");
    std::fs::create_dir(&data_dir)?;
    drop(Database::create(config(&data_dir))?);

    // BranchRefStore::open on an existing refs dir.
    let temp = tempfile::tempdir()?;
    let refs = temp.path().join("refs");
    std::fs::create_dir(&refs)?;
    drop(BranchRefStore::open(&refs)?);

    // Snapshot / commit-log open on existing layouts.
    let temp = tempfile::tempdir()?;
    drop(SnapshotRegistry::open(temp.path().join("registry.bin"))?);
    drop(CommitLog::open(temp.path().join("commit.log"))?);
    Ok(())
}

/// R4f: `DiskStore::with_cache_capacity` on a PRE-MADE existing store directory
/// succeeds and observes EXACTLY ONE normalized parent sync (`new` shares this
/// mode). D2: the fence is unconditional even though the store already existed.
#[test]
fn r4f_store_open_existing_observes_exactly_one_sync() -> TestResult {
    let _guard = journal::test_guard();
    let shard = tempfile::tempdir()?;
    let store_dir = shard.path().join("store");
    std::fs::create_dir(&store_dir)?;

    journal::reset_observer();
    let _store = DiskStore::with_cache_capacity(&store_dir, 16)?;

    assert_eq!(
        journal::sync_count(shard.path()),
        1,
        "opening an existing store must fence its entry exactly once"
    );
    assert_eq!(
        journal::total_sync_count(),
        1,
        "and no other directory fsync"
    );
    Ok(())
}

// --- R4g: open-mode fence pins -------------------------------------------------

/// R4g: `Database::open` fences the `data_dir` entry under the lock, and the
/// OBSERVER counts exactly one such fence.
#[test]
fn r4g_open_fences_data_dir_exactly_once() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let data_dir = temp.path().join("db");
    drop(Database::create(config(&data_dir))?);

    journal::reset_observer();
    let opened = Database::open(&data_dir)?;

    assert_eq!(
        journal::sync_count(temp.path()),
        1,
        "Database::open must fence the data_dir entry exactly once"
    );
    assert_eq!(
        journal::total_sync_count(),
        1,
        "and perform no other directory fsync at open"
    );
    drop(opened);
    Ok(())
}

/// R4g: an injected open-mode fence failure refuses without removing anything.
#[test]
fn r4g_open_fence_failure_removes_nothing() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let data_dir = temp.path().join("db");
    drop(Database::create(config(&data_dir))?);
    let config_before = std::fs::read(data_dir.join("config.json"))?;

    journal::arm_failpoint();
    let opened = Database::open(&data_dir);
    assert!(
        matches!(opened, Err(DatabaseError::DirectoryCreate(_))),
        "an injected open-mode fence failure must refuse, got {opened:?}"
    );
    assert!(
        data_dir.exists(),
        "a refused open must NEVER remove the existing directory"
    );
    assert_eq!(
        std::fs::read(data_dir.join("config.json"))?,
        config_before,
        "the existing database must survive byte-identically"
    );
    Ok(())
}

// --- R1a: journal coverage naming BOTH native install sites --------------------

/// R1a coverage (domain-seat sharpening 1): the journal covers BOTH native
/// install sites — `write_atomic_noclobber_unfenced` (create) AND
/// `write_atomic_unfenced` (cas-replace) — so neither slips through as the
/// unenumerated one. Proven by rewinding each under LOSSY: the create is REMOVED,
/// the cas-replace is RESTORED to its captured preimage.
#[test]
fn r1a_journal_covers_both_native_install_sites() -> TestResult {
    let _guard = journal::test_guard();

    // Site 1 — write_atomic_noclobber_unfenced (create). Rewinds by REMOVAL.
    let temp = tempfile::tempdir()?;
    let refs = temp.path().join("refs");
    let mut store = BranchRefStore::open(&refs)?;
    journal::reset();
    journal::set_lossy(true);
    store.create(record_at("branch", 0x11))?;
    journal::set_lossy(false);
    assert!(
        journal::pending_len() >= 1,
        "the create install must be journalled"
    );
    journal::cut()?;
    assert!(
        BranchRefStore::open(&refs)?.get("branch").is_none(),
        "the cut must REMOVE the unfenced create install (noclobber site)"
    );

    // Site 2 — write_atomic_unfenced (cas-replace). Rewinds by PREIMAGE RESTORE.
    let temp = tempfile::tempdir()?;
    let refs = temp.path().join("refs");
    let mut store = BranchRefStore::open(&refs)?;
    store.create(record_at("branch", 0x22))?; // durable head 0x22
    journal::reset();
    journal::set_lossy(true);
    store.advance(
        "branch",
        1,
        0,
        &[(0, Hash::from_bytes([0x33; 32]))],
        Vec::new(),
        2,
    )?;
    journal::set_lossy(false);
    assert!(
        journal::pending_len() >= 1,
        "the cas-replace install must be journalled"
    );
    journal::cut()?;
    let reopened = BranchRefStore::open(&refs)?;
    let head = reopened
        .get("branch")
        .ok_or("branch must still exist")?
        .shards[0]
        .head;
    assert_eq!(
        head,
        Hash::from_bytes([0x22; 32]),
        "the cut must RESTORE the cas-replace install's preimage (write_atomic_unfenced site)"
    );
    Ok(())
}