feat(tranquil-store/gauntlet): op surface, oracle, workload for eventlog & reads

Lewis: May this revision serve well! <lu5a@proton.me>

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,
 };

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<Op>,
 }
@@ -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<Op> {
         self.ops
     }
@@ -51,10 +100,99 @@ impl OpStream {
         self.ops.is_empty()
     }
 
-    pub fn shrink(&self) -> Option<OpStream> {
-        (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<Item = OpStream> + '_ {
+        let len = self.ops.len();
+        let chunk_sizes: Vec<usize> = 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()
+        );
     }
 }

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<Cid>,
     mst_node_cids: Vec<CidBytes>,
+    synced_events: Vec<EventExpectation>,
+    unsynced_events: Vec<EventExpectation>,
+    last_synced_seq: Option<EventSequence>,
+    last_retention_cutoff_us: Option<u64>,
 }
 
 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<EventSequence> {
+        self.last_synced_seq
+    }
+
+    pub fn last_retention_cutoff_us(&self) -> Option<u64> {
+        self.last_retention_cutoff_us
+    }
 }
 
 pub(super) fn try_cid_to_fixed(cid: &Cid) -> Result<CidBytes, CidFormatError> {

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<CollectionName>,
     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,
 }