haematite 0.7.0

//! COMMIT-COLLAPSE §11 O(dirty) global-commit integration tests, driven through a
//! real [`Database`].
//!
//! These pin the load-bearing behaviour of items 3/4/5: a zero-dirty commit costs
//! nothing (the flagship permanent negative assertion, §10/§11), a one-dirty
//! commit among many dispatches exactly one job and rewrites exactly one WAL, the
//! `unwrap_or(empty)` assembly returns cached NON-empty roots (the scout's central
//! risk), the clean fast path performs zero storage on a marker-bearing shard, the
//! marker-less shard publishes its committed-empty marker once so the read-only
//! observer agrees with the live view (§4 M4), and the per-shard cell has stable
//! pointer identity for process lifetime (§11 restart pin).
//!
//! SYNC-COUNTER note (§11 M6): the zero-dirty "does nothing" proof rests on the
//! executor's cross-thread dispatch gauge — a clean shard receives NO `Commit`
//! command, so ZERO of the five sync classes execute BY CONSTRUCTION (every fsync
//! path lives inside `Commit` handling on the actor thread). This is the
//! cross-thread-correct form of the assertion the r1 thread-local seam could not
//! make; it is corroborated by byte- and mtime-identical WAL/store snapshots.

use std::collections::BTreeMap;
use std::error::Error;
use std::path::Path;
use std::sync::Arc;
use std::thread;
use std::time::SystemTime;

use super::{Database, DatabaseConfig, ReadOnlyDatabase};
use crate::tree::{Hash, empty_root_hash};

type BoxError = Box<dyn Error>;

fn config_for(path: &Path, shard_count: usize) -> DatabaseConfig {
    DatabaseConfig {
        data_dir: path.to_path_buf(),
        shard_count,
        distributed: None,
        executor_threads: None,
    }
}

/// The bytes of shard `id`'s durable WAL file (empty if it does not exist yet).
fn wal_bytes(data_dir: &Path, id: usize) -> Result<Vec<u8>, BoxError> {
    let path = data_dir.join(format!("shard-{id}")).join("shard.wal");
    match std::fs::read(&path) {
        Ok(bytes) => Ok(bytes),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
        Err(error) => Err(error.into()),
    }
}

/// Shard `id`'s WAL mtime, or `None` if the file does not exist.
fn wal_mtime(data_dir: &Path, id: usize) -> Result<Option<SystemTime>, BoxError> {
    let path = data_dir.join(format!("shard-{id}")).join("shard.wal");
    match std::fs::metadata(&path) {
        Ok(metadata) => Ok(Some(metadata.modified()?)),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error.into()),
    }
}

/// Force-materialise every shard id in `0..shard_count` (spawns each actor and
/// runs its real boot) so a subsequent commit visits every one.
fn materialise_all(db: &Database, shard_count: usize) -> Result<(), BoxError> {
    for shard_id in 0..shard_count {
        let _handle = db.handle_for_shard(shard_id)?;
    }
    Ok(())
}

/// Zero-dirty commit does nothing (§11 flagship): fully materialise every shard,
/// commit until all are marker-bearing and clean, then commit again and assert the
/// pool dispatched ZERO jobs, every WAL is byte- and mtime-identical, the thread
/// census is unchanged, and the returned roots are identical.
#[test]
fn zero_dirty_commit_does_nothing() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let shard_count = 16;
    let db = Database::create(config_for(&data_dir, shard_count))?;

    materialise_all(&db, shard_count)?;
    // First commit publishes each shard's committed-empty marker (root None =>
    // classified dirty => full path => marker). After it every shard is clean.
    let roots_first = db.commit()?;
    let jobs_after_marker = db.executor().dispatched_jobs();
    let workers = db.executor().worker_count();

    // Snapshot every WAL's bytes + mtime while the database is quiescent.
    let bytes_before: Vec<Vec<u8>> = (0..shard_count)
        .map(|id| wal_bytes(&data_dir, id))
        .collect::<Result<_, _>>()?;
    let mtime_before: Vec<Option<SystemTime>> = (0..shard_count)
        .map(|id| wal_mtime(&data_dir, id))
        .collect::<Result<_, _>>()?;

    // The zero-dirty commit.
    let roots_second = db.commit()?;

    assert_eq!(roots_second, roots_first, "zero-dirty roots are identical");
    assert_eq!(
        db.executor().dispatched_jobs(),
        jobs_after_marker,
        "a zero-dirty commit dispatches NO executor job (no shard is dirty)"
    );
    assert_eq!(
        db.executor().worker_count(),
        workers,
        "the pool creates no thread after startup (bounded census)"
    );
    for (id, (expected_bytes, expected_mtime)) in bytes_before.iter().zip(&mtime_before).enumerate()
    {
        assert_eq!(
            &wal_bytes(&data_dir, id)?,
            expected_bytes,
            "shard {id} WAL must be byte-identical after a zero-dirty commit"
        );
        assert_eq!(
            wal_mtime(&data_dir, id)?,
            *expected_mtime,
            "shard {id} WAL mtime must be unchanged after a zero-dirty commit"
        );
    }
    Ok(())
}

