haematite 0.7.0

//! Tests for the root-advance event seam (`docs/design/ROOT-ADVANCE-SEAM.md` §9):
//! DIRECT-EMIT unit tests + INTEGRATION tests over a real [`Database`] (census §3).

use crate::db::DatabaseConfig;
use crate::tree::Hash;

use std::error::Error;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Barrier, Mutex, MutexGuard, PoisonError};
use std::thread;

use super::{Database, RootAdvance, RootAdvanceSeam, RootTransition, in_emission};

type BoxError = Box<dyn Error>;
type Seen = Arc<Mutex<Vec<RootAdvance>>>;

/// Poison-adopting lock so tests never `unwrap()` a mutex (crate-denied).
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
    mutex.lock().unwrap_or_else(PoisonError::into_inner)
}

/// A runtime panic that is NOT the `panic!` macro (crate-denied): index an empty
/// vector at its length. Used to model a subscriber that panics during delivery.
/// The `u8` return is never produced — the indexing diverges.
fn boom() -> u8 {
    let empty: Vec<u8> = Vec::new();
    let index = empty.len();
    empty[index]
}

fn collector() -> (Seen, impl Fn(RootAdvance) + Send + Sync + 'static) {
    let seen: Seen = Arc::default();
    let sink = Arc::clone(&seen);
    (seen, move |event| lock(&sink).push(event))
}

fn drain(seen: &Seen) -> Vec<RootAdvance> {
    lock(seen).clone()
}

fn gens(seen: &Seen) -> Vec<u64> {
    drain(seen)
        .into_iter()
        .map(|event| event.advance_gen)
        .collect()
}

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

fn transition(prior: u8, new: u8, generation: u64) -> RootTransition {
    RootTransition {
        prior_root: hash_of(prior),
        new_root: hash_of(new),
        advance_gen: generation,
    }
}

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

fn subscribe_seam(seam: &Arc<RootAdvanceSeam>) -> Seen {
    let (seen, sink) = collector();
    seam.subscribe(Arc::new(sink));
    seen
}

fn join<T>(handle: thread::JoinHandle<T>) -> Result<T, BoxError> {
    handle
        .join()
        .map_err(|_| "worker thread panicked".to_owned().into())
}

/// R3 wall: a write from inside a callback refuses typed, which BREAKS the
/// commit->tell->commit recursion (the callback fires a bounded number of times).
#[test]
fn reentrancy_wall_refuses_write_in_callback_and_bounds_recursion() -> Result<(), BoxError> {
    let (_dir, db) = temp_db(1)?;
    let db = Arc::new(db);
    let calls = Arc::new(AtomicUsize::new(0));
    let inner: Arc<Mutex<Vec<String>>> = Arc::default();

    let db_in = Arc::clone(&db);
    let calls_in = Arc::clone(&calls);
    let inner_in = Arc::clone(&inner);
    let sub = db.subscribe_root_advance(move |_event| {
        calls_in.fetch_add(1, Ordering::SeqCst);
        // Fail-first: unguarded, this inner commit would recurse commit->tell->commit
        // and self-deadlock on the held emission mutex. The wall must refuse it.
        assert!(in_emission(), "in-emission flag set during a callback");
        db_in.put(b"k2".to_vec(), b"v2".to_vec()).ok();
        lock(&inner_in).push(format!("{:?}", db_in.commit().err()));
    });

    db.put(b"k1".to_vec(), b"v1".to_vec())?;
    db.commit()?;

    assert_eq!(
        calls.load(Ordering::SeqCst),
        1,
        "callback fires once (no recursion)"
    );
    let refusals = lock(&inner).clone();
    assert_eq!(refusals.len(), 1);
    assert!(
        refusals[0].contains("WriteDuringRootAdvanceEmission"),
        "inner write must refuse typed, got {}",
        refusals[0]
    );
    drop(sub);
    Ok(())
}

/// Fail-first, structural: the shard emission mutex is HELD during a callback, so a
/// same-thread re-entrant tell would self-deadlock — `try_lock` proving it is held
/// is the deadlock the wall closes, demonstrated without hanging the test.
#[test]
fn emission_mutex_is_held_during_callback_so_reentrant_tell_would_deadlock() {
    let seam = RootAdvanceSeam::new();
    let state = seam.shard_state(0);
    let probe = Arc::clone(&state);
    let held = Arc::new(AtomicBool::new(false));
    let observed = Arc::clone(&held);
    seam.subscribe(Arc::new(move |_event: RootAdvance| {
        observed.store(probe.emission.try_lock().is_err(), Ordering::SeqCst);
    }));
    seam.emit(0, &state, transition(0, 1, 1));
    assert!(held.load(Ordering::SeqCst), "mutex held in callback");
}

