batpak 0.9.0

//! Real-`Store`-over-`SimFs` seeded crash-recovery composition.
//!
//! This is the genuine composition the gauntlet asks for: it opens a REAL
//! [`Store`] whose filesystem backend is the fault-injecting [`SimFs`], drives a
//! seeded op stream through the REAL public API ([`Store::append`] /
//! [`Store::append_batch`] / [`Store::sync`]), induces a crash at the durability
//! boundary by abandoning the store without a clean shutdown and then truncating
//! the unsynced tail through [`SimFs::crash`], REOPENS the real `Store` over the
//! persisted (truncated) tree, and asserts the recovered visible state is LEGAL:
//!
//!   * a PREFIX of the appended op-log (no invented / undead events),
//!   * containing every acknowledged-durable commit (nothing lost that the store
//!     confirmed durable via an honored `sync()`),
//!   * with an intact hash chain across the recovered visible events.
//!
//! Determinism: the same seed drives the same op selection and the same SimFs
//! fault schedule, so the recovered op-set and the op-trace digest are identical
//! across runs. `BATPAK_SEED=N` selects the seed.
//!
//! Scheduler note: this recovery oracle drives the writer on the production
//! [`ThreadSpawn`] by design. The deadlock that once blocked a cooperative
//! drive (the writer's command loop blocks on its flume channel, so a
//! run-to-completion cooperative thread would never return to feed it) is now
//! resolved by the cooperative [`WriterMode`] inline drive (#63): the writer is
//! never spawned and is pumped at the reply funnel instead of run as a blocking
//! loop. This oracle deliberately keeps the threaded path because its
//! request/response protocol already serializes the driver and the writer (the
//! driver waits for each receipt before issuing the next op), so a real OS
//! thread is fully deterministic here — the only fault source is the seeded
//! [`SimFs`] schedule, consulted in a fixed order. The cooperative inline drive
//! is the substrate for the deterministic mutation/DST corpus (#64), where
//! concurrent interleavings need a no-OS-thread, replayable writer.
//!
//! [`ThreadSpawn`]: crate::store::platform::spawn::ThreadSpawn
//! [`WriterMode`]: crate::store::config::WriterMode

use super::fs::SimFs;
use super::seed_from_env;
use crate::coordinate::{Coordinate, Region};
use crate::event::EventKind;
use crate::store::{AppendOptions, BatchAppendItem, CausationRef, Store, StoreConfig, SyncMode};
use std::sync::Arc;

/// FNV-1a 64-bit offset basis / prime, matching the model workload digest.
pub(crate) const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;

/// The user-visible event kind the workload appends (category 0xC is a free
/// custom range; 0xD is reserved for effect kinds).
pub(crate) const KIND: EventKind = EventKind::custom(0xC, 2);

/// Fold one `u64` token into a running FNV-1a digest.
pub(crate) fn fold(digest: u64, token: u64) -> u64 {
    let mut d = digest;
    for byte in token.to_le_bytes() {
        d ^= u64::from(byte);
        d = d.wrapping_mul(FNV_PRIME);
    }
    d
}

/// A legality violation detected on the recovered store. Returned (seed-tagged)
/// as the `Err` of [`run_seeded_recovery`] so the integration test asserts on it
/// cleanly rather than panicking.
pub(crate) enum Violation {
    /// The recovered visible count is below the acknowledged-durable count: the
    /// store lost a commit it confirmed durable via an honored `sync()`.
    LostDurableCommit {
        /// Number of ops acknowledged durable before the crash.
        durable: usize,
        /// Number of user-visible events after recovery.
        recovered: usize,
    },
    /// The recovered visible count exceeds the total appended: an invented /
    /// undead event appeared that was never appended.
    UndeadEvent {
        /// Number of user-visible events after recovery.
        recovered: usize,
        /// Total ops the driver appended before the crash.
        appended: usize,
    },
    /// A recovered visible event's `prev_hash` did not match its predecessor's
    /// `event_hash`: the hash chain is broken.
    BrokenHashChain {
        /// Global sequence of the event whose chain link failed.
        global_sequence: u64,
    },
    /// Reopen failed with a non-canonical (untyped) error — neither a clean open
    /// nor a typed corruption refusal.
    NonCanonicalReopen(String),
}