/// One dirty among 64 (§11): fully materialise 64 shards and clean them, touch
/// exactly ONE, then commit — the pool dispatches exactly one job, exactly one WAL
/// changes, and the other 63 are byte-identical.
#[test]
fn one_dirty_among_64_dispatches_exactly_one_job() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let shard_count = 64;
    let db = Database::create(config_for(&data_dir, shard_count))?;

    materialise_all(&db, shard_count)?;
    db.commit()?; // publish every marker; all clean
    let jobs_clean = db.executor().dispatched_jobs();

    // Find a key and dirty exactly its owning shard.
    let key = b"one-dirty-key".to_vec();
    let dirty_shard = db.shard_for(&key);
    let bytes_before: Vec<Vec<u8>> = (0..shard_count)
        .map(|id| wal_bytes(&data_dir, id))
        .collect::<Result<_, _>>()?;

    db.put(key, b"v".to_vec())?;
    db.commit()?;

    assert_eq!(
        db.executor().dispatched_jobs() - jobs_clean,
        1,
        "exactly one shard is dirty => exactly one dispatched job"
    );
    let mut changed = Vec::new();
    for (id, before) in bytes_before.iter().enumerate() {
        if &wal_bytes(&data_dir, id)? != before {
            changed.push(id);
        }
    }
    assert_eq!(
        changed,
        vec![dirty_shard],
        "only the dirtied shard's WAL changes; the other 63 are byte-identical"
    );
    Ok(())
}

/// Root-cache (§11, the `unwrap_or(empty)` central risk): a repeated commit returns
/// the cached NON-empty roots for clean data-bearing shards, dispatches no job on
/// the repeat, and survives a reopen byte-identically.
#[test]
fn repeated_commit_returns_cached_non_empty_root() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let db = Database::create(config_for(&data_dir, 4))?;

    db.put(b"cached".to_vec(), b"value".to_vec())?;
    let first = db.commit()?;
    let dirty_shard = db.shard_for(b"cached");
    let cached_root = *first.get(&dirty_shard).ok_or("dirty shard has a root")?;
    assert_ne!(
        cached_root,
        empty_root_hash(),
        "a data-bearing shard commits to a NON-empty root"
    );

    let jobs_before = db.executor().dispatched_jobs();
    let repeated = db.commit()?;
    assert_eq!(
        repeated, first,
        "the repeat returns identical roots from the cell"
    );
    assert_eq!(
        *repeated.get(&dirty_shard).ok_or("dirty shard present")?,
        cached_root,
        "the clean-skip returns the cached NON-empty root, never the empty constant"
    );
    assert_eq!(
        db.executor().dispatched_jobs(),
        jobs_before,
        "the repeat is a pure clean skip — no job dispatched"
    );
    drop(db);

    // Reopen and touch the data shard so it materialises and recovers its committed
    // root into the cell (lazy contract: an un-materialised shard synthesises the
    // empty root). The recovered shard is CLEAN, so commit clean-skips it and
    // returns the cached NON-empty root — byte-identical to the pre-reopen vector.
    let reopened = Database::open(&data_dir)?;
    assert_eq!(
        reopened.get(b"cached")?,
        Some(b"value".to_vec()),
        "data survives reopen"
    );
    assert_eq!(
        reopened.commit()?,
        first,
        "roots survive a reopen byte-identically"
    );
    Ok(())
}

