diff --git a/CHANGELOG.md b/CHANGELOG.md index caf85147..5643aa84 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -134,6 +134,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 wall-clock timeout with zero actual test failures. Bumped to 30m to match the Windows Test step's headroom; removes recurring false-negative reds on rebased branches. +### Fixed — Disk-offload: crash recovery no longer resurrects deleted cold keys, nor drops the AOF after a spill (PR #TBD) + +- **Deleted/flushed cold keys stayed deleted only until the next crash.** A + DEL/UNLINK/FLUSHALL/FLUSHDB of a spilled key tombstones the in-memory + ColdIndex, but the manifest entry stays `Active` until the orphan sweep. A + kill-9 inside that window lost the tombstone, and boot-time replay could not + re-apply it: the replayed DEL ran against databases whose `cold_index` was + `None` — a silent cold-plane no-op — so the key resurrected via cold + read-through (measured 85–97/200 probes under `--appendonly yes` + + `--disk-offload`). Two detach points fixed: (a) `recover_shard_v3_pitr` now + attaches the Phase-3-rebuilt ColdIndex to the databases BEFORE Phase 4 WAL + replay (was: stashed on `RecoveryResult`, merged only after recovery + returned); (b) the per-shard/multi-part AOF replay now keeps cold wiring + live during incr replay — main.rs re-attaches it after the pre-replay hot + wipe, and `replay_per_shard`/`replay_multi_part` bridge it across + `rdb::load`'s wholesale database swap (which silently dropped it). New + crash suite `tests/crash_recovery_cold_del_resurrection.rs` (DEL + FLUSHALL + scenarios, kill-9 inside the pre-sweep window). +- **Any disk-offload spill silently discarded the AOF on the next restart.** + The "WAL replayed 0 commands → fall back to the AOF" gate counted vector + + file-lifecycle records: after a spill wrote `FileCreate` records into the + shard WAL, the gate saw a "non-empty" WAL, skipped `appendonly.aof` (the + only complete KV history — `--wal-kv-log` is auto-off when the AOF is the + authority), and every KV write since the last snapshot was lost. The gate + now keys on KV `Command` records only. ### Fixed — Conn-plane: monoio central listener joins the SO_REUSEPORT group (PR #250) diff --git a/src/main.rs b/src/main.rs index d739554e..aee367ea 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1174,12 +1174,13 @@ fn main() -> anyhow::Result<()> { // (eviction.rs::next_spill_file_id_seed): the seed keeps recovered cold // files immutable so these preserved entries stay valid until the // steady-state cascade refreshes them. - let preserved_cold_wiring: Vec< + type PreservedColdWiring = Vec< Vec<( Option, Option, )>, - > = if disk_offload_base.is_some() { + >; + let mut preserved_cold_wiring: PreservedColdWiring = if disk_offload_base.is_some() { shards .iter_mut() .map(|s| { @@ -1192,6 +1193,23 @@ fn main() -> anyhow::Result<()> { } else { Vec::new() }; + // Consuming re-attach: first call restores the wiring, later calls no-op. + // Called (a) after each replay branch's hot wipe and BEFORE its replay — + // replayed DEL/UNLINK/FLUSH*/EXPIRE-past must tombstone the cold plane, + // and with `cold_index == None` those paths are silent cold no-ops, so a + // key deleted in the AOF tail resurrects via cold read-through after + // restart — and (b) unconditionally after the whole block for the + // branches that never replay (fresh boot, tokio single-shard). + // `rdb::load`'s wholesale `*live = temp` swap inside the replay is + // bridged separately (take/restore around the load in shard_replay.rs). + fn reattach_cold_wiring(shards: &mut [moon::shard::Shard], wiring: &mut PreservedColdWiring) { + for (shard, dbs) in shards.iter_mut().zip(std::mem::take(wiring)) { + for (db, (cold_shard_dir, cold_index)) in shard.databases.iter_mut().zip(dbs) { + db.cold_shard_dir = cold_shard_dir; + db.cold_index = cold_index; + } + } + } if config.appendonly == "yes" && let Some(ref dir) = persistence_dir { @@ -1220,6 +1238,8 @@ fn main() -> anyhow::Result<()> { for db in shards[0].databases.iter_mut() { db.clear(); } + // Cold wiring live during replay — see reattach_cold_wiring. + reattach_cold_wiring(&mut shards, &mut preserved_cold_wiring); let loaded = moon::persistence::aof_manifest::replay_multi_part( &mut shards[0].databases, manifest, @@ -1280,6 +1300,8 @@ fn main() -> anyhow::Result<()> { db.clear(); } } + // Cold wiring live during replay — see reattach_cold_wiring. + reattach_cold_wiring(&mut shards, &mut preserved_cold_wiring); // Borrow each shard's `databases` mutably and route through // `replay_per_shard`. The split_at_mut walk constructs a @@ -1470,19 +1492,14 @@ fn main() -> anyhow::Result<()> { // removed: multi-shard PerShard AOF is now loaded on tokio too, and the // single-shard tokio warn is emitted inline above.) - // ── B-2: re-attach cold-tier wiring dropped by the rdb::load swap ─────── - // Restore the `cold_shard_dir` + rebuilt `cold_index` captured before the - // AOF replay block. Without this, the handler reads a database with - // `cold_shard_dir = None`, so every read-through of a re-evicted key misses - // and disk-offload silently loses data across a restart. - if !preserved_cold_wiring.is_empty() { - for (shard, dbs) in shards.iter_mut().zip(preserved_cold_wiring) { - for (db, (cold_shard_dir, cold_index)) in shard.databases.iter_mut().zip(dbs) { - db.cold_shard_dir = cold_shard_dir; - db.cold_index = cold_index; - } - } - } + // ── B-2: re-attach cold-tier wiring for the branches that never replayed ─ + // The replay branches above already consumed the wiring (it must be live + // DURING replay so DEL/FLUSH tombstone the cold plane); this covers fresh + // boots and the tokio single-shard path. Without it, the handler reads a + // database with `cold_shard_dir = None`, so every read-through of a + // re-evicted key misses and disk-offload silently loses data across a + // restart. No-op when already consumed. + reattach_cold_wiring(&mut shards, &mut preserved_cold_wiring); // Extract databases from all shards and wrap in ShardDatabases let all_dbs: Vec> = shards diff --git a/src/persistence/aof_manifest/shard_replay.rs b/src/persistence/aof_manifest/shard_replay.rs index 08bb5849..a6ce0652 100644 --- a/src/persistence/aof_manifest/shard_replay.rs +++ b/src/persistence/aof_manifest/shard_replay.rs @@ -6,6 +6,40 @@ use super::*; +/// Cold-tier wiring (`cold_shard_dir` + `cold_index`) captured per database so +/// it survives the `rdb::load` swap. See [`take_cold_wiring`]. +type ColdWiring = Vec<( + Option, + Option, +)>; + +/// Detach cold-tier wiring from every database before an `rdb::load`. +/// +/// `rdb::load` loads the base RDB into fresh `Database::new()` temporaries and +/// swaps them wholesale into the live databases (`*live = temp`). That swap +/// silently drops `cold_shard_dir` and the recovery-rebuilt `cold_index` — +/// live-tier topology, not part of the RDB hot snapshot. Capture both here and +/// re-attach with [`restore_cold_wiring`] BEFORE the incr replay: replayed +/// DEL/UNLINK/FLUSH*/EXPIRE-past must tombstone the cold plane, and with +/// `cold_index == None` those paths (`remove_counting_cold()`/`clear()`) are +/// silent cold no-ops, so a key deleted in the incr tail resurrects via cold +/// read-through after restart (its manifest entry stays Active until the +/// orphan sweep). +fn take_cold_wiring(databases: &mut [crate::storage::Database]) -> ColdWiring { + databases + .iter_mut() + .map(|db| (db.cold_shard_dir.take(), db.cold_index.take())) + .collect() +} + +/// Re-attach wiring captured by [`take_cold_wiring`]. +fn restore_cold_wiring(databases: &mut [crate::storage::Database], wiring: ColdWiring) { + for (db, (cold_shard_dir, cold_index)) in databases.iter_mut().zip(wiring) { + db.cold_shard_dir = cold_shard_dir; + db.cold_index = cold_index; + } +} + /// Replay multi-part AOF: load base RDB then replay incremental RESP. /// /// Returns total keys/commands loaded. @@ -16,10 +50,15 @@ pub fn replay_multi_part( ) -> Result { let mut total = 0usize; - // Load base RDB + // Load base RDB. Cold wiring must survive the swap AND be live for the + // incr replay below (replayed DEL/FLUSH must tombstone the cold plane) — + // see take_cold_wiring. let base_path = manifest.base_path(); if base_path.exists() { - match crate::persistence::rdb::load(databases, &base_path) { + let cold_wiring = take_cold_wiring(databases); + let load_result = crate::persistence::rdb::load(databases, &base_path); + restore_cold_wiring(databases, cold_wiring); + match load_result { Ok(n) => { info!( "AOF base RDB loaded: {} keys from {}", @@ -439,9 +478,15 @@ pub fn replay_per_shard( let mut shard_max_lsn: u64 = 0; let mut shard_ordered: Vec = Vec::new(); - // Load this shard's base RDB. + // Load this shard's base RDB. Cold wiring must survive the + // swap AND be live for the incr replay below (replayed + // DEL/FLUSH must tombstone the cold plane) — see + // take_cold_wiring. if base_path.exists() { - match crate::persistence::rdb::load(*databases, &base_path) { + let cold_wiring = take_cold_wiring(databases); + let load_result = crate::persistence::rdb::load(*databases, &base_path); + restore_cold_wiring(databases, cold_wiring); + match load_result { Ok(n) => { info!( "AOF shard-{} base RDB loaded: {} keys from {}", diff --git a/src/persistence/recovery.rs b/src/persistence/recovery.rs index c7cba2e5..6153df26 100644 --- a/src/persistence/recovery.rs +++ b/src/persistence/recovery.rs @@ -48,8 +48,9 @@ pub struct RecoveryResult { pub cold_segments: Vec<(u64, std::path::PathBuf)>, /// Number of cold segments discovered. pub cold_segments_loaded: usize, - /// Cold index rebuilt from heap DataFiles (None if no KvLeaf entries). - pub cold_index: Option, + // NOTE: the ColdIndex rebuilt in Phase 3 is attached directly to + // `databases[0]` BEFORE Phase 4 replay (never returned here) so that + // replayed deletes tombstone the cold plane — see the Phase 3 comment. } /// 6-phase recovery protocol for disk-offload mode. @@ -329,7 +330,20 @@ pub fn recover_shard_v3_pitr( shard_id, cold_idx.len() ); - result.cold_index = Some(cold_idx); + // Attach to the database BEFORE Phase 4 replay. Replayed + // DEL/UNLINK/FLUSH*/EXPIRE-past must tombstone the cold plane: + // with `cold_index == None`, `remove_counting_cold()`/`clear()` + // are silent no-ops (`as_mut()` on None), so a key deleted in + // the WAL tail resurrects via cold read-through after restart + // whenever the crash lands inside the pre-orphan-sweep window + // (the manifest entry is still Active until the sweep). + if let Some(db0) = databases.first_mut() { + db0.cold_shard_dir = Some(shard_dir.to_path_buf()); + match db0.cold_index.as_mut() { + Some(existing) => existing.merge(cold_idx), + None => db0.cold_index = Some(cold_idx), + } + } } // Crash-orphan sweep: heap files written but never registered in // the manifest (crash between spill write and manifest commit) @@ -369,6 +383,14 @@ pub fn recover_shard_v3_pitr( } // ── Phase 4: WAL REPLAY ─────────────────────────────────────────── + // KV `Command` records counted separately from vector/file-lifecycle + // records: the Phase 4b "WAL gave us no KV history → fall back to the + // AOF" gate must key on THIS count. Gating on the aggregate + // `commands_replayed` silently skipped the AOF (the only complete KV + // history — `wal_kv_log` is auto-off when the AOF is the authority) + // whenever disk-offload spills had written FileCreate records into the + // WAL: every KV write since the last snapshot was lost on restart. + let mut kv_commands_replayed = 0usize; let wal_dir = shard_dir.join("wal-v3"); if wal_dir.exists() { let mut selected_db = 0usize; @@ -395,6 +417,7 @@ pub fn recover_shard_v3_pitr( &mut selected_db, ); result.commands_replayed += 1; + kv_commands_replayed += 1; } } } @@ -584,28 +607,38 @@ pub fn recover_shard_v3_pitr( } // ── Phase 4b: legacy-dir AOF / WAL v3 FALLBACK ───────────────────── - // When the disk-offload WAL v3 replay (Phase 4, above) produced 0 - // commands and a legacy persistence directory is available, fall back to - // it. `appendonly.aof` is the AUTHORITY there (post-#211, WAL v3's KV - // coverage is intentionally partial: `--wal-kv-log` default-off, and - // connection-local writes bypass it even when on — a non-empty WAL v3 - // must never shadow the AOF). Only when NO AOF exists at all does the - // legacy-mode WAL v3 directory (`v2_dir/shard-N/wal-v3/`, distinct from - // `shard_dir`'s Phase-4 WAL despite the shared name) serve as a - // last-resort partial recovery source. The WAL v2 rung (per-shard - // `shard-N.wal`) was removed in the pre-1.0 WAL-v3-only format freeze - // and is never replayed by this build. - if result.commands_replayed == 0 { + // When the disk-offload WAL v3 replay (Phase 4, above) produced 0 KV + // `Command` records and a legacy persistence directory is available, fall + // back to it. The gate is `kv_commands_replayed` — NOT the aggregate + // `commands_replayed`, which also counts vector + file-lifecycle records: + // disk-offload spills write FileCreate records into the same WAL, and + // gating on the aggregate skipped the AOF (the only complete KV history) + // on every post-spill restart. `appendonly.aof` is the AUTHORITY there + // (post-#211, WAL v3's KV coverage is intentionally partial: + // `--wal-kv-log` default-off, and connection-local writes bypass it even + // when on — a non-empty WAL v3 must never shadow the AOF). Only when NO + // AOF exists at all does the legacy-mode WAL v3 directory + // (`v2_dir/shard-N/wal-v3/`, distinct from `shard_dir`'s Phase-4 WAL + // despite the shared name) serve as a last-resort partial recovery + // source. The WAL v2 rung (per-shard `shard-N.wal`) was removed in the + // pre-1.0 WAL-v3-only format freeze and is never replayed by this build. + // + // Known follow-up: when the Phase-4 WAL DID carry some KV records + // (`--wal-kv-log on` / CDC active) alongside an existing AOF, this gate + // keeps the WAL's partial KV view and still skips the AOF (replaying + // both would double-apply non-idempotent commands). Resolving that needs + // an AOF-first redesign of Phase 4, tracked in the roadmap. + if kv_commands_replayed == 0 { if let Some(v2_dir) = v2_persistence_dir { let aof_path = v2_dir.join("appendonly.aof"); if aof_path.exists() { info!( - "Shard {}: WAL empty, falling back to AOF replay from {:?}", + "Shard {}: WAL carried no KV commands, falling back to AOF replay from {:?}", shard_id, aof_path ); match crate::persistence::aof::replay_aof(databases, &aof_path, engine) { Ok(n) => { - result.commands_replayed = n; + result.commands_replayed += n; info!("Shard {}: AOF fallback replayed {} commands", shard_id, n); } Err(e) => { @@ -626,7 +659,7 @@ pub fn recover_shard_v3_pitr( ) { Ok(0) => {} Ok(n) => { - result.commands_replayed = n; + result.commands_replayed += n; tracing::warn!( "Shard {}: no appendonly.aof found — replayed {} records \ from legacy-mode WAL v3 at {:?} as a LAST-RESORT \ @@ -1236,6 +1269,44 @@ mod tests { ); } + #[test] + fn test_spill_lifecycle_records_do_not_suppress_aof_fallback() { + // Disk-offload spills write FileCreate/FileDelete/FileTierChange + // records into the shard's own wal-v3 — but NO KV Command records + // (`--wal-kv-log` is auto-off when the AOF is the authority). The + // Phase 4b gate must key on KV Command records only: gating on the + // aggregate count made any post-spill restart skip the AOF entirely, + // losing every KV write since the last snapshot (and resurrecting + // deleted cold keys, since the AOF'd DELs never replayed). + let tmp = tempfile::tempdir().unwrap(); + let shard_dir = tmp.path().join("shard-0"); + let offload_wal_dir = shard_dir.join("wal-v3"); + std::fs::create_dir_all(&offload_wal_dir).unwrap(); + let mut wal_data = make_v3_header(0); + // One spill lifecycle record: 16-byte payload (file_id + page_offset). + write_wal_v3_record(&mut wal_data, 1, WalRecordType::FileCreate, &[0u8; 16]); + std::fs::write(offload_wal_dir.join("000000000001.wal"), &wal_data).unwrap(); + + let v2_dir = tmp.path().join("legacy"); + std::fs::create_dir_all(&v2_dir).unwrap(); + let aof_payload = b"*1\r\n$4\r\nPING\r\n*1\r\n$4\r\nPING\r\n"; // 2 KV records + std::fs::write(v2_dir.join("appendonly.aof"), aof_payload).unwrap(); + + let mut databases = vec![Database::new()]; + let engine = crate::persistence::replay::DispatchReplayEngine::new(); + let result = + recover_shard_v3_with_fallback(&mut databases, 0, &shard_dir, &engine, Some(&v2_dir)) + .unwrap(); + + assert_eq!( + result.commands_replayed, 3, + "1 FileCreate lifecycle record + 2 AOF commands must all be \ + replayed -- got {} (1 means the lifecycle record suppressed the \ + AOF fallback and every KV write was dropped)", + result.commands_replayed + ); + } + #[test] fn test_legacy_wal_v3_last_resort_when_no_aof() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/shard/mod.rs b/src/shard/mod.rs index 43e58110..9eb3baaf 100644 --- a/src/shard/mod.rs +++ b/src/shard/mod.rs @@ -170,6 +170,11 @@ impl Shard { ); // Initialize cold_index + cold_shard_dir on all databases // so cold_read_through can find keys spilled to NVMe. + // The recovered index itself was already attached to + // databases[0] BEFORE Phase 4 replay (inside + // recover_shard_v3_pitr) so replayed DEL/FLUSH/EXPIRE + // tombstone the cold plane; here we only backfill empty + // indexes on the remaining databases. { let cold_dir = shard_dir.clone(); for db in &mut self.databases { @@ -179,11 +184,6 @@ impl Shard { Some(crate::storage::tiered::cold_index::ColdIndex::new()); } } - if let Some(recovered_ci) = result.cold_index { - if let Some(ref mut ci) = self.databases[0].cold_index { - ci.merge(recovered_ci); - } - } } // Vector recovery: the `Shard`-owned `vector_store` diff --git a/tests/crash_recovery_cold_del_resurrection.rs b/tests/crash_recovery_cold_del_resurrection.rs new file mode 100644 index 00000000..420ca901 --- /dev/null +++ b/tests/crash_recovery_cold_del_resurrection.rs @@ -0,0 +1,411 @@ +//! CRASH-COLD-DEL: deleted/flushed spilled keys must NOT resurrect after a crash. +//! +//! Guards the replay-time cold-plane tombstone gap. Under `--appendonly yes` +//! (the only config where KV writes are durably logged — `--appendonly no` +//! skips the WAL writer entirely, so post-crash resurrection there is the +//! documented no-AOF RPO, not a bug), DEL/FLUSHALL are replayed at boot from +//! the per-shard AOF (`replay_per_shard`, main.rs). Pre-fix, the B-2 +//! cold-wiring preservation in main.rs `take()`s `cold_index` off every +//! database BEFORE the replay and re-attaches it only AFTER, so the replayed +//! DEL/FLUSHALL hit `remove_counting_cold`/`clear()` as silent no-ops against +//! the cold plane (src/storage/db.rs — `as_mut()` on None). The same gap +//! existed in `recover_shard_v3_pitr` Phase 4 WAL replay (ColdIndex built in +//! Phase 3 but attached only after recovery returned). +//! +//! Failure scenario (RED without the fix): +//! 1. SET probes → filler load evicts them to the cold tier (.mpf + manifest +//! commit, durable). +//! 2. DEL every probe. Live path tombstones the in-memory ColdIndex and queues +//! the backing file for the orphan sweep — but the manifest entry stays +//! Active until the sweep runs (held off here via a 1h sweep interval). +//! The DEL is durably logged in the per-shard AOF incr file. +//! 3. SIGKILL inside the pre-sweep window. +//! 4. Restart: v3 recovery Phase 3 rebuilds the ColdIndex from the +//! still-Active manifest entries; the per-shard AOF replay then replays +//! SET (hot) and DEL (hot removed; cold no-op — index detached). +//! 5. GET probe → hot miss → cold read-through returns the DELETED value. +//! +//! GREEN with the fix (cold wiring live during incr replay, preserved across +//! the base-RDB swap inside `replay_per_shard`): the replayed DEL tombstones +//! the cold entry; every post-crash GET is nil. +//! +//! Run with (monoio default — matches CI): +//! cargo build --release +//! cargo test --release --test crash_recovery_cold_del_resurrection -- --ignored +//! +//! Requires: built release binary, `redis-cli` on PATH. + +#![cfg(any(feature = "runtime-monoio", feature = "runtime-tokio"))] + +use std::io::Write; +use std::process::{Child, Command, Stdio}; +use std::time::Duration; + +const PROBE_COUNT: usize = 200; +const PROBE_VALUE_LEN: usize = 500; +/// Filler keys written after the probes to push memory past the disk-offload +/// threshold, forcing the (older) probe keys to be evicted to the cold tier. +const FILLER_COUNT: usize = 16_000; +const FILLER_VALUE_LEN: usize = 600; +/// 8 MiB total across 4 shards (2 MiB/shard); disk-offload spills at +/// 0.85 × maxmemory. Matches crash_recovery_disk_offload_no_aof.rs, whose +/// recovery floor proves most probes land (and stay) cold under this load. +const MAXMEMORY_BYTES: usize = 8 * 1024 * 1024; +const SHARDS: usize = 4; +/// Seconds to let the async spill/manifest ticks drain so the probes' cold +/// entries are manifest-committed before we DEL them. +const SETTLE_AFTER_FILLER: u64 = 8; +/// Seconds after the DELs for the per-shard AOF writer (appendfsync everysec) +/// to make them durable before the SIGKILL. +const SETTLE_AFTER_DEL: u64 = 3; + +fn unique_port() -> u16 { + use std::net::TcpListener; + let listener = TcpListener::bind("127.0.0.1:0").expect("bind to port 0"); + let port = listener.local_addr().expect("local addr").port(); + drop(listener); + port +} + +fn unique_dir(suffix: &str) -> std::path::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-cold-del-{}-{}-{}", + std::process::id(), + suffix, + nanos + )) +} + +fn start_moon(port: u16, dir: &std::path::Path) -> Child { + let off_dir = dir.join("off"); + std::fs::create_dir_all(&off_dir).expect("create off dir"); + Command::new("./target/release/moon") + .args([ + "--port", + &port.to_string(), + "--shards", + &SHARDS.to_string(), + "--maxmemory", + &MAXMEMORY_BYTES.to_string(), + "--maxmemory-policy", + "allkeys-lru", + "--disk-offload", + "enable", + "--disk-offload-dir", + off_dir.to_str().expect("off dir utf8"), + // `yes` is the bug surface: KV writes (incl. the DELs) are durably + // logged in the per-shard AOF and replayed at boot. Under + // `--appendonly no` the WAL writer is skipped entirely + // ("WAL skipped (appendonly=no)") — nothing replays, and cold + // resurrection there is the documented no-AOF RPO, not this bug. + "--appendonly", + "yes", + // Hold the pre-sweep window open for the whole test: the DEL's + // manifest tombstone is deferred to this sweep, and the bug under + // test lives precisely in the crash-before-sweep window. + "--cold-orphan-sweep-interval-secs", + "3600", + // The diskfull guard write-flags DEL/FLUSHALL too (documented + // trade-off) and dev/CI machines routinely sit under 5% free — + // with the guard active the DELs under test would be REJECTED + // (MOONERR diskfull) and never reach the AOF, silently gutting + // the test (redis-cli exits 0 on error replies). + "--disk-free-min-pct", + "0", + "--dir", + ]) + .arg(dir) + // Captured to a log file so a CI flake produces a real diagnostic + // (never Stdio::null()). + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create moon stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create moon stderr log")) + .spawn() + .expect("spawn moon (run `cargo build --release` with default features first)") +} + +fn wait_for_port(port: u16) { + for _ in 0..80 { + if std::net::TcpStream::connect(format!("127.0.0.1:{}", port)).is_ok() { + std::thread::sleep(Duration::from_millis(200)); + return; + } + std::thread::sleep(Duration::from_millis(100)); + } + panic!("moon did not start within 8s on port {}", port); +} + +/// See crash_recovery_disk_offload_no_aof.rs: SO_REUSEPORT makes a bind-probe +/// useless; poll until connect is REFUSED twice in a row. +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 std::net::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)); + } + } + } +} + +const RESTART_ATTEMPTS: usize = 6; + +/// Start moon, retrying the transient rebind EADDRINUSE self-shutdown race +/// (see crash_recovery_disk_offload_no_aof.rs for the full rationale). +fn start_moon_alive(port: u16, dir: &std::path::Path) -> Child { + for attempt in 1..=RESTART_ATTEMPTS { + let mut child = start_moon(port, dir); + let mut up = false; + for _ in 0..80 { + if let Ok(Some(_status)) = child.try_wait() { + break; + } + if std::net::TcpStream::connect(format!("127.0.0.1:{}", port)).is_ok() { + std::thread::sleep(Duration::from_millis(200)); + up = true; + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + if up { + return child; + } + let _ = child.kill(); + let _ = child.wait(); + if attempt < RESTART_ATTEMPTS { + std::thread::sleep(Duration::from_millis(300)); + } + } + panic!( + "moon failed to start+serve on port {} after {} attempts", + port, RESTART_ATTEMPTS + ); +} + +fn probe_key(i: usize) -> String { + format!("probe:{}", i) +} + +/// redis-cli exits 0 even when the server replies an error (MOONERR/ERR land +/// on stdout/stderr, not the exit status) — every mutation helper must check +/// the reply text too, or a server-side rejection (e.g. the diskfull guard) +/// silently guts the test. +fn assert_no_error_reply(op: &str, out: &std::process::Output) { + let stdout = String::from_utf8_lossy(&out.stdout); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!( + out.status.success() && !stdout.contains("MOONERR") && !stdout.starts_with("ERR"), + "redis-cli {} rejected by server: stdout={} stderr={}", + op, + stdout.trim(), + stderr.trim() + ); +} + +fn redis_set(port: u16, key: &str, value: &str) { + let out = Command::new("redis-cli") + .args(["-p", &port.to_string(), "SET", key, value]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .expect("redis-cli SET"); + assert_no_error_reply(&format!("SET {}", key), &out); +} + +fn redis_get(port: u16, key: &str) -> Option { + let out = Command::new("redis-cli") + .args(["-p", &port.to_string(), "GET", key]) + .output() + .expect("redis-cli GET"); + if !out.status.success() { + return None; + } + let s = String::from_utf8_lossy(&out.stdout).trim().to_string(); + if s.is_empty() || s == "(nil)" { + None + } else { + Some(s) + } +} + +/// DEL the probes in batches of 50 keys per redis-cli invocation. +fn redis_del_probes(port: u16) { + for chunk in (0..PROBE_COUNT).collect::>().chunks(50) { + let mut args = vec!["-p".to_string(), port.to_string(), "DEL".to_string()]; + args.extend(chunk.iter().map(|i| probe_key(*i))); + let out = Command::new("redis-cli") + .args(&args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .expect("redis-cli DEL"); + assert_no_error_reply("DEL batch", &out); + } +} + +fn redis_flushall(port: u16) { + let out = Command::new("redis-cli") + .args(["-p", &port.to_string(), "FLUSHALL"]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .expect("redis-cli FLUSHALL"); + assert_no_error_reply("FLUSHALL", &out); +} + +/// Pipelined filler SETs to push memory past the offload threshold. +fn write_filler(port: u16) { + let mut stream = + std::net::TcpStream::connect(format!("127.0.0.1:{}", port)).expect("connect for filler"); + stream.set_write_timeout(Some(Duration::from_secs(30))).ok(); + let val = "F".repeat(FILLER_VALUE_LEN); + let mut buf: Vec = Vec::with_capacity(64 * 1024); + for i in 0..FILLER_COUNT { + let key = format!("filler:{}", i); + let cmd = format!( + "*3\r\n$3\r\nSET\r\n${}\r\n{}\r\n${}\r\n{}\r\n", + key.len(), + key, + val.len(), + val + ); + buf.extend_from_slice(cmd.as_bytes()); + if buf.len() >= 64 * 1024 { + stream.write_all(&buf).expect("filler write"); + buf.clear(); + } + } + if !buf.is_empty() { + stream.write_all(&buf).expect("filler tail write"); + } + stream.flush().ok(); +} + +fn count_heap_files(dir: &std::path::Path) -> usize { + let off = dir.join("off"); + fn walk(p: &std::path::Path, acc: &mut usize) { + if let Ok(rd) = std::fs::read_dir(p) { + 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(&off, &mut acc); + acc +} + +/// Shared body: load → spill → delete via `delete_fn` → crash → restart → +/// assert zero resurrections. +fn run_scenario(suffix: &str, delete_fn: impl Fn(u16)) { + let port = unique_port(); + let dir = unique_dir(suffix); + std::fs::create_dir_all(&dir).expect("create test dir"); + + // Round 1: load, spill, delete, crash. + let mut server = start_moon(port, &dir); + wait_for_port(port); + + let probe_val = "P".repeat(PROBE_VALUE_LEN); + for i in 0..PROBE_COUNT { + redis_set(port, &probe_key(i), &probe_val); + } + write_filler(port); + std::thread::sleep(Duration::from_secs(SETTLE_AFTER_FILLER)); + + // Precondition: the probes really were spilled to the cold tier. Without + // heap files on disk this test has no power (nothing to resurrect). + let heap_files = count_heap_files(&dir); + assert!( + heap_files > 0, + "precondition failed: no heap-*.mpf files — filler did not force a spill" + ); + + delete_fn(port); + // Pre-crash sanity: the deletion must be visible on the LIVE server. + // Guards against server-side rejection (e.g. a still-active write guard) + // silently gutting the test — the crash assertions below are meaningless + // if the keys were never deleted in the first place. + assert!( + redis_get(port, &probe_key(0)).is_none(), + "precondition failed: probe:0 still readable after delete — the \ + deletion was rejected server-side (write guard?)" + ); + std::thread::sleep(Duration::from_secs(SETTLE_AFTER_DEL)); + + // Hard crash inside the pre-sweep window (sweep interval is 1h). + server.kill().expect("SIGKILL moon"); + let _ = server.wait(); + wait_for_port_down(port); + + // Round 2: recover and check for resurrections. + let mut server2 = start_moon_alive(port, &dir); + + let mut resurrected = 0usize; + let mut example = String::new(); + for i in 0..PROBE_COUNT { + if redis_get(port, &probe_key(i)).is_some() { + resurrected += 1; + if example.is_empty() { + example = probe_key(i); + } + } + } + + server2.kill().expect("kill moon round 2"); + let _ = server2.wait(); + // Keep the data dir + server logs for post-mortem when the assertion is + // about to fail (or when explicitly requested via MOON_TEST_KEEP=1). + if resurrected == 0 && std::env::var("MOON_TEST_KEEP").is_err() { + let _ = std::fs::remove_dir_all(&dir); + } else { + eprintln!("preserved test dir for diagnosis: {}", dir.display()); + } + + assert_eq!( + resurrected, 0, + "{} deleted key(s) resurrected from the cold tier after crash recovery \ + (first: {}); Phase-4 replay did not tombstone the cold plane \ + (heap files at kill time: {})", + resurrected, example, heap_files + ); +} + +/// DEL'd spilled keys must stay deleted across a kill-9 (pre-sweep window). +#[test] +#[ignore] // requires ./target/release/moon + redis-cli; run with -- --ignored +fn deleted_cold_keys_stay_deleted_after_crash() { + run_scenario("del", redis_del_probes); +} + +/// FLUSHALL must clear the cold plane across a kill-9 too (Database::clear()'s +/// cold_all path has the same None-during-replay gap as DEL). +#[test] +#[ignore] // requires ./target/release/moon + redis-cli; run with -- --ignored +fn flushed_cold_keys_stay_flushed_after_crash() { + run_scenario("flushall", |port| redis_flushall(port)); +}