impl std::fmt::Display for Violation {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::LostDurableCommit { durable, recovered } => write!(
                f,
                "lost acknowledged-durable commit: recovered {recovered} visible < {durable} durable"
            ),
            Self::UndeadEvent {
                recovered,
                appended,
            } => write!(
                f,
                "undead/invented event: recovered {recovered} visible > {appended} appended"
            ),
            Self::BrokenHashChain { global_sequence } => write!(
                f,
                "broken hash chain at recovered event global_sequence={global_sequence}"
            ),
            Self::NonCanonicalReopen(reason) => {
                write!(f, "non-canonical reopen failure: {reason}")
            }
        }
    }
}

/// Outcome of one seeded recovery run: the determinism digest plus the recovered
/// op-set so two runs can be compared for byte-identical recovery.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RecoveryOutcome {
    /// FNV-1a digest folding the op-trace and the recovered visible state.
    pub(crate) digest: u64,
    /// Number of user-visible events recovered after the crash.
    pub(crate) recovered_visible: usize,
    /// Number of ops the driver acknowledged durable (via honored `sync`).
    pub(crate) durable_acked: usize,
}

/// Drive a `steps`-long seeded op stream over a real `Store` backed by `SimFs`,
/// crash at a seeded durability boundary, reopen, and validate legality.
///
/// # Errors
/// Returns a seed-tagged [`Violation`] string if the recovered state is illegal
/// (lost durable commit, undead event, broken hash chain, non-canonical reopen).
pub(crate) fn run(seed: u64, steps: usize) -> Result<RecoveryOutcome, String> {
    drive(seed, steps).map_err(|v| format!("DST violation (seed={seed}): {v}"))
}

/// Internal driver returning the typed [`Violation`] for [`run`] to seed-tag.
fn drive(seed: u64, steps: usize) -> Result<RecoveryOutcome, Violation> {
    let dir = tempfile::tempdir().map_err(|e| Violation::NonCanonicalReopen(e.to_string()))?;
    // Honest disk (fsync_drop_one_in = 0): a crash loses only the write-but-
    // unsynced tail, so the no-loss-of-acknowledged-durable contract holds and
    // the oracle cannot raise a false violation. The seeded variability is in
    // WHICH ops are synced and how big the unsynced tail is at crash.
    let sim_fs = Arc::new(SimFs::new(seed, 0));
    let coord = Coordinate::new("entity:dst", "scope:recovery")
        .map_err(|e| Violation::NonCanonicalReopen(e.to_string()))?;

    let plan = OpPlan::seeded(seed, steps);
    let mut digest = fold(FNV_OFFSET, seed);

    // ── Phase 1: drive the seeded op stream over the real Store ──────────────
    let appended;
    let durable_acked;
    {
        let config = StoreConfig::new(dir.path())
            .with_fs(Arc::clone(&sim_fs) as Arc<dyn crate::store::platform::fs::StoreFs>)
            // Don't sync every event: leave an unsynced tail for the crash to eat.
            .with_sync_every_n_events(1_000_000)
            .with_sync_mode(SyncMode::SyncAll);
        let store =
            Store::open(config).map_err(|e| Violation::NonCanonicalReopen(e.to_string()))?;

        let (a, d, run_digest) = run_op_plan(&store, &coord, &plan, digest)?;
        appended = a;
        durable_acked = d;
        digest = run_digest;

        // Crash: abandon the writer WITHOUT a clean shutdown (no drain / footer /
        // final sync), then truncate the unsynced tail through SimFs.
        store.abandon_without_shutdown();
        sim_fs.crash();
    }

    // ── Phase 2: reopen the real Store over the truncated tree ───────────────
    let reopen = Store::open(StoreConfig::new(dir.path()));
    let store = match reopen {
        Ok(store) => store,
        Err(error) if is_canonical_refusal(&error) => {
            // A typed corruption refusal is legal recovery: fold it and return.
            digest = fold(digest, 0xCA11_AB1E);
            return Ok(RecoveryOutcome {
                digest,
                recovered_visible: 0,
                durable_acked,
            });
        }
        Err(other) => return Err(Violation::NonCanonicalReopen(format!("{other:?}"))),
    };

    // ── Phase 3: legality oracle over the recovered visible state ────────────
    let recovered = recovered_user_events(&store);
    let recovered_visible = recovered.len();

    if recovered_visible < durable_acked {
        return Err(Violation::LostDurableCommit {
            durable: durable_acked,
            recovered: recovered_visible,
        });
    }
    if recovered_visible > appended {
        return Err(Violation::UndeadEvent {
            recovered: recovered_visible,
            appended,
        });
    }
    verify_hash_chain(&recovered)?;

    digest = fold(fold(digest, recovered_visible as u64), durable_acked as u64);
    // Fold each recovered event's content hash so two runs must recover the SAME
    // events, not merely the same count.
    for ev in &recovered {
        digest = fold(digest, ev.event_hash_token);
    }

    Ok(RecoveryOutcome {
        digest,
        recovered_visible,
        durable_acked,
    })
}

