mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-07-20 06:52:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(store): torn hint-file tail should be recoverable on reopen

Browse Source 
Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-05-31 00:22:21 +03:00
parent ea106d5246
commit 31ee12ecd3
9 changed files with 245 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tranquil-store/Cargo.toml
+3 -1
View File
@@ -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]]
crates/tranquil-store/benches/blockstore.rs
+4
View File
@@ -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;
crates/tranquil-store/src/bin/tranquil_gauntlet.rs
+70 -37
View File
@@ -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<usize>,
#[serde(default)]
max_file_size: Vec<u64>,
#[serde(default)]
advance_time: Vec<u32>,
#[serde(default)]
advance_max_secs: Vec<u32>,
}
#[derive(Debug, Clone, Copy, Default)]
@@ -235,6 +243,8 @@ struct SweepAxisValues {
restart_every_n_ops: Option<usize>,
commit_batch_size: Option<usize>,
max_file_size: Option<u64>,
advance_time: Option<u32>,
advance_max_secs: Option<u32>,
}
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<SweepAxisValues> {
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<T: Copy>(
acc: Vec<SweepAxisValues>,
values: &[T],
set: impl Fn(&mut SweepAxisValues, T) + Copy,
) -> Vec<SweepAxisValues> {
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<T: Copy>(values: &[T]) -> Vec<Option<T>> {
if values.is_empty() {
vec![None]
@@ -613,7 +631,22 @@ fn install_interrupt(rt: &Runtime) -> Arc<AtomicBool> {
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()
crates/tranquil-store/src/blockstore/group_commit.rs
+7 -2
View File
@@ -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<S: StorageIO>(
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 => {
crates/tranquil-store/src/blockstore/hint.rs
+19 -6
View File
@@ -469,12 +469,16 @@ impl<'a, S: StorageIO> HintFileReader<'a, S> {
}
pub fn resume(io: &'a S, fd: FileId, position: HintOffset) -> io::Result<Self> {
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();
crates/tranquil-store/src/blockstore/store.rs
+80
View File
@@ -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"),
crates/tranquil-store/src/gauntlet/overrides.rs
+27 -1
View File
@@ -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<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub restart_every_n_ops: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub advance_time: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub advance_max_secs: Option<u32>,
#[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;
crates/tranquil-store/src/gauntlet/runner.rs
+26 -8
View File
@@ -397,7 +397,7 @@ async fn run_inner_real_on_root(
}
};
if config.writer_concurrency.0 > 1 {
run_inner_generic_concurrent::<RealIO, SystemClock, _, _>(
run_inner_generic_concurrent::<RealIO, SystemClock, _, _, _>(
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::<RealIO, SystemClock, _, _>(
run_inner_generic::<RealIO, SystemClock, _, _, _>(
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::<Arc<SimulatedIO>, SimClock, _, _>(
run_inner_generic_concurrent::<Arc<SimulatedIO>, 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::<Arc<SimulatedIO>, SimClock, _, _>(
run_inner_generic::<Arc<SimulatedIO>, 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<S: StorageIO + Send + Sync + 'static>(
}
#[allow(clippy::too_many_arguments)]
async fn run_inner_generic<S, C, Open, Crash>(
async fn run_inner_generic<S, C, Open, Crash, Quiesce>(
config: GauntletConfig,
op_stream: OpStream,
ops_counter: Arc<AtomicUsize>,
@@ -539,12 +545,14 @@ async fn run_inner_generic<S, C, Open, Crash>(
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<Harness<S, C>, String>,
Crash: FnMut(),
Quiesce: Fn(bool),
{
let mut oracle = Oracle::new();
let mut violations: Vec<InvariantViolation> = 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<S, C, Open, Crash>(
async fn run_inner_generic_concurrent<S, C, Open, Crash, Quiesce>(
config: GauntletConfig,
op_stream: OpStream,
ops_counter: Arc<AtomicUsize>,
@@ -1671,12 +1682,14 @@ async fn run_inner_generic_concurrent<S, C, Open, Crash>(
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<Harness<S, C>, String>,
Crash: FnMut(),
Quiesce: Fn(bool),
{
let ops: Vec<Op> = 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::<Arc<SimulatedIO>, SimClock, _, _>(
let report = run_inner_generic::<Arc<SimulatedIO>, 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::<Arc<SimulatedIO>, SimClock, _, _>(
let report = run_inner_generic_concurrent::<Arc<SimulatedIO>, SimClock, _, _, _>(
cfg,
OpStream::empty(),
Arc::new(AtomicUsize::new(0)),
@@ -2039,6 +2056,7 @@ mod tests {
true,
Duration::ZERO,
clock,
|_| {},
)
.await;
crates/tranquil-store/src/sim.rs
+9 -3
View File
@@ -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<SimulatedIO>,
prev: bool,
}
impl PristineGuard {
pub fn new(sim: Arc<SimulatedIO>, 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);
}
}