From 31ee12ecd39a60fad679b67bdb853bd8f1ee142a Mon Sep 17 00:00:00 2001 From: Lewis Date: Sun, 31 May 2026 00:22:21 +0300 Subject: [PATCH] fix(store): torn hint-file tail should be recoverable on reopen Lewis: May this revision serve well! --- crates/tranquil-store/Cargo.toml | 4 +- crates/tranquil-store/benches/blockstore.rs | 4 + .../src/bin/tranquil_gauntlet.rs | 107 ++++++++++++------ .../src/blockstore/group_commit.rs | 9 +- crates/tranquil-store/src/blockstore/hint.rs | 25 +++- crates/tranquil-store/src/blockstore/store.rs | 80 +++++++++++++ .../tranquil-store/src/gauntlet/overrides.rs | 28 ++++- crates/tranquil-store/src/gauntlet/runner.rs | 34 ++++-- crates/tranquil-store/src/sim.rs | 12 +- 9 files changed, 245 insertions(+), 58 deletions(-) diff --git a/crates/tranquil-store/Cargo.toml b/crates/tranquil-store/Cargo.toml index 9e9c6ec..cfbc81c 100644 --- a/crates/tranquil-store/Cargo.toml +++ b/crates/tranquil-store/Cargo.toml @@ -38,11 +38,13 @@ tempfile = { version = "3", optional = true } clap = { workspace = true, optional = true } toml = { version = "0.8", optional = true } tracing-subscriber = { workspace = true, features = ["env-filter"], optional = true } +tikv-jemallocator = { version = "0.6", optional = true } libc = "0.2" [features] test-harness = ["dep:tempfile"] -gauntlet-cli = ["test-harness", "dep:clap", "dep:toml", "dep:tracing-subscriber"] +jemalloc = ["dep:tikv-jemallocator"] +gauntlet-cli = ["test-harness", "dep:clap", "dep:toml", "dep:tracing-subscriber", "jemalloc"] gauntlet-jemalloc-prof = [] [[bin]] diff --git a/crates/tranquil-store/benches/blockstore.rs b/crates/tranquil-store/benches/blockstore.rs index 4a13ec5..75aa7e3 100644 --- a/crates/tranquil-store/benches/blockstore.rs +++ b/crates/tranquil-store/benches/blockstore.rs @@ -15,6 +15,10 @@ use tranquil_store::blockstore::{ }; use tranquil_store::{RealIO, SystemClock}; +#[cfg(feature = "jemalloc")] +#[global_allocator] +static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + const DAG_CBOR_CODEC: u64 = 0x71; const SHA2_256_CODE: u64 = 0x12; diff --git a/crates/tranquil-store/src/bin/tranquil_gauntlet.rs b/crates/tranquil-store/src/bin/tranquil_gauntlet.rs index 055153c..0e52cf2 100644 --- a/crates/tranquil-store/src/bin/tranquil_gauntlet.rs +++ b/crates/tranquil-store/src/bin/tranquil_gauntlet.rs @@ -14,6 +14,10 @@ use tranquil_store::gauntlet::{ shrink::{DEFAULT_MAX_SHRINK_ITERATIONS, shrink_failure}, }; +#[cfg(feature = "jemalloc")] +#[global_allocator] +static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + const MAX_HOURS: f64 = 1.0e6; const DEFAULT_SWEEP_RUN_CAP: u64 = 10_000; @@ -223,6 +227,10 @@ struct SweepAxes { commit_batch_size: Vec, #[serde(default)] max_file_size: Vec, + #[serde(default)] + advance_time: Vec, + #[serde(default)] + advance_max_secs: Vec, } #[derive(Debug, Clone, Copy, Default)] @@ -235,6 +243,8 @@ struct SweepAxisValues { restart_every_n_ops: Option, commit_batch_size: Option, max_file_size: Option, + advance_time: Option, + advance_max_secs: Option, } impl SweepAxisValues { @@ -263,51 +273,59 @@ impl SweepAxisValues { if let Some(v) = self.max_file_size { o.store.max_file_size = Some(v); } + if let Some(v) = self.advance_time { + o.advance_time = Some(v); + } + if let Some(v) = self.advance_max_secs { + o.advance_max_secs = Some(v); + } } } impl SweepAxes { fn axis_values(&self) -> Vec { - expand(&self.writer_concurrency) - .into_iter() - .flat_map(|wc| { - expand(&self.key_space).into_iter().flat_map(move |ks| { - expand(&self.value_bytes).into_iter().flat_map(move |vb| { - expand(&self.fault_density_scale) - .into_iter() - .flat_map(move |fds| { - expand(&self.fault_density_uniform).into_iter().flat_map( - move |fdu| { - expand(&self.restart_every_n_ops).into_iter().flat_map( - move |rc| { - expand(&self.commit_batch_size) - .into_iter() - .flat_map(move |cb| { - expand(&self.max_file_size).into_iter().map( - move |mfs| SweepAxisValues { - writer_concurrency: wc, - key_space: ks, - value_bytes: vb, - fault_density_scale: fds, - fault_density_uniform: fdu, - restart_every_n_ops: rc, - commit_batch_size: cb, - max_file_size: mfs, - }, - ) - }) - }, - ) - }, - ) - }) - }) - }) - }) - .collect() + let base = vec![SweepAxisValues::default()]; + let base = cross(base, &self.writer_concurrency, |a, v| a.writer_concurrency = Some(v)); + let base = cross(base, &self.key_space, |a, v| a.key_space = Some(v)); + let base = cross(base, &self.value_bytes, |a, v| a.value_bytes = Some(v)); + let base = cross(base, &self.fault_density_scale, |a, v| { + a.fault_density_scale = Some(v) + }); + let base = cross(base, &self.fault_density_uniform, |a, v| { + a.fault_density_uniform = Some(v) + }); + let base = cross(base, &self.restart_every_n_ops, |a, v| { + a.restart_every_n_ops = Some(v) + }); + let base = cross(base, &self.commit_batch_size, |a, v| { + a.commit_batch_size = Some(v) + }); + let base = cross(base, &self.max_file_size, |a, v| a.max_file_size = Some(v)); + let base = cross(base, &self.advance_time, |a, v| a.advance_time = Some(v)); + cross(base, &self.advance_max_secs, |a, v| { + a.advance_max_secs = Some(v) + }) } } +fn cross( + acc: Vec, + values: &[T], + set: impl Fn(&mut SweepAxisValues, T) + Copy, +) -> Vec { + acc.into_iter() + .flat_map(|base| { + expand(values).into_iter().map(move |opt| { + let mut next = base; + if let Some(v) = opt { + set(&mut next, v); + } + next + }) + }) + .collect() +} + fn expand(values: &[T]) -> Vec> { if values.is_empty() { vec![None] @@ -613,7 +631,22 @@ fn install_interrupt(rt: &Runtime) -> Arc { flag } +fn raise_fd_limit() { + let mut lim = libc::rlimit { + rlim_cur: 0, + rlim_max: 0, + }; + let read = unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut lim) } == 0; + if read && lim.rlim_cur < lim.rlim_max { + lim.rlim_cur = lim.rlim_max; + unsafe { + let _ = libc::setrlimit(libc::RLIMIT_NOFILE, &lim); + } + } +} + fn main() -> ExitCode { + raise_fd_limit(); let _ = tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() diff --git a/crates/tranquil-store/src/blockstore/group_commit.rs b/crates/tranquil-store/src/blockstore/group_commit.rs index 2d7759a..69ea0c6 100644 --- a/crates/tranquil-store/src/blockstore/group_commit.rs +++ b/crates/tranquil-store/src/blockstore/group_commit.rs @@ -14,7 +14,7 @@ use crate::io::{FileId, OpenOptions, StorageIO}; use super::data_file::{CID_SIZE, DataFileWriter, ReadBlockRecord, decode_block_record}; use super::hash_index::{BlockIndex, BlockIndexError, CheckpointPositions}; -use super::hint::{HintFileWriter, hint_file_path}; +use super::hint::{HINT_RECORD_SIZE, HintFileWriter, hint_file_path}; use super::manager::DataFileManager; use super::types::{ BlockLocation, BlockOffset, BlockstoreSnapshot, CommitEpoch, DataFileId, EpochCounter, @@ -605,13 +605,18 @@ fn initialize_active_state( let hint_path = hint_file_path(data_dir, wc.file_id); let hint_fd = manager.io().open(&hint_path, OpenOptions::read_write())?; let hint_size = manager.io().file_size(hint_fd)?; + let aligned_hint = hint_size - hint_size % HINT_RECORD_SIZE as u64; + if aligned_hint != hint_size { + manager.io().truncate(hint_fd, aligned_hint)?; + manager.io().sync(hint_fd)?; + } Ok(ActiveState { file_id: wc.file_id, fd, position, hint_fd, - hint_position: HintOffset::new(hint_size), + hint_position: HintOffset::new(aligned_hint), }) } None => { diff --git a/crates/tranquil-store/src/blockstore/hint.rs b/crates/tranquil-store/src/blockstore/hint.rs index 3ecdb29..23ef90c 100644 --- a/crates/tranquil-store/src/blockstore/hint.rs +++ b/crates/tranquil-store/src/blockstore/hint.rs @@ -469,12 +469,16 @@ impl<'a, S: StorageIO> HintFileReader<'a, S> { } pub fn resume(io: &'a S, fd: FileId, position: HintOffset) -> io::Result { - assert!( - position.raw().is_multiple_of(HINT_RECORD_SIZE as u64), - "hint resume position {} not aligned to HINT_RECORD_SIZE {}", - position.raw(), - HINT_RECORD_SIZE, - ); + if !position.raw().is_multiple_of(HINT_RECORD_SIZE as u64) { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "hint resume position {} not aligned to HINT_RECORD_SIZE {}", + position.raw(), + HINT_RECORD_SIZE, + ), + )); + } let file_size = io.file_size(fd)?; Ok(Self { io, @@ -1200,6 +1204,15 @@ mod tests { assert_eq!(valid_count, 2); } + #[test] + fn hint_reader_resume_rejects_unaligned_position_without_panicking() { + let (sim, fd) = setup(); + match HintFileReader::resume(&sim, fd, HintOffset::new(HINT_RECORD_SIZE as u64 + 5)) { + Err(e) => assert_eq!(e.kind(), io::ErrorKind::InvalidData), + Ok(_) => panic!("non-aligned resume position must surface as a recoverable error"), + } + } + #[test] fn hint_reader_empty_file() { let (sim, fd) = setup(); diff --git a/crates/tranquil-store/src/blockstore/store.rs b/crates/tranquil-store/src/blockstore/store.rs index c4bc7f4..a981cd8 100644 --- a/crates/tranquil-store/src/blockstore/store.rs +++ b/crates/tranquil-store/src/blockstore/store.rs @@ -954,6 +954,86 @@ mod tests { ); } + #[test] + fn reopen_recovers_from_torn_hint_tail_without_aborting() { + use crate::blockstore::HINT_RECORD_SIZE; + use std::io::Write; + + let tmp = tempfile::TempDir::new().unwrap(); + let data_dir = tmp.path().join("data"); + let index_dir = tmp.path().join("index"); + let cfg = || BlockStoreConfig::new(data_dir.clone(), index_dir.clone()); + + let payload_a = b"alpha-block-payload".to_vec(); + let payload_b = b"bravo-block-payload".to_vec(); + let cid_a = crate::blockstore::hash_to_cid_bytes(&payload_a); + let cid_b = crate::blockstore::hash_to_cid_bytes(&payload_b); + + { + let store = TranquilBlockStore::open(cfg()).unwrap(); + store + .apply_commit_blocking(vec![(cid_a, payload_a.clone())], vec![]) + .unwrap(); + store.apply_commit_blocking(vec![], vec![]).unwrap(); + } + + let hint_path = std::fs::read_dir(&data_dir) + .unwrap() + .filter_map(|e| e.ok().map(|e| e.path())) + .find(|p| p.extension().and_then(|s| s.to_str()) == Some("tqh")) + .expect("active hint file must exist after commit"); + + let clean = std::fs::metadata(&hint_path).unwrap().len(); + assert_eq!( + clean % HINT_RECORD_SIZE as u64, + 0, + "precondition: synced hint file must be record-aligned, got {clean}" + ); + { + let mut f = std::fs::OpenOptions::new() + .append(true) + .open(&hint_path) + .unwrap(); + f.write_all(&[0x5Au8; 50]).unwrap(); + f.sync_all().unwrap(); + } + assert_eq!( + std::fs::metadata(&hint_path).unwrap().len() % HINT_RECORD_SIZE as u64, + 50, + "torn tail must leave a non-aligned hint file" + ); + + { + let store = TranquilBlockStore::open(cfg()).unwrap(); + assert!( + store.get_block_sync(&cid_a).unwrap().is_some(), + "block A lost after torn-tail recovery" + ); + store + .apply_commit_blocking(vec![(cid_b, payload_b.clone())], vec![]) + .unwrap(); + } + + let healed = std::fs::metadata(&hint_path).unwrap().len(); + assert_eq!( + healed % HINT_RECORD_SIZE as u64, + 0, + "recovery must realign the hint file, got {healed}" + ); + + { + let store = TranquilBlockStore::open(cfg()).unwrap(); + assert!( + store.get_block_sync(&cid_a).unwrap().is_some(), + "block A lost across second reopen" + ); + assert!( + store.get_block_sync(&cid_b).unwrap().is_some(), + "block B lost across second reopen" + ); + } + } + fn instant_policy(max_attempts: u8) -> OpenRetryPolicy { OpenRetryPolicy { max_attempts: NonZeroU8::new(max_attempts).expect("max_attempts must be nonzero"), diff --git a/crates/tranquil-store/src/gauntlet/overrides.rs b/crates/tranquil-store/src/gauntlet/overrides.rs index 3cbdc7d..4658037 100644 --- a/crates/tranquil-store/src/gauntlet/overrides.rs +++ b/crates/tranquil-store/src/gauntlet/overrides.rs @@ -4,7 +4,7 @@ use super::runner::{ GauntletConfig, IoBackend, MaxFileSize, OpInterval, RestartPolicy, RunLimits, ShardCount, WallMs, WriterConcurrency, }; -use super::workload::{KeySpaceSize, OpCount, SizeDistribution, ValueBytes}; +use super::workload::{AdvanceMaxSecs, KeySpaceSize, OpCount, SizeDistribution, ValueBytes}; use crate::sim::FaultConfig; #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] @@ -26,6 +26,10 @@ pub struct ConfigOverrides { pub fault_density_uniform: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub restart_every_n_ops: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub advance_time: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub advance_max_secs: Option, #[serde(default, skip_serializing_if = "StoreOverrides::is_empty")] pub store: StoreOverrides, } @@ -110,6 +114,12 @@ impl ConfigOverrides { RestartPolicy::EveryNOps(OpInterval(n)) }; } + if let Some(n) = self.advance_time { + cfg.workload.weights.advance_time = n; + } + if let Some(n) = self.advance_max_secs { + cfg.workload.advance_max_secs = AdvanceMaxSecs(n.max(1)); + } if let Some(n) = self.store.max_file_size { cfg.store.max_file_size = MaxFileSize(n); } @@ -225,6 +235,22 @@ mod tests { assert!(matches!(cfg.io, IoBackend::Real)); } + #[test] + fn advance_time_overrides_inject_time_travel() { + use crate::gauntlet::op::Seed; + use crate::gauntlet::scenarios::{Scenario, config_for}; + let mut cfg = config_for(Scenario::FirehoseFanout, Seed(1)); + assert_eq!(cfg.workload.weights.advance_time, 0); + let o = ConfigOverrides { + advance_time: Some(40), + advance_max_secs: Some(1_209_600), + ..ConfigOverrides::default() + }; + o.apply_to(&mut cfg); + assert_eq!(cfg.workload.weights.advance_time, 40); + assert_eq!(cfg.workload.advance_max_secs.0, 1_209_600); + } + #[test] fn fault_density_uniform_forces_simulated_backend() { use crate::gauntlet::op::Seed; diff --git a/crates/tranquil-store/src/gauntlet/runner.rs b/crates/tranquil-store/src/gauntlet/runner.rs index fe62c7b..464b2e6 100644 --- a/crates/tranquil-store/src/gauntlet/runner.rs +++ b/crates/tranquil-store/src/gauntlet/runner.rs @@ -397,7 +397,7 @@ async fn run_inner_real_on_root( } }; if config.writer_concurrency.0 > 1 { - run_inner_generic_concurrent::( + run_inner_generic_concurrent::( config, ops, ops_counter, @@ -408,10 +408,11 @@ async fn run_inner_real_on_root( tolerate_op_errors, reopen_backoff, SystemClock, + |_| {}, ) .await } else { - run_inner_generic::( + run_inner_generic::( config, ops, ops_counter, @@ -422,6 +423,7 @@ async fn run_inner_real_on_root( tolerate_op_errors, reopen_backoff, SystemClock, + |_| {}, ) .await } @@ -473,8 +475,10 @@ async fn run_inner_simulated( }; let sim_for_crash = Arc::clone(&sim); let crash = move || sim_for_crash.crash(); + let sim_for_quiesce = Arc::clone(&sim); + let quiesce = move |on: bool| sim_for_quiesce.set_pristine_mode(on); if config.writer_concurrency.0 > 1 { - run_inner_generic_concurrent::, SimClock, _, _>( + run_inner_generic_concurrent::, SimClock, _, _, _>( config, ops, ops_counter, @@ -485,10 +489,11 @@ async fn run_inner_simulated( tolerate_errors, Duration::ZERO, clock, + quiesce, ) .await } else { - run_inner_generic::, SimClock, _, _>( + run_inner_generic::, SimClock, _, _, _>( config, ops, ops_counter, @@ -499,6 +504,7 @@ async fn run_inner_simulated( tolerate_errors, Duration::ZERO, clock, + quiesce, ) .await } @@ -528,7 +534,7 @@ pub(super) fn open_eventlog( } #[allow(clippy::too_many_arguments)] -async fn run_inner_generic( +async fn run_inner_generic( config: GauntletConfig, op_stream: OpStream, ops_counter: Arc, @@ -539,12 +545,14 @@ async fn run_inner_generic( tolerate_op_errors: bool, reopen_backoff: Duration, clock: C, + quiesce_faults: Quiesce, ) -> GauntletReport where S: StorageIO + Send + Sync + 'static, C: Clock, Open: FnMut(usize) -> Result, String>, Crash: FnMut(), + Quiesce: Fn(bool), { let mut oracle = Oracle::new(); let mut violations: Vec = Vec::new(); @@ -617,6 +625,7 @@ where let n = restarts_counter.fetch_add(1, Ordering::Relaxed) + 1; let live = harness.as_ref().expect("just reopened"); let before = violations.len(); + quiesce_faults(true); violations.extend( run_quick_check( &live.store, @@ -628,6 +637,7 @@ where ) .await, ); + quiesce_faults(false); if violations.len() > before { halt_ops = true; } @@ -643,6 +653,7 @@ where } } + quiesce_faults(true); if !halt_ops && tolerate_op_errors && harness.is_some() { crash(); oracle.record_crash(); @@ -1660,7 +1671,7 @@ fn compute_chunks( } #[allow(clippy::too_many_arguments)] -async fn run_inner_generic_concurrent( +async fn run_inner_generic_concurrent( config: GauntletConfig, op_stream: OpStream, ops_counter: Arc, @@ -1671,12 +1682,14 @@ async fn run_inner_generic_concurrent( tolerate_op_errors: bool, reopen_backoff: Duration, clock: C, + quiesce_faults: Quiesce, ) -> GauntletReport where S: StorageIO + Send + Sync + 'static, C: Clock, Open: FnMut(usize) -> Result, String>, Crash: FnMut(), + Quiesce: Fn(bool), { let ops: Vec = op_stream.into_vec(); let total_ops = ops.len(); @@ -1806,6 +1819,7 @@ where let n = restarts_counter.fetch_add(1, Ordering::Relaxed) + 1; let live = harness.as_ref().expect("just reopened"); let before = violations.len(); + quiesce_faults(true); violations.extend( run_quick_check( &live.store, @@ -1817,6 +1831,7 @@ where ) .await, ); + quiesce_faults(false); if violations.len() > before { halt_ops = true; } @@ -1833,6 +1848,7 @@ where } } + quiesce_faults(true); if !halt_ops && tolerate_op_errors && harness.is_some() { crash(); oracle.record_crash(); @@ -1978,7 +1994,7 @@ mod tests { let clock = sim.clock(); let attempts = Arc::new(AtomicUsize::new(0)); - let report = run_inner_generic::, SimClock, _, _>( + let report = run_inner_generic::, SimClock, _, _, _>( cfg, OpStream::empty(), Arc::new(AtomicUsize::new(0)), @@ -1994,6 +2010,7 @@ mod tests { true, Duration::ZERO, clock, + |_| {}, ) .await; @@ -2023,7 +2040,7 @@ mod tests { let clock = sim.clock(); let attempts = Arc::new(AtomicUsize::new(0)); - let report = run_inner_generic_concurrent::, SimClock, _, _>( + let report = run_inner_generic_concurrent::, SimClock, _, _, _>( cfg, OpStream::empty(), Arc::new(AtomicUsize::new(0)), @@ -2039,6 +2056,7 @@ mod tests { true, Duration::ZERO, clock, + |_| {}, ) .await; diff --git a/crates/tranquil-store/src/sim.rs b/crates/tranquil-store/src/sim.rs index 8c894f1..015469e 100644 --- a/crates/tranquil-store/src/sim.rs +++ b/crates/tranquil-store/src/sim.rs @@ -376,6 +376,10 @@ impl SimulatedIO { self.pristine_mode.store(on, Ordering::Relaxed); } + pub fn pristine_mode(&self) -> bool { + self.pristine_mode.load(Ordering::Relaxed) + } + fn jitter(&self) { let max_ns = self.effective_fault_config().latency_distribution_ns.0; let extra_ns = match max_ns { @@ -458,18 +462,20 @@ impl SimulatedIO { pub struct PristineGuard { sim: Arc, + prev: bool, } impl PristineGuard { pub fn new(sim: Arc, on: bool) -> Self { - sim.set_pristine_mode(on); - Self { sim } + let prev = sim.pristine_mode(); + sim.set_pristine_mode(on || prev); + Self { sim, prev } } } impl Drop for PristineGuard { fn drop(&mut self) { - self.sim.set_pristine_mode(false); + self.sim.set_pristine_mode(self.prev); } }