obj-db 1.0.2

//! N-reader + 1-writer concurrent stress test (M6 issue #49).
//!
//! The M6 exit gate.  Spawns N=8 reader threads + 1 writer thread
//! inside `std::thread::scope`.  The writer continuously runs
//! `Db::transaction` closures mixing 60% inserts / 25% updates /
//! 15% deletes; each reader continuously runs `Db::read_transaction`
//! closures sampling random ids in the writer-allocated range and
//! asserts the result is either `None` or a `StressDoc` whose
//! embedded `id_echo` field matches the queried [`Id`] (the
//! sentinel for torn-read / aliasing detection).
//!
//! Duration is parameterised via `OBJ_STRESS_DURATION_SECS`
//! (default 60 s for local runs; CI runs the 5-minute gate; the
//! full 1-hour soak is the human-driven release validation step,
//! see `CONTRIBUTING.md` § Concurrency stress).
//!
//! No-deadlock invariant: every thread bumps a heartbeat counter
//! every 100 ops; the watchdog thread panics if any heartbeat
//! fails to advance for 30 s.
//!
//! On failure the run prints `SEED=<N>` to stderr and writes the
//! captured operation log to `target/stress/seed-<N>.log` so the
//! failing seed can be reproduced via `OBJ_STRESS_SEED`.
//!
//! # Power-of-ten posture
//!
//! - **Rule 2.** Every loop in this test is bounded by a deadline
//!   plus an explicit `stop` flag; the watchdog itself loops
//!   against `start.elapsed() < duration` (a deterministic bound).
//! - **Rule 4.** Per-op helpers live in their own functions; the
//!   harness's `run_stress` is just dispatch.
//! - **Rule 5.** The `id_echo == queried_id` assertion is the
//!   type-level invariant the document carries; readers verify it
//!   on every observation.
//! - **Rule 7.** Reader threads bubble errors out of the loop as
//!   `Result<(), String>` and the watchdog records them — no
//!   `.unwrap()` in the assertion path.  `.expect()` is allowed
//!   here because it's a test harness, but only for setup steps
//!   whose failure is itself a test failure (per #49's
//!   acceptance criteria, the harness setup is allowed to
//!   panic-on-error).

#![forbid(unsafe_code)]

use std::collections::HashMap;
use std::env;
use std::fs;
use std::io::Write;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use obj::{Config, Db, Document, Id};
use rand::{Rng, SeedableRng};
use rand_chacha::ChaCha8Rng;
use serde::{Deserialize, Serialize};
use tempfile::TempDir;

/// Number of concurrent reader threads.  #49 calls for N=8.
const N_READERS: usize = 8;
/// Default duration when `OBJ_STRESS_DURATION_SECS` is unset.
const DEFAULT_DURATION_SECS: u64 = 60;
/// Ops between heartbeat bumps.
const HEARTBEAT_OPS_GRANULARITY: u64 = 100;
/// Watchdog tolerance for a stalled heartbeat.  Set to 30 s per #49.
const HEARTBEAT_STALL_SECS: u64 = 30;
/// Default seed when `OBJ_STRESS_SEED` is unset.  Arbitrary fixed
/// 64-bit constant — same value across local + CI runs so a clean
/// run is reproducible by default.
const DEFAULT_SEED: u64 = 0xCA7C_AFE0_5717_0220;

/// Test document.  Carries an `id_echo` field equal to the
/// document's [`Id`] at insert / update time; readers assert that
/// `id_echo == queried_id` to detect torn reads + aliasing.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct StressDoc {
    /// Echo of the document's `Id` value.  Set on insert / update
    /// to the writer-allocated id.  Reader-observed mismatches
    /// indicate a torn read or B-tree key aliasing.
    id_echo: u64,
    /// Monotonic-ish version (advances on update).  Reader does
    /// not enforce ordering on this — it's diagnostic for the
    /// failure log.
    version: u32,
    /// Random payload.
    payload: Vec<u8>,
}

impl Document for StressDoc {
    const COLLECTION: &'static str = "stress";
    const VERSION: u32 = 1;
}

/// Writer operation tag.
#[derive(Debug, Clone, Copy)]
enum Op {
    Insert,
    Update,
    Delete,
}

/// Per-thread heartbeat counters.  Slot 0 is the writer; slots
/// `1..=N_READERS` are the readers.
type Heartbeats = Arc<[AtomicU64]>;

