test(store): generic consistency checker, gauntlet fault/read

Lewis: May this revision serve well! <lu5a@proton.me>

crates/tranquil-store/src/consistency.rs

@@ -245,11 +245,16 @@ impl Default for ConsistencyCheckOptions {
     }
 }
 
-pub fn verify_store_consistency<S: StorageIO + 'static>(
-    blockstore: &TranquilBlockStore<RealIO, SystemClock>,
+pub fn verify_store_consistency<BS, BC, ES>(
+    blockstore: &TranquilBlockStore<BS, BC>,
     metastore: &Metastore,
-    eventlog: &EventLog<S>,
-) -> ConsistencyReport {
+    eventlog: &EventLog<ES>,
+) -> ConsistencyReport
+where
+    BS: StorageIO + Send + Sync + 'static,
+    BC: Clock,
+    ES: StorageIO + 'static,
+{
     verify_store_consistency_with_options(
         blockstore,
         metastore,
@@ -258,12 +263,17 @@ pub fn verify_store_consistency<S: StorageIO + 'static>(
     )
 }
 
-pub fn verify_store_consistency_with_options<S: StorageIO + 'static>(
-    blockstore: &TranquilBlockStore<RealIO, SystemClock>,
+pub fn verify_store_consistency_with_options<BS, BC, ES>(
+    blockstore: &TranquilBlockStore<BS, BC>,
     metastore: &Metastore,
-    eventlog: &EventLog<S>,
+    eventlog: &EventLog<ES>,
     options: ConsistencyCheckOptions,
-) -> ConsistencyReport {
+) -> ConsistencyReport
+where
+    BS: StorageIO + Send + Sync + 'static,
+    BC: Clock,
+    ES: StorageIO + 'static,
+{
     let mut report = ConsistencyReport::default();
 
     let block_index = blockstore.block_index();
@@ -567,8 +577,8 @@ fn check_cursor_vs_eventlog<S: StorageIO + 'static>(
     }
 }
 
-fn check_orphan_data_files(
-    blockstore: &TranquilBlockStore<RealIO, SystemClock>,
+fn check_orphan_data_files<BS: StorageIO + Send + Sync + 'static, BC: Clock>(
+    blockstore: &TranquilBlockStore<BS, BC>,
     block_index: &BlockIndex,
     report: &mut ConsistencyReport,
 ) {
@@ -600,8 +610,8 @@ fn check_orphan_data_files(
     });
 }
 
-fn check_missing_indexed_files(
-    blockstore: &TranquilBlockStore<RealIO, SystemClock>,
+fn check_missing_indexed_files<BS: StorageIO + Send + Sync + 'static, BC: Clock>(
+    blockstore: &TranquilBlockStore<BS, BC>,
     block_index: &BlockIndex,
     report: &mut ConsistencyReport,
 ) {
@@ -623,8 +633,8 @@ fn check_missing_indexed_files(
         .for_each(|(fid, _)| report.missing_indexed_files.push(*fid));
 }
 
-fn check_orphan_hint_files(
-    blockstore: &TranquilBlockStore<RealIO, SystemClock>,
+fn check_orphan_hint_files<BS: StorageIO + Send + Sync + 'static, BC: Clock>(
+    blockstore: &TranquilBlockStore<BS, BC>,
     report: &mut ConsistencyReport,
 ) {
     let data_files: HashSet<DataFileId> = match blockstore.list_data_files() {

crates/tranquil-store/src/gauntlet/chaos_walker.rs

@@ -77,6 +77,52 @@ pub fn mst_get_tolerant<S: StorageIO + Send + Sync + 'static, C: Clock>(
     }
 }
 
+pub fn walk_mst_entries_tolerant<S: StorageIO + Send + Sync + 'static, C: Clock>(
+    store: &TranquilBlockStore<S, C>,
+    root: Cid,
+    lost: &HashSet<CidBytes>,
+) -> Result<Option<Vec<(String, CidBytes)>>, String> {
+    let mut to_visit: Vec<Cid> = vec![root];
+    let mut visited: HashSet<CidBytes> = HashSet::new();
+    let mut entries: Vec<(String, CidBytes)> = Vec::new();
+
+    while let Some(cid) = to_visit.pop() {
+        let cid_bytes = try_cid_to_fixed(&cid).map_err(|e| format!("cid format: {e}"))?;
+        if !visited.insert(cid_bytes) {
+            continue;
+        }
+        if lost.contains(&cid_bytes) {
+            return Ok(None);
+        }
+        let node = match store.get_block_sync(&cid_bytes) {
+            Ok(Some(bytes)) => match serde_ipld_dagcbor::from_slice::<NodeData>(&bytes) {
+                Ok(n) => n,
+                Err(_) => return Ok(None),
+            },
+            Ok(None) => return Ok(None),
+            Err(_) => return Ok(None),
+        };
+        let keys = full_keys(&node)?;
+        keys.iter().zip(node.entries.iter()).try_for_each(
+            |(key, entry)| -> Result<(), String> {
+                let value =
+                    try_cid_to_fixed(&entry.value).map_err(|e| format!("cid format: {e}"))?;
+                entries.push((key.clone(), value));
+                Ok(())
+            },
+        )?;
+        if let Some(left) = node.left {
+            to_visit.push(left);
+        }
+        node.entries
+            .iter()
+            .filter_map(|e| e.tree)
+            .for_each(|t| to_visit.push(t));
+    }
+
+    Ok(Some(entries))
+}
+
 fn read_node<S: StorageIO + Send + Sync + 'static, C: Clock>(
     store: &TranquilBlockStore<S, C>,
     cid_bytes: &CidBytes,

