haematite 0.7.0

//! Unit tests for the COMMIT-COLLAPSE §2.2 commit-state cell.

use super::ShardCommitState;
use crate::tree::Hash;

fn hash(byte: u8) -> Hash {
    Hash::of(&[byte])
}

/// A fresh cell is marker-less with equal gens and zero incarnation — the restart
/// protocol overwrites everything but `advance_gen` before the first command.
#[test]
fn initial_snapshot_is_markerless_and_clean_by_gens() {
    let cell = ShardCommitState::new();
    let snapshot = cell.snapshot();
    assert_eq!(snapshot.root, None);
    assert_eq!(snapshot.dirty_gen, snapshot.committed_gen);
    assert_eq!(snapshot.incarnation, 0);
    assert!(!snapshot.recovering);
    assert!(!snapshot.unreconciled);
    assert_eq!(snapshot.advance_gen, 0);
}

/// Restart protocol step 1: recovering is raised and the incarnation bumps.
#[test]
fn begin_recovery_raises_recovering_and_bumps_incarnation() {
    let cell = ShardCommitState::new();
    cell.begin_recovery();
    let first = cell.snapshot();
    assert!(first.recovering);
    assert_eq!(first.incarnation, 1);
    cell.publish_recovered(Some(hash(1)), true);
    cell.begin_recovery();
    assert_eq!(cell.snapshot().incarnation, 2, "each (re)spawn bumps once");
}

/// Restart protocol step 3: a recovered EMPTY buffer publishes clean at the root;
/// a recovered NON-empty buffer publishes dirty. `advance_gen` is preserved.
#[test]
fn publish_recovered_sets_dirty_iff_buffer_nonempty_and_preserves_advance_gen() {
    let cell = ShardCommitState::new();
    // Advance once so there is a non-zero advance_gen to preserve.
    assert_eq!(cell.publish_committed(Some(hash(9)), true), Some(1));

    cell.begin_recovery();
    cell.publish_recovered(Some(hash(1)), true);
    let clean = cell.snapshot();
    assert!(!clean.recovering);
    assert_eq!(clean.root, Some(hash(1)));
    assert_eq!(
        clean.committed_gen, clean.dirty_gen,
        "empty buffer => clean"
    );
    assert_eq!(clean.advance_gen, 1, "advance_gen survives the incarnation");

    cell.begin_recovery();
    cell.publish_recovered(Some(hash(2)), false);
    let dirty = cell.snapshot();
    assert_ne!(
        dirty.committed_gen, dirty.dirty_gen,
        "non-empty recovered buffer => dirty"
    );
    assert_eq!(dirty.advance_gen, 1, "advance_gen still preserved");
}

/// The buffer-driven transition: dirty while non-empty, clean once emptied (the
/// rollback-to-empty and dirty-origin rows of §2.1).
#[test]
fn sync_from_buffer_tracks_emptiness() {
    let cell = ShardCommitState::new();
    cell.publish_recovered(Some(hash(0)), true); // clean baseline
    cell.sync_from_buffer(false);
    assert_ne!(
        cell.snapshot().committed_gen,
        cell.snapshot().dirty_gen,
        "non-empty buffer => dirty"
    );
    // Redundant dirty publish does not run gens away from each other unboundedly.
    let after_first = cell.snapshot().dirty_gen;
    cell.sync_from_buffer(false);
    assert_eq!(
        cell.snapshot().dirty_gen,
        after_first,
        "already dirty is idempotent"
    );
    cell.sync_from_buffer(true);
    assert_eq!(
        cell.snapshot().committed_gen,
        cell.snapshot().dirty_gen,
        "emptied buffer => clean"
    );
}

/// A commit that advances the root bumps `advance_gen` by exactly one and returns
/// it; a non-advancing commit returns None and bumps nothing. Both publish clean.
#[test]
fn publish_committed_bumps_advance_gen_iff_advanced() {
    let cell = ShardCommitState::new();
    cell.sync_from_buffer(false); // make it dirty first
    assert_eq!(cell.publish_committed(Some(hash(1)), true), Some(1));
    let after = cell.snapshot();
    assert_eq!(after.root, Some(hash(1)));
    assert_eq!(after.committed_gen, after.dirty_gen, "commit => clean");
    assert_eq!(after.advance_gen, 1);

    // A clean commit (no advance): no gen, no bump, still clean.
    assert_eq!(cell.publish_committed(Some(hash(1)), false), None);
    assert_eq!(cell.snapshot().advance_gen, 1);

    // Another advance: exactly one more.
    cell.sync_from_buffer(false);
    assert_eq!(cell.publish_committed(Some(hash(2)), true), Some(2));
}

