test(store): gauntlet MST-repairable & misdirected-write scenario

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

crates/tranquil-store/src/bin/tranquil_gauntlet.rs

@@ -285,7 +285,9 @@ impl SweepAxisValues {
 impl SweepAxes {
     fn axis_values(&self) -> Vec<SweepAxisValues> {
         let base = vec![SweepAxisValues::default()];
-        let base = cross(base, &self.writer_concurrency, |a, v| a.writer_concurrency = Some(v));
+        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| {

crates/tranquil-store/src/gauntlet/invariants.rs

@@ -35,6 +35,7 @@ impl InvariantSet {
     pub const INDEX_BACKED_BY_DISK: Self = Self(1 << 13);
     pub const HINT_BACKED_BY_DATA: Self = Self(1 << 14);
     pub const INDEX_BLOCKS_READABLE: Self = Self(1 << 15);
+    pub const MST_REPAIRABLE: Self = Self(1 << 16);
 
     const ALL_KNOWN: u32 = Self::REFCOUNT_CONSERVATION.0
         | Self::REACHABILITY.0
@@ -51,7 +52,8 @@ impl InvariantSet {
         | Self::TOMBSTONE_BOUND.0
         | Self::INDEX_BACKED_BY_DISK.0
         | Self::HINT_BACKED_BY_DATA.0
-        | Self::INDEX_BLOCKS_READABLE.0;
+        | Self::INDEX_BLOCKS_READABLE.0
+        | Self::MST_REPAIRABLE.0;
 
     pub const fn contains(self, other: Self) -> bool {
         (self.0 & other.0) == other.0
@@ -293,6 +295,62 @@ impl<S: StorageIO + Send + Sync + 'static, C: Clock> Invariant<S, C> for ReadAft
     }
 }
 
+pub struct MstRepairable;
+
+#[async_trait]
+impl<S: StorageIO + Send + Sync + 'static, C: Clock> Invariant<S, C> for MstRepairable {
+    fn name(&self) -> &'static str {
+        "MstRepairable"
+    }
+
+    async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> {
+        let Some(root) = ctx.root else {
+            if ctx.oracle.live_count() == 0 {
+                return Ok(());
+            }
+            return Err(InvariantViolation {
+                invariant: "MstRepairable",
+                detail: format!(
+                    "oracle has {} live records but store has no root after repair",
+                    ctx.oracle.live_count()
+                ),
+            });
+        };
+        let mst = Mst::load(ctx.store.clone(), root, None);
+        let entries: Vec<(String, CidBytes)> = ctx
+            .oracle
+            .live_records()
+            .map(|(c, r, v)| (format!("{}/{}", c.0, r.0), *v))
+            .collect();
+
+        let mut violations: Vec<String> = Vec::new();
+        for (key, expected) in &entries {
+            match mst.get(key).await {
+                Ok(Some(cid)) => match try_cid_to_fixed(&cid) {
+                    Ok(actual) if actual == *expected => {}
+                    Ok(actual) => violations.push(format!(
+                        "{key}: MST cid {} != oracle cid {} after repair",
+                        hex_short(&actual),
+                        hex_short(expected),
+                    )),
+                    Err(e) => violations.push(format!("{key}: unexpected CID format: {e}")),
+                },
+                Ok(None) => violations.push(format!("{key}: MST returned None after repair")),
+                Err(e) => violations.push(format!("{key}: mst.get error after repair: {e}")),
+            }
+        }
+
+        if violations.is_empty() {
+            Ok(())
+        } else {
+            Err(InvariantViolation {
+                invariant: "MstRepairable",
+                detail: violations.join("; "),
+            })
+        }
+    }
+}
+
 pub struct CompactionIdempotent;
 
 #[async_trait]
@@ -926,6 +984,7 @@ pub fn invariants_for<S: StorageIO + Send + Sync + 'static, C: Clock>(
             Box::new(AckedWritePersistence),
         ),
         (InvariantSet::READ_AFTER_WRITE, Box::new(ReadAfterWrite)),
+        (InvariantSet::MST_REPAIRABLE, Box::new(MstRepairable)),
         (
             InvariantSet::COMPACTION_IDEMPOTENT,
             Box::new(CompactionIdempotent),

crates/tranquil-store/src/gauntlet/runner.rs

@@ -672,6 +672,14 @@ where
     }
 
     let end_of_run_set = config.invariants.without(InvariantSet::RESTART_IDEMPOTENT);
+    if !halt_ops
+        && config.invariants.contains(InvariantSet::MST_REPAIRABLE)
+        && let Some(live) = harness.as_ref()
+        && let Some(v) = attempt_structural_repair(&live.store, &oracle, root).await
+    {
+        violations.push(v);
+        halt_ops = true;
+    }
     if !halt_ops && let Some(live) = harness.as_ref() {
         match refresh_oracle_graph(&live.store, &mut oracle, root).await {
             Ok(()) => {
@@ -1024,6 +1032,36 @@ pub(super) async fn refresh_oracle_graph<S: StorageIO + Send + Sync + 'static, C
     }
 }
 
+async fn attempt_structural_repair<S: StorageIO + Send + Sync + 'static, C: Clock>(
+    store: &Arc<TranquilBlockStore<S, C>>,
+    oracle: &Oracle,
+    root: Option<Cid>,
+) -> Option<InvariantViolation> {
+    let r = root?;
+    let entries: Vec<(String, Cid)> = oracle
+        .live_records()
+        .filter_map(|(c, rk, v)| {
+            Cid::try_from(&v[..])
+                .ok()
+                .map(|cid| (format!("{}/{}", c.0, rk.0), cid))
+        })
+        .collect();
+    match crate::blockstore::rebuild_and_repair_mst(store, &entries, r).await {
+        Ok(outcome) => {
+            tracing::info!(
+                nodes_repaired = outcome.nodes_repaired,
+                nodes_total = outcome.nodes_total,
+                "structural repair complete"
+            );
+            None
+        }
+        Err(e) => Some(InvariantViolation {
+            invariant: "MstRepairFailed",
+            detail: e.to_string(),
+        }),
+    }
+}
+
 enum RestartAction {
     None,
     Clean,
@@ -1867,6 +1905,14 @@ where
     }
 
     let end_of_run_set = config.invariants.without(InvariantSet::RESTART_IDEMPOTENT);
+    if !halt_ops
+        && config.invariants.contains(InvariantSet::MST_REPAIRABLE)
+        && let Some(live) = harness.as_ref()
+        && let Some(v) = attempt_structural_repair(&live.store, &oracle, root).await
+    {
+        violations.push(v);
+        halt_ops = true;
+    }
     if !halt_ops && let Some(live) = harness.as_ref() {
         match refresh_oracle_graph(&live.store, &mut oracle, root).await {
             Ok(()) => {

crates/tranquil-store/src/gauntlet/scenarios.rs

@@ -26,6 +26,7 @@ pub enum Scenario {
     ModerateFaults,
     AggressiveFaults,
     TornPages,
+    MisdirectedWrites,
     Fsyncgate,
     FirehoseFanout,
     ContendedReaders,
@@ -50,6 +51,7 @@ impl Scenario {
             Self::ModerateFaults => "ModerateFaults",
             Self::AggressiveFaults => "AggressiveFaults",
             Self::TornPages => "TornPages",
+            Self::MisdirectedWrites => "MisdirectedWrites",
             Self::Fsyncgate => "Fsyncgate",
             Self::FirehoseFanout => "FirehoseFanout",
             Self::ContendedReaders => "ContendedReaders",
@@ -74,6 +76,7 @@ impl Scenario {
             Self::ModerateFaults => "moderate-faults",
             Self::AggressiveFaults => "aggressive-faults",
             Self::TornPages => "torn-pages",
+            Self::MisdirectedWrites => "misdirected-writes",
             Self::Fsyncgate => "fsyncgate",
             Self::FirehoseFanout => "firehose-fanout",
             Self::ContendedReaders => "contended-readers",
@@ -104,6 +107,7 @@ impl Scenario {
                 "Simulated IO with aggressive fault config. CrashAtSyscall restarts."
             }
             Self::TornPages => "Torn-page faults only, 20k ops.",
+            Self::MisdirectedWrites => "Misdirected-write faults only, 20k ops.",
             Self::Fsyncgate => "Fsync-drop faults only, 10k ops.",
             Self::FirehoseFanout => {
                 "Eventlog-heavy workload with FSYNC_ORDERING / MONOTONIC_SEQ / TOMBSTONE_BOUND invariants."
@@ -145,6 +149,7 @@ impl Scenario {
         Self::ModerateFaults,
         Self::AggressiveFaults,
         Self::TornPages,
+        Self::MisdirectedWrites,
         Self::Fsyncgate,
         Self::FirehoseFanout,
         Self::ContendedReaders,
@@ -220,6 +225,7 @@ pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig {
         Scenario::ModerateFaults => moderate_faults(seed),
         Scenario::AggressiveFaults => aggressive_faults(seed),
         Scenario::TornPages => torn_pages(seed),
+        Scenario::MisdirectedWrites => misdirected_writes(seed),
         Scenario::Fsyncgate => fsyncgate(seed),
         Scenario::FirehoseFanout => firehose_fanout(seed),
         Scenario::ContendedReaders => contended_readers(seed),
@@ -634,6 +640,26 @@ fn torn_pages(seed: Seed) -> GauntletConfig {
     }
 }
 
+fn misdirected_writes(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Simulated {
+            fault: FaultConfig::misdirected_only(),
+        },
+        workload: sim_microbench_workload(),
+        op_count: OpCount(20_000),
+        invariants: InvariantSet::MST_REPAIRABLE,
+        limits: RunLimits {
+            max_wall_ms: Some(WallMs(5 * 60_000)),
+        },
+        restart_policy: RestartPolicy::Never,
+        store: sim_store(),
+        eventlog: None,
+        writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
+    }
+}
+
 fn fsyncgate(seed: Seed) -> GauntletConfig {
     GauntletConfig {
         seed,

crates/tranquil-store/src/sim.rs

@@ -114,6 +114,13 @@ impl FaultConfig {
         }
     }
 
+    pub fn misdirected_only() -> Self {
+        Self {
+            misdirected_write_probability: Probability::new(0.25),
+            ..Self::none()
+        }
+    }
+
     pub fn fsyncgate_only() -> Self {
         Self {
             delayed_io_error_probability: Probability::new(0.05),