test(store): D gauntlet faults, crash-loss oracley, recoverable scenarios

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

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

@@ -146,7 +146,7 @@ impl<S: StorageIO + Send + Sync + 'static, C: Clock> Invariant<S, C> for Refcoun
 
         let inverse: Vec<String> = index
             .iter()
-            .filter(|(cid, _)| !live_set.contains(*cid))
+            .filter(|(cid, _)| !live_set.contains(*cid) && !ctx.oracle.lost_blocks().contains(*cid))
             .map(|(cid, r)| format!("orphan cid {} refcount {}", hex_short(cid), r))
             .collect();
 
@@ -501,10 +501,12 @@ impl<S: StorageIO + Send + Sync + 'static, C: Clock> Invariant<S, C> for IndexBl
 
     async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> {
         let store_c = ctx.store.clone();
+        let lost = ctx.oracle.lost_blocks().clone();
         let result = tokio::task::spawn_blocking(move || {
             let entries = store_c.block_index().live_entries_snapshot();
             let unreadable: Vec<String> = entries
                 .iter()
+                .filter(|(cid, _)| !lost.contains(cid))
                 .take(INDEX_READABLE_SAMPLE_CAP)
                 .filter_map(|(cid, _)| match store_c.get_block_sync(cid) {
                     Ok(Some(_)) => None,
@@ -900,21 +902,37 @@ impl<S: StorageIO + Send + Sync + 'static, C: Clock> Invariant<S, C> for FsyncOr
         if !missing.is_empty() {
             let mut sorted = missing;
             sorted.sort_unstable();
+            let ts_by_seq: HashMap<u64, u64> = ctx
+                .oracle
+                .synced_events()
+                .iter()
+                .map(|e| (e.seq.raw(), e.timestamp_us))
+                .collect();
+            let cutoff = ctx.oracle.last_retention_cutoff_us();
+            let detail: Vec<String> = sorted
+                .iter()
+                .take(5)
+                .map(|seq| {
+                    let ts = ts_by_seq.get(seq).copied().unwrap_or(0);
+                    let below = cutoff.is_some_and(|c| ts < c);
+                    format!("seq {seq} ts {ts} below_cutoff {below}")
+                })
+                .collect();
             violations.push(format!(
-                "{} acked events lost on disk, lowest missing seq {}",
+                "{} acked events lost on disk, cutoff {cutoff:?}: {}",
                 sorted.len(),
-                sorted[0]
+                detail.join(", ")
             ));
         }
 
-        if let Some(last_synced) = ctx.oracle.last_synced_seq()
+        let retained_synced_max = ctx.oracle.synced_events().iter().map(|e| e.seq.raw()).max();
+        if let Some(expected) = retained_synced_max
             && el.synced_seq.raw() != 0
-            && el.synced_seq.raw() < last_synced.raw()
+            && el.synced_seq.raw() < expected
         {
             violations.push(format!(
-                "writer synced_seq {} below oracle last_synced_seq {}",
-                el.synced_seq.raw(),
-                last_synced.raw()
+                "writer synced_seq {} below retained acked seq {expected}",
+                el.synced_seq.raw()
             ));
         }
 
@@ -946,11 +964,14 @@ impl<S: StorageIO + Send + Sync + 'static, C: Clock> Invariant<S, C> for Tombsto
         };
 
         let active = el.segments.last().copied();
+        let retention_active = ctx.oracle.last_retention_active_segment();
 
         let stale: Vec<String> = el
             .segment_last_ts
             .iter()
-            .filter(|(id, last_ts)| Some(*id) != active && *last_ts < cutoff_us)
+            .filter(|(id, last_ts)| {
+                Some(*id) != active && Some(*id) != retention_active && *last_ts < cutoff_us
+            })
             .map(|(id, last_ts)| format!("segment {id} last_ts {last_ts} < cutoff {cutoff_us}"))
             .collect();
 

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

@@ -4,7 +4,7 @@ use cid::Cid;
 
 use super::op::{CollectionName, EventKind, RecordKey};
 use crate::blockstore::CidBytes;
-use crate::eventlog::EventSequence;
+use crate::eventlog::{EventSequence, SegmentId};
 
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 #[error("unexpected CID encoding: got {actual} bytes, expected 36 for sha256 CIDv1")]
@@ -18,6 +18,7 @@ pub struct EventExpectation {
     pub timestamp_us: u64,
     pub kind: EventKind,
     pub did_hash: u32,
+    pub segment: SegmentId,
 }
 
 #[derive(Debug, Default)]
@@ -29,6 +30,7 @@ pub struct Oracle {
     unsynced_events: Vec<EventExpectation>,
     last_synced_seq: Option<EventSequence>,
     last_retention_cutoff_us: Option<u64>,
+    last_retention_active_segment: Option<SegmentId>,
     lost_blocks: HashSet<CidBytes>,
 }
 
@@ -129,9 +131,18 @@ impl Oracle {
         self.unsynced_events.clear();
     }
 
-    pub fn record_retention(&mut self, cutoff_us: u64) {
+    pub fn forget_events_in_segments(&mut self, lost: &HashSet<SegmentId>) {
+        if lost.is_empty() {
+            return;
+        }
+        self.synced_events.retain(|e| !lost.contains(&e.segment));
+        self.unsynced_events.retain(|e| !lost.contains(&e.segment));
+    }
+
+    pub fn record_retention(&mut self, cutoff_us: u64, active_segment: Option<SegmentId>) {
         self.synced_events.retain(|e| e.timestamp_us >= cutoff_us);
         self.last_retention_cutoff_us = Some(cutoff_us);
+        self.last_retention_active_segment = active_segment;
     }
 
     pub fn synced_events(&self) -> &[EventExpectation] {
@@ -149,6 +160,10 @@ impl Oracle {
     pub fn last_retention_cutoff_us(&self) -> Option<u64> {
         self.last_retention_cutoff_us
     }
+
+    pub fn last_retention_active_segment(&self) -> Option<SegmentId> {
+        self.last_retention_active_segment
+    }
 }
 
 pub(super) fn try_cid_to_fixed(cid: &Cid) -> Result<CidBytes, CidFormatError> {

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

@@ -15,13 +15,13 @@ use super::op::{DidSeed, EventKind, Op, OpStream, PayloadSeed, RetentionSecs, Se
 use super::oracle::{CidFormatError, EventExpectation, Oracle, hex_short, try_cid_to_fixed};
 use super::workload::{Lcg, OpCount, SizeDistribution, ValueBytes, WorkloadModel};
 use crate::blockstore::{
-    BlockStoreConfig, CidBytes, CompactionError, GroupCommitConfig, TranquilBlockStore,
+    BlockStoreConfig, CidBytes, CompactionError, DataFileId, GroupCommitConfig, TranquilBlockStore,
     hash_to_cid, hash_to_cid_bytes,
 };
 use crate::clock::{Clock, SimClock, SystemClock};
 use crate::eventlog::{
     DEFAULT_INDEX_INTERVAL, DidHash, EventLogWriter, EventTypeTag, MAX_EVENT_PAYLOAD, SegmentId,
-    SegmentManager, SegmentReader, ValidEvent,
+    SegmentManager, SegmentReader, ValidEvent, parse_segment_id,
 };
 use crate::io::{RealIO, StorageIO};
 use crate::sim::{FaultConfig, PristineGuard, SimulatedIO};
@@ -404,7 +404,7 @@ async fn run_inner_real_on_root(
             op_errors_counter,
             restarts_counter,
             open,
-            || {},
+            Vec::new,
             tolerate_op_errors,
             reopen_backoff,
             SystemClock,
@@ -419,7 +419,7 @@ async fn run_inner_real_on_root(
             op_errors_counter,
             restarts_counter,
             open,
-            || {},
+            Vec::new,
             tolerate_op_errors,
             reopen_backoff,
             SystemClock,
@@ -438,7 +438,8 @@ async fn run_inner_simulated(
     restarts_counter: Arc<AtomicUsize>,
 ) -> GauntletReport {
     let dir = tempfile::TempDir::new().expect("tempdir");
-    let cfg = blockstore_config(dir.path(), &config.store);
+    let mut cfg = blockstore_config(dir.path(), &config.store);
+    cfg.group_commit.synchronous = config.writer_concurrency.0 == 1;
     let tolerate_errors = fault.injects_errors() || config.tolerate_op_errors;
     let eventlog_cfg = config.eventlog;
     let segments_dir = segments_subdir(dir.path());
@@ -551,7 +552,7 @@ where
     S: StorageIO + Send + Sync + 'static,
     C: Clock,
     Open: FnMut(usize) -> Result<Harness<S, C>, String>,
-    Crash: FnMut(),
+    Crash: FnMut() -> Vec<PathBuf>,
     Quiesce: Fn(bool),
 {
     let mut oracle = Oracle::new();
@@ -614,8 +615,8 @@ where
             continue;
         }
         if crashing {
-            crash();
-            oracle.record_crash();
+            let removed = crash();
+            record_crash_losses(&removed, harness.as_ref(), &mut oracle);
         }
         shutdown_harness(&mut harness);
         match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await
@@ -655,8 +656,8 @@ where
 
     quiesce_faults(true);
     if !halt_ops && tolerate_op_errors && harness.is_some() {
-        crash();
-        oracle.record_crash();
+        let removed = crash();
+        record_crash_losses(&removed, harness.as_ref(), &mut oracle);
         shutdown_harness(&mut harness);
         match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await
         {
@@ -794,6 +795,41 @@ fn shutdown_harness<S: StorageIO + Send + Sync + 'static, C: Clock>(
     let _ = harness.take();
 }
 
+fn record_crash_losses<S: StorageIO + Send + Sync + 'static, C: Clock>(
+    removed: &[PathBuf],
+    harness: Option<&Harness<S, C>>,
+    oracle: &mut Oracle,
+) {
+    mark_block_crash_losses(removed, harness, oracle);
+    mark_event_crash_losses(removed, oracle);
+    oracle.record_crash();
+}
+
+fn mark_block_crash_losses<S: StorageIO + Send + Sync + 'static, C: Clock>(
+    removed: &[PathBuf],
+    harness: Option<&Harness<S, C>>,
+    oracle: &mut Oracle,
+) {
+    let Some(h) = harness else { return };
+    let data_dir = h.store.data_dir();
+    let lost: Vec<CidBytes> = removed
+        .iter()
+        .filter(|p| p.parent() == Some(data_dir))
+        .filter_map(|p| p.file_stem()?.to_str()?.parse::<u32>().ok())
+        .map(DataFileId::new)
+        .flat_map(|fid| h.store.block_index().cids_in_file(fid))
+        .collect();
+    if !lost.is_empty() {
+        oracle.mark_blocks_lost(lost);
+    }
+}
+
+fn mark_event_crash_losses(removed: &[PathBuf], oracle: &mut Oracle) {
+    let lost: std::collections::HashSet<SegmentId> =
+        removed.iter().filter_map(|p| parse_segment_id(p)).collect();
+    oracle.forget_events_in_segments(&lost);
+}
+
 const MAX_REOPEN_ATTEMPTS: usize = 5;
 
 async fn reopen_with_recovery<S, C, Open, Crash>(
@@ -806,7 +842,7 @@ where
     S: StorageIO + Send + Sync + 'static,
     C: Clock,
     Open: FnMut(usize) -> Result<Harness<S, C>, String>,
-    Crash: FnMut(),
+    Crash: FnMut() -> Vec<PathBuf>,
 {
     let mut errors: Vec<String> = Vec::new();
     for attempt in 0..MAX_REOPEN_ATTEMPTS {
@@ -1234,11 +1270,13 @@ pub(super) async fn apply_op<S: StorageIO + Send + Sync + 'static, C: Clock>(
             let ts_before = clock.unix_micros().raw();
             match el.writer.append_with_clock(did_hash, tag, payload, clock) {
                 Ok(seq) => {
+                    let segment = el.writer.active_segment_id();
                     oracle.record_event_append(EventExpectation {
                         seq,
                         timestamp_us: ts_before,
                         kind: *event_kind,
                         did_hash: did_hash.raw(),
+                        segment,
                     });
                     let _ = el.writer.rotate_if_needed();
                     Ok(())
@@ -1307,7 +1345,7 @@ fn externally_delete_data_file<S: StorageIO + Send + Sync + 'static, C: Clock>(
         Ok(files) => files,
         Err(_) => return Ok(Vec::new()),
     };
-    candidates.retain(|fid| active.is_none_or(|a| *fid < a));
+    candidates.retain(|fid| active.is_some_and(|a| *fid < a));
     if candidates.is_empty() {
         return Ok(Vec::new());
     }
@@ -1349,7 +1387,7 @@ fn run_retention<S: StorageIO + Send + Sync + 'static, C: Clock>(
                 _ => Ok(()),
             }
         })?;
-    oracle.record_retention(cutoff_us);
+    oracle.record_retention(cutoff_us, active_id);
     Ok(())
 }
 
@@ -1480,11 +1518,12 @@ fn compact_by_liveness<S: StorageIO + Send + Sync + 'static, C: Clock>(
     let liveness = store
         .compaction_liveness(0)
         .map_err(|e| OpError::CompactFile(format!("compaction_liveness: {e}")))?;
-    let targets: Vec<_> = liveness
+    let mut targets: Vec<_> = liveness
         .iter()
         .filter(|(_, info)| info.total_blocks > 0 && info.ratio() < COMPACT_LIVENESS_CEILING)
         .map(|(&fid, _)| fid)
         .collect();
+    targets.sort_unstable();
     targets
         .into_iter()
         .try_for_each(|fid| match store.compact_file(fid, 0) {
@@ -1575,12 +1614,14 @@ async fn apply_op_concurrent<S: StorageIO + Send + Sync + 'static, C: Clock>(
             };
             match el.writer.append_with_clock(did_hash, tag, payload, clock) {
                 Ok(seq) => {
+                    let segment = el.writer.active_segment_id();
                     let _ = el.writer.rotate_if_needed();
                     state.oracle.record_event_append(EventExpectation {
                         seq,
                         timestamp_us: ts_before,
                         kind: *event_kind,
                         did_hash: did_hash.raw(),
+                        segment,
                     });
                     Ok(())
                 }
@@ -1726,7 +1767,7 @@ where
     S: StorageIO + Send + Sync + 'static,
     C: Clock,
     Open: FnMut(usize) -> Result<Harness<S, C>, String>,
-    Crash: FnMut(),
+    Crash: FnMut() -> Vec<PathBuf>,
     Quiesce: Fn(bool),
 {
     let ops: Vec<Op> = op_stream.into_vec();
@@ -1840,8 +1881,8 @@ where
             RestartAction::None => {}
             RestartAction::Clean | RestartAction::Crash => {
                 if matches!(action, RestartAction::Crash) {
-                    crash();
-                    oracle.record_crash();
+                    let removed = crash();
+                    record_crash_losses(&removed, harness.as_ref(), &mut oracle);
                 }
                 shutdown_harness(&mut harness);
                 match reopen_with_recovery(
@@ -1888,8 +1929,8 @@ where
 
     quiesce_faults(true);
     if !halt_ops && tolerate_op_errors && harness.is_some() {
-        crash();
-        oracle.record_crash();
+        let removed = crash();
+        record_crash_losses(&removed, harness.as_ref(), &mut oracle);
         shutdown_harness(&mut harness);
         match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await
         {
@@ -2052,7 +2093,7 @@ mod tests {
                 store_cfg,
                 clock.clone(),
             ),
-            || {},
+            Vec::new,
             true,
             Duration::ZERO,
             clock,
@@ -2098,7 +2139,7 @@ mod tests {
                 store_cfg,
                 clock.clone(),
             ),
-            || {},
+            Vec::new,
             true,
             Duration::ZERO,
             clock,

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

@@ -34,6 +34,8 @@ pub enum Scenario {
     FlakyDevice,
     ExternalCorruption,
     RetentionTimeTravel,
+    EventlogTimeTravelChaos,
+    BlockChurnRecoverable,
 }
 
 impl Scenario {
@@ -59,6 +61,8 @@ impl Scenario {
             Self::FlakyDevice => "FlakyDevice",
             Self::ExternalCorruption => "ExternalCorruption",
             Self::RetentionTimeTravel => "RetentionTimeTravel",
+            Self::EventlogTimeTravelChaos => "EventlogTimeTravelChaos",
+            Self::BlockChurnRecoverable => "BlockChurnRecoverable",
         }
     }
 
@@ -84,6 +88,8 @@ impl Scenario {
             Self::FlakyDevice => "flaky-device",
             Self::ExternalCorruption => "external-corruption",
             Self::RetentionTimeTravel => "retention-time-travel",
+            Self::EventlogTimeTravelChaos => "eventlog-time-travel-chaos",
+            Self::BlockChurnRecoverable => "block-churn-recoverable",
         }
     }
 
@@ -125,6 +131,12 @@ impl Scenario {
             Self::RetentionTimeTravel => {
                 "Eventlog retention under logical time travel: random multi-day AdvanceTime jumps interleaved with append/sync/retention. TOMBSTONE_BOUND across fake weeks."
             }
+            Self::EventlogTimeTravelChaos => {
+                "Eventlog crash-recovery under logical time travel with recoverable faults. Expect 100% clean: any violation is a recovery bug, not the single-copy detection limit."
+            }
+            Self::BlockChurnRecoverable => {
+                "Block-only churn under recoverable faults with crashes. Deterministic vehicle for refcount/reachability recovery bugs without the single-copy corruption-detection noise."
+            }
         }
     }
 
@@ -157,6 +169,8 @@ impl Scenario {
         Self::FlakyDevice,
         Self::ExternalCorruption,
         Self::RetentionTimeTravel,
+        Self::EventlogTimeTravelChaos,
+        Self::BlockChurnRecoverable,
     ];
 }
 
@@ -233,6 +247,8 @@ pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig {
         Scenario::FlakyDevice => flaky_device(seed),
         Scenario::ExternalCorruption => external_corruption(seed),
         Scenario::RetentionTimeTravel => retention_time_travel(seed),
+        Scenario::EventlogTimeTravelChaos => eventlog_time_travel_chaos(seed),
+        Scenario::BlockChurnRecoverable => block_churn_recoverable(seed),
     }
 }
 
@@ -892,3 +908,65 @@ fn retention_time_travel(seed: Seed) -> GauntletConfig {
         tolerate_op_errors: false,
     }
 }
+
+fn eventlog_time_travel_chaos(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Simulated {
+            fault: FaultConfig::recoverable(),
+        },
+        workload: WorkloadModel {
+            weights: OpWeights {
+                add: 15,
+                compact: 2,
+                checkpoint: 3,
+                append_event: 35,
+                sync_event_log: 15,
+                run_retention: 10,
+                advance_time: 20,
+                ..OpWeights::default()
+            },
+            size_distribution: SizeDistribution::Fixed(ValueBytes(96)),
+            collections: default_collections(),
+            key_space: KeySpaceSize(400),
+            did_space: DidSpaceSize(64),
+            retention_max_secs: RetentionMaxSecs(604_800),
+            advance_max_secs: AdvanceMaxSecs(259_200),
+        },
+        op_count: OpCount(20_000),
+        invariants: sim_invariants()
+            | InvariantSet::MONOTONIC_SEQ
+            | InvariantSet::FSYNC_ORDERING
+            | InvariantSet::TOMBSTONE_BOUND,
+        limits: RunLimits {
+            max_wall_ms: Some(WallMs(10 * 60_000)),
+        },
+        restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(2_000)),
+        store: sim_store(),
+        eventlog: Some(EventLogConfig {
+            max_segment_size: MaxSegmentSize(16 * 1024),
+        }),
+        writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
+    }
+}
+
+fn block_churn_recoverable(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Simulated {
+            fault: FaultConfig::recoverable(),
+        },
+        workload: sim_microbench_workload(),
+        op_count: OpCount(20_000),
+        invariants: sim_invariants(),
+        limits: RunLimits {
+            max_wall_ms: Some(WallMs(10 * 60_000)),
+        },
+        restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(2_000)),
+        store: sim_store(),
+        eventlog: None,
+        writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
+    }
+}

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, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
 use crate::clock::{Clock, SimClock};
@@ -128,6 +128,15 @@ impl FaultConfig {
         }
     }
 
+    pub fn recoverable() -> Self {
+        Self {
+            misdirected_write_probability: Probability::ZERO,
+            misdirected_read_probability: Probability::ZERO,
+            bit_flip_on_read_probability: Probability::ZERO,
+            ..Self::moderate()
+        }
+    }
+
     pub fn injects_errors(&self) -> bool {
         self.partial_write_probability.is_nonzero()
             || self.bit_flip_on_read_probability.is_nonzero()
@@ -257,33 +266,66 @@ struct SimState {
     fds: HashMap<FileId, SimFd>,
     dirs_durable: HashSet<PathBuf>,
     op_log: Vec<OpRecord>,
-    rng_counter: u64,
     next_fd_id: u64,
     next_storage_id: u64,
     pending_syncs: VecDeque<PendingSync>,
     pending_deletes: Vec<PendingDelete>,
 }
 
+const TAG_OPEN_IO: u64 = 100;
+const TAG_READ_IO: u64 = 101;
+const TAG_READ_MISDIR: u64 = 102;
+const TAG_READ_DRIFT_SECTORS: u64 = 103;
+const TAG_READ_DRIFT_DIR: u64 = 104;
+const TAG_READ_BITFLIP: u64 = 105;
+const TAG_READ_FLIP_POS: u64 = 106;
+const TAG_READ_FLIP_BIT: u64 = 107;
+const TAG_WRITE_IO: u64 = 110;
+const TAG_WRITE_TORN: u64 = 111;
+const TAG_WRITE_TORN_SECTORS: u64 = 112;
+const TAG_WRITE_PARTIAL: u64 = 113;
+const TAG_WRITE_PARTIAL_LEN: u64 = 114;
+const TAG_WRITE_MISDIR: u64 = 115;
+const TAG_WRITE_DRIFT_SECTORS: u64 = 116;
+const TAG_WRITE_DRIFT_DIR: u64 = 117;
+const TAG_SYNC_IO: u64 = 120;
+const TAG_SYNC_FAILURE: u64 = 121;
+const TAG_SYNC_DELAYED: u64 = 122;
+const TAG_SYNCDIR_IO: u64 = 130;
+const TAG_SYNCDIR_FAILURE: u64 = 131;
+const TAG_LATENCY: u64 = 140;
+
+fn path_stream(path: &Path) -> u64 {
+    let key = path.file_name().unwrap_or(path.as_os_str());
+    key.to_string_lossy()
+        .bytes()
+        .fold(0xcbf2_9ce4_8422_2325, |h, b| {
+            (h ^ u64::from(b)).wrapping_mul(0x0000_0100_0000_01b3)
+        })
+}
+
 impl SimState {
-    fn next_random(&mut self, seed: u64) -> f64 {
-        let counter = self.rng_counter;
-        self.rng_counter += 1;
-        let mixed = splitmix64(seed.wrapping_add(counter));
-        (mixed >> 11) as f64 / (1u64 << 53) as f64
+    fn fault_hash(seed: u64, stream: u64, key: u64, tag: u64) -> u64 {
+        let h = splitmix64(seed ^ 0x9E37_79B9_7F4A_7C15);
+        let h = splitmix64(h ^ stream.wrapping_mul(0xD6E8_FEB8_6659_FD93));
+        let h = splitmix64(h ^ key);
+        splitmix64(h ^ tag)
     }
 
-    fn next_random_usize(&mut self, seed: u64, max: usize) -> usize {
+    fn fault_unit(seed: u64, stream: u64, key: u64, tag: u64) -> f64 {
+        (Self::fault_hash(seed, stream, key, tag) >> 11) as f64 / (1u64 << 53) as f64
+    }
+
+    fn fault_below(seed: u64, stream: u64, key: u64, tag: u64, probability: Probability) -> bool {
+        probability.is_nonzero() && Self::fault_unit(seed, stream, key, tag) < probability.raw()
+    }
+
+    fn fault_usize(seed: u64, stream: u64, key: u64, tag: u64, max: usize) -> usize {
         if max == 0 {
-            return 0;
+            0
+        } else {
+            (Self::fault_hash(seed, stream, key, tag) as usize) % max
         }
-        let counter = self.rng_counter;
-        self.rng_counter += 1;
-        let mixed = splitmix64(seed.wrapping_add(counter));
-        (mixed as usize) % max
-    }
-
-    fn should_fault(&mut self, seed: u64, probability: Probability) -> bool {
-        probability.is_nonzero() && self.next_random(seed) < probability.raw()
     }
 
     fn alloc_fd_id(&mut self) -> FileId {
@@ -340,7 +382,6 @@ pub struct SimulatedIO {
     fault_config: FaultConfig,
     pristine_mode: AtomicBool,
     rng_seed: u64,
-    latency_counter: AtomicU64,
     clock: SimClock,
 }
 
@@ -353,7 +394,6 @@ impl SimulatedIO {
                 fds: HashMap::new(),
                 dirs_durable: HashSet::new(),
                 op_log: Vec::new(),
-                rng_counter: 0,
                 next_fd_id: 1,
                 next_storage_id: 1,
                 pending_syncs: VecDeque::new(),
@@ -362,7 +402,6 @@ impl SimulatedIO {
             fault_config,
             pristine_mode: AtomicBool::new(false),
             rng_seed: seed,
-            latency_counter: AtomicU64::new(0),
             clock: SimClock::new(seed),
         }
     }
@@ -387,14 +426,11 @@ impl SimulatedIO {
         self.pristine_mode.load(Ordering::Relaxed)
     }
 
-    fn jitter(&self) {
+    fn jitter(&self, stream: u64, key: u64) {
         let max_ns = self.effective_fault_config().latency_distribution_ns.0;
         let extra_ns = match max_ns {
             0 => 0,
-            max => {
-                let c = self.latency_counter.fetch_add(1, Ordering::Relaxed);
-                splitmix64(self.rng_seed.wrapping_add(c)) % max
-            }
+            max => SimState::fault_hash(self.rng_seed, stream, key, TAG_LATENCY) % max,
         };
         self.clock
             .advance(Duration::from_nanos(BASE_IO_SERVICE_NS + extra_ns));
@@ -404,7 +440,7 @@ impl SimulatedIO {
         Self::new(seed, FaultConfig::none())
     }
 
-    pub fn crash(&self) {
+    pub fn crash(&self) -> Vec<PathBuf> {
         let mut state = self.state.lock().unwrap();
 
         state.fds.clear();
@@ -417,13 +453,20 @@ impl SimulatedIO {
             }
         });
 
-        let orphaned: Vec<StorageId> = state
+        let orphaned: HashSet<StorageId> = state
             .storage
             .iter()
             .filter(|(_, s)| !s.dir_entry_durable)
             .map(|(sid, _)| *sid)
             .collect();
 
+        let removed_paths: Vec<PathBuf> = state
+            .paths
+            .iter()
+            .filter(|(_, sid)| orphaned.contains(sid))
+            .map(|(path, _)| path.clone())
+            .collect();
+
         orphaned.iter().for_each(|sid| {
             state.storage.remove(sid);
         });
@@ -435,6 +478,8 @@ impl SimulatedIO {
             s.buffered = s.durable.clone();
             s.io_poisoned = false;
         });
+
+        removed_paths
     }
 
     pub fn op_log(&self) -> Vec<OpRecord> {
@@ -492,7 +537,20 @@ impl StorageIO for SimulatedIO {
         let mut state = self.state.lock().unwrap();
         let seed = self.rng_seed;
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        let open_stream = path_stream(path);
+        let existing_size = state
+            .paths
+            .get(path)
+            .and_then(|sid| state.storage.get(sid))
+            .map(|s| s.buffered.len() as u64)
+            .unwrap_or(0);
+        if SimState::fault_below(
+            seed,
+            open_stream,
+            existing_size,
+            TAG_OPEN_IO,
+            fault.io_error_probability,
+        ) {
             return Err(io::Error::other("simulated EIO on open"));
         }
 
@@ -571,24 +629,38 @@ impl StorageIO for SimulatedIO {
     }
 
     fn read_at(&self, id: FileId, offset: u64, buf: &mut [u8]) -> io::Result<usize> {
-        self.jitter();
         let fault = self.effective_fault_config();
         let mut state = self.state.lock().unwrap();
         let sid = state.require_readable(id)?;
         let seed = self.rng_seed;
+        let stream = sid.0;
+        self.jitter(stream, offset);
 
         if state.storage.get(&sid).is_some_and(|s| s.io_poisoned) {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        if SimState::fault_below(
+            seed,
+            stream,
+            offset,
+            TAG_READ_IO,
+            fault.io_error_probability,
+        ) {
             return Err(io::Error::other("simulated EIO on read"));
         }
 
-        let read_offset = if state.should_fault(seed, fault.misdirected_read_probability) {
-            let drift_sectors = state.next_random_usize(seed, 8) + 1;
+        let read_offset = if SimState::fault_below(
+            seed,
+            stream,
+            offset,
+            TAG_READ_MISDIR,
+            fault.misdirected_read_probability,
+        ) {
+            let drift_sectors =
+                SimState::fault_usize(seed, stream, offset, TAG_READ_DRIFT_SECTORS, 8) + 1;
             let drift = (drift_sectors * SECTOR_BYTES) as u64;
-            if state.next_random(seed) < 0.5 {
+            if SimState::fault_unit(seed, stream, offset, TAG_READ_DRIFT_DIR) < 0.5 {
                 offset.saturating_sub(drift)
             } else {
                 offset.saturating_add(drift)
@@ -614,9 +686,17 @@ impl StorageIO for SimulatedIO {
         let to_read = buf.len().min(available);
         buf[..to_read].copy_from_slice(&storage.buffered[off..off + to_read]);
 
-        if state.should_fault(seed, fault.bit_flip_on_read_probability) && to_read > 0 {
-            let flip_pos = state.next_random_usize(seed, to_read);
-            let flip_bit = state.next_random_usize(seed, 8);
+        if to_read > 0
+            && SimState::fault_below(
+                seed,
+                stream,
+                offset,
+                TAG_READ_BITFLIP,
+                fault.bit_flip_on_read_probability,
+            )
+        {
+            let flip_pos = SimState::fault_usize(seed, stream, offset, TAG_READ_FLIP_POS, to_read);
+            let flip_bit = SimState::fault_usize(seed, stream, offset, TAG_READ_FLIP_BIT, 8);
             buf[flip_pos] ^= 1 << flip_bit;
         }
 
@@ -629,27 +709,47 @@ impl StorageIO for SimulatedIO {
     }
 
     fn write_at(&self, id: FileId, offset: u64, buf: &[u8]) -> io::Result<usize> {
-        self.jitter();
         let fault = self.effective_fault_config();
         let mut state = self.state.lock().unwrap();
         let sid = state.require_writable(id)?;
         let seed = self.rng_seed;
+        let stream = sid.0;
+        self.jitter(stream, offset);
 
         if state.storage.get(&sid).is_some_and(|s| s.io_poisoned) {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        if SimState::fault_below(
+            seed,
+            stream,
+            offset,
+            TAG_WRITE_IO,
+            fault.io_error_probability,
+        ) {
             return Err(io::Error::other("simulated EIO on write"));
         }
 
-        let torn_len = if buf.len() > 1 && state.should_fault(seed, fault.torn_page_probability) {
+        let torn_len = if buf.len() > 1
+            && SimState::fault_below(
+                seed,
+                stream,
+                offset,
+                TAG_WRITE_TORN,
+                fault.torn_page_probability,
+            ) {
             let page_base = (offset as usize) - ((offset as usize) % TORN_PAGE_BYTES);
             let page_end = page_base + TORN_PAGE_BYTES;
             let cap = page_end.saturating_sub(offset as usize).min(buf.len());
             let max_sectors = cap / SECTOR_BYTES;
             (max_sectors >= 2).then(|| {
-                let n = state.next_random_usize(seed, max_sectors - 1) + 1;
+                let n = SimState::fault_usize(
+                    seed,
+                    stream,
+                    offset,
+                    TAG_WRITE_TORN_SECTORS,
+                    max_sectors - 1,
+                ) + 1;
                 n * SECTOR_BYTES
             })
         } else {
@@ -658,18 +758,34 @@ impl StorageIO for SimulatedIO {
 
         let actual_len = match torn_len {
             Some(n) => n,
-            None if buf.len() > 1 && state.should_fault(seed, fault.partial_write_probability) => {
-                let partial = state.next_random_usize(seed, buf.len());
+            None if buf.len() > 1
+                && SimState::fault_below(
+                    seed,
+                    stream,
+                    offset,
+                    TAG_WRITE_PARTIAL,
+                    fault.partial_write_probability,
+                ) =>
+            {
+                let partial =
+                    SimState::fault_usize(seed, stream, offset, TAG_WRITE_PARTIAL_LEN, buf.len());
                 partial.max(1)
             }
             None => buf.len(),
         };
 
-        let misdirected = state.should_fault(seed, fault.misdirected_write_probability);
+        let misdirected = SimState::fault_below(
+            seed,
+            stream,
+            offset,
+            TAG_WRITE_MISDIR,
+            fault.misdirected_write_probability,
+        );
         let write_offset = if misdirected {
-            let drift_sectors = state.next_random_usize(seed, 8) + 1;
+            let drift_sectors =
+                SimState::fault_usize(seed, stream, offset, TAG_WRITE_DRIFT_SECTORS, 8) + 1;
             let drift = (drift_sectors * SECTOR_BYTES) as u64;
-            if state.next_random(seed) < 0.5 {
+            if SimState::fault_unit(seed, stream, offset, TAG_WRITE_DRIFT_DIR) < 0.5 {
                 offset.saturating_sub(drift)
             } else {
                 offset.saturating_add(drift)
@@ -698,21 +814,33 @@ impl StorageIO for SimulatedIO {
     }
 
     fn sync(&self, id: FileId) -> io::Result<()> {
-        self.jitter();
         let fault = self.effective_fault_config();
         let mut state = self.state.lock().unwrap();
         let sid = state.require_open(id)?;
         let seed = self.rng_seed;
+        let stream = sid.0;
+        let fsize = state
+            .storage
+            .get(&sid)
+            .map(|s| s.buffered.len() as u64)
+            .unwrap_or(0);
+        self.jitter(stream, fsize);
 
         if state.storage.get(&sid).is_some_and(|s| s.io_poisoned) {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        if SimState::fault_below(seed, stream, fsize, TAG_SYNC_IO, fault.io_error_probability) {
             return Err(io::Error::other("simulated EIO on sync"));
         }
 
-        if state.should_fault(seed, fault.sync_failure_probability) {
+        if SimState::fault_below(
+            seed,
+            stream,
+            fsize,
+            TAG_SYNC_FAILURE,
+            fault.sync_failure_probability,
+        ) {
             state.op_log.push(OpRecord::Sync {
                 fd: id,
                 succeeded: false,
@@ -720,7 +848,13 @@ impl StorageIO for SimulatedIO {
             return Err(io::Error::other("simulated dropped fsync"));
         }
 
-        let poison_after = state.should_fault(seed, fault.delayed_io_error_probability);
+        let poison_after = SimState::fault_below(
+            seed,
+            stream,
+            fsize,
+            TAG_SYNC_DELAYED,
+            fault.delayed_io_error_probability,
+        );
         let reorder_window = fault.sync_reorder_window.0 as usize;
 
         let evicted = if reorder_window > 0 {
@@ -838,7 +972,7 @@ impl StorageIO for SimulatedIO {
     }
 
     fn barrier(&self) -> io::Result<()> {
-        self.jitter();
+        self.jitter(0, 0);
         let mut state = self.state.lock().unwrap();
         let drained: Vec<PendingSync> = state.pending_syncs.drain(..).collect();
         drained.into_iter().for_each(|p| {
@@ -855,12 +989,31 @@ impl StorageIO for SimulatedIO {
         let mut state = self.state.lock().unwrap();
         let seed = self.rng_seed;
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        let dir_stream = path_stream(path);
+        let dir_entries = state
+            .paths
+            .keys()
+            .filter(|p| p.parent() == Some(path))
+            .count() as u64;
+
+        if SimState::fault_below(
+            seed,
+            dir_stream,
+            dir_entries,
+            TAG_SYNCDIR_IO,
+            fault.io_error_probability,
+        ) {
             return Err(io::Error::other("simulated EIO on sync_dir"));
         }
 
         let dir_path = path.to_path_buf();
-        let actually_persisted = !state.should_fault(seed, fault.dir_sync_failure_probability);
+        let actually_persisted = !SimState::fault_below(
+            seed,
+            dir_stream,
+            dir_entries,
+            TAG_SYNCDIR_FAILURE,
+            fault.dir_sync_failure_probability,
+        );
 
         if actually_persisted {
             state.dirs_durable.insert(dir_path.clone());