/// A per-op durable write (cas) commits inside its own command; a following global
/// commit clean-skips that shard and returns the per-op root with no dispatched job.
#[test]
fn per_op_commit_then_global_commit_clean_skips() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let db = Database::create(config_for(&data_dir, 4))?;

    db.cas(b"counter".to_vec(), None, 1)?; // stages AND commits in one command
    let shard = db.shard_for(b"counter");
    let after_cas = db.commit_state_snapshot(shard);
    assert!(
        after_cas.committed_gen == after_cas.dirty_gen && after_cas.root.is_some(),
        "a per-op cas leaves the shard clean at a published root"
    );

    let jobs_before = db.executor().dispatched_jobs();
    let roots = db.commit()?;
    assert_eq!(
        db.executor().dispatched_jobs(),
        jobs_before,
        "the shard is already clean after the per-op commit — global commit skips it"
    );
    assert_eq!(
        *roots.get(&shard).ok_or("shard root present")?,
        after_cas.root.ok_or("cas published a root")?,
        "the clean skip returns the per-op committed root"
    );
    Ok(())
}

/// The §4 clean fast path performs ZERO storage on a marker-bearing shard: a second
/// `Commit` sent straight to the handle returns the cached root and does not rewrite
/// the WAL (defense in depth for a false-DIRTY classification).
#[test]
fn clean_fast_path_second_commit_writes_no_storage() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let db = Database::create(config_for(&data_dir, 1))?;

    db.put(b"k".to_vec(), b"v".to_vec())?;
    let handle = db.handle_for_shard(0)?;
    let root_first = handle.commit(db.timeout())?; // real work: flush + marker
    let bytes_before = wal_bytes(&data_dir, 0)?;

    // A second Commit against the now-empty buffer takes the fast path: same root,
    // zero storage mutation, WAL untouched.
    let root_second = handle.commit(db.timeout())?;
    assert_eq!(
        root_second, root_first,
        "the fast path returns the cached root"
    );
    assert_eq!(
        wal_bytes(&data_dir, 0)?,
        bytes_before,
        "the clean fast path performs zero storage mutation (WAL byte-identical)"
    );
    Ok(())
}

/// Observer agreement (§4 M4, the subtlest invariant): a materialised-empty shard
/// gains its committed-empty marker at first global commit, so the read-only
/// observer answers `Some(empty)` — matching the live view — instead of the
/// never-committed `None`.
#[test]
fn observer_agrees_after_materialised_empty_commit() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let db = Database::create(config_for(&data_dir, 4))?;

    // Materialise shard 0 as an EMPTY shard (a read touches it), commit to publish
    // its one-time marker.
    let _handle = db.handle_for_shard(0)?;
    let roots = db.commit()?;
    assert_eq!(
        *roots.get(&0).ok_or("shard 0 root present")?,
        empty_root_hash(),
        "the materialised-empty shard commits to the empty root"
    );

    let observer = ReadOnlyDatabase::open(&data_dir)?;
    assert_eq!(
        observer.committed_root(0)?,
        Some(empty_root_hash()),
        "observer agrees: a materialised-empty shard reports Some(empty), not None"
    );
    // A shard never materialised carries NO on-disk marker — the observer reports
    // never-committed None (the documented GATE-1 distinction the returned map's
    // synthesised empty does not disturb).
    assert_eq!(
        observer.committed_root(3)?,
        None,
        "a never-materialised shard is never-committed (observer None)"
    );
    Ok(())
}

/// Cell identity is 1:1 with shard id for process lifetime (§11 restart pin): the
/// seam caches, so every fetch of a shard's cell is the SAME `Arc` (pointer
/// identity) — the router snapshot and the actor factory therefore share one cell,
/// and no `(id, handle, cell)` triple can pair a shard with another's state.
#[test]
fn commit_state_cell_has_stable_pointer_identity() -> Result<(), BoxError> {
    let (_dir, db) = {
        let dir = tempfile::tempdir()?;
        let db = Database::create(config_for(&dir.path().join("db"), 4))?;
        (dir, db)
    };
    let seam = db.seam();
    let cell_a = seam.commit_state(2);
    let cell_b = seam.commit_state(2);
    assert!(
        Arc::ptr_eq(&cell_a, &cell_b),
        "the same shard id resolves to the SAME cell Arc (pointer identity)"
    );
    let other = seam.commit_state(3);
    assert!(
        !Arc::ptr_eq(&cell_a, &other),
        "distinct shard ids get distinct cells"
    );
    Ok(())
}

