From 9b2cfb3a7e1db54df2e1f3cf3c0fd468caba0c9f Mon Sep 17 00:00:00 2001 From: Lewis Date: Wed, 29 Apr 2026 20:23:21 +0300 Subject: [PATCH] fix(tranquil-store): durable-tail recovery + sync semantics Lewis: May this revision serve well! --- crates/tranquil-pds/src/state.rs | 17 +- .../src/bin/tranquil_gauntlet.rs | 83 ++++- crates/tranquil-store/src/blockstore/mod.rs | 2 +- crates/tranquil-store/src/blockstore/store.rs | 306 +++++++++++++++++- crates/tranquil-store/src/eventlog/writer.rs | 58 ++++ crates/tranquil-store/src/gauntlet/farm.rs | 85 ++++- crates/tranquil-store/src/gauntlet/runner.rs | 209 ++++++++++-- crates/tranquil-store/src/sim.rs | 18 +- crates/tranquil-store/tests/gauntlet_smoke.rs | 112 +++++++ crates/tranquil-store/tests/sim_eventlog.rs | 22 +- 10 files changed, 841 insertions(+), 71 deletions(-) diff --git a/crates/tranquil-pds/src/state.rs b/crates/tranquil-pds/src/state.rs index 71b63c6..6ba45c1 100644 --- a/crates/tranquil-pds/src/state.rs +++ b/crates/tranquil-pds/src/state.rs @@ -523,13 +523,16 @@ fn wire_tranquil_store( let metastore = Metastore::open(&metastore_dir, metastore_config).expect("failed to open metastore"); - let blockstore = TranquilBlockStore::open(BlockStoreConfig { - data_dir: blockstore_data_dir, - index_dir: blockstore_index_dir, - max_file_size: store_cfg.max_blockstore_file_size, - group_commit: Default::default(), - shard_count: tranquil_store::blockstore::DEFAULT_SHARD_COUNT, - }) + let blockstore = TranquilBlockStore::open_with_retry( + BlockStoreConfig { + data_dir: blockstore_data_dir, + index_dir: blockstore_index_dir, + max_file_size: store_cfg.max_blockstore_file_size, + group_commit: Default::default(), + shard_count: tranquil_store::blockstore::DEFAULT_SHARD_COUNT, + }, + tranquil_store::blockstore::OpenRetryPolicy::default(), + ) .expect("failed to open blockstore"); let event_log = EventLog::open( diff --git a/crates/tranquil-store/src/bin/tranquil_gauntlet.rs b/crates/tranquil-store/src/bin/tranquil_gauntlet.rs index d3bff4b..055153c 100644 --- a/crates/tranquil-store/src/bin/tranquil_gauntlet.rs +++ b/crates/tranquil-store/src/bin/tranquil_gauntlet.rs @@ -64,6 +64,16 @@ enum Cmd { #[arg(long)] config: Option, + /// Tempdir parent for `IoBackend::Real` seeds only - ignored for + /// flaky-mount and simulated backends. Repeatable; each rayon worker + /// thread is pinned to one root so concurrent seeds on different + /// threads land on different mounts. Default `/tmp`. Also reads + /// colon-separated paths from `GAUNTLET_SCRATCH_ROOTS`. Set + /// `RAYON_NUM_THREADS=N` to cap workers; for full distribution pass + /// one root per worker. + #[arg(long)] + scratch_root: Vec, + /// Skip shrinking when dumping regressions. #[arg(long)] no_shrink: bool, @@ -95,6 +105,13 @@ enum Cmd { #[arg(long)] dump_regressions: Option, + /// Same as `farm --scratch-root`: tempdir parent for + /// `IoBackend::Real` seeds only, pinned per worker thread. Ignored + /// for flaky-mount and simulated backends. Repeatable; reads + /// colon-separated paths from `GAUNTLET_SCRATCH_ROOTS`. + #[arg(long)] + scratch_root: Vec, + /// Skip shrinking when dumping regressions. #[arg(long)] no_shrink: bool, @@ -159,6 +176,8 @@ struct ConfigFile { #[serde(default)] dump_regressions: Option, #[serde(default)] + scratch_roots: Vec, + #[serde(default)] overrides: ConfigOverrides, } @@ -178,6 +197,8 @@ struct SweepConfigFile { #[serde(default)] dump_regressions: Option, #[serde(default)] + scratch_roots: Vec, + #[serde(default)] base_overrides: ConfigOverrides, #[serde(default)] axes: SweepAxes, @@ -405,6 +426,7 @@ struct FarmPlan { seeds: u64, hours: Option, dump_regressions: Option, + scratch_roots: Vec, overrides: ConfigOverrides, shrink: bool, shrink_budget: usize, @@ -418,6 +440,7 @@ fn resolve_farm( hours: Option, dump_regressions: Option, config: Option, + scratch_root: Vec, shrink: bool, shrink_budget: usize, ) -> Result { @@ -443,6 +466,11 @@ fn resolve_farm( } let dump_regressions = dump_regressions.or_else(|| file.as_ref().and_then(|f| f.dump_regressions.clone())); + let file_scratch_roots = file + .as_ref() + .map(|f| f.scratch_roots.clone()) + .unwrap_or_default(); + let scratch_roots = resolve_scratch_roots(scratch_root, file_scratch_roots)?; let overrides = file.map(|f| f.overrides).unwrap_or_default(); Ok(FarmPlan { scenario, @@ -450,12 +478,50 @@ fn resolve_farm( seeds, hours, dump_regressions, + scratch_roots, overrides, shrink, shrink_budget, }) } +const SCRATCH_ROOTS_ENV: &str = "GAUNTLET_SCRATCH_ROOTS"; + +fn resolve_scratch_roots( + cli: Vec, + config_file: Vec, +) -> Result, String> { + let env_roots: Vec = std::env::var(SCRATCH_ROOTS_ENV) + .ok() + .filter(|s| !s.is_empty()) + .map(|s| s.split(':').map(PathBuf::from).collect()) + .unwrap_or_default(); + let candidate: Vec = if !cli.is_empty() { + cli + } else if !config_file.is_empty() { + config_file + } else { + env_roots + }; + candidate + .into_iter() + .map(|p| validate_scratch_root(&p).map(|_| p)) + .collect() +} + +fn validate_scratch_root(path: &Path) -> Result<(), String> { + match path.metadata() { + Ok(m) if m.is_dir() => {} + Ok(_) => return Err(format!("scratch root not a directory: {}", path.display())), + Err(e) => return Err(format!("scratch root {}: {e}", path.display())), + } + tempfile::Builder::new() + .prefix(".tranquil-gauntlet-probe-") + .tempfile_in(path) + .map(|_| ()) + .map_err(|e| format!("scratch root {} not writable: {e}", path.display())) +} + fn validate_hours(h: f64) -> Result<(), String> { if !h.is_finite() || h <= 0.0 { return Err(format!("invalid --hours={h}: must be positive and finite")); @@ -564,6 +630,7 @@ fn main() -> ExitCode { hours, dump_regressions, config, + scratch_root, no_shrink, shrink_budget, } => { @@ -574,6 +641,7 @@ fn main() -> ExitCode { hours, dump_regressions, config, + scratch_root, !no_shrink, shrink_budget, ) { @@ -625,6 +693,7 @@ fn main() -> ExitCode { seed_start, seeds, dump_regressions, + scratch_root, no_shrink, shrink_budget, max_runs, @@ -634,6 +703,7 @@ fn main() -> ExitCode { seed_start, seeds, dump_regressions, + scratch_root, !no_shrink, shrink_budget, max_runs, @@ -659,17 +729,20 @@ struct SweepPlan { seed_start: u64, seeds: u64, dump_regressions: Option, + scratch_roots: Vec, shrink: bool, shrink_budget: usize, base_overrides: ConfigOverrides, axes: Vec, } +#[allow(clippy::too_many_arguments)] fn resolve_sweep( config: PathBuf, seed_start: Option, seeds: Option, dump_regressions: Option, + scratch_root: Vec, shrink: bool, shrink_budget: usize, max_runs: u64, @@ -687,6 +760,7 @@ fn resolve_sweep( return Err("--shrink-budget must be greater than zero".to_string()); } let dump_regressions = dump_regressions.or(file.dump_regressions.clone()); + let scratch_roots = resolve_scratch_roots(scratch_root, file.scratch_roots.clone())?; let axes = file.axes.axis_values(); if axes.is_empty() { return Err("sweep produced no combinations".to_string()); @@ -705,6 +779,7 @@ fn resolve_sweep( seed_start, seeds, dump_regressions, + scratch_roots, shrink, shrink_budget, base_overrides: file.base_overrides, @@ -749,6 +824,7 @@ fn run_sweep(plan: SweepPlan, rt: &Runtime, interrupt: Arc) -> ExitC seed_start, seeds, dump_regressions, + scratch_roots, shrink, shrink_budget, base_overrides, @@ -782,12 +858,13 @@ fn run_sweep(plan: SweepPlan, rt: &Runtime, interrupt: Arc) -> ExitC axis_values.apply_to(&mut overrides); let combo_start = Instant::now(); let overrides_for_farm = overrides.clone(); - let reports = farm::run_many_timed( + let reports = farm::run_many_timed_with_scratch_roots( move |s| { let mut cfg = config_for(scenario, s); overrides_for_farm.apply_to(&mut cfg); cfg }, + &scratch_roots, (seed_start..end).map(Seed), ); let combo_wall = combo_start.elapsed(); @@ -840,6 +917,7 @@ fn run_farm(plan: FarmPlan, rt: &Runtime, interrupt: Arc) -> ExitCod seeds, hours, dump_regressions, + scratch_roots, overrides, shrink, shrink_budget, @@ -871,12 +949,13 @@ fn run_farm(plan: FarmPlan, rt: &Runtime, interrupt: Arc) -> ExitCod }; let overrides_ref = &overrides; let batch_start = Instant::now(); - let reports = farm::run_many_timed( + let reports = farm::run_many_timed_with_scratch_roots( |s| { let mut cfg = config_for(scenario, s); overrides_ref.apply_to(&mut cfg); cfg }, + &scratch_roots, (next_seed..end).map(Seed), ); let batch_wall = batch_start.elapsed(); diff --git a/crates/tranquil-store/src/blockstore/mod.rs b/crates/tranquil-store/src/blockstore/mod.rs index 2bc9c03..6d9191f 100644 --- a/crates/tranquil-store/src/blockstore/mod.rs +++ b/crates/tranquil-store/src/blockstore/mod.rs @@ -27,7 +27,7 @@ pub use hint::{ pub use manager::{CachedHandle, DEFAULT_MAX_FILE_SIZE, DataFileManager}; pub use reader::{BlockStoreReader, ReadError}; pub use store::QuiesceGuard; -pub use store::{BlockStoreConfig, DEFAULT_SHARD_COUNT, TranquilBlockStore}; +pub use store::{BlockStoreConfig, DEFAULT_SHARD_COUNT, OpenRetryPolicy, TranquilBlockStore}; pub use types::{ BlockLength, BlockLocation, BlockOffset, BlockstoreSnapshot, CidBytes, CollectionResult, CommitEpoch, CompactionResult, DataFileId, EpochCounter, HintOffset, IndexEntry, LivenessInfo, diff --git a/crates/tranquil-store/src/blockstore/store.rs b/crates/tranquil-store/src/blockstore/store.rs index 2859644..114164a 100644 --- a/crates/tranquil-store/src/blockstore/store.rs +++ b/crates/tranquil-store/src/blockstore/store.rs @@ -1,7 +1,9 @@ use std::collections::HashMap; use std::io; +use std::num::NonZeroU8; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::Duration; use bytes::Bytes; use cid::Cid; @@ -150,6 +152,24 @@ impl Drop for WriterHandle { } } +#[derive(Clone, Copy, Debug)] +pub struct OpenRetryPolicy { + pub max_attempts: NonZeroU8, + pub initial_backoff: Duration, + pub max_backoff: Duration, +} + +impl Default for OpenRetryPolicy { + fn default() -> Self { + const DEFAULT_MAX_ATTEMPTS: NonZeroU8 = NonZeroU8::new(5).unwrap(); + Self { + max_attempts: DEFAULT_MAX_ATTEMPTS, + initial_backoff: Duration::from_millis(100), + max_backoff: Duration::from_secs(2), + } + } +} + impl TranquilBlockStore { pub fn open(config: BlockStoreConfig) -> Result { Self::open_with_hook(config, None) @@ -161,6 +181,50 @@ impl TranquilBlockStore { ) -> Result { Self::open_with_io_hook(config, RealIO::new, post_sync_hook) } + + pub fn open_with_retry( + config: BlockStoreConfig, + policy: OpenRetryPolicy, + ) -> Result { + retry_with_backoff(policy, &mut |_| Self::open(config.clone())) + } +} + +fn retry_with_backoff(policy: OpenRetryPolicy, op: &mut F) -> Result +where + F: FnMut(u8) -> Result, +{ + retry_attempt(policy, op, 0, policy.initial_backoff) +} + +fn retry_attempt( + policy: OpenRetryPolicy, + op: &mut F, + attempt: u8, + backoff: Duration, +) -> Result +where + F: FnMut(u8) -> Result, +{ + match op(attempt) { + Ok(t) => Ok(t), + Err(e) if attempt + 1 >= policy.max_attempts.get() => Err(e), + Err(e) => { + tracing::warn!( + attempt, + error = %e, + backoff_ms = u64::try_from(backoff.as_millis()).unwrap_or(u64::MAX), + "blockstore open failed, retrying" + ); + std::thread::sleep(backoff); + retry_attempt( + policy, + op, + attempt + 1, + (backoff * 2).min(policy.max_backoff), + ) + } + } } impl TranquilBlockStore { @@ -331,15 +395,7 @@ impl TranquilBlockStore { let scan_pos = &mut { start_offset }; let (scanned_entries, last_valid_end) = std::iter::from_fn(|| { match super::data_file::decode_block_record(io, fd, *scan_pos, file_size) { - Err(e) => { - tracing::warn!( - file_id = %file_id, - offset = scan_pos.raw(), - error = %e, - "IO error during recovery scan, stopping" - ); - None - } + Err(e) => Some(Err(e)), Ok(None) => None, Ok(Some(ReadBlockRecord::Valid { offset, @@ -354,7 +410,7 @@ impl TranquilBlockStore { let record_size = BLOCK_RECORD_OVERHEAD as u64 + u64::from(raw_len); let new_end = offset.advance(record_size); *scan_pos = new_end; - Some(( + Some(Ok(( cid_bytes, BlockLocation { file_id, @@ -362,20 +418,30 @@ impl TranquilBlockStore { length, }, new_end, - )) + ))) } Ok(Some(ReadBlockRecord::Corrupted { .. } | ReadBlockRecord::Truncated { .. })) => { None } } }) - .fold( + .try_fold( (Vec::new(), start_offset), - |(mut entries, _), (cid, loc, new_end)| { + |(mut entries, _), item: io::Result<_>| { + let (cid, loc, new_end) = item?; entries.push((cid, loc)); - (entries, new_end) + Ok::<_, io::Error>((entries, new_end)) }, - ); + ) + .map_err(|e| { + tracing::warn!( + file_id = %file_id, + offset = scan_pos.raw(), + error = %e, + "IO error during recovery scan, aborting to preserve durable tail" + ); + RepoError::storage(e) + })?; if file_size > last_valid_end.raw() { tracing::info!( @@ -713,3 +779,213 @@ impl TranquilBlockStore { Ok(self.index.get(&cid_bytes).map(|entry| entry.refcount.raw())) } } + +#[cfg(test)] +mod tests { + use super::*; + + use std::collections::HashMap; + use std::sync::Mutex; + use std::sync::atomic::{AtomicBool, Ordering}; + + use crate::blockstore::data_file::{ + BLOCK_FORMAT_VERSION, BLOCK_HEADER_SIZE, BLOCK_MAGIC, encode_block_record, + }; + use crate::blockstore::manager::DATA_FILE_EXTENSION; + use crate::io::FileId; + + struct EioOnReadAtRange { + inner: RealIO, + target_path: PathBuf, + target_min: u64, + target_max: u64, + fired: AtomicBool, + fd_paths: Mutex>, + } + + impl StorageIO for EioOnReadAtRange { + fn open(&self, path: &Path, opts: OpenOptions) -> io::Result { + let fd = self.inner.open(path, opts)?; + self.fd_paths + .lock() + .unwrap() + .insert(fd, path.to_path_buf()); + Ok(fd) + } + + fn close(&self, fd: FileId) -> io::Result<()> { + self.fd_paths.lock().unwrap().remove(&fd); + self.inner.close(fd) + } + + fn read_at(&self, fd: FileId, offset: u64, buf: &mut [u8]) -> io::Result { + let path_match = self.fd_paths.lock().unwrap().get(&fd).cloned(); + let in_target_range = path_match.as_ref() == Some(&self.target_path) + && offset >= self.target_min + && offset <= self.target_max; + if in_target_range && !self.fired.swap(true, Ordering::SeqCst) { + return Err(io::Error::other("simulated EIO on read")); + } + self.inner.read_at(fd, offset, buf) + } + + fn write_at(&self, fd: FileId, offset: u64, buf: &[u8]) -> io::Result { + self.inner.write_at(fd, offset, buf) + } + + fn sync(&self, fd: FileId) -> io::Result<()> { + self.inner.sync(fd) + } + + fn file_size(&self, fd: FileId) -> io::Result { + self.inner.file_size(fd) + } + + fn truncate(&self, fd: FileId, size: u64) -> io::Result<()> { + self.inner.truncate(fd, size) + } + + fn rename(&self, from: &Path, to: &Path) -> io::Result<()> { + self.inner.rename(from, to) + } + + fn delete(&self, path: &Path) -> io::Result<()> { + self.inner.delete(path) + } + + fn mkdir(&self, path: &Path) -> io::Result<()> { + self.inner.mkdir(path) + } + + fn sync_dir(&self, path: &Path) -> io::Result<()> { + self.inner.sync_dir(path) + } + + fn list_dir(&self, path: &Path) -> io::Result> { + self.inner.list_dir(path) + } + } + + #[test] + fn scan_and_index_does_not_truncate_acked_block_on_transient_eio() { + let tmp = tempfile::TempDir::new().unwrap(); + let data_dir = tmp.path().join("data"); + let index_dir = tmp.path().join("index"); + std::fs::create_dir_all(&data_dir).unwrap(); + std::fs::create_dir_all(&index_dir).unwrap(); + + let file_id = DataFileId::new(0); + let file_path = data_dir.join(format!("{file_id}.{DATA_FILE_EXTENSION}")); + + let setup = RealIO::new(); + let fd = setup.open(&file_path, OpenOptions::read_write()).unwrap(); + let mut header = [0u8; BLOCK_HEADER_SIZE]; + header[..4].copy_from_slice(&BLOCK_MAGIC); + header[4] = BLOCK_FORMAT_VERSION; + setup.write_all_at(fd, 0, &header).unwrap(); + + let cid_a = [0xAAu8; CID_SIZE]; + let data_a = vec![1u8; 64]; + let block_a_offset = BlockOffset::new(BLOCK_HEADER_SIZE as u64); + let len_a = + encode_block_record(&setup, fd, block_a_offset, &cid_a, &data_a).unwrap(); + + let block_b_offset_raw = BLOCK_HEADER_SIZE as u64 + len_a; + let block_b_offset = BlockOffset::new(block_b_offset_raw); + let cid_b = [0xBBu8; CID_SIZE]; + let data_b = vec![2u8; 64]; + let len_b = encode_block_record(&setup, fd, block_b_offset, &cid_b, &data_b).unwrap(); + + setup.sync(fd).unwrap(); + setup.close(fd).unwrap(); + drop(setup); + + let total_size = block_b_offset_raw + len_b; + assert_eq!(std::fs::metadata(&file_path).unwrap().len(), total_size); + + let wrapper = EioOnReadAtRange { + inner: RealIO::new(), + target_path: file_path.clone(), + target_min: block_b_offset_raw, + target_max: block_b_offset_raw + (BLOCK_RECORD_OVERHEAD as u64) - 1, + fired: AtomicBool::new(false), + fd_paths: Mutex::new(HashMap::new()), + }; + + let index = BlockIndex::open(&index_dir).unwrap(); + + let result = TranquilBlockStore::::replay_single_file( + &wrapper, + &data_dir, + &index, + file_id, + BlockOffset::new(BLOCK_HEADER_SIZE as u64), + ); + + assert!( + result.is_err(), + "replay must surface transient EIO instead of silently truncating" + ); + + let post_size = std::fs::metadata(&file_path).unwrap().len(); + assert_eq!( + post_size, total_size, + "scan truncated durable acked block past EIO point: expected {total_size} bytes, got {post_size}" + ); + } + + fn instant_policy(max_attempts: u8) -> OpenRetryPolicy { + OpenRetryPolicy { + max_attempts: NonZeroU8::new(max_attempts).expect("max_attempts must be nonzero"), + initial_backoff: Duration::ZERO, + max_backoff: Duration::ZERO, + } + } + + #[test] + fn retry_with_backoff_succeeds_on_first_attempt() { + let calls = std::sync::atomic::AtomicUsize::new(0); + let result = retry_with_backoff(instant_policy(5), &mut |_| { + calls.fetch_add(1, Ordering::Relaxed); + Ok::(42) + }); + assert_eq!(result.expect("ok"), 42); + assert_eq!(calls.load(Ordering::Relaxed), 1); + } + + #[test] + fn retry_with_backoff_recovers_after_transient_failures() { + let calls = std::sync::atomic::AtomicUsize::new(0); + let result = retry_with_backoff(instant_policy(5), &mut |_| { + let n = calls.fetch_add(1, Ordering::Relaxed); + if n >= 2 { + Ok::(7) + } else { + Err(RepoError::storage(io::Error::other("transient EIO"))) + } + }); + assert_eq!(result.expect("ok"), 7); + assert_eq!(calls.load(Ordering::Relaxed), 3); + } + + #[test] + fn retry_with_backoff_gives_up_after_max_attempts() { + let calls = std::sync::atomic::AtomicUsize::new(0); + let result: Result = retry_with_backoff(instant_policy(3), &mut |_| { + calls.fetch_add(1, Ordering::Relaxed); + Err(RepoError::storage(io::Error::other("permanent EIO"))) + }); + assert!(result.is_err(), "expected exhaustion error"); + assert_eq!(calls.load(Ordering::Relaxed), 3); + } + + #[test] + fn retry_with_backoff_passes_attempt_index_to_op() { + let observed = std::sync::Mutex::new(Vec::::new()); + let _result: Result<(), RepoError> = retry_with_backoff(instant_policy(4), &mut |attempt| { + observed.lock().unwrap().push(attempt); + Err(RepoError::storage(io::Error::other("EIO"))) + }); + assert_eq!(*observed.lock().unwrap(), vec![0, 1, 2, 3]); + } +} diff --git a/crates/tranquil-store/src/eventlog/writer.rs b/crates/tranquil-store/src/eventlog/writer.rs index ecc8777..283dae6 100644 --- a/crates/tranquil-store/src/eventlog/writer.rs +++ b/crates/tranquil-store/src/eventlog/writer.rs @@ -1196,4 +1196,62 @@ mod tests { assert!(writer.rotate_if_needed().unwrap().is_none()); } + + #[test] + fn sync_must_not_certify_durability_when_io_sync_silently_drops() { + use crate::sim::{FaultConfig, Probability}; + + let sim = Arc::new(SimulatedIO::new( + 0, + FaultConfig { + sync_failure_probability: Probability::new(1.0), + ..FaultConfig::none() + }, + )); + sim.set_pristine_mode(true); + + let mgr = Arc::new( + SegmentManager::new(Arc::clone(&sim), PathBuf::from("/segments"), 64 * 1024).unwrap(), + ); + + let mut writer = + EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD) + .unwrap(); + + sim.set_pristine_mode(false); + + writer + .append( + DidHash::from_did("did:plc:bug2"), + EventTypeTag::COMMIT, + b"bug2-payload".to_vec(), + ) + .unwrap(); + + assert!( + writer.sync().is_err(), + "sync must surface dropped fsync as an error" + ); + let claimed_synced = writer.synced_seq(); + assert_eq!( + claimed_synced.raw(), + 0, + "synced_seq must not advance past a failed sync" + ); + drop(writer); + + mgr.shutdown(); + sim.crash(); + sim.set_pristine_mode(true); + + let reopened = + EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD) + .unwrap(); + let actually_durable = reopened.current_seq(); + + assert!( + actually_durable >= claimed_synced, + "writer claimed sync through {claimed_synced} but post-crash recovery only reaches {actually_durable}" + ); + } } diff --git a/crates/tranquil-store/src/gauntlet/farm.rs b/crates/tranquil-store/src/gauntlet/farm.rs index 9d3a6f0..d8875b2 100644 --- a/crates/tranquil-store/src/gauntlet/farm.rs +++ b/crates/tranquil-store/src/gauntlet/farm.rs @@ -1,5 +1,6 @@ use std::cell::RefCell; use std::panic::{AssertUnwindSafe, catch_unwind}; +use std::path::PathBuf; use std::time::{Duration, Instant}; use rayon::prelude::*; @@ -44,6 +45,17 @@ pub fn run_many_timed( make_config: F, seeds: impl IntoIterator, ) -> Vec<(GauntletReport, Duration)> +where + F: Fn(Seed) -> GauntletConfig + Sync + Send, +{ + run_many_timed_with_scratch_roots(make_config, &[], seeds) +} + +pub fn run_many_timed_with_scratch_roots( + make_config: F, + scratch_roots: &[PathBuf], + seeds: impl IntoIterator, +) -> Vec<(GauntletReport, Duration)> where F: Fn(Seed) -> GauntletConfig + Sync + Send, { @@ -51,10 +63,14 @@ where seeds .into_par_iter() .map(|s| { + let scratch = scratch_for_thread(scratch_roots, rayon::current_thread_index()); let start = Instant::now(); let outcome = catch_unwind(AssertUnwindSafe(|| { let cfg = make_config(s); - let gauntlet = Gauntlet::new(cfg).expect("build gauntlet"); + let mut gauntlet = Gauntlet::new(cfg).expect("build gauntlet"); + if let Some(root) = scratch { + gauntlet = gauntlet.with_scratch_root(root); + } with_runtime(|rt| rt.block_on(gauntlet.run())) })); let report = outcome.unwrap_or_else(|payload| { @@ -66,6 +82,14 @@ where .collect() } +fn scratch_for_thread(roots: &[PathBuf], thread_idx: Option) -> Option { + if roots.is_empty() { + None + } else { + Some(roots[thread_idx.unwrap_or(0) % roots.len()].clone()) + } +} + fn panic_report(seed: Seed, payload: Box) -> GauntletReport { let msg = payload .downcast_ref::<&'static str>() @@ -84,3 +108,62 @@ fn panic_report(seed: Seed, payload: Box) -> GauntletR ops: OpStream::empty(), } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn scratch_for_thread_returns_none_when_roots_empty() { + assert!(scratch_for_thread(&[], Some(0)).is_none()); + assert!(scratch_for_thread(&[], Some(7)).is_none()); + assert!(scratch_for_thread(&[], None).is_none()); + } + + #[test] + fn scratch_for_thread_round_robins_across_roots() { + let roots = vec![ + PathBuf::from("/scratch/a"), + PathBuf::from("/scratch/b"), + PathBuf::from("/scratch/c"), + ]; + let assigned: Vec = (0..7) + .map(|i| scratch_for_thread(&roots, Some(i)).expect("scratch path")) + .collect(); + assert_eq!( + assigned, + vec![ + PathBuf::from("/scratch/a"), + PathBuf::from("/scratch/b"), + PathBuf::from("/scratch/c"), + PathBuf::from("/scratch/a"), + PathBuf::from("/scratch/b"), + PathBuf::from("/scratch/c"), + PathBuf::from("/scratch/a"), + ] + ); + } + + #[test] + fn scratch_for_thread_with_single_root_returns_same_path() { + let roots = vec![PathBuf::from("/scratch/only")]; + (0..5).for_each(|i| { + assert_eq!( + scratch_for_thread(&roots, Some(i)), + Some(PathBuf::from("/scratch/only")) + ); + }); + } + + #[test] + fn scratch_for_thread_falls_back_to_root_zero_outside_pool() { + let roots = vec![ + PathBuf::from("/scratch/a"), + PathBuf::from("/scratch/b"), + ]; + assert_eq!( + scratch_for_thread(&roots, None), + Some(PathBuf::from("/scratch/a")) + ); + } +} diff --git a/crates/tranquil-store/src/gauntlet/runner.rs b/crates/tranquil-store/src/gauntlet/runner.rs index 6752aed..b8ff830 100644 --- a/crates/tranquil-store/src/gauntlet/runner.rs +++ b/crates/tranquil-store/src/gauntlet/runner.rs @@ -178,6 +178,7 @@ pub struct SharedState { pub struct Gauntlet { config: GauntletConfig, + scratch_root: Option, } #[derive(Debug, thiserror::Error)] @@ -185,7 +186,15 @@ pub enum GauntletBuildError {} impl Gauntlet { pub fn new(config: GauntletConfig) -> Result { - Ok(Self { config }) + Ok(Self { + config, + scratch_root: None, + }) + } + + pub fn with_scratch_root(mut self, root: PathBuf) -> Self { + self.scratch_root = Some(root); + self } pub fn generate_ops(&self) -> OpStream { @@ -211,6 +220,7 @@ impl Gauntlet { let ops_counter = Arc::new(AtomicUsize::new(0)); let op_errors_counter = Arc::new(AtomicUsize::new(0)); let restarts_counter = Arc::new(AtomicUsize::new(0)); + let scratch_root = self.scratch_root; let fut: std::pin::Pin + Send>> = match self.config.io { IoBackend::Real => Box::pin(run_inner_real( @@ -219,6 +229,7 @@ impl Gauntlet { ops_counter.clone(), op_errors_counter.clone(), restarts_counter.clone(), + scratch_root, )), IoBackend::RealWithFlaky { flaky } => Box::pin(run_inner_real_with_flaky( self.config, @@ -269,8 +280,12 @@ async fn run_inner_real( ops_counter: Arc, op_errors_counter: Arc, restarts_counter: Arc, + scratch_root: Option, ) -> GauntletReport { - let dir = tempfile::TempDir::new().expect("tempdir"); + let dir = match scratch_root.as_deref() { + Some(parent) => tempfile::TempDir::new_in(parent).expect("tempdir in scratch root"), + None => tempfile::TempDir::new().expect("tempdir"), + }; let root = dir.path().to_path_buf(); let report = run_inner_real_on_root( config, @@ -519,22 +534,24 @@ where let mut oracle = Oracle::new(); let mut violations: Vec = Vec::new(); - let mut harness: Option> = match open(0) { - Ok(h) => Some(h), - Err(e) => { - return GauntletReport { - seed: config.seed, - ops_executed: OpsExecuted(0), - op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)), - restarts: RestartCount(0), - violations: vec![InvariantViolation { - invariant: "OpenStore", - detail: format!("initial open: {e}"), - }], - ops: OpStream::empty(), - }; - } - }; + let mut harness: Option> = + match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await + { + Ok(h) => Some(h), + Err(e) => { + return GauntletReport { + seed: config.seed, + ops_executed: OpsExecuted(0), + op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)), + restarts: RestartCount(0), + violations: vec![InvariantViolation { + invariant: "OpenStore", + detail: format!("initial open: {e}"), + }], + ops: OpStream::empty(), + }; + } + }; let mut root: Option = None; let mut restart_rng = Lcg::new(Seed(config.seed.0 ^ 0xA5A5_A5A5_A5A5_A5A5)); let mut sample_rng = Lcg::new(Seed(config.seed.0 ^ 0x5A5A_5A5A_5A5A_5A5A)); @@ -1582,22 +1599,24 @@ where let mut sample_rng = Lcg::new(Seed(config.seed.0 ^ 0x5A5A_5A5A_5A5A_5A5A)); let chunks = compute_chunks(config.restart_policy, total_ops, &mut restart_rng); - let mut harness: Option> = match open(0) { - Ok(h) => Some(h), - Err(e) => { - return GauntletReport { - seed: config.seed, - ops_executed: OpsExecuted(0), - op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)), - restarts: RestartCount(0), - violations: vec![InvariantViolation { - invariant: "OpenStore", - detail: format!("initial open: {e}"), - }], - ops: OpStream::empty(), - }; - } - }; + let mut harness: Option> = + match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await + { + Ok(h) => Some(h), + Err(e) => { + return GauntletReport { + seed: config.seed, + ops_executed: OpsExecuted(0), + op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)), + restarts: RestartCount(0), + violations: vec![InvariantViolation { + invariant: "OpenStore", + detail: format!("initial open: {e}"), + }], + ops: OpStream::empty(), + }; + } + }; let mut root: Option = None; let mut oracle = Oracle::new(); let mut halt_ops = false; @@ -1806,3 +1825,125 @@ where ops: OpStream::empty(), } } + +#[cfg(test)] +mod tests { + use super::*; + + fn minimal_config() -> GauntletConfig { + GauntletConfig { + seed: Seed(0), + io: IoBackend::Real, + workload: WorkloadModel::default(), + op_count: OpCount(0), + invariants: InvariantSet::EMPTY, + limits: RunLimits { + max_wall_ms: Some(WallMs(30_000)), + }, + restart_policy: RestartPolicy::Never, + store: StoreConfig { + max_file_size: MaxFileSize(8 * 1024), + group_commit: GroupCommitConfig::default(), + shard_count: ShardCount(1), + }, + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } + } + + fn flaky_open( + attempts: Arc, + sim: Arc, + store_cfg: BlockStoreConfig, + ) -> impl FnMut(usize) -> Result>, String> + Send + 'static { + move |_attempt: usize| -> Result>, String> { + let n = attempts.fetch_add(1, Ordering::Relaxed); + if n == 0 { + return Err("simulated EIO on initial open".to_string()); + } + let factory_sim = Arc::clone(&sim); + let make_io = move || Arc::clone(&factory_sim); + TranquilBlockStore::>::open_with_io(store_cfg.clone(), make_io) + .map(|s| Harness { + store: Arc::new(s), + eventlog: None, + }) + .map_err(|e| e.to_string()) + } + } + + #[tokio::test] + async fn run_inner_generic_retries_initial_open_on_transient_io_error() { + let dir = tempfile::TempDir::new().expect("tempdir"); + let cfg = minimal_config(); + let store_cfg = blockstore_config(dir.path(), &cfg.store); + let sim: Arc = Arc::new(SimulatedIO::pristine(0)); + let attempts = Arc::new(AtomicUsize::new(0)); + + let report = run_inner_generic::, _, _>( + cfg, + OpStream::empty(), + Arc::new(AtomicUsize::new(0)), + Arc::new(AtomicUsize::new(0)), + Arc::new(AtomicUsize::new(0)), + flaky_open(Arc::clone(&attempts), Arc::clone(&sim), store_cfg), + || {}, + true, + Duration::ZERO, + ) + .await; + + let opens: Vec<&InvariantViolation> = report + .violations + .iter() + .filter(|v| v.invariant == "OpenStore") + .collect(); + assert!( + opens.is_empty(), + "expected initial open to retry, got OpenStore violations: {opens:?}" + ); + let total = attempts.load(Ordering::Relaxed); + assert!( + total >= 2, + "expected at least one retry after first failure, attempts={total}" + ); + } + + #[tokio::test] + async fn run_inner_generic_concurrent_retries_initial_open_on_transient_io_error() { + let dir = tempfile::TempDir::new().expect("tempdir"); + let mut cfg = minimal_config(); + cfg.writer_concurrency = WriterConcurrency(2); + let store_cfg = blockstore_config(dir.path(), &cfg.store); + let sim: Arc = Arc::new(SimulatedIO::pristine(0)); + let attempts = Arc::new(AtomicUsize::new(0)); + + let report = run_inner_generic_concurrent::, _, _>( + cfg, + OpStream::empty(), + Arc::new(AtomicUsize::new(0)), + Arc::new(AtomicUsize::new(0)), + Arc::new(AtomicUsize::new(0)), + flaky_open(Arc::clone(&attempts), Arc::clone(&sim), store_cfg), + || {}, + true, + Duration::ZERO, + ) + .await; + + let opens: Vec<&InvariantViolation> = report + .violations + .iter() + .filter(|v| v.invariant == "OpenStore") + .collect(); + assert!( + opens.is_empty(), + "expected initial open to retry, got OpenStore violations: {opens:?}" + ); + let total = attempts.load(Ordering::Relaxed); + assert!( + total >= 2, + "expected at least one retry after first failure, attempts={total}" + ); + } +} diff --git a/crates/tranquil-store/src/sim.rs b/crates/tranquil-store/src/sim.rs index 2fe3d15..a096561 100644 --- a/crates/tranquil-store/src/sim.rs +++ b/crates/tranquil-store/src/sim.rs @@ -692,12 +692,18 @@ impl StorageIO for SimulatedIO { return Err(io::Error::other("simulated EIO on sync")); } - let sync_succeeded = !state.should_fault(seed, fault.sync_failure_probability); - let poison_after = sync_succeeded - && state.should_fault(seed, fault.delayed_io_error_probability); + if state.should_fault(seed, fault.sync_failure_probability) { + state.op_log.push(OpRecord::Sync { + fd: id, + succeeded: false, + }); + return Err(io::Error::other("simulated dropped fsync")); + } + + let poison_after = state.should_fault(seed, fault.delayed_io_error_probability); let reorder_window = fault.sync_reorder_window.0 as usize; - let evicted = if sync_succeeded && reorder_window > 0 { + let evicted = if reorder_window > 0 { let snapshot = state.storage.get(&sid).unwrap().buffered.clone(); state.pending_syncs.push_back(PendingSync { storage_id: sid, @@ -723,7 +729,7 @@ impl StorageIO for SimulatedIO { let storage = state.storage.get_mut(&sid).unwrap(); - if sync_succeeded && reorder_window == 0 { + if reorder_window == 0 { storage.durable = storage.buffered.clone(); } if poison_after { @@ -732,7 +738,7 @@ impl StorageIO for SimulatedIO { state.op_log.push(OpRecord::Sync { fd: id, - succeeded: sync_succeeded, + succeeded: true, }); Ok(()) } diff --git a/crates/tranquil-store/tests/gauntlet_smoke.rs b/crates/tranquil-store/tests/gauntlet_smoke.rs index 1810368..aa0a49d 100644 --- a/crates/tranquil-store/tests/gauntlet_smoke.rs +++ b/crates/tranquil-store/tests/gauntlet_smoke.rs @@ -460,3 +460,115 @@ async fn mst_restart_churn_single_seed() { assert_clean(&report); assert!(report.restarts.0 >= 1); } + +#[tokio::test] +async fn torn_pages_only_completes_within_budget() { + let cfg = GauntletConfig { + seed: Seed(0), + io: IoBackend::Simulated { + fault: FaultConfig::torn_pages_only(), + }, + workload: WorkloadModel { + weights: OpWeights { + add: 80, + delete: 10, + compact: 5, + checkpoint: 5, + ..OpWeights::default() + }, + size_distribution: SizeDistribution::Fixed(ValueBytes(128)), + collections: vec![ + CollectionName("app.bsky.feed.post".to_string()), + CollectionName("app.bsky.feed.like".to_string()), + ], + key_space: KeySpaceSize(500), + did_space: DidSpaceSize(32), + retention_max_secs: RetentionMaxSecs(3600), + }, + op_count: OpCount(2_000), + invariants: InvariantSet::REFCOUNT_CONSERVATION + | InvariantSet::REACHABILITY + | InvariantSet::ACKED_WRITE_PERSISTENCE + | InvariantSet::READ_AFTER_WRITE + | InvariantSet::RESTART_IDEMPOTENT, + limits: RunLimits { + max_wall_ms: Some(WallMs(60_000)), + }, + restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(500)), + store: StoreConfig { + max_file_size: MaxFileSize(16 * 1024), + group_commit: GroupCommitConfig { + verify_persisted_blocks: true, + ..GroupCommitConfig::default() + }, + shard_count: ShardCount(1), + }, + eventlog: None, + writer_concurrency: WriterConcurrency(1), + }; + let report = Gauntlet::new(cfg).expect("build gauntlet").run().await; + let budget_violations: Vec<&str> = report + .violations + .iter() + .filter(|v| v.invariant == "WallClockBudget") + .map(|v| v.detail.as_str()) + .collect(); + assert!( + budget_violations.is_empty(), + "torn-pages exceeded budget: {budget_violations:?}; ops_executed={}", + report.ops_executed.0 + ); + assert_eq!( + report.ops_executed.0, 2_000, + "expected all ops to execute under torn-pages-only faults" + ); +} + +#[tokio::test] +async fn real_io_gauntlet_uses_scratch_root_for_tempdir() { + let scratch = tempfile::TempDir::new().expect("scratch dir"); + let scratch_path = scratch.path().to_path_buf(); + let cfg = fast_sanity_config(Seed(11)); + let report = Gauntlet::new(cfg) + .expect("build gauntlet") + .with_scratch_root(scratch_path.clone()) + .run() + .await; + assert_clean(&report); + let entries: Vec = std::fs::read_dir(&scratch_path) + .expect("read scratch") + .filter_map(|e| e.ok().map(|e| e.path())) + .collect(); + assert!( + entries.is_empty(), + "scratch root must be empty after gauntlet drop, found: {entries:?}" + ); +} + +#[test] +fn farm_run_many_timed_with_scratch_roots_honors_assignment() { + let scratch = tempfile::TempDir::new().expect("scratch dir"); + let root_a = scratch.path().join("a"); + let root_b = scratch.path().join("b"); + std::fs::create_dir_all(&root_a).expect("mkdir a"); + std::fs::create_dir_all(&root_b).expect("mkdir b"); + let roots = vec![root_a.clone(), root_b.clone()]; + let reports = farm::run_many_timed_with_scratch_roots( + |seed| fast_sanity_config(seed), + &roots, + (0..2).map(Seed), + ); + assert_eq!(reports.len(), 2); + reports.iter().for_each(|(r, _)| assert_clean(r)); + [&root_a, &root_b].iter().for_each(|root| { + let leftover: Vec = std::fs::read_dir(root) + .expect("read scratch root") + .filter_map(|e| e.ok().map(|e| e.path())) + .collect(); + assert!( + leftover.is_empty(), + "scratch root {} must be empty after farm completes, found: {leftover:?}", + root.display() + ); + }); +} diff --git a/crates/tranquil-store/tests/sim_eventlog.rs b/crates/tranquil-store/tests/sim_eventlog.rs index 7241b93..c50320f 100644 --- a/crates/tranquil-store/tests/sim_eventlog.rs +++ b/crates/tranquil-store/tests/sim_eventlog.rs @@ -1025,7 +1025,10 @@ fn aggressive_faults_group_sync_recovery() { #[test] fn sync_synced_seq_must_match_durable_valid_prefix() { - sim_seed_range().into_par_iter().for_each(|seed| { + let asserted = std::sync::atomic::AtomicU64::new(0); + let range = sim_seed_range(); + let total = range.end - range.start; + range.into_par_iter().for_each(|seed| { let fault_config = FaultConfig { partial_write_probability: Probability::new(0.05), torn_page_probability: Probability::new(0.01), @@ -1037,8 +1040,9 @@ fn sync_synced_seq_must_match_durable_valid_prefix() { let sim = SimulatedIO::new(seed, fault_config); let mgr = setup_manager(sim, 64 * 1024); - let mut writer = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD) - .unwrap_or_else(|e| panic!("seed {seed}: open writer failed: {e}")); + let Ok(mut writer) = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD) else { + return; + }; let event_count = 10u64; (1..=event_count).for_each(|i| { @@ -1067,13 +1071,21 @@ fn sync_synced_seq_must_match_durable_valid_prefix() { let durable_max = valid.last().map(|e| e.seq.raw()).unwrap_or(0); + asserted.fetch_add(1, std::sync::atomic::Ordering::Relaxed); assert!( synced_through <= durable_max, - "seed {seed}: sync acked seq {synced_through} but durable valid prefix only reaches {durable_max} \ - (events written: {event_count}, valid_prefix.len()={})", + "seed {seed}: sync acked seq {synced_through} but durable valid prefix only reaches {durable_max}, events written: {event_count}, valid_prefix.len()={}", valid.len() ); }); + + let asserted = asserted.load(std::sync::atomic::Ordering::Relaxed); + if total >= 50 { + assert!( + asserted * 2 >= total, + "fewer than half of {total} seeds reached the durability assertion: {asserted}" + ); + } } #[test]