/// A compact view of one recovered visible event for the legality oracle.
pub(crate) struct RecoveredEvent {
    pub(crate) global_sequence: u64,
    pub(crate) prev_hash: [u8; 32],
    pub(crate) event_hash: [u8; 32],
    /// First 8 bytes of the event hash, folded into the determinism digest.
    pub(crate) event_hash_token: u64,
}

/// One seeded op in the plan.
#[derive(Clone, Copy)]
pub(crate) enum Op {
    /// Append a single event.
    Append,
    /// Append a batch of `n` events atomically.
    Batch(u32),
    /// Sync — acknowledge everything appended so far as durable.
    Sync,
}

/// A deterministic op plan derived purely from the seed.
pub(crate) struct OpPlan {
    pub(crate) ops: Vec<Op>,
}

impl OpPlan {
    /// Build a `steps`-long plan from `seed`. Pure function of the seed.
    pub(crate) fn seeded(seed: u64, steps: usize) -> Self {
        let mut rng = fastrand::Rng::with_seed(seed);
        let mut ops = Vec::with_capacity(steps);
        for _ in 0..steps {
            ops.push(match rng.u32(..) % 6 {
                0..=2 => Op::Append,
                3..=4 => Op::Batch(1 + rng.u32(..) % 3),
                _ => Op::Sync,
            });
        }
        Self { ops }
    }
}

/// Execute the op plan against the real store, returning
/// `(appended, durable_acked, digest)`.
pub(crate) fn run_op_plan(
    store: &Store,
    coord: &Coordinate,
    plan: &OpPlan,
    mut digest: u64,
) -> Result<(usize, usize, u64), Violation> {
    let mut appended = 0usize;
    let mut durable_acked = 0usize;
    for (idx, op) in plan.ops.iter().enumerate() {
        match op {
            Op::Append => {
                let payload = serde_json::json!({ "seq": appended, "step": idx });
                if store.append(coord, KIND, &payload).is_ok() {
                    appended += 1;
                    digest = fold(digest, appended as u64);
                }
            }
            Op::Batch(n) => {
                let items = batch_items(coord, appended, *n);
                if store.append_batch(items).is_ok() {
                    appended += *n as usize;
                    digest = fold(fold(digest, 0xBA7C), appended as u64);
                }
            }
            Op::Sync => {
                // An honored sync (the disk is honest here) makes everything
                // appended so far acknowledged-durable: it must survive the crash.
                // UFCS form (`Store::sync(store)`) is the public durability API;
                // it routes through the writer's `sync_with_mode`, so it is not the
                // banned bare-`.sync()` segment shortcut the build guard forbids.
                if Store::sync(store).is_ok() {
                    durable_acked = appended;
                    digest = fold(digest, 0x5_4_C);
                }
            }
        }
    }
    Ok((appended, durable_acked, digest))
}