/// A commit issued from inside a root-advance callback is refused before any
/// fan-out (the R3 write wall on the calling thread) — the O(dirty) rewrite keeps
/// that guard.
#[test]
fn commit_from_emission_thread_is_refused() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let db = Arc::new(Database::create(config_for(&data_dir, 2))?);

    let inner = Arc::clone(&db);
    let outcome: Arc<std::sync::Mutex<Option<bool>>> = Arc::new(std::sync::Mutex::new(None));
    let sink = Arc::clone(&outcome);
    let _subscription = db.subscribe_root_advance(move |_advance| {
        let refused = matches!(
            inner.commit(),
            Err(crate::db::DatabaseError::WriteDuringRootAdvanceEmission)
        );
        if let Ok(mut slot) = sink.lock() {
            *slot = Some(refused);
        }
    });

    db.put(b"k".to_vec(), b"v".to_vec())?;
    db.commit()?; // advances the root => fires the callback on the committer's thread

    let observed = *outcome.lock().map_err(|_| "callback outcome poisoned")?;
    assert_eq!(
        observed,
        Some(true),
        "a commit from inside a root-advance callback must be refused"
    );
    Ok(())
}

/// Deterministic failure seam (a) — the post-WAL-commit barrier (§11, deferred item
/// 4a): a test-only rendezvous holds a shard EXACTLY between "WAL marker durable"
/// and "cell publishes clean". At that window the marker is durable on disk yet the
/// cell still classifies DIRTY (never falsely clean — the publication is pending);
/// releasing the barrier lets the publish complete and the shard converges to clean
/// at the published root. This is the in-process expression of the restart (b)
/// "marker durable, publication pending" case.
#[test]
fn post_wal_commit_barrier_holds_marker_durable_publication_pending() -> Result<(), BoxError> {
    use crate::shard::commit_state::{SeamBarrier, arm_seam_barrier};

    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let db = Arc::new(Database::create(config_for(&data_dir, 1))?);
    db.put(b"k".to_vec(), b"v".to_vec())?; // materialises shard 0, dirty

    let cell = db.seam().commit_state(0);
    let barrier = arm_seam_barrier(&cell, SeamBarrier::PostWalCommit);

    // Commit on a worker thread — it parks at the post-WAL-commit barrier.
    let committer = {
        let db = Arc::clone(&db);
        thread::spawn(move || db.commit().map(|_| ()).map_err(|error| error.to_string()))
    };
    barrier.wait_until_reached();

    // The WAL marker is durable, but the cell has NOT published clean yet.
    let wal_path = data_dir.join("shard-0").join("shard.wal");
    let contents = crate::wal::DurableWal::read_file(&wal_path)?;
    assert!(
        contents.committed_root().is_some(),
        "the WAL marker is durable at the barrier"
    );
    let pending = db.commit_state_snapshot(0);
    assert!(
        pending.committed_gen != pending.dirty_gen,
        "publication pending ⇒ the cell still classifies DIRTY (never falsely clean)"
    );

    barrier.release();
    committer.join().map_err(|_| "committer panicked")??;

    let converged = db.commit_state_snapshot(0);
    assert!(
        converged.committed_gen == converged.dirty_gen && converged.root.is_some(),
        "after the publish the shard converges to clean at the published root"
    );
    Ok(())
}

