diff --git a/.gitignore b/.gitignore
index c0db42dfe..b81325344 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,4 +2,6 @@
.claude/*
!.claude/skills/
.docs-cache
-.beads/issues.jsonl
\ No newline at end of file
+.beads/issues.jsonl
+# antithesis-research scratchbook (working notes, never committed)
+scratchbook/
diff --git a/Cargo.lock b/Cargo.lock
index 3068fad1c..82e140e6f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1374,9 +1374,14 @@ dependencies = [
"byte-unit",
"lading",
"lading-antithesis",
+ "lading-capture",
"lading-payload",
"rand 0.10.1",
+ "rustc-hash",
+ "serde_json",
"serde_yaml",
+ "tempfile",
+ "uuid",
]
[[package]]
diff --git a/test/antithesis/.gitignore b/test/antithesis/.gitignore
index eeee66e4a..922cbd44b 100644
--- a/test/antithesis/.gitignore
+++ b/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/
diff --git a/test/antithesis/harness/Cargo.toml b/test/antithesis/harness/Cargo.toml
index 2da18465d..56711f508 100644
--- a/test/antithesis/harness/Cargo.toml
+++ b/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 }
diff --git a/test/antithesis/harness/src/bin/anytime_capture_consistent.rs b/test/antithesis/harness/src/bin/anytime_capture_consistent.rs
new file mode 100644
index 000000000..77f868a8e
--- /dev/null
+++ b/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 `.jsonl` and `.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");
+ }
+}
diff --git a/test/antithesis/harness/src/capture.rs b/test/antithesis/harness/src/capture.rs
new file mode 100644
index 000000000..e3e2533ac
--- /dev/null
+++ b/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 = 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) {
+ 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::>()
+ .join("\n")
+ }
+
+ fn valid_lines() -> Vec {
+ 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);
+ }
+}
diff --git a/test/antithesis/harness/src/lib.rs b/test/antithesis/harness/src/lib.rs
index 383d0fc81..ffc38fa33 100644
--- a/test/antithesis/harness/src/lib.rs
+++ b/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;
diff --git a/test/antithesis/scenarios/general/Dockerfile b/test/antithesis/scenarios/general/Dockerfile
index c867ccfea..614682efa 100644
--- a/test/antithesis/scenarios/general/Dockerfile
+++ b/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"]
diff --git a/test/antithesis/scenarios/general/docker-compose.yaml b/test/antithesis/scenarios/general/docker-compose.yaml
index a6b99d782..6eea8063d 100644
--- a/test/antithesis/scenarios/general/docker-compose.yaml
+++ b/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,6 +67,9 @@ 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
@@ -71,3 +77,6 @@ services:
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:
diff --git a/test/antithesis/scenarios/general/lading-entrypoint.sh b/test/antithesis/scenarios/general/lading-entrypoint.sh
index 7e0202483..35316187f 100644
--- a/test/antithesis/scenarios/general/lading-entrypoint.sh
+++ b/test/antithesis/scenarios/general/lading-entrypoint.sh
@@ -14,8 +14,13 @@ while [ ! -f "${CONFIG_DIR}/ready" ]; do
done
echo "lading: config ready, starting" >&2
+# --capture-path satisfies lading's telemetry requirement and writes the capture
+# the anytime_capture_consistent checker validates for crash-consistency after
+# node faults. flush every 1s so a kill has a fresh file and many kill-in-flight
+# opportunities. Capture lives on its own volume so it survives the killed container.
exec /usr/local/bin/lading \
--no-target \
--experiment-duration-infinite \
- --prometheus-addr 0.0.0.0:9102 \
+ --capture-path /capture/capture.jsonl \
+ --capture-flush-seconds 1 \
--config-path "${CONFIG_DIR}/lading.yaml"