haematite 0.7.0

//! COMMIT-COLLAPSE §2/§11 dirty-signal + restart-protocol integration tests, driven
//! through a real [`Database`]. The cell is read via the test-only
//! [`Database::commit_state_snapshot`], which reads the SAME process-lifetime cell
//! the shard actor writes.
//!
//! The second-lander seam integration (clean-skip emits no gen; a dirty commit
//! emits exactly one tell with the cell's gen) lives in `db/root_advance_tests.rs`
//! (the `advance_invariant_*`, `concurrent_appends_*`, and `told_advance` pins);
//! those stay green under the §8 absorb, which is the absorb's acceptance test.

use std::error::Error;

use crate::db::{Database, DatabaseConfig};
use crate::shard::commit_state::CommitSnapshot;

type BoxError = Box<dyn Error>;

fn single_shard_db() -> Result<(tempfile::TempDir, Database), BoxError> {
    let dir = tempfile::tempdir()?;
    let db = Database::create(DatabaseConfig {
        data_dir: dir.path().join("db"),
        shard_count: 1,
        distributed: None,
        executor_threads: None,
    })?;
    Ok((dir, db))
}

/// The §5 classification predicate, read off a detached snapshot.
fn is_dirty(snapshot: &CommitSnapshot) -> bool {
    snapshot.recovering
        || snapshot.unreconciled
        || snapshot.committed_gen != snapshot.dirty_gen
        || snapshot.root.is_none()
}

/// A buffered put dirties the shard; a commit cleans it and publishes the root.
#[test]
fn put_dirties_the_shard_and_commit_cleans_it() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.put(b"k".to_vec(), b"v".to_vec())?;
    let dirty = db.commit_state_snapshot(0);
    assert!(is_dirty(&dirty), "a buffered put must classify dirty");

    db.commit()?;
    let clean = db.commit_state_snapshot(0);
    assert!(!is_dirty(&clean), "commit must classify clean");
    assert!(clean.root.is_some(), "commit publishes the root");
    assert_eq!(clean.committed_gen, clean.dirty_gen);
    Ok(())
}

/// A stamped-tombstone delete is a buffered put => dirty (the §2.1 delete row).
#[test]
fn delete_dirties_the_shard() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.put(b"k".to_vec(), b"v".to_vec())?;
    db.commit()?;
    assert!(!is_dirty(&db.commit_state_snapshot(0)));

    db.delete(b"k".to_vec())?;
    assert!(
        is_dirty(&db.commit_state_snapshot(0)),
        "a buffered tombstone must classify dirty"
    );
    Ok(())
}

/// `append` stages AND commits in one command: after it the shard is clean at a new
/// root, and the cell's `advance_gen` bumped by one.
#[test]
fn append_advances_and_leaves_clean() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.append(b"stream".to_vec(), vec![b"e".to_vec()], 0)?;
    let snapshot = db.commit_state_snapshot(0);
    assert!(!is_dirty(&snapshot), "append commits => clean");
    assert!(snapshot.root.is_some());
    assert_eq!(snapshot.advance_gen, 1, "append advanced the root once");
    Ok(())
}

/// `cas` stages AND commits in one command: clean afterwards, one advance.
#[test]
fn cas_advances_and_leaves_clean() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.cas(b"c".to_vec(), None, 1)?;
    let snapshot = db.commit_state_snapshot(0);
    assert!(!is_dirty(&snapshot), "cas commits => clean");
    assert_eq!(snapshot.advance_gen, 1);
    Ok(())
}

/// An empty commit on a marker-bearing clean shard stays clean and does not advance
/// (the cell precursor to the §4 clean fast path and the §8 clean-skip).
#[test]
fn empty_commit_stays_clean_without_advancing() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.put(b"k".to_vec(), b"v".to_vec())?;
    db.commit()?;
    let advance_after_first = db.commit_state_snapshot(0).advance_gen;

    db.commit()?; // empty
    let snapshot = db.commit_state_snapshot(0);
    assert!(!is_dirty(&snapshot));
    assert_eq!(
        snapshot.advance_gen, advance_after_first,
        "an empty commit does not bump advance_gen"
    );
    Ok(())
}

/// Materialising a shard runs the restart protocol once: incarnation reaches 1.
#[test]
fn materialise_runs_restart_protocol_once() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.put(b"k".to_vec(), b"v".to_vec())?; // materialises shard 0
    assert_eq!(
        db.commit_state_snapshot(0).incarnation,
        1,
        "first spawn bumps the incarnation to 1"
    );
    Ok(())
}