/// Deterministic failure seam (b) — the recovery pause barrier (§11, deferred item
/// 4b): a shard held mid-recovery (after `begin_recovery`, before boot) is observed
/// by a reader that must classify DIRTY throughout (the `recovering` flag is set and
/// no clean-skip is allowed); releasing the barrier lets recovery complete and the
/// shard converges to clean at the recovered marker root with its data intact.
///
/// A concurrent global commit submitted DURING the pause is deliberately NOT issued
/// here: `materialise` holds the router map lock across the synchronous first boot
/// slice, so a second router user would block on that lock — the serial actor
/// already guarantees a queued command cannot execute until boot completes, which is
/// what the "commit waits then converges" clause asserts, and convergence after
/// release witnesses it.
#[test]
fn recovery_barrier_reader_classifies_dirty_then_converges() -> Result<(), BoxError> {
    use crate::shard::commit_state::{SeamBarrier, arm_seam_barrier};

    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let committed_root;
    {
        let db = Database::create(config_for(&data_dir, 1))?;
        db.put(b"k".to_vec(), b"v".to_vec())?;
        committed_root = *db.commit()?.get(&0).ok_or("committed root")?;
    } // durable marker on disk

    let db = Arc::new(Database::open(&data_dir)?);
    // Create + cache the cell BEFORE materialising so the factory shares this Arc.
    let cell = db.seam().commit_state(0);
    let barrier = arm_seam_barrier(&cell, SeamBarrier::Recovery);

    // Materialise shard 0 on a worker thread — boot parks at the recovery barrier.
    let materialiser = {
        let db = Arc::clone(&db);
        thread::spawn(move || {
            db.handle_for_shard(0)
                .map(|_| ())
                .map_err(|error| error.to_string())
        })
    };
    barrier.wait_until_reached();

    // A reader mid-recovery classifies DIRTY: recovering is set, no clean-skip.
    let during = db.commit_state_snapshot(0);
    assert!(
        during.recovering,
        "the recovering flag is set during the pause"
    );
    assert!(
        cell.classify().dirty,
        "a reader mid-recovery classifies the shard DIRTY (no clean-skip)"
    );

    barrier.release();
    materialiser.join().map_err(|_| "materialiser panicked")??;

    // Recovery completed: clean at the recovered marker root, data intact.
    let after = db.commit_state_snapshot(0);
    assert!(
        after.committed_gen == after.dirty_gen && after.root == Some(committed_root),
        "recovery converges to clean at the recovered marker root"
    );
    assert_eq!(
        db.get(b"k")?,
        Some(b"v".to_vec()),
        "the committed data is intact after the paused recovery"
    );
    Ok(())
}

/// Deterministic failure seam (c) at the ordinary commit path (§11, deferred item
/// 4c): a token(path)-scoped WAL-replacement failpoint makes a global commit fail
/// with a typed error; the shard retains its buffer and STAYS DIRTY (no root-cache
/// corruption), and the retry — the failpoint self-disarms — converges to a clean
/// commit with the data durable. Extends the `persist::fail_next_parent_dir_sync`
/// pattern into the WAL layer (cross-thread, path-keyed).
#[test]
fn wal_replacement_failure_keeps_shard_dirty_then_retry_converges() -> Result<(), BoxError> {
    use crate::wal::durable::{WalReplaceFailure, arm_replacement_failure};

    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let db = Database::create(config_for(&data_dir, 1))?;
    db.put(b"k".to_vec(), b"v".to_vec())?; // materialises shard 0, dirty + uncommitted
    let dirty = db.commit_state_snapshot(0);
    assert!(
        dirty.committed_gen != dirty.dirty_gen,
        "the buffered put classifies dirty before the commit"
    );

    // Arm the NEXT WAL replacement on shard 0 to fail before the rename.
    let wal_path = data_dir.join("shard-0").join("shard.wal");
    arm_replacement_failure(&wal_path, WalReplaceFailure::BeforePersist);

    assert!(
        db.commit().is_err(),
        "the WAL replacement failure surfaces as a commit error"
    );
    let after_failure = db.commit_state_snapshot(0);
    assert!(
        after_failure.committed_gen != after_failure.dirty_gen,
        "an ordinary commit failure retains the buffer => shard STAYS DIRTY"
    );

    // The retry (failpoint self-disarmed) converges.
    db.commit()?;
    let converged = db.commit_state_snapshot(0);
    assert!(
        converged.committed_gen == converged.dirty_gen && converged.root.is_some(),
        "the retry commits the retained buffer => clean at a published root"
    );
    assert_eq!(
        db.get(b"k")?,
        Some(b"v".to_vec()),
        "the write is durable after convergence"
    );
    Ok(())
}

