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"