From c80a525e0dc9ffb20bb6b1af6bc6335f0b5e1533 Mon Sep 17 00:00:00 2001 From: Lewis Date: Thu, 16 Apr 2026 20:26:57 +0300 Subject: [PATCH] feat(tranquil-store/gauntlet): op surface, oracle, workload for eventlog & reads Lewis: May this revision serve well! --- crates/tranquil-store/src/gauntlet/mod.rs | 28 +++- crates/tranquil-store/src/gauntlet/op.rs | 158 ++++++++++++++++-- crates/tranquil-store/src/gauntlet/oracle.rs | 58 ++++++- .../tranquil-store/src/gauntlet/workload.rs | 108 ++++++++++-- 4 files changed, 321 insertions(+), 31 deletions(-) diff --git a/crates/tranquil-store/src/gauntlet/mod.rs b/crates/tranquil-store/src/gauntlet/mod.rs index 57c650a..8920601 100644 --- a/crates/tranquil-store/src/gauntlet/mod.rs +++ b/crates/tranquil-store/src/gauntlet/mod.rs @@ -2,19 +2,31 @@ pub mod farm; pub mod invariants; pub mod op; pub mod oracle; +pub mod overrides; +pub mod regression; pub mod runner; pub mod scenarios; +pub mod shrink; pub mod workload; -pub use invariants::{Invariant, InvariantSet, InvariantViolation, invariants_for}; -pub use op::{CollectionName, Op, OpStream, RecordKey, Seed, ValueSeed}; -pub use oracle::Oracle; +pub use invariants::{ + EventLogSnapshot, Invariant, InvariantSet, InvariantViolation, SnapshotEvent, invariants_for, +}; +pub use op::{ + CollectionName, DidSeed, EventKind, Op, OpStream, PayloadSeed, RecordKey, RetentionSecs, Seed, + ValueSeed, +}; +pub use oracle::{EventExpectation, Oracle}; +pub use overrides::{ConfigOverrides, GroupCommitOverrides, StoreOverrides}; +pub use regression::{RegressionRecord, RegressionViolation, default_root as regression_root}; pub use runner::{ - Gauntlet, GauntletBuildError, GauntletConfig, GauntletReport, IoBackend, MaxFileSize, OpIndex, - OpInterval, OpsExecuted, RestartCount, RestartPolicy, RunLimits, ShardCount, StoreConfig, - WallMs, + EventLogConfig, Gauntlet, GauntletBuildError, GauntletConfig, GauntletReport, Harness, + IoBackend, MaxFileSize, MaxSegmentSize, OpErrorCount, OpIndex, OpInterval, OpsExecuted, + RestartCount, RestartPolicy, RunLimits, ShardCount, StoreConfig, WallMs, WriterConcurrency, }; -pub use scenarios::{Scenario, config_for}; +pub use scenarios::{Scenario, UnknownScenario, config_for}; +pub use shrink::{ShrinkOutcome, shrink_failure}; pub use workload::{ - ByteRange, KeySpaceSize, OpCount, OpWeights, SizeDistribution, ValueBytes, WorkloadModel, + ByteRange, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, SizeDistribution, + ValueBytes, WorkloadModel, }; diff --git a/crates/tranquil-store/src/gauntlet/op.rs b/crates/tranquil-store/src/gauntlet/op.rs index 6645ab4..0c047f4 100644 --- a/crates/tranquil-store/src/gauntlet/op.rs +++ b/crates/tranquil-store/src/gauntlet/op.rs @@ -1,16 +1,35 @@ -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct Seed(pub u64); -#[derive(Debug, Clone, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct CollectionName(pub String); -#[derive(Debug, Clone, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct RecordKey(pub String); -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct ValueSeed(pub u32); -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct DidSeed(pub u32); + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct PayloadSeed(pub u32); + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct RetentionSecs(pub u32); + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub enum EventKind { + Commit, + Identity, + Account, + Sync, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum Op { AddRecord { collection: CollectionName, @@ -23,9 +42,31 @@ pub enum Op { }, Compact, Checkpoint, + AppendEvent { + did_seed: DidSeed, + event_kind: EventKind, + payload_seed: PayloadSeed, + }, + SyncEventLog, + RunRetention { + max_age_secs: RetentionSecs, + }, + ReadRecord { + collection: CollectionName, + rkey: RecordKey, + }, + ReadBlock { + value_seed: ValueSeed, + }, } -#[derive(Debug, Clone)] +impl Op { + pub const fn is_read_only(&self) -> bool { + matches!(self, Op::ReadRecord { .. } | Op::ReadBlock { .. }) + } +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct OpStream { ops: Vec, } @@ -35,6 +76,14 @@ impl OpStream { Self { ops } } + pub fn empty() -> Self { + Self { ops: Vec::new() } + } + + pub fn as_slice(&self) -> &[Op] { + &self.ops + } + pub fn into_vec(self) -> Vec { self.ops } @@ -51,10 +100,99 @@ impl OpStream { self.ops.is_empty() } - pub fn shrink(&self) -> Option { - (self.ops.len() >= 2).then(|| { - let half = self.ops.len() / 2; - OpStream::from_vec(self.ops[..half].to_vec()) + pub fn shrink_candidates(&self) -> impl Iterator + '_ { + let len = self.ops.len(); + let chunk_sizes: Vec = std::iter::successors((len >= 2).then_some(len / 2), |&s| { + (s >= 2).then_some(s / 2) }) + .collect(); + + let chunk_candidates = chunk_sizes.into_iter().flat_map(move |chunk_size| { + let count = len.div_ceil(chunk_size); + (0..count).map(move |i| { + let start = i * chunk_size; + let end = (start + chunk_size).min(len); + let mut reduced = Vec::with_capacity(len - (end - start)); + reduced.extend_from_slice(&self.ops[..start]); + reduced.extend_from_slice(&self.ops[end..]); + OpStream::from_vec(reduced) + }) + }); + + let single_candidates = (0..len).map(move |i| { + let mut reduced = self.ops.clone(); + reduced.remove(i); + OpStream::from_vec(reduced) + }); + + chunk_candidates.chain(single_candidates) + } + + pub fn shrink_to_fixpoint(mut self, mut fails: impl FnMut(&OpStream) -> bool) -> OpStream { + loop { + let next = self.shrink_candidates().find(|c| !c.is_empty() && fails(c)); + match next { + Some(smaller) => self = smaller, + None => return self, + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn stream(n: usize) -> OpStream { + OpStream::from_vec( + (0..n) + .map(|i| Op::AddRecord { + collection: CollectionName("c".into()), + rkey: RecordKey(format!("{i:04}")), + value_seed: ValueSeed(i as u32), + }) + .collect(), + ) + } + + fn contains_index(s: &OpStream, target: u32) -> bool { + s.iter() + .any(|op| matches!(op, Op::AddRecord { value_seed, .. } if value_seed.0 == target)) + } + + #[test] + fn shrink_candidates_nonempty_for_len_ge_2() { + let s = stream(8); + let count = s.shrink_candidates().count(); + assert!(count > 0); + } + + #[test] + fn shrink_candidates_empty_for_len_0() { + let s = OpStream::from_vec(Vec::new()); + assert_eq!(s.shrink_candidates().count(), 0); + } + + #[test] + fn shrink_candidates_includes_every_single_removal() { + let s = stream(5); + let singles: Vec<_> = s.shrink_candidates().filter(|c| c.len() == 4).collect(); + assert!( + singles.len() >= 5, + "expected at least 5 size-4 candidates, got {}", + singles.len() + ); + } + + #[test] + fn shrink_to_fixpoint_converges_to_culprit() { + let s = stream(64); + let shrunk = s.shrink_to_fixpoint(|c| contains_index(c, 17)); + assert!(contains_index(&shrunk, 17)); + assert!( + shrunk.len() < 4, + "expected shrink to close on culprit, got {} ops", + shrunk.len() + ); } } diff --git a/crates/tranquil-store/src/gauntlet/oracle.rs b/crates/tranquil-store/src/gauntlet/oracle.rs index 0d80567..08beaa2 100644 --- a/crates/tranquil-store/src/gauntlet/oracle.rs +++ b/crates/tranquil-store/src/gauntlet/oracle.rs @@ -2,8 +2,9 @@ use std::collections::HashMap; use cid::Cid; -use super::op::{CollectionName, RecordKey}; +use super::op::{CollectionName, EventKind, RecordKey}; use crate::blockstore::CidBytes; +use crate::eventlog::EventSequence; #[derive(Debug, thiserror::Error, PartialEq, Eq)] #[error("unexpected CID encoding: got {actual} bytes, expected 36 for sha256 CIDv1")] @@ -11,11 +12,23 @@ pub struct CidFormatError { pub actual: usize, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EventExpectation { + pub seq: EventSequence, + pub timestamp_us: u64, + pub kind: EventKind, + pub did_hash: u32, +} + #[derive(Debug, Default)] pub struct Oracle { live: HashMap<(CollectionName, RecordKey), CidBytes>, current_root: Option, mst_node_cids: Vec, + synced_events: Vec, + unsynced_events: Vec, + last_synced_seq: Option, + last_retention_cutoff_us: Option, } impl Oracle { @@ -36,6 +49,10 @@ impl Oracle { self.live.remove(&(coll.clone(), rkey.clone())) } + pub fn contains_record(&self, coll: &CollectionName, rkey: &RecordKey) -> bool { + self.live.contains_key(&(coll.clone(), rkey.clone())) + } + pub fn set_root(&mut self, root: Cid) { self.current_root = Some(root); } @@ -71,6 +88,45 @@ impl Oracle { .map(|(c, r, v)| (format!("record {}/{}", c.0, r.0), *v)); nodes.chain(records).collect() } + + pub fn record_event_append(&mut self, event: EventExpectation) { + self.unsynced_events.push(event); + } + + pub fn record_event_sync(&mut self, synced_through: EventSequence) { + let (promoted, remaining): (Vec<_>, Vec<_>) = self + .unsynced_events + .drain(..) + .partition(|e| e.seq <= synced_through); + self.synced_events.extend(promoted); + self.unsynced_events = remaining; + self.last_synced_seq = Some(synced_through); + } + + pub fn record_crash(&mut self) { + self.unsynced_events.clear(); + } + + pub fn record_retention(&mut self, cutoff_us: u64) { + self.synced_events.retain(|e| e.timestamp_us >= cutoff_us); + self.last_retention_cutoff_us = Some(cutoff_us); + } + + pub fn synced_events(&self) -> &[EventExpectation] { + &self.synced_events + } + + pub fn unsynced_events(&self) -> &[EventExpectation] { + &self.unsynced_events + } + + pub fn last_synced_seq(&self) -> Option { + self.last_synced_seq + } + + pub fn last_retention_cutoff_us(&self) -> Option { + self.last_retention_cutoff_us + } } pub(super) fn try_cid_to_fixed(cid: &Cid) -> Result { diff --git a/crates/tranquil-store/src/gauntlet/workload.rs b/crates/tranquil-store/src/gauntlet/workload.rs index 9277d53..a08fb85 100644 --- a/crates/tranquil-store/src/gauntlet/workload.rs +++ b/crates/tranquil-store/src/gauntlet/workload.rs @@ -1,4 +1,7 @@ -use super::op::{CollectionName, Op, OpStream, RecordKey, Seed, ValueSeed}; +use super::op::{ + CollectionName, DidSeed, EventKind, Op, OpStream, PayloadSeed, RecordKey, RetentionSecs, Seed, + ValueSeed, +}; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct ValueBytes(pub u32); @@ -9,17 +12,34 @@ pub struct KeySpaceSize(pub u32); #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct OpCount(pub usize); -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, Default)] pub struct OpWeights { pub add: u32, pub delete: u32, pub compact: u32, pub checkpoint: u32, + pub append_event: u32, + pub sync_event_log: u32, + pub run_retention: u32, + pub read_record: u32, + pub read_block: u32, } impl OpWeights { pub const fn total(&self) -> u32 { - self.add + self.delete + self.compact + self.checkpoint + self.add + + self.delete + + self.compact + + self.checkpoint + + self.append_event + + self.sync_event_log + + self.run_retention + + self.read_record + + self.read_block + } + + pub const fn touches_eventlog(&self) -> bool { + self.append_event > 0 || self.sync_event_log > 0 || self.run_retention > 0 } } @@ -51,14 +71,46 @@ impl ByteRange { pub enum SizeDistribution { Fixed(ValueBytes), Uniform(ByteRange), + HeavyTail(ByteRange), } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct DidSpaceSize(pub u32); + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct RetentionMaxSecs(pub u32); + #[derive(Debug, Clone)] pub struct WorkloadModel { pub weights: OpWeights, pub size_distribution: SizeDistribution, pub collections: Vec, pub key_space: KeySpaceSize, + pub did_space: DidSpaceSize, + pub retention_max_secs: RetentionMaxSecs, +} + +impl Default for WorkloadModel { + fn default() -> Self { + Self { + weights: OpWeights { + add: 80, + delete: 10, + compact: 5, + checkpoint: 5, + append_event: 0, + sync_event_log: 0, + run_retention: 0, + read_record: 0, + read_block: 0, + }, + size_distribution: SizeDistribution::Fixed(ValueBytes(64)), + collections: vec![CollectionName("app.bsky.feed.post".to_string())], + key_space: KeySpaceSize(200), + did_space: DidSpaceSize(32), + retention_max_secs: RetentionMaxSecs(3600), + } + } } impl WorkloadModel { @@ -77,23 +129,46 @@ impl WorkloadModel { let coll = self.collections[rng.next_usize() % self.collections.len()].clone(); let rkey = RecordKey(format!("{:06}", rng.next_u32() % self.key_space.0.max(1))); - let (a, d, c) = ( - self.weights.add, - self.weights.add + self.weights.delete, - self.weights.add + self.weights.delete + self.weights.compact, - ); + let w = &self.weights; + let t1 = w.add; + let t2 = t1 + w.delete; + let t3 = t2 + w.compact; + let t4 = t3 + w.checkpoint; + let t5 = t4 + w.append_event; + let t6 = t5 + w.sync_event_log; + let t7 = t6 + w.run_retention; + let t8 = t7 + w.read_record; + match bucket { - b if b < a => Op::AddRecord { + b if b < t1 => Op::AddRecord { collection: coll, rkey, value_seed: ValueSeed(rng.next_u32()), }, - b if b < d => Op::DeleteRecord { + b if b < t2 => Op::DeleteRecord { collection: coll, rkey, }, - b if b < c => Op::Compact, - _ => Op::Checkpoint, + b if b < t3 => Op::Compact, + b if b < t4 => Op::Checkpoint, + b if b < t5 => Op::AppendEvent { + did_seed: DidSeed(rng.next_u32() % self.did_space.0.max(1)), + event_kind: event_kind_for(rng.next_u32()), + payload_seed: PayloadSeed(rng.next_u32()), + }, + b if b < t6 => Op::SyncEventLog, + b if b < t7 => Op::RunRetention { + max_age_secs: RetentionSecs( + rng.next_u32() % self.retention_max_secs.0.max(1), + ), + }, + b if b < t8 => Op::ReadRecord { + collection: coll, + rkey, + }, + _ => Op::ReadBlock { + value_seed: ValueSeed(rng.next_u32()), + }, } }) .collect(); @@ -101,6 +176,15 @@ impl WorkloadModel { } } +fn event_kind_for(n: u32) -> EventKind { + match n & 0b11 { + 0 => EventKind::Commit, + 1 => EventKind::Identity, + 2 => EventKind::Account, + _ => EventKind::Sync, + } +} + pub struct Lcg { state: u64, }