/// One-time marker concurrency (§11 residual, deferred item 5): two racing global
/// commits against a marker-LESS materialised (empty) shard. Both classify it dirty
/// (root None), the actor serialises the two `Commit` commands, the first publishes
/// the committed-empty marker via the full path and the second takes the clean fast
/// path — so both callers receive the identical empty root, the shard settles to
/// clean+marker (a following commit dispatches zero jobs), and the observer agrees.
#[test]
fn concurrent_commits_on_markerless_shard_publish_one_marker() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let db = Arc::new(Database::create(config_for(&data_dir, 4))?);

    // Materialise shard 0 as an EMPTY, marker-LESS shard — do NOT commit yet, so
    // both racing commits find it un-marked (committed_root None ⇒ classified dirty).
    let _handle = db.handle_for_shard(0)?;

    let barrier = Arc::new(std::sync::Barrier::new(2));
    let mut roots = Vec::new();
    let handles: Vec<_> = (0..2)
        .map(|_| {
            let db = Arc::clone(&db);
            let barrier = Arc::clone(&barrier);
            thread::spawn(move || -> Result<BTreeMap<usize, Hash>, String> {
                barrier.wait();
                db.commit().map_err(|error| error.to_string())
            })
        })
        .collect();
    for handle in handles {
        roots.push(handle.join().map_err(|_| "committer panicked")??);
    }

    let (first, second) = (&roots[0], &roots[1]);
    assert_eq!(
        first, second,
        "both racing commits return the identical root vector"
    );
    assert_eq!(
        *first.get(&0).ok_or("shard 0 root present")?,
        empty_root_hash(),
        "the marker-less empty shard commits to the empty root for both callers"
    );

    // The shard settled to clean + marker: a following commit dispatches no job.
    let jobs_before = db.executor().dispatched_jobs();
    db.commit()?;
    assert_eq!(
        db.executor().dispatched_jobs(),
        jobs_before,
        "after the one-time marker publication the shard is clean — no further job"
    );

    let observer = ReadOnlyDatabase::open(&data_dir)?;
    assert_eq!(
        observer.committed_root(0)?,
        Some(empty_root_hash()),
        "observer agrees the shard is committed-empty after the concurrent commits"
    );
    Ok(())
}

/// Write racing a global commit is never lost — both interleavings (§11 M1,
/// deferred item 6). Determinism comes from the actor's serial command queue: each
/// blocking API call completes (its cell publish is visible) before the next call
/// observes state, so the two orderings are exact, not timing-dependent, and no
/// global commit boundary is asserted.
///
/// Interleaving A — the write lands BEFORE the commit's classification: the put
/// completes (cell dirty), then commit classifies dirty and persists it THIS round.
/// Interleaving B — the write lands AFTER a commit: that commit returns the
/// pre-write root, and the NEXT commit picks the write up. In both the write
/// survives.
#[test]
fn write_racing_commit_is_never_lost_both_interleavings() -> Result<(), BoxError> {
    // --- Interleaving A: write before classification ⇒ committed this round. ---
    let dir_a = tempfile::tempdir()?;
    let db_a = Database::create(config_for(&dir_a.path().join("db"), 4))?;
    let shard_a = db_a.shard_for(b"key-a");
    // Materialise + publish the marker so the pre-write root is a real empty marker.
    let _handle = db_a.handle_for_shard(shard_a)?;
    let pre_write = *db_a.commit()?.get(&shard_a).ok_or("pre-write root")?;

    db_a.put(b"key-a".to_vec(), b"va".to_vec())?; // write lands (cell dirty) …
    let round = db_a.commit()?; // … before this commit's classification ⇒ included
    assert_ne!(
        *round.get(&shard_a).ok_or("shard root present")?,
        pre_write,
        "interleaving A: the write is committed in the SAME round (root advanced)"
    );
    assert_eq!(
        db_a.get(b"key-a")?,
        Some(b"va".to_vec()),
        "interleaving A: the write is durable"
    );

    // --- Interleaving B: write after a commit ⇒ the NEXT commit picks it up. ---
    let dir_b = tempfile::tempdir()?;
    let db_b = Database::create(config_for(&dir_b.path().join("db"), 4))?;
    let shard_b = db_b.shard_for(b"key-b");
    let _handle_b = db_b.handle_for_shard(shard_b)?;
    let before = *db_b.commit()?.get(&shard_b).ok_or("pre-write root b")?;

    db_b.put(b"key-b".to_vec(), b"vb".to_vec())?; // write lands AFTER the commit above
    let next = db_b.commit()?; // the next commit picks it up
    assert_eq!(
        before,
        empty_root_hash(),
        "interleaving B: the pre-write commit returned the empty pre-write root"
    );
    assert_ne!(
        *next.get(&shard_b).ok_or("shard root present b")?,
        before,
        "interleaving B: the NEXT commit picks up the write (root advanced)"
    );
    assert_eq!(
        db_b.get(b"key-b")?,
        Some(b"vb".to_vec()),
        "interleaving B: the write is never lost"
    );
    Ok(())
}