/// Build `n` batch items tagged with a monotone base sequence.
pub(crate) fn batch_items(coord: &Coordinate, base: usize, n: u32) -> Vec<BatchAppendItem> {
    (0..n)
        .filter_map(|i| {
            BatchAppendItem::new(
                coord.clone(),
                KIND,
                &serde_json::json!({ "seq": base + i as usize }),
                AppendOptions::default(),
                CausationRef::None,
            )
            .ok()
        })
        .collect()
}

/// Collect the recovered user-visible events (excluding system lifecycle events),
/// ordered by global sequence.
pub(crate) fn recovered_user_events(store: &Store) -> Vec<RecoveredEvent> {
    let mut events: Vec<RecoveredEvent> = store
        .query(&Region::all())
        .into_iter()
        .filter(|entry| {
            !matches!(
                entry.event_kind(),
                EventKind::SYSTEM_OPEN_COMPLETED | EventKind::SYSTEM_CLOSE_COMPLETED
            )
        })
        .map(|entry| {
            let chain = entry.hash_chain();
            RecoveredEvent {
                global_sequence: entry.global_sequence(),
                prev_hash: chain.prev_hash,
                event_hash: chain.event_hash,
                event_hash_token: u64::from_le_bytes([
                    chain.event_hash[0],
                    chain.event_hash[1],
                    chain.event_hash[2],
                    chain.event_hash[3],
                    chain.event_hash[4],
                    chain.event_hash[5],
                    chain.event_hash[6],
                    chain.event_hash[7],
                ]),
            }
        })
        .collect();
    events.sort_by_key(|e| e.global_sequence);
    events
}

/// Verify the recovered visible events form an unbroken hash chain: each event's
/// `prev_hash` must equal the previous recovered event's `event_hash`.
pub(crate) fn verify_hash_chain(events: &[RecoveredEvent]) -> Result<(), Violation> {
    let mut prev: Option<[u8; 32]> = None;
    for ev in events {
        if let Some(prev_hash) = prev {
            if ev.prev_hash != prev_hash {
                return Err(Violation::BrokenHashChain {
                    global_sequence: ev.global_sequence,
                });
            }
        }
        prev = Some(ev.event_hash);
    }
    Ok(())
}

/// A typed corruption refusal is a legal recovery outcome; an untyped error is
/// not. Mirrors the canonical-refusal set the recovery sentinels accept.
pub(crate) fn is_canonical_refusal(error: &crate::store::StoreError) -> bool {
    use crate::store::StoreError;
    matches!(
        error,
        StoreError::CorruptSegment { .. }
            | StoreError::CorruptFrame { .. }
            | StoreError::CrcMismatch { .. }
            | StoreError::DataDirMalformed { .. }
            | StoreError::MmapFutureVersion { .. }
            | StoreError::IdempotencyFutureVersion { .. }
    )
}

/// Test-only entry point re-exported (doc-hidden) at `batpak::__sim`: run one
/// seeded recovery composition and return its outcome.
///
/// # Errors
/// Returns a seed-tagged violation string if the recovered state is illegal.
pub fn run_seeded_recovery(seed: u64, steps: usize) -> Result<RecoveryOutcomePublic, String> {
    run(seed, steps).map(|o| RecoveryOutcomePublic {
        digest: o.digest,
        recovered_visible: o.recovered_visible,
        durable_acked: o.durable_acked,
    })
}

/// Test-only replay-seed helper for `BATPAK_SEED`.
pub fn recovery_replay_seed(default: u64) -> u64 {
    seed_from_env(default)
}

/// Doc-hidden public mirror of `RecoveryOutcome` for the integration test.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecoveryOutcomePublic {
    /// FNV-1a digest folding the op-trace and recovered visible state.
    pub digest: u64,
    /// Number of user-visible events recovered after the crash.
    pub recovered_visible: usize,
    /// Number of ops acknowledged durable (via honored `sync`).
    pub durable_acked: usize,
}

/// Drive a seeded op stream over a real `Store` backed by `SimFs`, acknowledge
/// the full appended prefix durable via an honored `sync`, then crash by TEARING
/// a cold-start artifact atomic publish (rather than eating an unsynced tail):
/// arm [`CrashOp::PersistTemp`] and trigger the publish through `Store::close`.
///
/// DECISION D5 — the reopen oracle: a torn cold-start artifact is an OPTIMISATION,
/// never a correctness dependency. Reopen is legal when it is either a canonical
/// corruption refusal OR a success whose recovered visible state
/// (`durable_acked <= recovered_visible <= appended`, hash chain intact) has lost
/// no acknowledged-durable commit. The store falls back to a full segment scan
/// over the intact (honestly-fsynced) segments, so the torn artifact costs a
/// slow open, not a lost commit.
///
/// Same shape as [`drive`]; the only differences are the crash trigger (torn
/// publish vs unsynced tail) and that the full appended prefix is acknowledged
/// durable up front.
#[cfg(test)]
fn drive_torn_publish(seed: u64, steps: usize) -> Result<RecoveryOutcome, Violation> {
    use super::fs::CrashOp;
    use crate::store::platform::fs::StoreFs;

    let dir = tempfile::tempdir().map_err(|e| Violation::NonCanonicalReopen(e.to_string()))?;
    // Honest disk: the ONLY fault is the torn artifact publish. The segments
    // holding acknowledged-durable commits are fsynced honestly, so the no-loss
    // contract must hold across the reopen full-scan.
    let sim_fs = Arc::new(SimFs::new(seed, 0));
    let coord = Coordinate::new("entity:dst", "scope:torn-publish")
        .map_err(|e| Violation::NonCanonicalReopen(e.to_string()))?;

    let plan = OpPlan::seeded(seed, steps);
    let mut digest = fold(FNV_OFFSET, seed);

    let appended;
    let durable_acked;
    {
        let config = StoreConfig::new(dir.path())
            .with_fs(Arc::clone(&sim_fs) as Arc<dyn StoreFs>)
            .with_sync_every_n_events(1_000_000)
            .with_sync_mode(SyncMode::SyncAll);
        let store =
            Store::open(config).map_err(|e| Violation::NonCanonicalReopen(e.to_string()))?;

        let (a, _plan_durable, run_digest) = run_op_plan(&store, &coord, &plan, digest)?;
        appended = a;
        digest = run_digest;

        // Acknowledge the FULL appended prefix durable via an honored sync, so
        // every appended commit MUST survive regardless of the torn artifact.
        Store::sync(&store).map_err(|e| Violation::NonCanonicalReopen(e.to_string()))?;
        durable_acked = appended;
        digest = fold(digest, 0x5_4_C);

        // Arm the atomic-publish fault, then trigger the cold-start artifact
        // publish via close(). close() drains the writer, flushes the durable
        // idempotency store, then writes the checkpoint/mmap artifact — both now
        // routed through StoreFs::persist_temp_with_parent_sync. The first such
        // publish tears, so close() returns Err with the artifact un-published.
        sim_fs.arm_fault_on(CrashOp::PersistTemp, 1);
        let close_result = store.close();
        debug_assert!(
            close_result.is_err(),
            "the armed PersistTemp fault must tear a cold-start artifact publish during close"
        );
        drop(close_result);

        // Truncate any write-but-unsynced tail (e.g. the post-sync close marker).
        // The durable (synced) segment prefix stays intact.
        sim_fs.crash();
    }

    // Reopen over the persisted tree. The torn/absent cold-start artifact forces a
    // full segment scan; D5 says that is legal recovery, never a lost commit.
    let reopen = Store::open(StoreConfig::new(dir.path()));
    let store = match reopen {
        Ok(store) => store,
        Err(error) if is_canonical_refusal(&error) => {
            digest = fold(digest, 0xCA11_AB1E);
            return Ok(RecoveryOutcome {
                digest,
                recovered_visible: 0,
                durable_acked,
            });
        }
        Err(other) => return Err(Violation::NonCanonicalReopen(format!("{other:?}"))),
    };

    let recovered = recovered_user_events(&store);
    let recovered_visible = recovered.len();
    if recovered_visible < durable_acked {
        return Err(Violation::LostDurableCommit {
            durable: durable_acked,
            recovered: recovered_visible,
        });
    }
    if recovered_visible > appended {
        return Err(Violation::UndeadEvent {
            recovered: recovered_visible,
            appended,
        });
    }
    verify_hash_chain(&recovered)?;

    digest = fold(fold(digest, recovered_visible as u64), durable_acked as u64);
    for ev in &recovered {
        digest = fold(digest, ev.event_hash_token);
    }
    Ok(RecoveryOutcome {
        digest,
        recovered_visible,
        durable_acked,
    })
}

/// Seed-tagged wrapper over [`drive_torn_publish`], mirroring [`run`].
#[cfg(test)]
fn run_torn_publish(seed: u64, steps: usize) -> Result<RecoveryOutcome, String> {
    drive_torn_publish(seed, steps)
        .map_err(|v| format!("torn-publish DST violation (seed={seed}): {v}"))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn same_seed_recovers_identically() {
        let a = run(0x5EED_0001, 48).expect("legal recovery");
        let b = run(0x5EED_0001, 48).expect("legal recovery");
        assert_eq!(
            a, b,
            "PROPERTY: identical seeds must recover the same state with the same digest"
        );
    }

    #[test]
    fn different_seeds_diverge() {
        let a = run(0x5EED_0002, 48).expect("legal recovery");
        let b = run(0x5EED_0003, 48).expect("legal recovery");
        // Over 48 mixed ops the recovered op-set + digest should differ almost
        // surely; if a rare collision occurs, the digests still must be internally
        // consistent (both legal), so we assert on the digest inequality which is
        // the determinism witness's discriminating signal.
        assert_ne!(
            a.digest, b.digest,
            "PROPERTY: distinct seeds should (almost surely) diverge in recovered digest"
        );
    }

    #[test]
    fn recovery_preserves_durable_prefix_and_legality() {
        let outcome = run(0x5EED_0004, 64).expect("legal recovery");
        assert!(
            outcome.recovered_visible >= outcome.durable_acked,
            "PROPERTY: every acknowledged-durable commit must survive the crash"
        );
    }

    #[test]
    fn torn_publish_same_seed_recovers_identically() {
        let a = run_torn_publish(0x7091_0001, 48).expect("legal torn-publish recovery");
        let b = run_torn_publish(0x7091_0001, 48).expect("legal torn-publish recovery");
        assert_eq!(
            a, b,
            "PROPERTY: identical seeds must recover the same state with the same digest \
             after a torn cold-start artifact publish"
        );
    }

    #[test]
    fn torn_publish_preserves_durable_prefix_via_full_scan() {
        // A torn cold-start artifact publish must never lose an acknowledged-durable
        // commit: reopen falls back to a full segment scan over the intact segments.
        let outcome = run_torn_publish(0x7091_0002, 64).expect("legal torn-publish recovery");
        assert!(
            outcome.recovered_visible >= outcome.durable_acked,
            "PROPERTY: a torn cold-start artifact is an optimization, not a correctness \
             dependency — every acknowledged-durable commit must survive it"
        );
    }
}