haematite 0.7.0

//! FENCE-CHAIN r8 R4a — dual-polarity walls for L1, L2, L3.
//!
//! Each wall exercises a real production fence site in BOTH polarities:
//!   * FAILPOINT — the directory-entry fsync fails, so the operation refuses with
//!     its typed error BEFORE any durable claim (the fence is load-bearing on the
//!     error path).
//!   * LOSSY + CUT — the fsync is silently suppressed (the "fence patched away"
//!     case), so the install's directory-entry mutation stays PENDING in the
//!     journal; the CUT then rewinds it, and the loss is observable — the exact
//!     assertion the R3 crash pin makes fails here in miniature. A positive
//!     control (real fence → nothing pending → CUT is a no-op → state intact)
//!     proves the LOSSY branch's loss is caused by the missing fence, not the CUT.

use std::error::Error;
use std::sync::Arc;
use std::time::Duration;

use beamr::module::ModuleRegistry;
use beamr::scheduler::{Scheduler, SchedulerConfig};

use super::journal;
use crate::shard::router::{ShardMode, ShardRouter, shard_dir};
use crate::store::{DiskStore, NodeStore, StoreError};
use crate::tree::{LeafNode, Node};

type TestResult = Result<(), Box<dyn Error>>;
const TIMEOUT: Duration = Duration::from_secs(5);

fn scheduler() -> Result<Arc<Scheduler>, Box<dyn Error>> {
    Scheduler::new(SchedulerConfig::default(), Arc::new(ModuleRegistry::new()))
        .map(Arc::new)
        .map_err(|message| -> Box<dyn Error> { message.into() })
}

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

// --- L1: router / `MaterialiseError` ------------------------------------------

/// L1 FAILPOINT polarity: an injected `data_dir` fence failure aborts
/// materialisation with `MaterialiseError` before the shard is usable.
#[test]
fn wall_l1_failpoint_refuses_materialisation() -> TestResult {
    let _guard = journal::test_guard();
    let dir = tempfile::tempdir()?;
    let router = ShardRouter::new(
        scheduler()?,
        dir.path(),
        4,
        crate::tree::TreePolicy::V1_DEFAULT,
        ShardMode::Create,
        crate::db::root_advance::RootAdvanceSeam::new(),
    )
    .ok_or("non-zero shard_count")?;

    journal::arm_failpoint();
    let outcome = router.handle_for_shard(0);

    assert!(
        outcome.is_err(),
        "an injected data_dir fence failure must refuse materialisation"
    );
    assert!(
        router.materialised_shard_ids().is_empty(),
        "a refused materialisation must leave no usable shard (no durable claim)"
    );
    router.shutdown_all(TIMEOUT);
    Ok(())
}

/// L1 LOSSY + CUT polarity: with the `data_dir` fence suppressed the `shard-{id}`
/// entry stays pending; the CUT rewinds it and the shard directory vanishes — the
/// silent-empty-shard the R3 L1 pin catches. The positive control shows a real
/// fence clears the obligation so the CUT is a no-op.
#[test]
fn wall_l1_lossy_cut_loses_shard_entry() -> TestResult {
    let _guard = journal::test_guard();
    let dir = tempfile::tempdir()?;
    let shard0 = shard_dir(dir.path(), 0);

    // Positive control: a real fence clears the pending shard-entry obligation.
    let router = ShardRouter::new(
        scheduler()?,
        dir.path(),
        4,
        crate::tree::TreePolicy::V1_DEFAULT,
        ShardMode::Create,
        crate::db::root_advance::RootAdvanceSeam::new(),
    )
    .ok_or("non-zero shard_count")?;
    router.handle_for_shard(0).map_err(|error| error.message)?;
    assert_eq!(
        journal::pending_len(),
        0,
        "a real data_dir fence must leave nothing pending"
    );
    router.shutdown_all(TIMEOUT);
    journal::cut()?;
    assert!(shard0.is_dir(), "a fenced shard entry survives the CUT");

    // Falsifier: the fence is patched away, so the shard entry is rewound.
    let dir = tempfile::tempdir()?;
    let shard0 = shard_dir(dir.path(), 0);
    let router = ShardRouter::new(
        scheduler()?,
        dir.path(),
        4,
        crate::tree::TreePolicy::V1_DEFAULT,
        ShardMode::Create,
        crate::db::root_advance::RootAdvanceSeam::new(),
    )
    .ok_or("non-zero shard_count")?;
    journal::set_lossy(true);
    router.handle_for_shard(0).map_err(|error| error.message)?;
    journal::set_lossy(false);
    assert!(
        journal::pending_len() >= 1,
        "an unfenced shard entry must remain pending in the journal"
    );
    router.shutdown_all(TIMEOUT);
    journal::cut()?;
    assert!(
        !shard0.exists(),
        "the CUT must rewind the unfenced shard-0 entry (silent empty shard)"
    );
    Ok(())
}