crates/tranquil-store/src/gauntlet/op.rs

@@ -67,6 +67,7 @@ pub enum Op {
     ReadBlock {
         value_seed: ValueSeed,
     },
+    MstList,
     ExternalDeleteDataFile {
         choice: FileChoice,
     },
@@ -74,7 +75,10 @@ pub enum Op {
 
 impl Op {
     pub const fn is_read_only(&self) -> bool {
-        matches!(self, Op::ReadRecord { .. } | Op::ReadBlock { .. })
+        matches!(
+            self,
+            Op::ReadRecord { .. } | Op::ReadBlock { .. } | Op::MstList
+        )
     }
 }
 

crates/tranquil-store/src/gauntlet/oracle.rs

@@ -56,6 +56,10 @@ impl Oracle {
         self.live.contains_key(&(coll.clone(), rkey.clone()))
     }
 
+    pub fn expected_record_cid(&self, coll: &CollectionName, rkey: &RecordKey) -> Option<CidBytes> {
+        self.live.get(&(coll.clone(), rkey.clone())).copied()
+    }
+
     pub fn set_root(&mut self, root: Cid) {
         self.current_root = Some(root);
     }

crates/tranquil-store/src/gauntlet/runner.rs

@@ -153,6 +153,14 @@ pub(super) enum OpError {
     EventLogSync(String),
     #[error("eventlog retention: {0}")]
     EventLogRetention(String),
+    #[error("read validation: {0}")]
+    ReadValidation(String),
+}
+
+impl OpError {
+    pub(super) fn is_hard_violation(&self) -> bool {
+        matches!(self, OpError::ReadValidation(_))
+    }
 }
 
 pub struct EventLogState<S: StorageIO + Send + Sync + 'static> {
@@ -593,6 +601,14 @@ where
         let root_before = root;
         match apply_op(live, &mut root, &mut oracle, op, &config.workload, &clock).await {
             Ok(()) => {}
+            Err(e) if e.is_hard_violation() => {
+                violations.push(InvariantViolation {
+                    invariant: "ReadValidation",
+                    detail: format!("op {idx}: {e}"),
+                });
+                halt_ops = true;
+                continue;
+            }
             Err(e) => {
                 if tolerate_op_errors {
                     op_errors_counter.fetch_add(1, Ordering::Relaxed);
@@ -1196,6 +1212,85 @@ fn did_hash_for_seed(seed: DidSeed) -> DidHash {
     DidHash::from_did(&format!("did:plc:gauntlet{:08x}", seed.0))
 }
 
+fn validate_block_self_consistent<S: StorageIO + Send + Sync + 'static, C: Clock>(
+    store: &Arc<TranquilBlockStore<S, C>>,
+    requested: CidBytes,
+) -> Result<(), OpError> {
+    match store.get_block_sync(&requested) {
+        Ok(Some(bytes)) => {
+            let actual = hash_to_cid_bytes(&bytes);
+            if actual == requested {
+                Ok(())
+            } else {
+                Err(OpError::ReadValidation(format!(
+                    "block served under cid {} hashes to {}: store returned content not matching its address",
+                    hex_short(&requested),
+                    hex_short(&actual),
+                )))
+            }
+        }
+        Ok(None) => Ok(()),
+        Err(_) => Ok(()),
+    }
+}
+
+fn mst_list_violation(
+    expected: &std::collections::HashMap<String, CidBytes>,
+    actual: &std::collections::HashMap<String, CidBytes>,
+) -> Option<String> {
+    if expected == actual {
+        return None;
+    }
+    let missing: Vec<String> = expected
+        .iter()
+        .filter(|(k, v)| actual.get(*k) != Some(v))
+        .map(|(k, _)| k.clone())
+        .take(8)
+        .collect();
+    let extra: Vec<String> = actual
+        .iter()
+        .filter(|(k, v)| expected.get(*k) != Some(v))
+        .map(|(k, _)| k.clone())
+        .take(8)
+        .collect();
+    Some(format!(
+        "MST list walk mismatch: oracle has {} live records, walk found {}; missing-or-wrong {missing:?}; extra-or-wrong {extra:?}",
+        expected.len(),
+        actual.len(),
+    ))
+}
+
+async fn validate_read_record<S: StorageIO + Send + Sync + 'static, C: Clock>(
+    store: &Arc<TranquilBlockStore<S, C>>,
+    root: Cid,
+    key: &str,
+    expected: Option<CidBytes>,
+    has_lost_blocks: bool,
+) -> Result<(), OpError> {
+    let mst = Mst::load(store.clone(), root, None);
+    let Ok(Some(value_cid)) = mst.get(key).await else {
+        return Ok(());
+    };
+    let actual = try_cid_to_fixed(&value_cid)?;
+    match expected {
+        Some(x) if !has_lost_blocks && actual != x => {
+            return Err(OpError::ReadValidation(format!(
+                "{key}: MST returned cid {} but oracle holds {}",
+                hex_short(&actual),
+                hex_short(&x),
+            )));
+        }
+        None if !has_lost_blocks => {
+            return Err(OpError::ReadValidation(format!(
+                "{key}: MST holds a record with cid {} but the oracle has no live record at this key",
+                hex_short(&actual),
+            )));
+        }
+        _ => {}
+    }
+    validate_block_self_consistent(store, actual)
+}
+
 pub(super) async fn apply_op<S: StorageIO + Send + Sync + 'static, C: Clock>(
     harness: &mut Harness<S, C>,
     root: &mut Option<Cid>,
@@ -1311,15 +1406,41 @@ pub(super) async fn apply_op<S: StorageIO + Send + Sync + 'static, C: Clock>(
         Op::ReadRecord { collection, rkey } => {
             let Some(r) = *root else { return Ok(()) };
             let key = format!("{}/{}", collection.0, rkey.0);
-            let mst = Mst::load(harness.store.clone(), r, None);
-            let _ = mst.get(&key).await;
-            Ok(())
+            let expected = oracle.expected_record_cid(collection, rkey);
+            validate_read_record(&harness.store, r, &key, expected, oracle.has_lost_blocks()).await
         }
         Op::ReadBlock { value_seed } => {
             let record_bytes = make_record_bytes(*value_seed, workload.size_distribution);
             let record_cid = hash_to_cid_bytes(&record_bytes);
-            let _ = harness.store.get_block_sync(&record_cid);
-            Ok(())
+            validate_block_self_consistent(&harness.store, record_cid)
+        }
+        Op::MstList => {
+            let Some(r) = *root else {
+                return Ok(());
+            };
+            let store = harness.store.clone();
+            let lost = oracle.lost_blocks().clone();
+            let walked = tokio::task::spawn_blocking(move || {
+                super::chaos_walker::walk_mst_entries_tolerant(&store, r, &lost)
+            })
+            .await
+            .map_err(|e| OpError::Join(e.to_string()))?;
+            if oracle.has_lost_blocks() {
+                return Ok(());
+            }
+            let entries = match walked {
+                Ok(Some(entries)) => entries,
+                Ok(None) | Err(_) => return Ok(()),
+            };
+            let actual: std::collections::HashMap<String, CidBytes> = entries.into_iter().collect();
+            let expected: std::collections::HashMap<String, CidBytes> = oracle
+                .live_records()
+                .map(|(c, rk, v)| (format!("{}/{}", c.0, rk.0), *v))
+                .collect();
+            match mst_list_violation(&expected, &actual) {
+                None => Ok(()),
+                Some(detail) => Err(OpError::ReadValidation(detail)),
+            }
         }
         Op::ExternalDeleteDataFile { choice } => {
             let s = harness.store.clone();
@@ -1664,14 +1785,28 @@ async fn apply_op_concurrent<S: StorageIO + Send + Sync + 'static, C: Clock>(
                 return Ok(());
             };
             let key = format!("{}/{}", collection.0, rkey.0);
-            let mst = Mst::load(shared.store.clone(), r, None);
-            let _ = mst.get(&key).await;
-            Ok(())
+            validate_read_record(&shared.store, r, &key, None, true).await
         }
         Op::ReadBlock { value_seed } => {
             let record_bytes = make_record_bytes(*value_seed, workload.size_distribution);
             let record_cid = hash_to_cid_bytes(&record_bytes);
-            let _ = shared.store.get_block_sync(&record_cid);
+            validate_block_self_consistent(&shared.store, record_cid)
+        }
+        Op::MstList => {
+            let r = { shared.write.lock().await.root };
+            let Some(r) = r else {
+                return Ok(());
+            };
+            let store = shared.store.clone();
+            let _ = tokio::task::spawn_blocking(move || {
+                super::chaos_walker::walk_mst_entries_tolerant(
+                    &store,
+                    r,
+                    &std::collections::HashSet::new(),
+                )
+            })
+            .await
+            .map_err(|e| OpError::Join(e.to_string()))?;
             Ok(())
         }
         Op::ExternalDeleteDataFile { choice } => {
@@ -1712,6 +1847,12 @@ async fn writer_task<S: StorageIO + Send + Sync + 'static, C: Clock>(
             Ok(()) => {
                 ops_counter.fetch_max(idx + 1, Ordering::Relaxed);
             }
+            Err(e) if e.is_hard_violation() => {
+                return Some(InvariantViolation {
+                    invariant: "ReadValidation",
+                    detail: format!("op {idx}: {e}"),
+                });
+            }
             Err(e) => {
                 if tolerate_op_errors {
                     op_errors_counter.fetch_add(1, Ordering::Relaxed);

crates/tranquil-store/src/gauntlet/scenarios.rs

@@ -36,6 +36,8 @@ pub enum Scenario {
     RetentionTimeTravel,
     EventlogTimeTravelChaos,
     BlockChurnRecoverable,
+    ReadCorruption,
+    InlineCommit,
 }
 
 impl Scenario {
@@ -63,6 +65,8 @@ impl Scenario {
             Self::RetentionTimeTravel => "RetentionTimeTravel",
             Self::EventlogTimeTravelChaos => "EventlogTimeTravelChaos",
             Self::BlockChurnRecoverable => "BlockChurnRecoverable",
+            Self::ReadCorruption => "ReadCorruption",
+            Self::InlineCommit => "InlineCommit",
         }
     }
 
@@ -90,6 +94,8 @@ impl Scenario {
             Self::RetentionTimeTravel => "retention-time-travel",
             Self::EventlogTimeTravelChaos => "eventlog-time-travel-chaos",
             Self::BlockChurnRecoverable => "block-churn-recoverable",
+            Self::ReadCorruption => "read-corruption",
+            Self::InlineCommit => "inline-commit",
         }
     }
 
@@ -137,6 +143,12 @@ impl Scenario {
             Self::BlockChurnRecoverable => {
                 "Block-only churn under recoverable faults with crashes. Deterministic vehicle for refcount/reachability recovery bugs without the single-copy corruption-detection noise."
             }
+            Self::ReadCorruption => {
+                "Read-heavy workload under misdirected-read and bit-flip faults. Validates ReadRecord/ReadBlock results against the oracle at op time: the store must never serve content that does not match the requested address."
+            }
+            Self::InlineCommit => {
+                "Synchronous inline group-commit on the real backend with persisted-block verification and frequent restarts. Drives the GroupCommitConfig synchronous + verify_persisted_blocks path that production single-writer commits use."
+            }
         }
     }
 
@@ -171,6 +183,8 @@ impl Scenario {
         Self::RetentionTimeTravel,
         Self::EventlogTimeTravelChaos,
         Self::BlockChurnRecoverable,
+        Self::ReadCorruption,
+        Self::InlineCommit,
     ];
 }
 
@@ -249,6 +263,8 @@ pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig {
         Scenario::RetentionTimeTravel => retention_time_travel(seed),
         Scenario::EventlogTimeTravelChaos => eventlog_time_travel_chaos(seed),
         Scenario::BlockChurnRecoverable => block_churn_recoverable(seed),
+        Scenario::ReadCorruption => read_corruption(seed),
+        Scenario::InlineCommit => inline_commit(seed),
     }
 }
 
@@ -957,7 +973,22 @@ fn block_churn_recoverable(seed: Seed) -> GauntletConfig {
         io: IoBackend::Simulated {
             fault: FaultConfig::recoverable(),
         },
-        workload: sim_microbench_workload(),
+        workload: WorkloadModel {
+            weights: OpWeights {
+                add: 70,
+                delete: 10,
+                compact: 5,
+                checkpoint: 5,
+                mst_list: 10,
+                ..OpWeights::default()
+            },
+            size_distribution: SizeDistribution::Fixed(ValueBytes(128)),
+            collections: default_collections(),
+            key_space: KeySpaceSize(500),
+            did_space: DidSpaceSize(32),
+            retention_max_secs: RetentionMaxSecs(3600),
+            advance_max_secs: AdvanceMaxSecs(7200),
+        },
         op_count: OpCount(20_000),
         invariants: sim_invariants(),
         limits: RunLimits {
@@ -970,3 +1001,71 @@ fn block_churn_recoverable(seed: Seed) -> GauntletConfig {
         tolerate_op_errors: false,
     }
 }
+
+fn read_corruption(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Simulated {
+            fault: FaultConfig::read_faults(),
+        },
+        workload: WorkloadModel {
+            weights: OpWeights {
+                add: 35,
+                delete: 3,
+                compact: 2,
+                read_record: 40,
+                read_block: 12,
+                mst_list: 8,
+                ..OpWeights::default()
+            },
+            size_distribution: SizeDistribution::Fixed(ValueBytes(128)),
+            collections: default_collections(),
+            key_space: KeySpaceSize(300),
+            did_space: DidSpaceSize(32),
+            retention_max_secs: RetentionMaxSecs(3600),
+            advance_max_secs: AdvanceMaxSecs(7200),
+        },
+        op_count: OpCount(20_000),
+        invariants: sim_invariants(),
+        limits: RunLimits {
+            max_wall_ms: Some(WallMs(10 * 60_000)),
+        },
+        restart_policy: RestartPolicy::Never,
+        store: sim_store(),
+        eventlog: None,
+        writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
+    }
+}
+
+fn inline_commit(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Real,
+        workload: block_workload(
+            block_weights(80, 10, 5, 5),
+            SizeDistribution::Fixed(ValueBytes(96)),
+            KeySpaceSize(300),
+        ),
+        op_count: OpCount(10_000),
+        invariants: phase2_invariants(),
+        limits: RunLimits {
+            max_wall_ms: Some(WallMs(120_000)),
+        },
+        restart_policy: RestartPolicy::EveryNOps(OpInterval(1_000)),
+        store: StoreConfig {
+            max_file_size: MaxFileSize(8 * 1024),
+            group_commit: GroupCommitConfig {
+                synchronous: true,
+                verify_persisted_blocks: true,
+                checkpoint_interval_ms: 100,
+                checkpoint_write_threshold: 16,
+                ..GroupCommitConfig::default()
+            },
+            shard_count: ShardCount(1),
+        },
+        eventlog: None,
+        writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
+    }
+}

