use crate::clock::{Clock, SimClock};
use std::cell::RefCell;
use std::fmt;
use std::rc::Rc;
const DEFAULT_BUGGIFY_PPM: u64 = 10_000;
const PPM_DENOMINATOR: u64 = 1_000_000;
pub const SECTOR_BYTES: u64 = 512;
pub const FAULT_ENV: &str = "REDDB_DST_FAULT";
thread_local! {
static ACTIVE: RefCell<Option<Rc<RefCell<ActiveSimulation>>>> = const { RefCell::new(None) };
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum FaultClass {
TornWrite,
MisdirectedWrite,
BitRot,
LostWrite,
}
impl FaultClass {
pub const ALL: [Self; 4] = [
Self::TornWrite,
Self::MisdirectedWrite,
Self::BitRot,
Self::LostWrite,
];
pub const fn name(self) -> &'static str {
match self {
Self::TornWrite => "torn_write",
Self::MisdirectedWrite => "misdirected_write",
Self::BitRot => "bit_rot",
Self::LostWrite => "lost_write",
}
}
pub fn from_name(name: &str) -> Option<Self> {
Self::ALL.into_iter().find(|class| class.name() == name)
}
const fn index(self) -> usize {
match self {
Self::TornWrite => 0,
Self::MisdirectedWrite => 1,
Self::BitRot => 2,
Self::LostWrite => 3,
}
}
}
impl fmt::Display for FaultClass {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.name())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FaultDecision {
TornWrite { persisted: u64 },
MisdirectedWrite { actual_offset: u64 },
BitRot { byte_offset: u64, bit: u8 },
LostWrite,
}
impl FaultDecision {
pub const fn class(self) -> FaultClass {
match self {
Self::TornWrite { .. } => FaultClass::TornWrite,
Self::MisdirectedWrite { .. } => FaultClass::MisdirectedWrite,
Self::BitRot { .. } => FaultClass::BitRot,
Self::LostWrite => FaultClass::LostWrite,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FaultRecord {
pub class: FaultClass,
pub file: String,
pub offset: u64,
pub length: u64,
pub decision: FaultDecision,
}
impl FaultRecord {
pub fn new(file: impl Into<String>, offset: u64, length: u64, decision: FaultDecision) -> Self {
Self {
class: decision.class(),
file: file.into(),
offset,
length,
decision,
}
}
}
impl fmt::Display for FaultRecord {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"class={} file={} offset={} length={}",
self.class, self.file, self.offset, self.length
)?;
match self.decision {
FaultDecision::TornWrite { persisted } => write!(f, " persisted={persisted}"),
FaultDecision::MisdirectedWrite { actual_offset } => {
write!(f, " actual_offset={actual_offset}")
}
FaultDecision::BitRot { byte_offset, bit } => {
write!(f, " byte_offset={byte_offset} bit={bit}")
}
FaultDecision::LostWrite => Ok(()),
}
}
}
#[derive(Debug, Clone, Copy)]
pub struct SimulationContext {
seed: u64,
clock_start_ms: u64,
buggify_ppm: u64,
fault_ppm: [u64; 4],
}
impl SimulationContext {
pub fn new(seed: u64) -> Self {
Self {
seed,
clock_start_ms: seed,
buggify_ppm: DEFAULT_BUGGIFY_PPM,
fault_ppm: [0; 4],
}
}
pub fn with_buggify_ppm(seed: u64, buggify_ppm: u64) -> Self {
Self {
buggify_ppm: buggify_ppm.min(PPM_DENOMINATOR),
..Self::new(seed)
}
}
#[must_use]
pub fn with_fault_class(mut self, class: FaultClass, ppm: u64) -> Self {
self.fault_ppm[class.index()] = ppm.min(PPM_DENOMINATOR);
self
}
pub fn fault_ppm(&self, class: FaultClass) -> u64 {
self.fault_ppm[class.index()]
}
pub fn install(self) -> SimulationGuard {
let active = Rc::new(RefCell::new(ActiveSimulation::new(self)));
let previous = ACTIVE.with(|slot| slot.replace(Some(Rc::clone(&active))));
SimulationGuard { active, previous }
}
}
pub struct SimulationGuard {
active: Rc<RefCell<ActiveSimulation>>,
previous: Option<Rc<RefCell<ActiveSimulation>>>,
}
impl SimulationGuard {
pub fn trace(&self) -> Vec<u8> {
self.active.borrow().trace.clone()
}
pub fn fault_log(&self) -> Vec<FaultRecord> {
self.active.borrow().fault_log.clone()
}
pub fn fault_log_lines(&self) -> String {
self.fault_log()
.iter()
.map(|record| format!("{record}\n"))
.collect()
}
}
impl Drop for SimulationGuard {
fn drop(&mut self) {
let previous = self.previous.take();
ACTIVE.with(|slot| {
slot.replace(previous);
});
}
}
struct ActiveSimulation {
clock: SimClock,
rng: SplitMix64,
buggify_ppm: u64,
fault_ppm: [u64; 4],
tick: u64,
trace: Vec<u8>,
fault_log: Vec<FaultRecord>,
}
impl ActiveSimulation {
fn new(context: SimulationContext) -> Self {
let mut trace = Vec::new();
trace.extend_from_slice(format!("seed={}\n", context.seed).as_bytes());
Self {
clock: SimClock::from_seed(context.clock_start_ms),
rng: SplitMix64::new(context.seed ^ 0x4255_4747_4946_595F), buggify_ppm: context.buggify_ppm,
fault_ppm: context.fault_ppm,
tick: 0,
trace,
fault_log: Vec::new(),
}
}
fn boundary(&mut self, env: &str, point: &str, ppm: u64) -> bool {
self.tick = self.tick.saturating_add(1);
self.clock.advance_ms(1);
let roll = self.rng.below(PPM_DENOMINATOR);
let threshold = ppm.min(PPM_DENOMINATOR);
let fired = roll < threshold;
self.trace.extend_from_slice(
format!(
"tick={} clock_ms={} env={} point={} roll={} ppm={} fired={}\n",
self.tick,
self.clock.now_unix_millis(),
env,
point,
roll,
threshold,
fired
)
.as_bytes(),
);
fired
}
}
pub fn buggify(env: &str, point: &str) -> bool {
ACTIVE.with(|slot| {
let Some(active) = slot.borrow().clone() else {
return false;
};
let ppm = { active.borrow().buggify_ppm };
let fired = active.borrow_mut().boundary(env, point, ppm);
fired
})
}
pub fn buggify_at(env: &str, point: &str, ppm: u64) -> bool {
ACTIVE.with(|slot| {
let Some(active) = slot.borrow().clone() else {
return false;
};
let fired = active.borrow_mut().boundary(env, point, ppm);
fired
})
}
pub fn is_active() -> bool {
ACTIVE.with(|slot| slot.borrow().is_some())
}
pub fn roll_fault(class: FaultClass, offset: u64, length: u64) -> Option<FaultDecision> {
ACTIVE.with(|slot| {
let active = slot.borrow().clone()?;
let mut active = active.borrow_mut();
let ppm = active.fault_ppm[class.index()];
if ppm == 0 || !active.boundary(FAULT_ENV, class.name(), ppm) {
return None;
}
let decision = derive_fault(class, offset, length, &mut || active.rng.next_u64());
Some(decision)
})
}
pub fn record_fault(record: FaultRecord) {
ACTIVE.with(|slot| {
if let Some(active) = slot.borrow().clone() {
active.borrow_mut().fault_log.push(record);
}
});
}
pub fn derive_fault(
class: FaultClass,
offset: u64,
length: u64,
next: &mut dyn FnMut() -> u64,
) -> FaultDecision {
match class {
FaultClass::TornWrite => FaultDecision::TornWrite {
persisted: torn_prefix(offset, length, next),
},
FaultClass::MisdirectedWrite => {
let delta = SECTOR_BYTES * (1 + next() % 4);
let backwards = next() % 2 == 1;
let actual_offset = if backwards && offset >= delta {
offset - delta
} else {
offset.saturating_add(delta)
};
FaultDecision::MisdirectedWrite { actual_offset }
}
FaultClass::BitRot => {
let byte_offset = offset + if length == 0 { 0 } else { next() % length };
let bit = u8::try_from(next() % 8).unwrap_or(0);
FaultDecision::BitRot { byte_offset, bit }
}
FaultClass::LostWrite => FaultDecision::LostWrite,
}
}
fn torn_prefix(offset: u64, length: u64, next: &mut dyn FnMut() -> u64) -> u64 {
if length == 0 {
return 0;
}
let end = offset + length;
let first_boundary = (offset / SECTOR_BYTES + 1) * SECTOR_BYTES;
if first_boundary >= end {
return next() % length;
}
let boundaries = (end - 1 - first_boundary) / SECTOR_BYTES + 1;
let chosen = first_boundary + (next() % boundaries) * SECTOR_BYTES;
chosen - offset
}
#[derive(Debug, Clone)]
struct SplitMix64 {
state: u64,
}
impl SplitMix64 {
fn new(seed: u64) -> Self {
Self { state: seed }
}
fn next_u64(&mut self) -> u64 {
self.state = self.state.wrapping_add(0x9E37_79B9_7F4A_7C15);
let mut z = self.state;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^ (z >> 31)
}
fn below(&mut self, bound: u64) -> u64 {
if bound == 0 {
0
} else {
self.next_u64() % bound
}
}
}
#[macro_export]
macro_rules! buggify {
($env:expr, $point:expr) => {{
#[cfg(debug_assertions)]
{
$crate::dst::buggify($env, $point)
}
#[cfg(not(debug_assertions))]
{
let _ = ($env, $point);
false
}
}};
}
#[macro_export]
macro_rules! buggify_fault {
($class:expr, $offset:expr, $length:expr) => {{
#[cfg(debug_assertions)]
{
$crate::dst::roll_fault($class, $offset, $length)
}
#[cfg(not(debug_assertions))]
{
let _ = ($class, $offset, $length);
::core::option::Option::<$crate::dst::FaultDecision>::None
}
}};
}
#[cfg(test)]
mod tests {
use crate::dst::{FaultClass, FaultDecision, FaultRecord, SECTOR_BYTES};
use crate::SimulationContext;
const ENV: &str = "REDDB_EMBEDDED_RDB_CRASH_AT";
fn trace_for(seed: u64) -> Vec<u8> {
let context = SimulationContext::with_buggify_ppm(seed, 1_000_000);
let guard = context.install();
assert!(crate::buggify!(ENV, "wal_after_frame_write"));
assert!(crate::buggify!(ENV, "wal_after_frame_sync"));
assert!(crate::buggify!(ENV, "wal_after_superblock_write"));
guard.trace()
}
#[test]
fn same_seed_produces_byte_identical_buggify_trace() {
let first = trace_for(0x5EED);
let second = trace_for(0x5EED);
assert_eq!(first, second);
}
#[test]
fn buggify_can_fire_at_named_crash_boundary() {
let context = SimulationContext::with_buggify_ppm(123, 1_000_000);
let guard = context.install();
assert!(crate::buggify!(ENV, "snapshot_after_manifest_write"));
let trace = String::from_utf8(guard.trace()).unwrap();
assert!(trace.contains("env=REDDB_EMBEDDED_RDB_CRASH_AT"));
assert!(trace.contains("point=snapshot_after_manifest_write"));
assert!(trace.contains("fired=true"));
}
#[test]
fn fault_classes_round_trip_through_their_names() {
for class in FaultClass::ALL {
assert_eq!(FaultClass::from_name(class.name()), Some(class));
}
assert_eq!(FaultClass::from_name("not_a_class"), None);
}
#[test]
fn every_fault_class_is_off_by_default() {
let context = SimulationContext::new(7);
for class in FaultClass::ALL {
assert_eq!(context.fault_ppm(class), 0, "{class} must default to off");
}
let guard = context.install();
for class in FaultClass::ALL {
assert_eq!(crate::buggify_fault!(class, 0, 64), None);
}
assert!(guard.fault_log().is_empty());
assert!(!String::from_utf8(guard.trace())
.unwrap()
.contains("env=REDDB_DST_FAULT"));
}
#[test]
fn each_class_is_individually_triggerable_by_name_and_ppm() {
for armed in FaultClass::ALL {
let guard = SimulationContext::new(99)
.with_fault_class(armed, 1_000_000)
.install();
for class in FaultClass::ALL {
let decision = crate::buggify_fault!(class, SECTOR_BYTES + 8, 64);
if class == armed {
let decision = decision.unwrap_or_else(|| panic!("{armed} must fire at 1e6"));
assert_eq!(decision.class(), armed);
} else {
assert_eq!(
decision, None,
"{class} must stay off while {armed} is armed"
);
}
}
let trace = String::from_utf8(guard.trace()).unwrap();
assert!(trace.contains(&format!("point={armed}")));
}
}
#[test]
fn a_torn_write_crossing_a_sector_is_cut_at_a_sector_boundary() {
let guard = SimulationContext::new(5)
.with_fault_class(FaultClass::TornWrite, 1_000_000)
.install();
let offset = 8;
let Some(FaultDecision::TornWrite { persisted }) =
crate::buggify_fault!(FaultClass::TornWrite, offset, 1_200)
else {
panic!("torn_write must fire at 1e6")
};
assert!(
(offset + persisted) % SECTOR_BYTES == 0,
"cut at {persisted}"
);
assert!(persisted < 1_200, "a torn write always loses bytes");
drop(guard);
}
#[test]
fn a_torn_write_inside_one_sector_is_cut_at_a_strict_byte_prefix() {
let _guard = SimulationContext::new(6)
.with_fault_class(FaultClass::TornWrite, 1_000_000)
.install();
let Some(FaultDecision::TornWrite { persisted }) =
crate::buggify_fault!(FaultClass::TornWrite, 64, 48)
else {
panic!("torn_write must fire at 1e6")
};
assert!(persisted < 48, "a torn write always loses bytes");
}
#[test]
fn a_misdirected_write_lands_a_whole_number_of_sectors_away() {
let _guard = SimulationContext::new(11)
.with_fault_class(FaultClass::MisdirectedWrite, 1_000_000)
.install();
for offset in [0, 64, 4_096, 10_000] {
let Some(FaultDecision::MisdirectedWrite { actual_offset }) =
crate::buggify_fault!(FaultClass::MisdirectedWrite, offset, 64)
else {
panic!("misdirected_write must fire at 1e6")
};
assert_ne!(actual_offset, offset, "the write must land elsewhere");
assert_eq!(actual_offset.abs_diff(offset) % SECTOR_BYTES, 0);
}
}
#[test]
fn bit_rot_targets_a_byte_inside_the_read_region() {
let _guard = SimulationContext::new(13)
.with_fault_class(FaultClass::BitRot, 1_000_000)
.install();
for _ in 0..32 {
let Some(FaultDecision::BitRot { byte_offset, bit }) =
crate::buggify_fault!(FaultClass::BitRot, 100, 40)
else {
panic!("bit_rot must fire at 1e6")
};
assert!((100..140).contains(&byte_offset));
assert!(bit < 8);
}
}
#[test]
fn the_fault_log_records_class_file_offset_and_length() {
let guard = SimulationContext::new(21)
.with_fault_class(FaultClass::LostWrite, 1_000_000)
.install();
let decision = crate::buggify_fault!(FaultClass::LostWrite, 4_096, 512)
.expect("lost_write must fire at 1e6");
crate::dst::record_fault(FaultRecord::new("/db/wal.log", 4_096, 512, decision));
let log = guard.fault_log();
assert_eq!(log.len(), 1);
assert_eq!(log[0].class, FaultClass::LostWrite);
assert_eq!(log[0].file, "/db/wal.log");
assert_eq!(log[0].offset, 4_096);
assert_eq!(log[0].length, 512);
assert_eq!(
guard.fault_log_lines(),
"class=lost_write file=/db/wal.log offset=4096 length=512\n"
);
}
#[test]
fn same_seed_produces_an_identical_fault_schedule() {
fn schedule(seed: u64) -> Vec<String> {
let mut context = SimulationContext::new(seed);
for class in FaultClass::ALL {
context = context.with_fault_class(class, 300_000);
}
let guard = context.install();
for step in 0..64u64 {
for class in FaultClass::ALL {
let offset = step * 97;
if let Some(decision) = crate::buggify_fault!(class, offset, 64) {
crate::dst::record_fault(FaultRecord::new(
"/db/wal.log",
offset,
64,
decision,
));
}
}
}
guard.fault_log().iter().map(ToString::to_string).collect()
}
let first = schedule(0xDEAD);
assert!(!first.is_empty(), "the sweep must inject something");
assert_eq!(first, schedule(0xDEAD), "same seed → same fault schedule");
assert_ne!(first, schedule(0xBEEF), "different seeds must diverge");
}
#[test]
fn fault_classes_compose_with_the_crash_knob_deterministically() {
fn run(seed: u64) -> Vec<u8> {
let guard = SimulationContext::with_buggify_ppm(seed, 500_000)
.with_fault_class(FaultClass::TornWrite, 400_000)
.install();
for _ in 0..32 {
let _crashed = crate::buggify!(ENV, "wal_after_frame_write");
if let Some(decision) = crate::buggify_fault!(FaultClass::TornWrite, 64, 48) {
crate::dst::record_fault(FaultRecord::new("/db/wal.log", 64, 48, decision));
}
}
let mut out = guard.trace();
out.extend_from_slice(guard.fault_log_lines().as_bytes());
out
}
assert_eq!(
run(1234),
run(1234),
"crash + torn_write must replay exactly"
);
}
#[test]
fn no_fault_fires_without_an_installed_context() {
for class in FaultClass::ALL {
assert_eq!(crate::buggify_fault!(class, 0, 64), None);
}
}
#[test]
fn the_fault_macro_is_release_inert() {
let _guard = SimulationContext::new(3)
.with_fault_class(FaultClass::LostWrite, 1_000_000)
.install();
let fired = crate::buggify_fault!(FaultClass::LostWrite, 0, 8);
if cfg!(debug_assertions) {
assert_eq!(fired, Some(FaultDecision::LostWrite));
} else {
assert_eq!(fired, None, "release builds must never inject a fault");
}
}
}