/// The monotone filter: superseded advances (`gen` <= `last_told`) deliver ZERO
/// tells; delivered gens strictly increase (fail-first: a regression never appears).
#[test]
fn per_shard_filter_coalesces_superseded_and_strictly_increases() -> Result<(), BoxError> {
    let seam = RootAdvanceSeam::new();
    let state = seam.shard_state(0);
    let seen = subscribe_seam(&seam);

    seam.emit(0, &state, transition(0, 1, 1));
    seam.emit(0, &state, transition(1, 3, 3)); // jump — an intermediate was skipped
    seam.emit(0, &state, transition(1, 2, 2)); // superseded (2 <= 3) → zero tells
    seam.emit(0, &state, transition(3, 3, 3)); // equal gen → superseded → zero tells

    assert_eq!(gens(&seen), vec![1, 3], "superseded gens deliver nothing");
    let last = drain(&seen).pop().ok_or("no tell delivered")?;
    assert_eq!(last.new_root, hash_of(3), "last tell = latest root");
    Ok(())
}

/// Coalescing pin (§9.2): a test seam HOLDS the emission mutex to force two racing
/// emitters (gen 6, gen 5) into an uncontrolled order; delivered tells always
/// strictly increase and the highest gen (6) always delivers (5 coalesces if lost).
#[test]
fn adversarial_arrival_order_still_coalesces_to_highest_gen() -> Result<(), BoxError> {
    for _ in 0..200 {
        let seam = RootAdvanceSeam::new();
        let state = seam.shard_state(0);
        let seen = subscribe_seam(&seam);

        let guard = lock(&state.emission);
        let barrier = Arc::new(Barrier::new(3));
        let mut handles = Vec::new();
        for generation in [6_u64, 5_u64] {
            let seam = Arc::clone(&seam);
            let state = Arc::clone(&state);
            let barrier = Arc::clone(&barrier);
            handles.push(thread::spawn(move || {
                barrier.wait();
                let new = u8::try_from(generation).unwrap_or(0);
                seam.emit(0, &state, transition(0, new, generation));
            }));
        }
        barrier.wait();
        thread::yield_now(); // let both threads park on the mutex, then release
        drop(guard);
        for handle in handles {
            join(handle)?;
        }

        let delivered = gens(&seen);
        assert!(
            delivered.windows(2).all(|window| window[0] < window[1]),
            "delivered gens must strictly increase, got {delivered:?}"
        );
        assert_eq!(
            delivered.last(),
            Some(&6),
            "highest gen always delivers last"
        );
    }
    Ok(())
}

/// Integration: concurrent per-op appends to ONE shard deliver strictly
/// gen-increasing tells (actor serializes advances; filter serializes delivery).
#[test]
fn concurrent_appends_one_shard_deliver_strictly_increasing() -> Result<(), BoxError> {
    let (_dir, db) = temp_db(1)?;
    let db = Arc::new(db);
    let seen: Arc<Mutex<Vec<u64>>> = Arc::default();
    let sink = Arc::clone(&seen);
    let sub = db.subscribe_root_advance(move |event| lock(&sink).push(event.advance_gen));

    let barrier = Arc::new(Barrier::new(8));
    let mut handles = Vec::new();
    for worker in 0..8 {
        let db = Arc::clone(&db);
        let barrier = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            barrier.wait();
            db.append(
                format!("stream-{worker}").into_bytes(),
                vec![b"e".to_vec()],
                0,
            )
        }));
    }
    for handle in handles {
        join(handle)??;
    }

    let delivered = lock(&seen).clone();
    assert!(!delivered.is_empty(), "appends must produce tells");
    assert!(
        delivered.windows(2).all(|window| window[0] < window[1]),
        "delivered gens must strictly increase, got {delivered:?}"
    );
    drop(sub);
    Ok(())
}

/// The seam imposes NO cross-shard order: delivering shard 1 before shard 0 keeps
/// that order (and vice versa). Both interleavings are realized — the promise is
/// the ABSENCE of a cross-shard order.
#[test]
fn cross_shard_delivery_preserves_emission_order_not_a_shard_order() {
    for (first, second) in [(1_usize, 0_usize), (0_usize, 1_usize)] {
        let seam = RootAdvanceSeam::new();
        let state0 = seam.shard_state(0);
        let state1 = seam.shard_state(1);
        let seen = subscribe_seam(&seam);

        let (fst, snd) = if first == 0 {
            (&state0, &state1)
        } else {
            (&state1, &state0)
        };
        seam.emit(first, fst, transition(0, 1, 1));
        seam.emit(second, snd, transition(0, 1, 1));

        let order: Vec<usize> = drain(&seen)
            .into_iter()
            .map(|event| event.shard_id)
            .collect();
        assert_eq!(
            order,
            vec![first, second],
            "order follows emission, not shard id"
        );
    }
}