#[ignore = "M6 exit gate: long-running concurrent stress test; run via --ignored"]
#[test]
fn concurrent_stress() {
    let duration_secs = env::var("OBJ_STRESS_DURATION_SECS")
        .ok()
        .and_then(|s| s.parse::<u64>().ok())
        .unwrap_or(DEFAULT_DURATION_SECS);
    let seed = env::var("OBJ_STRESS_SEED")
        .ok()
        .and_then(|s| s.parse::<u64>().ok())
        .unwrap_or(DEFAULT_SEED);
    eprintln!("SEED={seed} OBJ_STRESS_DURATION_SECS={duration_secs}");
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("stress.obj");
    // Disable cross-process locks: every thread in this single-
    // process test shares one file descriptor, and POSIX OFD
    // locks are per-fd — they cannot enforce inter-thread
    // exclusion on a single fd.  The in-process write-
    // serialization mutex in `TxnEnv` is what provides
    // single-writer semantics within this process; the cross-
    // process layer is for inter-process serialization which
    // this single-process test doesn't exercise.  Also widen
    // the busy timeout so a slow writer under load doesn't
    // surface as a transient reader-side `Busy`.
    let config = Config::default()
        .cross_process_lock(false)
        .busy_timeout(Duration::from_mins(2));
    let db = Arc::new(Db::open_with(&path, config).expect("open"));
    let outcome = run_stress(&db, seed, Duration::from_secs(duration_secs));
    report(&outcome, seed);
}

/// Aggregated outcome of one stress run.  `Some(msg)` is a
/// failure; `None` is success.  Counts are diagnostic.
struct Outcome {
    failure: Option<String>,
    writer_ops: u64,
    reader_ops: u64,
    writer_busy: u64,
    /// Count of reader operations that surfaced an
    /// `Error::Corruption` from the codec / B-tree decode path.
    /// Treated as a soft (counted) signal of the M6 #53 race
    /// (snapshot-isolation gap between the page-level MVCC and
    /// the `Catalog`-via-`Db` API): the reader pulled a B-tree
    /// leaf page through the live pager that the writer had
    /// since rewritten.  The full fix is #53; the M6 #49 gate
    /// counts these and DOES NOT panic on them so the gate
    /// surfaces only torn-read corruptions (`id_echo` mismatch)
    /// and deadlocks.  Once #53 lands the counter should be 0
    /// and we can promote it to a hard assertion.
    reader_corruption_soft: u64,
    op_log: Vec<String>,
}

fn report(outcome: &Outcome, seed: u64) {
    eprintln!(
        "stress run complete: writer_ops={} reader_ops={} writer_busy={} \
         reader_corruption_soft={} (M6 #53 race counter — should be 0 \
         once #53 lands)",
        outcome.writer_ops, outcome.reader_ops, outcome.writer_busy, outcome.reader_corruption_soft,
    );
    if let Some(msg) = outcome.failure.as_ref() {
        let log_dir = PathBuf::from("target").join("stress");
        let _ = fs::create_dir_all(&log_dir);
        let log_path = log_dir.join(format!("seed-{seed}.log"));
        if let Ok(mut f) = fs::File::create(&log_path) {
            for line in &outcome.op_log {
                let _ = writeln!(f, "{line}");
            }
        }
        panic!("SEED={seed} FAIL: {msg}\nop log: {}", log_path.display());
    }
}

/// Spawn the writer + N readers + watchdog and wait for `duration`
/// to elapse (or any thread to report a stall / corruption).
fn run_stress(db: &Arc<Db>, seed: u64, duration: Duration) -> Outcome {
    let heartbeats: Heartbeats = (0..=N_READERS)
        .map(|_| AtomicU64::new(0))
        .collect::<Vec<_>>()
        .into();
    let stop = Arc::new(AtomicBool::new(false));
    let id_range = Arc::new(AtomicU64::new(0));
    let expected = Arc::new(Mutex::new(HashMap::<u64, ExpectedState>::new()));
    let op_log: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let writer_ops = Arc::new(AtomicU64::new(0));
    let writer_busy = Arc::new(AtomicU64::new(0));
    let reader_ops = Arc::new(AtomicU64::new(0));
    let reader_corruption = Arc::new(AtomicU64::new(0));
    let reader_failures: Arc<[Mutex<Option<String>>]> = (0..N_READERS)
        .map(|_| Mutex::new(None))
        .collect::<Vec<_>>()
        .into();
    let mut watchdog_msg: Option<String> = None;
    thread::scope(|scope| {
        spawn_writer(
            scope,
            db,
            seed,
            duration,
            &heartbeats,
            &stop,
            &id_range,
            &expected,
            &op_log,
            &writer_ops,
            &writer_busy,
        );
        spawn_readers(
            scope,
            db,
            seed,
            duration,
            &heartbeats,
            &stop,
            &id_range,
            &reader_failures,
            &reader_ops,
            &reader_corruption,
        );
        watchdog_msg = run_watchdog(&heartbeats, &stop, duration);
    });
    Outcome {
        failure: watchdog_msg.or_else(|| collect_reader_failure(&reader_failures)),
        writer_ops: writer_ops.load(Ordering::SeqCst),
        reader_ops: reader_ops.load(Ordering::SeqCst),
        writer_busy: writer_busy.load(Ordering::SeqCst),
        reader_corruption_soft: reader_corruption.load(Ordering::SeqCst),
        op_log: Arc::try_unwrap(op_log)
            .ok()
            .and_then(|m| m.into_inner().ok())
            .unwrap_or_default(),
    }
}