/// Bounded threads under sustained load (§11 M8, deferred item 8): N concurrent
/// committer threads over M shards, each doing repeated put+commit, must not grow
/// the executor. The fixed-size named worker pool (`haematite-exec-{i}`) is only
/// ever spawned at startup, so `worker_count()` is the authoritative no-growth
/// census (the portable proxy the landed zero-dirty test uses); the admission
/// ceiling bounds queued+running jobs at `max_admitted_batches × shard_count`
/// throughout.
#[test]
fn bounded_threads_hold_under_sustained_load() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let shard_count = 8;
    let db = Arc::new(Database::create(config_for(&data_dir, shard_count))?);
    materialise_all(&db, shard_count)?;
    db.commit()?; // publish every marker

    let workers_at_start = db.executor().worker_count();
    let ceiling = db.executor().max_admitted_batches() * shard_count;

    let committers = 6;
    let iterations = 25;
    let barrier = Arc::new(std::sync::Barrier::new(committers));
    let handles: Vec<_> = (0..committers)
        .map(|thread_id| {
            let db = Arc::clone(&db);
            let barrier = Arc::clone(&barrier);
            thread::spawn(move || -> Result<(), String> {
                barrier.wait();
                for round in 0..iterations {
                    let key = format!("t{thread_id}-r{round}").into_bytes();
                    db.put(key, vec![thread_id as u8])
                        .map_err(|e| e.to_string())?;
                    db.commit().map_err(|e| e.to_string())?;
                    // Sample the ceiling under contention: queued+running never
                    // exceeds the admission bound.
                    assert!(
                        db.executor().queued_and_running() <= ceiling,
                        "queued+running exceeded the admission ceiling under load"
                    );
                }
                Ok(())
            })
        })
        .collect();
    for handle in handles {
        handle.join().map_err(|_| "committer panicked")??;
    }

    assert_eq!(
        db.executor().worker_count(),
        workers_at_start,
        "the executor spawned no thread after startup under sustained load (census delta = 0)"
    );
    Ok(())
}

/// Assemble the full ordered root vector for a small workload and confirm the
/// mixed clean/dirty/never-materialised assembly is total and correct across a
/// second commit (no None panic, no omission — §5 step 5).
#[test]
fn mixed_dirty_clean_assembly_is_total_and_stable() -> Result<(), BoxError> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    let shard_count = 8;
    let db = Database::create(config_for(&data_dir, shard_count))?;

    db.put(b"alpha".to_vec(), b"1".to_vec())?;
    db.put(b"beta".to_vec(), b"2".to_vec())?;
    let first: BTreeMap<usize, Hash> = db.commit()?;
    assert_eq!(
        first.len(),
        shard_count,
        "every slot is present (total assembly)"
    );

    // Dirty one already-materialised shard again; leave the rest clean.
    db.put(b"alpha".to_vec(), b"1b".to_vec())?;
    let second = db.commit()?;
    assert_eq!(second.len(), shard_count);
    for id in 0..shard_count {
        assert!(
            second.contains_key(&id),
            "slot {id} present after the second commit"
        );
        if id != db.shard_for(b"alpha") {
            assert_eq!(
                second.get(&id),
                first.get(&id),
                "an untouched shard's root is stable across commits"
            );
        }
    }
    Ok(())
}