From 3d49e99cc3882c9f6101f18cc5e5e9ad0e657c7a Mon Sep 17 00:00:00 2001 From: Lewis Date: Tue, 2 Jun 2026 11:51:20 +0300 Subject: [PATCH] test(store): generic consistency checker, gauntlet fault/read Lewis: May this revision serve well! --- crates/tranquil-store/src/consistency.rs | 38 +++-- .../src/gauntlet/chaos_walker.rs | 46 +++++ crates/tranquil-store/src/gauntlet/op.rs | 6 +- crates/tranquil-store/src/gauntlet/oracle.rs | 4 + crates/tranquil-store/src/gauntlet/runner.rs | 159 +++++++++++++++++- .../tranquil-store/src/gauntlet/scenarios.rs | 101 ++++++++++- .../tranquil-store/src/gauntlet/workload.rs | 9 +- crates/tranquil-store/src/sim.rs | 58 ++++++- 8 files changed, 393 insertions(+), 28 deletions(-) diff --git a/crates/tranquil-store/src/consistency.rs b/crates/tranquil-store/src/consistency.rs index 97b18d8..709acf0 100644 --- a/crates/tranquil-store/src/consistency.rs +++ b/crates/tranquil-store/src/consistency.rs @@ -245,11 +245,16 @@ impl Default for ConsistencyCheckOptions { } } -pub fn verify_store_consistency( - blockstore: &TranquilBlockStore, +pub fn verify_store_consistency( + blockstore: &TranquilBlockStore, metastore: &Metastore, - eventlog: &EventLog, -) -> ConsistencyReport { + eventlog: &EventLog, +) -> 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( ) } -pub fn verify_store_consistency_with_options( - blockstore: &TranquilBlockStore, +pub fn verify_store_consistency_with_options( + blockstore: &TranquilBlockStore, metastore: &Metastore, - eventlog: &EventLog, + eventlog: &EventLog, 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( } } -fn check_orphan_data_files( - blockstore: &TranquilBlockStore, +fn check_orphan_data_files( + blockstore: &TranquilBlockStore, block_index: &BlockIndex, report: &mut ConsistencyReport, ) { @@ -600,8 +610,8 @@ fn check_orphan_data_files( }); } -fn check_missing_indexed_files( - blockstore: &TranquilBlockStore, +fn check_missing_indexed_files( + blockstore: &TranquilBlockStore, 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, +fn check_orphan_hint_files( + blockstore: &TranquilBlockStore, report: &mut ConsistencyReport, ) { let data_files: HashSet = match blockstore.list_data_files() { diff --git a/crates/tranquil-store/src/gauntlet/chaos_walker.rs b/crates/tranquil-store/src/gauntlet/chaos_walker.rs index dcf15a9..a7d71f2 100644 --- a/crates/tranquil-store/src/gauntlet/chaos_walker.rs +++ b/crates/tranquil-store/src/gauntlet/chaos_walker.rs @@ -77,6 +77,52 @@ pub fn mst_get_tolerant( } } +pub fn walk_mst_entries_tolerant( + store: &TranquilBlockStore, + root: Cid, + lost: &HashSet, +) -> Result>, String> { + let mut to_visit: Vec = vec![root]; + let mut visited: HashSet = 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::(&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( store: &TranquilBlockStore, cid_bytes: &CidBytes, diff --git a/crates/tranquil-store/src/gauntlet/op.rs b/crates/tranquil-store/src/gauntlet/op.rs index 7fd3062..9083324 100644 --- a/crates/tranquil-store/src/gauntlet/op.rs +++ b/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 + ) } } diff --git a/crates/tranquil-store/src/gauntlet/oracle.rs b/crates/tranquil-store/src/gauntlet/oracle.rs index 8320815..89a10a4 100644 --- a/crates/tranquil-store/src/gauntlet/oracle.rs +++ b/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 { + self.live.get(&(coll.clone(), rkey.clone())).copied() + } + pub fn set_root(&mut self, root: Cid) { self.current_root = Some(root); } diff --git a/crates/tranquil-store/src/gauntlet/runner.rs b/crates/tranquil-store/src/gauntlet/runner.rs index 36bb2aa..7428717 100644 --- a/crates/tranquil-store/src/gauntlet/runner.rs +++ b/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 { @@ -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( + store: &Arc>, + 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, + actual: &std::collections::HashMap, +) -> Option { + if expected == actual { + return None; + } + let missing: Vec = expected + .iter() + .filter(|(k, v)| actual.get(*k) != Some(v)) + .map(|(k, _)| k.clone()) + .take(8) + .collect(); + let extra: Vec = 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( + store: &Arc>, + root: Cid, + key: &str, + expected: Option, + 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( harness: &mut Harness, root: &mut Option, @@ -1311,15 +1406,41 @@ pub(super) async fn apply_op( 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 = entries.into_iter().collect(); + let expected: std::collections::HashMap = 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( 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( 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); diff --git a/crates/tranquil-store/src/gauntlet/scenarios.rs b/crates/tranquil-store/src/gauntlet/scenarios.rs index 0426986..823666f 100644 --- a/crates/tranquil-store/src/gauntlet/scenarios.rs +++ b/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, + } +} diff --git a/crates/tranquil-store/src/gauntlet/workload.rs b/crates/tranquil-store/src/gauntlet/workload.rs index 262fc32..48666fc 100644 --- a/crates/tranquil-store/src/gauntlet/workload.rs +++ b/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 { diff --git a/crates/tranquil-store/src/sim.rs b/crates/tranquil-store/src/sim.rs index 5a58d2a..ced6c32 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, 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, 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 { 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")); }