From 57336fa124052819a65d8600c312d55f07e5a702 Mon Sep 17 00:00:00 2001 From: Lewis Date: Sun, 19 Apr 2026 10:25:07 +0300 Subject: [PATCH] feat(tranquil-store/gauntlet): new invariants & scenarios Lewis: May this revision serve well! --- .../tranquil-store/src/blockstore/cid_util.rs | 47 ++ crates/tranquil-store/src/blockstore/mod.rs | 2 + crates/tranquil-store/src/blockstore/store.rs | 78 ++- .../tranquil-store/src/gauntlet/invariants.rs | 533 ++++++++++++++++- .../tranquil-store/src/gauntlet/scenarios.rs | 555 ++++++++++++++++-- 5 files changed, 1122 insertions(+), 93 deletions(-) create mode 100644 crates/tranquil-store/src/blockstore/cid_util.rs diff --git a/crates/tranquil-store/src/blockstore/cid_util.rs b/crates/tranquil-store/src/blockstore/cid_util.rs new file mode 100644 index 0000000..92616c6 --- /dev/null +++ b/crates/tranquil-store/src/blockstore/cid_util.rs @@ -0,0 +1,47 @@ +use cid::Cid; +use multihash::Multihash; +use sha2::{Digest, Sha256}; + +use super::data_file::CID_SIZE; + +pub const DAG_CBOR_CODEC: u64 = 0x71; +pub const SHA2_256_CODE: u64 = 0x12; + +pub fn hash_to_cid(data: &[u8]) -> Cid { + let mut hasher = Sha256::new(); + hasher.update(data); + let digest = hasher.finalize(); + let mh = Multihash::wrap(SHA2_256_CODE, &digest) + .expect("SHA-256 digest is 32 bytes, well within multihash capacity"); + Cid::new_v1(DAG_CBOR_CODEC, mh) +} + +pub fn hash_to_cid_bytes(data: &[u8]) -> [u8; CID_SIZE] { + let raw = hash_to_cid(data).to_bytes(); + raw.try_into() + .expect("CIDv1 + DAG-CBOR + SHA-256 always encodes to CID_SIZE bytes") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hash_to_cid_bytes_is_deterministic() { + let a = hash_to_cid_bytes(b"hello"); + let b = hash_to_cid_bytes(b"hello"); + assert_eq!(a, b); + } + + #[test] + fn hash_to_cid_bytes_diverges_on_single_byte_change() { + assert_ne!(hash_to_cid_bytes(b"abc"), hash_to_cid_bytes(b"abd")); + } + + #[test] + fn hash_to_cid_and_bytes_agree() { + let cid = hash_to_cid(b"payload"); + let raw: [u8; CID_SIZE] = cid.to_bytes().try_into().expect("36 bytes"); + assert_eq!(raw, hash_to_cid_bytes(b"payload")); + } +} diff --git a/crates/tranquil-store/src/blockstore/mod.rs b/crates/tranquil-store/src/blockstore/mod.rs index 40f4cc4..a594901 100644 --- a/crates/tranquil-store/src/blockstore/mod.rs +++ b/crates/tranquil-store/src/blockstore/mod.rs @@ -1,3 +1,4 @@ +mod cid_util; mod compaction; mod data_file; mod group_commit; @@ -8,6 +9,7 @@ mod reader; mod store; mod types; +pub use cid_util::{DAG_CBOR_CODEC, SHA2_256_CODE, hash_to_cid, hash_to_cid_bytes}; pub use compaction::CompactionError; pub use data_file::{ BLOCK_FORMAT_VERSION, BLOCK_HEADER_SIZE, BLOCK_MAGIC, BLOCK_RECORD_OVERHEAD, CID_SIZE, diff --git a/crates/tranquil-store/src/blockstore/store.rs b/crates/tranquil-store/src/blockstore/store.rs index 4f18ad1..6e88d77 100644 --- a/crates/tranquil-store/src/blockstore/store.rs +++ b/crates/tranquil-store/src/blockstore/store.rs @@ -8,12 +8,11 @@ use cid::Cid; use jacquard_repo::error::RepoError; use jacquard_repo::repo::CommitData; use jacquard_repo::storage::BlockStore; -use multihash::Multihash; -use sha2::{Digest, Sha256}; use crate::fsync_order::PostBlockstoreHook; use crate::io::{OpenOptions, RealIO, StorageIO}; +use super::cid_util::hash_to_cid; use super::compaction::CompactionError; use super::data_file::{BLOCK_RECORD_OVERHEAD, CID_SIZE, ReadBlockRecord}; use super::group_commit::{CommitError, CommitRequest, GroupCommitConfig, GroupCommitWriter}; @@ -25,9 +24,6 @@ use super::types::{ EpochCounter, LivenessInfo, WallClockMs, WriteCursor, }; -const DAG_CBOR_CODEC: u64 = 0x71; -const SHA2_256_CODE: u64 = 0x12; - fn cid_to_bytes(cid: &Cid) -> Result<[u8; CID_SIZE], RepoError> { let raw = cid.to_bytes(); let len = raw.len(); @@ -41,16 +37,6 @@ fn cid_to_bytes(cid: &Cid) -> Result<[u8; CID_SIZE], RepoError> { }) } -fn hash_and_cid(data: &[u8]) -> Result { - let mut hasher = Sha256::new(); - hasher.update(data); - let hash = hasher.finalize(); - let multihash = Multihash::wrap(SHA2_256_CODE, &hash).map_err(|e| { - RepoError::storage(io::Error::new(io::ErrorKind::InvalidData, e.to_string())) - })?; - Ok(Cid::new_v1(DAG_CBOR_CODEC, multihash)) -} - fn block_index_err_to_repo(e: super::hash_index::BlockIndexError) -> RepoError { RepoError::storage(io::Error::other(e.to_string())) } @@ -123,15 +109,26 @@ impl Drop for QuiesceGuard { } } -#[derive(Clone)] -pub struct TranquilBlockStore { +pub struct TranquilBlockStore { writer: Arc, - reader: Arc>, + reader: Arc>, index: Arc, epoch: EpochCounter, data_dir: PathBuf, } +impl Clone for TranquilBlockStore { + fn clone(&self) -> Self { + Self { + writer: Arc::clone(&self.writer), + reader: Arc::clone(&self.reader), + index: Arc::clone(&self.index), + epoch: self.epoch.clone(), + data_dir: self.data_dir.clone(), + } + } +} + struct WriterHandle { inner: parking_lot::Mutex>, } @@ -153,7 +150,7 @@ impl Drop for WriterHandle { } } -impl TranquilBlockStore { +impl TranquilBlockStore { pub fn open(config: BlockStoreConfig) -> Result { Self::open_with_hook(config, None) } @@ -162,6 +159,26 @@ impl TranquilBlockStore { config: BlockStoreConfig, post_sync_hook: Option>, ) -> Result { + Self::open_with_io_hook(config, RealIO::new, post_sync_hook) + } +} + +impl TranquilBlockStore { + pub fn open_with_io(config: BlockStoreConfig, make_io: F) -> Result + where + F: Fn() -> S + Send + Sync + Clone + 'static, + { + Self::open_with_io_hook(config, make_io, None) + } + + pub fn open_with_io_hook( + config: BlockStoreConfig, + make_io: F, + post_sync_hook: Option>, + ) -> Result + where + F: Fn() -> S + Send + Sync + Clone + 'static, + { if config.data_dir == config.index_dir { return Err(RepoError::storage(io::Error::new( io::ErrorKind::InvalidInput, @@ -173,7 +190,7 @@ impl TranquilBlockStore { let index = BlockIndex::open(&config.index_dir).map_err(RepoError::storage)?; - let io = RealIO::new(); + let io = make_io(); let (replayed, file_cursors) = super::hint::replay_hints_into_block_index( &io, @@ -195,8 +212,13 @@ impl TranquilBlockStore { let max_file_size = config.max_file_size; let shard_count = config.shard_count; let data_dir_for_closure = data_dir.clone(); + let make_io_for_manager = make_io.clone(); let make_manager = move || { - DataFileManager::new(RealIO::new(), data_dir_for_closure.clone(), max_file_size) + DataFileManager::new( + make_io_for_manager(), + data_dir_for_closure.clone(), + max_file_size, + ) }; let checkpoint_epoch = index.loaded_checkpoint_epoch(); @@ -214,7 +236,7 @@ impl TranquilBlockStore { let epoch = writer.epoch().clone(); let manager_for_reader = Arc::new(DataFileManager::new( - RealIO::new(), + make_io(), data_dir.clone(), max_file_size, )); @@ -234,7 +256,7 @@ impl TranquilBlockStore { }) } - fn recover_from_file_cursors( + fn recover_from_file_cursors( io: &S, data_dir: &Path, index: &BlockIndex, @@ -256,7 +278,7 @@ impl TranquilBlockStore { }) } - fn replay_single_file( + fn replay_single_file( io: &S, data_dir: &Path, index: &BlockIndex, @@ -284,7 +306,7 @@ impl TranquilBlockStore { result } - fn scan_and_index( + fn scan_and_index( io: &S, index: &BlockIndex, fd: crate::io::FileId, @@ -594,7 +616,7 @@ impl TranquilBlockStore { } } -impl BlockStore for TranquilBlockStore { +impl BlockStore for TranquilBlockStore { async fn get(&self, cid: &Cid) -> Result, RepoError> { let cid_bytes = cid_to_bytes(cid)?; let reader = Arc::clone(&self.reader); @@ -605,7 +627,7 @@ impl BlockStore for TranquilBlockStore { } async fn put(&self, data: &[u8]) -> Result { - let cid = hash_and_cid(data)?; + let cid = hash_to_cid(data); let cid_bytes = cid_to_bytes(&cid)?; self.send_put_blocks(vec![(cid_bytes, data.to_vec())]) .await?; @@ -666,7 +688,7 @@ impl BlockStore for TranquilBlockStore { } } -impl TranquilBlockStore { +impl TranquilBlockStore { pub async fn decrement_refs(&self, cids: &[Cid]) -> Result<(), RepoError> { if cids.is_empty() { return Ok(()); diff --git a/crates/tranquil-store/src/gauntlet/invariants.rs b/crates/tranquil-store/src/gauntlet/invariants.rs index a8fe7f8..08036c5 100644 --- a/crates/tranquil-store/src/gauntlet/invariants.rs +++ b/crates/tranquil-store/src/gauntlet/invariants.rs @@ -1,4 +1,5 @@ use std::collections::{HashMap, HashSet}; +use std::path::PathBuf; use std::sync::Arc; use async_trait::async_trait; @@ -6,7 +7,9 @@ use cid::Cid; use jacquard_repo::mst::Mst; use super::oracle::{Oracle, hex_short, try_cid_to_fixed}; -use crate::blockstore::{CidBytes, TranquilBlockStore}; +use crate::blockstore::{CidBytes, CompactionError, TranquilBlockStore, hash_to_cid_bytes}; +use crate::eventlog::{EventSequence, SegmentId}; +use crate::io::{RealIO, StorageIO}; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct InvariantSet(u32); @@ -18,12 +21,28 @@ impl InvariantSet { pub const ACKED_WRITE_PERSISTENCE: Self = Self(1 << 2); pub const READ_AFTER_WRITE: Self = Self(1 << 3); pub const RESTART_IDEMPOTENT: Self = Self(1 << 4); + pub const COMPACTION_IDEMPOTENT: Self = Self(1 << 5); + pub const NO_ORPHAN_FILES: Self = Self(1 << 6); + pub const BYTE_BUDGET: Self = Self(1 << 7); + pub const MANIFEST_EQUALS_REALITY: Self = Self(1 << 8); + pub const CHECKSUM_COVERAGE: Self = Self(1 << 9); + pub const MONOTONIC_SEQ: Self = Self(1 << 10); + pub const FSYNC_ORDERING: Self = Self(1 << 11); + pub const TOMBSTONE_BOUND: Self = Self(1 << 12); const ALL_KNOWN: u32 = Self::REFCOUNT_CONSERVATION.0 | Self::REACHABILITY.0 | Self::ACKED_WRITE_PERSISTENCE.0 | Self::READ_AFTER_WRITE.0 - | Self::RESTART_IDEMPOTENT.0; + | Self::RESTART_IDEMPOTENT.0 + | Self::COMPACTION_IDEMPOTENT.0 + | Self::NO_ORPHAN_FILES.0 + | Self::BYTE_BUDGET.0 + | Self::MANIFEST_EQUALS_REALITY.0 + | Self::CHECKSUM_COVERAGE.0 + | Self::MONOTONIC_SEQ.0 + | Self::FSYNC_ORDERING.0 + | Self::TOMBSTONE_BOUND.0; pub const fn contains(self, other: Self) -> bool { (self.0 & other.0) == other.0 @@ -49,33 +68,52 @@ impl std::ops::BitOr for InvariantSet { } } -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct InvariantViolation { pub invariant: &'static str, pub detail: String, } -pub struct InvariantCtx<'a> { - pub store: &'a Arc, +#[derive(Debug, Clone, Copy)] +pub struct SnapshotEvent { + pub seq: EventSequence, + pub timestamp_us: u64, + pub event_type_raw: u8, + pub did_hash: u32, +} + +#[derive(Debug, Clone)] +pub struct EventLogSnapshot { + pub segments_dir: PathBuf, + pub max_segment_size: u64, + pub synced_seq: EventSequence, + pub segments: Vec, + pub events: Vec, + pub segment_last_ts: Vec<(SegmentId, u64)>, +} + +pub struct InvariantCtx<'a, S: StorageIO + Send + Sync + 'static = RealIO> { + pub store: &'a Arc>, pub oracle: &'a Oracle, pub root: Option, + pub eventlog: Option<&'a EventLogSnapshot>, } #[async_trait] -pub trait Invariant: Send + Sync { +pub trait Invariant: Send + Sync { fn name(&self) -> &'static str; - async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation>; + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation>; } pub struct RefcountConservation; #[async_trait] -impl Invariant for RefcountConservation { +impl Invariant for RefcountConservation { fn name(&self) -> &'static str { "RefcountConservation" } - async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { let live: Vec<(String, CidBytes)> = ctx.oracle.live_cids_labeled(); let live_set: HashSet = live.iter().map(|(_, c)| *c).collect(); let index: HashMap = ctx @@ -116,12 +154,12 @@ impl Invariant for RefcountConservation { pub struct Reachability; #[async_trait] -impl Invariant for Reachability { +impl Invariant for Reachability { fn name(&self) -> &'static str { "Reachability" } - async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { let violations: Vec = ctx .oracle .live_cids_labeled() @@ -147,12 +185,12 @@ impl Invariant for Reachability { pub struct AckedWritePersistence; #[async_trait] -impl Invariant for AckedWritePersistence { +impl Invariant for AckedWritePersistence { fn name(&self) -> &'static str { "AckedWritePersistence" } - async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { let Some(root) = ctx.root else { if ctx.oracle.live_count() == 0 { return Ok(()); @@ -195,12 +233,12 @@ impl Invariant for AckedWritePersistence { pub struct ReadAfterWrite; #[async_trait] -impl Invariant for ReadAfterWrite { +impl Invariant for ReadAfterWrite { fn name(&self) -> &'static str { "ReadAfterWrite" } - async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { let Some(root) = ctx.root else { return Ok(()); }; @@ -246,13 +284,460 @@ impl Invariant for ReadAfterWrite { } } -pub fn invariants_for(set: InvariantSet) -> Vec> { +pub struct CompactionIdempotent; + +#[async_trait] +impl Invariant for CompactionIdempotent { + fn name(&self) -> &'static str { + "CompactionIdempotent" + } + + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + let store_a = ctx.store.clone(); + let first = tokio::task::spawn_blocking(move || compact_by_liveness(&store_a)) + .await + .map_err(|e| InvariantViolation { + invariant: "CompactionIdempotent", + detail: format!("first compaction join: {e}"), + })?; + if let Err(e) = first { + return Err(InvariantViolation { + invariant: "CompactionIdempotent", + detail: format!("first compaction: {e}"), + }); + } + + let pre = snapshot(ctx.store); + + let store_b = ctx.store.clone(); + let second = tokio::task::spawn_blocking(move || compact_by_liveness(&store_b)) + .await + .map_err(|e| InvariantViolation { + invariant: "CompactionIdempotent", + detail: format!("second compaction join: {e}"), + })?; + if let Err(e) = second { + return Err(InvariantViolation { + invariant: "CompactionIdempotent", + detail: format!("second compaction: {e}"), + }); + } + + let post = snapshot(ctx.store); + + if pre == post { + Ok(()) + } else { + Err(InvariantViolation { + invariant: "CompactionIdempotent", + detail: format!( + "second compaction changed observable state: pre={} entries, post={} entries", + pre.len(), + post.len(), + ), + }) + } + } +} + +fn snapshot( + store: &Arc>, +) -> Vec<(CidBytes, u32)> { + let mut v: Vec<(CidBytes, u32)> = store + .block_index() + .live_entries_snapshot() + .into_iter() + .map(|(c, r)| (c, r.raw())) + .collect(); + v.sort_unstable_by(|a, b| a.0.cmp(&b.0)); + v +} + +const COMPACT_LIVENESS_CEILING: f64 = 0.99; + +fn compact_by_liveness( + store: &TranquilBlockStore, +) -> Result<(), String> { + let liveness = store + .compaction_liveness(0) + .map_err(|e| format!("compaction_liveness: {e}"))?; + let targets: Vec<_> = liveness + .iter() + .filter(|(_, info)| info.total_blocks > 0 && info.ratio() < COMPACT_LIVENESS_CEILING) + .map(|(&fid, _)| fid) + .collect(); + targets + .into_iter() + .try_for_each(|fid| match store.compact_file(fid, 0) { + Ok(_) => Ok(()), + Err(CompactionError::ActiveFileCannotBeCompacted) => Ok(()), + Err(e) => Err(format!("{fid}: {e}")), + }) +} + +pub struct NoOrphanFiles; + +#[async_trait] +impl Invariant for NoOrphanFiles { + fn name(&self) -> &'static str { + "NoOrphanFiles" + } + + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + let store_c = ctx.store.clone(); + let result = tokio::task::spawn_blocking(move || { + let disk = store_c.list_data_files().map_err(|e| e.to_string())?; + let liveness = store_c.compaction_liveness(0).map_err(|e| e.to_string())?; + let orphans: Vec = disk + .iter() + .filter(|fid| !liveness.contains_key(fid)) + .map(|fid| format!("{fid}")) + .collect(); + Ok::<_, String>(orphans) + }) + .await + .map_err(|e| InvariantViolation { + invariant: "NoOrphanFiles", + detail: format!("join: {e}"), + })?; + + let orphans = result.map_err(|e| InvariantViolation { + invariant: "NoOrphanFiles", + detail: e, + })?; + + if orphans.is_empty() { + Ok(()) + } else { + Err(InvariantViolation { + invariant: "NoOrphanFiles", + detail: format!("files on disk missing from index: {}", orphans.join(", ")), + }) + } + } +} + +pub struct ByteBudget { + pub overhead_factor: f64, + pub floor_bytes: u64, +} + +impl Default for ByteBudget { + fn default() -> Self { + Self { + overhead_factor: 8.0, + floor_bytes: 1 << 20, + } + } +} + +#[async_trait] +impl Invariant for ByteBudget { + fn name(&self) -> &'static str { + "ByteBudget" + } + + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + let store = ctx.store.clone(); + let factor = self.overhead_factor; + let floor = self.floor_bytes; + tokio::task::spawn_blocking(move || { + let liveness = store.compaction_liveness(0).map_err(|e| e.to_string())?; + let live: u64 = liveness.values().map(|i| i.live_bytes).sum(); + let total: u64 = liveness.values().map(|i| i.total_bytes).sum(); + let budget = (live as f64 * factor) as u64 + floor; + if total <= budget { + Ok(()) + } else { + Err(format!( + "total_bytes {total} exceeds budget {budget}: live_bytes {live}, factor {factor}, floor {floor}" + )) + } + }) + .await + .map_err(|e| InvariantViolation { + invariant: "ByteBudget", + detail: format!("join: {e}"), + })? + .map_err(|e| InvariantViolation { + invariant: "ByteBudget", + detail: e, + }) + } +} + +pub struct ManifestEqualsReality; + +#[async_trait] +impl Invariant for ManifestEqualsReality { + fn name(&self) -> &'static str { + "ManifestEqualsReality" + } + + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + let store = ctx.store.clone(); + tokio::task::spawn_blocking(move || { + let listed = store.list_data_files().map_err(|e| e.to_string())?; + let liveness = store.compaction_liveness(0).map_err(|e| e.to_string())?; + + let mut violations: Vec = Vec::new(); + listed.iter().for_each(|fid| { + let path = store.data_file_path(*fid); + match std::fs::metadata(&path) { + Err(e) => violations.push(format!("{fid}: metadata {e}")), + Ok(meta) => { + let on_disk = meta.len(); + match liveness.get(fid) { + None => violations.push(format!( + "{fid}: listed on disk at {on_disk} B but not in index liveness" + )), + Some(info) if on_disk < info.total_bytes => { + violations.push(format!( + "{fid}: on-disk {on_disk} B < index total_bytes {}", + info.total_bytes + )); + } + Some(info) if on_disk > info.total_bytes => { + violations.push(format!( + "{fid}: on-disk {on_disk} B > index total_bytes {}, {} B unaccounted", + info.total_bytes, + on_disk - info.total_bytes + )); + } + Some(_) => {} + } + } + } + }); + + let listed_set: std::collections::HashSet<_> = listed.into_iter().collect(); + liveness.keys().for_each(|fid| { + if !listed_set.contains(fid) { + violations.push(format!("{fid}: in index liveness but missing on disk")); + } + }); + + if violations.is_empty() { + Ok(()) + } else { + Err(violations.join("; ")) + } + }) + .await + .map_err(|e| InvariantViolation { + invariant: "ManifestEqualsReality", + detail: format!("join: {e}"), + })? + .map_err(|e| InvariantViolation { + invariant: "ManifestEqualsReality", + detail: e, + }) + } +} + +pub struct ChecksumCoverage; + +#[async_trait] +impl Invariant for ChecksumCoverage { + fn name(&self) -> &'static str { + "ChecksumCoverage" + } + + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + let violations: Vec = ctx + .oracle + .live_cids_labeled() + .into_iter() + .filter_map(|(label, expected)| match ctx.store.get_block_sync(&expected) { + Ok(Some(bytes)) => { + let actual = hash_to_cid_bytes(&bytes); + (actual != expected).then(|| { + format!( + "{label}: silent corruption, bytes hash to {} but store returned them under {}", + hex_short(&actual), + hex_short(&expected), + ) + }) + } + Ok(None) => Some(format!( + "{label}: live CID {} missing from store", + hex_short(&expected) + )), + Err(e) => Some(format!( + "{label}: read error for live CID {}: {e}", + hex_short(&expected) + )), + }) + .collect(); + + if violations.is_empty() { + Ok(()) + } else { + Err(InvariantViolation { + invariant: "ChecksumCoverage", + detail: violations.join("; "), + }) + } + } +} + +pub struct MonotonicSeq; + +#[async_trait] +impl Invariant for MonotonicSeq { + fn name(&self) -> &'static str { + "MonotonicSeq" + } + + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + let Some(el) = ctx.eventlog else { + return Ok(()); + }; + let mut violations: Vec = Vec::new(); + el.events + .iter() + .zip(el.events.iter().skip(1)) + .for_each(|(prev, next)| match next.seq.raw() { + n if n == prev.seq.raw() + 1 => {} + n if n == prev.seq.raw() => violations.push(format!("duplicate seq {n}")), + n => violations.push(format!( + "gap: seq {} followed by {n}, expected {}", + prev.seq.raw(), + prev.seq.raw() + 1 + )), + }); + if ctx.oracle.last_retention_cutoff_us().is_none() + && let Some(first) = el.events.first() + && first.seq.raw() != 1 + { + violations.push(format!( + "first persisted seq is {}, expected 1", + first.seq.raw() + )); + } + let acked_max = ctx + .oracle + .synced_events() + .iter() + .map(|e| e.seq.raw()) + .max() + .unwrap_or(0); + let disk_max = el.events.last().map(|e| e.seq.raw()).unwrap_or(0); + if disk_max < acked_max { + violations.push(format!( + "acked seq {acked_max} missing on disk, disk max {disk_max}" + )); + } + if violations.is_empty() { + Ok(()) + } else { + Err(InvariantViolation { + invariant: "MonotonicSeq", + detail: violations.join("; "), + }) + } + } +} + +pub struct FsyncOrdering; + +#[async_trait] +impl Invariant for FsyncOrdering { + fn name(&self) -> &'static str { + "FsyncOrdering" + } + + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + let Some(el) = ctx.eventlog else { + return Ok(()); + }; + let mut violations: Vec = Vec::new(); + + let acked_seqs: HashSet = ctx + .oracle + .synced_events() + .iter() + .map(|e| e.seq.raw()) + .collect(); + let disk_seqs: HashSet = el.events.iter().map(|e| e.seq.raw()).collect(); + let missing: Vec = acked_seqs.difference(&disk_seqs).copied().collect(); + if !missing.is_empty() { + let mut sorted = missing; + sorted.sort_unstable(); + violations.push(format!( + "{} acked events lost on disk, lowest missing seq {}", + sorted.len(), + sorted[0] + )); + } + + if let Some(last_synced) = ctx.oracle.last_synced_seq() + && el.synced_seq.raw() != 0 + && el.synced_seq.raw() < last_synced.raw() + { + violations.push(format!( + "writer synced_seq {} below oracle last_synced_seq {}", + el.synced_seq.raw(), + last_synced.raw() + )); + } + + if violations.is_empty() { + Ok(()) + } else { + Err(InvariantViolation { + invariant: "FsyncOrdering", + detail: violations.join("; "), + }) + } + } +} + +pub struct TombstoneBound; + +#[async_trait] +impl Invariant for TombstoneBound { + fn name(&self) -> &'static str { + "TombstoneBound" + } + + async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + let Some(el) = ctx.eventlog else { + return Ok(()); + }; + let Some(cutoff_us) = ctx.oracle.last_retention_cutoff_us() else { + return Ok(()); + }; + + let active = el.segments.last().copied(); + + let stale: Vec = el + .segment_last_ts + .iter() + .filter(|(id, last_ts)| Some(*id) != active && *last_ts < cutoff_us) + .map(|(id, last_ts)| format!("segment {id} last_ts {last_ts} < cutoff {cutoff_us}")) + .collect(); + + if stale.is_empty() { + Ok(()) + } else { + Err(InvariantViolation { + invariant: "TombstoneBound", + detail: stale.join("; "), + }) + } + } +} + +pub fn invariants_for( + set: InvariantSet, +) -> Vec>> { let unknown = set.unknown_bits(); assert!( unknown == 0, "invariants_for: unknown InvariantSet bits 0x{unknown:x}; all bits must map to an impl" ); - let candidates: Vec<(InvariantSet, Box)> = vec![ + let candidates: Vec<(InvariantSet, Box>)> = vec![ ( InvariantSet::REFCOUNT_CONSERVATION, Box::new(RefcountConservation), @@ -263,6 +748,20 @@ pub fn invariants_for(set: InvariantSet) -> Vec> { Box::new(AckedWritePersistence), ), (InvariantSet::READ_AFTER_WRITE, Box::new(ReadAfterWrite)), + ( + InvariantSet::COMPACTION_IDEMPOTENT, + Box::new(CompactionIdempotent), + ), + (InvariantSet::NO_ORPHAN_FILES, Box::new(NoOrphanFiles)), + (InvariantSet::BYTE_BUDGET, Box::new(ByteBudget::default())), + ( + InvariantSet::MANIFEST_EQUALS_REALITY, + Box::new(ManifestEqualsReality), + ), + (InvariantSet::CHECKSUM_COVERAGE, Box::new(ChecksumCoverage)), + (InvariantSet::MONOTONIC_SEQ, Box::new(MonotonicSeq)), + (InvariantSet::FSYNC_ORDERING, Box::new(FsyncOrdering)), + (InvariantSet::TOMBSTONE_BOUND, Box::new(TombstoneBound)), ]; candidates .into_iter() diff --git a/crates/tranquil-store/src/gauntlet/scenarios.rs b/crates/tranquil-store/src/gauntlet/scenarios.rs index ebb031c..4bacba4 100644 --- a/crates/tranquil-store/src/gauntlet/scenarios.rs +++ b/crates/tranquil-store/src/gauntlet/scenarios.rs @@ -1,20 +1,98 @@ use super::invariants::InvariantSet; use super::op::{CollectionName, Seed}; use super::runner::{ - GauntletConfig, IoBackend, MaxFileSize, OpInterval, RestartPolicy, RunLimits, ShardCount, - StoreConfig, WallMs, + EventLogConfig, GauntletConfig, IoBackend, MaxFileSize, MaxSegmentSize, OpInterval, + RestartPolicy, RunLimits, ShardCount, StoreConfig, WallMs, WriterConcurrency, }; use super::workload::{ - KeySpaceSize, OpCount, OpWeights, SizeDistribution, ValueBytes, WorkloadModel, + ByteRange, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, SizeDistribution, + ValueBytes, WorkloadModel, }; use crate::blockstore::GroupCommitConfig; +use crate::sim::FaultConfig; -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Scenario { SmokePR, MstChurn, MstRestartChurn, FullStackRestart, + CatastrophicChurn, + HugeValues, + TinyBatches, + GiantBatches, + ManyFiles, + ModerateFaults, + AggressiveFaults, + TornPages, + Fsyncgate, + FirehoseFanout, + ContendedReaders, + ContendedWriters, +} + +impl Scenario { + pub const fn name(self) -> &'static str { + match self { + Self::SmokePR => "SmokePR", + Self::MstChurn => "MstChurn", + Self::MstRestartChurn => "MstRestartChurn", + Self::FullStackRestart => "FullStackRestart", + Self::CatastrophicChurn => "CatastrophicChurn", + Self::HugeValues => "HugeValues", + Self::TinyBatches => "TinyBatches", + Self::GiantBatches => "GiantBatches", + Self::ManyFiles => "ManyFiles", + Self::ModerateFaults => "ModerateFaults", + Self::AggressiveFaults => "AggressiveFaults", + Self::TornPages => "TornPages", + Self::Fsyncgate => "Fsyncgate", + Self::FirehoseFanout => "FirehoseFanout", + Self::ContendedReaders => "ContendedReaders", + Self::ContendedWriters => "ContendedWriters", + } + } + + pub fn from_name(name: &str) -> Option { + Self::ALL.iter().copied().find(|s| s.name() == name) + } + + pub const ALL: &'static [Scenario] = &[ + Self::SmokePR, + Self::MstChurn, + Self::MstRestartChurn, + Self::FullStackRestart, + Self::CatastrophicChurn, + Self::HugeValues, + Self::TinyBatches, + Self::GiantBatches, + Self::ManyFiles, + Self::ModerateFaults, + Self::AggressiveFaults, + Self::TornPages, + Self::Fsyncgate, + Self::FirehoseFanout, + Self::ContendedReaders, + Self::ContendedWriters, + ]; +} + +impl std::fmt::Display for Scenario { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.name()) + } +} + +#[derive(Debug, thiserror::Error)] +#[error("unknown scenario: {0}")] +pub struct UnknownScenario(pub String); + +impl std::str::FromStr for Scenario { + type Err = UnknownScenario; + + fn from_str(s: &str) -> Result { + Self::from_name(s).ok_or_else(|| UnknownScenario(s.to_string())) + } } pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig { @@ -23,6 +101,18 @@ pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig { Scenario::MstChurn => mst_churn(seed), Scenario::MstRestartChurn => mst_restart_churn(seed), Scenario::FullStackRestart => full_stack_restart(seed), + Scenario::CatastrophicChurn => catastrophic_churn(seed), + Scenario::HugeValues => huge_values(seed), + Scenario::TinyBatches => tiny_batches(seed), + Scenario::GiantBatches => giant_batches(seed), + Scenario::ManyFiles => many_files(seed), + Scenario::ModerateFaults => moderate_faults(seed), + Scenario::AggressiveFaults => aggressive_faults(seed), + Scenario::TornPages => torn_pages(seed), + Scenario::Fsyncgate => fsyncgate(seed), + Scenario::FirehoseFanout => firehose_fanout(seed), + Scenario::ContendedReaders => contended_readers(seed), + Scenario::ContendedWriters => contended_writers(seed), } } @@ -33,6 +123,31 @@ fn default_collections() -> Vec { ] } +fn block_weights(add: u32, delete: u32, compact: u32, checkpoint: u32) -> OpWeights { + OpWeights { + add, + delete, + compact, + checkpoint, + ..OpWeights::default() + } +} + +fn block_workload( + weights: OpWeights, + size_distribution: SizeDistribution, + key_space: KeySpaceSize, +) -> WorkloadModel { + WorkloadModel { + weights, + size_distribution, + collections: default_collections(), + key_space, + did_space: DidSpaceSize(32), + retention_max_secs: RetentionMaxSecs(3600), + } +} + fn tiny_store() -> StoreConfig { StoreConfig { max_file_size: MaxFileSize(4096), @@ -49,17 +164,11 @@ fn smoke_pr(seed: Seed) -> GauntletConfig { GauntletConfig { seed, io: IoBackend::Real, - workload: WorkloadModel { - weights: OpWeights { - add: 80, - delete: 0, - compact: 10, - checkpoint: 10, - }, - size_distribution: SizeDistribution::Fixed(ValueBytes(64)), - collections: default_collections(), - key_space: KeySpaceSize(200), - }, + workload: block_workload( + block_weights(80, 0, 10, 10), + SizeDistribution::Fixed(ValueBytes(64)), + KeySpaceSize(200), + ), op_count: OpCount(10_000), invariants: InvariantSet::REFCOUNT_CONSERVATION | InvariantSet::REACHABILITY @@ -71,6 +180,8 @@ fn smoke_pr(seed: Seed) -> GauntletConfig { }, restart_policy: RestartPolicy::EveryNOps(OpInterval(2_000)), store: tiny_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(1), } } @@ -78,17 +189,11 @@ fn mst_churn(seed: Seed) -> GauntletConfig { GauntletConfig { seed, io: IoBackend::Real, - workload: WorkloadModel { - weights: OpWeights { - add: 85, - delete: 0, - compact: 10, - checkpoint: 5, - }, - size_distribution: SizeDistribution::Fixed(ValueBytes(64)), - collections: default_collections(), - key_space: KeySpaceSize(2_000), - }, + workload: block_workload( + block_weights(85, 0, 10, 5), + SizeDistribution::Fixed(ValueBytes(64)), + KeySpaceSize(2_000), + ), op_count: OpCount(100_000), invariants: InvariantSet::REFCOUNT_CONSERVATION | InvariantSet::REACHABILITY @@ -100,6 +205,8 @@ fn mst_churn(seed: Seed) -> GauntletConfig { }, restart_policy: RestartPolicy::Never, store: tiny_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(1), } } @@ -107,17 +214,11 @@ fn mst_restart_churn(seed: Seed) -> GauntletConfig { GauntletConfig { seed, io: IoBackend::Real, - workload: WorkloadModel { - weights: OpWeights { - add: 85, - delete: 0, - compact: 10, - checkpoint: 5, - }, - size_distribution: SizeDistribution::Fixed(ValueBytes(64)), - collections: default_collections(), - key_space: KeySpaceSize(2_000), - }, + workload: block_workload( + block_weights(85, 0, 10, 5), + SizeDistribution::Fixed(ValueBytes(64)), + KeySpaceSize(2_000), + ), op_count: OpCount(100_000), invariants: InvariantSet::REFCOUNT_CONSERVATION | InvariantSet::REACHABILITY @@ -129,6 +230,8 @@ fn mst_restart_churn(seed: Seed) -> GauntletConfig { }, restart_policy: RestartPolicy::PoissonByOps(OpInterval(5_000)), store: tiny_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(1), } } @@ -136,17 +239,11 @@ fn full_stack_restart(seed: Seed) -> GauntletConfig { GauntletConfig { seed, io: IoBackend::Real, - workload: WorkloadModel { - weights: OpWeights { - add: 80, - delete: 0, - compact: 15, - checkpoint: 5, - }, - size_distribution: SizeDistribution::Fixed(ValueBytes(80)), - collections: default_collections(), - key_space: KeySpaceSize(500), - }, + workload: block_workload( + block_weights(80, 0, 15, 5), + SizeDistribution::Fixed(ValueBytes(80)), + KeySpaceSize(500), + ), op_count: OpCount(5_000), invariants: InvariantSet::REFCOUNT_CONSERVATION | InvariantSet::REACHABILITY @@ -162,5 +259,367 @@ fn full_stack_restart(seed: Seed) -> GauntletConfig { group_commit: GroupCommitConfig::default(), shard_count: ShardCount(1), }, + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn phase2_invariants() -> InvariantSet { + InvariantSet::REFCOUNT_CONSERVATION + | InvariantSet::REACHABILITY + | InvariantSet::ACKED_WRITE_PERSISTENCE + | InvariantSet::READ_AFTER_WRITE + | InvariantSet::RESTART_IDEMPOTENT + | InvariantSet::COMPACTION_IDEMPOTENT + | InvariantSet::BYTE_BUDGET + | InvariantSet::MANIFEST_EQUALS_REALITY + | InvariantSet::CHECKSUM_COVERAGE +} + +fn catastrophic_churn(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Real, + workload: block_workload( + block_weights(94, 0, 5, 1), + SizeDistribution::Fixed(ValueBytes(64)), + KeySpaceSize(200), + ), + op_count: OpCount(1_000_000), + invariants: phase2_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(30 * 60_000)), + }, + restart_policy: RestartPolicy::PoissonByOps(OpInterval(50_000)), + store: tiny_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn huge_values(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Real, + workload: block_workload( + block_weights(85, 5, 8, 2), + SizeDistribution::HeavyTail( + ByteRange::new(ValueBytes(256), ValueBytes(16 * 1024 * 1024)) + .expect("huge_values ByteRange"), + ), + KeySpaceSize(64), + ), + 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(10 * 60_000)), + }, + restart_policy: RestartPolicy::EveryNOps(OpInterval(500)), + store: StoreConfig { + max_file_size: MaxFileSize(32 * 1024 * 1024), + group_commit: GroupCommitConfig::default(), + shard_count: ShardCount(1), + }, + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn tiny_batches(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Real, + workload: block_workload( + block_weights(85, 0, 5, 10), + SizeDistribution::Fixed(ValueBytes(64)), + KeySpaceSize(500), + ), + op_count: OpCount(10_000), + invariants: phase2_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(120_000)), + }, + restart_policy: RestartPolicy::EveryNOps(OpInterval(2_000)), + store: StoreConfig { + max_file_size: MaxFileSize(4096), + group_commit: GroupCommitConfig { + max_batch_size: 1, + checkpoint_interval_ms: 100, + checkpoint_write_threshold: 1, + ..GroupCommitConfig::default() + }, + shard_count: ShardCount(1), + }, + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn giant_batches(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Real, + workload: block_workload( + block_weights(95, 0, 3, 2), + SizeDistribution::Fixed(ValueBytes(64)), + KeySpaceSize(5_000), + ), + op_count: OpCount(50_000), + invariants: phase2_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(10 * 60_000)), + }, + restart_policy: RestartPolicy::EveryNOps(OpInterval(10_000)), + store: StoreConfig { + max_file_size: MaxFileSize(16 * 1024 * 1024), + group_commit: GroupCommitConfig { + max_batch_size: 100_000, + checkpoint_interval_ms: 5_000, + checkpoint_write_threshold: 100_000, + ..GroupCommitConfig::default() + }, + shard_count: ShardCount(1), + }, + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn many_files(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Real, + workload: block_workload( + block_weights(80, 10, 5, 5), + SizeDistribution::Fixed(ValueBytes(128)), + KeySpaceSize(2_000), + ), + op_count: OpCount(200_000), + invariants: phase2_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(20 * 60_000)), + }, + restart_policy: RestartPolicy::PoissonByOps(OpInterval(5_000)), + store: StoreConfig { + max_file_size: MaxFileSize(256), + group_commit: GroupCommitConfig::default(), + shard_count: ShardCount(1), + }, + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn sim_invariants() -> InvariantSet { + InvariantSet::REFCOUNT_CONSERVATION + | InvariantSet::REACHABILITY + | InvariantSet::ACKED_WRITE_PERSISTENCE + | InvariantSet::READ_AFTER_WRITE + | InvariantSet::RESTART_IDEMPOTENT + | InvariantSet::NO_ORPHAN_FILES + | InvariantSet::BYTE_BUDGET + | InvariantSet::CHECKSUM_COVERAGE +} + +fn sim_microbench_workload() -> WorkloadModel { + block_workload( + block_weights(80, 10, 5, 5), + SizeDistribution::Fixed(ValueBytes(128)), + KeySpaceSize(500), + ) +} + +fn sim_store() -> StoreConfig { + StoreConfig { + max_file_size: MaxFileSize(16 * 1024), + group_commit: GroupCommitConfig::default(), + shard_count: ShardCount(1), + } +} + +fn moderate_faults(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Simulated { + fault: FaultConfig::moderate(), + }, + workload: sim_microbench_workload(), + op_count: OpCount(50_000), + invariants: sim_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(10 * 60_000)), + }, + restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(2_000)), + store: sim_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn aggressive_faults(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Simulated { + fault: FaultConfig::aggressive(), + }, + workload: sim_microbench_workload(), + op_count: OpCount(50_000), + invariants: sim_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(10 * 60_000)), + }, + restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(2_000)), + store: sim_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn torn_pages(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Simulated { + fault: FaultConfig::torn_pages_only(), + }, + workload: sim_microbench_workload(), + op_count: OpCount(20_000), + invariants: sim_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(5 * 60_000)), + }, + restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(1_000)), + store: sim_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn fsyncgate(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Simulated { + fault: FaultConfig::fsyncgate_only(), + }, + workload: sim_microbench_workload(), + op_count: OpCount(10_000), + invariants: sim_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(5 * 60_000)), + }, + restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(500)), + store: sim_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(1), + } +} + +fn firehose_fanout(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Simulated { + fault: FaultConfig::moderate(), + }, + workload: WorkloadModel { + weights: OpWeights { + add: 20, + compact: 2, + checkpoint: 3, + append_event: 60, + sync_event_log: 10, + run_retention: 5, + ..OpWeights::default() + }, + size_distribution: SizeDistribution::Fixed(ValueBytes(128)), + collections: default_collections(), + key_space: KeySpaceSize(500), + did_space: DidSpaceSize(64), + retention_max_secs: RetentionMaxSecs(60), + }, + op_count: OpCount(20_000), + invariants: sim_invariants() + | InvariantSet::MONOTONIC_SEQ + | InvariantSet::FSYNC_ORDERING + | InvariantSet::TOMBSTONE_BOUND, + limits: RunLimits { + max_wall_ms: Some(WallMs(10 * 60_000)), + }, + restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(2_000)), + store: sim_store(), + eventlog: Some(EventLogConfig { + max_segment_size: MaxSegmentSize(64 * 1024), + }), + writer_concurrency: WriterConcurrency(1), + } +} + +fn contended_readers(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Simulated { + fault: FaultConfig::moderate(), + }, + workload: WorkloadModel { + weights: OpWeights { + add: 15, + delete: 1, + compact: 2, + checkpoint: 2, + read_record: 60, + read_block: 20, + ..OpWeights::default() + }, + size_distribution: SizeDistribution::Fixed(ValueBytes(128)), + collections: default_collections(), + key_space: KeySpaceSize(400), + did_space: DidSpaceSize(32), + retention_max_secs: RetentionMaxSecs(3600), + }, + op_count: OpCount(20_000), + invariants: sim_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(10 * 60_000)), + }, + restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(2_000)), + store: sim_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(64), + } +} + +fn contended_writers(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Simulated { + fault: FaultConfig::moderate(), + }, + workload: WorkloadModel { + weights: OpWeights { + add: 85, + delete: 5, + compact: 3, + checkpoint: 2, + read_record: 4, + read_block: 1, + ..OpWeights::default() + }, + size_distribution: SizeDistribution::Fixed(ValueBytes(128)), + collections: default_collections(), + key_space: KeySpaceSize(1_000), + did_space: DidSpaceSize(32), + retention_max_secs: RetentionMaxSecs(3600), + }, + op_count: OpCount(20_000), + invariants: sim_invariants(), + limits: RunLimits { + max_wall_ms: Some(WallMs(10 * 60_000)), + }, + restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(2_000)), + store: sim_store(), + eventlog: None, + writer_concurrency: WriterConcurrency(32), } }