From 052b392b717c82b0fa18e8cdeae39bdf090b7fc9 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 21:29:26 +0700 Subject: [PATCH 1/7] test(replication): RED suite for Wave A plane replication (task #34) 8 black-box tests, all failing on the missing features by design: eviction parity (s1+s4), EVAL effects parity (s1+s4), EVAL-write AOF durability across kill-9, evicted-key resurrection from AOF replay, expiry DEL emission (offset must advance), SELECT-in-script loud error. author: Tin Dang --- tests/replication_planes.rs | 945 ++++++++++++++++++++++++++++++++++++ 1 file changed, 945 insertions(+) create mode 100644 tests/replication_planes.rs diff --git a/tests/replication_planes.rs b/tests/replication_planes.rs new file mode 100644 index 00000000..c79aaa6b --- /dev/null +++ b/tests/replication_planes.rs @@ -0,0 +1,945 @@ +//! Wave A plane-replication RED suite (task #34, `.planning/rfcs/plane-replication-design.md`). +//! +//! Today (before Wave A) three classes of master-side key removal reach +//! NEITHER the AOF plane NOR the replication plane: +//! +//! 1. Active-expiry DELs (background TTL sweep). +//! 2. Eviction plain-drops (`--maxmemory` + an evicting policy). +//! 3. Lua `redis.call` write *effects* (EVAL/EVALSHA carry no WRITE +//! command-metadata flag, so the generic AOF/replication gate never +//! sees them; the script's own writes vanish on restart AND never +//! reach an attached replica). +//! +//! `redis.call('SELECT', ...)` inside a script also silently corrupts the +//! wrong db today instead of erroring loudly. +//! +//! Six black-box scenarios pin these gaps down. Run: +//! +//! ```text +//! MOON_BIN=./target/release/moon \ +//! cargo test --test replication_planes -- --ignored --nocapture +//! ``` +//! +//! Server dirs use `tempfile::tempdir()` (respects `$TMPDIR`, which on this +//! host resolves to the boot volume with ample free space) PLUS +//! `--disk-free-min-pct 0` on every spawn, matching +//! `tests/replication_multishard.rs` — belt and suspenders against the +//! diskfull guard (`/Volumes/Games` itself hovers near the 5% floor). + +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::thread; +use std::time::Duration; + +fn moon_bin() -> String { + std::env::var("MOON_BIN").unwrap_or_else(|_| "./target/release/moon".to_string()) +} + +/// Spawn moon with explicit `--shards` plus arbitrary extra CLI args (mirrors +/// `tests/replication_hardening.rs::start_moon`'s extra-args signature). +/// Always includes `--disk-free-min-pct 0` (repo harness rule) unless the +/// caller already passed that flag. +fn start_moon(port: u16, dir: &str, shards: usize, extra: &[&str]) -> Child { + let port_s = port.to_string(); + let shards_s = shards.to_string(); + let mut full: Vec<&str> = vec![ + "--port", + &port_s, + "--shards", + &shards_s, + "--dir", + dir, + "--disk-free-min-pct", + "0", + ]; + full.extend_from_slice(extra); + Command::new(moon_bin()) + .args(&full) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("Failed to start moon (set MOON_BIN to a built binary)") +} + +/// Send one inline command and return the raw reply text (bulk bodies +/// inlined; `$-1` nil yields an empty string). +fn send_cmd(addr: &str, cmd: &str) -> String { + let Ok(mut stream) = TcpStream::connect(addr) else { + return String::new(); + }; + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + stream + .write_all(format!("{}\r\n", cmd).as_bytes()) + .expect("write"); + stream.flush().ok(); + let mut reader = BufReader::new(&stream); + read_one_reply(&mut reader) +} + +fn read_one_reply(reader: &mut R) -> String { + let mut out = String::new(); + let mut line = String::new(); + loop { + line.clear(); + match reader.read_line(&mut line) { + Ok(0) | Err(_) => break, + Ok(_) => { + let trimmed = line.trim_end_matches("\r\n").trim_end_matches('\n'); + if trimmed.starts_with('+') || trimmed.starts_with('-') || trimmed.starts_with(':') + { + out.push_str(trimmed); + break; + } + if let Some(rest) = trimmed.strip_prefix('$') { + let len: i64 = rest.trim().parse().unwrap_or(-1); + if len < 0 { + break; + } + let mut buf = vec![0u8; (len as usize) + 2]; + if reader.read_exact(&mut buf).is_ok() { + out.push_str(&String::from_utf8_lossy(&buf[..len as usize])); + } + break; + } + // Ignore array/other headers — not needed by these helpers. + } + } + } + out +} + +/// Send a RESP-array command (needed for args that contain spaces/binary, +/// e.g. Lua script bodies) and return the raw wire reply as text. +fn send_resp(addr: &str, parts: &[&str]) -> String { + let Ok(mut stream) = TcpStream::connect(addr) else { + return String::new(); + }; + stream + .set_read_timeout(Some(Duration::from_millis(500))) + .ok(); + let mut out = format!("*{}\r\n", parts.len()).into_bytes(); + for p in parts { + out.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + out.extend_from_slice(p.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + if stream.write_all(&out).is_err() { + return String::new(); + } + stream.flush().ok(); + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + let deadline = std::time::Instant::now() + Duration::from_millis(600); + while std::time::Instant::now() < deadline { + match stream.read(&mut chunk) { + Ok(0) => break, + Ok(n) => buf.extend_from_slice(&chunk[..n]), + Err(_) => { + if !buf.is_empty() { + break; + } + } + } + } + String::from_utf8_lossy(&buf).into_owned() +} + +/// Run a burst of pipelined inline commands on one fresh connection and wait +/// for all replies (order-preserving, no interleave risk). +fn pipeline(addr: &str, cmds: &[String]) { + let Ok(mut stream) = TcpStream::connect(addr) else { + panic!("pipeline: connect failed to {addr}"); + }; + stream.set_read_timeout(Some(Duration::from_secs(20))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + let mut buf = String::new(); + for c in cmds { + buf.push_str(c); + buf.push_str("\r\n"); + } + stream.write_all(buf.as_bytes()).expect("write"); + stream.flush().ok(); + for _ in cmds { + read_one_reply(&mut reader); + } +} + +fn dbsize(addr: &str) -> i64 { + send_cmd(addr, "DBSIZE") + .strip_prefix(':') + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(-1) +} + +/// Pull the body out of a raw `$\r\n\r\n` bulk-string wire reply +/// (as returned by `send_resp`). Falls back to a trimmed copy of the input +/// for non-bulk replies. +fn bulk_body(raw: &str) -> String { + if let Some(rest) = raw.strip_prefix('$') + && let Some(nl) = rest.find("\r\n") + && let Ok(len) = rest[..nl].trim().parse::() + && let Some(body) = rest.get(nl + 2..nl + 2 + len) + { + return body.to_string(); + } + raw.trim().to_string() +} + +fn master_repl_offset(addr: &str) -> i64 { + let info = send_cmd(addr, "INFO replication"); + info.lines() + .find_map(|l| l.strip_prefix("master_repl_offset:")) + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(-1) +} + +fn wait_until bool>(timeout: Duration, f: F) -> bool { + let deadline = std::time::Instant::now() + timeout; + while std::time::Instant::now() < deadline { + if f() { + return true; + } + thread::sleep(Duration::from_millis(100)); + } + false +} + +fn await_ready(addr: &str) { + assert!( + wait_until(Duration::from_secs(15), || send_cmd(addr, "PING") + .starts_with("+PONG")), + "server at {} did not become ready", + addr + ); +} + +fn await_link_up(replica_addr: &str) { + assert!( + wait_until(Duration::from_secs(15), || send_cmd( + replica_addr, + "INFO replication" + ) + .contains("master_link_status:up")), + "replica {} link did not come up", + replica_addr + ); +} + +struct Guard(Vec); +impl Drop for Guard { + fn drop(&mut self) { + for c in &mut self.0 { + let _ = c.kill(); + let _ = c.wait(); + } + } +} + +/// SAFETY: `pid` is a live child PID we spawned ourselves; SIGKILL is always +/// a valid signal to send. Mirrors `tests/replication_hardening.rs`. +fn sigkill(child: &mut Child) { + // SAFETY: see doc comment above. + let ret = unsafe { libc::kill(child.id() as i32, libc::SIGKILL) }; + assert_eq!(ret, 0, "libc::kill failed"); + let _ = child.wait(); +} + +// ============================================================================ +// Scenario 1: eviction_parity — plain-dropped (evicted) keys on the master +// must also disappear on the replica. Today `record_reason_del` doesn't +// exist, so eviction plain-drops never reach the replication plane: the +// replica keeps every key the master ever sent it, even ones the master +// itself has since evicted. +// ============================================================================ + +fn run_eviction_parity(shards: usize, master_port: u16, replica_port: u16) { + let mdir = tempfile::tempdir().expect("mdir"); + let rdir = tempfile::tempdir().expect("rdir"); + // Small whole-instance cap + allkeys-lru: writes past the cap force + // plain-drop eviction of older keys. `--disk-offload disable` per the + // Wave A eviction-emission scope (plain-drops only, no spill/cold tier). + const MAXMEMORY: &str = "262144"; // 256KB + let master = start_moon( + master_port, + mdir.path().to_str().unwrap(), + shards, + &[ + "--maxmemory", + MAXMEMORY, + "--maxmemory-policy", + "allkeys-lru", + "--disk-offload", + "disable", + "--appendonly", + "no", + ], + ); + let replica = start_moon( + replica_port, + rdir.path().to_str().unwrap(), + 1, + &["--appendonly", "no"], + ); + let mut guard = Guard(vec![master, replica]); + let m = format!("127.0.0.1:{}", master_port); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&m); + await_ready(&r); + + // Attach the replica FIRST (before the eviction-triggering writes), so + // every write (and every eviction, once implemented) is observed live. + assert!(send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)).starts_with("+OK")); + await_link_up(&r); + + // ~4x the cap in raw value bytes (200B values), comfortably past 256KB + // so eviction is guaranteed to fire repeatedly, not just once. + const N: usize = 5000; + let value = "v".repeat(200); + for burst in 0..(N / 100) { + let cmds: Vec = (0..100) + .map(|i| format!("SET ev:{} {}", burst * 100 + i, value)) + .collect(); + pipeline(&m, &cmds); + } + + // Let the master settle (writes done, no more evictions in flight). + thread::sleep(Duration::from_millis(500)); + let master_final = dbsize(&m); + assert!( + master_final < N as i64, + "setup invariant broken: master did not evict anything (dbsize={}, wrote {}) — \ + raise write volume or lower --maxmemory", + master_final, + N + ); + + // FEATURE ASSERTION: replica must converge to the SAME (evicted) dbsize, + // not the full write count. + assert!( + wait_until(Duration::from_secs(10), || dbsize(&r) == master_final), + "replica did not converge to master's evicted dbsize: master={} replica={} \ + (wrote {} keys) — eviction plain-drops are not propagated to the replica", + master_final, + dbsize(&r), + N + ); + + // Spot-check: any key missing on the master (evicted) must also be + // missing on the replica. + let mut master_missing = Vec::new(); + for i in 0..N { + if send_cmd(&m, &format!("GET ev:{}", i)).is_empty() { + master_missing.push(i); + if master_missing.len() >= 20 { + break; + } + } + } + assert!( + !master_missing.is_empty(), + "setup invariant broken: could not find any evicted key on the master" + ); + for i in &master_missing { + assert!( + send_cmd(&r, &format!("GET ev:{}", i)).is_empty(), + "key ev:{} was evicted on master but still present on replica", + i + ); + } + + guard.0.clear(); +} + +#[test] +#[ignore] +fn eviction_parity_shards1() { + run_eviction_parity(1, 17301, 17302); +} + +#[test] +#[ignore] +fn eviction_parity_shards4() { + run_eviction_parity(4, 17311, 17312); +} + +// ============================================================================ +// Scenario 2: eval_effects_parity — a Lua script's `redis.call` write +// effects (SET x3 + INCR x5 on distinct keys) must reach an attached +// replica, for both EVAL and EVALSHA. Today EVAL/EVALSHA carry no WRITE +// command-metadata flag, so the generic replication gate never sees them — +// the replica never observes the script's writes at all. +// ============================================================================ + +/// EVAL/EVALSHA do NOT cross-shard forward (`src/server/conn/handler_monoio/ +/// dispatch.rs::try_handle_eval` always runs on `ctx.shard_id`, the shard +/// that accepted the TCP connection): `validate_keys_same_shard` requires +/// every key to hash to THAT shard specifically, not merely to agree with +/// each other. A single connection is pinned to whichever shard SO_REUSEPORT +/// handed it, so a hardcoded hash tag has only a `1/num_shards` chance of +/// matching. Probe on ONE fixed connection (so the shard binding never +/// changes mid-probe) by trying tag suffixes until the real EVAL succeeds — +/// the first non-CROSSSLOT reply IS the real call, so this costs nothing +/// extra on the (common) shards=1 case. +fn eval_find_shard_and_run( + addr: &str, + script: &str, + tag_prefix: &str, + key_suffixes: &[&str], + argv: &[&str], +) -> (String, String) { + let mut stream = TcpStream::connect(addr).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + for attempt in 0..64 { + let tag = format!("{}{}", tag_prefix, attempt); + let keys: Vec = key_suffixes + .iter() + .map(|s| format!("{{{}}}:{}", tag, s)) + .collect(); + let mut parts: Vec<&str> = vec!["EVAL", script]; + let numkeys = keys.len().to_string(); + parts.push(&numkeys); + for k in &keys { + parts.push(k); + } + for a in argv { + parts.push(a); + } + let mut out = format!("*{}\r\n", parts.len()).into_bytes(); + for p in &parts { + out.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + out.extend_from_slice(p.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + stream.write_all(&out).expect("write"); + stream.flush().ok(); + let reply = read_one_reply(&mut reader); + if !reply.contains("CROSSSLOT") { + return (tag, reply); + } + } + panic!( + "eval_find_shard_and_run: no hash tag among 64 tried landed on this connection's shard \ + (tag_prefix={})", + tag_prefix + ); +} + +/// Same shard-matching probe, for EVALSHA (script must already be loaded on +/// every shard — `SCRIPT LOAD` fans out, see `try_handle_script`). +fn evalsha_find_shard_and_run( + addr: &str, + sha: &str, + tag_prefix: &str, + key_suffixes: &[&str], + argv: &[&str], +) -> (String, String) { + let mut stream = TcpStream::connect(addr).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + for attempt in 0..64 { + let tag = format!("{}{}", tag_prefix, attempt); + let keys: Vec = key_suffixes + .iter() + .map(|s| format!("{{{}}}:{}", tag, s)) + .collect(); + let mut parts: Vec<&str> = vec!["EVALSHA", sha]; + let numkeys = keys.len().to_string(); + parts.push(&numkeys); + for k in &keys { + parts.push(k); + } + for a in argv { + parts.push(a); + } + let mut out = format!("*{}\r\n", parts.len()).into_bytes(); + for p in &parts { + out.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + out.extend_from_slice(p.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + stream.write_all(&out).expect("write"); + stream.flush().ok(); + let reply = read_one_reply(&mut reader); + if !reply.contains("CROSSSLOT") { + return (tag, reply); + } + } + panic!( + "evalsha_find_shard_and_run: no hash tag among 64 tried landed on this connection's \ + shard (tag_prefix={})", + tag_prefix + ); +} + +fn run_eval_effects_parity(shards: usize, master_port: u16, replica_port: u16) { + let mdir = tempfile::tempdir().expect("mdir"); + let rdir = tempfile::tempdir().expect("rdir"); + let master = start_moon( + master_port, + mdir.path().to_str().unwrap(), + shards, + &["--appendonly", "no"], + ); + let replica = start_moon( + replica_port, + rdir.path().to_str().unwrap(), + 1, + &["--appendonly", "no"], + ); + let mut guard = Guard(vec![master, replica]); + let m = format!("127.0.0.1:{}", master_port); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&m); + await_ready(&r); + assert!(send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)).starts_with("+OK")); + await_link_up(&r); + + // EVAL variant: 3 SETs + 5 INCRs, 8 distinct keys, single db (db0). + let eval_script = "redis.call('SET', KEYS[1], ARGV[1]) \ + redis.call('SET', KEYS[2], ARGV[2]) \ + redis.call('SET', KEYS[3], ARGV[3]) \ + redis.call('INCR', KEYS[4]) \ + redis.call('INCR', KEYS[5]) \ + redis.call('INCR', KEYS[6]) \ + redis.call('INCR', KEYS[7]) \ + redis.call('INCR', KEYS[8]) \ + return 'OK'"; + // Hash-tagged keys: on a multi-shard master a script's KEYS must all + // route to ONE shard (cross-shard scripts are rejected with CROSSSLOT). + // `eval_find_shard_and_run` probes tag suffixes on ONE fixed connection + // until it lands on that connection's own shard. + let key_suffixes = ["s1", "s2", "s3", "i1", "i2", "i3", "i4", "i5"]; + let (evg, reply) = + eval_find_shard_and_run(&m, eval_script, "evg", &key_suffixes, &["va", "vb", "vc"]); + assert!( + reply.contains("OK"), + "setup: EVAL should succeed, got: {}", + reply + ); + let evg_set_keys: Vec = ["s1", "s2", "s3"] + .iter() + .map(|s| format!("{{{}}}:{}", evg, s)) + .collect(); + let evg_incr_keys: Vec = ["i1", "i2", "i3", "i4", "i5"] + .iter() + .map(|s| format!("{{{}}}:{}", evg, s)) + .collect(); + for k in &evg_set_keys { + assert!( + !send_cmd(&m, &format!("GET {}", k)).is_empty(), + "setup invariant broken: master missing {} after EVAL", + k + ); + } + for k in &evg_incr_keys { + assert_eq!( + send_cmd(&m, &format!("GET {}", k)), + "1", + "setup invariant broken: master {} should be 1 after EVAL INCR", + k + ); + } + + // EVALSHA variant: SCRIPT LOAD then EVALSHA, distinct hash-tagged key + // group so the two sub-scenarios can't mask each other. SCRIPT LOAD fans + // the sha out to every shard (`try_handle_script`), so it can be issued + // on any connection. + let sha_reply = send_resp(&m, &["SCRIPT", "LOAD", eval_script]); + let sha = bulk_body(&sha_reply); + assert_eq!( + sha.len(), + 40, + "setup: SCRIPT LOAD should return a 40-char sha1, got: {}", + sha_reply + ); + let (eshg, reply) = + evalsha_find_shard_and_run(&m, &sha, "eshg", &key_suffixes, &["wa", "wb", "wc"]); + assert!( + reply.contains("OK"), + "setup: EVALSHA should succeed, got: {}", + reply + ); + let eshg_set_keys: Vec = ["s1", "s2", "s3"] + .iter() + .map(|s| format!("{{{}}}:{}", eshg, s)) + .collect(); + let eshg_incr_keys: Vec = ["i1", "i2", "i3", "i4", "i5"] + .iter() + .map(|s| format!("{{{}}}:{}", eshg, s)) + .collect(); + + // FEATURE ASSERTION: replica must converge on every key from BOTH + // sub-scenarios. + let all_set_keys: Vec = evg_set_keys + .iter() + .chain(eshg_set_keys.iter()) + .cloned() + .collect(); + let all_set_vals = ["va", "vb", "vc", "wa", "wb", "wc"]; + let all_incr_keys: Vec = evg_incr_keys + .iter() + .chain(eshg_incr_keys.iter()) + .cloned() + .collect(); + + let converged = wait_until(Duration::from_secs(10), || { + all_set_keys + .iter() + .zip(all_set_vals.iter()) + .all(|(k, v)| send_cmd(&r, &format!("GET {}", k)) == *v) + && all_incr_keys + .iter() + .all(|k| send_cmd(&r, &format!("GET {}", k)) == "1") + }); + if !converged { + let missing_sets: Vec<_> = all_set_keys + .iter() + .filter(|k| send_cmd(&r, &format!("GET {}", k)).is_empty()) + .collect(); + let missing_incrs: Vec<_> = all_incr_keys + .iter() + .filter(|k| send_cmd(&r, &format!("GET {}", k)).is_empty()) + .collect(); + panic!( + "replica never observed the Lua script's write effects — \ + missing SET keys: {:?}, missing INCR keys: {:?} \ + (EVAL/EVALSHA writes are not propagated to the replica)", + missing_sets, missing_incrs + ); + } + + guard.0.clear(); +} + +#[test] +#[ignore] +fn eval_effects_parity_shards1() { + run_eval_effects_parity(1, 17321, 17322); +} + +#[test] +#[ignore] +fn eval_effects_parity_shards4() { + run_eval_effects_parity(4, 17331, 17332); +} + +// ============================================================================ +// Scenario 3: eval_writes_survive_restart — standalone durability bug. No +// replica involved: a Lua script's writes must survive a kill -9 + restart +// against the SAME `--appendonly yes` dir. Today the AOF gate never sees +// EVAL/EVALSHA's inner writes either (same missing-WRITE-flag root cause as +// scenario 2), so they are lost on restart. +// ============================================================================ + +#[test] +#[ignore] +fn eval_writes_survive_restart() { + let (port,) = (17341,); + let dir = tempfile::tempdir().expect("dir"); + let mut master = start_moon( + port, + dir.path().to_str().unwrap(), + 1, + &["--appendonly", "yes"], + ); + let m = format!("127.0.0.1:{}", port); + await_ready(&m); + + let script = "redis.call('SET', KEYS[1], ARGV[1]) \ + redis.call('SET', KEYS[2], ARGV[2]) \ + redis.call('SET', KEYS[3], ARGV[3]) \ + return 'OK'"; + let reply = send_resp( + &m, + &[ + "EVAL", script, "3", "surv:a", "surv:b", "surv:c", "va", "vb", "vc", + ], + ); + assert!( + reply.contains("OK"), + "setup: EVAL should succeed, got: {}", + reply + ); + for (k, v) in [("surv:a", "va"), ("surv:b", "vb"), ("surv:c", "vc")] { + assert_eq!( + send_cmd(&m, &format!("GET {}", k)), + v, + "setup invariant broken: master missing {} before kill", + k + ); + } + + // Give the AOF writer a moment to flush/fsync before the kill (WAL/AOF + // group-commit cadence; not testing the fsync timing itself here). + thread::sleep(Duration::from_millis(500)); + sigkill(&mut master); + + let master2 = start_moon( + port, + dir.path().to_str().unwrap(), + 1, + &["--appendonly", "yes"], + ); + let _guard = Guard(vec![master2]); + await_ready(&m); + + // FEATURE ASSERTION: EVAL-written keys must exist after restart. + for (k, v) in [("surv:a", "va"), ("surv:b", "vb"), ("surv:c", "vc")] { + let got = send_cmd(&m, &format!("GET {}", k)); + assert_eq!( + got, v, + "key {} lost across kill-9 + restart (EVAL write did not reach the AOF plane): \ + got {:?}, want {:?}", + k, got, v + ); + } +} + +// ============================================================================ +// Scenario 4: evicted_keys_stay_dead_after_restart — an evicted key must +// stay evicted after a kill -9 + restart against the SAME `--appendonly yes` +// dir. Today eviction plain-drops never reach the AOF plane either, so the +// AOF replay resurrects every key the eviction sweep ever dropped. +// ============================================================================ + +#[test] +#[ignore] +fn evicted_keys_stay_dead_after_restart() { + let (port,) = (17351,); + let dir = tempfile::tempdir().expect("dir"); + const MAXMEMORY: &str = "262144"; // 256KB + let mut master = start_moon( + port, + dir.path().to_str().unwrap(), + 1, + &[ + "--appendonly", + "yes", + "--maxmemory", + MAXMEMORY, + "--maxmemory-policy", + "allkeys-lru", + "--disk-offload", + "disable", + ], + ); + let m = format!("127.0.0.1:{}", port); + await_ready(&m); + + const N: usize = 5000; + let value = "v".repeat(200); + for burst in 0..(N / 100) { + let cmds: Vec = (0..100) + .map(|i| format!("SET ek:{} {}", burst * 100 + i, value)) + .collect(); + pipeline(&m, &cmds); + } + thread::sleep(Duration::from_millis(500)); + + // Find a confirmed-evicted key (allkeys-lru evicts oldest-touched first, + // so the earliest-written keys are the most likely victims — scan for + // one to avoid hardcoding LRU internals). + let evicted_key = (0..N).find(|i| send_cmd(&m, &format!("GET ek:{}", i)).is_empty()); + let Some(evicted_key) = evicted_key else { + panic!( + "setup invariant broken: master did not evict anything (raise write volume or lower --maxmemory)" + ); + }; + let pre_kill_dbsize = dbsize(&m); + assert!( + pre_kill_dbsize < N as i64, + "setup invariant broken: dbsize {} should be less than {} written keys", + pre_kill_dbsize, + N + ); + + thread::sleep(Duration::from_millis(500)); + sigkill(&mut master); + + let master2 = start_moon( + port, + dir.path().to_str().unwrap(), + 1, + &[ + "--appendonly", + "yes", + "--maxmemory", + MAXMEMORY, + "--maxmemory-policy", + "allkeys-lru", + "--disk-offload", + "disable", + ], + ); + let _guard = Guard(vec![master2]); + await_ready(&m); + + // FEATURE ASSERTION: the evicted key must still be gone. + let got = send_cmd(&m, &format!("GET ek:{}", evicted_key)); + assert!( + got.is_empty(), + "evicted key ek:{} resurrected after kill-9 + restart (got {:?}) — \ + eviction plain-drops are not propagated to the AOF plane", + evicted_key, + got + ); + + // FEATURE ASSERTION: dbsize must not balloon back up past a small slack + // (a handful of keys written in the final unflushed AOF tail before the + // kill are tolerable; a full resurrection of all evicted keys is not). + let post_restart_dbsize = dbsize(&m); + const SLACK: i64 = 20; + assert!( + post_restart_dbsize <= pre_kill_dbsize + SLACK, + "dbsize ballooned after restart: pre-kill={} post-restart={} (slack={}) — \ + evicted keys resurrected from AOF replay", + pre_kill_dbsize, + post_restart_dbsize, + SLACK + ); +} + +// ============================================================================ +// Scenario 5: expiry_del_propagates — active-expiry DELs must be visible on +// the replication plane (proven via master_repl_offset advancing after the +// SETs settle), not just via the replica's own independent TTL sweep +// silently masking the gap (both sides end up empty either way — that alone +// does NOT prove the DEL was replicated). +// ============================================================================ + +#[test] +#[ignore] +fn expiry_del_propagates() { + let (master_port, replica_port) = (17361, 17362); + let mdir = tempfile::tempdir().expect("mdir"); + let rdir = tempfile::tempdir().expect("rdir"); + let master = start_moon( + master_port, + mdir.path().to_str().unwrap(), + 1, + &["--appendonly", "no"], + ); + let replica = start_moon( + replica_port, + rdir.path().to_str().unwrap(), + 1, + &["--appendonly", "no"], + ); + let mut guard = Guard(vec![master, replica]); + let m = format!("127.0.0.1:{}", master_port); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&m); + await_ready(&r); + assert!(send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)).starts_with("+OK")); + await_link_up(&r); + + const N: usize = 50; + let cmds: Vec = (0..N).map(|i| format!("SET xk:{} v PX 200", i)).collect(); + pipeline(&m, &cmds); + assert!( + wait_until(Duration::from_secs(5), || dbsize(&r) == N as i64), + "setup invariant broken: replica did not receive the {} SETs before expiry", + N + ); + assert_eq!(dbsize(&m), N as i64, "setup: master dbsize after SETs"); + + // Baseline offset AFTER the SETs have settled (excludes SET/PEXPIRE + // writes themselves — only the subsequent active-expiry DELs, if any, + // should move the needle from here). + let baseline_offset = master_repl_offset(&m); + + // Let the 200ms TTLs expire (master active-expiry sweep runs at 100ms + // cadence, comfortably inside this 2s window either way). + thread::sleep(Duration::from_secs(2)); + + assert!( + wait_until(Duration::from_secs(5), || dbsize(&m) == 0), + "master did not actively expire all keys: dbsize={}", + dbsize(&m) + ); + assert!( + wait_until(Duration::from_secs(5), || dbsize(&r) == 0), + "replica did not converge to 0 keys (either via replicated DELs or its own \ + independent expiry): dbsize={}", + dbsize(&r) + ); + + // FEATURE ASSERTION: the offset must have moved — proof that the master + // actually EMITTED replication records for the active-expiry DELs, + // rather than relying entirely on the replica's own independent sweep + // (which would leave both sides at dbsize 0 while the offset stayed + // frozen at `baseline_offset`). + let after_offset = master_repl_offset(&m); + assert!( + after_offset > baseline_offset, + "master_repl_offset did not advance after active expiry (baseline={}, after={}) — \ + active-expiry DELs are not emitted to the replication plane \ + (both sides reaching dbsize 0 is being masked by the replica's own independent TTL sweep)", + baseline_offset, + after_offset + ); + + guard.0.clear(); +} + +// ============================================================================ +// Scenario 6: select_in_script_errors — `redis.call('SELECT', ...)` inside a +// script must return a loud error mentioning SELECT, and no write must land +// anywhere. Today SELECT inside a script silently succeeds and the +// subsequent write completes (landing in the ORIGINAL db regardless of the +// SELECT target — a silent cross-db bug that Wave A converts into a loud +// script error instead). +// ============================================================================ + +#[test] +#[ignore] +fn select_in_script_errors() { + let (port,) = (17371,); + let dir = tempfile::tempdir().expect("dir"); + let master = start_moon( + port, + dir.path().to_str().unwrap(), + 1, + &["--appendonly", "no"], + ); + let _guard = Guard(vec![master]); + let m = format!("127.0.0.1:{}", port); + await_ready(&m); + + let script = "redis.call('SELECT', '1') \ + redis.call('SET', KEYS[1], ARGV[1]) \ + return 'done'"; + let reply = send_resp(&m, &["EVAL", script, "1", "sel:key", "sel:val"]); + + // FEATURE ASSERTION: must be a RESP error mentioning SELECT. + assert!( + reply.starts_with('-') && reply.to_uppercase().contains("SELECT"), + "EVAL with redis.call('SELECT', ...) should return a loud error mentioning SELECT, \ + got: {:?}", + reply + ); + + // FEATURE ASSERTION: the SET must never have landed anywhere — the + // script should abort AT the SELECT call, before the SET ever runs. + let in_db0 = send_cmd(&m, "GET sel:key"); + assert!( + in_db0.is_empty(), + "SELECT-then-SET script should abort before the SET executes, but sel:key exists \ + in db0: {:?}", + in_db0 + ); + let mut stream = TcpStream::connect(&m).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + stream.write_all(b"SELECT 1\r\n").unwrap(); + read_one_reply(&mut reader); + stream.write_all(b"GET sel:key\r\n").unwrap(); + let db1_val = read_one_reply(&mut reader); + assert!( + db1_val.is_empty(), + "sel:key should not exist in db1 either: {:?}", + db1_val + ); +} From f543183c78eab28434b294c261548ea7233e15bb Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 22:10:49 +0700 Subject: [PATCH 2/7] feat(replication): dual-plane DEL emission for expiry + eviction plain-drops MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave A part 1 (task #34): a key removed by active TTL expiry or by memory-pressure eviction previously vanished from the master's own keyspace without telling either durability plane. An attached replica kept serving the stale key forever, and a kill -9 + restart against --appendonly yes resurrected it from the AOF replay (the AOF recorded the original SET but never the removal). Add src/replication/reason_del.rs with two call shapes for the two places a reason-based removal happens: - record_reason_del — shard-event-loop context (background active expiry, background eviction tick). Reuses shard::spsc_handler::wal_append_and_fanout verbatim, the same mechanism ShardMessage::SwapDb already uses for a synthetic write with no originating client command: one call does the WAL v3 log, the replication backlog append + offset advance + deferred live fan-out, and the AOF pool append. - record_reason_del_conn — connection-handler context (the monoio inline SET fast path and the generic per-command write-eviction gate). Mirrors handler_monoio::ft::record_local_write_db's mechanics (fused SELECT for multi-shard masters, emit-on-change SELECT tracking for single-shard masters, backlog append + offset advance + deferred ShardMessage::ReplicaLiveFanout push) and adds the AOF leg record_local_write_db omits, since a synthetic eviction DEL has no client command to hang the existing per-command AOF gate off of. Wire both flavors into every genuine plain-drop site, while leaving every hot-to-cold spill/tiering path emitting nothing (a spilled key stays cold-readable, not deleted): - server/expiration.rs: expire_cycle / expire_cycle_direct gain a removed-key sink parameter, invoked only for the whole-key TTL sweep (hash-field TTL reaps are out of scope for this wave). - shard/timers.rs: run_active_expiry and run_eviction thread the shard-loop's WAL/backlog/fanout/AOF handles through to the sink. - shard/persistence_tick.rs: run_eviction_tick and handle_memory_pressure thread the same handles to the sync-spill fallback branch that has no manifest (a true plain-drop); the durable-manifest branch is left calling the original non-reporting function since that victim is never actually deleted. - storage/eviction.rs: evict_one_with_spill gains an on_plain_drop sink, fired only when spill was None for that victim (captured before spill is consumed). try_evict_if_needed_budget, try_evict_if_needed_with_spill_and_total_budget, and try_evict_if_needed_async_spill_with_total_budget each split into a _reporting variant plus a thin no-op-sink wrapper preserving the original public signature, so out-of-scope callers (Lua bridge, tokio-only handlers, storage/db_quota.rs's per-db quota path, shard/spsc_two_db.rs's cross-db COPY destination eviction) need no changes. - server/conn/blocking.rs (try_inline_dispatch) and server/conn/handler_monoio/mod.rs (run_write_eviction_gate): the two connection-handler-context write-eviction gates now emit via record_reason_del_conn. - shard/spsc_handler.rs: spsc_eviction_gate gains the same sink parameter; all 6 cross-shard write call sites wire it to record_reason_del. Also fixes a pre-existing replication gap uncovered while chasing the eviction-parity tests: try_inline_dispatch's monoio SET fast path only ever fed the AOF, never the replication backlog/fan-out at all (an attached replica never saw a lone `SET foo bar` under --disk-offload disable, independent of eviction). can_inline_writes now also requires !fanout_hint_active(), falling back to the fully- replicating generic dispatch path once any replica has ever attached — the same "fall back once the fast path can't do enough" precedent ctx.spill_sender.is_none() already established for disk-offload. Turns 4 of the committed RED tests in tests/replication_planes.rs green: eviction_parity_shards1, eviction_parity_shards4, evicted_keys_stay_dead_after_restart, expiry_del_propagates. The 4 Lua/EVAL scenarios (eval_effects_parity_shards1/4, eval_writes_survive_restart, select_in_script_errors) are untouched and remain red, owned by a follow-up agent in src/scripting/. Verified: cargo test --release --lib (4115 passed) and cargo test --no-default-features --features runtime-tokio,jemalloc --lib (3316 passed) both clean; cargo clippy -D warnings clean on both feature matrices; cargo fmt --check clean; replication_multishard (9/9), replication_streaming (7/7), replication_hardening (5/5), aof_multidb_kill9 (4/4), and oom_bypass_closure (8/8) all still green. Known follow-ups left as documented no-op gaps, out of scope for this wave: hash-field TTL reaps, Lua write effects, storage/db_quota.rs's per-db maxmemory eviction, and spsc_two_db.rs's cross-db COPY destination eviction do not yet emit reason-DEL records. Refs: task #34, .planning/rfcs/plane-replication-design.md (Wave A) author: Tin Dang --- CHANGELOG.md | 37 +++++ src/command/hash/mod.rs | 6 +- src/replication/mod.rs | 1 + src/replication/reason_del.rs | 201 ++++++++++++++++++++++++++ src/server/conn/blocking.rs | 27 +++- src/server/conn/handler_monoio/mod.rs | 39 ++++- src/server/expiration.rs | 34 +++-- src/shard/event_loop.rs | 50 ++++++- src/shard/persistence_tick.rs | 74 +++++++++- src/shard/spsc_handler.rs | 97 ++++++++++++- src/shard/spsc_two_db.rs | 10 ++ src/shard/timers.rs | 121 +++++++++++++++- src/storage/db_quota.rs | 9 +- src/storage/eviction.rs | 92 +++++++++++- 14 files changed, 765 insertions(+), 33 deletions(-) create mode 100644 src/replication/reason_del.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index fb36f40d..53fa59bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,43 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added — Wave A part 1: `record_reason_del` dual-plane DEL emission (task #34) + +Master-side key removals for a reason OTHER than a client write command +(active TTL expiry, `--maxmemory` eviction plain-drops) previously reached +NEITHER the AOF plane nor the replication plane: the key vanished from the +master's own keyspace, but an attached replica kept serving it forever, and +a `kill -9` + restart against `--appendonly yes` resurrected it from the AOF +replay (the AOF never recorded the DEL, only the original SET/write). + +- New `replication::reason_del::record_reason_del` (shard-event-loop + context, reuses `wal_append_and_fanout` — the same mechanism the + `ShardMessage::SwapDb` synthetic-command record already uses) and + `record_reason_del_conn` (connection-handler context, mirrors + `handler_monoio::ft::record_local_write_db`'s backlog/offset/fan-out + mechanics and adds the AOF leg that helper omits). Both emit a real + `DEL ` RESP record, respecting the fused-`SELECT` multi-db framing + every ordinary write already uses. +- Wired into: active expiry's whole-key sweep (`expire_cycle`), background + eviction's plain-drop path (`timers::run_eviction`, + `persistence_tick::handle_memory_pressure`'s no-manifest fallback), the + inline fast-path SET eviction gate, the generic per-command write-eviction + gate, and the SPSC cross-shard write-eviction gate. Every call site is + restricted to `spill.is_none()` plain drops — a spilled/cold-tiered key is + NOT a delete and must never be reported here. +- Fixed a related pre-existing gap found while wiring this: with + `--disk-offload disable`, `server::conn::blocking::try_inline_dispatch` + (the monoio inline SET fast path) fed the AOF but never the replication + backlog/fan-out at all — a plain `SET` on such a master silently never + reached an attached replica. `can_inline_writes` now also gates on + `!replication::state::fanout_hint_active()`, falling back to the generic + dispatch path (which replicates correctly) once any replica has ever + attached. +- Known Wave-A-scoped gaps (documented, not silent): hash-field TTL reaps, + Lua `redis.call` write effects, per-db quota (`--db-maxmemory`) eviction, + and cross-db `COPY ... DB n` destination eviction do not yet emit — + tracked as follow-ups. + ### Fixed — test-harness port-flake sweep (task #18) 33 integration suites that spawn a real `moon` process shared a copy-pasted diff --git a/src/command/hash/mod.rs b/src/command/hash/mod.rs index a0e91cf3..65e785d4 100644 --- a/src/command/hash/mod.rs +++ b/src/command/hash/mod.rs @@ -1093,7 +1093,7 @@ mod tests { let exp = db.now_ms() + 1_000; expire_field_at(&mut db, b"h", b"f", exp); db.set_cached_now_ms_for_test(exp + 1); - expire_cycle_direct(&mut db); + expire_cycle_direct(&mut db, &mut |_| {}); assert_eq!(hget(&mut db, &make_args(&[b"h", b"f"])), Frame::Null); assert_eq!(hlen(&mut db, &make_args(&[b"h"])), Frame::Integer(1)); assert_eq!( @@ -1110,7 +1110,7 @@ mod tests { let exp = db.now_ms() + 1_000; expire_field_at(&mut db, b"h", b"f", exp); db.set_cached_now_ms_for_test(exp + 1); - expire_cycle_direct(&mut db); + expire_cycle_direct(&mut db, &mut |_| {}); // After reaping the only TTL'd field, encoding must downgrade to plain Hash. // HEXPIRE XX on the surviving field must return -2 (no TTL). let r = hexpire( @@ -1128,7 +1128,7 @@ mod tests { let exp = db.now_ms() + 1_000; expire_field_at(&mut db, b"h", b"f", exp); db.set_cached_now_ms_for_test(exp + 1); - expire_cycle_direct(&mut db); + expire_cycle_direct(&mut db, &mut |_| {}); assert_eq!( hgetall(&mut db, &make_args(&[b"h"])), Frame::Array(framevec![]) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 09ba39bf..b6a7e18b 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -4,5 +4,6 @@ pub mod backlog; pub mod graph_sync; pub mod handshake; pub mod master; +pub mod reason_del; pub mod replica; pub mod state; diff --git a/src/replication/reason_del.rs b/src/replication/reason_del.rs new file mode 100644 index 00000000..237701e9 --- /dev/null +++ b/src/replication/reason_del.rs @@ -0,0 +1,201 @@ +//! `record_reason_del` — dual-plane (AOF + replication) `DEL ` emission +//! for master-side key removals that happen for a reason OTHER than a client +//! write command: active TTL expiry and eviction plain-drops (task #34, +//! `.planning/rfcs/plane-replication-design.md`, Wave A). +//! +//! Before this module, three classes of removal reached neither plane: the +//! key vanished from the master's own keyspace, but an attached replica kept +//! serving it forever, and a `kill -9` + restart against `--appendonly yes` +//! resurrected it from the AOF replay (the AOF never recorded the DEL, only +//! the original SET). Both gaps are closed by threading this helper's two +//! call shapes into every genuine plain-drop site: +//! +//! - [`record_reason_del`] — shard-event-loop context (background active +//! expiry, background eviction tick). Reuses +//! [`crate::shard::spsc_handler::wal_append_and_fanout`] verbatim — the +//! exact mechanism the `ShardMessage::SwapDb` synthetic-command record +//! already uses for a write with no originating client command. One call +//! does the WAL v3 log, the replication backlog append + offset advance + +//! deferred live fan-out, AND the AOF pool append (with its own SELECT +//! injection, PR #282), so a background-tick DEL gets bit-for-bit the same +//! durability and replica-delivery guarantees as an ordinary write. +//! +//! - [`record_reason_del_conn`] — connection-handler context (the inline +//! fast-path SET eviction gate and the generic per-command write-eviction +//! gate). These run off a per-connection monoio task, not the shard event +//! loop, so they only have `Option>>` (not +//! the shard loop's pre-extracted `SharedBacklog`/`OffsetHandle`/ +//! `Vec`) — mirrors +//! `handler_monoio::ft::record_local_write_db`'s mechanics (backlog +//! append + offset advance under one read-lock + deferred +//! `ShardMessage::ReplicaLiveFanout` self-queue push, same fused-`SELECT` +//! rule for multi-shard masters) and adds the AOF leg that +//! `record_local_write_db` deliberately omits (its callers already ride +//! the generic per-command AOF gate; a synthetic eviction DEL has no +//! client command to hang that off of). +//! +//! ⚠ Both flavors push to `shard::self_msg` (the live fan-out relay) or +//! touch shard-owned state — monoio shard threads only, same restriction as +//! every other producer in `self_msg`. +//! +//! ⚠ Callers MUST only invoke either flavor for a removal where the key is +//! truly gone — a spilled/cold-tiered entry is NOT a delete (see the +//! plain-drop vs. spill distinction in `storage::eviction`); emitting here +//! for a spill would fabricate a phantom DEL that a replica or AOF replay +//! would apply against a key the master can still serve from its cold tier. + +use bytes::Bytes; + +use crate::persistence::aof::{AOF_SPSC_BACKPRESSURE_BOUND, AofWriterPool, serialize_command}; +use crate::protocol::Frame; +#[cfg(feature = "runtime-monoio")] +use crate::replication::state::ReplicationState; + +/// Serialize `DEL ` as a RESP command record — the same wire form +/// `aof::serialize_command` produces for any ordinary client command, so +/// replica apply and AOF replay treat it identically to a real client DEL. +fn serialize_del(key: &[u8]) -> Bytes { + let frame = Frame::Array(crate::framevec![ + Frame::BulkString(Bytes::from_static(b"DEL")), + Frame::BulkString(Bytes::copy_from_slice(key)), + ]); + serialize_command(&frame) +} + +/// Shard-event-loop-context DEL emission (active expiry, background +/// eviction tick). See module docs. +/// +/// `db` is the logical db the removed key lived in — threaded through to +/// `wal_append_and_fanout` so its SELECT-on-db-change bookkeeping (and the +/// AOF writer's own db-scoped SELECT injection) stay correct for multi-db +/// deployments, exactly as every ordinary write already relies on. +#[allow(clippy::too_many_arguments)] +pub(crate) fn record_reason_del( + key: &[u8], + db: usize, + wal_writer: &mut Option, + repl_backlog: &crate::replication::backlog::SharedBacklog, + replica_txs: &mut Vec, + repl_state: &Option, + shard_id: usize, + aof_pool: Option<&std::sync::Arc>, + wal_kv_log: bool, +) { + let serialized = serialize_del(key); + // Fire-and-forget-on-loss like every other background-tick record (the + // client that originally SET the key is long gone; there is no response + // frame to fail loud through). A dropped AOF append here is already + // counted + `error!`-logged inside `send_append_bounded_blocking`. + let mut aof_budget = AOF_SPSC_BACKPRESSURE_BOUND; + let _ = crate::shard::spsc_handler::wal_append_and_fanout( + &serialized, + db, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + &mut aof_budget, + ); +} + +/// Connection-handler-context DEL emission (inline fast-path SET eviction +/// gate, generic per-command write-eviction gate). See module docs. +/// +/// Monoio-only: both call sites (`server::conn::blocking::try_inline_dispatch`, +/// `server::conn::handler_monoio::run_write_eviction_gate`) are +/// `#[cfg(feature = "runtime-monoio")]`-gated — master-side PSYNC is +/// monoio-only (CLAUDE.md), so a tokio-runtime build never needs this leg. +#[cfg(feature = "runtime-monoio")] +#[allow(clippy::too_many_arguments)] +pub(crate) fn record_reason_del_conn( + repl_state: &Option>>, + shard_id: usize, + num_shards: usize, + aof_pool: Option<&std::sync::Arc>, + db: usize, + key: &[u8], +) { + let del_bytes = serialize_del(key); + // Cheap first gate (one Relaxed load): skip the replication leg entirely + // until a replica has ever begun attaching — mirrors + // `handler_monoio::ft::replication_fanout_active`'s first check. + if crate::replication::state::fanout_hint_active() { + push_record_db(repl_state, shard_id, num_shards, db, del_bytes.clone()); + } + if let Some(pool) = aof_pool { + let mut budget = AOF_SPSC_BACKPRESSURE_BOUND; + let _ = pool.send_append_bounded_blocking(shard_id, 0, db, del_bytes, &mut budget); + } +} + +/// Db-aware record push — mirrors `handler_monoio::ft::record_local_write_db` +/// exactly (fused `SELECT` prefix for multi-shard masters, emit-on-change +/// `SELECT` tracking for single-shard masters), parameterized on a raw +/// `repl_state` handle instead of `&ConnectionContext` so it is usable from +/// call sites that don't carry a full connection context (the inline +/// dispatch fast path only threads the individual fields it needs). +#[cfg(feature = "runtime-monoio")] +fn push_record_db( + repl_state: &Option>>, + shard_id: usize, + num_shards: usize, + db: usize, + bytes: Bytes, +) { + if num_shards > 1 { + let select = crate::persistence::aof::serialize_select_record(db); + let mut combined = Vec::with_capacity(select.len() + bytes.len()); + combined.extend_from_slice(&select); + combined.extend_from_slice(&bytes); + push_record(repl_state, shard_id, Bytes::from(combined)); + return; + } + let needs_select = repl_state.as_ref().is_some_and(|rs| { + rs.read().is_ok_and(|g| { + g.stream_db.get(shard_id).is_some_and(|slot| { + if slot.load(std::sync::atomic::Ordering::Relaxed) != db as i64 { + slot.store(db as i64, std::sync::atomic::Ordering::Relaxed); + true + } else { + false + } + }) + }) + }); + if needs_select { + push_record( + repl_state, + shard_id, + crate::persistence::aof::serialize_select_record(db), + ); + } + push_record(repl_state, shard_id, bytes); +} + +/// One backlog-append + offset-advance + deferred-fanout record push — +/// mirrors `handler_monoio::ft::record_local_write` exactly. +#[cfg(feature = "runtime-monoio")] +fn push_record( + repl_state: &Option>>, + shard_id: usize, + bytes: Bytes, +) { + let mut end_offset = u64::MAX; + if let Some(rs) = repl_state.as_ref() { + if let Ok(g) = rs.read() { + if let Some(slot) = g.per_shard_backlogs.get(shard_id) { + if let Some(backlog) = slot.lock().as_mut() { + backlog.append(&bytes); + } + } + end_offset = g.increment_shard_offset(shard_id, bytes.len() as u64); + } + } + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { + bytes, + end_offset, + }); +} diff --git a/src/server/conn/blocking.rs b/src/server/conn/blocking.rs index fafb5153..c5830b80 100644 --- a/src/server/conn/blocking.rs +++ b/src/server/conn/blocking.rs @@ -1382,7 +1382,32 @@ pub(crate) fn try_inline_dispatch( if !crate::storage::eviction::inline_write_can_skip_eviction(est, budget) { let rt = runtime_config.read(); let oom = crate::shard::slice::with_shard_db(selected_db, |db| { - crate::storage::eviction::try_evict_if_needed_budget(db, &rt, budget).is_err() + crate::storage::eviction::try_evict_if_needed_budget_reporting( + db, + &rt, + budget, + // task #34 (Wave A): the inline SET fast path is the + // ONLY write-eviction gate that can plain-drop a key + // with zero AOF/replication visibility today — the + // key vanishes from RAM but the client's `+OK` and any + // attached replica/AOF replay never learn it happened. + // `can_inline_writes` (this path's caller) already + // forces the generic dispatch path once a replica has + // ever attached (`fanout_hint_active`), so this sink's + // replication leg is a cheap no-op in that case; the + // AOF leg still fires unconditionally. + &mut |key| { + crate::replication::reason_del::record_reason_del_conn( + repl_state, + shard_id, + num_shards, + aof_pool.as_ref(), + selected_db, + key, + ); + }, + ) + .is_err() }); drop(rt); if oom { diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index f3f3f80a..6f5b8cfe 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -24,7 +24,7 @@ use crate::protocol::Frame; use crate::shard::dispatch::key_to_shard; use crate::shard::mesh::ChannelMesh; use crate::storage::eviction::{ - try_evict_if_needed_async_spill_budget, try_evict_if_needed_budget, + try_evict_if_needed_async_spill_budget, try_evict_if_needed_budget_reporting, }; use crate::workspace::{strip_workspace_prefix_from_response, workspace_rewrite_args}; @@ -96,7 +96,21 @@ fn run_write_eviction_gate( ctx.spill_file_id.set(fid); res } else { - try_evict_if_needed_budget(db, &rt, budget) + // task #34 (Wave A): plain-drop eviction on the generic per-command + // write path (the local-shard leg of any write when disk-offload is + // off/unwired) — emit a dual-plane DEL for every victim so an + // attached replica and the AOF replay converge with the master's + // own eviction decision. + try_evict_if_needed_budget_reporting(db, &rt, budget, &mut |key| { + crate::replication::reason_del::record_reason_del_conn( + &ctx.repl_state, + ctx.shard_id, + ctx.num_shards, + ctx.aof_pool.as_ref(), + sel_db, + key, + ); + }) }; // WS6 fix (HIGH, adversarial review 2026-07-08): a command that can only // shrink memory (HDEL, SREM, LPOP, ...) must never be REJECTED by either @@ -628,12 +642,31 @@ pub(crate) async fn handle_connection_sharded_monoio< // CLIENT TRACKING invalidation hook, so ANY tracking client in // the process (not just this conn) forces writes through the // normal path. One relaxed atomic load; free when tracking is off. + // + // task #34 (Wave A) fix: `try_inline_dispatch`'s SET fast path + // (`server::conn::blocking`) does NOT feed the replication + // backlog/fan-out at all — only `record_local_write`/ + // `record_local_write_db` (the generic dispatch path) do. + // Before this gate, a master with `--disk-offload disable` and + // an attached replica silently dropped every plain `SET` from + // the replication stream (verified: a lone `SET foo bar` never + // reached the replica). `!fanout_hint_active()` — the same + // cheap, never-cleared-once-true Relaxed load + // `handler_monoio::ft::replication_fanout_active` uses as its + // first gate — forces plain SET onto the generic dispatch path + // for the rest of the process once any replica has ever begun + // attaching, matching the existing `ctx.spill_sender.is_none()` + // precedent of "fall back to the full path when it must do more + // than this fast path knows how to do". GET inlining is + // unaffected (`is_get` in `try_inline_dispatch` doesn't check + // `can_inline_writes`). let can_inline_writes = conn.acl_skip_allowed() && !conn.in_multi && !conn.tracking_state.enabled && !crate::tracking::tracking_active() && !is_replica - && ctx.spill_sender.is_none(); + && ctx.spill_sender.is_none() + && !crate::replication::state::fanout_hint_active(); let inlined = try_inline_dispatch_loop( &mut read_buf, &mut write_buf, diff --git a/src/server/expiration.rs b/src/server/expiration.rs index a2e5df51..8c9b90c3 100644 --- a/src/server/expiration.rs +++ b/src/server/expiration.rs @@ -28,7 +28,12 @@ pub async fn run_active_expiration(db: SharedDatabases, shutdown: CancellationTo _ = interval.tick() => { for lock in db.iter() { let mut guard = lock.write(); - expire_cycle(&mut *guard); + // task #34 (Wave A): tokio's active-expiry sweep does not + // emit `record_reason_del` — master-side PSYNC is + // monoio-only (CLAUDE.md), so a tokio-runtime process is + // never a replication master; a no-op sink preserves + // today's behavior exactly. + expire_cycle(&mut guard, &mut |_| {}); } } _ = shutdown.cancelled() => { @@ -43,7 +48,17 @@ pub async fn run_active_expiration(db: SharedDatabases, shutdown: CancellationTo /// /// Shards call this directly on their owned databases without going through /// the `SharedDatabases` wrapper (no Arc/RwLock needed in shared-nothing mode). -pub fn expire_cycle_direct(db: &mut Database) { +/// +/// `on_removed` fires once per whole-key removal from the probabilistic +/// sweep (sweep 1) — task #34 (Wave A): the shard event loop's +/// `run_active_expiry` uses this to emit a dual-plane `DEL` record for every +/// key the sweep actually deletes, so an attached replica and the AOF replay +/// both observe the master's own expiry decision instead of racing their own +/// independent TTL sweeps. Hash-field TTL reaps (sweep 2) deliberately do +/// NOT fire `on_removed` in Wave A — see the RFC's accepted-gap note +/// (replicas run the identical reaper against the identical TTLs, a bounded +/// divergence documented in CHANGELOG rather than wired up here). +pub fn expire_cycle_direct(db: &mut Database, on_removed: &mut dyn FnMut(&[u8])) { // Fast path: if the DB-level flag latches "no expiring keys", skip the // O(N) `keys_with_expiry()` scan entirely. Discovered by flamegraph: // with 100K TTL-less keys, the per-tick scan was consuming ~26% of @@ -53,7 +68,7 @@ pub fn expire_cycle_direct(db: &mut Database) { if !db.maybe_has_expiring_keys() { return; } - expire_cycle(db); + expire_cycle(db, on_removed); } /// Run one probabilistic expiration cycle on a single database. @@ -71,7 +86,7 @@ pub fn expire_cycle_direct(db: &mut Database) { /// `maybe_has_expiring_keys` is cleared only when **both** sweeps return /// empty, so a database with hash-field TTLs but no whole-key TTLs is not /// incorrectly short-circuited on the next tick. -fn expire_cycle(db: &mut Database) { +fn expire_cycle(db: &mut Database, on_removed: &mut dyn FnMut(&[u8])) { let start = Instant::now(); let budget = Duration::from_millis(1); let mut rng = rand::rng(); @@ -90,6 +105,7 @@ fn expire_cycle(db: &mut Database) { for key in &sampled { if db.is_key_expired(key.as_bytes()) { db.remove(key.as_bytes()); + on_removed(key.as_bytes()); expired_count += 1; } } @@ -158,7 +174,7 @@ mod tests { db.set(key, Entry::new_string(Bytes::from_static(b"v"))); } - expire_cycle(&mut db); + expire_cycle(&mut db, &mut |_| {}); // All expired keys should be removed for i in 0..10 { @@ -190,7 +206,7 @@ mod tests { db.set_string(Bytes::from_static(b"k1"), Bytes::from_static(b"v1")); db.set_string(Bytes::from_static(b"k2"), Bytes::from_static(b"v2")); - expire_cycle(&mut db); + expire_cycle(&mut db, &mut |_| {}); // Nothing should change assert_eq!(db.len(), 2); @@ -199,7 +215,7 @@ mod tests { #[test] fn test_expire_cycle_empty_db() { let mut db = Database::new(); - expire_cycle(&mut db); + expire_cycle(&mut db, &mut |_| {}); assert_eq!(db.len(), 0); } @@ -242,7 +258,7 @@ mod tests { // The hash key still exists; "f" is expired, "g" is live. assert_eq!(db.len(), 1); - expire_cycle(&mut db); + expire_cycle(&mut db, &mut |_| {}); // Key must still exist (g is alive). assert_eq!(db.len(), 1); @@ -257,7 +273,7 @@ mod tests { let mut db = Database::new(); seed_hash_with_expired_field(&mut db, b"h", &[(b"f", b"v")], b"f"); - expire_cycle(&mut db); + expire_cycle(&mut db, &mut |_| {}); // Key must be entirely removed. assert_eq!(db.len(), 0); diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 7a2d0971..284997b3 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -1647,7 +1647,18 @@ impl super::Shard { } // Cooperative active expiry + MQ triggers _ = expiry_interval.0.tick() => { - timers::run_active_expiry(&shard_databases, shard_id); + timers::run_active_expiry( + &shard_databases, shard_id, + &mut wal_writer, &repl_backlog, &mut replica_txs, &repl_offsets, + aof_pool.as_ref(), + match wal_kv_log_mode { + crate::config::WalKvLogMode::On => true, + crate::config::WalKvLogMode::Off => false, + crate::config::WalKvLogMode::Auto => { + !appendonly_enabled || !cdc_registry.is_empty() + } + }, + ); // MQ trigger check: fire debounced triggers timers::fire_pending_mq_triggers( &shard_databases, @@ -1670,6 +1681,15 @@ impl super::Shard { &mut wal_writer, &script_cache_rc, &spill_file_id, + &repl_backlog, &mut replica_txs, &repl_offsets, + aof_pool.as_ref(), + match wal_kv_log_mode { + crate::config::WalKvLogMode::On => true, + crate::config::WalKvLogMode::Off => false, + crate::config::WalKvLogMode::Auto => { + !appendonly_enabled || !cdc_registry.is_empty() + } + }, ); // Reap idle io_uring connections (tokio+io_uring path). @@ -2158,7 +2178,22 @@ impl super::Shard { } // expiry + eviction + MQ triggers: every 100ms (100 ticks) if monoio_tick_counter % 100 == 0 { - timers::run_active_expiry(&shard_databases, shard_id); + timers::run_active_expiry( + &shard_databases, + shard_id, + &mut wal_writer, + &repl_backlog, + &mut replica_txs, + &repl_offsets, + aof_pool.as_ref(), + match wal_kv_log_mode { + crate::config::WalKvLogMode::On => true, + crate::config::WalKvLogMode::Off => false, + crate::config::WalKvLogMode::Auto => { + !appendonly_enabled || !cdc_registry.is_empty() + } + }, + ); persistence_tick::run_eviction_tick( spill_thread.as_ref(), &mut shard_manifest, @@ -2171,6 +2206,17 @@ impl super::Shard { &mut wal_writer, &script_cache_rc, &spill_file_id, + &repl_backlog, + &mut replica_txs, + &repl_offsets, + aof_pool.as_ref(), + match wal_kv_log_mode { + crate::config::WalKvLogMode::On => true, + crate::config::WalKvLogMode::Off => false, + crate::config::WalKvLogMode::Auto => { + !appendonly_enabled || !cdc_registry.is_empty() + } + }, ); // MQ trigger check: fire debounced triggers timers::fire_pending_mq_triggers( diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs index f747f1c0..0e1116c4 100644 --- a/src/shard/persistence_tick.rs +++ b/src/shard/persistence_tick.rs @@ -295,6 +295,14 @@ pub(crate) fn run_eviction_tick( wal_v3_writer: &mut Option, script_cache: &std::rc::Rc>, spill_file_id: &std::rc::Rc>, + // task #34 (Wave A): `record_reason_del` handles for plain-dropped + // eviction victims — threaded straight through to `timers::run_eviction` + // and `handle_memory_pressure`'s sync-spill-fallback branch. + repl_backlog: &crate::replication::backlog::SharedBacklog, + replica_txs: &mut Vec, + repl_state: &Option, + aof_pool: Option<&std::sync::Arc>, + wal_kv_log: bool, ) { if let Some(spill_t) = spill_thread { apply_spill_completions(spill_t, shard_manifest, shard_databases, shard_id); @@ -375,9 +383,24 @@ pub(crate) fn run_eviction_tick( next_file_id, wal_v3_writer, spill_thread, + repl_backlog, + replica_txs, + repl_state, + aof_pool, + wal_kv_log, ); } else { - super::timers::run_eviction(shard_databases, shard_id, runtime_config); + super::timers::run_eviction( + shard_databases, + shard_id, + runtime_config, + wal_v3_writer, + repl_backlog, + replica_txs, + repl_state, + aof_pool, + wal_kv_log, + ); } // Sync file ID back to the shared Cell so connection handlers see it. @@ -540,6 +563,7 @@ pub(crate) fn should_run_pressure_cascade( /// /// Called from eviction timer tick when `disk_offload_enabled` is true and /// `should_run_pressure_cascade()` returns true. +#[allow(clippy::too_many_arguments)] pub(crate) fn handle_memory_pressure( page_cache: &Option, shard_databases: &std::sync::Arc, @@ -550,6 +574,12 @@ pub(crate) fn handle_memory_pressure( next_file_id: &mut u64, wal_v3: &mut Option, spill_thread: Option<&crate::storage::tiered::spill_thread::SpillThread>, + // task #34 (Wave A): see `run_eviction_tick`. + repl_backlog: &crate::replication::backlog::SharedBacklog, + replica_txs: &mut Vec, + repl_state: &Option, + aof_pool: Option<&std::sync::Arc>, + wal_kv_log: bool, ) { // Step 1: PageCache eviction -- evict up to 16 cold frames per tick. // This is the cheapest operation: no disk I/O, just invalidates cached pages. @@ -650,7 +680,7 @@ pub(crate) fn handle_memory_pressure( let sender = spill_t.sender(); for i in 0..db_count { crate::shard::slice::with_shard_db(i, |db| { - let _ = crate::storage::eviction::try_evict_if_needed_async_spill_with_total_budget( + let _ = crate::storage::eviction::try_evict_if_needed_async_spill_with_total_budget_reporting( db, &rt, &sender, @@ -660,6 +690,25 @@ pub(crate) fn handle_memory_pressure( i, budget, shard_manifest.as_mut(), + // task #34 (Wave A): only the no-manifest, + // `--appendonly no` plain-drop fallback + // inside this function ever calls this sink + // (the async-spill and durable-batch + // branches leave a cold/AOF-recoverable + // copy and never invoke it). + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + i, + wal_v3, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ); }); } @@ -675,6 +724,10 @@ pub(crate) fn handle_memory_pressure( manifest, next_file_id, }; + // Durable spill (manifest reachable): the + // victim stays cold-readable, never a plain + // drop — must NOT emit (unchanged, no-op-sink + // call). let _ = crate::storage::eviction::try_evict_if_needed_with_spill_and_total_budget( db, &rt, @@ -683,8 +736,23 @@ pub(crate) fn handle_memory_pressure( budget, ); } else { - let _ = crate::storage::eviction::try_evict_if_needed_with_spill_and_total_budget( + // No manifest reachable: this IS the plain + // -drop path (task #34, Wave A) — emit. + let _ = crate::storage::eviction::try_evict_if_needed_with_spill_and_total_budget_reporting( db, &rt, None, total_mem, budget, + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + i, + wal_v3, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ); } }); diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index bc4e05d4..ca2c8383 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -24,7 +24,7 @@ use crate::runtime::channel; use crate::storage::Database; use crate::storage::entry::CachedClock; use crate::storage::eviction::{ - try_evict_if_needed_async_spill_budget, try_evict_if_needed_budget, + try_evict_if_needed_async_spill_budget, try_evict_if_needed_budget_reporting, }; use crate::storage::tiered::spill_thread::SpillRequest; @@ -46,6 +46,7 @@ use super::shared_databases::ShardDatabases; /// directly, bypassing the connection handlers' write path entirely — without /// this gate a scatter-gather write could grow a remote shard's memory past /// `maxmemory` without limit. +#[allow(clippy::too_many_arguments)] pub(super) fn spsc_eviction_gate( db: &mut Database, db_idx: usize, @@ -55,6 +56,14 @@ pub(super) fn spsc_eviction_gate( spill_sender: Option<&flume::Sender>, spill_file_id: &Rc>, disk_offload_dir: Option<&std::path::Path>, + // task #34 (Wave A): fires once per plain-dropped (non-spill) victim so + // the caller can emit a dual-plane DEL record. Callers build this from + // `record_reason_del` (event-loop-context flavor) — `drain_spsc_shared` + // already has the shard loop's pre-extracted `SharedBacklog`/ + // `OffsetHandle`/`Vec` in scope for its own + // `wal_append_and_fanout` calls right next to each `spsc_eviction_gate` + // call site. + on_plain_drop: &mut dyn FnMut(&[u8]), ) -> Result<(), crate::protocol::Frame> { let rt = runtime_config.read(); let budget = shard_databases.elastic_budget(shard_id); @@ -66,7 +75,7 @@ pub(super) fn spsc_eviction_gate( spill_file_id.set(fid); res } else { - try_evict_if_needed_budget(db, &rt, budget) + try_evict_if_needed_budget_reporting(db, &rt, budget, on_plain_drop) }; global_result?; // WS5b: per-db quota, additive and finer-grained than the whole-instance @@ -663,6 +672,20 @@ pub(crate) fn handle_shard_message_shared( spill_sender, spill_file_id, disk_offload_dir, + // task #34 (Wave A): cross-shard write leg. + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + db_idx, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ) { oom_frame = Some(oom); return; @@ -896,6 +919,20 @@ pub(crate) fn handle_shard_message_shared( spill_sender, spill_file_id, disk_offload_dir, + // task #34 (Wave A): cross-shard write leg. + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + db_idx, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ) { results.push(oom); continue; @@ -1081,6 +1118,20 @@ pub(crate) fn handle_shard_message_shared( spill_sender, spill_file_id, disk_offload_dir, + // task #34 (Wave A): cross-shard write leg. + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + db_idx, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ) { results.push(oom); continue; @@ -1309,6 +1360,20 @@ pub(crate) fn handle_shard_message_shared( spill_sender, spill_file_id, disk_offload_dir, + // task #34 (Wave A): cross-shard write leg. + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + db_idx, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ) { oom_frame = Some(oom); return; @@ -1500,6 +1565,20 @@ pub(crate) fn handle_shard_message_shared( spill_sender, spill_file_id, disk_offload_dir, + // task #34 (Wave A): cross-shard write leg. + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + db_idx, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ) { results.push(oom); continue; @@ -1686,6 +1765,20 @@ pub(crate) fn handle_shard_message_shared( spill_sender, spill_file_id, disk_offload_dir, + // task #34 (Wave A): cross-shard write leg. + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + db_idx, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ) { results.push(oom); continue; diff --git a/src/shard/spsc_two_db.rs b/src/shard/spsc_two_db.rs index 5e5ec374..abc83778 100644 --- a/src/shard/spsc_two_db.rs +++ b/src/shard/spsc_two_db.rs @@ -100,6 +100,15 @@ pub(crate) fn try_two_db_intercept( // the same eviction gate as any other write. Gate on the // DESTINATION db (the one that grows) before copy_core. if evict_active { + // task #34 (Wave A): a plain-dropped victim here (cross-db + // `COPY ... DB n` growing the destination past budget) is + // NOT wired to `record_reason_del` yet — this function + // deliberately does no WAL/AOF/replication I/O ("no + // persistence here" per the module doc), and threading + // those handles through for the comparatively rare + // cross-db COPY path is left as a follow-up alongside the + // other documented Wave-A gaps (db-quota eviction, Lua + // effects). A no-op sink preserves pre-#34 behavior. if let Err(oom) = crate::shard::spsc_handler::spsc_eviction_gate( dst, ca.dst_db, @@ -109,6 +118,7 @@ pub(crate) fn try_two_db_intercept( spill_sender, spill_file_id, disk_offload_dir, + &mut |_| {}, ) { return oom; } diff --git a/src/shard/timers.rs b/src/shard/timers.rs index e8fa3577..5a578ada 100644 --- a/src/shard/timers.rs +++ b/src/shard/timers.rs @@ -14,11 +14,41 @@ use super::shared_databases::ShardDatabases; /// Run cooperative active expiry across all databases. /// Shard 0 also updates the RSS gauge (once per expiry cycle, ~100ms). -pub(crate) fn run_active_expiry(shard_databases: &Arc, shard_id: usize) { +/// +/// task #34 (Wave A): the `record_reason_del` handles are threaded through +/// so every key the sweep actually removes gets a dual-plane (AOF + +/// replication) `DEL` record — mirrors the `ShardMessage::SwapDb` synthetic +/// -command pattern via `crate::replication::reason_del::record_reason_del`. +/// Zero-cost when neither an AOF pool nor a replica/backlog is wired: the +/// helper's own fast-path checks (`wal_fanout_has_work` inside +/// `wal_append_and_fanout`) skip the serialization + fan-out entirely. +#[allow(clippy::too_many_arguments)] +pub(crate) fn run_active_expiry( + shard_databases: &Arc, + shard_id: usize, + wal_writer: &mut Option, + repl_backlog: &crate::replication::backlog::SharedBacklog, + replica_txs: &mut Vec, + repl_state: &Option, + aof_pool: Option<&Arc>, + wal_kv_log: bool, +) { let db_count = shard_databases.db_count(); for i in 0..db_count { crate::shard::slice::with_shard_db(i, |db| { - crate::server::expiration::expire_cycle_direct(db); + crate::server::expiration::expire_cycle_direct(db, &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + i, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }); }); } // Update RSS gauge on shard 0 only, once per second (not every 100ms tick). @@ -37,10 +67,22 @@ pub(crate) fn run_active_expiry(shard_databases: &Arc, shard_id: } /// Run background eviction if maxmemory (or a per-db quota) is configured. +/// +/// task #34 (Wave A): plain-dropped victims (no disk-offload spill in this +/// path — `--disk-offload disable`, or disk-offload's own budget cascade +/// handles the spill-capable case elsewhere) get a dual-plane `DEL` record +/// via the same `record_reason_del` handles `run_active_expiry` uses. +#[allow(clippy::too_many_arguments)] pub(crate) fn run_eviction( shard_databases: &Arc, shard_id: usize, runtime_config: &Arc>, + wal_writer: &mut Option, + repl_backlog: &crate::replication::backlog::SharedBacklog, + replica_txs: &mut Vec, + repl_state: &Option, + aof_pool: Option<&Arc>, + wal_kv_log: bool, ) { let rt = runtime_config.read(); if rt.maxmemory > 0 { @@ -86,8 +128,21 @@ pub(crate) fn run_eviction( for i in 0..db_count { crate::shard::slice::with_shard_db(i, |db| { let before = db.estimated_memory(); - let _ = crate::storage::eviction::try_evict_if_needed_with_spill_and_total_budget( + let _ = crate::storage::eviction::try_evict_if_needed_with_spill_and_total_budget_reporting( db, &rt, None, remaining, budget, + &mut |key| { + crate::replication::reason_del::record_reason_del( + key, + i, + wal_writer, + repl_backlog, + replica_txs, + repl_state, + shard_id, + aof_pool, + wal_kv_log, + ); + }, ); let freed = before.saturating_sub(db.estimated_memory()); remaining = remaining.saturating_sub(freed); @@ -472,6 +527,28 @@ mod tests { use bytes::Bytes; use std::sync::atomic::Ordering; + /// No-replica/no-AOF `record_reason_del` handles for `run_eviction` + /// unit tests below — these tests exercise the eviction budget math + /// only, not task #34's dual-plane emission (covered end-to-end by + /// `tests/replication_planes.rs`). + struct NoOpReasonDelHandles { + wal_writer: Option, + repl_backlog: crate::replication::backlog::SharedBacklog, + replica_txs: Vec, + repl_state: Option, + } + + impl NoOpReasonDelHandles { + fn new() -> Self { + Self { + wal_writer: None, + repl_backlog: std::sync::Arc::new(parking_lot::Mutex::new(None)), + replica_txs: Vec::new(), + repl_state: None, + } + } + } + /// A3 review fix (CRITICAL, both adversarial reviews): the shard-wide /// vector term must gate an AGGREGATE check (Σ all dbs' KV + vector), /// not be re-added to every logical db's independent check. The per-db @@ -518,7 +595,18 @@ mod tests { shared.store_memory_per_shard[0] .vector .store(700 * 1024, Ordering::Relaxed); - run_eviction(&shared, 0, &runtime_config); + let mut h = NoOpReasonDelHandles::new(); + run_eviction( + &shared, + 0, + &runtime_config, + &mut h.wal_writer, + &h.repl_backlog, + &mut h.replica_txs, + &h.repl_state, + None, + false, + ); let len0 = crate::shard::slice::with_shard_db(0, |db| db.len()); let len1 = crate::shard::slice::with_shard_db(1, |db| db.len()); @@ -569,7 +657,18 @@ mod tests { let runtime_config = Arc::new(parking_lot::RwLock::new(rt)); // KV alone (~tens of KB) is far under the 1 MiB budget: no eviction. - run_eviction(&shared, 0, &runtime_config); + let mut h = NoOpReasonDelHandles::new(); + run_eviction( + &shared, + 0, + &runtime_config, + &mut h.wal_writer, + &h.repl_backlog, + &mut h.replica_txs, + &h.repl_state, + None, + false, + ); let before = crate::shard::slice::with_shard_db(0, |db| db.len()); assert_eq!(before, 100, "KV under budget must not evict"); @@ -578,7 +677,17 @@ mod tests { shared.store_memory_per_shard[0] .vector .store(2 * 1024 * 1024, Ordering::Relaxed); - run_eviction(&shared, 0, &runtime_config); + run_eviction( + &shared, + 0, + &runtime_config, + &mut h.wal_writer, + &h.repl_backlog, + &mut h.replica_txs, + &h.repl_state, + None, + false, + ); let after = crate::shard::slice::with_shard_db(0, |db| db.len()); assert!( after < before, diff --git a/src/storage/db_quota.rs b/src/storage/db_quota.rs index dd65fea0..d508584d 100644 --- a/src/storage/db_quota.rs +++ b/src/storage/db_quota.rs @@ -240,7 +240,14 @@ pub fn check_db_maxmemory( let before = db.estimated_memory(); // No disk-offload spill integration for db-quota eviction — see the // module doc's "Known limitation" note. - if !eviction::evict_one_with_spill(db, config, &policy, None) { + // + // task #34 (Wave A): db-quota plain-drops are NOT wired into + // `record_reason_del` yet (a no-op sink here matches pre-#34 + // behavior exactly) — `--db-maxmemory` is a separate, narrower + // knob from the whole-instance `--maxmemory` gates Wave A covers; + // tracked as a follow-up alongside the other documented Wave-A gaps + // (hash-field TTL reaps, Lua script effects). + if !eviction::evict_one_with_spill(db, config, &policy, None, &mut |_| {}) { return Err(db_quota_error(db_index, budget)); } let after = db.estimated_memory(); diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs index 48d5ae55..828130d9 100644 --- a/src/storage/eviction.rs +++ b/src/storage/eviction.rs @@ -323,12 +323,31 @@ pub fn try_evict_if_needed_budget( config: &RuntimeConfig, budget_override: usize, ) -> Result<(), Frame> { - try_evict_if_needed_with_spill_and_total_budget( + try_evict_if_needed_budget_reporting(db, config, budget_override, &mut |_| {}) +} + +/// Like [`try_evict_if_needed_budget`], but reports every plain-dropped +/// (non-spill) victim key to `on_plain_drop` — task #34 (Wave A): +/// connection-handler-context write-eviction gates (inline SET fast path, +/// generic per-command write path, SPSC cross-shard write path) use this to +/// emit a dual-plane DEL record for keys the caller's eviction just deleted +/// with no cold-tier copy. Callers that don't care (Lua bridge, the tokio +/// handlers, unit tests) keep calling [`try_evict_if_needed_budget`], which +/// forwards here with a no-op sink — behavior-identical to before this +/// parameter existed. +pub fn try_evict_if_needed_budget_reporting( + db: &mut Database, + config: &RuntimeConfig, + budget_override: usize, + on_plain_drop: &mut dyn FnMut(&[u8]), +) -> Result<(), Frame> { + try_evict_if_needed_with_spill_and_total_budget_reporting( db, config, None, db.estimated_memory(), budget_override, + on_plain_drop, ) } @@ -369,11 +388,37 @@ pub fn try_evict_if_needed_with_spill_and_total( /// `ShardDatabases::recompute_elastic_budget`) lets a hot shard borrow /// headroom donated by under-budget siblings before evicting. pub fn try_evict_if_needed_with_spill_and_total_budget( + db: &mut Database, + config: &RuntimeConfig, + spill: Option<&mut SpillContext<'_>>, + total_memory: usize, + budget_override: usize, +) -> Result<(), Frame> { + try_evict_if_needed_with_spill_and_total_budget_reporting( + db, + config, + spill, + total_memory, + budget_override, + &mut |_| {}, + ) +} + +/// Like [`try_evict_if_needed_with_spill_and_total_budget`], but reports +/// every plain-dropped (non-spill) victim key to `on_plain_drop` — task #34 +/// (Wave A): background-tick callers (active-expiry-adjacent eviction sweep, +/// memory-pressure-cascade sync-spill fallback) use this to emit a +/// dual-plane DEL record. `on_plain_drop` fires ONLY when `spill` was `None` +/// for that victim — a spilled entry stays cold-readable, not deleted, so it +/// must never be reported here (see `storage::eviction`'s spill-vs-plain-drop +/// distinction, module docs of `replication::reason_del`). +pub fn try_evict_if_needed_with_spill_and_total_budget_reporting( db: &mut Database, config: &RuntimeConfig, mut spill: Option<&mut SpillContext<'_>>, total_memory: usize, budget_override: usize, + on_plain_drop: &mut dyn FnMut(&[u8]), ) -> Result<(), Frame> { if config.maxmemory == 0 { return Ok(()); @@ -397,7 +442,7 @@ pub fn try_evict_if_needed_with_spill_and_total_budget( return Err(oom_error()); } let before = db.estimated_memory(); - if !evict_one_with_spill(db, config, &policy, spill.as_deref_mut()) { + if !evict_one_with_spill(db, config, &policy, spill.as_deref_mut(), on_plain_drop) { return Err(oom_error()); } let after = db.estimated_memory(); @@ -598,6 +643,38 @@ pub fn try_evict_if_needed_async_spill_with_total_budget( db_index: usize, budget_override: usize, manifest: Option<&mut ShardManifest>, +) -> Result<(), Frame> { + try_evict_if_needed_async_spill_with_total_budget_reporting( + db, + config, + sender, + shard_dir, + next_file_id, + total_memory, + db_index, + budget_override, + manifest, + &mut |_| {}, + ) +} + +/// Like [`try_evict_if_needed_async_spill_with_total_budget`], but reports +/// every plain-dropped (non-spill) victim key to `on_plain_drop` — task #34 +/// (Wave A). Only the `config.appendonly != "yes"` + no-manifest fallback +/// branch below ever plain-drops; the durable-batch and async-spill branches +/// never call `on_plain_drop` (both leave a cold-tier/AOF-recoverable copy). +#[allow(clippy::too_many_arguments)] +pub fn try_evict_if_needed_async_spill_with_total_budget_reporting( + db: &mut Database, + config: &RuntimeConfig, + sender: &flume::Sender, + shard_dir: &Path, + next_file_id: &mut u64, + total_memory: usize, + db_index: usize, + budget_override: usize, + manifest: Option<&mut ShardManifest>, + on_plain_drop: &mut dyn FnMut(&[u8]), ) -> Result<(), Frame> { if config.maxmemory == 0 { return Ok(()); @@ -631,7 +708,7 @@ pub fn try_evict_if_needed_async_spill_with_total_budget( return Err(oom_error()); } let before = db.estimated_memory(); - if !evict_one_with_spill(db, config, &policy, None) { + if !evict_one_with_spill(db, config, &policy, None, on_plain_drop) { return Err(oom_error()); } let after = db.estimated_memory(); @@ -1005,7 +1082,13 @@ pub(crate) fn evict_one_with_spill( config: &RuntimeConfig, policy: &EvictionPolicy, spill: Option<&mut SpillContext<'_>>, + on_plain_drop: &mut dyn FnMut(&[u8]), ) -> bool { + // Captured before `spill` is consumed below — determines whether this + // victim (if any) is a true delete (no cold-tier copy, `on_plain_drop` + // fires) or a hot→cold tier transition (spill wrote a durable copy + // first; the key stays cold-readable, never reported). + let is_plain_drop = spill.is_none(); // Find victim key using policy-specific sampling let victim = match policy { EvictionPolicy::NoEviction => None, @@ -1059,6 +1142,9 @@ pub(crate) fn evict_one_with_spill( db.remove(key.as_bytes()); crate::admin::metrics_setup::record_eviction(); + if is_plain_drop { + on_plain_drop(key.as_bytes()); + } true } From a8b7473fcd0a1dbed924f89e69a5502fe06520c2 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 22:58:55 +0700 Subject: [PATCH 3/7] feat(replication): Lua script write-effect replication + SELECT lockout (Wave A part 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part 1 (f543183c) gave background expiry/eviction plain-drops a dual-plane DEL emission path but left Lua scripting untouched. Before this commit, a Lua script's `redis.call`/`redis.pcall` write effects reached NEITHER durability plane: EVAL/EVALSHA deliberately carry no WRITE command-metadata flag, so the generic per-command AOF/replication gate in handler_monoio never saw them. A script's writes vanished on kill -9 + restart against --appendonly yes, and an attached replica never observed a script's writes at all. Separately, redis.call('SELECT', ...) inside a script silently corrupted state: dispatch's SELECT handler mutated only a local variable, so every subsequent write in the script kept landing in the ORIGINAL db while looking like it had switched. scripting::bridge::make_redis_call_fn now records every successfully- executed, WRITE-flagged inner command to both planes itself, immediately after each redis.call/redis.pcall returns — not batched to script end, so a script that writes two keys and then errors on a third still durably records the first two. New replication::reason_del::record_effect_write (sharing a record_bytes_conn core refactored out of part 1's record_reason_del_conn) does the emission, reusing the identical fused- SELECT/backlog/offset/fan-out mechanics every other write path already uses, recording the verbatim cmd+args the script invoked. LuaEvictionCtx (built once per shard at Lua-VM setup, already caching the OOM eviction handles) now also carries num_shards/repl_state/aof_pool for this emission. Every production construction site picked up the three new arguments: the 4 call sites in shard::conn_accept (tokio, monoio, and their migrated-connection twins) plus server::conn::core::ConnectionContext::build_lua_eviction_ctx, which is shared by EVAL/EVALSHA and FCALL/FCALL_RO — both route through the same bridge closure, so FCALL gets the identical effect emission with no extra wiring. redis.call('SELECT', ...) inside any script now fails loud with "ERR SELECT inside scripts is not supported by moon yet" instead of silently corrupting state, intercepted before dispatch so no partial write from the same call ever lands. A real multi-db-scripts feature is a follow-up. Deliberately did NOT flip EVAL/EVALSHA to WRITE in the command metadata table: that would route the literal "EVAL