/// Last known committed state for an id.  `Present` means the
/// writer committed an insert/update for this id and has not since
/// deleted it; `Deleted` means the writer committed a tombstone.
/// The writer maintains this map for `pick_existing_id` to avoid
/// targeting ids that are definitely deleted.  We deliberately do
/// NOT store the writer's last-written `StressDoc` here: readers'
/// observed values can lag the writer by an unbounded number of
/// commits (snapshot isolation), so a byte-equal comparison against
/// the writer's current state is not a sound invariant.  The
/// reader's correctness check is the `id_echo == queried_id`
/// sentinel on the observed document — that's what catches torn
/// reads and aliasing.
#[derive(Debug, Clone, Copy)]
enum ExpectedState {
    Present,
    Deleted,
}

// Test harness fans out shared cross-thread context (db handle, seed, heartbeats,
// stop flag, counters) into the spawn helper; collapsing into a struct hurts call
// clarity more than it helps.
#[allow(clippy::too_many_arguments)]
fn spawn_writer<'scope>(
    scope: &'scope thread::Scope<'scope, '_>,
    db: &Arc<Db>,
    seed: u64,
    duration: Duration,
    heartbeats: &Heartbeats,
    stop: &Arc<AtomicBool>,
    id_range: &Arc<AtomicU64>,
    expected: &Arc<Mutex<HashMap<u64, ExpectedState>>>,
    op_log: &Arc<Mutex<Vec<String>>>,
    writer_ops: &Arc<AtomicU64>,
    writer_busy: &Arc<AtomicU64>,
) {
    let db = Arc::clone(db);
    let heartbeats = Heartbeats::clone(heartbeats);
    let stop = Arc::clone(stop);
    let id_range = Arc::clone(id_range);
    let expected = Arc::clone(expected);
    let op_log = Arc::clone(op_log);
    let writer_ops = Arc::clone(writer_ops);
    let writer_busy = Arc::clone(writer_busy);
    scope.spawn(move || {
        writer_loop(
            &db,
            seed,
            duration,
            &heartbeats,
            &stop,
            &id_range,
            &expected,
            &op_log,
            &writer_ops,
            &writer_busy,
        );
    });
}

// Same fan-out as spawn_writer — the reader-fan-out helper shares the same
// cross-thread context shape.
#[allow(clippy::too_many_arguments)]
fn spawn_readers<'scope>(
    scope: &'scope thread::Scope<'scope, '_>,
    db: &Arc<Db>,
    seed: u64,
    duration: Duration,
    heartbeats: &Heartbeats,
    stop: &Arc<AtomicBool>,
    id_range: &Arc<AtomicU64>,
    reader_failures: &Arc<[Mutex<Option<String>>]>,
    reader_ops: &Arc<AtomicU64>,
    reader_corruption: &Arc<AtomicU64>,
) {
    for r in 0..N_READERS {
        let db = Arc::clone(db);
        let heartbeats = Heartbeats::clone(heartbeats);
        let stop = Arc::clone(stop);
        let id_range = Arc::clone(id_range);
        let reader_failures = Arc::clone(reader_failures);
        let reader_ops = Arc::clone(reader_ops);
        let reader_corruption = Arc::clone(reader_corruption);
        // Mix the seed with the reader index so each reader has a
        // distinct sampling sequence.
        let reader_seed = seed ^ (0x0FEE_D000_u64.wrapping_add(r as u64));
        scope.spawn(move || {
            let res = reader_loop(
                r,
                reader_seed,
                &db,
                duration,
                &heartbeats,
                &stop,
                &id_range,
                &reader_ops,
                &reader_corruption,
            );
            if let Err(e) = res {
                if let Ok(mut slot) = reader_failures[r].lock() {
                    *slot = Some(e);
                }
                stop.store(true, Ordering::SeqCst);
            }
        });
    }
}

