haematite 0.7.0

//! Migration fence crash-pins (§9, r2 B2): a crash injected at EVERY step
//! boundary leaves the directory fenced — a normal writer/observer open refuses
//! [`DatabaseError::MigrationInProgress`], and an old binary reads
//! `format_version = 2` (so it would refuse `FormatVersionTooNew`) — and a rerun
//! resumes idempotently, converging to the identical roots.
//!
//! Crash injection mirrors the house in-process idiom (`branch/crash_window_
//! tests.rs`): perform every durable write up to the boundary, then STOP (the
//! writer lock drops exactly as a killed process would release it), then cold-
//! reopen and assert. The boundary is named by the same string the orchestrator
//! checks, so each named test maps 1:1 to a step boundary.

use std::error::Error;

use crate::db::{Database, DatabaseError, ReadOnlyDatabase};
use crate::tree::Hash;

use super::test_support::{
    Entries, Res, build_v1_branch, build_v1_fixture, format_version, fresh_build, is_fenced,
    oversized, oversized_shards, read_all_roots, read_branch_head, v2_default,
};
use super::{MigrateOptions, MigrationOutcome, migrate_chunking, migrate_with_crash};

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

/// The fresh-build canonical v2 root of every shard — the convergence target a
/// resumed forward migration must reach.
fn v2_canonical(shards: &[Entries]) -> Res<Vec<Option<Hash>>> {
    shards
        .iter()
        .map(|entries| fresh_build(entries, v2_default()).map(Some))
        .collect()
}

/// Assert the directory is fenced: writer open refuses `MigrationInProgress`,
/// and the on-disk `format_version` is 2 (what an old binary reads and refuses).
fn assert_fenced_refusal(data: &std::path::Path) -> TestResult {
    assert!(is_fenced(data)?, "the fence must be present in config.json");
    assert!(
        matches!(
            Database::open(data),
            Err(DatabaseError::MigrationInProgress { .. })
        ),
        "a normal writer open must refuse a fenced directory with MigrationInProgress"
    );
    assert_eq!(
        format_version(data)?,
        Some(2),
        "the stamp stays 2 while fenced — an old binary refuses FormatVersionTooNew"
    );
    Ok(())
}

/// Forward crash at `boundary`: fence holds, then a rerun resumes and converges
/// to the canonical v2 roots.
fn forward_crash_resumes(boundary: &str) -> TestResult {
    let dir = tempfile::tempdir()?;
    let data = dir.path().join("db");
    let shards = oversized_shards(2, 3);
    build_v1_fixture(&data, &shards)?;
    let canonical = v2_canonical(&shards)?;

    let stopped = migrate_with_crash(&MigrateOptions::migrate(data.clone()), boundary)?;
    assert!(
        stopped.is_none(),
        "{boundary}: the crash must stop the run early"
    );
    assert_fenced_refusal(&data)?;

    let report = migrate_chunking(&MigrateOptions::migrate(data.clone()))?;
    assert_eq!(
        report.outcome,
        MigrationOutcome::MigratedToV2,
        "{boundary}: the rerun must complete the forward migration"
    );
    assert!(
        !is_fenced(&data)?,
        "{boundary}: the resumed run removes the fence"
    );
    assert_eq!(
        read_all_roots(&data, shards.len())?,
        canonical,
        "{boundary}: the resumed run must converge to the canonical v2 roots"
    );
    Ok(())
}

#[test]
fn crash_after_forward_fence_resumes() -> TestResult {
    forward_crash_resumes("after_forward_fence")
}

#[test]
fn crash_after_first_shard_resumes() -> TestResult {
    forward_crash_resumes("after_shard_0")
}

#[test]
fn crash_after_last_shard_resumes() -> TestResult {
    forward_crash_resumes("after_shard_1")
}

#[test]
fn crash_after_all_shards_resumes() -> TestResult {
    forward_crash_resumes("after_all_shards")
}

#[test]
fn crash_before_fence_removal_resumes() -> TestResult {
    forward_crash_resumes("before_fence_removal")
}

/// Abort crash at `boundary`: reach a completed v2 directory, start an abort
/// that crashes at `boundary`, assert the fence holds (format STAYS 2), then a
/// rerun resumes the abort and converges to the byte-identical v1 roots.
fn abort_crash_resumes(boundary: &str) -> TestResult {
    let dir = tempfile::tempdir()?;
    let data = dir.path().join("db");
    let shards = oversized_shards(2, 3);
    let v1_roots = build_v1_fixture(&data, &shards)?;
    migrate_chunking(&MigrateOptions::migrate(data.clone()))?;

    let stopped = migrate_with_crash(&MigrateOptions::abort(data.clone()), boundary)?;
    assert!(
        stopped.is_none(),
        "{boundary}: the crash must stop the abort early"
    );
    assert_fenced_refusal(&data)?;

    let report = migrate_chunking(&MigrateOptions::abort(data.clone()))?;
    assert_eq!(
        report.outcome,
        MigrationOutcome::AbortedToV1,
        "{boundary}: the rerun must complete the abort"
    );
    assert!(
        !is_fenced(&data)?,
        "{boundary}: the resumed abort removes the fence"
    );
    assert_eq!(
        format_version(&data)?,
        Some(1),
        "{boundary}: the format flips to 1 last"
    );
    assert_eq!(
        read_all_roots(&data, shards.len())?,
        v1_roots.iter().copied().map(Some).collect::<Vec<_>>(),
        "{boundary}: the resumed abort must converge to the byte-identical v1 roots"
    );
    Ok(())
}

#[test]
fn crash_after_abort_flip_resumes() -> TestResult {
    abort_crash_resumes("after_abort_flip")
}

#[test]
fn crash_after_first_abort_shard_resumes() -> TestResult {
    abort_crash_resumes("after_abort_shard_0")
}

#[test]
fn crash_before_v1_finalize_resumes() -> TestResult {
    abort_crash_resumes("before_v1_finalize")
}

/// Step 4 crash boundary: a crash AFTER a branch head is advanced leaves the
/// fence up; a rerun resumes and converges (shards AND the branch head reach
/// their canonical v2 shape).
#[test]
fn crash_after_branch_head_resumes() -> TestResult {
    let dir = tempfile::tempdir()?;
    let data = dir.path().join("db");
    let shards = oversized_shards(2, 3);
    build_v1_fixture(&data, &shards)?;
    let puts: Entries = vec![
        (b"branch-0".to_vec(), oversized(50)),
        (b"branch-1".to_vec(), oversized(51)),
    ];
    build_v1_branch(&data, 0, &puts)?;

    let stopped = migrate_with_crash(
        &MigrateOptions::migrate(data.clone()),
        "after_branch_head_0",
    )?;
    assert!(
        stopped.is_none(),
        "the crash must stop the run after the branch head"
    );
    assert_fenced_refusal(&data)?;

    let report = migrate_chunking(&MigrateOptions::migrate(data.clone()))?;
    assert_eq!(report.outcome, MigrationOutcome::MigratedToV2);
    assert_eq!(read_all_roots(&data, shards.len())?, v2_canonical(&shards)?);
    assert_eq!(
        read_branch_head(&data, "main", 0)?,
        fresh_build(&puts, v2_default())?,
        "the resumed run must converge the branch head to canonical v2"
    );
    Ok(())
}

/// §4.2 step 8: an OBSERVER open of a fenced directory also refuses — serving a
/// half-migrated shard set as a coherent database is exactly what the fence
/// prevents.
#[test]
fn observer_open_refuses_a_fenced_directory() -> TestResult {
    let dir = tempfile::tempdir()?;
    let data = dir.path().join("db");
    build_v1_fixture(&data, &oversized_shards(2, 3))?;
    let stopped = migrate_with_crash(&MigrateOptions::migrate(data.clone()), "after_shard_0")?;
    assert!(stopped.is_none());

    assert!(
        ReadOnlyDatabase::open(&data).is_err(),
        "a read-only observer must refuse a fenced directory (§4.2 step 8)"
    );
    Ok(())
}