diff --git a/CHANGELOG.md b/CHANGELOG.md index de96ca6e..bd476080 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,72 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added — cross-plane kill-9 crash-matrix suite (kernel M3 stage 1 / G1) + +New `tests/crash_matrix_cross_plane.rs` + `tests/crash_matrix_cross_plane/` +harness: 40 kill-9 crash-recovery cells spanning every persistence plane +(KV/AOF, KV disk-offload, graph/vector/WS/MQ WAL v3, cross-store MULTI/EXEC, +mixed-plane concurrent workloads, and a checkpoint-`Finalize`-window kill) +across the `{shards=1, shards=4} x {appendonly, disk-offload}` matrix. This +is a RED-first tripwire suite — 29 cells run GREEN by default; 11 known-RED +cells are gated behind a runtime guard (`harness::red_guard`, checked via +`MOON_CRASH_MATRIX_RED=1`) rather than `#[ignore = "RED: ..."]`, because the +suite's own `--ignored` invocation convention (every cell needs a release +binary) makes an ignore-reason string alone non-functional as a gate. A +`MOON_CRASH_MATRIX_ITERS` env var (default 1) enables ad-hoc soak runs for +probabilistic findings. + +A second review round caught a vacuous-precondition bug and a gating +granularity bug before this suite could be trusted as a tripwire: +`kv_spilled_isolated`'s filler (3000×512B against a 4 MiB cap) never +actually forced a cold-tier spill at either shard count, silently +degenerating to the same coverage as `kv_isolated` — fixed by mirroring +`crash_recovery_cold_del_resurrection.rs`'s proven filler ratios +(16,000×600B against 8 MiB) plus a hard `heap-*.mpf` precondition assert so +a future load-parameter regression fails loud instead of vacuously passing. +`txn_isolated` was split into `txn_isolated_committed` (still RED, task #52 +below) and `txn_isolated_atomicity` (a transaction queued but killed before +`EXEC` must apply nothing) — the two halves have unrelated root causes and +sharing one `red_guard` hid a working, GREEN regression tripwire on +`prod_s1`/`prod_s4` by default; the atomicity half now runs ungated on +those two configs (still RED on `legacy_yes_s1`, same pre-existing +legacy-graph gap as everything else there). + +Two real, unresolved findings surfaced and are intentionally left RED (not +fixed — that is out of scope for this stage) for the kernel M3 backlog: + +- **Cross-store TXN graph leg not durable (task #52):** a committed + `MULTI`/`EXEC` mixing `SET` + `GRAPH.ADDNODE` survives kill-9 on the KV + leg (PR #247's `persist_txn_aof`) but the graph leg is lost. Deterministic + repro (3x consecutive at shards=1 and shards=4): kill immediately after + the post-`EXEC` WAL-v3 sync wait, with `--checkpoint-timeout 3600` so no + periodic checkpoint can incidentally rescue the graph leg. See + `scenarios::txn_isolated_committed`'s doc comment for the minimal repro. +- **Checkpoint-`Finalize` window can total-loss the graph plane (new):** + some kill offsets in the 0-150ms window after `BGSAVE` lose the *entire* + already-wal-v3-synced graph batch (not just an unsynced tail) while + KV/vector/WS/MQ all survive in the same run. Probabilistic + (~1-in-7 to 1-in-12 per iteration) — a single default run can report + false-green even with `MOON_CRASH_MATRIX_RED=1`; reproduce reliably with + `MOON_CRASH_MATRIX_RED=1 MOON_CRASH_MATRIX_ITERS=20`. Strong P0 candidate + for kernel M3 stage 2 (K2). See `scenarios::mixed_mid_checkpoint`'s doc + comment. + +Also confirmed GREEN (not RED, contrary to the brief's grouping at the +design-doc level): `WS DROP` durability under kill-9 — `WorkspaceDrop` WAL +records replay correctly in order, unlike the sibling MQ generic-`DEL` +resurrection gap (still RED, same bug class as the already-fixed KV/vector +cold-plane resurrection, PR #257, not yet applied to MQ). + +`tests/common/mod.rs` gains `find_moon_binary()`/`sigkill()`/ +`wait_for_port_down()` shared helpers this suite depends on (the latter now +panics on loop exhaustion instead of silently returning — a tripwire +codebase should not have silent-pass verification helpers). Shared helpers +added; migration of the 5 existing `crash_recovery_*.rs` suites' own local +copies to the shared versions is deferred, not done in this stage. + +Test-only; no production code changed in this stage. + ### Added — memory + tier accounting spine (kernel M2 stage 2 / K4) `resident_bytes()` is now implemented by every storage plane, and the diff --git a/tests/common/mod.rs b/tests/common/mod.rs index febbf368..841d14fd 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -25,6 +25,7 @@ use std::collections::HashSet; use std::net::TcpStream; +use std::path::PathBuf; use std::process::Child; use std::sync::{LazyLock, Mutex}; use std::time::{Duration, Instant}; @@ -108,3 +109,104 @@ pub fn spawn_listening(mut spawn: impl FnMut(u16) -> Child) -> (Child, u16) { not a port race; read the server stderr log in the test's --dir" ); } + +// --------------------------------------------------------------------------- +// Kernel M3 / G1 (task #18 follow-up): consolidated binary-resolution + kill +// helpers. Every crash suite used to carry its own copy of these three +// functions (grepped: `crash_recovery_graph_durability.rs`, +// `crash_recovery_wal_recycle_legacy.rs`, `crash_recovery_vector_durability.rs` +// duplicated all three; `crash_recovery_mq_effects.rs` / +// `crash_recovery_temporal_mq.rs` already used `spawn_listening` above but +// still kept a local `find_moon_binary`). Consolidated here as the mechanical, +// behavior-preserving union of every variant found (MOON_BIN override with a +// non-empty guard, then the compile-time `CARGO_BIN_EXE_moon` path Cargo sets +// for this test binary, then `target/{release,debug}/moon` as a last resort +// for ad-hoc `cargo test --test ` invocations outside the normal +// harness). `tests/aof_multidb_kill9.rs`'s `moon_bin()` returns a bare +// `./target/release/moon` default with NO binary-existence check and no +// `CARGO_BIN_EXE_moon` fallback — a real behavioral difference, so it is +// deliberately left alone here rather than force-migrated (see the kernel M3 +// brief, Stage 1, decision #5: "if any suite needs behavioral adaptation, +// skip it there and note it for a follow-up task instead"). +// --------------------------------------------------------------------------- + +/// Resolve the `moon` server binary for crash/integration suites. +/// +/// Precedence: `MOON_BIN` env var (only if non-empty and the path exists) → +/// `CARGO_BIN_EXE_moon` (the exact binary Cargo built for THIS test run — +/// right profile, right `CARGO_TARGET_DIR`, right `.exe` suffix on Windows) +/// → `target/release/moon` → `target/debug/moon`. Panics with a actionable +/// message if none resolve. +pub fn find_moon_binary() -> PathBuf { + if let Ok(bin) = std::env::var("MOON_BIN") + && !bin.trim().is_empty() + { + let p = PathBuf::from(&bin); + if p.exists() { + return p; + } + } + let cargo_bin = PathBuf::from(env!("CARGO_BIN_EXE_moon")); + if cargo_bin.exists() { + return cargo_bin; + } + let manifest = env!("CARGO_MANIFEST_DIR"); + let release = PathBuf::from(format!("{manifest}/target/release/moon")); + if release.exists() { + return release; + } + let debug = PathBuf::from(format!("{manifest}/target/debug/moon")); + if debug.exists() { + return debug; + } + panic!( + "No moon binary found. Build with `cargo build --release` or set \ + MOON_BIN=/path/to/moon." + ); +} + +/// SIGKILL a spawned child and reap it (never SIGTERM — SIGTERM + +/// SO_REUSEPORT is a documented hang, see CLAUDE.md / the harness-speed +/// gotcha ledger). `Child::kill()` is documented to send SIGKILL on Unix, +/// so no raw `libc::kill` (and no `unsafe`, no cfg split) is needed. +pub fn sigkill(child: &mut Child) { + let _ = child.kill(); + let _ = child.wait(); +} + +/// Wait until nothing is accepting on `port` — required before a same-port +/// restart, or the new listener can race the dying process's socket +/// teardown. moon binds `SO_REUSEPORT` per shard, so a plain bind-based +/// check is useless here; two consecutive refused connects is the signal. +pub fn wait_for_port_down(port: u16) { + let addr = format!("127.0.0.1:{port}"); + let mut consecutive_refused = 0; + for _ in 0..120 { + match TcpStream::connect_timeout(&addr.parse().expect("addr"), Duration::from_millis(100)) { + Ok(_) => { + consecutive_refused = 0; + std::thread::sleep(Duration::from_millis(100)); + } + Err(_) => { + consecutive_refused += 1; + if consecutive_refused >= 2 { + return; + } + std::thread::sleep(Duration::from_millis(50)); + } + } + } + // No silent-pass verification helpers in a tripwire codebase (review + // round 3, P2): a caller that proceeds to bind the SAME port after this + // returns without the port actually being down races the dying + // process's socket teardown, which can manifest as a flaky bind failure + // or — worse — a same-port respawn silently talking to the wrong + // (still-dying) process. Loop exhaustion after 120 iterations (~18s + // worst case) is not a benign timeout here; it means the port never + // went down and every caller's assumption is false. + panic!( + "wait_for_port_down: port {port} never stopped accepting connections \ + after 120 poll iterations (~18s) — the old process may still be \ + alive, or SO_REUSEPORT is masking a listener that never exited" + ); +} diff --git a/tests/crash_matrix_cross_plane.rs b/tests/crash_matrix_cross_plane.rs new file mode 100644 index 00000000..22ac8210 --- /dev/null +++ b/tests/crash_matrix_cross_plane.rs @@ -0,0 +1,244 @@ +//! Kernel M3 / G1: cross-plane kill-9 crash-matrix, RED-first. +//! +//! Execution brief: `.planning/reviews/kernel-m3-brief-2026-07-12.md` +//! (Stage 1). This suite is the tripwire for Stage 2 (K2 — unified +//! per-shard floor register): "no data loss, provably, across every plane" +//! is an empty claim until something checks it, and every prior kernel +//! milestone shipped the failing test FIRST, then fixed it — never the +//! other way around (task #41/#42/#43's history). **Stage 1 does not fix +//! anything** — it exists to produce an honest RED/GREEN inventory before +//! K2 touches the recycle-floor invariant. +//! +//! Internal naming: `cross_plane__` (NOT a `g1_` prefix — +//! `tests/crash_recovery_graph_durability.rs` already has its own internal +//! `g1_.../g5_...` scenario numbering; a second, differently-scoped "G1" in +//! a different file with the same prefix would confuse `cargo test g1` +//! filtering and any future grep — brief §Stage 1). +//! +//! # Axes +//! +//! - **Plane:** KV (incl. spilled collections), graph (structural + +//! GraphTemporal via `TEMPORAL.INVALIDATE`), vector, WS, MQ (incl. +//! DLQ/PEL/triggers), cross-store TXN (`MULTI`/`EXEC` spanning ≥2 planes). +//! FTS/TEMPORAL(node upsert) are WAL-recycle non-participants per the +//! brief §1.2 and are out of scope here. +//! - **Workload:** (a) single-plane isolated writes (regression baseline) +//! and (b) concurrent mixed writes across ALL planes on the SAME shard +//! (`tests/crash_matrix_cross_plane/mixed.rs` — the genuinely new +//! coverage; no existing suite does this). +//! - **Kill point:** after a synced marker at a deterministic or randomized +//! offset in the write stream (`kill_after_marker`, the baseline pattern +//! from `crash_recovery_graph_durability.rs`); mid-checkpoint-Finalize +//! (`kill_mid_checkpoint` — see the honesty note below); mid-Pass-C-recycle +//! (`kill_mid_pass_c`, tiny `--wal-segment-size`/`--max-wal-size`); mid a +//! cross-store `MULTI` (`kill_mid_txn` — kill before `EXEC`, proving +//! atomicity: nothing queued must apply); genuinely-concurrent mid-burst +//! (`mixed::start_concurrent_burst`, no synchronization at all). +//! +//! **Honesty note on `kill_mid_checkpoint`:** this codebase has no +//! fault-injection hooks for "kill exactly at Finalize step N" (verified — +//! `src/shard/persistence_tick.rs`'s `CheckpointAction::Finalize` arm has +//! no test-only interrupt points). What this suite does instead is +//! probabilistic: trigger `BGSAVE`, then kill after a short randomized +//! sleep drawn across the window a small checkpoint typically completes +//! in. Repeated / soaked runs (`MOON_CRASH_MATRIX_ITERS`) sample different +//! points in that window; a single run is NOT a proof any specific +//! Finalize step (1-6) was hit. This is a real coverage gap relative to +//! the brief's phrasing ("each individually") — flagged explicitly rather +//! than silently claimed. +//! +//! - **Config:** 8 cells = `appendonly {yes,no}` × `disk-offload +//! {enable,disable}` × `shards {1,4}`. Curated per the brief: full +//! plane×workload matrix on the 2 production cells (`PROD_S1`, `PROD_S4`) +//! and the 2 legacy cells at shards=1 only (`LEGACY_YES_S1`, +//! `LEGACY_NO_S1`); 1-2 representative spot-checks on the remaining 4 +//! cells (`tests_spot.rs`). `appendonly=no` cells have NO durability +//! contract on ANY plane (not just KV — `persistence_dir`, and therefore +//! the WAL writer, is only constructed when `appendonly=="yes" || +//! save.is_some()`), so those cells assert liveness/no-corruption only. +//! +//! # RED/GREEN cell inventory +//! +//! Final state after 7 full runs on the Linux VM (`orb run -m moon-dev`, +//! ELF release binary) and three coordinator review rounds that caught 12 +//! harness bugs total (wrong durability-log-per-plane assumptions, +//! `VALID_AT` syntax, internal-node-id-vs-property-value comparison, a +//! double-firing `BGSAVE` assertion, an invalid `FT.SEARCH` no-corruption +//! probe, an ineffective `#[ignore = "RED: ..."]` gating convention, a +//! vacuous kv-spill filler, a shared `red_guard` masking an unrelated GREEN +//! atomicity claim, a missing `GRAPH.CREATE` sync-wait in the split-out +//! atomicity check, a substring-based benign-error allowlist) — see git +//! history for the fix-by-fix trail. **40 cells total: 29 GREEN by +//! default, 11 RED.** +//! +//! RED cells are gated by `harness::red_guard` — NOT by +//! `#[ignore = "RED: ..."]` alone, because this suite's own execution +//! convention is `cargo test ... --ignored` (every cell needs a release +//! binary), and `--ignored` RUNS every ignored test regardless of its +//! reason string. `red_guard` early-returns (printing `SKIP: RED cell +//! (...)`) unless `MOON_CRASH_MATRIX_RED=1` is set, so the default +//! `--ignored` run is green-only (a clean tripwire for regressions) while +//! every RED reproduction stays one env var away on demand: +//! +//! ```text +//! cargo test --release --test crash_matrix_cross_plane -- --ignored --test-threads=1 # green-only (default) +//! MOON_CRASH_MATRIX_RED=1 cargo test --release --test crash_matrix_cross_plane -- --ignored --test-threads=1 # RED cells run too +//! ``` +//! +//! **RED cell 1 — legacy-mode graph reconstruction (one root cause, 6 +//! cells):** `cross_plane_legacy_yes_s1_graph_isolated`, +//! `cross_plane_legacy_yes_s1_txn_isolated_committed`, +//! `cross_plane_legacy_yes_s1_txn_isolated_atomicity`, +//! `cross_plane_legacy_yes_s1_mixed_all_planes_synced`, +//! `cross_plane_legacy_yes_s1_mixed_all_planes_mid_pass_c`, plus the +//! shard-count spot-check `cross_plane_spot_legacy_s4_graph_isolated` (6 +//! cells, one cause). PR #288's own changelog: graph command replay from +//! WAL v3 does not reconstruct the graph in legacy (`--disk-offload +//! disable`) mode even with zero Pass C interference — a pre-existing gap +//! unrelated to WAL recycling, deliberately not asserted by +//! `tests/crash_recovery_wal_recycle_legacy.rs` either. Every graph read +//! post-restart errors `ERR graph not found`. Not this stage's job to fix. +//! +//! **RED cell 2 — MQ resurrection via generic DEL:** +//! `cross_plane_seeded_red_mq_generic_del_resurrection`. PR #291's +//! changelog: durable MQ streams deleted via generic `DEL` are resurrected +//! — with full content — by MQ WAL replay after a restart; there is no MQ +//! tombstone record (same bug class as the already-fixed KV/vector +//! cold-plane resurrection, PR #257, not yet applied to MQ). Kernel M3 +//! brief §1.4's WS half of this finding turned out GREEN — +//! `cross_plane_seeded_red_ws_drop_durability` passes cleanly (WorkspaceDrop +//! WAL records DO replay in order); it stays an ordinary `#[ignore]` (needs +//! a release binary), not RED. +//! +//! **RED cell 3 — cross-store TXN graph leg not durable, task #52 (NEW, 2 +//! cells):** `cross_plane_prod_s1_txn_isolated_committed`, +//! `cross_plane_prod_s4_txn_isolated_committed`. A committed cross-store +//! transaction (`SET` + `GRAPH.ADDNODE` in one `MULTI`/`EXEC`) survives +//! kill-9 on the KV leg (PR #247's `persist_txn_aof`) but the graph leg is +//! lost. Reviewed and confirmed NOT a harness artifact; confirmed +//! deterministic 3× consecutive at both shard counts after a task #52 +//! hardening pass (own crash cycle per round, `--checkpoint-timeout 3600`, +//! kill immediately after the sync-wait — see +//! `scenarios::txn_isolated_committed`'s doc for the full analysis and +//! minimal repro). New P1 finding for kernel M3. The SIBLING atomicity +//! claim (`cross_plane_prod_s1_txn_isolated_atomicity` / +//! `_s4_txn_isolated_atomicity` — a transaction queued but killed BEFORE +//! `EXEC` must apply NOTHING) is a genuinely different, unrelated claim and +//! is GREEN on these two configs (review round 3, P1: it used to share this +//! same `red_guard`, incorrectly hiding a working regression tripwire by +//! default — split into its own ungated `#[test]` per config). It stays RED +//! on `legacy_yes_s1` only, folded into RED cell 1 above (same +//! legacy-graph-reconstruction root cause — confirmed on the VM, not +//! assumed: `GRAPH.CREATE` itself never survives a restart there, so the +//! atomicity check's own graph-leg assertion errors "graph not found" +//! before it can ever evaluate atomicity). +//! +//! **RED cell 4 — checkpoint-Finalize window total-losses the graph plane, +//! NEW (2 cells):** `cross_plane_prod_s1_mixed_all_planes_mid_checkpoint`, +//! `cross_plane_prod_s4_mixed_all_planes_mid_checkpoint`. Found via +//! `MOON_CRASH_MATRIX_ITERS` soak, NOT the default single-iteration run — +//! this is a PROBABILISTIC finding (some, not all, kill offsets in the +//! 0-150ms post-`BGSAVE` window trigger it — confirmed absent in the +//! default `MOON_CRASH_MATRIX_RED=1`/`ITERS=1` full-suite run that produced +//! this doc's final numbers, exactly as this caution note warns). Some +//! offsets cause TOTAL loss of the graph plane's content — not the unsynced +//! tail, the FULL `MixedPlan::default()` batch that was already confirmed +//! durable via a wal-v3 sync-wait BEFORE `BGSAVE` was even issued — while +//! KV, vector, WS, and MQ all survive in the same run. Reviewed and +//! confirmed NOT a harness artifact (KV assertion runs first, +//! unconditionally, in `assert_mixed_truth_recovered`, and never fails +//! here). Reproduced via `MOON_CRASH_MATRIX_ITERS=20` in isolation at both +//! shard counts (prod_s1: hit at iteration 11/20; prod_s4: hit at iteration +//! 7/20) — confirming this is a real, load-sensitive timing window, not a +//! one-off VM hiccup from the full-suite run that first surfaced it. See +//! `scenarios::mixed_mid_checkpoint`'s doc for the full analysis +//! (consistent with, though not proven at the step level to be, a +//! `persist_graph_at_checkpoint`-vs-`wal.recycle_segments_before` ordering +//! gap in `CheckpointAction::Finalize`). Strong P0 candidate for kernel M3 +//! stage 2 (K2, the unified per-shard floor register); not this stage's +//! job to fix. **Caution when reproducing:** because this is probabilistic, +//! `MOON_CRASH_MATRIX_RED=1` alone with the default `MOON_CRASH_MATRIX_ITERS=1` +//! can report green by chance — use +//! `MOON_CRASH_MATRIX_RED=1 MOON_CRASH_MATRIX_ITERS=20` to reproduce +//! reliably. +//! +//! **Every other cell (29/40) is GREEN** — including, notably, +//! `kv_isolated`/`kv_spilled_isolated`/`vector_isolated` on ALL configs (KV +//! and vector-HSET durability via the AOF; `kv_spilled_isolated`'s filler +//! sizing was hardened in review round 3 — see +//! `scenarios::kv_spilled_isolated`'s doc — and now passes its own hard +//! `heap-*.mpf` precondition on both shard counts), `graph_isolated` on +//! both `prod_s1`/`prod_s4` (structural writes + GraphTemporal +//! `TEMPORAL.INVALIDATE`, incl. a VALID_AT positive control — checkpoint-backed +//! mode does NOT share legacy mode's gap), `mq_isolated` everywhere (PR +//! #291's effect-record fix holds), `txn_isolated_atomicity` on +//! `prod_s1`/`prod_s4` (queuing without `EXEC` then killing applies nothing +//! — verified 3× consecutive), `mixed_all_planes_synced`/`mid_pass_c` on +//! both production shard counts, `mixed_all_planes_mid_checkpoint` at the +//! default single-iteration sample (RED cell 4 above is probabilistic — +//! most single samples don't hit the window), and +//! `mixed_all_planes_concurrent_burst_no_corruption` on all 3 configs it +//! runs on (no plane's on-disk structures were ever corrupted badly enough +//! to error post-restart, beyond the expected "schema object never made it +//! to disk before the kill" not-found case). +//! +//! # Soak cadence +//! +//! `MOON_CRASH_MATRIX_ITERS` (default `1`) — set to e.g. `100` to repeat the +//! curated cells' probabilistic kill points (`kill_mid_checkpoint`, +//! `kill_mid_pass_c`, `mixed::start_concurrent_burst`) that many times in a +//! loop inside the SAME `#[test]` fn, sampling more of the timing window +//! each rerun. No new CI workflow — this is a manual/on-demand env-gated +//! knob, run from a VM-local `--dir` (never `/Volumes/Games` directly — see +//! `gotcha_vm_diskfull_shared_volume` / `gotcha_vm_tmpfs_target_dirs_full`), +//! matching `scripts/bench-soak-7d.sh`'s local-only cadence (brief §Stage 1 +//! cadence, Open Decision #4). +//! +//! # Running +//! +//! ```text +//! cargo build --release +//! cargo test --release --test crash_matrix_cross_plane -- --ignored --test-threads=1 +//! MOON_CRASH_MATRIX_ITERS=100 cargo test --release --test crash_matrix_cross_plane \ +//! soaked -- --ignored --test-threads=1 +//! ``` +//! +//! `--test-threads=1`: every scenario spawns a real server process and +//! drives it with multiple threads/connections; running scenarios +//! concurrently multiplies port/tmpdir/VM-resource pressure for no benefit +//! (crash suites are I/O- and wall-clock-bound, not CPU-bound). + +#![cfg(any(feature = "runtime-monoio", feature = "runtime-tokio"))] +#![allow(clippy::unwrap_used)] + +mod common; + +#[path = "crash_matrix_cross_plane/harness.rs"] +mod harness; +#[path = "crash_matrix_cross_plane/mixed.rs"] +mod mixed; +#[path = "crash_matrix_cross_plane/planes.rs"] +mod planes; +#[path = "crash_matrix_cross_plane/resp.rs"] +mod resp; +#[path = "crash_matrix_cross_plane/scenarios.rs"] +mod scenarios; +#[path = "crash_matrix_cross_plane/tests_legacy.rs"] +mod tests_legacy; +#[path = "crash_matrix_cross_plane/tests_prod.rs"] +mod tests_prod; +#[path = "crash_matrix_cross_plane/tests_seeded_red.rs"] +mod tests_seeded_red; +#[path = "crash_matrix_cross_plane/tests_spot.rs"] +mod tests_spot; + +/// How many times a soak-eligible scenario repeats its probabilistic kill +/// window. `1` (default) = the normal PR-gated single pass; the brief's +/// ×100 soak is opt-in via this env var, never a hardcoded loop. +pub(crate) fn soak_iters() -> u32 { + std::env::var("MOON_CRASH_MATRIX_ITERS") + .ok() + .and_then(|s| s.parse().ok()) + .filter(|&n| n > 0) + .unwrap_or(1) +} diff --git a/tests/crash_matrix_cross_plane/harness.rs b/tests/crash_matrix_cross_plane/harness.rs new file mode 100644 index 00000000..a058dd9b --- /dev/null +++ b/tests/crash_matrix_cross_plane/harness.rs @@ -0,0 +1,479 @@ +//! Spawn/crash/restart harness + the curated config-cell matrix. +//! +//! Reuses `tests/common::{reserve_port, spawn_listening, find_moon_binary, +//! sigkill, wait_for_port_down}` (kernel M3 brief, Stage 1) instead of a +//! seventh hand-rolled copy of these primitives. + +#![allow(dead_code)] + +use std::path::{Path, PathBuf}; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +use crate::common; + +/// One curated config cell. `label` feeds the `cross_plane__` +/// test-fn naming convention the brief specifies (avoids the `g1_` prefix +/// collision documented in `crash_recovery_graph_durability.rs`). +#[derive(Clone, Copy)] +pub struct Config { + pub appendonly: bool, + pub disk_offload: bool, + pub shards: u32, + pub label: &'static str, +} + +impl Config { + /// Production cell A — the config the brief calls out as "the config + /// that matters most": checkpoint-backed, AOF durability, single shard. + pub const PROD_S1: Config = Config { + appendonly: true, + disk_offload: true, + shards: 1, + label: "prod_s1", + }; + /// Production cell B — same durability contract, 4 shards (the + /// cross-shard axis for the production config). + pub const PROD_S4: Config = Config { + appendonly: true, + disk_offload: true, + shards: 4, + label: "prod_s4", + }; + /// Legacy cell A — no checkpoint/snapshot protocol at all + /// (`segment_holds_plane_history` is the ONLY thing preventing WAL + /// recycle from eating plane history in this mode). Brief: legacy cells + /// run at shards=1 only ("no cross-shard novelty for this suite's + /// purpose"). + pub const LEGACY_YES_S1: Config = Config { + appendonly: true, + disk_offload: false, + shards: 1, + label: "legacy_yes_s1", + }; + /// Legacy cell B — `appendonly=no` skips the WAL writer entirely + /// (`persistence_dir` is only constructed when `appendonly=="yes" || + /// save.is_some()`, `src/config.rs`), so NO plane has any durability + /// contract here, not just KV. Assertions in this cell are deliberately + /// "must survive a crash without corrupting/hanging", never "must + /// recover data" — see the brief's "vacuous by design, not a bug" note. + pub const LEGACY_NO_S1: Config = Config { + appendonly: false, + disk_offload: false, + shards: 1, + label: "legacy_no_s1", + }; + /// Spot-check cell — disk-offload enabled but no durability backstop + /// (`appendonly=no`, no `--save`): `disk_offload_spill_inert` in + /// `src/config.rs` documents this as a loud no-op, not a crash, and + /// `persistence_dir` is still None so the WAL writer never exists here + /// either. Same "vacuous by design" contract as `LEGACY_NO_S1`. + pub const SPOT_OFFLOAD_NO_S1: Config = Config { + appendonly: false, + disk_offload: true, + shards: 1, + label: "spot_offload_no_s1", + }; + pub const SPOT_OFFLOAD_NO_S4: Config = Config { + appendonly: false, + disk_offload: true, + shards: 4, + label: "spot_offload_no_s4", + }; + /// Spot-check cell — legacy mode at shards=4 (the brief's curation + /// excludes legacy from the full plane×workload matrix at shards=4, but + /// still wants 1-2 representative scenarios here). + pub const SPOT_LEGACY_S4: Config = Config { + appendonly: true, + disk_offload: false, + shards: 4, + label: "spot_legacy_s4", + }; + pub const SPOT_LEGACY_NO_S4: Config = Config { + appendonly: false, + disk_offload: false, + shards: 4, + label: "spot_legacy_no_s4", + }; + + fn appendonly_arg(&self) -> &'static str { + if self.appendonly { "yes" } else { "no" } + } + + fn disk_offload_arg(&self) -> &'static str { + if self.disk_offload { + "enable" + } else { + "disable" + } + } + + /// True for the two cells with NO durability contract on ANY plane + /// (`persistence_dir` never constructed — see `LEGACY_NO_S1`'s doc). + pub fn no_durability_contract(&self) -> bool { + !self.appendonly + } +} + +// --------------------------------------------------------------------------- +// Spawn / crash / restart +// --------------------------------------------------------------------------- + +pub fn unique_dir(suffix: &str) -> PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + std::env::temp_dir().join(format!( + "moon-xplane-{}-{}-{}", + std::process::id(), + suffix, + nanos + )) +} + +/// Kills the wrapped child on drop (SIGKILL + waitpid via `common::sigkill`), +/// plus a best-effort `pkill -9 -f ` backstop keyed on this test's +/// unique temp dir (never a bare port/process-name pattern — see the +/// SIGTERM+SO_REUSEPORT / leaked-busy-poller gotchas in CLAUDE.md). +pub struct ServerGuard { + pub child: Child, + dir_marker: String, +} + +impl ServerGuard { + pub fn new(child: Child, dir: &Path) -> Self { + Self { + child, + dir_marker: dir.to_string_lossy().into_owned(), + } + } +} + +impl Drop for ServerGuard { + fn drop(&mut self) { + if matches!(self.child.try_wait(), Ok(None)) { + common::sigkill(&mut self.child); + } else { + let _ = self.child.wait(); + } + let _ = Command::new("pkill") + .args(["-9", "-f"]) + .arg(&self.dir_marker) + .output(); + } +} + +/// Spawn moon under `cfg`, with `extra` appended verbatim (WAL-size overrides +/// for the Pass-C kill point, `--maxmemory`/`--maxmemory-policy` for the +/// spilled-KV scenario, etc). Fresh `--dir` per test, `--disk-free-min-pct 0` +/// (DEL/FLUSH are write-flagged by the diskfull guard and this repo's dev +/// volumes routinely sit near the 5% line). Returns the guard and the port +/// actually bound (`spawn_listening` may respawn on a fresh port internally +/// after losing a bind race). +pub fn spawn_moon_on(dir: &Path, cfg: &Config, extra: &[&str]) -> (ServerGuard, u16) { + std::fs::create_dir_all(dir).expect("create test dir"); + let (child, port) = common::spawn_listening(|port| { + let mut args: Vec = vec![ + "--port".into(), + port.to_string(), + "--dir".into(), + dir.to_string_lossy().into_owned(), + "--shards".into(), + cfg.shards.to_string(), + "--appendonly".into(), + cfg.appendonly_arg().into(), + "--disk-offload".into(), + cfg.disk_offload_arg().into(), + "--disk-free-min-pct".into(), + "0".into(), + ]; + for &e in extra { + args.push(e.into()); + } + Command::new(common::find_moon_binary()) + .args(&args) + .env("RUST_LOG", "moon=info") + .stdout( + std::fs::File::options() + .append(true) + .create(true) + .open(dir.join("moon.stdout.log")) + .expect("stdout log"), + ) + .stderr( + std::fs::File::options() + .append(true) + .create(true) + .open(dir.join("moon.stderr.log")) + .expect("stderr log"), + ) + .spawn() + .unwrap_or_else(|e| { + panic!( + "spawn moon (dir={}) failed: {e}. Build with `cargo build \ + --release` or set MOON_BIN.", + dir.display() + ) + }) + }); + (ServerGuard::new(child, dir), port) +} + +/// SIGKILL + wait for the port to actually stop accepting (SO_REUSEPORT +/// makes a plain bind-check useless). +pub fn crash(mut guard: ServerGuard, port: u16) { + common::sigkill(&mut guard.child); + drop(guard); // idempotent (already reaped) + pkill backstop + common::wait_for_port_down(port); +} + +// --------------------------------------------------------------------------- +// WAL v3 durability-wait primitives +// --------------------------------------------------------------------------- + +/// Scan every shard's `wal-v3/` dir for `needle`. Used instead of a +/// per-shard variant because this suite deliberately does not pin write +/// routing to a specific shard index at shards=4 (hash-tag co-location picks +/// *some* shard, and the suite doesn't need to know which). +pub fn wait_for_wal_v3_bytes_any_shard(dir: &Path, num_shards: u32, needle: &[u8]) { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if wal_v3_contains_bytes_any_shard(dir, num_shards, needle) { + return; + } + assert!( + Instant::now() < deadline, + "no shard's wal-v3/ dir under {} ever contained marker {:?} — \ + writes are not reaching the WAL (or --appendonly no, where no \ + WAL writer exists at all)", + dir.display(), + String::from_utf8_lossy(needle), + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +pub fn wal_v3_contains_bytes_any_shard(dir: &Path, num_shards: u32, needle: &[u8]) -> bool { + for shard_id in 0..num_shards { + let wal_dir = dir.join(format!("shard-{shard_id}")).join("wal-v3"); + if let Ok(entries) = std::fs::read_dir(&wal_dir) { + for entry in entries.flatten() { + if let Ok(bytes) = std::fs::read(entry.path()) + && bytes.windows(needle.len()).any(|w| w == needle) + { + return true; + } + } + } + } + false +} + +// --------------------------------------------------------------------------- +// AOF durability-wait primitives (KV plane, incl. vector-HSET — brief §1.2: +// "vector durability rides the same path as any other HSET, normal Command +// WAL/AOF replay"). Confirmed empirically on the VM: plain SET commands +// NEVER appear in `wal-v3`, only in the AOF incr file; `wal-v3` is +// exclusively graph/vector-index-lifecycle/WS/MQ/temporal EFFECT records. +// Layout differs by shard count (confirmed via a live `--shards 4` smoke +// spawn): shards==1 uses a FLAT `appendonlydir/moon.aof.1.incr.aof` (no +// `shard-0/` subdirectory at all); shards>1 uses +// `appendonlydir/shard-/moon.aof.1.incr.aof` per shard. Scans both shapes +// unconditionally rather than branching on `num_shards` so a config change +// can never silently point this at the wrong layout. +// --------------------------------------------------------------------------- + +pub fn wait_for_aof_bytes_any_shard(dir: &Path, needle: &[u8]) { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if aof_contains_bytes_any_shard(dir, needle) { + return; + } + assert!( + Instant::now() < deadline, + "no AOF file under {}/appendonlydir ever contained marker {:?} — \ + writes are not reaching the AOF (or --appendonly no, where no \ + AOF writer exists at all)", + dir.display(), + String::from_utf8_lossy(needle), + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +/// Wait for EVERY needle in `needles` to appear (independently, possibly in +/// different files) under `dir/appendonlydir`. Use this instead of a single +/// concatenated needle when a command's arguments are logged as SEPARATE +/// RESP bulk strings (e.g. `SET key val` — the AOF never contains the +/// literal substring `"key=val"`, only `key` and `val` as distinct frames). +pub fn wait_for_aof_bytes_all_any_shard(dir: &Path, needles: &[&[u8]]) { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if needles + .iter() + .all(|needle| aof_contains_bytes_any_shard(dir, needle)) + { + return; + } + assert!( + Instant::now() < deadline, + "no AOF file under {}/appendonlydir ever contained all of {:?} — \ + writes are not reaching the AOF (or --appendonly no, where no \ + AOF writer exists at all)", + dir.display(), + needles + .iter() + .map(|n| String::from_utf8_lossy(n).into_owned()) + .collect::>(), + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +pub fn aof_contains_bytes_any_shard(dir: &Path, needle: &[u8]) -> bool { + let aof_root = dir.join("appendonlydir"); + if scan_aof_dir_for_bytes(&aof_root, needle) { + return true; + } + if let Ok(entries) = std::fs::read_dir(&aof_root) { + for entry in entries.flatten() { + if entry.path().is_dir() && scan_aof_dir_for_bytes(&entry.path(), needle) { + return true; + } + } + } + false +} + +fn scan_aof_dir_for_bytes(dir: &Path, needle: &[u8]) -> bool { + let Ok(entries) = std::fs::read_dir(dir) else { + return false; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.extension().is_some_and(|e| e == "aof") + && let Ok(bytes) = std::fs::read(&path) + && bytes.windows(needle.len()).any(|w| w == needle) + { + return true; + } + } + false +} + +pub fn wal_v3_total_bytes(dir: &Path, num_shards: u32) -> u64 { + let mut total = 0u64; + for shard_id in 0..num_shards { + let wal_dir = dir.join(format!("shard-{shard_id}")).join("wal-v3"); + if let Ok(entries) = std::fs::read_dir(&wal_dir) { + for entry in entries.flatten() { + total += std::fs::metadata(entry.path()) + .map(|m| m.len()) + .unwrap_or(0); + } + } + } + total +} + +/// Poll the shard-0 control file (`shard-0/shard-0.control`) until its +/// content changes from `before` — the moment a checkpoint Finalize has +/// written a new `last_checkpoint_lsn` (pattern: +/// `crash_recovery_graph_durability.rs`'s g5 scenario). +pub fn control_file_path(dir: &Path) -> PathBuf { + dir.join("shard-0").join("shard-0.control") +} + +pub fn wait_for_control_file_change(dir: &Path, before: &Option>, timeout: Duration) { + let path = control_file_path(dir); + let deadline = Instant::now() + timeout; + loop { + let now = std::fs::read(&path).ok(); + if now.is_some() && &now != before { + return; + } + assert!( + Instant::now() < deadline, + "control file {} never advanced — checkpoint Finalize did not run \ + within {timeout:?}", + path.display(), + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +/// Recursively count `heap-*.mpf` cold-tier data files anywhere under +/// `dir` (kv_spill's on-disk format, `src/storage/tiered/kv_spill.rs` — +/// `/shard-/data/heap-.mpf`; +/// `effective_disk_offload_dir` falls back to `--dir` when +/// `--disk-offload-dir` is not set, which is this harness's default, so a +/// plain recursive walk from the test's unique dir finds them without +/// needing to know the exact shard layout). A hard precondition for +/// every disk-offload scenario: without any heap files on disk, filler +/// pressure never forced a spill and the scenario silently degenerates to +/// ordinary AOF-replay coverage (already proven by `kv_isolated`) — proven +/// empirically in review round 3 (task #52 P0) against a too-small filler. +/// Pattern: `crash_recovery_cold_del_resurrection.rs::count_heap_files`. +pub fn count_heap_mpf_files(dir: &Path) -> usize { + fn walk(p: &Path, acc: &mut usize) { + let Ok(rd) = std::fs::read_dir(p) else { + return; + }; + for entry in rd.flatten() { + let path = entry.path(); + if path.is_dir() { + walk(&path, acc); + } else if path + .file_name() + .and_then(|n| n.to_str()) + .is_some_and(|n| n.starts_with("heap-") && n.ends_with(".mpf")) + { + *acc += 1; + } + } + } + let mut acc = 0; + walk(dir, &mut acc); + acc +} + +/// Gate for RED cells (kernel M3 brief, coordinator review round 3): the +/// suite's own execution convention is `cargo test ... --ignored` (every +/// test needs a built release binary, so ALL of them carry `#[ignore]`) — +/// but `--ignored` RUNS every ignored test, so annotating a RED cell with +/// `#[ignore = "RED: ..."]` does NOT exclude it from the default run; it +/// just documents intent while the cell still executes and fails. Every RED +/// cell calls this FIRST and returns early unless `MOON_CRASH_MATRIX_RED=1` +/// is set, so the default `--ignored` run is green-only (a clean tripwire +/// for regressions) while RED reproductions stay one env var away on +/// demand. `#[ignore]` on RED cells stays bare (or the generic "requires a +/// release binary" reason) — this guard is the actual RED/GREEN gate now. +pub fn red_guard(reason: &str) -> bool { + if std::env::var("MOON_CRASH_MATRIX_RED").is_err() { + eprintln!("SKIP: RED cell ({reason}) — set MOON_CRASH_MATRIX_RED=1 to run"); + return false; + } + true +} + +/// Cheap xorshift PRNG seeded from the wall clock — good enough for "pick a +/// random marker index to kill after" jitter; no external `rand` crate +/// dependency needed for this. +pub fn jitter_u64(seed_salt: u64) -> u64 { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0); + let mut x = nanos ^ seed_salt ^ (std::process::id() as u64).wrapping_mul(0x9E3779B97F4A7C15); + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + x +} + +/// Random value in `[lo, hi)` using [`jitter_u64`]. +pub fn jitter_range(seed_salt: u64, lo: u64, hi: u64) -> u64 { + assert!(hi > lo); + lo + jitter_u64(seed_salt) % (hi - lo) +} diff --git a/tests/crash_matrix_cross_plane/mixed.rs b/tests/crash_matrix_cross_plane/mixed.rs new file mode 100644 index 00000000..b6a845f1 --- /dev/null +++ b/tests/crash_matrix_cross_plane/mixed.rs @@ -0,0 +1,392 @@ +//! Concurrent mixed-writes-across-all-planes workload generator. +//! +//! This is the genuinely new coverage the kernel M3 brief calls out: no +//! existing suite drives KV + graph + vector + WS + MQ writes CONCURRENTLY +//! against the same shard (the closest precedent, +//! `tests/replication_planes.rs`'s Lua-writes-survive-kill-9 test, is +//! single-plane-at-a-time even though the script itself touches several +//! command families). Four threads, one per plane, each on its own +//! connection, barrier-started so they overlap on the wire; every write +//! lands on the SAME shard via a shared `{tag}` hash-tag baked into every +//! plane's routing key (WS is shard-0-pinned regardless of tag — Wave B). + +#![allow(dead_code)] + +use std::collections::BTreeSet; +use std::path::Path; +use std::sync::{Arc, Barrier}; +use std::thread; +use std::time::Duration; + +use crate::harness; +use crate::planes::*; +use crate::resp::Conn; + +pub struct MixedPlan { + pub kv_n: u64, + pub graph_n: u64, + pub vec_n: u64, + pub mq_n: u64, +} + +impl Default for MixedPlan { + fn default() -> Self { + MixedPlan { + kv_n: 20, + graph_n: 20, + vec_n: 20, + mq_n: 20, + } + } +} + +/// Ground truth for the PREFIX of each plane's ops that was waited-for on +/// disk before the kill. Ops beyond this prefix were sent but not +/// necessarily durable — a kill can land before, during, or after their WAL +/// append, so they are deliberately not asserted either way. +pub struct MixedTruth { + pub tag: String, + pub kv_synced: Vec<(String, String)>, + pub graph_name: String, + pub graph_ids_synced: BTreeSet, + pub vec_idx: String, + pub vec_synced: Vec<(String, Vec)>, + pub ws_name: String, + pub ws_id: Vec, + pub mq_queue: String, + pub mq_synced: u64, +} + +/// Run the four plane-writer threads concurrently to completion (schema ops +/// — GRAPH.CREATE / FT.CREATE / MQ.CREATE / WS.CREATE — happen once, +/// up-front, sequentially, since they are one-shot setup, not part of the +/// "sustained write load"). `sync_fraction` controls how much of EACH +/// plane's completed op stream is waited-for on disk (via a per-plane +/// marker on the last synced op) before this function returns — the +/// "kill at a random/periodic offset in the write stream" kill point from +/// the brief. `sync_fraction` is chosen by the caller (fixed 1.0 for a +/// full-durability check, or `harness::jitter_range` for a randomized +/// partial one). +pub fn run_mixed_workload_synced( + dir: &Path, + port: u16, + num_shards: u32, + tag: &str, + plan: MixedPlan, + sync_fraction: f64, +) -> MixedTruth { + let graph_name = graph_name(tag); + let vec_idx = vec_index_name(tag); + let vec_prefix = format!("{{{tag}}}:vec:"); + let ws_name = format!("{tag}-ws"); + let mq_queue = mq_queue_name(tag); + + // One-shot schema setup, sequential (not part of the concurrent burst). + let mut setup = Conn::open(port); + graph_create(&mut setup, &graph_name); + ft_create_hnsw(&mut setup, &vec_idx, &vec_prefix); + mq_create(&mut setup, &mq_queue, 3); + let ws_id = ws_create(&mut setup, &ws_name); + + let barrier = Arc::new(Barrier::new(4)); + + // KV writer. + let (kv_port, kv_tag, kv_n, kv_barrier) = (port, tag.to_string(), plan.kv_n, barrier.clone()); + let kv_handle = thread::spawn(move || -> Vec<(String, String)> { + let mut c = Conn::open(kv_port); + kv_barrier.wait(); + (0..kv_n) + .map(|i| { + let key = kv_key(&kv_tag, i); + let val = format!("v{i}"); + kv_set(&mut c, &key, &val); + (key, val) + }) + .collect() + }); + + // Graph writer. `graph_addnode`'s RETURN value is the internal node + // HANDLE, which legitimately renumbers across a restart (WAL replay can + // reassign shard-encoded handles) — collect the "id" PROPERTY value + // (`i`, the value actually queryable post-restart via `RETURN n.id`) + // instead, matching how every other scenario in this suite verifies + // graph durability. + let (g_port, g_name, g_n, g_barrier) = + (port, graph_name.clone(), plan.graph_n, barrier.clone()); + let graph_handle = thread::spawn(move || -> Vec { + let mut c = Conn::open(g_port); + g_barrier.wait(); + (0..g_n) + .map(|i| { + graph_addnode(&mut c, &g_name, "MixN", i); + i as i64 + }) + .collect() + }); + + // Vector writer. + let (v_port, v_prefix, v_n, v_barrier) = + (port, vec_prefix.clone(), plan.vec_n, barrier.clone()); + let vec_handle = thread::spawn(move || -> Vec<(String, Vec)> { + let mut c = Conn::open(v_port); + v_barrier.wait(); + (0..v_n) + .map(|i| { + let key = format!("{v_prefix}{i}"); + let blob = vec_blob(i); + hset_vec(&mut c, &key, &blob); + (key, blob) + }) + .collect() + }); + + // MQ writer. + let (m_port, m_queue, m_n, m_barrier) = (port, mq_queue.clone(), plan.mq_n, barrier); + let mq_handle = thread::spawn(move || -> u64 { + let mut c = Conn::open(m_port); + m_barrier.wait(); + for i in 0..m_n { + mq_push(&mut c, &m_queue, "seq", &i.to_string()); + } + m_n + }); + + let kv_all = kv_handle.join().expect("kv writer panicked"); + let graph_ids: BTreeSet = graph_handle + .join() + .expect("graph writer panicked") + .into_iter() + .collect(); + let vec_all = vec_handle.join().expect("vector writer panicked"); + let mq_pushed = mq_handle.join().expect("mq writer panicked"); + + let cut = |n: u64| -> u64 { ((n as f64) * sync_fraction).floor() as u64 }; + let kv_cut = cut(kv_all.len() as u64) as usize; + let graph_cut = cut(graph_ids.len() as u64) as usize; + let vec_cut = cut(vec_all.len() as u64) as usize; + let mq_cut = cut(mq_pushed); + + // Sync each plane's prefix by writing ONE extra marker op tagged with + // the cut index and waiting for it on the WAL — proves everything + // strictly before it (same connection, same shard, in-order append) is + // durable too. + let mut sync_conn = Conn::open(port); + if kv_cut > 0 { + // KV rides the AOF, not wal-v3 (confirmed empirically — plain SETs + // never appear in wal-v3 at all). + let marker = format!("XPLANE-KV-SYNC-{tag}-{kv_cut}"); + kv_set(&mut sync_conn, &format!("{{{tag}}}:kvsync"), &marker); + harness::wait_for_aof_bytes_any_shard(dir, marker.as_bytes()); + } + if graph_cut > 0 { + // Marker must be a property VALUE, not a label (labels are not + // preserved verbatim in the wal-v3 record — see + // `planes::graph_addnode_marker`'s doc). + let marker = format!("XPLANE-GRAPH-SYNC-{tag}-{graph_cut}"); + graph_addnode_marker(&mut sync_conn, &graph_name, &marker); + harness::wait_for_wal_v3_bytes_any_shard(dir, num_shards, marker.as_bytes()); + } + if vec_cut > 0 { + // Vector-HSET durability rides the AOF too (brief §1.2), not wal-v3. + let marker_key = format!("{vec_prefix}sync"); + let marker_blob = vec_blob(777); + hset_vec(&mut sync_conn, &marker_key, &marker_blob); + harness::wait_for_aof_bytes_any_shard(dir, marker_key.as_bytes()); + } + if mq_cut > 0 { + let marker = format!("XPLANE-MQ-SYNC-{tag}-{mq_cut}"); + mq_push(&mut sync_conn, &mq_queue, "sync", &marker); + harness::wait_for_wal_v3_bytes_any_shard(dir, num_shards, marker.as_bytes()); + } + // WS.CREATE above is a single fast op on its own — always wait for it + // directly rather than deriving a cut. + harness::wait_for_wal_v3_bytes_any_shard(dir, num_shards, ws_name.as_bytes()); + + MixedTruth { + tag: tag.to_string(), + kv_synced: kv_all.into_iter().take(kv_cut).collect(), + graph_name, + graph_ids_synced: graph_ids.into_iter().take(graph_cut).collect(), + vec_idx, + vec_synced: vec_all.into_iter().take(vec_cut).collect(), + ws_name, + ws_id, + mq_queue, + mq_synced: mq_cut, + } +} + +pub fn assert_mixed_truth_recovered(c: &mut Conn, truth: &MixedTruth) { + for (k, v) in &truth.kv_synced { + assert!( + kv_get_eq(c, k, v), + "mixed-workload KV key {k} must survive (synced prefix)" + ); + } + if !truth.graph_ids_synced.is_empty() { + let got = graph_query_ids(c, &truth.graph_name, "MATCH (n:MixN) RETURN n.id"); + assert!( + truth.graph_ids_synced.is_subset(&got), + "mixed-workload graph ids {:?} must be a subset of recovered {:?} \ + (graph plane {})", + truth.graph_ids_synced, + got, + truth.graph_name, + ); + } + for (key, blob) in &truth.vec_synced { + assert!( + vec_search_contains(c, &truth.vec_idx, 5, blob, key), + "mixed-workload vector key {key} must survive (synced prefix, index {})", + truth.vec_idx, + ); + } + assert!( + ws_list_contains(c, &truth.ws_id), + "mixed-workload workspace {} must survive", + truth.ws_name, + ); + if truth.mq_synced > 0 { + assert!( + xlen(c, &truth.mq_queue) >= truth.mq_synced as i64, + "mixed-workload MQ queue {} must have at least {} entries after restart", + truth.mq_queue, + truth.mq_synced, + ); + } +} + +/// RAII guard (coordinator review round 3, item 3): filters the EXPECTED +/// "server died mid-write" panic messages (`Conn::fill`'s `read from +/// server` / `connection closed mid-frame`, `write cmd`/`write pipeline`) +/// out of stderr for as long as it is held, restoring the previous hook on +/// drop. Deliberately scoped TIGHTLY around a single +/// `concurrent_burst_no_corruption` iteration (not installed process-wide, +/// not left in place for the rest of this long-lived test binary) — this +/// test binary runs 37 tests sequentially in one process, so a +/// process-wide swap would risk silently swallowing a GENUINE +/// disconnect-shaped panic in some unrelated LATER test. Any panic message +/// that doesn't match the known-benign substrings is forwarded to the +/// previous hook unchanged, so real bugs stay visible. +/// +/// REQUIRES `--test-threads=1` (this suite's documented invocation — see +/// the crate-root module doc): `std::panic::set_hook`/`take_hook` are +/// PROCESS-GLOBAL, not per-thread. If two `#[test]` functions ran +/// concurrently and both installed/dropped this guard around overlapping +/// windows, the `take_hook`/`set_hook` pairs could interleave and clobber +/// each other's `prev` chain. This does NOT corrupt test *results* — a +/// panic still unwinds and fails its own test regardless of which hook +/// happened to print it — it only garbles WHICH hook prints a given panic +/// message, i.e. stderr readability, not correctness. Still worth avoiding: +/// this suite already requires `--test-threads=1` for an unrelated reason +/// (every cell spawns/kills real server processes bound to dynamically +/// reserved ports; concurrent cells competing for OS resources is its own +/// source of flakiness), so this guard's global-hook sharing is inert in +/// practice, not accidentally safe. +type PanicHook = dyn Fn(&std::panic::PanicHookInfo<'_>) + Sync + Send; + +pub struct BenignDisconnectPanicFilter { + prev: Option>, +} + +impl BenignDisconnectPanicFilter { + pub fn install() -> Self { + let prev: std::sync::Arc = std::panic::take_hook().into(); + let forward = prev.clone(); + std::panic::set_hook(Box::new(move |info| { + let payload = info.payload(); + let msg = payload + .downcast_ref::<&str>() + .copied() + .or_else(|| payload.downcast_ref::().map(String::as_str)) + .unwrap_or(""); + let benign = msg.contains("read from server") + || msg.contains("connection closed mid-frame") + || msg.contains("write cmd") + || msg.contains("write pipeline"); + if !benign { + forward(info); + } + })); + Self { prev: Some(prev) } + } +} + +impl Drop for BenignDisconnectPanicFilter { + fn drop(&mut self) { + if let Some(prev) = self.prev.take() { + std::panic::set_hook(Box::new(move |info| prev(info))); + } + } +} + +/// Genuinely-concurrent variant: start the four writer threads and return +/// while they are STILL mid-burst (no join, no markers, no sync wait) — the +/// purest form of "random offset in the write stream". The caller MUST +/// SIGKILL the server immediately after this returns (any delay narrows the +/// "mid-flight" window). Because nothing is synchronized, this makes NO +/// durability claim about which ops landed; callers only assert the server +/// comes back up cleanly and every plane answers well-formed queries +/// afterward (no corruption / no crash-loop / no wedged replay). Writer +/// threads are detached (not joined) — once the socket dies each iteration's +/// `.expect()`/`assert_eq!` panics and unwinds that thread only (Cargo +/// forces `panic = "unwind"` for test binaries regardless of the release +/// profile's `panic = "abort"`), which is harmless noise on stderr (silenced +/// by `BenignDisconnectPanicFilter` when the caller installs it), not a +/// test failure (nothing here ever calls `.join().unwrap()`). +pub fn start_concurrent_burst(port: u16, tag: &str) { + let graph_name = graph_name(tag); + let vec_idx = vec_index_name(tag); + let vec_prefix = format!("{{{tag}}}:vec:"); + let mq_queue = mq_queue_name(tag); + + let mut setup = Conn::open(port); + graph_create(&mut setup, &graph_name); + ft_create_hnsw(&mut setup, &vec_idx, &vec_prefix); + mq_create(&mut setup, &mq_queue, 3); + drop(setup); + + let barrier = Arc::new(Barrier::new(4)); + + let (kv_port, kv_tag, b) = (port, tag.to_string(), barrier.clone()); + thread::spawn(move || { + let mut c = Conn::open(kv_port); + b.wait(); + for i in 0..100_000u64 { + kv_set(&mut c, &kv_key(&kv_tag, i), "v"); + } + }); + + let (g_port, g_name, b) = (port, graph_name, barrier.clone()); + thread::spawn(move || { + let mut c = Conn::open(g_port); + b.wait(); + for i in 0..100_000u64 { + graph_addnode(&mut c, &g_name, "BurstN", i); + } + }); + + let (v_port, v_prefix, b) = (port, vec_prefix, barrier.clone()); + thread::spawn(move || { + let mut c = Conn::open(v_port); + b.wait(); + for i in 0..100_000u64 { + let key = format!("{v_prefix}{i}"); + hset_vec(&mut c, &key, &vec_blob(i)); + } + }); + + let (m_port, m_queue, b) = (port, mq_queue, barrier); + thread::spawn(move || { + let mut c = Conn::open(m_port); + b.wait(); + for i in 0..100_000u64 { + mq_push(&mut c, &m_queue, "seq", &i.to_string()); + } + }); + + // Let the burst run for a short, randomized window before returning — + // genuinely "somewhere mid-stream", never a marker-synced boundary. + let jitter_ms = harness::jitter_range(0xB0F5, 50, 400); + thread::sleep(Duration::from_millis(jitter_ms)); +} diff --git a/tests/crash_matrix_cross_plane/planes.rs b/tests/crash_matrix_cross_plane/planes.rs new file mode 100644 index 00000000..89007235 --- /dev/null +++ b/tests/crash_matrix_cross_plane/planes.rs @@ -0,0 +1,375 @@ +//! Per-plane write/verify helpers shared by every scenario module. +//! +//! Every helper takes a `tag: &str` used as a RESP2 hash-tag (`{tag}`) baked +//! into the plane's routing key, so the mixed-workload scenarios can force +//! KV/graph/vector/MQ writes onto the SAME shard at `--shards 4` (WS is +//! always shard-0-pinned regardless of tag — see `ws_create`'s doc). + +#![allow(dead_code)] + +use std::collections::BTreeSet; + +use crate::resp::{Conn, Resp, as_array, as_int}; + +// --------------------------------------------------------------------------- +// KV (incl. spilled collections) +// --------------------------------------------------------------------------- + +pub fn kv_key(tag: &str, n: u64) -> String { + format!("{{{tag}}}:kv:{n}") +} + +pub fn kv_set(c: &mut Conn, key: &str, val: &str) { + assert!(c.cmd_s(&["SET", key, val]).is_ok(), "SET {key} failed"); +} + +pub fn kv_get_eq(c: &mut Conn, key: &str, expected: &str) -> bool { + matches!(c.cmd_s(&["GET", key]), Resp::Bulk(Some(b)) if b == expected.as_bytes()) +} + +pub fn kv_del(c: &mut Conn, key: &str) { + c.cmd_s(&["DEL", key]); +} + +/// Fill `count` filler keys under a DIFFERENT (untagged) prefix, each +/// `value_len` bytes, to push shard memory past the disk-offload spill +/// threshold — forces the tagged probe keys written earlier to evict to the +/// cold tier. Pattern: `tests/crash_recovery_cold_del_resurrection.rs`'s +/// `write_filler` — pipelined (one `pipeline_s` batch per chunk, not one +/// round trip per key) so 16,000+ filler writes complete in a couple of +/// seconds rather than minutes; caller MUST verify the spill actually +/// happened via a hard precondition (`harness::count_heap_mpf_files`) rather +/// than assuming byte-count math alone forced it — see review round 3 (task +/// #52 P0): a prior 3000×512B filler against a 4 MiB cap silently never +/// spilled at all and the cell degenerated to plain AOF-replay coverage. +pub fn kv_spill_filler(c: &mut Conn, prefix: &str, count: u64, value_len: usize) { + let value = "f".repeat(value_len); + const BATCH: u64 = 2000; + let mut i = 0u64; + while i < count { + let end = (i + BATCH).min(count); + let keys: Vec = (i..end).map(|n| format!("{prefix}:filler:{n}")).collect(); + let cmds: Vec> = keys + .iter() + .map(|k| vec!["SET", k.as_str(), value.as_str()]) + .collect(); + c.pipeline_s(&cmds); + i = end; + } +} + +// --------------------------------------------------------------------------- +// Graph (structural + GraphTemporal) +// --------------------------------------------------------------------------- + +pub fn graph_name(tag: &str) -> String { + format!("{{{tag}}}graph") +} + +pub fn graph_create(c: &mut Conn, name: &str) { + assert_eq!( + c.cmd_s(&["GRAPH.CREATE", name]), + Resp::Simple("OK".into()), + "GRAPH.CREATE {name}" + ); +} + +pub fn graph_addnode(c: &mut Conn, name: &str, label: &str, id: u64) -> i64 { + let id_s = id.to_string(); + match c.cmd_s(&["GRAPH.ADDNODE", name, label, "id", &id_s]) { + Resp::Int(h) => h, + other => panic!("GRAPH.ADDNODE {name} {label} id={id} failed: {other:?}"), + } +} + +/// Write `marker` as a property VALUE (not a label) on a fresh node, then +/// return. Property values are preserved literally in the WAL v3 record; +/// labels are NOT (dictionary/handle-encoded on the rewritten internal +/// command form — confirmed via `xxd` on a real +/// `shard-0/wal-v3/000000000001.wal` segment). Every marker-sync in this +/// suite that targets the graph plane MUST go through this helper, never +/// `graph_addnode`'s `label` argument, or the WAL-wait below will time out +/// waiting for bytes that are never written verbatim. +pub fn graph_addnode_marker(c: &mut Conn, name: &str, marker: &str) { + match c.cmd_s(&["GRAPH.ADDNODE", name, "Marker", "tag", marker]) { + Resp::Int(_) => {} + other => panic!("GRAPH.ADDNODE {name} Marker tag={marker} failed: {other:?}"), + } +} + +pub fn graph_addedge(c: &mut Conn, name: &str, src: i64, dst: i64) { + let (s, d) = (src.to_string(), dst.to_string()); + match c.cmd_s(&["GRAPH.ADDEDGE", name, &s, &d, "E"]) { + Resp::Int(_) => {} + other => panic!("GRAPH.ADDEDGE {name} {src}->{dst} failed: {other:?}"), + } +} + +pub fn graph_query_ids(c: &mut Conn, name: &str, cypher: &str) -> BTreeSet { + match c.cmd_s(&["GRAPH.QUERY", name, cypher]) { + Resp::Array(Some(outer)) => match outer.get(1) { + Some(Resp::Array(Some(rows))) => rows + .iter() + .map(|r| match r { + Resp::Array(Some(cells)) => match cells.first() { + Some(Resp::Int(i)) => *i, + Some(Resp::Bulk(Some(b))) => { + String::from_utf8_lossy(b).parse().expect("int cell") + } + other => panic!("unexpected cell: {other:?}"), + }, + other => panic!("malformed row: {other:?}"), + }) + .collect(), + other => panic!("malformed query reply, rows slot = {other:?}"), + }, + Resp::Error(e) => panic!("query {cypher:?} errored: {e}"), + other => panic!("malformed query reply: {other:?}"), + } +} + +/// Same as [`graph_query_ids`] but with an explicit `VALID_AT ` clause. +/// `VALID_AT` is a SEPARATE `GRAPH.QUERY` argument, not part of the Cypher +/// text itself — embedding it into the Cypher string (e.g. `"... RETURN +/// n.id VALID_AT 123"`) is a parse error (`ERR Cypher parse error: ... +/// expected clause keyword ... found Ident(...)`). Confirmed against the +/// working pattern in `tests/crash_recovery_temporal_mq.rs`. +pub fn graph_query_ids_valid_at( + c: &mut Conn, + name: &str, + cypher: &str, + valid_at_ms: &str, +) -> BTreeSet { + match c.cmd_s(&["GRAPH.QUERY", name, cypher, "VALID_AT", valid_at_ms]) { + Resp::Array(Some(outer)) => match outer.get(1) { + Some(Resp::Array(Some(rows))) => rows + .iter() + .map(|r| match r { + Resp::Array(Some(cells)) => match cells.first() { + Some(Resp::Int(i)) => *i, + Some(Resp::Bulk(Some(b))) => { + String::from_utf8_lossy(b).parse().expect("int cell") + } + other => panic!("unexpected cell: {other:?}"), + }, + other => panic!("malformed row: {other:?}"), + }) + .collect(), + other => panic!("malformed query reply, rows slot = {other:?}"), + }, + Resp::Error(e) => panic!("query {cypher:?} VALID_AT {valid_at_ms} errored: {e}"), + other => panic!("malformed query reply: {other:?}"), + } +} + +pub fn graph_write(c: &mut Conn, name: &str, cypher: &str) { + if let Resp::Error(e) = c.cmd_s(&["GRAPH.QUERY", name, cypher]) { + panic!("write query {cypher:?} on {name} errored: {e}"); + } +} + +/// `TEMPORAL.INVALIDATE NODE ` — GraphTemporal WAL record; +/// structurally covered by the graph engine's `snapshot_lsn` floor (kernel +/// M3 brief §1.3) but currently guarded like WS/MQ by +/// `segment_holds_plane_history` pre-K2. +pub fn temporal_invalidate_node(c: &mut Conn, node_id: &str, graph: &str) { + assert_eq!( + c.cmd_s(&["TEMPORAL.INVALIDATE", node_id, "NODE", graph]), + Resp::Simple("OK".into()), + "TEMPORAL.INVALIDATE {node_id} on {graph}" + ); +} + +// --------------------------------------------------------------------------- +// Vector +// --------------------------------------------------------------------------- + +pub const VEC_DIM: usize = 8; + +pub fn vec_index_name(tag: &str) -> String { + format!("{tag}vecidx") +} + +pub fn vec_key(tag: &str, n: u64) -> String { + format!("{{{tag}}}:vec:{n}") +} + +/// Deterministic float32 blob from a seed — no external RNG dependency. +pub fn vec_blob(seed: u64) -> Vec { + let mut out = Vec::with_capacity(VEC_DIM * 4); + for i in 0..VEC_DIM { + let v = ((seed.wrapping_mul(31).wrapping_add(i as u64)) % 1000) as f32 / 1000.0; + out.extend_from_slice(&v.to_le_bytes()); + } + out +} + +pub fn ft_create_hnsw(c: &mut Conn, idx: &str, prefix: &str) { + let dim_s = VEC_DIM.to_string(); + let reply = c.cmd(&[ + b"FT.CREATE", + idx.as_bytes(), + b"ON", + b"HASH", + b"PREFIX", + b"1", + prefix.as_bytes(), + b"SCHEMA", + b"vec", + b"VECTOR", + b"HNSW", + b"8", + b"TYPE", + b"FLOAT32", + b"DIM", + dim_s.as_bytes(), + b"DISTANCE_METRIC", + b"L2", + ]); + assert_eq!(reply, Resp::Simple("OK".into()), "FT.CREATE {idx} failed"); +} + +pub fn hset_vec(c: &mut Conn, key: &str, blob: &[u8]) { + match c.cmd(&[b"HSET", key.as_bytes(), b"vec", blob]) { + Resp::Int(_) => {} + other => panic!("HSET {key} failed: {other:?}"), + } +} + +/// True if `key` is among the top-`k` KNN results against `blob` — enough +/// signal for a crash-durability check (does the vector plane know about +/// this key at all after restart), not a recall benchmark. +pub fn vec_search_contains(c: &mut Conn, idx: &str, k: u32, blob: &[u8], key: &str) -> bool { + let query = format!("*=>[KNN {k} @vec $B]"); + let r = c.cmd(&[ + b"FT.SEARCH", + idx.as_bytes(), + query.as_bytes(), + b"PARAMS", + b"2", + b"B", + blob, + b"DIALECT", + b"2", + ]); + let items = as_array(&r); + items[1..] + .iter() + .step_by(2) + .any(|v| matches!(v, Resp::Bulk(Some(b)) if b == key.as_bytes())) +} + +// --------------------------------------------------------------------------- +// WS (workspace registry — process-global, shard-0-pinned per Wave B) +// --------------------------------------------------------------------------- + +pub fn ws_create(c: &mut Conn, name: &str) -> Vec { + match c.cmd_s(&["WS", "CREATE", name]) { + Resp::Bulk(Some(id)) => id, + other => panic!("WS CREATE {name} failed: {other:?}"), + } +} + +pub fn ws_drop(c: &mut Conn, id: &[u8]) { + let id_s = String::from_utf8_lossy(id).into_owned(); + match c.cmd_s(&["WS", "DROP", &id_s]) { + Resp::Simple(_) | Resp::Int(_) => {} + other => panic!("WS DROP {id_s} failed: {other:?}"), + } +} + +pub fn ws_list_contains(c: &mut Conn, id: &[u8]) -> bool { + match c.cmd_s(&["WS", "LIST"]) { + Resp::Array(Some(entries)) => entries.iter().any(|e| match e { + Resp::Array(Some(cells)) => { + matches!(cells.first(), Some(Resp::Bulk(Some(b))) if b == id) + } + _ => false, + }), + other => panic!("WS LIST failed: {other:?}"), + } +} + +// --------------------------------------------------------------------------- +// MQ (incl. DLQ / PEL / triggers) +// --------------------------------------------------------------------------- + +pub fn mq_queue_name(tag: &str) -> String { + format!("{{{tag}}}mq") +} + +pub fn mq_create(c: &mut Conn, q: &str, max_delivery: u32) { + let md = max_delivery.to_string(); + assert_eq!( + c.cmd_s(&["MQ", "CREATE", q, "MAXDELIVERY", &md]), + Resp::Simple("OK".into()), + "MQ CREATE {q}" + ); +} + +pub fn mq_push(c: &mut Conn, q: &str, field: &str, val: &str) -> Vec { + match c.cmd_s(&["MQ", "PUSH", q, field, val]) { + Resp::Bulk(Some(id)) => id, + other => panic!("MQ PUSH {q} failed: {other:?}"), + } +} + +pub fn mq_pop(c: &mut Conn, q: &str, count: u32) -> Vec { + let cnt = count.to_string(); + as_array(&c.cmd_s(&["MQ", "POP", q, "COUNT", &cnt])).to_vec() +} + +pub fn mq_ack(c: &mut Conn, q: &str, ids: &[Vec]) -> i64 { + let id_strs: Vec = ids + .iter() + .map(|i| String::from_utf8_lossy(i).into_owned()) + .collect(); + let mut parts: Vec<&str> = vec!["MQ", "ACK", q]; + parts.extend(id_strs.iter().map(|s| s.as_str())); + as_int(&c.cmd_s(&parts)) +} + +pub fn mq_dlqlen(c: &mut Conn, q: &str) -> i64 { + as_int(&c.cmd_s(&["MQ", "DLQLEN", q])) +} + +pub fn mq_trigger(c: &mut Conn, q: &str, callback: &str, debounce_ms: u32) { + let ms = debounce_ms.to_string(); + assert_eq!( + c.cmd_s(&["MQ", "TRIGGER", q, callback, "DEBOUNCE", &ms]), + Resp::Simple("OK".into()), + "MQ TRIGGER {q}" + ); +} + +pub fn xlen(c: &mut Conn, key: &str) -> i64 { + as_int(&c.cmd_s(&["XLEN", key])) +} + +// --------------------------------------------------------------------------- +// Cross-store TXN (MULTI/EXEC spanning ≥2 planes) +// --------------------------------------------------------------------------- + +/// Queue a mixed KV+graph write inside MULTI/EXEC and execute it. Returns +/// the EXEC reply (caller inspects it for the golden invariant: an +/// EXEC-committed write must survive exactly like a non-MULTI write). +pub fn txn_kv_graph_write( + c: &mut Conn, + kv_key: &str, + kv_val: &str, + graph: &str, + label: &str, + node_id: u64, +) -> Resp { + assert_eq!(c.cmd_s(&["MULTI"]), Resp::Simple("OK".into())); + let id_s = node_id.to_string(); + assert_eq!( + c.cmd_s(&["SET", kv_key, kv_val]), + Resp::Simple("QUEUED".into()) + ); + assert_eq!( + c.cmd_s(&["GRAPH.ADDNODE", graph, label, "id", &id_s]), + Resp::Simple("QUEUED".into()) + ); + c.cmd_s(&["EXEC"]) +} diff --git a/tests/crash_matrix_cross_plane/resp.rs b/tests/crash_matrix_cross_plane/resp.rs new file mode 100644 index 00000000..250e6bdd --- /dev/null +++ b/tests/crash_matrix_cross_plane/resp.rs @@ -0,0 +1,222 @@ +//! Minimal self-contained RESP2 client (pattern: +//! `tests/crash_recovery_mq_effects.rs::Conn` / `tests/shardslice_live.rs`). +//! Deliberately duplicated rather than shared — every other crash suite in +//! this repo carries its own copy too (each has slightly different framing +//! needs: RESP3 maps, pipelining, etc.), and unifying RESP clients is a +//! separate, larger refactor than the harness-primitive consolidation this +//! stage takes on (see the module doc in `crash_matrix_cross_plane.rs`). + +#![allow(dead_code)] + +use std::io::{Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::time::{Duration, Instant}; + +#[derive(Debug, Clone, PartialEq)] +pub enum Resp { + Simple(String), + Error(String), + Int(i64), + Bulk(Option>), + Array(Option>), +} + +impl Resp { + pub fn flat(&self) -> String { + match self { + Resp::Simple(s) | Resp::Error(s) => s.clone(), + Resp::Int(i) => i.to_string(), + Resp::Bulk(Some(b)) => String::from_utf8_lossy(b).into_owned(), + Resp::Bulk(None) => "".into(), + Resp::Array(Some(items)) => items.iter().map(Resp::flat).collect::>().join(" "), + Resp::Array(None) => "".into(), + } + } + + pub fn is_ok(&self) -> bool { + matches!(self, Resp::Simple(s) if s == "OK") + } +} + +pub fn connect(port: u16, deadline: Duration) -> TcpStream { + let addr = format!("127.0.0.1:{port}") + .to_socket_addrs() + .expect("parse addr") + .next() + .expect("one addr"); + let start = Instant::now(); + loop { + match TcpStream::connect_timeout(&addr, Duration::from_millis(200)) { + Ok(s) => { + s.set_read_timeout(Some(Duration::from_secs(15))).ok(); + s.set_write_timeout(Some(Duration::from_secs(15))).ok(); + return s; + } + Err(_) if start.elapsed() < deadline => { + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => panic!("server never accepted on port {port}: {e}"), + } + } +} + +/// Poll `PING` until the server answers `PONG` — a successful TCP accept +/// only means "listening", not "serving" (recovery/replay can still be +/// running Phase B before the dispatch loop comes up). +pub fn wait_ready(port: u16) -> TcpStream { + let mut s = connect(port, Duration::from_secs(30)); + let start = Instant::now(); + loop { + s.write_all(b"PING\r\n").expect("write PING"); + let mut buf = [0u8; 64]; + if let Ok(n) = s.read(&mut buf) + && n > 0 + && buf[..n].windows(4).any(|w| w == b"PONG") + { + return s; + } + assert!( + start.elapsed() < Duration::from_secs(30), + "server accepted TCP but never answered PING on port {port}" + ); + std::thread::sleep(Duration::from_millis(100)); + s = connect(port, Duration::from_secs(10)); + } +} + +pub struct Conn { + s: TcpStream, + buf: Vec, + pos: usize, +} + +impl Conn { + pub fn new(s: TcpStream) -> Self { + Conn { + s, + buf: Vec::with_capacity(16 * 1024), + pos: 0, + } + } + + pub fn open(port: u16) -> Self { + Conn::new(connect(port, Duration::from_secs(15))) + } + + pub fn cmd(&mut self, parts: &[&[u8]]) -> Resp { + let mut req = Vec::with_capacity(128); + req.extend_from_slice(format!("*{}\r\n", parts.len()).as_bytes()); + for p in parts { + req.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + req.extend_from_slice(p); + req.extend_from_slice(b"\r\n"); + } + self.s.write_all(&req).expect("write cmd"); + self.frame() + } + + pub fn cmd_s(&mut self, parts: &[&str]) -> Resp { + let v: Vec<&[u8]> = parts.iter().map(|p| p.as_bytes()).collect(); + self.cmd(&v) + } + + /// Send every command without waiting for a reply, then read all replies + /// in order (pipelining — used to send `MULTI`-body writes as a burst + /// right before a kill so the harness controls exactly what got queued + /// on the wire before the process died). + pub fn pipeline_s(&mut self, cmds: &[Vec<&str>]) -> Vec { + let mut buf = Vec::new(); + for c in cmds { + let mut req = Vec::with_capacity(64); + req.extend_from_slice(format!("*{}\r\n", c.len()).as_bytes()); + for p in c { + let bytes = p.as_bytes(); + req.extend_from_slice(format!("${}\r\n", bytes.len()).as_bytes()); + req.extend_from_slice(bytes); + req.extend_from_slice(b"\r\n"); + } + buf.extend_from_slice(&req); + } + self.s.write_all(&buf).expect("write pipeline"); + cmds.iter().map(|_| self.frame()).collect() + } + + fn fill(&mut self) { + let mut chunk = [0u8; 16 * 1024]; + let n = self.s.read(&mut chunk).expect("read from server"); + assert!(n > 0, "connection closed mid-frame"); + self.buf.extend_from_slice(&chunk[..n]); + } + + fn line(&mut self) -> String { + loop { + if let Some(rel) = self.buf[self.pos..].windows(2).position(|w| w == b"\r\n") { + let line = + String::from_utf8_lossy(&self.buf[self.pos..self.pos + rel]).into_owned(); + self.pos += rel + 2; + return line; + } + self.fill(); + } + } + + fn exact(&mut self, n: usize) -> Vec { + while self.buf.len() - self.pos < n + 2 { + self.fill(); + } + let out = self.buf[self.pos..self.pos + n].to_vec(); + self.pos += n + 2; + out + } + + /// Read one RESP frame. Used both for command replies and for + /// unsolicited pub/sub pushes on a SUBSCRIBEd connection. + pub fn frame(&mut self) -> Resp { + if self.pos > 0 && self.pos == self.buf.len() { + self.buf.clear(); + self.pos = 0; + } + let line = self.line(); + let (tag, rest) = line.split_at(1); + match tag { + "+" => Resp::Simple(rest.to_string()), + "-" => Resp::Error(rest.to_string()), + ":" => Resp::Int(rest.parse().unwrap_or(0)), + "$" => { + let n: i64 = rest.parse().unwrap_or(-1); + if n < 0 { + Resp::Bulk(None) + } else { + Resp::Bulk(Some(self.exact(n as usize))) + } + } + "*" => { + let n: i64 = rest.parse().unwrap_or(-1); + if n < 0 { + Resp::Array(None) + } else { + let mut items = Vec::with_capacity(n as usize); + for _ in 0..n { + items.push(self.frame()); + } + Resp::Array(Some(items)) + } + } + other => panic!("unexpected RESP tag {other:?} in line {line:?}"), + } + } +} + +pub fn as_array(r: &Resp) -> &[Resp] { + match r { + Resp::Array(Some(items)) => items, + other => panic!("expected array, got {other:?}"), + } +} + +pub fn as_int(r: &Resp) -> i64 { + match r { + Resp::Int(i) => *i, + other => panic!("expected integer, got {other:?}"), + } +} diff --git a/tests/crash_matrix_cross_plane/scenarios.rs b/tests/crash_matrix_cross_plane/scenarios.rs new file mode 100644 index 00000000..634e2934 --- /dev/null +++ b/tests/crash_matrix_cross_plane/scenarios.rs @@ -0,0 +1,967 @@ +//! Shared scenario bodies, parametrized by [`Config`]. Thin `#[test]` +//! wrappers in `tests_prod.rs` / `tests_legacy.rs` / `tests_spot.rs` / +//! `tests_seeded_red.rs` bind each body to a specific config cell and own +//! the `cross_plane__` naming + any `#[ignore = "RED: ..."]` +//! annotation. Bodies themselves always assert the FULL expected contract — +//! whether a given cell is RED or GREEN is discovered by actually running +//! it, never hard-coded here. + +#![allow(dead_code)] + +use std::time::Duration; + +use crate::harness::{self, Config}; +use crate::mixed::{self, MixedPlan}; +use crate::planes::*; +use crate::resp::Conn; + +// --------------------------------------------------------------------------- +// Single-plane isolated baselines +// --------------------------------------------------------------------------- + +/// KV isolated: write a small batch, sync a randomly-chosen prefix boundary +/// (the "periodic markers / kill at a random offset" kill point), kill, +/// verify exactly that prefix survived. `appendonly=no` cells have no +/// durability contract — liveness only. +pub fn kv_isolated(cfg: &Config) { + let dir = harness::unique_dir(&format!("kviso-{}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c = Conn::open(port); + let tag = "kviso"; + const N: u64 = 12; + + if cfg.no_durability_contract() { + for i in 0..N { + kv_set(&mut c, &kv_key(tag, i), &format!("v{i}")); + } + harness::crash(guard, port); + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + // No durability claim; just prove the server serves cleanly. + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + drop(guard2); + return; + } + + let kill_at = harness::jitter_range(1001, 1, N + 1); + let mut written = Vec::new(); + for i in 0..kill_at { + let key = kv_key(tag, i); + let val = format!("v{i}"); + kv_set(&mut c, &key, &val); + written.push((key, val)); + } + // The AOF stores `SET key val` as two SEPARATE RESP bulk strings, never + // the concatenated literal "key=val" — wait for both independently. + let (last_key, last_val) = written.last().expect("kill_at >= 1"); + harness::wait_for_aof_bytes_all_any_shard(&dir, &[last_key.as_bytes(), last_val.as_bytes()]); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + for (key, val) in &written { + assert!( + kv_get_eq(&mut c2, key, val), + "cell {}: KV key {key} must survive up to the synced offset {kill_at}/{N}", + cfg.label + ); + } + drop(guard2); +} + +/// KV spilled: filler pushes the probe keys to the cold tier before they are +/// deleted-and-resynced; only meaningful when disk-offload is enabled. +/// +/// Filler sizing mirrors `crash_recovery_cold_del_resurrection.rs` exactly +/// (`MAXMEMORY_BYTES` = 8 MiB, `FILLER_COUNT` × `FILLER_VALUE_LEN` = +/// 16,000×600B ≈ 9.6 MiB, `SETTLE_AFTER_FILLER` = 8s) rather than an +/// independently-invented ratio. `--maxmemory` is divided evenly across +/// shards (`RuntimeConfig::maxmemory_per_shard`, `src/config.rs`), so at +/// `shards=1` the whole 8 MiB cap lives on one shard (trivially exceeded by +/// unscoped filler) and at `shards=4` each shard's ~2 MiB cap is exceeded by +/// the filler's average ~2.4 MiB/shard scatter — the SAME total-byte ratio +/// the reference suite already validated forces a spill at its own +/// `SHARDS=4` config, so no per-shard-count filler scaling is needed on top +/// of matching the reference's absolute numbers. A prior 3000×512B filler +/// (~1.5 MiB) against a 4 MiB cap never came close to the 0.85× spill +/// threshold at either shard count — confirmed on the VM via +/// `reclamation_cold_segments:0` and zero `heap-*.mpf` files — and the cell +/// silently degenerated to ordinary AOF-replay coverage (already proven by +/// `kv_isolated`), a real false-GREEN (review round 3, task #52 P0). The +/// `count_heap_mpf_files` precondition assert below is the hard backstop: +/// any future load-parameter regression now fails LOUD instead of vacuously +/// passing. +pub fn kv_spilled_isolated(cfg: &Config) { + assert!( + cfg.disk_offload, + "kv_spilled_isolated requires disk-offload enable" + ); + const MAXMEMORY_BYTES: u64 = 8 * 1024 * 1024; + const FILLER_COUNT: u64 = 16_000; + const FILLER_VALUE_LEN: usize = 600; + const SETTLE_AFTER_FILLER: u64 = 8; + + let dir = harness::unique_dir(&format!("kvspill-{}", cfg.label)); + let maxmemory_s = MAXMEMORY_BYTES.to_string(); + let extra = [ + "--maxmemory", + maxmemory_s.as_str(), + "--maxmemory-policy", + "allkeys-lru", + ]; + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &extra); + let mut c = Conn::open(port); + let tag = "kvspill"; + + const PROBES: u64 = 10; + let mut probes = Vec::new(); + for i in 0..PROBES { + let key = kv_key(tag, i); + let val = format!("probe-{i}"); + kv_set(&mut c, &key, &val); + probes.push((key, val)); + } + // Push probes to the cold tier. + kv_spill_filler(&mut c, "kvspillfiller", FILLER_COUNT, FILLER_VALUE_LEN); + std::thread::sleep(Duration::from_secs(SETTLE_AFTER_FILLER)); + + // Hard precondition (review round 3, task #52 P0): without any + // heap-*.mpf files on disk, filler pressure never forced a spill and + // this scenario has no power — it would silently degenerate to + // ordinary AOF-replay coverage, already proven by `kv_isolated`. + let heap_files = harness::count_heap_mpf_files(&dir); + assert!( + heap_files > 0, + "cell {}: precondition failed: no heap-*.mpf files — filler did not \ + force a spill", + cfg.label + ); + + let marker = "XPLANE-KVSPILL-MARKER".to_string(); + kv_set(&mut c, &format!("{{{tag}}}:marker"), &marker); + harness::wait_for_aof_bytes_any_shard(&dir, marker.as_bytes()); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &extra); + let mut c2 = Conn::open(port2); + for (key, val) in &probes { + assert!( + kv_get_eq(&mut c2, key, val), + "cell {}: spilled KV probe {key} must survive kill-9 (task #41)", + cfg.label + ); + } + drop(guard2); +} + +/// Graph isolated: structural writes (ADDNODE/ADDEDGE + Cypher CREATE/SET/ +/// DELETE) + GraphTemporal (`TEMPORAL.INVALIDATE`). SEEDED RED in legacy +/// mode (brief §1.4 / PR #288 changelog) — the body still asserts the full +/// contract; the `#[ignore]` annotation lives on the legacy-cell wrapper. +pub fn graph_isolated(cfg: &Config) { + let dir = harness::unique_dir(&format!("graphiso-{}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c = Conn::open(port); + let name = graph_name("graphiso"); + + graph_create(&mut c, &name); + let handles: Vec = (0..6) + .map(|i| graph_addnode(&mut c, &name, "N", i)) + .collect(); + for i in 0..6 { + graph_addedge(&mut c, &name, handles[i], handles[(i + 1) % 6]); + } + graph_write(&mut c, &name, "CREATE (n:W {id: 100})"); + graph_write(&mut c, &name, "MATCH (n:N {id: 3}) SET n.score = 42"); + graph_write(&mut c, &name, "MATCH (n:N {id: 5}) DELETE n"); + + // GraphTemporal: invalidate node 0. + let node0_id = match graph_query_ids(&mut c, &name, "MATCH (n:N {id: 0}) RETURN n.id").len() { + 1 => handles[0].to_string(), + _ => panic!("node id=0 must exist before invalidation"), + }; + temporal_invalidate_node(&mut c, &node0_id, &name); + + if cfg.no_durability_contract() { + harness::crash(guard, port); + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + drop(guard2); + return; + } + + let marker = "XPLANE-GRAPH-MARKER".to_string(); + graph_addnode_marker(&mut c, &name, &marker); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker.as_bytes()); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert_eq!( + graph_query_ids(&mut c2, &name, "MATCH (n:N) RETURN n.id"), + (0..5).collect(), + "cell {}: structural writes (ADDNODE/ADDEDGE + Cypher DELETE) must \ + survive kill-9", + cfg.label + ); + assert_eq!( + graph_query_ids(&mut c2, &name, "MATCH (n:W) RETURN n.id"), + std::collections::BTreeSet::from([100]), + "cell {}: Cypher CREATE must survive kill-9", + cfg.label + ); + assert_eq!( + graph_query_ids(&mut c2, &name, "MATCH (n:N {id: 3}) RETURN n.score"), + std::collections::BTreeSet::from([42]), + "cell {}: Cypher SET must survive kill-9", + cfg.label + ); + // VALID_AT is a separate GRAPH.QUERY argument, not part of the Cypher + // text (see `graph_query_ids_valid_at`'s doc). + let visible_far_future = graph_query_ids_valid_at( + &mut c2, + &name, + "MATCH (n:N {id: 0}) RETURN n.id", + "9000000000000", + ); + assert!( + visible_far_future.is_empty(), + "cell {}: TEMPORAL.INVALIDATE (GraphTemporal) must survive kill-9 — \ + node 0 must NOT be visible at a far-future VALID_AT after restart, \ + got {visible_far_future:?}", + cfg.label + ); + // Positive control (review round 3, P2): without this, the negative + // check above would PASS vacuously if VALID_AT plumbing itself were + // broken and always returned empty regardless of invalidation state. + // Node 1 was never invalidated or deleted, so it MUST still be visible + // at the same far-future VALID_AT. + let node1_visible_far_future = graph_query_ids_valid_at( + &mut c2, + &name, + "MATCH (n:N {id: 1}) RETURN n.id", + "9000000000000", + ); + assert_eq!( + node1_visible_far_future, + std::collections::BTreeSet::from([1]), + "cell {}: VALID_AT positive control — node 1 (never invalidated) \ + must be visible at a far-future VALID_AT", + cfg.label + ); + drop(guard2); +} + +/// Vector isolated: FT.CREATE + HSET batch, synced prefix, kill, verify. +pub fn vector_isolated(cfg: &Config) { + let dir = harness::unique_dir(&format!("veciso-{}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c = Conn::open(port); + let tag = "veciso"; + let idx = vec_index_name(tag); + let prefix = format!("{{{tag}}}:vec:"); + ft_create_hnsw(&mut c, &idx, &prefix); + + if cfg.no_durability_contract() { + for i in 0..10u64 { + hset_vec(&mut c, &format!("{prefix}{i}"), &vec_blob(i)); + } + harness::crash(guard, port); + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + drop(guard2); + return; + } + + const N: u64 = 10; + let kill_at = harness::jitter_range(2002, 1, N + 1); + let mut written = Vec::new(); + for i in 0..kill_at { + let key = format!("{prefix}{i}"); + let blob = vec_blob(i); + hset_vec(&mut c, &key, &blob); + written.push((key, blob)); + } + // Vector durability rides the same path as any other HSET (brief §1.2: + // "normal Command WAL/AOF replay of the hash that gets auto-indexed") — + // wait on the AOF, not wal-v3. + harness::wait_for_aof_bytes_any_shard(&dir, written.last().expect("kill_at >= 1").0.as_bytes()); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + for (key, blob) in &written { + assert!( + vec_search_contains(&mut c2, &idx, 5, blob, key), + "cell {}: vector key {key} must survive up to the synced offset \ + {kill_at}/{N}", + cfg.label + ); + } + drop(guard2); +} + +/// WS isolated: WS CREATE, synced, kill, verify WS LIST. +pub fn ws_isolated(cfg: &Config) { + let dir = harness::unique_dir(&format!("wsiso-{}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c = Conn::open(port); + let ws_name = "wsiso-workspace"; + let ws_id = ws_create(&mut c, ws_name); + + if cfg.no_durability_contract() { + harness::crash(guard, port); + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + drop(guard2); + return; + } + + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, ws_name.as_bytes()); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert!( + ws_list_contains(&mut c2, &ws_id), + "cell {}: WS CREATE must survive kill-9", + cfg.label + ); + drop(guard2); +} + +/// MQ isolated: stream content, ACK-subset, cursor resume, DLQ routing, +/// trigger registration (pattern: `crash_recovery_mq_effects.rs`, condensed). +pub fn mq_isolated(cfg: &Config) { + let dir = harness::unique_dir(&format!("mqiso-{}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c = Conn::open(port); + let ordersq = "mqiso-ordersq"; + let dlqtestq = "mqiso-dlqtestq"; + + mq_create(&mut c, ordersq, 0); + for i in 1..=5u32 { + mq_push(&mut c, ordersq, "seq", &i.to_string()); + } + let popped = mq_pop(&mut c, ordersq, 3); + assert_eq!(popped.len(), 3, "must claim exactly 3 messages"); + let claimed_ids: Vec> = popped + .iter() + .map(|entry| match entry { + crate::resp::Resp::Array(Some(cells)) => match &cells[0] { + crate::resp::Resp::Bulk(Some(id)) => id.clone(), + other => panic!("unexpected id cell: {other:?}"), + }, + other => panic!("unexpected pop entry: {other:?}"), + }) + .collect(); + let acked = mq_ack(&mut c, ordersq, &claimed_ids[0..2]); + assert_eq!(acked, 2, "must ack exactly 2 messages"); + + mq_create(&mut c, dlqtestq, 1); + mq_push(&mut c, dlqtestq, "f", "a"); + mq_push(&mut c, dlqtestq, "f", "b"); + let dlq_popped = mq_pop(&mut c, dlqtestq, 2); + assert_eq!( + dlq_popped.len(), + 0, + "MAXDELIVERY 1 routes both straight to DLQ" + ); + assert_eq!(mq_dlqlen(&mut c, dlqtestq), 2); + + if cfg.no_durability_contract() { + harness::crash(guard, port); + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + drop(guard2); + return; + } + + let marker = "XPLANE-MQ-MARKER".to_string(); + mq_push(&mut c, ordersq, "final", &marker); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker.as_bytes()); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert_eq!( + xlen(&mut c2, ordersq), + 6, + "cell {}: MQ.PUSH'd messages (incl. the sync marker) must survive kill-9", + cfg.label + ); + let resumed = mq_pop(&mut c2, ordersq, 10); + assert_eq!( + resumed.len(), + 3, + "cell {}: cursor must resume after msg 3 (msgs 4, 5, marker remain)", + cfg.label + ); + assert_eq!( + mq_dlqlen(&mut c2, dlqtestq), + 2, + "cell {}: DLQ-routed entries must survive kill-9", + cfg.label + ); + drop(guard2); +} + +/// Cross-store TXN, split into two independently-gated functions (review +/// round 3, P1: the two halves have UNRELATED root causes and must not +/// share one `red_guard` — see each function's own doc): +/// +/// - [`txn_isolated_committed`]: a committed transaction spanning +/// KV+graph must survive kill-9 exactly like a non-transactional write +/// (extends `sharded_multi_exec_durability.rs`'s single-plane proof +/// cross-plane). RED on `prod_s1`/`prod_s4` — task #52. +/// - [`txn_isolated_atomicity`]: a transaction queued but killed BEFORE it +/// commits must apply NOTHING (atomicity — the "kill mid cross-store TXN" +/// kill point). GREEN on `prod_s1`/`prod_s4` (unrelated to task #52) — +/// gating it behind the SAME `red_guard` as the committed-exec half would +/// skip a working, unrelated regression tripwire by default. Still RED on +/// `legacy_yes_s1` (same pre-existing legacy-graph-WAL-replay gap as +/// every other legacy graph cell — `GRAPH.CREATE` itself never +/// reconstructs there, so any post-restart `GRAPH.QUERY` errors "graph +/// not found" regardless of transaction semantics; empirically confirmed, +/// not assumed — see `tests_legacy.rs`). +/// +/// Each half runs its OWN spawn/crash/restart cycle (task #52 review round: +/// interleaving them in one server lifetime inserted extra round-trips — +/// queuing the atomicity round's transaction — between the committed-exec +/// round's durability sync-wait and the kill, an avoidable timing confound +/// for a RED-cell repro that needs to be deterministic). +/// +/// REAL RED FINDING (kernel M3, task #52 — reviewed and confirmed NOT a +/// harness artifact): on `prod_s1`/`prod_s4` (checkpoint-backed, +/// appendonly=yes), round (1)'s KV leg survives kill-9 but the graph leg +/// does NOT. PR #247's `persist_txn_aof` covers the KV leg of a cross-store +/// transaction; the graph leg's node-create — queued and committed inside +/// the SAME transaction — is not durably replayed. Ruled out as a false +/// negative from this suite's own known harness bugs: the verification +/// query (`MATCH (n:TxnN {id: 1}) RETURN n.id`) matches on the property +/// VALUE, not an internal node handle (unaffected by the mixed-workload +/// node-id-renumbering bug), does not use `VALID_AT` (unaffected by that +/// parse-error bug), and both the KV-leg AOF wait and the graph-leg wal-v3 +/// wait (via `graph_addnode_marker`, issued and synced AFTER the +/// transaction commits) complete successfully before the kill — so the +/// graph leg's own WAL flush tick had run by kill time, yet the node is +/// still gone post-restart. +/// +/// Determinism conditions (task #52 — the repro is now shaped to remove the +/// two plausible sources of flakiness): (a) `--checkpoint-timeout 3600` +/// rules out a periodic checkpoint Finalize "accidentally" persisting the +/// graph leg via an unrelated snapshot in the window between the commit and +/// the kill (the flag's own default is already 300s vs. this scenario's +/// sub-second runtime, so this mostly documents intent rather than changing +/// behavior — but removes the ambiguity outright); (b) the kill happens +/// IMMEDIATELY after the sync-wait completes, with NO further commands +/// issued in between (round (2)'s queued-uncommitted transaction now runs +/// in its own separate server lifetime, never sharing a process with round +/// (1)). Verified 3× consecutive reproduction at both shards=1 and +/// shards=4 (kernel M3 stage 1 hardening pass) after this restructuring. +/// +/// Minimal repro: +/// ```text +/// MULTI +/// SET k v +/// GRAPH.ADDNODE g Label id 1 +/// EXEC +/// +/// kill -9 IMMEDIATELY (no further commands); restart +/// GET k -> v (KV leg survives) +/// GRAPH.QUERY g "MATCH (n:Label {id: 1}) RETURN n.id" -> empty (graph leg lost) +/// ``` +pub fn txn_isolated_committed(cfg: &Config) { + let dir = harness::unique_dir(&format!("txniso-committed-{}", cfg.label)); + // Long checkpoint-timeout: see this function's own "Determinism + // conditions" doc above — rules out a periodic checkpoint as the rescuer. + let extra = ["--checkpoint-timeout", "3600"]; + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &extra); + let mut c = Conn::open(port); + let graph = graph_name("txniso"); + graph_create(&mut c, &graph); + + let committed_key = "{txniso}:committed"; + let commit_reply = + txn_kv_graph_write(&mut c, committed_key, "committed-val", &graph, "TxnN", 1); + match commit_reply { + crate::resp::Resp::Array(Some(items)) => { + assert_eq!(items.len(), 2, "commit returns 2 replies") + } + other => panic!("commit must return an array: {other:?}"), + } + + if cfg.no_durability_contract() { + harness::crash(guard, port); + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &extra); + let mut c2 = Conn::open(port2); + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + drop(guard2); + return; + } + + // Two separate log files, two separate writers (KV leg -> AOF, graph + // leg -> wal-v3) — each needs its own same-family sync check; waiting + // on only one does not prove the other's flush tick has also run. The + // committed value proves the KV leg; a follow-up graph marker (issued + // AFTER the commit, same connection) proves the graph leg, since + // wal-v3 appends are strictly ordered per shard. + harness::wait_for_aof_bytes_any_shard(&dir, b"committed-val"); + let graph_marker = "XPLANE-TXN-GRAPH-SYNC".to_string(); + graph_addnode_marker(&mut c, &graph, &graph_marker); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, graph_marker.as_bytes()); + // Kill IMMEDIATELY — no further commands between the sync-wait + // completing and the kill (task #52 determinism requirement). + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &extra); + let mut c2 = Conn::open(port2); + assert!( + kv_get_eq(&mut c2, committed_key, "committed-val"), + "cell {}: committed cross-store transaction (KV leg) must survive \ + kill-9", + cfg.label + ); + assert_eq!( + graph_query_ids(&mut c2, &graph, "MATCH (n:TxnN {id: 1}) RETURN n.id"), + std::collections::BTreeSet::from([1]), + "cell {}: committed cross-store transaction (graph leg) must \ + survive kill-9", + cfg.label + ); + drop(guard2); +} + +/// A transaction queued but killed BEFORE it commits must apply NOTHING +/// (atomicity — the "kill mid cross-store TXN" kill point). Own +/// spawn/crash/restart cycle, fully independent of +/// [`txn_isolated_committed`] — see that function's doc for why. +/// +/// Deliberately NOT gated behind the task #52 `red_guard` on +/// `prod_s1`/`prod_s4` (review round 3, P1): this is a genuinely different +/// claim from the committed half (nothing was ever queued to durably apply, +/// vs. a committed transaction losing a leg) and is GREEN there — gating it +/// behind the same reason as the committed half would skip a working, +/// unrelated regression tripwire by default. IS gated on `legacy_yes_s1` +/// via `LEGACY_GRAPH_RED_REASON` (see `tests_legacy.rs`): `graph_create` +/// itself never survives a restart in legacy mode (the same pre-existing +/// WAL-v3-replay gap every other legacy graph cell hits), so the graph-leg +/// assertion below panics on "ERR graph not found" rather than evaluating +/// the atomicity claim at all — confirmed on the VM, not assumed. +/// +/// The initial `GRAPH.CREATE` MUST sync-wait before the MULTI is queued and +/// the kill fires (an earlier draft of this function skipped that wait, +/// treating "kill as close to queued as possible" as license to skip ALL +/// sync-waiting — that was wrong): without it, the graph's own existence +/// races the same WAL flush tick as everything else, so a post-restart +/// `GRAPH.QUERY` can error "ERR graph not found" for a reason that has +/// NOTHING to do with the atomicity claim under test, on every config +/// (confirmed on the VM — the earlier unwaited version failed identically +/// on `prod_s1`/`prod_s4`/`legacy_yes_s1` alike, which was the tell that it +/// was a harness bug, not a finding). Waiting for `GRAPH.CREATE` does not +/// weaken the kill point: the transaction itself is still queued and killed +/// with zero settle time. +pub fn txn_isolated_atomicity(cfg: &Config) { + let dir = harness::unique_dir(&format!("txniso-queued-{}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c = Conn::open(port); + let graph = graph_name("txnisoq"); + graph_create(&mut c, &graph); + if !cfg.no_durability_contract() { + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, graph.as_bytes()); + } + + let queued_key = "{txnisoq}:queued-uncommitted"; + c.pipeline_s(&[ + vec!["MULTI"], + vec!["SET", queued_key, "must-not-persist"], + vec!["GRAPH.ADDNODE", &graph, "TxnN", "id", "2"], + ]); + // Deliberately never committed, no wait for the transaction itself — + // kill as close to "queued" as possible (only the PRECEDING + // GRAPH.CREATE is sync-waited, see above). + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + + if cfg.no_durability_contract() { + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + drop(guard2); + return; + } + + assert!( + matches!( + c2.cmd_s(&["GET", queued_key]), + crate::resp::Resp::Bulk(None) + ), + "cell {}: a transaction queued but never committed must apply \ + NOTHING (KV leg) — atomicity violated if this key exists", + cfg.label + ); + assert_eq!( + graph_query_ids(&mut c2, &graph, "MATCH (n:TxnN {id: 2}) RETURN n.id"), + std::collections::BTreeSet::new(), + "cell {}: a transaction queued but never committed must apply \ + NOTHING (graph leg) — atomicity violated if this node exists", + cfg.label + ); + drop(guard2); +} + +// --------------------------------------------------------------------------- +// Concurrent mixed-workload scenarios (the genuinely new coverage) +// --------------------------------------------------------------------------- + +/// Concurrent writes across KV/graph/vector/WS/MQ on ONE shard, a randomized +/// synced prefix per plane, kill, verify. +pub fn mixed_synced(cfg: &Config) { + for iter in 0..crate::soak_iters() { + let dir = harness::unique_dir(&format!("mixedsync-{}-{iter}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let tag = format!("mixedsync{iter}"); + let sync_fraction = if cfg.no_durability_contract() { + 0.0 + } else { + harness::jitter_range(3003 + iter as u64, 3, 10) as f64 / 10.0 + }; + let truth = mixed::run_mixed_workload_synced( + &dir, + port, + cfg.shards, + &tag, + MixedPlan::default(), + sync_fraction, + ); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + if cfg.no_durability_contract() { + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + } else { + mixed::assert_mixed_truth_recovered(&mut c2, &truth); + } + drop(guard2); + } +} + +/// Mixed workload + kill in the probabilistic checkpoint-Finalize window +/// (see the module doc's honesty note — this is NOT per-step fault +/// injection). Only meaningful when `disk_offload` is enabled (Finalize is +/// the checkpoint-backed-mode-only code path, brief §1.1). +/// +/// REAL RED FINDING (kernel M3, NEW — found via `MOON_CRASH_MATRIX_ITERS` +/// soak, reviewed and confirmed NOT a harness artifact): on both +/// `prod_s1` and `prod_s4`, some kill offsets in the 0-150ms +/// post-`BGSAVE` window cause TOTAL loss of the graph plane's content — +/// not partial loss of the unsynced tail, but the FULL set of already +/// wal-v3-synced nodes (the entire `MixedPlan::default()` graph batch, +/// confirmed durable via `graph_addnode_marker`'s wal-v3 wait BEFORE +/// `BGSAVE` is even issued) comes back completely EMPTY. KV, vector, WS, +/// and MQ all survive in the same run (`assert_mixed_truth_recovered` +/// checks KV first, unconditionally, before the graph check that panics — +/// KV never fails here). Reproduced via `MOON_CRASH_MATRIX_ITERS=20` in +/// isolation (not full-suite — the full-suite run's contention only +/// makes this window easier to hit, it isn't the cause): prod_s1 hit at +/// iteration 11/20, prod_s4 at iteration 7/20 — a real, load-sensitive but +/// unambiguous timing window, not a one-off VM hiccup. Consistent with +/// (though not proven to be exactly) a checkpoint `Finalize` step ordering +/// gap between `persist_graph_at_checkpoint`'s snapshot write and +/// `wal.recycle_segments_before`'s WAL-v3 segment recycle +/// (`src/shard/persistence_tick.rs` ~1182-1301) — if recycle can run (or +/// its effect become visible after a kill) before the graph snapshot is +/// durably on disk, the WAL copy of the data is gone AND the snapshot +/// never had it either. Not proven at the step level (no fault-injection +/// hooks exist — see the module doc's `kill_mid_checkpoint` honesty note); +/// this is exactly the class of finding that note anticipated soak testing +/// would surface. This is a strong candidate P0 for kernel M3 stage 2 (K2, +/// the unified per-shard floor register) to close, not this stage's job to +/// fix. +pub fn mixed_mid_checkpoint(cfg: &Config) { + assert!( + cfg.disk_offload, + "mixed_mid_checkpoint requires disk-offload enable" + ); + for iter in 0..crate::soak_iters() { + let dir = harness::unique_dir(&format!("mixedckpt-{}-{iter}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let tag = format!("mixedckpt{iter}"); + let truth = mixed::run_mixed_workload_synced( + &dir, + port, + cfg.shards, + &tag, + MixedPlan::default(), + 1.0, + ); + + let mut c = Conn::open(port); + let before = std::fs::read(harness::control_file_path(&dir)).ok(); + // BGSAVE replies `+Background saving started\r\n` (see + // `command::persistence::bgsave_start_sharded`), NOT `+OK` — a + // single call, checked once, accepting either a simple-string + // acknowledgement or the (benign, racy) already-in-progress error. + let bgsave_reply = c.cmd_s(&["BGSAVE"]); + assert!( + matches!(bgsave_reply, crate::resp::Resp::Simple(_)) + || matches!(&bgsave_reply, crate::resp::Resp::Error(e) + if e.to_lowercase().contains("already in progress")), + "cell {}: BGSAVE must be accepted (or report already-in-progress), \ + got {bgsave_reply:?}", + cfg.label + ); + // Randomized short sleep across the window a small checkpoint + // typically finalizes in — see module doc's honesty note. + let jitter_ms = harness::jitter_range(4004 + iter as u64, 0, 150); + std::thread::sleep(Duration::from_millis(jitter_ms)); + let _ = &before; // best-effort signal only, not asserted pre-kill + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + mixed::assert_mixed_truth_recovered(&mut c2, &truth); + drop(guard2); + } +} + +/// Mixed workload + kill mid-Pass-C-recycle: tiny `--wal-segment-size` / +/// `--max-wal-size` force several autovacuum Pass C ticks within seconds +/// (pattern: `crash_recovery_wal_recycle_legacy.rs`). This is the kill +/// point most directly relevant to K2 (Stage 2) — it targets exactly the +/// recycle-floor invariant the brief's whole milestone is about. +pub fn mixed_mid_pass_c(cfg: &Config) { + for iter in 0..crate::soak_iters() { + let dir = harness::unique_dir(&format!("mixedpassc-{}-{iter}", cfg.label)); + let extra = [ + "--wal-segment-size", + "32kb", + "--max-wal-size", + "96kb", + "--autovacuum-interval-secs", + "1", + ]; + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &extra); + let tag = format!("mixedpassc{iter}"); + + // Sync a small mixed batch first so there is real plane history in + // the earliest (soon-to-be-recycle-eligible) segment. + let truth = mixed::run_mixed_workload_synced( + &dir, + port, + cfg.shards, + &tag, + MixedPlan { + kv_n: 5, + graph_n: 5, + vec_n: 5, + mq_n: 5, + }, + 1.0, + ); + + // Blow past the tiny WAL ceiling with padded GRAPH writes (Pass C / + // `--max-wal-size` pressure is a wal-v3-only concept — plain KV SETs + // ride the AOF and would never grow wal-v3 at all, so padding must + // land on a plane that actually appends there; pattern: + // `crash_recovery_wal_recycle_legacy.rs`), then give the 1s + // autovacuum tick several chances to run before killing. + let mut c = Conn::open(port); + let padding = "x".repeat(2048); + let pad_graph = graph_name(&format!("{tag}pad")); + graph_create(&mut c, &pad_graph); + for i in 0..200u64 { + c.cmd_s(&[ + "GRAPH.ADDNODE", + &pad_graph, + "Bulk", + "pad", + &padding, + "id", + &i.to_string(), + ]); + } + std::thread::sleep(Duration::from_secs(5)); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &extra); + let mut c2 = Conn::open(port2); + if cfg.no_durability_contract() { + assert!(matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_))); + } else { + mixed::assert_mixed_truth_recovered(&mut c2, &truth); + } + drop(guard2); + } +} + +/// Genuinely-concurrent mid-burst kill: no synchronization at all. Proves +/// only "no corruption" (every plane still answers well-formed queries) — +/// see `mixed::start_concurrent_burst`'s doc for why no content assertion +/// is possible here. +pub fn concurrent_burst_no_corruption(cfg: &Config) { + for iter in 0..crate::soak_iters() { + // Scoped to this ONE iteration (see the guard's own doc for why + // process-wide would be unsafe in a 37-test binary) — the burst + // writer threads dying at kill -9 with connection-reset/mid-frame + // panics is expected noise, not signal; this keeps stderr readable + // without risking silently swallowing a genuine panic elsewhere. + let _panic_filter = mixed::BenignDisconnectPanicFilter::install(); + let dir = harness::unique_dir(&format!("burst-{}-{iter}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let tag = format!("burst{iter}"); + mixed::start_concurrent_burst(port, &tag); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert!( + matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_)), + "cell {}: server must come back up cleanly after a mid-burst kill", + cfg.label + ); + // No-corruption spot checks: every plane must either answer cleanly + // OR fail with the ordinary "this schema object doesn't exist" + // error — that is a legitimate outcome here, not corruption, since + // `start_concurrent_burst` deliberately does NOT sync-wait the + // one-shot GRAPH.CREATE/FT.CREATE/MQ.CREATE calls before returning + // (nothing in this scenario is synced — see its own doc). A crash + // landing before those schema ops reach disk is a real, valid + // outcome, not a bug. What must NEVER happen is any OTHER error — + // that would mean the crash corrupted a plane's on-disk structures + // badly enough that even a fresh boot can't parse them, which is + // the actual invariant this scenario exists to check. + assert!( + !is_unexpected_plane_error( + &c2.cmd_s(&[ + "GRAPH.QUERY", + &graph_name(&tag), + "MATCH (n:BurstN) RETURN count(n)" + ]), + &["ERR graph not found"], + ), + "cell {}: graph plane must not error (beyond ordinary \ + not-found) after a mid-burst kill", + cfg.label + ); + // `FT.SEARCH ... "*" LIMIT 0 1` is not valid syntax for this engine + // at all (it requires a `*=>[KNN k @field $B]` clause — see + // `planes::vec_search_contains`) and would error unconditionally + // regardless of corruption, so it cannot serve as a no-corruption + // probe. `FT.INFO` needs no query vector and answers cleanly with + // just "Unknown Index name" when the index was never durable — + // exactly the same benign-not-found shape as the graph/MQ checks. + // Exact strings (`src/command/vector_search/ft_info.rs`, + // `ft_aggregate.rs`/`hybrid.rs`/`ft_search/dispatch.rs`), not bare + // substrings (review round 3, P2): a substring match on "unknown + // index" would ALSO classify a hypothetical future corruption + // message like "unknown index format" as benign — every not-found + // error this engine emits is a short, fixed, exact literal (never a + // longer message with a variable/diagnostic suffix), so exact + // equality is strictly correct here and closes that door. + assert!( + !is_unexpected_plane_error( + &c2.cmd_s(&["FT.INFO", &vec_index_name(&tag)]), + &[ + "ERR unknown index", + "Unknown Index name", + "ERR no such index" + ], + ), + "cell {}: vector plane must not error (beyond ordinary \ + not-found) after a mid-burst kill", + cfg.label + ); + // XLEN on a never-durable queue replies Int(0), so no benign + // not-found error strings need allowlisting — any error reply at + // all is an unexpected MQ-plane failure. (The previous + // `as_int_or_zero(..) >= 0` form was vacuous: it mapped error + // replies to 0 and could never fail.) + let mq_reply = c2.cmd_s(&["XLEN", &mq_queue_name(&tag)]); + assert!( + !is_unexpected_plane_error(&mq_reply, &[]), + "cell {}: MQ plane must not error after a mid-burst kill (got {mq_reply:?})", + cfg.label + ); + drop(guard2); + } +} + +/// True if `r` is an error reply whose text does NOT EXACTLY match (modulo +/// case/whitespace) any of the benign `allowed_exact` messages — i.e. a +/// genuine, unexpected plane error rather than an ordinary "this schema +/// object was never durable before the kill" not-found response. +/// +/// Exact matching, not `.contains()` (review round 3, P2): every not-found +/// error this engine emits (`"ERR graph not found"`, `"ERR unknown +/// index"`, `"Unknown Index name"`, `"ERR no such index"`) is a short, +/// fixed, literal `Frame::Error` with no variable suffix — confirmed via a +/// full-repo grep, not assumed. A substring check would ALSO classify a +/// hypothetical future corruption message like `"unknown index format"` as +/// benign purely because it happens to start with the same words; exact +/// equality closes that door without weakening today's coverage (every +/// caller's allowlist already contains the exact real strings). +fn is_unexpected_plane_error(r: &crate::resp::Resp, allowed_exact: &[&str]) -> bool { + match r { + crate::resp::Resp::Error(msg) => { + let lower = msg.trim().to_lowercase(); + !allowed_exact + .iter() + .any(|needle| lower == needle.to_lowercase()) + } + _ => false, + } +} + +/// `appendonly=no` cells: no plane has a durability contract at all +/// (`persistence_dir` never constructed). Assert the documented no-RPO +/// contract explicitly — writes across every plane, kill -9, restart must +/// be clean (no crash-loop, no corrupted half-state) with an EMPTY +/// keyspace, never a silent partial-persist. +pub fn no_durability_liveness(cfg: &Config) { + assert!(cfg.no_durability_contract()); + let dir = harness::unique_dir(&format!("nodur-{}", cfg.label)); + let (guard, port) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c = Conn::open(port); + let tag = "nodur"; + + kv_set(&mut c, &kv_key(tag, 0), "v0"); + let graph = graph_name(tag); + graph_create(&mut c, &graph); + graph_addnode(&mut c, &graph, "N", 0); + let idx = vec_index_name(tag); + ft_create_hnsw(&mut c, &idx, &format!("{{{tag}}}:vec:")); + hset_vec(&mut c, &format!("{{{tag}}}:vec:0"), &vec_blob(0)); + let mq = mq_queue_name(tag); + mq_create(&mut c, &mq, 0); + mq_push(&mut c, &mq, "f", "v"); + let ws_id = ws_create(&mut c, "nodur-ws"); + std::thread::sleep(Duration::from_millis(500)); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]); + let mut c2 = Conn::open(port2); + assert!( + matches!(c2.cmd_s(&["PING"]), crate::resp::Resp::Simple(_)), + "cell {}: server must restart cleanly even with zero WAL history", + cfg.label + ); + assert!( + matches!( + c2.cmd_s(&["GET", &kv_key(tag, 0)]), + crate::resp::Resp::Bulk(None) + ), + "cell {}: appendonly=no must not silently persist KV — this cell's \ + entire contract is documented no-RPO, a survivor here would be a \ + worse bug (undocumented partial persistence)", + cfg.label + ); + let _ = ws_id; // WS is process-global; not expected to survive a restart either way here + drop(guard2); +} diff --git a/tests/crash_matrix_cross_plane/tests_legacy.rs b/tests/crash_matrix_cross_plane/tests_legacy.rs new file mode 100644 index 00000000..2f922bed --- /dev/null +++ b/tests/crash_matrix_cross_plane/tests_legacy.rs @@ -0,0 +1,121 @@ +//! Legacy cells (`--disk-offload disable`), shards=1 only per the brief +//! ("legacy mode has no cross-shard novelty for this suite's purpose"). +//! +//! 5 cells here are RED, all one root cause: PR #288's own changelog states +//! graph command replay from WAL v3 does not reconstruct the graph in +//! legacy mode even with zero Pass C interference — a pre-existing gap +//! unrelated to WAL recycling, deliberately not asserted by +//! `tests/crash_recovery_wal_recycle_legacy.rs` for the same reason. This +//! suite DOES assert it (the shared `scenarios::graph_isolated` / +//! `mixed_synced` / `mixed_mid_pass_c` / `txn_isolated_committed` / +//! `txn_isolated_atomicity` bodies make no exceptions), so each surfaces the +//! same "ERR graph not found" — `harness::red_guard` skips them by default +//! (set `MOON_CRASH_MATRIX_RED=1` to reproduce); see that function's doc for +//! why a plain `#[ignore = "RED: ..."]` annotation does NOT achieve this +//! (`--ignored` runs ignored tests). + +use crate::harness::{self, Config}; +use crate::scenarios; + +const LEGACY_GRAPH_RED_REASON: &str = "legacy-mode (--disk-offload disable) graph WAL replay \ + does not reconstruct the graph even with zero Pass C interference — \ + PR #288 / task #43 changelog; tests/crash_recovery_wal_recycle_legacy.rs \ + deliberately does not assert this either. Pre-existing gap, not this \ + stage's job to fix — tracked separately."; + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_legacy_yes_s1_kv_isolated() { + scenarios::kv_isolated(&Config::LEGACY_YES_S1); +} + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_legacy_yes_s1_graph_isolated() { + if !harness::red_guard(LEGACY_GRAPH_RED_REASON) { + return; + } + scenarios::graph_isolated(&Config::LEGACY_YES_S1); +} + +#[test] +#[ignore] +fn cross_plane_legacy_yes_s1_vector_isolated() { + scenarios::vector_isolated(&Config::LEGACY_YES_S1); +} + +#[test] +#[ignore] +fn cross_plane_legacy_yes_s1_ws_isolated() { + scenarios::ws_isolated(&Config::LEGACY_YES_S1); +} + +#[test] +#[ignore] +fn cross_plane_legacy_yes_s1_mq_isolated() { + scenarios::mq_isolated(&Config::LEGACY_YES_S1); +} + +// One root cause (legacy-mode graph WAL replay never reconstructs the +// graph — same gap as `cross_plane_legacy_yes_s1_graph_isolated`) surfaces +// in the 4 cells below too: each hits the same "ERR graph not found" the +// isolated graph scenario does. This includes BOTH txn_isolated halves +// (review round 3, P1 restructuring, task #52) — unlike prod_s1/prod_s4, +// the atomicity half is NOT independently green here: `graph_create` itself +// never survives a restart in legacy mode, so the atomicity scenario's +// graph-leg assertion panics on "ERR graph not found" before it can ever +// evaluate the atomicity claim. Confirmed on the VM, not assumed. + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_legacy_yes_s1_txn_isolated_committed() { + if !harness::red_guard(LEGACY_GRAPH_RED_REASON) { + return; + } + scenarios::txn_isolated_committed(&Config::LEGACY_YES_S1); +} + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_legacy_yes_s1_txn_isolated_atomicity() { + if !harness::red_guard(LEGACY_GRAPH_RED_REASON) { + return; + } + scenarios::txn_isolated_atomicity(&Config::LEGACY_YES_S1); +} + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_legacy_yes_s1_mixed_all_planes_synced() { + if !harness::red_guard(LEGACY_GRAPH_RED_REASON) { + return; + } + scenarios::mixed_synced(&Config::LEGACY_YES_S1); +} + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_legacy_yes_s1_mixed_all_planes_mid_pass_c() { + if !harness::red_guard(LEGACY_GRAPH_RED_REASON) { + return; + } + scenarios::mixed_mid_pass_c(&Config::LEGACY_YES_S1); +} + +#[test] +#[ignore] +fn cross_plane_legacy_yes_s1_mixed_all_planes_concurrent_burst_no_corruption() { + scenarios::concurrent_burst_no_corruption(&Config::LEGACY_YES_S1); +} + +// --------------------------------------------------------------------------- +// legacy_no_s1: appendonly=no + disk-offload=disable — NO durability contract +// on any plane (persistence_dir never constructed). Liveness/no-corruption +// only, per the brief's "vacuous by design, not a bug" note. +// --------------------------------------------------------------------------- + +#[test] +#[ignore] +fn cross_plane_legacy_no_s1_no_durability_contract_liveness() { + scenarios::no_durability_liveness(&Config::LEGACY_NO_S1); +} diff --git a/tests/crash_matrix_cross_plane/tests_prod.rs b/tests/crash_matrix_cross_plane/tests_prod.rs new file mode 100644 index 00000000..12820263 --- /dev/null +++ b/tests/crash_matrix_cross_plane/tests_prod.rs @@ -0,0 +1,193 @@ +//! Production cells: the full plane×workload matrix on the 2 configs the +//! brief calls out as "the config that matters most" (disk-offload enable, +//! appendonly yes, shards {1,4}). Naming: `cross_plane__` +//! (flat function names, per the brief's explicit suggestion — avoids the +//! `g1_` prefix collision `crash_recovery_graph_durability.rs` already has). + +use crate::harness::{self, Config}; +use crate::scenarios; + +const TXN_GRAPH_LEG_RED_REASON: &str = "cross-store TXN graph leg not durable, task #52. A \ + committed cross-store transaction (SET+GRAPH.ADDNODE) survives kill-9 \ + on the KV leg (PR #247's persist_txn_aof) but the graph leg is lost — \ + reviewed and confirmed not a harness artifact (property-value query, \ + no VALID_AT, both legs' WAL flush confirmed synced before the kill), \ + and confirmed deterministic 3x consecutive at both shards=1 and \ + shards=4 after the task #52 hardening pass (own crash cycle per \ + round, --checkpoint-timeout 3600, kill immediately post-sync-wait). \ + Minimal repro in scenarios::txn_isolated_committed's doc comment. \ + P1 finding for kernel M3 — tracked separately, not this stage's job to \ + fix. Gates ONLY the committed-transaction half (review round 3, P1): \ + the atomicity half (`cross_plane_*_txn_isolated_atomicity`) is a \ + genuinely different, unrelated, GREEN claim and runs ungated."; + +/// NEW finding (kernel M3, not a harness artifact — see +/// `scenarios::mixed_mid_checkpoint`'s doc for the full analysis): some +/// kill offsets in the post-`BGSAVE` window cause TOTAL loss of the graph +/// plane's already-wal-v3-synced content. PROBABILISTIC, not +/// deterministic — a single default run (`MOON_CRASH_MATRIX_ITERS=1`) has +/// a real chance of NOT hitting the window and reporting green even with +/// `MOON_CRASH_MATRIX_RED=1` set. Reproduce reliably with +/// `MOON_CRASH_MATRIX_RED=1 MOON_CRASH_MATRIX_ITERS=20` (confirmed hit +/// within 20 iterations on both shard counts during this stage's +/// investigation — prod_s1 at iteration 11/20, prod_s4 at iteration +/// 7/20). +const MID_CHECKPOINT_GRAPH_LOSS_RED_REASON: &str = "checkpoint-Finalize window can total-loss the graph plane, NEW. Some \ + kill offsets in the 0-150ms post-BGSAVE window lose the FULL synced \ + graph batch (not just the unsynced tail) while KV/vector/WS/MQ all \ + survive in the same run — reviewed and confirmed not a harness \ + artifact. PROBABILISTIC (~1-in-7 to 1-in-12 per MOON_CRASH_MATRIX_ITERS \ + sample in this stage's investigation) — reproduce reliably with \ + MOON_CRASH_MATRIX_RED=1 MOON_CRASH_MATRIX_ITERS=20, a single default \ + run may report green by chance. Strong P0 candidate for kernel M3 \ + stage 2 (K2). Tracked separately, not this stage's job to fix."; + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_prod_s1_kv_isolated() { + scenarios::kv_isolated(&Config::PROD_S1); +} + +#[test] +#[ignore] +fn cross_plane_prod_s1_kv_spilled_isolated() { + scenarios::kv_spilled_isolated(&Config::PROD_S1); +} + +#[test] +#[ignore] +fn cross_plane_prod_s1_graph_isolated() { + scenarios::graph_isolated(&Config::PROD_S1); +} + +#[test] +#[ignore] +fn cross_plane_prod_s1_vector_isolated() { + scenarios::vector_isolated(&Config::PROD_S1); +} + +#[test] +#[ignore] +fn cross_plane_prod_s1_ws_isolated() { + scenarios::ws_isolated(&Config::PROD_S1); +} + +#[test] +#[ignore] +fn cross_plane_prod_s1_mq_isolated() { + scenarios::mq_isolated(&Config::PROD_S1); +} + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_prod_s1_txn_isolated_committed() { + if !harness::red_guard(TXN_GRAPH_LEG_RED_REASON) { + return; + } + scenarios::txn_isolated_committed(&Config::PROD_S1); +} + +/// Ungated (review round 3, P1) — atomicity is unrelated to task #52's +/// graph-leg-durability finding and is GREEN here; see +/// `scenarios::txn_isolated_atomicity`'s doc for why it must not share the +/// committed half's `red_guard`. +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_prod_s1_txn_isolated_atomicity() { + scenarios::txn_isolated_atomicity(&Config::PROD_S1); +} + +#[test] +#[ignore] +fn cross_plane_prod_s1_mixed_all_planes_synced() { + scenarios::mixed_synced(&Config::PROD_S1); +} + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_prod_s1_mixed_all_planes_mid_checkpoint() { + if !harness::red_guard(MID_CHECKPOINT_GRAPH_LOSS_RED_REASON) { + return; + } + scenarios::mixed_mid_checkpoint(&Config::PROD_S1); +} + +#[test] +#[ignore] +fn cross_plane_prod_s1_mixed_all_planes_concurrent_burst_no_corruption() { + scenarios::concurrent_burst_no_corruption(&Config::PROD_S1); +} + +#[test] +#[ignore] +fn cross_plane_prod_s4_kv_isolated() { + scenarios::kv_isolated(&Config::PROD_S4); +} + +#[test] +#[ignore] +fn cross_plane_prod_s4_kv_spilled_isolated() { + scenarios::kv_spilled_isolated(&Config::PROD_S4); +} + +#[test] +#[ignore] +fn cross_plane_prod_s4_graph_isolated() { + scenarios::graph_isolated(&Config::PROD_S4); +} + +#[test] +#[ignore] +fn cross_plane_prod_s4_vector_isolated() { + scenarios::vector_isolated(&Config::PROD_S4); +} + +#[test] +#[ignore] +fn cross_plane_prod_s4_ws_isolated() { + scenarios::ws_isolated(&Config::PROD_S4); +} + +#[test] +#[ignore] +fn cross_plane_prod_s4_mq_isolated() { + scenarios::mq_isolated(&Config::PROD_S4); +} + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_prod_s4_txn_isolated_committed() { + if !harness::red_guard(TXN_GRAPH_LEG_RED_REASON) { + return; + } + scenarios::txn_isolated_committed(&Config::PROD_S4); +} + +/// Ungated (review round 3, P1) — see +/// `cross_plane_prod_s1_txn_isolated_atomicity`'s doc. +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_prod_s4_txn_isolated_atomicity() { + scenarios::txn_isolated_atomicity(&Config::PROD_S4); +} + +#[test] +#[ignore] +fn cross_plane_prod_s4_mixed_all_planes_synced() { + scenarios::mixed_synced(&Config::PROD_S4); +} + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_prod_s4_mixed_all_planes_mid_checkpoint() { + if !harness::red_guard(MID_CHECKPOINT_GRAPH_LOSS_RED_REASON) { + return; + } + scenarios::mixed_mid_checkpoint(&Config::PROD_S4); +} + +#[test] +#[ignore] +fn cross_plane_prod_s4_mixed_all_planes_concurrent_burst_no_corruption() { + scenarios::concurrent_burst_no_corruption(&Config::PROD_S4); +} diff --git a/tests/crash_matrix_cross_plane/tests_seeded_red.rs b/tests/crash_matrix_cross_plane/tests_seeded_red.rs new file mode 100644 index 00000000..e2f91511 --- /dev/null +++ b/tests/crash_matrix_cross_plane/tests_seeded_red.rs @@ -0,0 +1,153 @@ +//! Seeded RED cell 2 (brief §1.4): MQ/WS resurrection via delete-then- +//! restart. Run on `Config::PROD_S1` (appendonly=yes, disk-offload=enable, +//! shards=1) — the config where MQ effect-record durability itself is +//! ALREADY fixed (PR #291), so a resurrection here isolates exactly the one +//! remaining variable: no MQ/WS tombstone record exists yet, so a generic +//! delete does not stop WAL replay from re-materializing the deleted +//! content on the next restart (same bug class as the already-fixed +//! KV/vector cold-plane resurrection, PR #257). +//! +//! Both writes-a-marker-after-the-delete-and-waits-for-it, never for the +//! delete's own payload bytes directly — `MQ CREATE`/`MQ PUSH` records +//! already contain the queue name, so waiting for that substring after a +//! `DEL` would false-positive-match the EARLIER create/push records. A +//! later, unrelated marker proves (by WAL append ordering) that everything +//! before it — including the delete — reached disk. + +use crate::harness::{self, Config}; +use crate::planes::*; +use crate::resp::Conn; + +/// MQ: `DEL ` (generic keyspace delete, NOT an `MQ.*` command) must +/// tombstone the stream so it does not resurrect on the next restart. +/// +/// RED (expected, per PR #291's changelog and brief §1.4): `replay_mq_wal` +/// has no way to represent "this queue was deleted after these pushes" — +/// there is no MqDelete/tombstone WAL record — so replay re-materializes +/// the full pre-delete stream content regardless of the `DEL`. +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_seeded_red_mq_generic_del_resurrection() { + if !harness::red_guard( + "MQ streams deleted via generic DEL are not tombstoned and \ + resurrect with full content on the next restart — no MQ tombstone \ + WAL record exists yet (PR #291 changelog, kernel M3 brief §1.4). \ + Same bug class as the already-fixed KV/vector cold-plane \ + resurrection (PR #257), not yet applied to MQ. Tracked \ + separately — not this stage's job to fix.", + ) { + return; + } + let cfg = Config::PROD_S1; + let dir = harness::unique_dir("seededred-mq"); + let (guard, port) = harness::spawn_moon_on(&dir, &cfg, &[]); + let mut c = Conn::open(port); + let q = "seededred-mq-queue"; + + mq_create(&mut c, q, 0); + for i in 1..=5u32 { + mq_push(&mut c, q, "seq", &i.to_string()); + } + let marker1 = "SEEDEDRED-MQ-PRE-DEL-SYNC".to_string(); + mq_push(&mut c, q, "final", &marker1); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker1.as_bytes()); + + assert_eq!( + xlen(&mut c, q), + 6, + "sanity: 6 messages must be visible before DEL" + ); + c.cmd_s(&["DEL", q]); + assert_eq!( + xlen(&mut c, q), + 0, + "sanity: DEL removed the key immediately" + ); + + // Sync a LATER, unrelated write so its position proves the DEL (which + // precedes it in append order) is also durable. `DEL` is a generic + // keyspace command — same log family as any other keyspace write, i.e. + // the AOF, not wal-v3 (confirmed empirically: plain commands never + // appear in wal-v3 at all) — so the follow-up sync marker must be + // AOF-family too, or it proves nothing about the DEL's own durability. + let marker2 = "SEEDEDRED-MQ-POST-DEL-SYNC".to_string(); + kv_set(&mut c, "seededred-mq-post-del-marker", &marker2); + harness::wait_for_aof_bytes_any_shard(&dir, marker2.as_bytes()); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, &cfg, &[]); + let mut c2 = Conn::open(port2); + assert_eq!( + xlen(&mut c2, q), + 0, + "RED: MQ stream {q:?} must NOT resurrect after restart — DEL was \ + issued (and synced) before the crash, so replay re-materializing \ + any content here is exactly the task #43-class resurrection bug" + ); + drop(guard2); +} + +/// WS analogue: `WS DROP ` (the workspace registry's OWN delete +/// operation — `WorkspaceRegistry` is process-global, not a normal keyspace +/// key, so a *generic* `DEL` does not apply to it the way it does to an MQ +/// stream) must not resurrect the workspace on the next restart. +/// +/// This does NOT assume a RED label — `replay_workspace_wal` +/// (`src/shard/shared_databases.rs`) was read directly for this brief and +/// DOES process `WorkspaceDrop` records in on-disk order, so this specific +/// scenario is plausibly already GREEN; what the brief's §1.4 citation +/// verifies is MQ, grouped with WS only as "the same bug CLASS" at the +/// design-doc level. Run it and report the actual outcome rather than +/// asserting a predetermined answer — if this turns out GREEN, that is +/// itself useful signal (the WS half of "MQ/WS resurrection" may already be +/// closed, or may need a differently-shaped reproduction); if RED, annotate +/// with what actually failed. +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_seeded_red_ws_drop_durability() { + let cfg = Config::PROD_S1; + let dir = harness::unique_dir("seededred-ws"); + let (guard, port) = harness::spawn_moon_on(&dir, &cfg, &[]); + let mut c = Conn::open(port); + let name = "seededred-ws-workspace"; + + let ws_id = ws_create(&mut c, name); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, name.as_bytes()); + assert!( + ws_list_contains(&mut c, &ws_id), + "sanity: workspace visible before DROP" + ); + + ws_drop(&mut c, &ws_id); + assert!( + !ws_list_contains(&mut c, &ws_id), + "sanity: DROP removed it immediately (live path)" + ); + + // Sync a LATER, unrelated wal-v3-family write (same log file, same + // per-shard append order as WorkspaceDrop itself) so its WAL position + // proves the DROP is also durable. A KV/AOF-family marker would NOT + // prove this — the AOF and wal-v3 writers flush independently, so + // waiting on the AOF gives no guarantee wal-v3 has caught up to the + // corresponding point. + let marker = "SEEDEDRED-WS-POST-DROP-SYNC".to_string(); + let sync_graph = "seededred-ws-sync-graph"; + graph_create(&mut c, sync_graph); + graph_addnode_marker(&mut c, sync_graph, &marker); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker.as_bytes()); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, &cfg, &[]); + let mut c2 = Conn::open(port2); + let resurrected = ws_list_contains(&mut c2, &ws_id); + assert!( + !resurrected, + "workspace {name:?} resurrected after WS DROP + kill-9 — \ + WorkspaceDrop WAL record did not replay (or replayed out of order \ + relative to WorkspaceCreate); this is the WS half of the brief \ + §1.4 'MQ/WS resurrection' gap and should be annotated \ + #[ignore = \"RED: ...\"] with this exact failure mode if it \ + reproduces on the VM" + ); + drop(guard2); +} diff --git a/tests/crash_matrix_cross_plane/tests_spot.rs b/tests/crash_matrix_cross_plane/tests_spot.rs new file mode 100644 index 00000000..6922a410 --- /dev/null +++ b/tests/crash_matrix_cross_plane/tests_spot.rs @@ -0,0 +1,62 @@ +//! Spot-check cells: the 4 remaining config cells outside the curated +//! production/legacy set (brief: "spot-check the remaining cells with 1-2 +//! representative scenarios each"). + +use crate::harness::{self, Config}; +use crate::scenarios; + +// --------------------------------------------------------------------------- +// appendonly=no + disk-offload=enable (spill is a documented loud no-op +// without a durability backstop — src/config.rs `disk_offload_spill_inert`; +// persistence_dir is None here too, so this is the same "no durability +// contract on any plane" shape as the legacy_no_s1 cell). +// --------------------------------------------------------------------------- + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_spot_offload_no_s1_no_durability_contract_liveness() { + scenarios::no_durability_liveness(&Config::SPOT_OFFLOAD_NO_S1); +} + +#[test] +#[ignore] +fn cross_plane_spot_offload_no_s4_no_durability_contract_liveness() { + scenarios::no_durability_liveness(&Config::SPOT_OFFLOAD_NO_S4); +} + +// --------------------------------------------------------------------------- +// Legacy mode at shards=4 (excluded from the full matrix — "legacy mode has +// no cross-shard novelty" — but the brief still wants 1-2 representative +// scenarios). graph_isolated here reproduces the SAME RED gap as +// legacy_yes_s1 (the bug is legacy-mode WAL replay, not shard-count-specific; +// spot-checking at shards=4 confirms the gap is not accidentally +// shard-count-dependent). Gated by `harness::red_guard` — see that +// function's doc for why `#[ignore = "RED: ..."]` alone does not skip it. +// --------------------------------------------------------------------------- + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_spot_legacy_s4_graph_isolated() { + if !harness::red_guard( + "legacy-mode (--disk-offload disable) graph WAL replay does not \ + reconstruct the graph — same gap as \ + cross_plane_legacy_yes_s1_graph_isolated, confirmed shard-count- \ + independent (PR #288 / task #43 changelog). Not this stage's job \ + to fix — tracked separately.", + ) { + return; + } + scenarios::graph_isolated(&Config::SPOT_LEGACY_S4); +} + +#[test] +#[ignore] +fn cross_plane_spot_legacy_s4_mq_isolated() { + scenarios::mq_isolated(&Config::SPOT_LEGACY_S4); +} + +#[test] +#[ignore] +fn cross_plane_spot_legacy_no_s4_no_durability_contract_liveness() { + scenarios::no_durability_liveness(&Config::SPOT_LEGACY_NO_S4); +}