/// Only `prior != new` is tell-eligible and consumes a gen: empty/clean commits
/// emit NOTHING and bump NO gen; a real commit emits once with a gen one above.
#[test]
fn advance_invariant_clean_commit_emits_nothing_real_commit_bumps_one() -> Result<(), BoxError> {
    let (_dir, db) = temp_db(1)?;
    let (seen, sink) = collector();
    let sub = db.subscribe_root_advance(sink);

    db.commit()?; // empty commit → no advance, no tell
    assert!(drain(&seen).is_empty(), "empty commit must emit nothing");

    db.put(b"a".to_vec(), b"1".to_vec())?;
    db.commit()?;
    let first = drain(&seen);
    assert_eq!(first.len(), 1, "one real commit → one tell");
    assert_eq!(first[0].advance_gen, 1);
    assert_eq!(first[0].prior_root, crate::tree::empty_root_hash());
    assert_ne!(first[0].prior_root, first[0].new_root);

    db.commit()?; // clean commit → still nothing, gen unchanged
    assert_eq!(
        drain(&seen).len(),
        1,
        "clean commit must not emit or bump gen"
    );

    db.put(b"b".to_vec(), b"2".to_vec())?;
    db.commit()?;
    let all = drain(&seen);
    assert_eq!(all.len(), 2);
    assert_eq!(all[1].advance_gen, 2, "gen exactly one above the last");
    assert_eq!(
        all[1].prior_root, all[0].new_root,
        "consecutive delivered tells chain"
    );
    drop(sub);
    Ok(())
}

/// Inert seam: an idle soak (subscriber, no writes) delivers zero tells; with ZERO
/// subscribers, advancing commits succeed and the wall never raises. The seam owns
/// no thread/timer/fd (structural: [`RootAdvanceSeam`] has no such field).
#[test]
fn zero_subscriber_and_idle_soak_are_inert() -> Result<(), BoxError> {
    let (_dir, db) = temp_db(2)?;
    let (seen, sink) = collector();
    let sub = db.subscribe_root_advance(sink);
    for _ in 0..50 {
        db.commit()?;
    }
    assert!(drain(&seen).is_empty(), "idle commits must be inert");
    drop(sub);

    let (_dir2, db2) = temp_db(2)?;
    for round in 0..50_u32 {
        db2.put(
            format!("k{round}").into_bytes(),
            round.to_be_bytes().to_vec(),
        )?;
        db2.commit()?;
        assert!(!in_emission(), "no callback runs with zero subscribers");
    }
    Ok(())
}

/// A told advance is always durable (emission is sequenced AFTER the durable commit
/// reply): the tell's `new_root` equals the root a reopened database recovers.
#[test]
fn told_advance_is_durable() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let (seen, sink) = collector();
    {
        let db = Database::create(DatabaseConfig {
            data_dir: data_dir.clone(),
            shard_count: 1,
            distributed: None,
            executor_threads: None,
        })?;
        let sub = db.subscribe_root_advance(sink);
        db.put(b"durable".to_vec(), b"value".to_vec())?;
        let roots = db.commit()?;
        let told = drain(&seen);
        assert_eq!(told.len(), 1);
        assert_eq!(Some(&told[0].new_root), roots.get(&0));
        drop(sub);
    }
    // Reopen: the data is durable, and once shard 0 is materialised its recovered
    // committed root is exactly the told `new_root`.
    let reopened = Database::open(&data_dir)?;
    assert_eq!(
        reopened.get(b"durable")?,
        Some(b"value".to_vec()),
        "data durable across reopen"
    );
    let recovered = reopened.commit()?;
    let told_root = drain(&seen).first().map(|event| event.new_root);
    assert_eq!(
        recovered.get(&0).copied(),
        told_root,
        "told root is the durable root"
    );
    Ok(())
}

