Antithesis scenario demonstrates termination consistency

.gitignore

@@ -2,4 +2,6 @@
 .claude/*
 !.claude/skills/
 .docs-cache
-.beads/issues.jsonl
+.beads/issues.jsonl
+# antithesis-research scratchbook (working notes, never committed)
+scratchbook/

test/antithesis/.gitignore

@@ -1,2 +1,5 @@
 # Rendered compose files produced by bin/launch.sh at submit time.
 scenarios/*/.launch/
+
+# antithesis-research scratchbook: working notes, never committed.
+scratchbook/

test/antithesis/harness/Cargo.toml

@@ -13,6 +13,10 @@ doctest = false
 name = "first_sample_config"
 path = "src/bin/first_sample_config.rs"
 
+[[bin]]
+name = "anytime_capture_consistent"
+path = "src/bin/anytime_capture_consistent.rs"
+
 [lints]
 workspace = true
 
@@ -24,4 +28,11 @@ antithesis_sdk = { workspace = true, features = ["full", "rand_v0_10"] }
 byte-unit = { workspace = true, features = ["std"] }
 rand = { workspace = true, features = ["thread_rng", "std_rng"] }
 serde_yaml = { workspace = true }
+serde_json = { workspace = true }
 anyhow = { workspace = true }
+lading-capture = { path = "../../../lading_capture" }
+
+[dev-dependencies]
+uuid = { workspace = true }
+rustc-hash = { workspace = true }
+tempfile = { workspace = true }

test/antithesis/harness/src/bin/anytime_capture_consistent.rs

@@ -0,0 +1,84 @@
+//! Antithesis `anytime_` command: assert lading's capture is crash-consistent,
+//! for every capture format lading can emit (JSONL, Parquet, or Multi = both).
+//!
+//! node faults hard-kill lading (SIGKILL), so the whole lading container dies and
+//! only artifacts on the shared `capture` volume survive. This checker runs in the
+//! (never-faulted) workload container and fires whenever Antithesis chooses,
+//! including right after a kill/restart. It scans the capture directory and
+//! validates each file by its format:
+//!   * JSONL: a valid parseable prefix (a torn final line -- the interrupted
+//!     write -- is tolerated; a broken earlier line or a broken invariant is not).
+//!   * Parquet: footer-terminated, so a hard kill leaves it unreadable -- that is
+//!     expected, not corruption; but a *readable* Parquet must be consistent.
+//!   * Multi: both of the above, on `<base>.jsonl` and `<base>.parquet`.
+//!
+//! It always exits 0; findings are reported as named assertions.
+
+use std::fs;
+
+/// Directory lading writes capture files into. Overridable via `CAPTURE_DIR`.
+const DEFAULT_CAPTURE_DIR: &str = "/capture";
+/// Non-vacuity floor: prove the checker saw a substantive capture at least once
+/// across the run, so the always-invariants are not passing on empty files.
+const MIN_RECORDS: usize = 10;
+
+fn main() {
+    lading_antithesis::init();
+
+    let dir = std::env::var("CAPTURE_DIR").unwrap_or_else(|_| DEFAULT_CAPTURE_DIR.to_string());
+    let Ok(entries) = fs::read_dir(&dir) else {
+        // Capture directory not present yet this tick; nothing to check.
+        return;
+    };
+
+    let mut checked_any = false;
+    for entry in entries.flatten() {
+        let path = entry.path();
+        match path.extension().and_then(|e| e.to_str()) {
+            Some("jsonl") => {
+                let Ok(content) = fs::read_to_string(&path) else {
+                    continue;
+                };
+                let r = harness::capture::check_consistency(&content);
+                checked_any = true;
+                lading_antithesis::always!(
+                    r.torn_before_final == 0,
+                    "jsonl capture has no torn record before the final line",
+                    { "parsed": r.parsed, "torn_before_final": r.torn_before_final }
+                );
+                lading_antithesis::always!(
+                    r.invariants_hold,
+                    "jsonl capture fetch_index and per-series time stay monotonic",
+                    { "fetch_index_errors": r.fetch_index_errors, "per_series_errors": r.per_series_errors }
+                );
+                lading_antithesis::sometimes!(
+                    r.parsed >= MIN_RECORDS,
+                    "jsonl capture accumulated records across the run",
+                    { "parsed": r.parsed }
+                );
+            }
+            Some("parquet") => {
+                let r = harness::capture::check_parquet(&path);
+                checked_any = true;
+                // A readable Parquet must be internally consistent. An unreadable
+                // one is the expected result of a hard kill (footer only on clean
+                // close), so it is not a violation.
+                lading_antithesis::always!(
+                    !r.readable || r.invariants_hold,
+                    "readable parquet capture is internally consistent",
+                    { "readable": r.readable, "records": r.records }
+                );
+                lading_antithesis::sometimes!(
+                    r.readable && r.records >= MIN_RECORDS,
+                    "parquet capture finalized and readable across the run",
+                    { "readable": r.readable, "records": r.records }
+                );
+            }
+            _ => {}
+        }
+    }
+
+    if checked_any {
+        lading_antithesis::reachable!("capture consistency checker validated a capture file");
+    }
+}