crates/tranquil-store/src/gauntlet/workload.rs

@@ -25,6 +25,7 @@ pub struct OpWeights {
     pub run_retention: u32,
     pub read_record: u32,
     pub read_block: u32,
+    pub mst_list: u32,
     pub external_delete_data_file: u32,
     pub advance_time: u32,
 }
@@ -40,6 +41,7 @@ impl OpWeights {
             + self.run_retention
             + self.read_record
             + self.read_block
+            + self.mst_list
             + self.external_delete_data_file
             + self.advance_time
     }
@@ -113,6 +115,7 @@ impl Default for WorkloadModel {
                 run_retention: 0,
                 read_record: 0,
                 read_block: 0,
+                mst_list: 0,
                 external_delete_data_file: 0,
                 advance_time: 0,
             },
@@ -152,7 +155,8 @@ impl WorkloadModel {
                 let t7 = t6 + w.run_retention;
                 let t8 = t7 + w.read_record;
                 let t9 = t8 + w.read_block;
-                let t10 = t9 + w.external_delete_data_file;
+                let t10 = t9 + w.mst_list;
+                let t11 = t10 + w.external_delete_data_file;
 
                 match bucket {
                     b if b < t1 => Op::AddRecord {
@@ -184,7 +188,8 @@ impl WorkloadModel {
                     b if b < t9 => Op::ReadBlock {
                         value_seed: ValueSeed(rng.next_u32()),
                     },
-                    b if b < t10 => Op::ExternalDeleteDataFile {
+                    b if b < t10 => Op::MstList,
+                    b if b < t11 => Op::ExternalDeleteDataFile {
                         choice: FileChoice(rng.next_u32()),
                     },
                     _ => Op::AdvanceTime {

crates/tranquil-store/src/sim.rs

@@ -3,7 +3,7 @@ use std::io;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::sync::Mutex;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
 use std::time::Duration;
 
 use crate::clock::{Clock, SimClock};
@@ -137,6 +137,23 @@ impl FaultConfig {
         }
     }
 
+    pub fn read_faults() -> Self {
+        Self {
+            misdirected_read_probability: Probability::new(0.05),
+            bit_flip_on_read_probability: Probability::new(0.05),
+            io_error_probability: Probability::new(0.01),
+            ..Self::none()
+        }
+    }
+
+    pub fn read_corruption() -> Self {
+        Self {
+            misdirected_read_probability: Probability::new(0.05),
+            bit_flip_on_read_probability: Probability::new(0.05),
+            ..Self::none()
+        }
+    }
+
     pub fn injects_errors(&self) -> bool {
         self.partial_write_probability.is_nonzero()
             || self.bit_flip_on_read_probability.is_nonzero()
@@ -381,6 +398,9 @@ pub struct SimulatedIO {
     state: Mutex<SimState>,
     fault_config: FaultConfig,
     pristine_mode: AtomicBool,
+    write_crash_armed: AtomicBool,
+    write_crash_countdown: AtomicI64,
+    write_crashed: AtomicBool,
     rng_seed: u64,
     clock: SimClock,
 }
@@ -401,11 +421,35 @@ impl SimulatedIO {
             }),
             fault_config,
             pristine_mode: AtomicBool::new(false),
+            write_crash_armed: AtomicBool::new(false),
+            write_crash_countdown: AtomicI64::new(-1),
+            write_crashed: AtomicBool::new(false),
             rng_seed: seed,
             clock: SimClock::new(seed),
         }
     }
 
+    pub fn arm_write_crash(&self, after_writes: i64) {
+        self.write_crashed.store(false, Ordering::Relaxed);
+        self.write_crash_countdown
+            .store(after_writes, Ordering::Relaxed);
+        self.write_crash_armed.store(true, Ordering::Relaxed);
+    }
+
+    fn write_crash_fired(&self) -> bool {
+        if !self.write_crash_armed.load(Ordering::Relaxed) {
+            return false;
+        }
+        if self.write_crashed.load(Ordering::Relaxed) {
+            return true;
+        }
+        if self.write_crash_countdown.fetch_sub(1, Ordering::Relaxed) <= 0 {
+            self.write_crashed.store(true, Ordering::Relaxed);
+            return true;
+        }
+        false
+    }
+
     pub fn clock(&self) -> SimClock {
         self.clock.clone()
     }
@@ -443,6 +487,10 @@ impl SimulatedIO {
     pub fn crash(&self) -> Vec<PathBuf> {
         let mut state = self.state.lock().unwrap();
 
+        self.write_crash_armed.store(false, Ordering::Relaxed);
+        self.write_crashed.store(false, Ordering::Relaxed);
+        self.write_crash_countdown.store(-1, Ordering::Relaxed);
+
         state.fds.clear();
         state.pending_syncs.clear();
 
@@ -716,6 +764,10 @@ impl StorageIO for SimulatedIO {
         let stream = sid.0;
         self.jitter(stream, offset);
 
+        if self.write_crash_fired() {
+            return Err(io::Error::other("simulated crash mid-commit"));
+        }
+
         if state.storage.get(&sid).is_some_and(|s| s.io_poisoned) {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
@@ -826,6 +878,10 @@ impl StorageIO for SimulatedIO {
             .unwrap_or(0);
         self.jitter(stream, fsize);
 
+        if self.write_crashed.load(Ordering::Relaxed) {
+            return Err(io::Error::other("simulated crash mid-commit"));
+        }
+
         if state.storage.get(&sid).is_some_and(|s| s.io_poisoned) {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }