From 70cdb5bfef25e2ded8a10ebe0d2cf270c6b1b08b Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 05:27:04 +0700 Subject: [PATCH 1/3] =?UTF-8?q?test(shard):=20RED=20=E2=80=94=20Pass=20C?= =?UTF-8?q?=20recycles=20the=20sole=20durable=20copy=20of=20legacy-mode=20?= =?UTF-8?q?WAL=20history=20(task=20#43)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds tests/crash_recovery_wal_recycle_legacy.rs, a live kill-9 + restart integration test proving that autovacuum's Pass C (`src/shard/autovacuum.rs`) deletes WAL v3 segments that are the SOLE durable copy of graph/workspace history when disk-offload is disabled (legacy mode). Pass C calls `WalWriterV3::recycle_aggressive(redo_lsn = wal.current_lsn())` whenever total WAL bytes exceed `--max-wal-size`, with no gate on whether a checkpoint protocol backs that floor. In legacy mode there is no periodic plane snapshot: `save_graph_store` only runs at graceful shutdown, and the workspace/MQ/temporal registries have no snapshot format at all — they replay purely from the WAL. Once the ceiling is breached, every sealed segment is deleted regardless of whether its records survive anywhere else. The test forces the condition deterministically: tiny --wal-segment-size / --max-wal-size plus a fast --autovacuum-interval-secs make Pass C run within seconds. It asserts two things: (1) byte-level survival — the earliest WorkspaceCreate and graph-node WAL records are still present on disk after Pass C has had several chances to run, and (2) round-trip survival — after a real kill -9 + restart, WS LIST still returns the workspace. Against current main, both fail: assertion (1) panics with "the WorkspaceCreate record ... was deleted from every on-disk WAL segment." Confirmed RED via `git stash` of the (separately staged) fix and rerunning this test: it fails with exactly that assertion. Restoring the fix turns it green. Disk-offload mode is a pre-existing pass (`g4`/`g5` in crash_recovery_graph_durability.rs, unmodified) — this test targets only the legacy-mode gap. Full graph reconstruction via GRAPH.QUERY after restart is deliberately not asserted in legacy mode: independent probing found graph command replay from WAL v3 does not reconstruct the graph in that mode even with zero Pass C interference (default --max-wal-size), a separate, pre-existing gap unrelated to WAL recycling and out of scope here. author: Tin Dang --- tests/crash_recovery_wal_recycle_legacy.rs | 560 +++++++++++++++++++++ 1 file changed, 560 insertions(+) create mode 100644 tests/crash_recovery_wal_recycle_legacy.rs diff --git a/tests/crash_recovery_wal_recycle_legacy.rs b/tests/crash_recovery_wal_recycle_legacy.rs new file mode 100644 index 00000000..634b23cd --- /dev/null +++ b/tests/crash_recovery_wal_recycle_legacy.rs @@ -0,0 +1,560 @@ +//! Task #43 (P1 data loss): autovacuum Pass C (`src/shard/autovacuum.rs`) +//! calls `WalWriterV3::recycle_aggressive(redo_lsn = wal.current_lsn())` +//! whenever total on-disk WAL bytes exceed `--max-wal-size`, REGARDLESS of +//! whether disk-offload (the only mode with a periodic checkpoint/graph +//! snapshot protocol) is enabled. +//! +//! In legacy mode (`--disk-offload disable`) there is no periodic plane +//! snapshot: `save_graph_store` runs only on graceful shutdown +//! (`src/shard/event_loop.rs` shutdown branch), and the workspace/MQ/temporal +//! registries have NO snapshot format at all — `replay_workspace_wal` / +//! `replay_mq_wal` / `replay_temporal_wal` rebuild them purely by re-scanning +//! the WAL. `current_lsn()` is the LSN *about to be assigned* (i.e. "assume +//! everything is durable already"), so `recycle_aggressive` deletes every +//! sealed segment once the ceiling is breached — including segments whose +//! records have no other durable form. A kill -9 after that recycle loses +//! that data permanently, even though replay (task #42, PR #286) now works +//! correctly for the records that remain. +//! +//! This test proves the loss end to end against a REAL server process +//! (pattern: `tests/crash_recovery_graph_durability.rs` G1/G4/G5): tiny +//! `--wal-segment-size` / `--max-wal-size` force several Pass C recycle +//! passes within seconds via a fast `--autovacuum-interval-secs`. +//! +//! Two independent assertions, deliberately kept separate: +//! +//! 1. **Byte-level survival** (the assertion scoped exactly to task #43): +//! after Pass C has had several chances to run, the WAL bytes for the +//! earliest graph write AND the earliest workspace write must still be +//! present on disk somewhere under `wal-v3/`. This is what +//! `recycle_aggressive` vs. "skip and warn" actually controls, and it is +//! provable without depending on any replay/reconstruction logic. +//! 2. **Round-trip survival** (kill -9 + restart): the workspace must still +//! be listed by `WS LIST`. `replay_workspace_wal` rebuilds the registry +//! purely from the WAL (no snapshot format exists for it — the exact +//! plane the task's design guidance called out), so this is an +//! end-to-end proof that legacy mode no longer loses it. +//! +//! Deliberately NOT asserted: full graph reconstruction via `GRAPH.QUERY` +//! after restart. Probing independently (see task notes) showed that graph +//! command replay in legacy (non-disk-offload) mode does not reconstruct +//! the graph even with default `--max-wal-size` and zero Pass C recycling — +//! a separate, pre-existing gap unrelated to WAL recycling (the WAL bytes +//! for the graph write are provably still on disk; the replay pipeline +//! just doesn't consume them into a `GraphStore` in this mode). Assertion 1 +//! above still proves task #43 is fixed for the graph plane's data at the +//! byte level, which is the layer Pass C operates on. +//! +//! Run with (monoio default — matches CI): +//! cargo build --release +//! cargo test --release --test crash_recovery_wal_recycle_legacy -- --ignored + +#![allow(clippy::unwrap_used)] + +use std::io::{BufReader, Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +// --------------------------------------------------------------------------- +// Binary resolution (pattern: crash_recovery_graph_durability.rs) +// --------------------------------------------------------------------------- + +fn find_moon_binary() -> PathBuf { + if let Ok(bin) = std::env::var("MOON_BIN") { + let p = PathBuf::from(bin); + if p.exists() { + return p; + } + } + 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." + ); +} + +fn unique_port() -> u16 { + use std::net::TcpListener; + let listener = TcpListener::bind("127.0.0.1:0").expect("bind :0"); + let port = listener.local_addr().expect("local_addr").port(); + drop(listener); + port +} + +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-walrecycle-{}-{}-{}", + std::process::id(), + suffix, + nanos + )) +} + +// --------------------------------------------------------------------------- +// Server spawn / crash / restart harness (pattern: crash_recovery_graph_durability.rs) +// --------------------------------------------------------------------------- + +struct ServerGuard { + child: Child, + dir_marker: String, +} + +impl ServerGuard { + 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)) { + sigkill(&mut self.child); + } else { + let _ = self.child.wait(); + } + let _ = Command::new("pkill") + .args(["-9", "-f"]) + .arg(&self.dir_marker) + .output(); + } +} + +#[cfg(unix)] +fn sigkill(child: &mut Child) { + let pid = child.id() as i32; + unsafe { + libc::kill(pid, libc::SIGKILL); + } + let _ = child.wait(); +} + +#[cfg(not(unix))] +fn sigkill(child: &mut Child) { + let _ = child.kill(); + let _ = child.wait(); +} + +/// Wait until nothing is accepting on `port` (SO_REUSEPORT makes a plain +/// bind-check useless; two consecutive refused connects = down). +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)); + } + } + } +} + +/// Spawn moon in LEGACY mode (`--disk-offload disable` — no checkpoint +/// protocol, no periodic graph snapshot) with a tiny WAL ceiling so Pass C +/// fires within seconds instead of needing 256 MB of writes. +fn spawn_moon_legacy_fast_recycle(port: u16, dir: &Path) -> ServerGuard { + std::fs::create_dir_all(dir).expect("create test dir"); + let child = Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--appendonly", + "yes", + "--disk-offload", + "disable", + "--disk-free-min-pct", + "0", + "--wal-segment-size", + "32kb", + "--max-wal-size", + "96kb", + "--autovacuum-interval-secs", + "1", + "--dir", + ]) + .arg(dir) + .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() + .expect("spawn moon (run `cargo build --release` first, or set MOON_BIN)"); + ServerGuard::new(child, dir) +} + +const RESTART_ATTEMPTS: usize = 6; + +fn start_moon_alive(port: u16, dir: &Path) -> ServerGuard { + for attempt in 1..=RESTART_ATTEMPTS { + let mut guard = spawn_moon_legacy_fast_recycle(port, dir); + let mut up = false; + for _ in 0..300 { + if let Ok(Some(_status)) = guard.child.try_wait() { + break; // self-terminated (rebind race) — retry + } + if TcpStream::connect(format!("127.0.0.1:{port}")).is_ok() { + up = true; + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + if up { + return guard; + } + drop(guard); + if attempt < RESTART_ATTEMPTS { + std::thread::sleep(Duration::from_millis(300)); + } + } + panic!("moon failed to start+serve on port {port} after {RESTART_ATTEMPTS} attempts"); +} + +fn crash(mut guard: ServerGuard, port: u16) { + sigkill(&mut guard.child); + drop(guard); // idempotent (already reaped) + pkill backstop + wait_for_port_down(port); +} + +/// Poll every `*.wal` file under `/shard-0/wal-v3/` until `needle` +/// appears somewhere on disk — the bounded, condition-based "my last write +/// reached the WAL fd" wait that makes the recycle race deterministic. +fn wait_for_wal_v3_bytes(dir: &Path, needle: &[u8]) { + let wal_dir = dir.join("shard-0").join("wal-v3"); + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if let Ok(entries) = std::fs::read_dir(&wal_dir) { + for entry in entries.flatten() { + if let Ok(bytes) = std::fs::read(entry.path()) { + if bytes.windows(needle.len()).any(|w| w == needle) { + return; + } + } + } + } + assert!( + Instant::now() < deadline, + "WAL v3 dir {} never contained marker {:?} — writes are not \ + reaching the v3 WAL", + wal_dir.display(), + String::from_utf8_lossy(needle), + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +/// Total bytes across every `*.wal` segment under `/shard-0/wal-v3/`. +fn wal_v3_total_bytes(dir: &Path) -> u64 { + let wal_dir = dir.join("shard-0").join("wal-v3"); + let mut total = 0u64; + 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 +} + +/// One-shot (non-blocking) check: does `needle` appear in any `*.wal` +/// segment under `/shard-0/wal-v3/` right now? +fn wal_v3_contains_bytes(dir: &Path, needle: &[u8]) -> bool { + let wal_dir = dir.join("shard-0").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()) { + if bytes.windows(needle.len()).any(|w| w == needle) { + return true; + } + } + } + } + false +} + +// --------------------------------------------------------------------------- +// Minimal RESP client (pattern: crash_recovery_graph_durability.rs) +// --------------------------------------------------------------------------- + +#[derive(Debug, PartialEq, Clone)] +enum V { + Simple(String), + Err(String), + Int(i64), + Bulk(Vec), + Arr(Vec), + Null, +} + +struct Client { + reader: BufReader, + writer: TcpStream, +} + +impl Client { + fn connect(port: u16) -> Self { + let addr = format!("127.0.0.1:{port}") + .to_socket_addrs() + .unwrap() + .next() + .unwrap(); + let start = Instant::now(); + let stream = loop { + match TcpStream::connect_timeout(&addr, Duration::from_millis(200)) { + Ok(s) => break s, + Err(_) if start.elapsed() < Duration::from_secs(30) => { + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => panic!("server never accepted on port {port}: {e}"), + } + }; + stream + .set_read_timeout(Some(Duration::from_secs(30))) + .unwrap(); + let writer = stream.try_clone().unwrap(); + Client { + reader: BufReader::new(stream), + writer, + } + } + + fn encode(args: &[&[u8]]) -> Vec { + let mut out = format!("*{}\r\n", args.len()).into_bytes(); + for a in args { + out.extend_from_slice(format!("${}\r\n", a.len()).as_bytes()); + out.extend_from_slice(a); + out.extend_from_slice(b"\r\n"); + } + out + } + + fn cmd(&mut self, args: &[&[u8]]) -> V { + self.writer.write_all(&Self::encode(args)).expect("write"); + self.parse() + } + + fn read_line(&mut self) -> String { + let mut line = Vec::new(); + let mut b = [0u8; 1]; + loop { + self.reader.read_exact(&mut b).expect("read byte"); + if b[0] == b'\n' { + break; + } + if b[0] != b'\r' { + line.push(b[0]); + } + } + String::from_utf8_lossy(&line).into_owned() + } + + fn parse(&mut self) -> V { + let line = self.read_line(); + let (t, rest) = line.split_at(1); + match t { + "+" => V::Simple(rest.to_string()), + "-" => V::Err(rest.to_string()), + ":" => V::Int(rest.parse().expect("int")), + "$" => { + let n: i64 = rest.parse().expect("bulk len"); + if n < 0 { + return V::Null; + } + let mut buf = vec![0u8; n as usize + 2]; + self.reader.read_exact(&mut buf).expect("bulk body"); + buf.truncate(n as usize); + V::Bulk(buf) + } + "*" => { + let n: i64 = rest.parse().expect("arr len"); + if n < 0 { + return V::Null; + } + (0..n).map(|_| self.parse()).collect::>().into() + } + "%" => { + let n: i64 = rest.parse().expect("map len"); + (0..n * 2).map(|_| self.parse()).collect::>().into() + } + "_" => V::Null, + other => panic!("unexpected RESP type byte {other:?} (line {line:?})"), + } + } +} + +impl From> for V { + fn from(v: Vec) -> Self { + V::Arr(v) + } +} + +const G: &[u8] = b"walrecycleg"; + +// --------------------------------------------------------------------------- +// Task #43: legacy-mode Pass C must not lose the sole durable copy of +// graph / workspace history. +// --------------------------------------------------------------------------- + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn t43_legacy_mode_pass_c_must_not_lose_ws_and_graph_history() { + let port = unique_port(); + let dir = unique_dir("t43"); + let guard = start_moon_alive(port, &dir); + let mut c = Client::connect(port); + + // 1. Earliest writes: a workspace (registry has NO snapshot format — + // replay_workspace_wal rebuilds it purely from the WAL) and a graph + // node with a unique marker property. Both land in the FIRST WAL + // segment, which — once the ceiling is breached below — is the + // first one Pass C's `recycle_aggressive` deletes. + let ws_id = match c.cmd(&[b"WS", b"CREATE", b"legacy-ws-t43"]) { + V::Bulk(id) => id, + other => panic!("WS CREATE failed: {other:?}"), + }; + + assert_eq!( + c.cmd(&[b"GRAPH.CREATE", G]), + V::Simple("OK".into()), + "GRAPH.CREATE" + ); + match c.cmd(&[b"GRAPH.ADDNODE", G, b"Early", b"tag", b"EARLY-MARKER-T43"]) { + V::Int(_) => {} + other => panic!("early marker ADDNODE failed: {other:?}"), + } + wait_for_wal_v3_bytes(&dir, b"EARLY-MARKER-T43"); + wait_for_wal_v3_bytes(&dir, b"legacy-ws-t43"); + + // Sanity: the workspace is visible right now, pre-recycle. + match c.cmd(&[b"WS", b"LIST"]) { + V::Arr(entries) => assert!( + entries.iter().any(|e| matches!(e, V::Arr(cells) + if cells.first() == Some(&V::Bulk(ws_id.clone())))), + "workspace not visible immediately after WS CREATE: {entries:?}" + ), + other => panic!("WS LIST failed: {other:?}"), + } + + // 2. Blow past `--max-wal-size 96kb` with padded ADDNODE writes so the + // early segment rotates out and becomes eligible for recycling. + // 2KB padding * 300 nodes ~= 600KB, several multiples of the 96KB + // ceiling and the 32KB segment size (~18+ rotations). + let padding = "x".repeat(2048); + for i in 0..300u64 { + match c.cmd(&[ + b"GRAPH.ADDNODE", + G, + b"Bulk", + b"pad", + padding.as_bytes(), + b"id", + i.to_string().as_bytes(), + ]) { + V::Int(_) => {} + other => panic!("bulk ADDNODE {i} failed: {other:?}"), + } + } + + // Confirm we actually forced growth past the ceiling (test-setup sanity, + // not the bug assertion itself). + let deadline = Instant::now() + Duration::from_secs(15); + loop { + if wal_v3_total_bytes(&dir) > 96 * 1024 { + break; + } + assert!( + Instant::now() < deadline, + "WAL never exceeded the 96KB ceiling — test setup did not \ + generate enough volume to exercise Pass C" + ); + std::thread::sleep(Duration::from_millis(50)); + } + + // 3. Give the 1s autovacuum tick several chances to run Pass C. + std::thread::sleep(Duration::from_secs(5)); + + // 4. Assertion 1 (byte-level, scoped exactly to task #43): the earliest + // WS + graph records must still be on disk somewhere under wal-v3/. + // This is what `recycle_aggressive` vs. "skip and warn" controls, + // independent of any replay/reconstruction correctness. + assert!( + wal_v3_contains_bytes(&dir, b"legacy-ws-t43"), + "task #43 REGRESSION: the WorkspaceCreate record for \ + 'legacy-ws-t43' was deleted from every on-disk WAL segment — Pass \ + C recycled it in legacy mode with no checkpoint/snapshot floor, \ + and the workspace registry has no other durable form" + ); + assert!( + wal_v3_contains_bytes(&dir, b"EARLY-MARKER-T43"), + "task #43 REGRESSION: the earliest graph node's WAL record was \ + deleted from every on-disk WAL segment — Pass C recycled it in \ + legacy mode with no checkpoint/graph-snapshot floor" + ); + + // 5. Kill -9 — NO graceful shutdown, so `save_graph_store` never runs + // and the WAL is the only durable form of the early writes. + crash(guard, port); + + // 6. Assertion 2 (round-trip): restart and prove the workspace registry + // — which replays purely from the WAL, no snapshot format exists — + // survived. (Full graph reconstruction via GRAPH.QUERY is + // deliberately not asserted here; see the module doc comment.) + let guard = start_moon_alive(port, &dir); + let mut c = Client::connect(port); + + let ws_list = match c.cmd(&[b"WS", b"LIST"]) { + V::Arr(entries) => entries, + other => panic!("WS LIST failed: {other:?}"), + }; + let ws_survived = ws_list.iter().any(|entry| match entry { + V::Arr(cells) => cells.first() == Some(&V::Bulk(ws_id.clone())), + _ => false, + }); + assert!( + ws_survived, + "task #43 REGRESSION: workspace 'legacy-ws-t43' (id {:?}) did not \ + survive kill-9 — Pass C recycled the WAL segment holding its \ + WorkspaceCreate record, and the workspace registry has no other \ + durable form. WS LIST after recovery: {ws_list:?}", + String::from_utf8_lossy(&ws_id), + ); + + drop(guard); +} From 6b0dc5d1e2a768d0d65d4b7eb6e40aada65ddb50 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 05:27:09 +0700 Subject: [PATCH 2/3] fix(shard): gate autovacuum Pass C WAL recycle on checkpoint-backed mode (task #43, P1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `AutovacuumDaemon::run_tick`'s Pass C called `WalWriterV3::recycle_aggressive` against `wal.current_lsn()` — the LSN about to be assigned, i.e. "every sealed segment is durable elsewhere" — whenever total WAL bytes exceeded `--max-wal-size`, regardless of whether disk-offload (the only mode with a real checkpoint protocol) was enabled. That assumption only holds when a checkpoint maintains a genuine redo_lsn and `persist_graph_at_checkpoint` snapshots the graph write-buffer before the floor advances. In legacy (non-disk-offload) mode there is no periodic plane snapshot at all: `save_graph_store` runs only at graceful shutdown, and the workspace/MQ/ temporal registries have no snapshot format whatsoever — they replay purely from the WAL. Pass C therefore deleted the sole durable copy of that history once the ceiling was breached; a kill -9 afterwards lost it permanently. Design choice (documented in autovacuum.rs and CHANGELOG): rather than inventing a snapshot format for workspace/MQ/temporal (explicitly out of scope, and the guidance's own fallback for exactly this case), Pass C now takes a `disk_offload_enabled` flag: - true (disk-offload / checkpoint-backed): unchanged — same `recycle_aggressive(current_lsn())` call as before this fix. - false (legacy mode): skip the recycle entirely. Emit a rate-limited (5 min) tracing::warn! and increment the new `RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL` counter (exposed as `reclamation_wal_recycle_blocked_no_checkpoint_total` in `INFO # Reclamation`) so operators can observe unbounded WAL growth in this mode instead of silently losing data. A partial snapshot-then-floor design (freeze+persist the graph plane via the already-callable `persist_graph_at_checkpoint`, then recycle only segments proven free of un-snapshotted workspace/MQ/temporal records) was considered and rejected: it would still leave those three planes unprotected without inventing a snapshot format for them, adding complexity without closing the gap. "Never trade data loss for disk space" wins; operators needing bounded legacy-mode WAL growth should enable `--disk-offload enable`. Both call sites in `event_loop.rs` (tokio select! and monoio tick loop) now pass `server_config.disk_offload_enabled()`. Verification: - tests/crash_recovery_wal_recycle_legacy.rs (previous commit) goes green. - Two new fast unit tests in autovacuum.rs (test_pass_c_legacy_mode_skips_recycle_and_counts_blocked, test_pass_c_disk_offload_mode_recycles_unchanged) exercise Pass C directly against a real WalWriterV3 without spawning a server, proving disk-offload mode's segment count still drops (byte-for-byte unchanged behavior) while legacy mode's does not. - crash_recovery_graph_durability.rs g4/g5 (disk-offload / checkpoint path) pass unmodified, confirming no behavior change outside legacy mode. - Full `cargo test --release --lib` (4145 passed), shardslice_live (6 passed), `cargo clippy` default and `--no-default-features --features runtime-tokio,jemalloc` (zero warnings), `cargo fmt --check` clean. CHANGELOG updated under [Unreleased]. author: Tin Dang --- CHANGELOG.md | 51 ++++++++ src/command/info_reclamation.rs | 14 ++- src/shard/autovacuum.rs | 209 +++++++++++++++++++++++++++++--- src/shard/event_loop.rs | 2 + 4 files changed, 259 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cec873b..4c2ddc1a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,57 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — autovacuum Pass C recycled the sole durable copy of graph/WS/MQ/temporal WAL history in legacy mode (task #43, P1) + +`AutovacuumDaemon::run_tick`'s Pass C (`src/shard/autovacuum.rs`) called +`WalWriterV3::recycle_aggressive(redo_lsn = wal.current_lsn())` — the LSN +*about to be assigned*, i.e. "everything sealed is durable elsewhere" — +whenever total on-disk WAL bytes exceeded `--max-wal-size` (default +256 MiB), **regardless of whether disk-offload was enabled**. That floor +only holds in disk-offload mode, where the checkpoint protocol maintains a +real `redo_lsn` and `persist_graph_at_checkpoint` snapshots the graph +write-buffer before the floor advances. In legacy (non-disk-offload) mode +there is no periodic plane snapshot at all: `save_graph_store` runs only at +graceful shutdown, and the workspace/MQ/temporal registries have **no +snapshot format whatsoever** — `replay_workspace_wal` / `replay_mq_wal` / +`replay_temporal_wal` rebuild them purely by re-scanning the WAL. Once the +ceiling was breached, Pass C deleted every sealed segment unconditionally, +including segments holding the sole durable copy of that history; a kill-9 +afterwards lost it permanently, even though replay itself (task #42, +PR #286) works correctly for whatever records survive. + +Fixed per the "never trade data loss for disk space" principle: `run_tick` +now takes a `disk_offload_enabled` flag. When `true`, behavior is byte-for- +byte unchanged (`recycle_aggressive` against `current_lsn()`, as before). +When `false`, Pass C no longer recycles: it skips the delete, emits a +rate-limited (5 min) `tracing::warn!`, and increments the new +`reclamation_wal_recycle_blocked_no_checkpoint_total` INFO counter +(`RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL`) so operators can see WAL +growth is unbounded in this mode. A snapshot-then-floor design (option A) +was considered and rejected for the workspace/MQ/temporal planes — they +have no snapshot format to floor against, and inventing one was explicitly +out of scope for this fix; only the graph plane has a callable snapshot +(`persist_graph_at_checkpoint`), so a partial-A fix would still have left +WS/MQ/temporal unprotected. Operators needing bounded WAL growth without +this trade-off should enable `--disk-offload enable`. + +New integration test `tests/crash_recovery_wal_recycle_legacy.rs` +(`t43_legacy_mode_pass_c_must_not_lose_ws_and_graph_history`, live kill -9 + +restart against tiny `--wal-segment-size`/`--max-wal-size` to force several +Pass C ticks within seconds) proves two things: (1) byte-level survival — +after Pass C has had several chances to run, the earliest workspace AND +graph WAL records are still present on disk; and (2) round-trip survival — +after a real kill -9 + restart, `WS LIST` still returns the workspace +(replay-only plane, no snapshot format, the exact risk called out above). +Full `GRAPH.QUERY` reconstruction after restart is deliberately not +asserted in legacy mode — independent probing found graph command replay +from WAL v3 does not reconstruct the graph in that mode even with zero +Pass C interference (default `--max-wal-size`), a separate, pre-existing +gap unrelated to WAL recycling; assertion (1) above still proves task #43 +is fixed at the layer Pass C actually operates on. Disk-offload mode is +unchanged, verified by the existing `crash_recovery_graph_durability.rs` +G4/G5 suite (checkpoint-driven recycle path) passing unmodified. + ### Fixed — cold-collection visibility: silent data loss on evicted Hash/List/Set/ZSet/Stream (task #41, P0) With disk-offload enabled (the default), the production eviction paths diff --git a/src/command/info_reclamation.rs b/src/command/info_reclamation.rs index 3c282e56..0beb7060 100644 --- a/src/command/info_reclamation.rs +++ b/src/command/info_reclamation.rs @@ -136,6 +136,14 @@ pub static RECL_AUTOVACUUM_SEGMENTS_COMPACTED_TOTAL: AtomicU64 = AtomicU64::new( /// TODO(P10→Wave2): wire from autovacuum scheduler. pub static RECL_AUTOVACUUM_THROTTLED_DUE_TO_LOAD: AtomicU64 = AtomicU64::new(0); +/// Cumulative count of Pass C (WAL recycle) ticks that found the WAL over +/// `--max-wal-size` but declined to recycle because no checkpoint/graph +/// snapshot protocol backs this WAL (legacy, non-disk-offload mode — +/// task #43, P1 data loss fix). Non-zero means the operator should either +/// enable `--disk-offload enable` or expect unbounded WAL growth past the +/// configured ceiling; it never means data was lost. +pub static RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL: AtomicU64 = AtomicU64::new(0); + /// Graph plan-cache hit count (cumulative). Used to compute hit_ratio. /// TODO(P10→Wave2): wire from PlanCache::get() on hit path. pub static RECL_PLAN_CACHE_HITS: AtomicU64 = AtomicU64::new(0); @@ -191,9 +199,11 @@ pub fn write_reclamation_section(buf: &mut String) { let _ = write!( buf, "reclamation_wal_bytes:{}\r\n\ - reclamation_wal_segments:{}\r\n", + reclamation_wal_segments:{}\r\n\ + reclamation_wal_recycle_blocked_no_checkpoint_total:{}\r\n", RECL_WAL_BYTES.load(Ordering::Relaxed), - RECL_WAL_SEGMENTS.load(Ordering::Relaxed) + RECL_WAL_SEGMENTS.load(Ordering::Relaxed), + RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL.load(Ordering::Relaxed) ); // -- Write stall: OR of disk-pressure (MA12), segment-backlog (MA1), and diff --git a/src/shard/autovacuum.rs b/src/shard/autovacuum.rs index 7d0578e2..eed834c0 100644 --- a/src/shard/autovacuum.rs +++ b/src/shard/autovacuum.rs @@ -10,7 +10,7 @@ //! |------|-------------|--------------| //! | a – MVCC | `prune_committed` + `sweep_zombies_mut` + `sweep_graph_zombies_mut` | P3 already ran (see `run_mvcc_sweep` in `timers.rs` on 1 s timer) | //! | b – Manifest GC | `gc_tombstones(retain_epochs, retain_secs)` | persistence dir not configured | -//! | c – WAL recycle | `recycle_aggressive(redo_lsn)` | WAL bytes ≤ max_wal_bytes | +//! | c – WAL recycle | `recycle_aggressive(redo_lsn)` (disk-offload only — task #43) | WAL bytes ≤ max_wal_bytes, or no checkpoint floor (legacy mode) | //! | d – Vector compact | `force_compact_if_needed(threshold)` | immutable segments ≤ threshold | //! | e – Graph compact | _no-op placeholder_ | TODO(P7): wire graph auto-merge | //! @@ -169,6 +169,10 @@ pub struct AutovacuumDaemon { last_budget_warn_at: Option, /// MA5: maintenance-window schedule (cron-based budget multipliers). pub maintenance_schedule: crate::shard::maintenance_schedule::MaintenanceSchedule, + /// Task #43: warn-rate limiter for the legacy-mode "WAL over ceiling but + /// no checkpoint floor to recycle against" condition — suppresses spam + /// while the WAL sits over `--max-wal-size` every tick. + last_wal_recycle_warn_at: Option, } impl AutovacuumDaemon { @@ -184,6 +188,7 @@ impl AutovacuumDaemon { last_run_at: None, last_budget_warn_at: None, maintenance_schedule: crate::shard::maintenance_schedule::MaintenanceSchedule::new(), + last_wal_recycle_warn_at: None, } } @@ -215,6 +220,19 @@ impl AutovacuumDaemon { /// * `wal_v3` — Per-shard WAL v3 writer (for WAL recycle pass). /// Pass `None` when persistence is not configured. /// * `max_wal_bytes` — WAL size ceiling from `--max-wal-size`. + /// * `disk_offload_enabled` — whether the checkpoint protocol (real + /// `redo_lsn` floor + periodic `persist_graph_at_checkpoint` snapshot) + /// is active for this shard. **Task #43 (P1 data loss)**: outside + /// disk-offload mode there is no periodic graph/workspace/MQ/temporal + /// snapshot — `save_graph_store` runs only at graceful shutdown, and + /// the workspace/MQ/temporal registries have no snapshot format at + /// all (they replay purely from the WAL). Pass C's `recycle_aggressive` + /// uses `wal.current_lsn()` (the LSN about to be assigned, i.e. + /// "assume every sealed segment is durable elsewhere") as its floor — + /// true only when a checkpoint backs it. When `false`, Pass C does + /// NOT recycle; it logs a rate-limited warning and increments + /// `RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL` instead. Disk-offload + /// mode behavior is unchanged from before this fix. /// * `manifest_retain_epochs` / `manifest_retain_secs` — P1 config. /// * `max_immutable_segments` — trigger threshold for vector compact pass. /// * `shutdown_requested` — if true, abort immediately after the current pass. @@ -225,6 +243,7 @@ impl AutovacuumDaemon { manifest: Option<&mut crate::persistence::manifest::ShardManifest>, wal_v3: Option<&mut crate::persistence::wal_v3::segment::WalWriterV3>, max_wal_bytes: u64, + disk_offload_enabled: bool, manifest_retain_epochs: u64, manifest_retain_secs: u64, max_immutable_segments: usize, @@ -333,25 +352,64 @@ impl AutovacuumDaemon { // The stats() call does a read_dir — acceptable at 30 s cadence. match wal.stats() { Ok(wal_stats) if max_wal_bytes > 0 && wal_stats.total_bytes > max_wal_bytes => { - let redo_lsn = wal.current_lsn(); - match wal.recycle_aggressive(redo_lsn) { - Ok(recycled) => { - let pass_ms = pass_start.elapsed().as_millis() as u64; - cumulative_ms += pass_ms; - if recycled.segments_recycled > 0 { - tracing::debug!( - segments_recycled = recycled.segments_recycled, - bytes_reclaimed = recycled.bytes_reclaimed, - pass_ms, - "autovacuum: pass C (WAL recycle) freed segments" + if disk_offload_enabled { + // Disk-offload mode: the checkpoint protocol maintains + // a real redo_lsn and persist_graph_at_checkpoint + // snapshots the graph write-buffer, so recycling + // everything before current_lsn is backed by an + // actual durable floor. UNCHANGED from before task #43. + let redo_lsn = wal.current_lsn(); + match wal.recycle_aggressive(redo_lsn) { + Ok(recycled) => { + let pass_ms = pass_start.elapsed().as_millis() as u64; + cumulative_ms += pass_ms; + if recycled.segments_recycled > 0 { + tracing::debug!( + segments_recycled = recycled.segments_recycled, + bytes_reclaimed = recycled.bytes_reclaimed, + pass_ms, + "autovacuum: pass C (WAL recycle) freed segments" + ); + } + } + Err(e) => { + tracing::warn!( + error = %e, + "autovacuum: pass C (WAL recycle) failed" ); } } - Err(e) => { + } else { + // Task #43 (P1 data loss) fix: legacy (non-disk-offload) + // mode has NO periodic checkpoint or plane snapshot — + // save_graph_store only runs at graceful shutdown, and + // the workspace/MQ/temporal registries have no + // snapshot format at all (replay is WAL-only). Every + // sealed segment may hold the SOLE durable copy of + // that history, so recycle_aggressive's "everything + // before current_lsn is safe" floor does not hold + // here. Never trade data loss for disk space: skip + // the recycle, warn (rate-limited), and count it. + crate::command::info_reclamation::RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL + .fetch_add(1, Ordering::Relaxed); + let should_warn = self + .last_wal_recycle_warn_at + .map_or(true, |t| now.duration_since(t) >= Duration::from_secs(300)); + if should_warn { tracing::warn!( - error = %e, - "autovacuum: pass C (WAL recycle) failed" + wal_bytes = wal_stats.total_bytes, + max_wal_bytes, + "autovacuum: pass C (WAL recycle) SKIPPED — WAL exceeds \ + --max-wal-size but this shard has no checkpoint floor \ + to recycle against (legacy/non-disk-offload mode); \ + recycling here would risk permanently losing graph/ \ + workspace/MQ/temporal history that has no snapshot \ + outside the WAL. Enable --disk-offload for bounded WAL \ + growth, or perform a graceful shutdown (SIGTERM/SHUTDOWN, \ + which persists the graph store) before the WAL grows \ + much further." ); + self.last_wal_recycle_warn_at = Some(now); } } } @@ -796,6 +854,127 @@ impl CompactionScheduler { mod tests { use super::*; + /// Count `*.wal` files under `wal_dir`. + fn count_wal_segments(wal_dir: &std::path::Path) -> usize { + std::fs::read_dir(wal_dir) + .map(|entries| { + entries + .filter_map(|e| e.ok()) + .filter(|e| e.file_name().to_string_lossy().ends_with(".wal")) + .count() + }) + .unwrap_or(0) + } + + /// Build a `WalWriterV3` with enough sealed (non-active) segments to + /// exceed `max_wal_bytes`, so Pass C's ceiling check fires. + fn wal_writer_over_ceiling( + wal_dir: &std::path::Path, + ) -> crate::persistence::wal_v3::segment::WalWriterV3 { + use crate::persistence::wal_v3::record::WalRecordType; + let mut writer = + crate::persistence::wal_v3::segment::WalWriterV3::new(0, wal_dir, 512).unwrap(); + for i in 0..80 { + writer.append(WalRecordType::Command, b"legacy-ws-t43 EARLY-MARKER-T43"); + if (i + 1) % 3 == 0 { + writer.flush_sync().unwrap(); + } + } + writer.flush_sync().unwrap(); + assert!( + writer.current_segment_sequence() >= 3, + "test setup must produce several sealed segments" + ); + writer + } + + /// Task #43 (P1 data loss): in legacy (non-disk-offload) mode, Pass C + /// must NOT delete sealed WAL segments — there is no checkpoint/graph + /// snapshot floor backing `recycle_aggressive`'s "everything before + /// current_lsn is durable elsewhere" assumption. + #[test] + fn test_pass_c_legacy_mode_skips_recycle_and_counts_blocked() { + let tmp = tempfile::tempdir().unwrap(); + let wal_dir = tmp.path().join("wal"); + let mut writer = wal_writer_over_ceiling(&wal_dir); + let before = count_wal_segments(&wal_dir); + + let before_blocked = + crate::command::info_reclamation::RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL + .load(Ordering::Relaxed); + + let mut vector_store = crate::vector::store::VectorStore::new(); + #[cfg(feature = "graph")] + let mut graph_store = crate::graph::store::GraphStore::new(); + let mut daemon = AutovacuumDaemon::new(AutovacuumConfig::default()); + daemon.run_tick( + &mut vector_store, + #[cfg(feature = "graph")] + &mut graph_store, + None, + Some(&mut writer), + 256, // max_wal_bytes: tiny, so the ceiling is already breached + false, // disk_offload_enabled: legacy mode — the fix under test + 0, + 0, + usize::MAX, + usize::MAX, + 1.0, + false, + ); + + let after = count_wal_segments(&wal_dir); + assert_eq!( + after, before, + "legacy mode must not delete any WAL segment (task #43)" + ); + let after_blocked = + crate::command::info_reclamation::RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL + .load(Ordering::Relaxed); + assert!( + after_blocked > before_blocked, + "legacy-mode skip must be observable via RECL_WAL_RECYCLE_BLOCKED_NO_CHECKPOINT_TOTAL" + ); + } + + /// Disk-offload mode is unchanged by task #43: Pass C still recycles + /// aggressively against `current_lsn()`, exactly as before this fix. + #[test] + fn test_pass_c_disk_offload_mode_recycles_unchanged() { + let tmp = tempfile::tempdir().unwrap(); + let wal_dir = tmp.path().join("wal"); + let mut writer = wal_writer_over_ceiling(&wal_dir); + let before = count_wal_segments(&wal_dir); + assert!(before >= 3, "test setup should produce 3+ segments"); + + let mut vector_store = crate::vector::store::VectorStore::new(); + #[cfg(feature = "graph")] + let mut graph_store = crate::graph::store::GraphStore::new(); + let mut daemon = AutovacuumDaemon::new(AutovacuumConfig::default()); + daemon.run_tick( + &mut vector_store, + #[cfg(feature = "graph")] + &mut graph_store, + None, + Some(&mut writer), + 256, // max_wal_bytes: tiny, so the ceiling is already breached + true, // disk_offload_enabled: checkpoint-backed mode — unchanged + 0, + 0, + usize::MAX, + usize::MAX, + 1.0, + false, + ); + + let after = count_wal_segments(&wal_dir); + assert!( + after < before, + "disk-offload mode must still recycle sealed segments \ + (before={before}, after={after}) — unchanged by task #43" + ); + } + #[test] fn test_latency_window_p95_empty() { let w = LatencyWindow::new(); diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 2210a222..47e65834 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -1743,6 +1743,7 @@ impl super::Shard { shard_manifest.as_mut(), wal_writer.as_mut(), server_config.max_wal_size_bytes(), + server_config.disk_offload_enabled(), server_config.manifest_tombstone_retain_epochs, server_config.manifest_tombstone_retain_secs, server_config.max_unflushed_immutable_segments as usize, @@ -2351,6 +2352,7 @@ impl super::Shard { shard_manifest.as_mut(), wal_writer.as_mut(), server_config.max_wal_size_bytes(), + server_config.disk_offload_enabled(), server_config.manifest_tombstone_retain_epochs, server_config.manifest_tombstone_retain_secs, server_config.max_unflushed_immutable_segments as usize, From d7e99f4c6f2cf46e50b6356cb2dafc942cc3ae29 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 05:54:26 +0700 Subject: [PATCH 3/3] docs(test): add SAFETY comment to sigkill unsafe block (task #43 review) CodeRabbit review finding on PR #288: the libc::kill FFI call in the crash_recovery_wal_recycle_legacy harness lacked the repo-mandated // SAFETY: justification. Documents the invariant (owned live Child, targeted pid, no memory-safety preconditions). author: Tin Dang --- tests/crash_recovery_wal_recycle_legacy.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/crash_recovery_wal_recycle_legacy.rs b/tests/crash_recovery_wal_recycle_legacy.rs index 634b23cd..67ec669a 100644 --- a/tests/crash_recovery_wal_recycle_legacy.rs +++ b/tests/crash_recovery_wal_recycle_legacy.rs @@ -139,6 +139,9 @@ impl Drop for ServerGuard { #[cfg(unix)] fn sigkill(child: &mut Child) { let pid = child.id() as i32; + // SAFETY: `pid` is the live child's own process id (owned `Child`, not yet + // waited), so the SIGKILL targets exactly the server this guard spawned; + // `libc::kill` has no memory-safety preconditions beyond a valid signal. unsafe { libc::kill(pid, libc::SIGKILL); }