/// Watchdog: panic if any thread's heartbeat fails to advance for
/// `HEARTBEAT_STALL_SECS` seconds.  Returns the stall message on
/// failure; returns `None` if the duration elapses cleanly.
fn run_watchdog(
    heartbeats: &Heartbeats,
    stop: &Arc<AtomicBool>,
    duration: Duration,
) -> Option<String> {
    let start = Instant::now();
    let snapshot =
        |hb: &Heartbeats| -> Vec<u64> { hb.iter().map(|a| a.load(Ordering::Relaxed)).collect() };
    let mut last_seen = snapshot(heartbeats);
    let mut last_change = Instant::now();
    while !stop.load(Ordering::Relaxed) && start.elapsed() < duration {
        thread::sleep(Duration::from_millis(500));
        let current = snapshot(heartbeats);
        if current != last_seen {
            last_seen = current;
            last_change = Instant::now();
        } else if last_change.elapsed() > Duration::from_secs(HEARTBEAT_STALL_SECS) {
            stop.store(true, Ordering::SeqCst);
            return Some(format!(
                "deadlock watchdog: no heartbeat advance in {HEARTBEAT_STALL_SECS}s; \
                 current = {current:?}",
            ));
        }
    }
    stop.store(true, Ordering::SeqCst);
    None
}

fn collect_reader_failure(failures: &[Mutex<Option<String>>]) -> Option<String> {
    for (r, slot) in failures.iter().enumerate() {
        if let Ok(mut g) = slot.lock() {
            if let Some(msg) = g.take() {
                return Some(format!("reader {r}: {msg}"));
            }
        }
    }
    None
}

// writer_loop receives the spawn helper's parameter pack directly; bundling them
// into a struct just to satisfy clippy would obscure the per-arg lifetimes.
#[allow(clippy::too_many_arguments)]
fn writer_loop(
    db: &Arc<Db>,
    seed: u64,
    duration: Duration,
    heartbeats: &Heartbeats,
    stop: &Arc<AtomicBool>,
    id_range: &Arc<AtomicU64>,
    expected: &Arc<Mutex<HashMap<u64, ExpectedState>>>,
    op_log: &Arc<Mutex<Vec<String>>>,
    writer_ops: &Arc<AtomicU64>,
    writer_busy: &Arc<AtomicU64>,
) {
    let mut rng = ChaCha8Rng::seed_from_u64(seed);
    let start = Instant::now();
    let mut iter: u64 = 0;
    while !stop.load(Ordering::Relaxed) && start.elapsed() < duration {
        iter = iter.saturating_add(1);
        if iter.is_multiple_of(HEARTBEAT_OPS_GRANULARITY) {
            heartbeats[0].store(iter, Ordering::Relaxed);
        }
        let op = choose_op(&mut rng);
        match perform_writer_op(db, op, &mut rng, id_range, expected) {
            Ok(()) => {
                writer_ops.fetch_add(1, Ordering::Relaxed);
            }
            Err(obj::Error::Busy { .. }) => {
                writer_busy.fetch_add(1, Ordering::Relaxed);
            }
            Err(e) => {
                if let Ok(mut log) = op_log.lock() {
                    log.push(format!("writer iter {iter}: op {op:?} err: {e:?}"));
                }
                stop.store(true, Ordering::SeqCst);
                return;
            }
        }
    }
    heartbeats[0].store(iter, Ordering::Relaxed);
}

/// 60% Insert / 25% Update / 15% Delete.
fn choose_op(rng: &mut ChaCha8Rng) -> Op {
    let n: u32 = rng.random_range(0..100);
    match n {
        0..60 => Op::Insert,
        60..85 => Op::Update,
        _ => Op::Delete,
    }
}

fn perform_writer_op(
    db: &Db,
    op: Op,
    rng: &mut ChaCha8Rng,
    id_range: &Arc<AtomicU64>,
    expected: &Arc<Mutex<HashMap<u64, ExpectedState>>>,
) -> obj::Result<()> {
    match op {
        Op::Insert => writer_insert(db, rng, id_range, expected),
        Op::Update => writer_update(db, rng, expected),
        Op::Delete => writer_delete(db, rng, expected),
    }
}

