diff --git a/crates/tranquil-store/src/gauntlet/invariants.rs b/crates/tranquil-store/src/gauntlet/invariants.rs index 961f4e3..6e9eaf1 100644 --- a/crates/tranquil-store/src/gauntlet/invariants.rs +++ b/crates/tranquil-store/src/gauntlet/invariants.rs @@ -146,7 +146,7 @@ impl Invariant for Refcoun let inverse: Vec = 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 Invariant 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 = 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 Invariant for FsyncOr if !missing.is_empty() { let mut sorted = missing; sorted.sort_unstable(); + let ts_by_seq: HashMap = 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 = 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 Invariant for Tombsto }; let active = el.segments.last().copied(); + let retention_active = ctx.oracle.last_retention_active_segment(); let stale: Vec = 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(); diff --git a/crates/tranquil-store/src/gauntlet/oracle.rs b/crates/tranquil-store/src/gauntlet/oracle.rs index 08c2b72..8320815 100644 --- a/crates/tranquil-store/src/gauntlet/oracle.rs +++ b/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, last_synced_seq: Option, last_retention_cutoff_us: Option, + last_retention_active_segment: Option, lost_blocks: HashSet, } @@ -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) { + 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) { 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 { self.last_retention_cutoff_us } + + pub fn last_retention_active_segment(&self) -> Option { + self.last_retention_active_segment + } } pub(super) fn try_cid_to_fixed(cid: &Cid) -> Result { diff --git a/crates/tranquil-store/src/gauntlet/runner.rs b/crates/tranquil-store/src/gauntlet/runner.rs index 0dfb09f..36bb2aa 100644 --- a/crates/tranquil-store/src/gauntlet/runner.rs +++ b/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, ) -> 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, String>, - Crash: FnMut(), + Crash: FnMut() -> Vec, 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( let _ = harness.take(); } +fn record_crash_losses( + removed: &[PathBuf], + harness: Option<&Harness>, + oracle: &mut Oracle, +) { + mark_block_crash_losses(removed, harness, oracle); + mark_event_crash_losses(removed, oracle); + oracle.record_crash(); +} + +fn mark_block_crash_losses( + removed: &[PathBuf], + harness: Option<&Harness>, + oracle: &mut Oracle, +) { + let Some(h) = harness else { return }; + let data_dir = h.store.data_dir(); + let lost: Vec = removed + .iter() + .filter(|p| p.parent() == Some(data_dir)) + .filter_map(|p| p.file_stem()?.to_str()?.parse::().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 = + 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( @@ -806,7 +842,7 @@ where S: StorageIO + Send + Sync + 'static, C: Clock, Open: FnMut(usize) -> Result, String>, - Crash: FnMut(), + Crash: FnMut() -> Vec, { let mut errors: Vec = Vec::new(); for attempt in 0..MAX_REOPEN_ATTEMPTS { @@ -1234,11 +1270,13 @@ pub(super) async fn apply_op( 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( 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( _ => Ok(()), } })?; - oracle.record_retention(cutoff_us); + oracle.record_retention(cutoff_us, active_id); Ok(()) } @@ -1480,11 +1518,12 @@ fn compact_by_liveness( 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( }; 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, String>, - Crash: FnMut(), + Crash: FnMut() -> Vec, Quiesce: Fn(bool), { let ops: Vec = 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, diff --git a/crates/tranquil-store/src/gauntlet/scenarios.rs b/crates/tranquil-store/src/gauntlet/scenarios.rs index 213b8e8..0426986 100644 --- a/crates/tranquil-store/src/gauntlet/scenarios.rs +++ b/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, + } +} diff --git a/crates/tranquil-store/src/sim.rs b/crates/tranquil-store/src/sim.rs index 6e32e3b..5a58d2a 100644 --- a/crates/tranquil-store/src/sim.rs +++ b/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, dirs_durable: HashSet, op_log: Vec, - rng_counter: u64, next_fd_id: u64, next_storage_id: u64, pending_syncs: VecDeque, pending_deletes: Vec, } +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 { let mut state = self.state.lock().unwrap(); state.fds.clear(); @@ -417,13 +453,20 @@ impl SimulatedIO { } }); - let orphaned: Vec = state + let orphaned: HashSet = state .storage .iter() .filter(|(_, s)| !s.dir_entry_durable) .map(|(sid, _)| *sid) .collect(); + let removed_paths: Vec = 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 { @@ -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 { - 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 { - 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 = 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());