/// Recovery row (§2.1) + restart protocol step 3: a WAL-durable but uncommitted
/// buffer recovers DIRTY on reopen, and the next commit persists it and cleans it.
#[test]
fn recovery_replay_reopens_dirty_then_commits() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    {
        let db = Database::create(DatabaseConfig {
            data_dir: data_dir.clone(),
            shard_count: 1,
            distributed: None,
            executor_threads: None,
        })?;
        db.put(b"k".to_vec(), b"v".to_vec())?; // durable in the WAL, NOT committed
        // drop without committing
    }
    let reopened = Database::open(&data_dir)?;
    // Materialise shard 0 (a read touches it), which runs recovery + the restart
    // protocol: the recovered buffer is non-empty, so the cell publishes DIRTY.
    assert_eq!(reopened.get(b"k")?, Some(b"v".to_vec()));
    let recovered = reopened.commit_state_snapshot(0);
    assert!(
        is_dirty(&recovered),
        "a recovered non-empty buffer must classify dirty"
    );
    assert_eq!(recovered.incarnation, 1, "the reopened actor spawned once");

    reopened.commit()?;
    assert!(
        !is_dirty(&reopened.commit_state_snapshot(0)),
        "committing the recovered buffer cleans the shard"
    );
    Ok(())
}

// ---------------------------------------------------------------------------
// COMMIT-COLLAPSE §11 (deferred item 1): rollback/restore matrix ×2.
//
// Every §2.1 rollback/restore site must republish the classification the RESTORED
// buffer implies: from a previously-CLEAN shard a rejected op leaves it clean
// (zero-dirty commit stays zero-work); from an already-DIRTY shard a rejected op
// leaves it dirty AND does not lose the prior buffered mutation. The failure is a
// natural precondition REJECTION (CAS expected-mismatch, append seq-conflict) —
// the same restore path a storage failure would take — so no failpoint is needed.
// shard_count = 1 so the setup write and the rejected op share shard 0.
// ---------------------------------------------------------------------------

/// CAS rejection (the single-key `rollback_staged` site) from a CLEAN shard leaves
/// it clean: the staged CAS write is rolled back, the buffer is empty again, the
/// cell republishes clean at the committed root.
#[test]
fn cas_rejection_from_clean_stays_clean() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.cas(b"c".to_vec(), None, 1)?; // create + commit counter => clean baseline
    assert!(!is_dirty(&db.commit_state_snapshot(0)), "clean baseline");

    // Wrong `expected` (actual is 1) => rejection => rollback.
    assert!(
        db.cas(b"c".to_vec(), Some(99), 2).is_err(),
        "CAS is rejected"
    );
    assert!(
        !is_dirty(&db.commit_state_snapshot(0)),
        "a rejected CAS restores an empty buffer => shard stays CLEAN"
    );
    // Zero-dirty commit stays zero-work: the counter is unchanged (CAS with the
    // true prior value 1 succeeds, proving the rejected CAS did not move it).
    db.commit()?;
    assert!(
        db.cas(b"c".to_vec(), Some(1), 5).is_ok(),
        "counter still 1 — the rejected CAS left it untouched"
    );
    Ok(())
}

/// CAS rejection from an already-DIRTY shard leaves it dirty and does not lose the
/// prior buffered put.
#[test]
fn cas_rejection_from_dirty_stays_dirty_and_preserves_prior() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.put(b"k".to_vec(), b"v".to_vec())?; // buffered, uncommitted => dirty
    assert!(is_dirty(&db.commit_state_snapshot(0)), "dirty baseline");

    assert!(
        db.cas(b"c".to_vec(), Some(99), 2).is_err(),
        "CAS on a missing counter is rejected"
    );
    assert!(
        is_dirty(&db.commit_state_snapshot(0)),
        "a rejected CAS leaves the pre-existing buffered put => shard stays DIRTY"
    );
    db.commit()?;
    assert_eq!(
        db.get(b"k")?,
        Some(b"v".to_vec()),
        "the prior buffered put is not lost by the rejected CAS"
    );
    Ok(())
}

/// Append rejection (the all-or-nothing `previous_buffer` restore site) from a CLEAN
/// shard leaves it clean.
#[test]
fn append_rejection_from_clean_stays_clean() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.append(b"s".to_vec(), vec![b"e0".to_vec()], 0)?; // stream now at seq 1, clean
    assert!(!is_dirty(&db.commit_state_snapshot(0)), "clean baseline");

    // Multi-entry append with a stale expected_seq (0, but the stream is at 1) =>
    // whole-batch rejection => previous_buffer restore.
    assert!(
        db.append(b"s".to_vec(), vec![b"e1".to_vec(), b"e2".to_vec()], 0)
            .is_err(),
        "a stale-seq append is rejected"
    );
    assert!(
        !is_dirty(&db.commit_state_snapshot(0)),
        "a rejected append restores the prior buffer => shard stays CLEAN"
    );
    assert_eq!(
        db.read_events(b"s")?,
        vec![b"e0".to_vec()],
        "no partial entries leaked from the rejected multi-entry append"
    );
    Ok(())
}

/// Append rejection from an already-DIRTY shard leaves it dirty and preserves the
/// prior buffered put.
#[test]
fn append_rejection_from_dirty_stays_dirty_and_preserves_prior() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    db.append(b"s".to_vec(), vec![b"e0".to_vec()], 0)?; // stream at seq 1, clean
    db.put(b"k".to_vec(), b"v".to_vec())?; // buffered put => dirty
    assert!(is_dirty(&db.commit_state_snapshot(0)), "dirty baseline");

    assert!(
        db.append(b"s".to_vec(), vec![b"e1".to_vec(), b"e2".to_vec()], 0)
            .is_err(),
        "stale-seq append is rejected"
    );
    assert!(
        is_dirty(&db.commit_state_snapshot(0)),
        "a rejected append leaves the buffered put => shard stays DIRTY"
    );
    db.commit()?;
    assert_eq!(
        db.get(b"k")?,
        Some(b"v".to_vec()),
        "the prior buffered put is not lost by the rejected append"
    );
    assert_eq!(
        db.read_events(b"s")?,
        vec![b"e0".to_vec()],
        "the stream is unchanged by the rejected append"
    );
    Ok(())
}

/// Restart case (b), in-process expression (§11 M2): a durable committed marker with
/// its cell publication "pending" at process end converges on reopen. After a commit
/// persists the marker root, the database is dropped; reopening re-runs the factory
/// boot (`begin_recovery` → recover → `publish_recovered`), which reads the marker and
/// publishes CLEAN at the recovered root — no stale-root window — with incarnation 1,
/// the data intact, and a following commit doing zero rework (clean skip returns the
/// recovered root).
///
/// In-process limit: a genuine OS-kill-then-same-`Arc` respawn (incarnation 1→2 on
/// the cached cell) is NOT expressible here — the router does not wire supervised
/// actor restart (CORE-008, deferred), and adding that production surface purely for
/// the test is out of scope. The reopen path exercises the IDENTICAL factory boot
/// and cell-publish code a respawn would run; the same-`Arc` incarnation continuity
/// is pinned separately by the cell unit tests + `commit_state_cell_has_stable_
/// pointer_identity`.
#[test]
fn restart_marker_durable_reopens_clean_and_converged() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let committed_root;
    {
        let db = Database::create(DatabaseConfig {
            data_dir: data_dir.clone(),
            shard_count: 1,
            distributed: None,
            executor_threads: None,
        })?;
        db.put(b"k".to_vec(), b"v".to_vec())?;
        committed_root = *db.commit()?.get(&0).ok_or("shard 0 committed root")?; // marker durable
    } // drop: the in-memory cell is gone, the durable marker remains

    let reopened = Database::open(&data_dir)?;
    assert_eq!(
        reopened.get(b"k")?,
        Some(b"v".to_vec()),
        "the committed data is intact after reopen (marker recovered)"
    );
    let recovered = reopened.commit_state_snapshot(0);
    assert!(
        !is_dirty(&recovered),
        "a shard with a durable marker recovers CLEAN (no stale-root window)"
    );
    assert_eq!(
        recovered.root,
        Some(committed_root),
        "the recovered cell carries the durable marker root"
    );
    assert_eq!(recovered.incarnation, 1, "the reopened actor spawned once");

    // Convergence: a following commit is a clean skip that returns the recovered
    // root — zero rework.
    let jobs_before = reopened.executor().dispatched_jobs();
    let roots = reopened.commit()?;
    assert_eq!(
        *roots.get(&0).ok_or("shard 0 root")?,
        committed_root,
        "the next commit returns the recovered marker root"
    );
    assert_eq!(
        reopened.executor().dispatched_jobs(),
        jobs_before,
        "convergence is a clean skip — no job dispatched"
    );
    Ok(())
}

/// `advance_gen` is monotonic and gap-free across a run of advancing commits, and
/// unchanged by the interleaved empty commits.
#[test]
fn advance_gen_is_monotonic_across_commits() -> Result<(), BoxError> {
    let (_dir, db) = single_shard_db()?;
    for round in 0..5_u32 {
        db.put(
            format!("k{round}").into_bytes(),
            round.to_be_bytes().to_vec(),
        )?;
        db.commit()?;
        db.commit()?; // empty — must not bump
        assert_eq!(
            db.commit_state_snapshot(0).advance_gen,
            u64::from(round) + 1,
            "advance_gen bumps exactly once per advancing commit"
        );
    }
    Ok(())
}