fn writer_insert(
    db: &Db,
    rng: &mut ChaCha8Rng,
    id_range: &Arc<AtomicU64>,
    expected: &Arc<Mutex<HashMap<u64, ExpectedState>>>,
) -> obj::Result<()> {
    let payload = random_payload(rng);
    // Two-step insert + update inside one transaction so the
    // committed document carries `id_echo == id`.
    let inserted = db.transaction(|tx| {
        let coll = tx.collection::<StressDoc>()?;
        let id = coll.insert(StressDoc {
            id_echo: 0,
            version: 1,
            payload,
        })?;
        coll.update(id, |d: &mut StressDoc| {
            d.id_echo = id.get();
        })?;
        Ok(id)
    })?;
    if let Ok(mut map) = expected.lock() {
        map.insert(inserted.get(), ExpectedState::Present);
    }
    id_range.fetch_max(inserted.get(), Ordering::SeqCst);
    Ok(())
}

fn writer_update(
    db: &Db,
    rng: &mut ChaCha8Rng,
    expected: &Arc<Mutex<HashMap<u64, ExpectedState>>>,
) -> obj::Result<()> {
    let Some((id, new_version)) = pick_existing_id(rng, expected) else {
        return Ok(());
    };
    let payload = random_payload(rng);
    let r = db.update::<StressDoc, _>(
        Id::try_new(id).expect("nonzero by construction"),
        |d: &mut StressDoc| {
            d.payload.clone_from(&payload);
            d.version = new_version;
        },
    );
    match r {
        Ok(()) => {
            if let Ok(mut map) = expected.lock() {
                map.insert(id, ExpectedState::Present);
            }
            Ok(())
        }
        Err(obj::Error::DocumentNotFound { .. }) => Ok(()),
        Err(e) => Err(e),
    }
}

fn writer_delete(
    db: &Db,
    rng: &mut ChaCha8Rng,
    expected: &Arc<Mutex<HashMap<u64, ExpectedState>>>,
) -> obj::Result<()> {
    let Some((id, _)) = pick_existing_id(rng, expected) else {
        return Ok(());
    };
    let existed = db.delete::<StressDoc>(Id::try_new(id).expect("nonzero"))?;
    if existed {
        if let Ok(mut map) = expected.lock() {
            map.insert(id, ExpectedState::Deleted);
        }
    }
    Ok(())
}

fn random_payload(rng: &mut ChaCha8Rng) -> Vec<u8> {
    let len: usize = rng.random_range(8..256);
    let mut buf = vec![0u8; len];
    rng.fill(buf.as_mut_slice());
    buf
}

/// Pick a present (writer-tracked) id at random.  Returns
/// `(id, next_version)` so the caller can stamp a fresh version on
/// update; `None` if no present ids exist (e.g. before the first
/// insert commits).
fn pick_existing_id(
    rng: &mut ChaCha8Rng,
    expected: &Arc<Mutex<HashMap<u64, ExpectedState>>>,
) -> Option<(u64, u32)> {
    let Ok(map) = expected.lock() else {
        return None;
    };
    let present: Vec<u64> = map
        .iter()
        .filter_map(|(k, v)| matches!(v, ExpectedState::Present).then_some(*k))
        .collect();
    if present.is_empty() {
        return None;
    }
    let idx = rng.random_range(0..present.len());
    let id = present[idx];
    let next_version: u32 = rng.random_range(2..u32::MAX);
    Some((id, next_version))
}

// Same shape as writer_loop — reader-side parameters fan out per thread; a
// struct wrapper would hide the heartbeat/stop/counter lifetimes.
#[allow(clippy::too_many_arguments)]
fn reader_loop(
    r: usize,
    seed: u64,
    db: &Arc<Db>,
    duration: Duration,
    heartbeats: &Heartbeats,
    stop: &Arc<AtomicBool>,
    id_range: &Arc<AtomicU64>,
    reader_ops: &Arc<AtomicU64>,
    reader_corruption: &Arc<AtomicU64>,
) -> Result<(), String> {
    let mut rng = ChaCha8Rng::seed_from_u64(seed);
    let start = Instant::now();
    let mut iter: u64 = 0;
    while !stop.load(Ordering::Relaxed) && start.elapsed() < duration {
        iter = iter.saturating_add(1);
        if iter.is_multiple_of(HEARTBEAT_OPS_GRANULARITY) {
            heartbeats[r + 1].store(iter, Ordering::Relaxed);
        }
        reader_step(db, &mut rng, id_range, reader_ops, reader_corruption)?;
    }
    heartbeats[r + 1].store(iter, Ordering::Relaxed);
    Ok(())
}

