haematite 0.7.0

//! FENCE-CHAIN r8 R4d — causal-retry pins.
//!
//! Each pin injects a one-shot fence failure, observes the typed refusal, then
//! RETRIES the same operation against the SAME directory and observes success —
//! the fence failure left the world in a state a fresh attempt recovers from
//! (causality), and the retry re-runs the unconditional fence rather than skipping
//! it. The ownership-transfer half of R4d lives in
//! `db/ownership_transfer_tests.rs` (it needs the db-internal F4 seam).

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

use crate::branch::{BranchRefStore, SnapshotRegistry};
use crate::db::{Database, DatabaseConfig};
use crate::shard::router::{ShardMode, ShardRouter, shard_dir};

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

use super::journal;

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

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 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() })
}

/// R4d: `Database::create` — fail the `data_dir` fence once, then retry the same
/// directory to success.
#[test]
fn r4d_causal_retry_data_dir() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let data_dir = temp.path().join("db");

    journal::arm_failpoint();
    assert!(
        Database::create(config(&data_dir)).is_err(),
        "the injected data_dir fence failure must refuse the first create"
    );

    // Retry against the same directory: the unconditional fence runs (no failpoint)
    // and the create succeeds.
    let database = Database::create(config(&data_dir))?;
    assert!(data_dir.join("config.json").exists());
    drop(database);
    Ok(())
}

/// R4d: `BranchRefStore::open` — fail the refs-dir fence once, then retry.
#[test]
fn r4d_causal_retry_refs_dir() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let refs = temp.path().join("refs");

    journal::arm_failpoint();
    assert!(
        BranchRefStore::open(&refs).is_err(),
        "the injected refs-dir fence failure must refuse the first open"
    );
    drop(BranchRefStore::open(&refs)?);
    Ok(())
}

/// R4d: `SnapshotRegistry::open` — fail the containing-dir fence once, then retry.
#[test]
fn r4d_causal_retry_snapshot_dir() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let path = temp.path().join("meta").join("registry.bin");

    journal::arm_failpoint();
    assert!(
        SnapshotRegistry::open(&path).is_err(),
        "the injected containing-dir fence failure must refuse the first open"
    );
    drop(SnapshotRegistry::open(&path)?);
    Ok(())
}

/// R4d: L1 materialisation — fail the `data_dir` fence once, then retry the SAME
/// router and shard to success (the retry re-fences).
#[test]
fn r4d_causal_retry_l1_materialise() -> TestResult {
    let _guard = journal::test_guard();
    let temp = tempfile::tempdir()?;
    let router = ShardRouter::new(
        scheduler()?,
        temp.path(),
        4,
        crate::tree::TreePolicy::V1_DEFAULT,
        ShardMode::Create,
        crate::db::root_advance::RootAdvanceSeam::new(),
    )
    .ok_or("non-zero shard_count")?;

    journal::arm_failpoint();
    assert!(
        router.handle_for_shard(0).is_err(),
        "the injected data_dir fence failure must refuse the first materialisation"
    );

    router
        .handle_for_shard(0)
        .map_err(|error| -> Box<dyn Error> { error.message.into() })?;
    assert!(shard_dir(temp.path(), 0).is_dir());
    assert_eq!(router.materialised_shard_ids(), vec![0]);
    router.shutdown_all(TIMEOUT);
    Ok(())
}