/// An advance that happens AFTER subscribe is delivered, never lost (the attach
/// protocol closes the window: a post-subscribe advance always arrives as a tell).
#[test]
fn advance_after_subscribe_is_delivered() -> Result<(), BoxError> {
    let (_dir, db) = temp_db(1)?;
    let db = Arc::new(db);
    let (seen, sink) = collector();
    let sub = db.subscribe_root_advance(sink);

    let writer = Arc::clone(&db);
    join(thread::spawn(move || -> Result<(), String> {
        writer
            .put(b"late".to_vec(), b"v".to_vec())
            .map_err(|e| e.to_string())?;
        writer.commit().map(drop).map_err(|e| e.to_string())
    }))??;

    let told = drain(&seen);
    assert_eq!(told.len(), 1, "a post-subscribe advance must be delivered");
    assert_eq!(told[0].advance_gen, 1);
    drop(sub);
    Ok(())
}

/// Panicking subscriber (§4): commit result unaffected; the LATER peer in the
/// snapshot is NOT told that tell (asserted starvation) but IS told the next
/// advance; the mutex stays healthy (poison-adopting); no gen regression.
#[test]
fn panicking_subscriber_starves_later_peer_for_that_tell_only() {
    let seam = RootAdvanceSeam::new();
    let state = seam.shard_state(0);

    // A: panics on its FIRST tell only. B: a collector, registered AFTER A so it is
    // strictly later in the snapshot (A runs first, then B).
    let a_panics = Arc::new(AtomicBool::new(true));
    let a_flag = Arc::clone(&a_panics);
    seam.subscribe(Arc::new(move |_event: RootAdvance| {
        if a_flag.swap(false, Ordering::SeqCst) {
            boom();
        }
    }));
    let seen_b = subscribe_seam(&seam);

    // First advance: A panics; emission unwinds on this thread. Catch it so the
    // commit-result-unaffected property is observable (only delivery unwound).
    let seam1 = Arc::clone(&seam);
    let state1 = Arc::clone(&state);
    let first = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        seam1.emit(0, &state1, transition(0, 1, 1));
    }));
    assert!(first.is_err(), "A's panic unwinds on the emitting thread");
    assert!(
        drain(&seen_b).is_empty(),
        "later peer B is STARVED of the tell A panicked on"
    );

    // Second advance: A does not panic; the mutex is healthy (poison-adopting), so
    // B is told, with a strictly greater gen (no regression).
    seam.emit(0, &state, transition(1, 2, 2));
    let told_b = drain(&seen_b);
    assert_eq!(told_b.len(), 1, "B is told on the next advance");
    assert_eq!(
        told_b[0].advance_gen, 2,
        "no gen regression after partial delivery"
    );
}

/// Row 4: a receiver-driven durable apply that advances the root tells once, on the
/// CALLER's (receiver's) thread, through the same seam path as every other row.
#[test]
fn replication_apply_advancing_root_tells_on_caller_thread() -> Result<(), BoxError> {
    use beamr::module::ModuleRegistry;
    use beamr::scheduler::{Scheduler, SchedulerConfig};

    use crate::shard::actor::ShardHandle;
    use crate::store::DiskStore;
    use crate::sync::ballot::Stamp;
    use crate::tree::{LeafNode, Node};

    let dir = tempfile::tempdir()?;
    let store_dir = dir.path().join("shard.store");
    let wal_path = dir.path().join("shard.wal");
    let mut store = DiskStore::new(&store_dir)?;
    store.put(&Node::Leaf(LeafNode::new(Vec::new())?))?;
    drop(store);

    let seam = RootAdvanceSeam::new();
    let seen = subscribe_seam(&seam);
    let on_caller_thread = Arc::new(AtomicBool::new(false));
    let probe = Arc::clone(&on_caller_thread);
    let caller = thread::current().id();
    seam.subscribe(Arc::new(move |_event: RootAdvance| {
        probe.store(thread::current().id() == caller, Ordering::SeqCst);
    }));

    let scheduler = Arc::new(
        Scheduler::new(SchedulerConfig::default(), Arc::new(ModuleRegistry::new()))
            .map_err(|message| -> BoxError { message.into() })?,
    );
    let handle = ShardHandle::spawn(
        scheduler,
        store_dir,
        wal_path,
        0,
        crate::tree::TreePolicy::V1_DEFAULT,
        Arc::clone(&seam),
    )?;

    handle.apply_durable(
        b"replicated".to_vec(),
        None,
        b"payload".to_vec(),
        None,
        Stamp::bottom(),
        std::time::Duration::from_secs(5),
    )?;

    let told = drain(&seen);
    assert_eq!(
        told.len(),
        1,
        "a replication apply that advances the root tells once"
    );
    assert_eq!(told[0].advance_gen, 1);
    assert!(
        on_caller_thread.load(Ordering::SeqCst),
        "callback runs on the caller's thread"
    );
    Ok(())
}