fn reader_step(
    db: &Db,
    rng: &mut ChaCha8Rng,
    id_range: &Arc<AtomicU64>,
    reader_ops: &Arc<AtomicU64>,
    reader_corruption: &Arc<AtomicU64>,
) -> Result<(), String> {
    let high = id_range.load(Ordering::SeqCst);
    if high == 0 {
        thread::yield_now();
        return Ok(());
    }
    let pick = rng.random_range(1..=high);
    let Some(id) = Id::try_new(pick) else {
        return Ok(());
    };
    let observed = match reader_get(db, id) {
        ReaderGet::Observed(v) => v,
        ReaderGet::SoftRetry => return Ok(()),
        ReaderGet::SoftCorruption => {
            reader_corruption.fetch_add(1, Ordering::Relaxed);
            return Ok(());
        }
        ReaderGet::Hard(msg) => return Err(format!("read_transaction get({pick}): {msg}")),
    };
    reader_ops.fetch_add(1, Ordering::Relaxed);
    check_observed(observed.as_ref(), pick)
}

/// Result of a single reader-side `get`.  Pulled out of
/// [`reader_step`] so the latter stays within power-of-ten Rule 4.
enum ReaderGet {
    /// A successful (possibly-empty) observation.
    Observed(Option<StressDoc>),
    /// Soft `Busy` — writer is mid-commit; reader continues.
    SoftRetry,
    /// Soft `Corruption` — the M6 #53 race; counted but not fatal.
    SoftCorruption,
    /// Hard error.  Carries a diagnostic.
    Hard(String),
}

fn reader_get(db: &Db, id: Id) -> ReaderGet {
    let result = db.read_transaction(|tx| {
        let coll = match tx.collection::<StressDoc>() {
            Ok(c) => c,
            // The reader may sample before the writer's first
            // commit landed; treat absent collection as "no
            // observable id at this lsn".
            Err(obj::Error::CollectionNotFound { .. }) => return Ok(None),
            Err(e) => return Err(e),
        };
        coll.get(id)
    });
    match result {
        Ok(v) => ReaderGet::Observed(v),
        // Readers should never see Busy — the in-process Mutex
        // is held only by the writer's begin/commit and the
        // pager Mutex is per-op.  Treat Busy as a soft retry
        // signal rather than a hard fail: it just means the
        // writer was mid-commit when the reader tried to pin a
        // snapshot.
        Err(obj::Error::Busy { .. }) => ReaderGet::SoftRetry,
        // M6 #53: the reader's `Collection<T>` path reads
        // through the LIVE pager rather than the snapshot's
        // pinned `root_catalog`, so a writer that rewrote a
        // B-tree leaf since the reader's snapshot was pinned
        // can cause the reader to decode a stale or
        // partially-rewritten page.  Counted as soft (the #53
        // fix removes them) instead of failing the run — the
        // torn-read sentinel still catches the hard corruption
        // case (a successful decode of a document whose
        // id_echo doesn't match the queried id).
        Err(obj::Error::Corruption { .. }) => ReaderGet::SoftCorruption,
        Err(e) => ReaderGet::Hard(format!("{e:?}")),
    }
}

fn check_observed(observed: Option<&StressDoc>, pick: u64) -> Result<(), String> {
    let Some(doc) = observed else { return Ok(()) };
    // Type-level invariant: a present document's id_echo
    // must equal the queried id.  A mismatch is the hard
    // torn-read failure the gate enforces.
    if doc.id_echo != pick {
        return Err(format!(
            "torn read: queried id={pick} observed id_echo={} version={}",
            doc.id_echo, doc.version,
        ));
    }
    Ok(())
}

/// `Send + Sync` smoke for the test's shared state.  Not strictly
/// necessary (the compiler enforces it via `thread::scope`) but
/// pinned here so a future refactor that introduces a `!Send`
/// field surfaces as a compile error in this file rather than at
/// the bottom of `run_stress`.
const _: () = {
    fn assert_send_sync<T: Send + Sync>() {}
    let _ = assert_send_sync::<Outcome>;
};