/// `mark_unreconciled` forces the full path (unreconciled flag set).
#[test]
fn mark_unreconciled_sets_the_flag() {
    let cell = ShardCommitState::new();
    cell.publish_recovered(Some(hash(0)), true);
    assert!(!cell.snapshot().unreconciled);
    cell.mark_unreconciled();
    assert!(cell.snapshot().unreconciled);
    // A subsequent successful commit clears it.
    cell.publish_committed(Some(hash(1)), true);
    assert!(!cell.snapshot().unreconciled);
}

/// Restart case (c), cell half: a reader that observes the cell mid-recovery
/// classifies the shard DIRTY throughout the recovery window (from `begin_recovery`
/// until `publish_recovered`), and only the completing publish returns it to a
/// classification driven by the recovered buffer. `classify()` is the exact reader
/// half global commit uses, so this pins "no clean-skip while recovering".
#[test]
fn classify_is_dirty_throughout_recovery() {
    let cell = ShardCommitState::new();
    cell.publish_committed(Some(hash(7)), true); // clean at a real root first
    assert!(!cell.classify().dirty, "clean before recovery");

    cell.begin_recovery();
    assert!(
        cell.classify().dirty,
        "a reader mid-recovery must classify DIRTY (recovering set)"
    );
    // A second observation still dirty — the window holds until the publish.
    assert!(cell.classify().dirty, "still dirty later in the window");

    cell.publish_recovered(Some(hash(7)), true); // recovered empty buffer => clean
    assert!(
        !cell.classify().dirty,
        "only the completing publish returns the shard to clean"
    );
}

/// Restart case (d): incarnation is a monotonic reinitialisation observable that
/// bumps once per (re)spawn on the SAME process-lifetime cell, while `advance_gen`
/// is PRESERVED across incarnations (never reset, never persisted) — so a
/// (id, handle, cell) triple observes a continuous gen sequence across actor
/// restarts. Exercised on ONE cell `Arc` to model the same-cell-across-incarnations
/// invariant (the router caches one cell per shard id for process lifetime).
#[test]
fn incarnation_bumps_and_advance_gen_survives_across_incarnations() {
    let cell = ShardCommitState::new();
    // Incarnation 1 commits and advances the root twice.
    cell.begin_recovery();
    cell.publish_recovered(None, true);
    assert_eq!(cell.snapshot().incarnation, 1);
    cell.sync_from_buffer(false);
    assert_eq!(cell.publish_committed(Some(hash(1)), true), Some(1));
    cell.sync_from_buffer(false);
    assert_eq!(cell.publish_committed(Some(hash(2)), true), Some(2));
    assert_eq!(cell.snapshot().advance_gen, 2);

    // Incarnation 2 (a restart): incarnation bumps, advance_gen carries forward.
    cell.begin_recovery();
    assert_eq!(
        cell.snapshot().incarnation,
        2,
        "each respawn bumps incarnation"
    );
    cell.publish_recovered(Some(hash(2)), true);
    assert_eq!(
        cell.snapshot().advance_gen,
        2,
        "advance_gen survives the incarnation (never reset)"
    );
    // The next advance continues the sequence — gap-free across the restart.
    cell.sync_from_buffer(false);
    assert_eq!(
        cell.publish_committed(Some(hash(3)), true),
        Some(3),
        "the advance sequence continues across incarnations without a gap"
    );
}

/// Poison is ADOPTED by readers/writers and the next write CLEARS it (§2.2 ⟨r4⟩).
#[test]
fn poison_is_adopted_then_cleared_by_the_next_write() {
    let cell = ShardCommitState::new();
    cell.publish_committed(Some(hash(1)), true);
    cell.force_poison_for_test();
    assert!(cell.is_poisoned(), "the mutex is poisoned");
    // A reader adopts the poison without unwinding and still sees the last snapshot.
    assert_eq!(cell.snapshot().root, Some(hash(1)));
    // The actor's next write recovers the guard, overwrites, and clears the poison.
    cell.sync_from_buffer(true);
    assert!(!cell.is_poisoned(), "the next write clears the poison");
}