/// Coordinator sharpening 2: the L1 fence must preserve the create-OR-open shared
/// semantic — an `Open` of a NEVER-materialised shard directory must succeed
/// through the fence (lazy reopen), not assume a create.
#[test]
fn wall_l1_open_of_never_materialised_shard_passes_the_fence() -> TestResult {
    let _guard = journal::test_guard();
    let dir = tempfile::tempdir()?;
    // Open mode over a data_dir whose shard-0 directory was never created.
    let router = ShardRouter::new(
        scheduler()?,
        dir.path(),
        4,
        crate::tree::TreePolicy::V1_DEFAULT,
        ShardMode::Open,
        crate::db::root_advance::RootAdvanceSeam::new(),
    )
    .ok_or("non-zero shard_count")?;
    assert!(!shard_dir(dir.path(), 0).exists());

    router
        .handle_for_shard(0)
        .map_err(|error| -> Box<dyn Error> { error.message.into() })?;

    assert!(
        shard_dir(dir.path(), 0).is_dir(),
        "opening a never-materialised shard must create and fence its entry"
    );
    assert_eq!(router.materialised_shard_ids(), vec![0]);
    router.shutdown_all(TIMEOUT);
    Ok(())
}

// --- L2: public DiskStore construction / `StoreError::Io` ----------------------

/// L2 FAILPOINT polarity through `with_cache_capacity` — the public constructor
/// that owns `ensure_directory`, so the wall covers both public doors. An injected
/// store-root fence failure refuses with `StoreError::Io`.
#[test]
fn wall_l2_failpoint_refuses_store_construction() -> TestResult {
    let _guard = journal::test_guard();
    let shard = tempfile::tempdir()?;
    let store_dir = shard.path().join("store");

    journal::arm_failpoint();
    let outcome = DiskStore::with_cache_capacity(&store_dir, 16);

    assert!(
        matches!(outcome, Err(StoreError::Io(_))),
        "an injected store-root fence failure must refuse with StoreError::Io, got {outcome:?}"
    );
    Ok(())
}

/// L2 LOSSY + CUT polarity: a suppressed store-root fence leaves the `store` entry
/// pending; the CUT rewinds it. Positive control: a real fence clears it.
#[test]
fn wall_l2_lossy_cut_loses_store_root_entry() -> TestResult {
    let _guard = journal::test_guard();

    let shard = tempfile::tempdir()?;
    let store_dir = shard.path().join("store");
    let _store = DiskStore::with_cache_capacity(&store_dir, 16)?;
    assert_eq!(
        journal::pending_len(),
        0,
        "a real store-root fence leaves nothing pending"
    );
    journal::cut()?;
    assert!(store_dir.is_dir(), "a fenced store root survives the CUT");

    let shard = tempfile::tempdir()?;
    let store_dir = shard.path().join("store");
    journal::set_lossy(true);
    let _store = DiskStore::with_cache_capacity(&store_dir, 16)?;
    journal::set_lossy(false);
    assert!(
        journal::pending_len() >= 1,
        "an unfenced store root must remain pending"
    );
    journal::cut()?;
    assert!(
        !store_dir.exists(),
        "the CUT must rewind the unfenced store-root entry"
    );
    Ok(())
}