test/antithesis/harness/src/capture.rs

@@ -0,0 +1,196 @@
+//! Crash-consistency checking for lading capture (JSONL) files.
+//!
+//! Under node faults lading is hard-killed (SIGKILL), possibly mid-write. A safe
+//! termination leaves the capture file a valid parseable prefix: a partial final
+//! line (the interrupted write) is tolerated, but any unparseable line before the
+//! final one, or any `fetch_index`/time invariant violation among the parsed
+//! records, is real corruption. Parsing and the invariant check reuse
+//! lading's own [`lading_capture`] `Line` and canonical `validate_lines`, so the
+//! oracle cannot drift from lading's real capture contract.
+
+use std::path::Path;
+
+use lading_capture::line::Line;
+use lading_capture::validate::jsonl::validate_lines;
+use lading_capture::validate::parquet::validate_parquet;
+
+/// Outcome of checking a capture file's crash-consistency.
+#[derive(Debug, Clone, Copy)]
+pub struct ConsistencyReport {
+    /// Complete lines that parsed as capture records.
+    pub parsed: usize,
+    /// Non-final lines that failed to parse. A torn final line is tolerated and
+    /// not counted here; anything here is real mid-file corruption.
+    pub torn_before_final: usize,
+    /// Whether the parsed records satisfy lading's capture invariants.
+    pub invariants_hold: bool,
+    /// `fetch_index`/time mapping violations among parsed records.
+    pub fetch_index_errors: u64,
+    /// Per-series (time / `fetch_index` monotonicity) violations.
+    pub per_series_errors: u64,
+}
+
+/// Check a capture file's contents for crash-consistency.
+#[must_use]
+pub fn check_consistency(content: &str) -> ConsistencyReport {
+    let raw: Vec<&str> = content.lines().collect();
+    let line_count = raw.len();
+    let mut parsed: Vec<Line> = Vec::new();
+    let mut torn_before_final = 0_usize;
+
+    for (idx, line) in raw.iter().enumerate() {
+        if line.trim().is_empty() {
+            continue;
+        }
+        match serde_json::from_str::<Line>(line) {
+            Ok(parsed_line) => parsed.push(parsed_line),
+            Err(_) => {
+                // A parse failure is real corruption only if it is not the final
+                // line; a torn final line is the tolerated interrupted write.
+                if idx + 1 != line_count {
+                    torn_before_final += 1;
+                }
+            }
+        }
+    }
+
+    let result = validate_lines(&parsed, None);
+    ConsistencyReport {
+        parsed: parsed.len(),
+        torn_before_final,
+        invariants_hold: result.is_valid(),
+        fetch_index_errors: u64::try_from(result.fetch_index_errors).unwrap_or(u64::MAX),
+        per_series_errors: u64::try_from(result.per_series_errors).unwrap_or(u64::MAX),
+    }
+}
+
+/// Outcome of checking a Parquet capture file.
+#[derive(Debug, Clone, Copy)]
+pub struct ParquetReport {
+    /// Whether the file was readable as Parquet (its footer is present). Parquet
+    /// writes the footer only on a clean close, so a hard kill mid-write leaves it
+    /// unreadable -- that is expected, not corruption.
+    pub readable: bool,
+    /// If readable, whether its capture invariants hold.
+    pub invariants_hold: bool,
+    /// If readable, the record count.
+    pub records: usize,
+}
+
+/// Validate a Parquet capture file. An unreadable file (missing footer) is
+/// reported via `readable: false` rather than an error, because an abrupt kill
+/// legitimately leaves Parquet without its footer; a *readable* Parquet, however,
+/// must satisfy the capture invariants.
+#[must_use]
+pub fn check_parquet(path: &Path) -> ParquetReport {
+    match validate_parquet(path, None) {
+        Ok(result) => ParquetReport {
+            readable: true,
+            invariants_hold: result.is_valid(),
+            records: usize::try_from(result.line_count).unwrap_or(usize::MAX),
+        },
+        Err(_) => ParquetReport {
+            readable: false,
+            invariants_hold: false,
+            records: 0,
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{check_consistency, check_parquet};
+    use lading_capture::line::{Line, LineValue, MetricKind};
+    use rustc_hash::FxHashMap;
+    use uuid::Uuid;
+
+    fn line(run_id: Uuid, time: u128, fetch_index: u64, metric: &str) -> Line {
+        Line {
+            run_id,
+            time,
+            fetch_index,
+            metric_name: metric.to_string(),
+            metric_kind: MetricKind::Counter,
+            value: LineValue::Int(fetch_index),
+            labels: FxHashMap::default(),
+            value_histogram: Vec::new(),
+        }
+    }
+
+    fn jsonl(lines: &[Line]) -> String {
+        lines
+            .iter()
+            .map(|l| serde_json::to_string(l).expect("serialize line"))
+            .collect::<Vec<_>>()
+            .join("\n")
+    }
+
+    fn valid_lines() -> Vec<Line> {
+        let run = Uuid::new_v4();
+        vec![
+            line(run, 1000, 0, "m.a"),
+            line(run, 2000, 1, "m.a"),
+            line(run, 3000, 2, "m.a"),
+        ]
+    }
+
+    #[test]
+    fn valid_capture_passes() {
+        let r = check_consistency(&jsonl(&valid_lines()));
+        assert_eq!(r.parsed, 3);
+        assert_eq!(r.torn_before_final, 0);
+        assert!(r.invariants_hold);
+    }
+
+    #[test]
+    fn torn_final_line_is_tolerated() {
+        let content = format!("{}\n{{\"run_id\":\"partial", jsonl(&valid_lines()));
+        let r = check_consistency(&content);
+        assert_eq!(r.parsed, 3, "the three complete lines still parse");
+        assert_eq!(r.torn_before_final, 0, "a partial final line is tolerated");
+        assert!(r.invariants_hold);
+    }
+
+    #[test]
+    fn torn_middle_line_is_corruption() {
+        let good = jsonl(&valid_lines());
+        let mut parts: Vec<&str> = good.lines().collect();
+        parts.insert(1, "{not valid json");
+        let content = parts.join("\n");
+        let r = check_consistency(&content);
+        assert!(
+            r.torn_before_final >= 1,
+            "a broken non-final line is flagged"
+        );
+    }
+
+    #[test]
+    fn non_monotonic_fetch_index_fails_invariants() {
+        let run = Uuid::new_v4();
+        // Same series, fetch_index goes backwards -> invariant violation.
+        let lines = vec![line(run, 1000, 5, "m.a"), line(run, 2000, 2, "m.a")];
+        let r = check_consistency(&jsonl(&lines));
+        assert!(!r.invariants_hold, "backwards fetch_index must fail");
+        assert!(r.per_series_errors >= 1 || r.fetch_index_errors >= 1);
+    }
+
+    #[test]
+    fn unreadable_parquet_is_reported_not_errored() {
+        use std::io::Write;
+        // A hard kill can leave a Parquet file without its footer. That is not a
+        // crash-consistency violation; check_parquet must report it as unreadable
+        // rather than error. (A well-formed Parquet's validity is covered by
+        // lading_capture's own validate::parquet tests.)
+        let mut f = tempfile::Builder::new()
+            .suffix(".parquet")
+            .tempfile()
+            .expect("temp file");
+        f.write_all(b"not a parquet footer").expect("write");
+        let r = check_parquet(f.path());
+        assert!(
+            !r.readable,
+            "a footer-less parquet is unreadable, not a crash"
+        );
+        assert!(!r.invariants_hold);
+    }
+}

test/antithesis/harness/src/lib.rs

@@ -6,4 +6,5 @@
 //! system-under-test boots from. The menu is built from lading's own
 //! `tcp::Config`, so it cannot drift from the real config schema.
 
+pub mod capture;
 pub mod config;

test/antithesis/scenarios/general/Dockerfile

@@ -72,9 +72,10 @@ RUN --mount=type=cache,target=/tools/target,id=antithesis-tools-target \
     --mount=type=cache,target=/root/.cargo/registry,id=cargo-registry \
     --mount=type=cache,target=/root/.cargo/git,id=cargo-git \
     cargo build --release --package sink --bin sink && \
-    cargo build --release --package harness --bin first_sample_config && \
+    cargo build --release --package harness --bin first_sample_config --bin anytime_capture_consistent && \
     cp /tools/target/release/sink /usr/local/bin/sink && \
-    cp /tools/target/release/first_sample_config /usr/local/bin/first_sample_config
+    cp /tools/target/release/first_sample_config /usr/local/bin/first_sample_config && \
+    cp /tools/target/release/anytime_capture_consistent /usr/local/bin/anytime_capture_consistent
 
 # ---------------------------------------------------------------------------
 # Runtime: instrumented lading (system under test).
@@ -121,4 +122,5 @@ COPY --chmod=755 test/antithesis/scenarios/general/workload/entrypoint.sh /entry
 # template structure; the compiled command binary is injected below.
 COPY --chmod=755 test/antithesis/scenarios/general/workload/test/ /opt/antithesis/test/
 COPY --from=tools-builder --chmod=755 /usr/local/bin/first_sample_config /opt/antithesis/test/v1/main/first_sample_config
+COPY --from=tools-builder --chmod=755 /usr/local/bin/anytime_capture_consistent /opt/antithesis/test/v1/main/anytime_capture_consistent
 ENTRYPOINT ["/entrypoint.sh"]

test/antithesis/scenarios/general/docker-compose.yaml

@@ -43,6 +43,9 @@ services:
       # first_sample_config (workload) writes this timeline's lading.yaml + ready
       # sentinel here; lading's entrypoint blocks on it, then boots under it.
       - shared:/shared:ro
+      # lading writes its capture here (rw); it is a named volume so it survives a
+      # node_termination of this container for the checker to validate.
+      - capture:/capture
     depends_on:
       sink:
         condition: service_healthy
@@ -64,10 +67,16 @@ services:
       CONFIG_DIR: "/shared"
     volumes:
       - shared:/shared
+      # Read-only: the checker (anytime_capture_consistent) validates lading's
+      # capture after node faults. workload is never faulted, so it survives.
+      - capture:/capture:ro
     depends_on:
       sink:
         condition: service_healthy
 
 volumes:
   # Carries the sampled lading.yaml + ready sentinel from workload to lading.
   shared:
+  # Carries lading's capture file; survives a node_termination of lading so the
+  # workload's checker can validate crash-consistency.
+  capture: