mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-07-27 13:22:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(tranquil-store): durable-tail recovery + sync semantics

Browse Source 
Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-04-30 11:11:28 +03:00
parent efd499bb26
commit 9b2cfb3a7e
10 changed files with 841 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tranquil-pds/src/state.rs
+10 -7
View File
@@ -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(
crates/tranquil-store/src/bin/tranquil_gauntlet.rs
+81 -2
View File
@@ -64,6 +64,16 @@ enum Cmd {
#[arg(long)]
config: Option<PathBuf>,
/// 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<PathBuf>,
/// Skip shrinking when dumping regressions.
#[arg(long)]
no_shrink: bool,
@@ -95,6 +105,13 @@ enum Cmd {
#[arg(long)]
dump_regressions: Option<PathBuf>,
/// 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<PathBuf>,
/// Skip shrinking when dumping regressions.
#[arg(long)]
no_shrink: bool,
@@ -159,6 +176,8 @@ struct ConfigFile {
#[serde(default)]
dump_regressions: Option<PathBuf>,
#[serde(default)]
scratch_roots: Vec<PathBuf>,
#[serde(default)]
overrides: ConfigOverrides,
}
@@ -178,6 +197,8 @@ struct SweepConfigFile {
#[serde(default)]
dump_regressions: Option<PathBuf>,
#[serde(default)]
scratch_roots: Vec<PathBuf>,
#[serde(default)]
base_overrides: ConfigOverrides,
#[serde(default)]
axes: SweepAxes,
@@ -405,6 +426,7 @@ struct FarmPlan {
seeds: u64,
hours: Option<f64>,
dump_regressions: Option<PathBuf>,
scratch_roots: Vec<PathBuf>,
overrides: ConfigOverrides,
shrink: bool,
shrink_budget: usize,
@@ -418,6 +440,7 @@ fn resolve_farm(
hours: Option<f64>,
dump_regressions: Option<PathBuf>,
config: Option<PathBuf>,
scratch_root: Vec<PathBuf>,
shrink: bool,
shrink_budget: usize,
) -> Result<FarmPlan, String> {
@@ -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<PathBuf>,
config_file: Vec<PathBuf>,
) -> Result<Vec<PathBuf>, String> {
let env_roots: Vec<PathBuf> = 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<PathBuf> = 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<PathBuf>,
scratch_roots: Vec<PathBuf>,
shrink: bool,
shrink_budget: usize,
base_overrides: ConfigOverrides,
axes: Vec<SweepAxisValues>,
}
#[allow(clippy::too_many_arguments)]
fn resolve_sweep(
config: PathBuf,
seed_start: Option<u64>,
seeds: Option<u64>,
dump_regressions: Option<PathBuf>,
scratch_root: Vec<PathBuf>,
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<AtomicBool>) -> 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<AtomicBool>) -> 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<AtomicBool>) -> ExitCod
seeds,
hours,
dump_regressions,
scratch_roots,
overrides,
shrink,
shrink_budget,
@@ -871,12 +949,13 @@ fn run_farm(plan: FarmPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> 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();
crates/tranquil-store/src/blockstore/mod.rs
+1 -1
View File
@@ -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,
crates/tranquil-store/src/blockstore/store.rs
+291 -15
View File
@@ -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<RealIO> {
pub fn open(config: BlockStoreConfig) -> Result<Self, RepoError> {
Self::open_with_hook(config, None)
@@ -161,6 +181,50 @@ impl TranquilBlockStore<RealIO> {
) -> Result<Self, RepoError> {
Self::open_with_io_hook(config, RealIO::new, post_sync_hook)
}
pub fn open_with_retry(
config: BlockStoreConfig,
policy: OpenRetryPolicy,
) -> Result<Self, RepoError> {
retry_with_backoff(policy, &mut |_| Self::open(config.clone()))
}
}
fn retry_with_backoff<T, F>(policy: OpenRetryPolicy, op: &mut F) -> Result<T, RepoError>
where
F: FnMut(u8) -> Result<T, RepoError>,
{
retry_attempt(policy, op, 0, policy.initial_backoff)
}
fn retry_attempt<T, F>(
policy: OpenRetryPolicy,
op: &mut F,
attempt: u8,
backoff: Duration,
) -> Result<T, RepoError>
where
F: FnMut(u8) -> Result<T, RepoError>,
{
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<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
@@ -331,15 +395,7 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
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<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
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<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
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<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
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<HashMap<FileId, PathBuf>>,
}
impl StorageIO for EioOnReadAtRange {
fn open(&self, path: &Path, opts: OpenOptions) -> io::Result<FileId> {
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<usize> {
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<usize> {
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<u64> {
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<Vec<PathBuf>> {
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::<EioOnReadAtRange>::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::<u8, RepoError>(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::<u8, RepoError>(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<u8, RepoError> = 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::<u8>::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]);
}
}
crates/tranquil-store/src/eventlog/writer.rs
+58
View File
@@ -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}"
);
}
}
crates/tranquil-store/src/gauntlet/farm.rs
+84 -1
View File
@@ -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<F>(
make_config: F,
seeds: impl IntoIterator<Item = Seed>,
) -> 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<F>(
make_config: F,
scratch_roots: &[PathBuf],
seeds: impl IntoIterator<Item = Seed>,
) -> 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<usize>) -> Option<PathBuf> {
if roots.is_empty() {
None
} else {
Some(roots[thread_idx.unwrap_or(0) % roots.len()].clone())
}
}
fn panic_report(seed: Seed, payload: Box<dyn std::any::Any + Send>) -> GauntletReport {
let msg = payload
.downcast_ref::<&'static str>()
@@ -84,3 +108,62 @@ fn panic_report(seed: Seed, payload: Box<dyn std::any::Any + Send>) -> 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<PathBuf> = (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"))
);
}
}
crates/tranquil-store/src/gauntlet/runner.rs
+175 -34
View File
@@ -178,6 +178,7 @@ pub struct SharedState<S: StorageIO + Send + Sync + 'static> {
pub struct Gauntlet {
config: GauntletConfig,
scratch_root: Option<PathBuf>,
}
#[derive(Debug, thiserror::Error)]
@@ -185,7 +186,15 @@ pub enum GauntletBuildError {}
impl Gauntlet {
pub fn new(config: GauntletConfig) -> Result<Self, GauntletBuildError> {
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<Box<dyn std::future::Future<Output = GauntletReport> + 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<AtomicUsize>,
op_errors_counter: Arc<AtomicUsize>,
restarts_counter: Arc<AtomicUsize>,
scratch_root: Option<PathBuf>,
) -> 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<InvariantViolation> = Vec::new();
let mut harness: Option<Harness<S>> = 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<Harness<S>> =
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<Cid> = 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<Harness<S>> = 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<Harness<S>> =
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<Cid> = 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<AtomicUsize>,
sim: Arc<SimulatedIO>,
store_cfg: BlockStoreConfig,
) -> impl FnMut(usize) -> Result<Harness<Arc<SimulatedIO>>, String> + Send + 'static {
move |_attempt: usize| -> Result<Harness<Arc<SimulatedIO>>, 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::<Arc<SimulatedIO>>::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<SimulatedIO> = Arc::new(SimulatedIO::pristine(0));
let attempts = Arc::new(AtomicUsize::new(0));
let report = run_inner_generic::<Arc<SimulatedIO>, _, _>(
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<SimulatedIO> = Arc::new(SimulatedIO::pristine(0));
let attempts = Arc::new(AtomicUsize::new(0));
let report = run_inner_generic_concurrent::<Arc<SimulatedIO>, _, _>(
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}"
);
}
}
crates/tranquil-store/src/sim.rs
+12 -6
View File
@@ -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(())
}
crates/tranquil-store/tests/gauntlet_smoke.rs
+112
View File
@@ -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::path::PathBuf> = 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::path::PathBuf> = 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()
);
});
}
crates/tranquil-store/tests/sim_eventlog.rs
+17 -5
View File
@@ -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]