// --- L3: barrier discipline ----------------------------------------------------

/// L3 FAILPOINT polarity: an injected barrier fence failure refuses, and the
/// store-root entry-fence obligation is RETAINED for the retry (R4c) — never
/// silently dropped.
#[test]
fn wall_l3_failpoint_refuses_and_retains() -> TestResult {
    let _guard = journal::test_guard();
    let shard = tempfile::tempdir()?;
    let store_dir = shard.path().join("store");
    let mut store = DiskStore::with_cache_capacity(&store_dir, 64)?;
    journal::reset(); // ignore the construction fence; isolate the barrier

    store.put(&leaf(b"k", b"v")?)?;
    assert_eq!(
        store.pending_entry_fences().len(),
        1,
        "a new prefix subdir must owe a store-root entry fence"
    );

    journal::arm_failpoint();
    let outcome = store.sync_dirty_dirs();
    assert!(
        matches!(outcome, Err(StoreError::Io(_))),
        "an injected barrier fence failure must refuse, got {outcome:?}"
    );
    assert_eq!(
        store.pending_entry_fences().len(),
        1,
        "a failed barrier must RETAIN the un-fenced store-root obligation (R4c)"
    );

    // The retry (no failpoint) discharges the retained obligation.
    store.sync_dirty_dirs()?;
    assert!(
        store.pending_entry_fences().is_empty(),
        "the retry must clear it"
    );
    Ok(())
}

/// L3 LOSSY + CUT polarity: a suppressed barrier fence leaves the new prefix
/// subdir's entry pending; the CUT rewinds the subdir (and the node beneath it).
/// Positive control: a real barrier clears it and the node survives.
#[test]
fn wall_l3_lossy_cut_loses_prefix_subdir_entry() -> TestResult {
    let _guard = journal::test_guard();

    // Positive control.
    let shard = tempfile::tempdir()?;
    let store_dir = shard.path().join("store");
    let mut store = DiskStore::with_cache_capacity(&store_dir, 64)?;
    journal::reset();
    let hash = store.put(&leaf(b"live", b"value")?)?;
    let node_path = store_dir_node_path(&store_dir, &hash);
    store.sync_dirty_dirs()?;
    assert_eq!(
        journal::pending_len(),
        0,
        "a real barrier leaves nothing pending"
    );
    journal::cut()?;
    assert!(
        node_path.exists(),
        "a fenced prefix subdir survives the CUT"
    );

    // Falsifier.
    let shard = tempfile::tempdir()?;
    let store_dir = shard.path().join("store");
    let mut store = DiskStore::with_cache_capacity(&store_dir, 64)?;
    journal::reset();
    let hash = store.put(&leaf(b"doomed", b"value")?)?;
    let node_path = store_dir_node_path(&store_dir, &hash);
    let prefix_dir = node_path
        .parent()
        .ok_or("node has a prefix dir")?
        .to_path_buf();
    journal::set_lossy(true);
    store.sync_dirty_dirs()?;
    journal::set_lossy(false);
    assert!(
        journal::pending_len() >= 1,
        "an unfenced prefix subdir must remain pending"
    );
    journal::cut()?;
    assert!(
        !prefix_dir.exists(),
        "the CUT must rewind the unfenced prefix subdir and the node beneath it"
    );
    Ok(())
}

/// The on-disk path a hash lands at under a store root (mirrors `DiskStore`'s
/// `node_path`: two-hex-char prefix subdir + remaining hex file name).
fn store_dir_node_path(
    store_dir: &std::path::Path,
    hash: &crate::tree::Hash,
) -> std::path::PathBuf {
    let hex = hash.to_string();
    let (prefix, file_name) = hex.split_at(2);
    store_dir.join(prefix).join(file_name)
}