From 8fe9c76539f4664a16bdafd854c174e9e1524ebd Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 00:32:28 +0700 Subject: [PATCH 1/5] =?UTF-8?q?feat(replication):=20v0.7=20R0=20=E2=80=94?= =?UTF-8?q?=20replica=20applies=20the=20replication=20stream=20end-to-end?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Foundational fix for the v0.7 replication milestone: before this change a REPLICAOF replica completed the PSYNC2 handshake but applied NOTHING — run_handshake_and_stream discarded the FULLRESYNC RDB bulk (logged "received … bytes" only) and stream_commands did buf.clear() after advancing the offset. A freshly attached replica reported DBSIZE 0. The gap was invisible because every replication_hardening.rs test is #[ignore]d and replication_test.rs only covers handshake/INFO/READONLY. What the replica now does (both runtimes, monoio + tokio): - Loads the full-resync RDB snapshot into its local shard (clear + load). - Parses the live RESP command stream incrementally and applies each write on the shard thread via the thread-local ShardSlice — no SPSC self-hop (the ChannelMesh has no self-slot). Tracks SELECT; skips PING/REPLCONF chatter. Bypasses the connection-layer read-only guard by construction, as a replica must. - MOVE and cross-db COPY ... DB n are applied through the same two-db core helpers the master's handler-level intercept uses (generic dispatch cannot apply them; it errors on MOVE and mis-targets COPY..DB). No eviction gate on apply — the replica follows the master. - A replicated command that fails to apply is logged loudly (silent divergence is the failure mode to fear), and a malformed frame drops the connection for a clean resync instead of guessing at resync points. Wire bug fixed: the replica read the diskless FULLRESYNC RDB bulk as len + 2 bytes, assuming a trailing CRLF that diskless replication does not send — stealing the first two bytes of the command stream and desyncing it. Now reads exactly len (capped at 64 GiB). The replication offset advances by bytes CONSUMED by complete frames, not the raw socket read count, so a partial trailing frame never skews the offset. Guards: a multi-shard replica refuses to start replication loudly (R0 topology is --shards 1) rather than diverging silently. Scope: single-shard, logical db 0. The master's live fanout does not yet stream SELECT (multi-db is task #22); per-shard WAIT/ACK (R1) and multi-shard PSYNC (R2) build on this foundation. Tests: new pure stream router replication::apply with 8 unit tests (framing, partial frames, SELECT tracking, chatter skip, fatal parse); new black-box e2e tests/replication_streaming.rs covering snapshot + live SET/DEL + MOVE/COPY-DB with an in-order stream sentinel. Part of v0.7 replication milestone (R0 → R1 → R2). author: Tin Dang --- src/replication/apply.rs | 413 +++++++++++++++++++++++++++++++++ src/replication/mod.rs | 1 + src/replication/replica.rs | 187 +++++++++++---- tests/replication_streaming.rs | 300 ++++++++++++++++++++++++ 4 files changed, 862 insertions(+), 39 deletions(-) create mode 100644 src/replication/apply.rs create mode 100644 tests/replication_streaming.rs diff --git a/src/replication/apply.rs b/src/replication/apply.rs new file mode 100644 index 000000000..d995d48a7 --- /dev/null +++ b/src/replication/apply.rs @@ -0,0 +1,413 @@ +//! Replication stream apply (R0): parse the master's RESP command stream and +//! apply each write to the local shard. +//! +//! The master fans out every write as `aof::serialize_command(cmd)` — a bare +//! RESP Array frame, identical to the AOF wire format (see +//! [`crate::shard::spsc_handler::wal_append_and_fanout`]). The replica +//! connection task runs ON its shard's thread, so for a single-shard deployment +//! every command targets the local shard and is applied directly through the +//! thread-local [`ShardSlice`](crate::shard::slice) via `with_shard` — there is +//! no SPSC self-hop (the ChannelMesh has no self-slot; local legs never go +//! through `spsc_send`). Multi-shard replica routing (hash each key to its +//! owning shard, broadcast keyless commands) is deferred to R2. +//! +//! **Read-only bypass:** the read-only-replica guard lives in the connection +//! layer (`try_enforce_readonly`), which the replica task never invokes. Applying +//! here therefore correctly bypasses it — a replica MUST apply whatever the +//! master streams regardless of its own read-only role. +//! +//! **Durability (R0 scope):** apply is in-memory only. The replica's own AOF is +//! NOT appended per replicated write; a replica recovers by re-syncing from its +//! master on restart (standard, safe). BGREWRITEAOF / RDB snapshots on the +//! replica still fold the applied in-memory state. Independent replica-side +//! incremental persistence is a documented follow-up. + +use std::sync::Arc; + +use bytes::BytesMut; + +use crate::protocol::{Frame, ParseConfig, parse}; + +/// One replicated data command, already resolved to its logical db. +#[derive(Debug)] +pub(crate) struct ReplCommand { + pub db_index: usize, + pub command: Arc, +} + +/// Outcome of draining complete frames out of the replication read buffer. +pub(crate) struct DrainResult { + /// Data commands to apply, in stream order. + pub commands: Vec, + /// Bytes consumed from `buf` — advance the replication offset by exactly + /// this (NOT by the raw socket read count, which may split a frame). + pub consumed: usize, + /// A frame failed to parse. The RESP replication stream is unframed and + /// cannot be safely resynced mid-stream, so the caller must drop the + /// connection; the reconnect path then negotiates a fresh resync. + pub fatal: bool, +} + +/// Drain every COMPLETE RESP command frame from `buf`, tracking `SELECT` into +/// `selected_db`. Any partial trailing frame is left in `buf` for the next read +/// (the parser does not consume incomplete frames), so `consumed` counts only +/// whole frames. +/// +/// `SELECT n` updates `selected_db` and is NOT emitted (carries no data). +/// Replication chatter (`PING`, `REPLCONF`) is skipped. Every other command is +/// emitted bound to the `selected_db` in effect when it was parsed. +pub(crate) fn drain_replicated_commands( + buf: &mut BytesMut, + selected_db: &mut usize, +) -> DrainResult { + let config = ParseConfig::default(); + let mut commands = Vec::new(); + let mut consumed = 0usize; + let mut fatal = false; + + loop { + if buf.is_empty() { + break; + } + let before = buf.len(); + match parse::parse(buf, &config) { + Ok(Some(frame)) => { + consumed += before - buf.len(); + classify(frame, selected_db, &mut commands); + } + // Incomplete trailing frame: parser left `buf` untouched — wait for + // the next socket read to complete it. + Ok(None) => break, + Err(_) => { + fatal = true; + break; + } + } + } + + DrainResult { + commands, + consumed, + fatal, + } +} + +/// Route a single parsed frame: absorb `SELECT`, drop chatter, or record a data +/// command bound to the current `selected_db`. +fn classify(frame: Frame, selected_db: &mut usize, out: &mut Vec) { + let Some((cmd, args)) = command_parts(&frame) else { + return; // non-array / empty — ignore (e.g. inline-newline keepalive) + }; + if cmd.eq_ignore_ascii_case(b"SELECT") { + if let Some(db) = args.first().and_then(frame_to_usize) { + *selected_db = db; + } + return; + } + // Keepalive / ack-negotiation frames the master may interleave — never data. + if cmd.eq_ignore_ascii_case(b"PING") || cmd.eq_ignore_ascii_case(b"REPLCONF") { + return; + } + out.push(ReplCommand { + db_index: *selected_db, + command: Arc::new(frame), + }); +} + +/// Borrow `(command_name, args)` out of a RESP Array frame. +fn command_parts(frame: &Frame) -> Option<(&[u8], &[Frame])> { + match frame { + Frame::Array(arr) if !arr.is_empty() => { + let name = match &arr[0] { + Frame::BulkString(s) => s.as_ref(), + Frame::SimpleString(s) => s.as_ref(), + _ => return None, + }; + Some((name, &arr[1..])) + } + _ => None, + } +} + +fn frame_to_usize(f: &Frame) -> Option { + match f { + Frame::BulkString(s) | Frame::SimpleString(s) => { + std::str::from_utf8(s).ok()?.trim().parse().ok() + } + Frame::Integer(n) if *n >= 0 => Some(*n as usize), + _ => None, + } +} + +/// Apply one replicated command to the local shard's database. +/// +/// Runs synchronously on the shard thread through the thread-local `ShardSlice` +/// (no `.await`), so it cannot interleave with the shard event loop's own +/// `with_shard` access on the same cooperative thread. Bypasses the read-only +/// guard by construction (see module docs). +/// +/// Returns `false` only if this thread has no initialized `ShardSlice` — which +/// would indicate the replica task was spawned off a shard thread (a wiring +/// bug); the caller logs and drops the stream. +pub(crate) fn apply_local(rc: &ReplCommand) -> bool { + use crate::command::{DispatchResult, dispatch as cmd_dispatch}; + use crate::shard::spsc_handler::extract_command_static; + + let Some((cmd, args)) = extract_command_static(&rc.command) else { + return true; // not an array command — nothing to apply (defensive) + }; + crate::shard::slice::try_with_shard(|s| { + let db_count = s.databases.len(); + if db_count == 0 { + return; + } + let db_idx = rc.db_index.min(db_count - 1); + if db_idx != rc.db_index { + // Replica configured with fewer logical dbs than the master: a + // high-index write is clamped rather than lost, but that is a + // divergence — surface it. + tracing::debug!( + "replica apply: db {} clamped to {} ({} dbs on this shard)", + rc.db_index, + db_idx, + db_count + ); + } + + // MOVE / cross-db COPY touch two databases at once and are intercepted + // BEFORE generic dispatch on the master (see `spsc_two_db`). Generic + // `dispatch()` cannot apply them — it returns an error for MOVE and + // silently mis-targets COPY..DB — so mirror the master's two-db + // intercept here. A replica never evicts on apply (it follows the + // master), so the destination-db eviction gate is skipped. + if cmd.eq_ignore_ascii_case(b"MOVE") || cmd.eq_ignore_ascii_case(b"COPY") { + if let Some(resp) = apply_two_db(cmd, args, &mut s.databases, db_idx, db_count) { + warn_on_error(cmd, &resp); + return; + } + // COPY with no DB clause / same-db COPY: fall through to dispatch. + } + + let db = &mut s.databases[db_idx]; + // Replica applies off the shard's periodic clock tick; refresh directly + // so command-relative expiries (EXPIRE/SETEX) compute against real time. + db.refresh_now(); + let mut selected = db_idx; + let resp = match cmd_dispatch(db, cmd, args, &mut selected, db_count) { + DispatchResult::Response(f) | DispatchResult::Quit(f) => f, + }; + warn_on_error(cmd, &resp); + }) + .is_some() +} + +/// A replicated write that fails to apply is a silent-divergence risk — log it +/// loudly instead of dropping it on the floor. (Read-only errors cannot occur +/// here: apply bypasses the connection-layer read-only guard by construction.) +fn warn_on_error(cmd: &[u8], resp: &Frame) { + if let Frame::Error(e) = resp { + tracing::warn!( + "replica apply: {} returned error, replica may diverge from master: {}", + String::from_utf8_lossy(cmd), + String::from_utf8_lossy(e) + ); + } +} + +/// Apply `MOVE` / cross-db `COPY ... DB n` on the replica using the same core +/// helpers as the master's two-db intercept. Returns `None` for a same-db / +/// no-`DB`-clause COPY (caller falls through to generic dispatch), `Some(resp)` +/// otherwise. +fn apply_two_db( + cmd: &[u8], + args: &[Frame], + databases: &mut [crate::storage::Database], + db_idx: usize, + db_count: usize, +) -> Option { + use crate::command::keyspace::move_cmd as ksmv; + + if cmd.eq_ignore_ascii_case(b"MOVE") { + let resp = match ksmv::parse_move_args(args, db_count) { + Err(e) => e, + Ok((_key, dst)) if dst == db_idx => Frame::Integer(0), + Ok((key, dst)) => ksmv::with_two_slice_dbs(databases, db_idx, dst, |src, dstdb| { + src.refresh_now(); + dstdb.refresh_now(); + ksmv::move_core(src, dstdb, &key) + }), + }; + return Some(resp); + } + + // COPY: `?` returns None (fall through to dispatch) for no-DB / same-db. + let copy_result = ksmv::parse_copy_db_args(args, db_idx, db_count)?; + let resp = match copy_result { + Err(e) => e, + Ok(ca) => ksmv::with_two_slice_dbs(databases, db_idx, ca.dst_db, |src, dst| { + src.refresh_now(); + dst.refresh_now(); + ksmv::copy_core(src, dst, &ca.src_key, &ca.dst_key, ca.replace) + }), + }; + Some(resp) +} + +/// Load a full-resync RDB snapshot into the local shard's databases, replacing +/// existing contents (full resync = authoritative master state). +/// +/// Returns the number of keys loaded, or an error if this thread has no +/// `ShardSlice` or the RDB is malformed. +pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { + match crate::shard::slice::try_with_shard(|s| { + for db in s.databases.iter_mut() { + db.clear(); + } + crate::persistence::redis_rdb::load_rdb(&mut s.databases, rdb) + }) { + Some(r) => r, + None => Err(anyhow::anyhow!( + "replica snapshot load: no ShardSlice on this thread" + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + + /// Build the RESP bytes for `SET key val` etc. (bare Array, AOF wire form). + fn resp_cmd(parts: &[&[u8]]) -> Vec { + let mut v = Vec::new(); + v.extend_from_slice(format!("*{}\r\n", parts.len()).as_bytes()); + for p in parts { + v.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + v.extend_from_slice(p); + v.extend_from_slice(b"\r\n"); + } + v + } + + fn cmd_name(rc: &ReplCommand) -> Vec { + match rc.command.as_ref() { + Frame::Array(a) => match &a[0] { + Frame::BulkString(s) => s.to_vec(), + _ => Vec::new(), + }, + _ => Vec::new(), + } + } + + #[test] + fn single_complete_command_consumes_all() { + let bytes = resp_cmd(&[b"SET", b"foo", b"bar"]); + let total = bytes.len(); + let mut buf = BytesMut::from(&bytes[..]); + let mut db = 0usize; + let r = drain_replicated_commands(&mut buf, &mut db); + assert_eq!(r.commands.len(), 1); + assert_eq!(cmd_name(&r.commands[0]), b"SET"); + assert_eq!(r.commands[0].db_index, 0); + assert_eq!(r.consumed, total); + assert!(!r.fatal); + assert!(buf.is_empty(), "whole frame must be consumed"); + } + + #[test] + fn two_back_to_back_commands() { + let mut bytes = resp_cmd(&[b"SET", b"a", b"1"]); + bytes.extend_from_slice(&resp_cmd(&[b"DEL", b"a"])); + let total = bytes.len(); + let mut buf = BytesMut::from(&bytes[..]); + let mut db = 0usize; + let r = drain_replicated_commands(&mut buf, &mut db); + assert_eq!(r.commands.len(), 2); + assert_eq!(cmd_name(&r.commands[0]), b"SET"); + assert_eq!(cmd_name(&r.commands[1]), b"DEL"); + assert_eq!(r.consumed, total); + assert!(buf.is_empty()); + } + + #[test] + fn partial_trailing_frame_is_retained() { + let full = resp_cmd(&[b"SET", b"a", b"1"]); + let complete_len = full.len(); + let mut bytes = full.clone(); + // Append a truncated second frame (header only, body missing). + bytes.extend_from_slice(b"*3\r\n$3\r\nSET\r\n$3\r\nfo"); + let mut buf = BytesMut::from(&bytes[..]); + let mut db = 0usize; + let r = drain_replicated_commands(&mut buf, &mut db); + // Only the first (complete) command is emitted; consumed excludes the + // partial tail, which stays buffered for the next read. + assert_eq!(r.commands.len(), 1); + assert_eq!(r.consumed, complete_len); + assert!(!r.fatal); + assert_eq!(&buf[..], b"*3\r\n$3\r\nSET\r\n$3\r\nfo"); + } + + #[test] + fn select_updates_db_and_is_not_emitted() { + let mut bytes = resp_cmd(&[b"SELECT", b"2"]); + bytes.extend_from_slice(&resp_cmd(&[b"SET", b"k", b"v"])); + let mut buf = BytesMut::from(&bytes[..]); + let mut db = 0usize; + let r = drain_replicated_commands(&mut buf, &mut db); + assert_eq!(r.commands.len(), 1, "SELECT must not be emitted as data"); + assert_eq!(cmd_name(&r.commands[0]), b"SET"); + assert_eq!(r.commands[0].db_index, 2, "SET must bind to selected db 2"); + assert_eq!(db, 2, "selected_db persists across drains"); + } + + #[test] + fn ping_and_replconf_are_skipped() { + let mut bytes = resp_cmd(&[b"PING"]); + bytes.extend_from_slice(&resp_cmd(&[b"REPLCONF", b"GETACK", b"*"])); + bytes.extend_from_slice(&resp_cmd(&[b"SET", b"k", b"v"])); + let mut buf = BytesMut::from(&bytes[..]); + let mut db = 0usize; + let r = drain_replicated_commands(&mut buf, &mut db); + assert_eq!(r.commands.len(), 1); + assert_eq!(cmd_name(&r.commands[0]), b"SET"); + assert!(buf.is_empty()); + } + + #[test] + fn malformed_frame_is_fatal() { + // A bulk-string length that cannot parse → hard parse error. + let bytes = b"*1\r\n$-5\r\nX\r\n".to_vec(); + let mut buf = BytesMut::from(&bytes[..]); + let mut db = 0usize; + let r = drain_replicated_commands(&mut buf, &mut db); + assert!(r.fatal, "unparseable frame must flag fatal for reconnect"); + } + + #[test] + fn empty_buffer_is_noop() { + let mut buf = BytesMut::new(); + let mut db = 3usize; + let r = drain_replicated_commands(&mut buf, &mut db); + assert!(r.commands.is_empty()); + assert_eq!(r.consumed, 0); + assert!(!r.fatal); + assert_eq!(db, 3); + } + + #[test] + fn integer_select_arg_parses() { + // SELECT sent with an integer arg instead of bulk string. + let frame = Frame::Array( + vec![ + Frame::BulkString(Bytes::from_static(b"SELECT")), + Frame::Integer(4), + ] + .into(), + ); + let mut db = 0usize; + let mut out = Vec::new(); + classify(frame, &mut db, &mut out); + assert_eq!(db, 4); + assert!(out.is_empty()); + } +} diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 65fbb9e79..bc63f2903 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -1,3 +1,4 @@ +pub mod apply; pub mod backlog; pub mod handshake; pub mod master; diff --git a/src/replication/replica.rs b/src/replication/replica.rs index 843262e10..157bfa3ce 100644 --- a/src/replication/replica.rs +++ b/src/replication/replica.rs @@ -35,6 +35,20 @@ pub struct ReplicaTaskConfig { #[cfg(feature = "runtime-tokio")] pub async fn run_replica_task(cfg: ReplicaTaskConfig) { let addr = format!("{}:{}", cfg.master_host, cfg.master_port); + // R0 streaming replication is single-shard only. A multi-shard replica would + // misread the master's single diskless RDB bulk and mis-route the command + // stream (see `apply::load_snapshot`, which is thread-local and clears all + // dbs per call), silently diverging. Refuse loudly rather than loop forever + // on a broken sync. Multi-shard replica apply is tracked for R2. + if cfg.num_shards != 1 { + tracing::error!( + "Replica: streaming replication currently supports single-shard only \ + (--shards 1); this node has {} shards. Not starting replication.", + cfg.num_shards + ); + return; + } + let mut backoff_ms = 500u64; const MAX_BACKOFF_MS: u64 = 30_000; @@ -189,15 +203,30 @@ async fn run_handshake_and_stream( } } - // Read N per-shard RDB bulk strings ($\r\n\r\n) - // Simplified for Phase 19: read and discard RDB bytes (snapshot load via ShardMessage in Plan 04) + // Load the full-resync RDB snapshot into the local shard's databases. + // + // R0 targets single-shard replication: `num_shards == 1`, so the master + // (`handle_psync_inline_single_shard`) sends exactly one diskless RDB + // bulk and we load it into this thread's ShardSlice. `load_snapshot` + // clears existing state first (full resync = authoritative). Multi-shard + // replicas (merged-RDB load) are R2. for shard_id in 0..cfg.num_shards { - let rdb_bytes = read_bulk_string(&mut stream).await?; - info!( - "Replica: received shard {} RDB snapshot ({} bytes)", - shard_id, - rdb_bytes.len() - ); + let rdb_bytes = read_rdb_bulk(&mut stream).await?; + match crate::replication::apply::load_snapshot(&rdb_bytes) { + Ok(keys) => info!( + "Replica: loaded shard {} RDB snapshot ({} bytes, {} keys)", + shard_id, + rdb_bytes.len(), + keys + ), + Err(e) => { + warn!( + "Replica: failed to load shard {} RDB snapshot: {}", + shard_id, e + ); + return Err(e); + } + } } // Enter streaming mode @@ -229,6 +258,7 @@ async fn run_handshake_and_stream( #[cfg(feature = "runtime-tokio")] async fn stream_commands(stream: &mut TcpStream, cfg: &ReplicaTaskConfig) -> anyhow::Result<()> { let mut buf = BytesMut::with_capacity(65536); + let mut selected_db = 0usize; loop { let n = stream.read_buf(&mut buf).await?; @@ -236,15 +266,29 @@ async fn stream_commands(stream: &mut TcpStream, cfg: &ReplicaTaskConfig) -> any return Err(anyhow::anyhow!("Master closed connection")); } - // Update local replication offset - if let Ok(rs) = cfg.repl_state.read() { - rs.master_repl_offset.fetch_add(n as u64, Ordering::Relaxed); + // Parse every complete RESP command in the buffer and apply it to the + // local shard. The replication offset advances by CONSUMED bytes (whole + // frames), never the raw socket read count — a read may split a frame. + let outcome = + crate::replication::apply::drain_replicated_commands(&mut buf, &mut selected_db); + for rc in &outcome.commands { + if !crate::replication::apply::apply_local(rc) { + return Err(anyhow::anyhow!( + "replica has no ShardSlice on this thread — cannot apply replication stream" + )); + } + } + if outcome.consumed > 0 { + if let Ok(rs) = cfg.repl_state.read() { + rs.master_repl_offset + .fetch_add(outcome.consumed as u64, Ordering::Relaxed); + } + } + if outcome.fatal { + return Err(anyhow::anyhow!( + "replication stream parse error — dropping connection to force resync" + )); } - - // Parse and dispatch commands from the received bytes - // Full RESP parsing and per-key dispatch routing is added in Plan 04. - // For now, buffer is accumulated and will be wired in Plan 04. - buf.clear(); } } @@ -257,6 +301,20 @@ pub async fn run_replica_task(cfg: ReplicaTaskConfig) { let addr: std::net::SocketAddr = format!("{}:{}", cfg.master_host, cfg.master_port) .parse() .expect("invalid master address"); + // R0 streaming replication is single-shard only. A multi-shard replica would + // misread the master's single diskless RDB bulk and mis-route the command + // stream (see `apply::load_snapshot`, which is thread-local and clears all + // dbs per call), silently diverging. Refuse loudly rather than loop forever + // on a broken sync. Multi-shard replica apply is tracked for R2. + if cfg.num_shards != 1 { + tracing::error!( + "Replica: streaming replication currently supports single-shard only \ + (--shards 1); this node has {} shards. Not starting replication.", + cfg.num_shards + ); + return; + } + let mut backoff_ms = 500u64; const MAX_BACKOFF_MS: u64 = 30_000; @@ -410,14 +468,27 @@ async fn run_handshake_and_stream( } } - // Read N per-shard RDB bulk strings + // Load the full-resync RDB snapshot into the local shard's databases. + // R0 = single-shard: the master sends one diskless RDB bulk, loaded into + // this thread's ShardSlice (clears existing state first — full resync is + // authoritative). Multi-shard merged-RDB load is R2. for shard_id in 0..cfg.num_shards { - let rdb_bytes = read_bulk_string(&mut stream).await?; - info!( - "Replica: received shard {} RDB snapshot ({} bytes)", - shard_id, - rdb_bytes.len() - ); + let rdb_bytes = read_rdb_bulk(&mut stream).await?; + match crate::replication::apply::load_snapshot(&rdb_bytes) { + Ok(keys) => info!( + "Replica: loaded shard {} RDB snapshot ({} bytes, {} keys)", + shard_id, + rdb_bytes.len(), + keys + ), + Err(e) => { + warn!( + "Replica: failed to load shard {} RDB snapshot: {}", + shard_id, e + ); + return Err(e); + } + } } if let Ok(mut rs) = cfg.repl_state.write() { @@ -449,6 +520,7 @@ async fn stream_commands( use monoio::io::AsyncReadRent; let mut buf = BytesMut::with_capacity(65536); + let mut selected_db = 0usize; loop { let tmp = vec![0u8; 65536]; @@ -459,11 +531,29 @@ async fn stream_commands( } buf.extend_from_slice(&tmp[..n]); - if let Ok(rs) = cfg.repl_state.read() { - rs.master_repl_offset.fetch_add(n as u64, Ordering::Relaxed); + // Parse every complete RESP command in the buffer and apply it to the + // local shard. Offset advances by CONSUMED bytes (whole frames), never + // the raw read count — a read may split a frame across boundaries. + let outcome = + crate::replication::apply::drain_replicated_commands(&mut buf, &mut selected_db); + for rc in &outcome.commands { + if !crate::replication::apply::apply_local(rc) { + return Err(anyhow::anyhow!( + "replica has no ShardSlice on this thread — cannot apply replication stream" + )); + } + } + if outcome.consumed > 0 { + if let Ok(rs) = cfg.repl_state.read() { + rs.master_repl_offset + .fetch_add(outcome.consumed as u64, Ordering::Relaxed); + } + } + if outcome.fatal { + return Err(anyhow::anyhow!( + "replication stream parse error — dropping connection to force resync" + )); } - - buf.clear(); } } @@ -495,9 +585,16 @@ async fn read_line(stream: &mut monoio::net::TcpStream) -> anyhow::Result\r\n\r\n) from a monoio TcpStream. +/// Read a diskless full-resync RDB payload (`$\r\n`) from a +/// monoio TcpStream. +/// +/// Unlike a normal RESP bulk string, the master's full-resync RDB is NOT +/// terminated with a trailing `\r\n` (diskless wire format — the bytes right +/// after the RDB are the live replication command stream). Reading `len + 2` +/// would steal the first two bytes of that stream and desync every command +/// after it, so this reads EXACTLY `len` bytes and stops. #[cfg(feature = "runtime-monoio")] -async fn read_bulk_string(stream: &mut monoio::net::TcpStream) -> anyhow::Result> { +async fn read_rdb_bulk(stream: &mut monoio::net::TcpStream) -> anyhow::Result> { use monoio::io::AsyncReadRent; let len_line = read_line(stream).await?; @@ -508,19 +605,24 @@ async fn read_bulk_string(stream: &mut monoio::net::TcpStream) -> anyhow::Result .ok() .and_then(|s| s.parse().ok()) .ok_or_else(|| anyhow::anyhow!("Invalid bulk length"))?; - let total = len + 2; // +2 for trailing \r\n - let mut data = Vec::with_capacity(total); - while data.len() < total { - let remaining = total - data.len(); + // Guard against a corrupted/hostile length header driving a huge allocation + // (the normal RESP parser's bulk cap is bypassed here for the diskless + // no-CRLF format). 64 GiB is far above any legitimate snapshot. + const MAX_RDB_BYTES: usize = 64 * 1024 * 1024 * 1024; + if len > MAX_RDB_BYTES { + anyhow::bail!("RDB bulk length {len} exceeds sanity cap {MAX_RDB_BYTES}"); + } + let mut data = Vec::with_capacity(len); + while data.len() < len { + let remaining = len - data.len(); let tmp = vec![0u8; remaining]; let (result, tmp) = stream.read(tmp).await; let n = result?; if n == 0 { - anyhow::bail!("EOF during read_bulk_string"); + anyhow::bail!("EOF during RDB read"); } data.extend_from_slice(&tmp[..n]); } - data.truncate(len); Ok(data) } @@ -542,9 +644,13 @@ async fn read_line(stream: &mut TcpStream) -> anyhow::Result> { } } -/// Read a RESP bulk string ($\r\n\r\n). +/// Read a diskless full-resync RDB payload (`$\r\n`). +/// +/// The master's full-resync RDB has NO trailing `\r\n` (diskless wire format); +/// reading `len + 2` would steal the first two bytes of the live command stream +/// that follows. Reads EXACTLY `len` bytes. #[cfg(feature = "runtime-tokio")] -async fn read_bulk_string(stream: &mut TcpStream) -> anyhow::Result> { +async fn read_rdb_bulk(stream: &mut TcpStream) -> anyhow::Result> { let len_line = read_line(stream).await?; if len_line.first() != Some(&b'$') { anyhow::bail!("Expected bulk string, got: {:?}", len_line); @@ -553,8 +659,11 @@ async fn read_bulk_string(stream: &mut TcpStream) -> anyhow::Result> { .ok() .and_then(|s| s.parse().ok()) .ok_or_else(|| anyhow::anyhow!("Invalid bulk length"))?; - let mut data = vec![0u8; len + 2]; // +2 for trailing \r\n + const MAX_RDB_BYTES: usize = 64 * 1024 * 1024 * 1024; + if len > MAX_RDB_BYTES { + anyhow::bail!("RDB bulk length {len} exceeds sanity cap {MAX_RDB_BYTES}"); + } + let mut data = vec![0u8; len]; stream.read_exact(&mut data).await?; - data.truncate(len); Ok(data) } diff --git a/tests/replication_streaming.rs b/tests/replication_streaming.rs new file mode 100644 index 000000000..ebff2b667 --- /dev/null +++ b/tests/replication_streaming.rs @@ -0,0 +1,300 @@ +//! R0 acceptance: single-shard streaming replication actually APPLIES the +//! master's snapshot AND its live command stream on the replica. +//! +//! These are black-box tests over two real `moon` processes. They are +//! `#[ignore]`d (like `replication_hardening.rs`) because they need a prebuilt +//! binary; run them explicitly: +//! +//! ```text +//! MOON_BIN=./target/release/moon \ +//! cargo test --test replication_streaming -- --ignored --nocapture +//! ``` +//! +//! Before R0 the replica discarded both the FULLRESYNC snapshot and the live +//! stream (`buf.clear()`), so a freshly-attached replica reported `DBSIZE 0`. +//! These tests lock in the fix. + +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()) +} + +fn start_moon(port: u16, dir: &str) -> Child { + Command::new(moon_bin()) + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--dir", + dir, + "--appendonly", + "no", + // /Volumes/Games hovers near the 5% diskfull guard; disable it so a + // low-free-space dev host does not turn writes into MOONERR diskfull. + "--disk-free-min-pct", + "0", + ]) + .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 (one logical RESP reply). +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); + 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'); + out.push_str(trimmed); + out.push('\n'); + if trimmed.starts_with('+') || trimmed.starts_with('-') || trimmed.starts_with(':') + { + break; + } + if let Some(rest) = trimmed.strip_prefix('$') { + let len: i64 = rest.trim().parse().unwrap_or(-1); + if len < 0 { + break; // $-1 nil + } + 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])); + out.push('\n'); + } + break; + } + } + } + } + out +} + +/// Read exactly one RESP reply from `reader`, returned as text (bulk bodies on +/// their own line; `$-1` nil yields an empty trailing line). +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 for these simple sequences. + } + } + } + out +} + +/// Run a sequence of commands on ONE connection (so `SELECT` persists) and +/// return the LAST reply as text. +fn send_seq(addr: &str, cmds: &[&str]) -> String { + let Ok(mut stream) = TcpStream::connect(addr) else { + return String::new(); + }; + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + for c in cmds { + if stream.write_all(format!("{}\r\n", c).as_bytes()).is_err() { + return String::new(); + } + } + stream.flush().ok(); + let mut reader = BufReader::new(&stream); + let mut last = String::new(); + for _ in 0..cmds.len() { + last = read_one_reply(&mut reader); + } + last +} + +/// GET `key` in logical db `db` (SELECT + GET on one connection). +fn get_in_db(addr: &str, db: usize, key: &str) -> Option { + let v = send_seq(addr, &[&format!("SELECT {}", db), &format!("GET {}", key)]); + let v = v.trim().to_string(); + if v.is_empty() { None } else { Some(v) } +} + +fn dbsize(addr: &str) -> i64 { + let resp = send_cmd(addr, "DBSIZE"); + resp.trim() + .trim_start_matches(':') + .trim() + .parse() + .unwrap_or(-1) +} + +fn get(addr: &str, key: &str) -> Option { + let resp = send_cmd(addr, &format!("GET {}", key)); + let v = resp.lines().nth(1).map(|s| s.to_string()); + v.filter(|s| !s.is_empty()) +} + +fn wait_until bool>(timeout: Duration, f: F) -> bool { + let start = std::time::Instant::now(); + while start.elapsed() < timeout { + if f() { + return true; + } + thread::sleep(Duration::from_millis(100)); + } + f() +} + +struct Killer(Child); +impl Drop for Killer { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// REPL-STREAM-01: replica applies the initial snapshot AND subsequent live +/// writes streamed from the master. +#[test] +#[ignore] +fn replica_applies_snapshot_and_live_stream() { + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + + let master_addr = "127.0.0.1:16700"; + let replica_addr = "127.0.0.1:16701"; + + let _master = Killer(start_moon(16700, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + + // Pre-connect state → must arrive via the FULLRESYNC snapshot. + send_cmd(master_addr, "SET before1 alpha"); + send_cmd(master_addr, "SET before2 beta"); + + let _replica = Killer(start_moon(16701, replica_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + + send_cmd(replica_addr, &format!("REPLICAOF 127.0.0.1 {}", 16700)); + + // Snapshot must land: both pre-connect keys visible on the replica. + let synced = wait_until(Duration::from_secs(10), || dbsize(replica_addr) >= 2); + assert!( + synced, + "replica did not load the FULLRESYNC snapshot (dbsize={})", + dbsize(replica_addr) + ); + assert_eq!(get(replica_addr, "before1").as_deref(), Some("alpha")); + assert_eq!(get(replica_addr, "before2").as_deref(), Some("beta")); + + // Post-connect write → must arrive via the LIVE stream. + send_cmd(master_addr, "SET after1 gamma"); + let streamed = wait_until(Duration::from_secs(10), || { + get(replica_addr, "after1").as_deref() == Some("gamma") + }); + assert!( + streamed, + "replica did not apply the live-streamed write (after1={:?})", + get(replica_addr, "after1") + ); + + // A live DEL must also propagate (delete, not just set). + send_cmd(master_addr, "DEL before1"); + let deleted = wait_until(Duration::from_secs(10), || { + get(replica_addr, "before1").is_none() + }); + assert!(deleted, "replica did not apply the live-streamed DEL"); + + // Confirm the replica reports replica role while still attached. + let info = send_cmd(replica_addr, "INFO replication"); + assert!( + info.contains("role:slave") || info.contains("role:replica"), + "replica INFO should report replica role, got:\n{}", + info + ); + + // MOVE and cross-db COPY are intercepted BEFORE generic dispatch on the + // master (two-db path); the replica must mirror that intercept or they + // silently diverge (generic dispatch errors on MOVE / mis-targets COPY..DB). + // Their destination db is a command ARGUMENT (self-describing), so they + // replicate even in db0-scoped R0. + send_cmd(master_addr, "SET mvkey moved"); + send_cmd(master_addr, "MOVE mvkey 1"); // db0 -> db1 + send_cmd(master_addr, "SET cpkey orig"); + send_cmd(master_addr, "COPY cpkey cpkey2 DB 1"); + send_cmd(master_addr, "SET twodb_done 1"); // in-order stream sentinel (db0) + + // The stream is applied in order, so once the sentinel is visible on the + // replica, MOVE + COPY have already been applied. (A bare is_none() check on + // mvkey would false-pass immediately, since mvkey starts absent.) + let sentinel = wait_until(Duration::from_secs(10), || { + get(replica_addr, "twodb_done").as_deref() == Some("1") + }); + assert!(sentinel, "two-db ordering sentinel never replicated"); + // MOVE's db0 side-effect (key left db0) is observable WITHOUT SELECT — + // SELECT is (wrongly) rejected on a read-only replica, task #23. + assert!( + get(replica_addr, "mvkey").is_none(), + "MOVE did not remove mvkey from replica db0" + ); + // COPY leaves the source key in db0. + assert_eq!(get(replica_addr, "cpkey").as_deref(), Some("orig")); + + // Promote the replica so the read-only guard lifts and SELECT works — then + // confirm both two-db writes landed in db1. + assert!( + send_cmd(replica_addr, "REPLICAOF NO ONE").starts_with("+OK"), + "REPLICAOF NO ONE should succeed" + ); + thread::sleep(Duration::from_millis(300)); + assert_eq!( + get_in_db(replica_addr, 1, "mvkey").as_deref(), + Some("moved"), + "MOVE did not land in replica db1" + ); + assert_eq!( + get_in_db(replica_addr, 1, "cpkey2").as_deref(), + Some("orig"), + "cross-db COPY did not land in replica db1" + ); +} From ebaa7a09ed86cbe83dbf07f75fb69a25ce4ea6b9 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 00:32:55 +0700 Subject: [PATCH 2/5] feat(replication): --repl-backlog-size flag + kill-based hardening teardown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two independent root causes made the replication_hardening suite unable to pass, discovered by running it against the R0 binary with stderr captured and stack sampling (the harness nulls server stderr): 1. --repl-backlog-size did not exist. full_resync_outside_backlog spawns its master with that flag, so the master exited at spawn with a clap error and the test failed its first sync assert against a dead server — a startup failure masquerading as a replication bug. The flag is Redis repl-backlog-size parity: per-shard backlog capacity in raw bytes (like --maxmemory), default 1 MiB, clamped to Redis's 16 KiB floor (the floor also guards ServerConfig's derived Default of 0). ReplicationState.backlog_capacity is now the single source of truth: - ensure_backlogs_allocated() reads it (capacity param removed; the two handshake-path callers previously hardcoded 1 MiB), - ShardMessage::RegisterReplica carries it so the SPSC lazy fallback-init cannot silently diverge from the handshake path, - INFO replication repl_backlog_size reports the real value instead of a hardcoded 1048576. 2. The harness teardown could never complete: it sent SHUTDOWN NOSAVE and blocked on Child::wait(), but moon has no working SHUTDOWN on the production path — the inline parse answers "unknown command", the dispatch arm at command/mod.rs:767 is an error stub, and no connection handler intercepts it. Every test therefore hung forever at cleanup, and a mid-test assert panic leaked live servers that wedged later tests. Teardown is now a kill-on-drop Guard (SIGKILL + wait), which is also panic-safe. Implementing a real SHUTDOWN [NOSAVE|SAVE] is tracked as follow-up work. With both fixes plus R0, the full hardening suite passes for the first time: 5/5 (partial resync, full resync outside backlog, network partition recovery, replica kill-restart parity, replica promotion). Tests: state.rs unit tests for the capacity default/clamp and for eviction kicking in exactly at the configured capacity. Part of v0.7 replication milestone (R0 → R1 → R2). author: Tin Dang --- CHANGELOG.md | 53 +++++++++++ src/config.rs | 8 ++ src/main.rs | 8 +- src/replication/handshake.rs | 5 +- src/replication/master.rs | 30 +++++- src/replication/state.rs | 60 +++++++++++- src/server/conn/handler_monoio/dispatch.rs | 4 +- src/server/embedded.rs | 9 +- src/server/listener.rs | 10 +- src/shard/dispatch.rs | 3 + src/shard/spsc_handler.rs | 12 ++- tests/observability_resident_bytes.rs | 5 +- tests/replication_hardening.rs | 105 ++++++++++----------- 13 files changed, 232 insertions(+), 80 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e05f8ea40..7d3ac1102 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,59 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 post-ACL, so the H-3 deniability guarantee is unchanged). Re-greens all 5 `client_tracking_invalidation` black-box tests on monoio. +### Added — replica now applies the replication stream end-to-end (v0.7 R0) + +- **Foundational fix**: before this change a `REPLICAOF` replica completed the + PSYNC2 handshake but applied **nothing** — `run_handshake_and_stream` + discarded the FULLRESYNC RDB (logged "received … bytes" only) and + `stream_commands` did `buf.clear()` after advancing the offset. A freshly + attached replica reported `DBSIZE 0`. This went unnoticed because every + `replication_hardening.rs` test is `#[ignore]`d and never runs in CI. +- The replica now (1) loads the full-resync RDB snapshot into its local shard + and (2) parses the live RESP command stream and applies each write, tracking + `SELECT`. Both runtime variants (monoio, tokio). Apply runs synchronously on + the shard thread via the thread-local `ShardSlice` (single-shard: every + command is local, no SPSC self-hop), bypassing the connection-layer read-only + guard as a replica must. +- Also fixes a wire bug: the replica read the diskless full-resync RDB bulk as + `len + 2` bytes (assuming a trailing `\r\n` that diskless replication does not + send), stealing the first two bytes of the command stream and desyncing it. + Now reads exactly `len`. The replication offset advances by bytes *consumed* + by complete frames, not the raw socket read count. +- `MOVE` and cross-db `COPY ... DB n` are applied through the same two-db core + helpers the master's handler-level intercept uses (generic dispatch cannot + apply them), so they replicate correctly. A replicated command that still + fails to apply is logged loudly rather than dropped silently. +- New pure, unit-tested stream router (`replication::apply`) and a black-box + streaming acceptance test (`tests/replication_streaming.rs`, covering + snapshot + live SET/DEL/MOVE/COPY). +- **Scope:** single-shard (`--shards 1`), logical **db 0**. A multi-shard + replica refuses to start replication loudly (rather than diverging silently). + Non-default-db writes are a known follow-up — the master's live fanout does + not yet stream `SELECT`, so `SELECT n; SET` replicates into db 0. Per-shard + WAIT/ACK and multi-shard PSYNC build on this in R1/R2. + +### Added — `--repl-backlog-size` (Redis `repl-backlog-size` parity) + +- New flag: per-shard replication backlog capacity in raw bytes (default + 1 MiB, clamped to the 16 KiB Redis floor). Bounds how far a disconnected + replica may fall behind and still partial-resync. Previously the capacity + was hardcoded at three sites (handshake allocation ×2, SPSC lazy fallback) + and `replication_hardening::full_resync_outside_backlog` failed at spawn + with `unexpected argument` — the master process never started. + `ReplicationState.backlog_capacity` is now the single source of truth, + carried into `ShardMessage::RegisterReplica` for the fallback-init and + reported truthfully by `INFO replication` `repl_backlog_size`. + +### Fixed — `replication_hardening` harness could never complete + +- Test teardown used `SHUTDOWN NOSAVE` + `Child::wait()`, but SHUTDOWN is not + implemented on the production path (the dispatch arm is an error stub and no + connection handler intercepts it) — every test hung forever at cleanup and + a mid-test assert failure leaked live servers that wedged later tests' + ports. Teardown is now a kill-on-drop guard (panic-safe). Implementing a + real `SHUTDOWN [NOSAVE|SAVE]` is tracked separately. + ### Fixed — `txn_kv_wiring` integration test port-collision flake - `start_txn_server` picked a port from a throwaway `bind(:0)` probe, dropped diff --git a/src/config.rs b/src/config.rs index 1db28a82e..b19b9b1c9 100644 --- a/src/config.rs +++ b/src/config.rs @@ -86,6 +86,14 @@ pub struct ServerConfig { #[arg(long = "tcp-backlog", default_value_t = 1024)] pub tcp_backlog: i32, + /// Replication backlog capacity in bytes per shard (raw bytes, like + /// --maxmemory). Bounds how far a disconnected replica can fall behind + /// and still partial-resync; older stream bytes are evicted and force a + /// full resync on reconnect. Matches Redis `repl-backlog-size` + /// (default 1 MiB). + #[arg(long = "repl-backlog-size", default_value_t = 1024 * 1024)] + pub repl_backlog_size: usize, + /// Require clients to authenticate with this password #[arg(long)] pub requirepass: Option, diff --git a/src/main.rs b/src/main.rs index 07bf945db..36e683b16 100644 --- a/src/main.rs +++ b/src/main.rs @@ -891,9 +891,11 @@ fn main() -> anyhow::Result<()> { // Create replication state -- load persisted repl_id or generate new one. let (repl_id, repl_id2) = moon::replication::state::load_replication_state(std::path::Path::new(&config.dir)); - let repl_state = std::sync::Arc::new(std::sync::RwLock::new( - moon::replication::state::ReplicationState::new(num_shards, repl_id, repl_id2), - )); + let repl_state = { + let mut rs = moon::replication::state::ReplicationState::new(num_shards, repl_id, repl_id2); + rs.set_backlog_capacity(config.repl_backlog_size); + std::sync::Arc::new(std::sync::RwLock::new(rs)) + }; // Register repl_state globally for INFO command queries. moon::admin::metrics_setup::set_global_repl_state(repl_state.clone()); diff --git a/src/replication/handshake.rs b/src/replication/handshake.rs index 52d0f89a9..434122fcf 100644 --- a/src/replication/handshake.rs +++ b/src/replication/handshake.rs @@ -91,7 +91,10 @@ pub fn build_info_replication(repl_state: &ReplicationState) -> String { s.push_str(&format!("master_repl_offset:{}\r\n", offset)); s.push_str("second_repl_offset:-1\r\n"); s.push_str("repl_backlog_active:1\r\n"); - s.push_str("repl_backlog_size:1048576\r\n"); + s.push_str(&format!( + "repl_backlog_size:{}\r\n", + repl_state.backlog_capacity + )); s.push_str("repl_backlog_first_byte_offset:1\r\n"); s.push_str(&format!("repl_backlog_histlen:{}\r\n", offset)); for (i, replica) in repl_state.replicas.iter().enumerate() { diff --git a/src/replication/master.rs b/src/replication/master.rs index 608741a5f..1afd3c405 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -421,6 +421,12 @@ async fn register_replica_with_shards( // Share the write_half across per-shard sender tasks let write_half = Arc::new(tokio::sync::Mutex::new(write_half)); + // `--repl-backlog-size`, carried in RegisterReplica for the lazy fallback-init. + let backlog_capacity = repl_state + .read() + .map(|g| g.backlog_capacity) + .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); + let channel_capacity = 1024; let mut shard_txs = Vec::with_capacity(num_shards); let mut ack_offsets = Vec::with_capacity(num_shards); @@ -432,7 +438,11 @@ async fn register_replica_with_shards( // Send RegisterReplica to the shard's SPSC if let Some(prod) = shard_producers.get_mut(shard_id) { - let msg = crate::shard::dispatch::ShardMessage::RegisterReplica { replica_id, tx }; + let msg = crate::shard::dispatch::ShardMessage::RegisterReplica { + replica_id, + tx, + backlog_capacity, + }; let _ = prod.try_push(msg); } @@ -498,6 +508,12 @@ async fn register_replica_with_shards( // Single-threaded cooperative scheduling ensures no concurrent borrows. let shared_stream: Rc> = Rc::new(RefCell::new(stream)); + // `--repl-backlog-size`, carried in RegisterReplica for the lazy fallback-init. + let backlog_capacity = repl_state + .read() + .map(|g| g.backlog_capacity) + .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); + let channel_capacity = 1024; let mut shard_txs = Vec::with_capacity(num_shards); let mut ack_offsets = Vec::with_capacity(num_shards); @@ -509,7 +525,11 @@ async fn register_replica_with_shards( // Send RegisterReplica to the shard's SPSC if let Some(prod) = shard_producers.get_mut(shard_id) { - let msg = crate::shard::dispatch::ShardMessage::RegisterReplica { replica_id, tx }; + let msg = crate::shard::dispatch::ShardMessage::RegisterReplica { + replica_id, + tx, + backlog_capacity, + }; let _ = prod.try_push(msg); } @@ -696,11 +716,17 @@ async fn register_replica_inline_single_shard( // tx into its local replica_txs Vec — the sole authority used by // wal_append_and_fanout for live write streaming. { + // `--repl-backlog-size`, carried in RegisterReplica for the lazy fallback-init. + let backlog_capacity = repl_state + .read() + .map(|g| g.backlog_capacity) + .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); let mut prods = dispatch_tx.borrow_mut(); if let Some(prod) = prods.get_mut(0) { let msg = crate::shard::dispatch::ShardMessage::RegisterReplica { replica_id, tx: tx.clone(), + backlog_capacity, }; if prod.try_push(msg).is_err() { anyhow::bail!("failed to push RegisterReplica onto shard 0 SPSC"); diff --git a/src/replication/state.rs b/src/replication/state.rs index f75ec32de..5b65b5487 100644 --- a/src/replication/state.rs +++ b/src/replication/state.rs @@ -7,6 +7,13 @@ pub use crate::replication::handshake::ReplicaHandshakeState; use crate::replication::backlog::SharedBacklog; +/// Default per-shard replication backlog capacity (1 MiB), Redis parity. +pub const DEFAULT_REPL_BACKLOG_SIZE: usize = 1024 * 1024; +/// Smallest accepted backlog capacity. Redis clamps `repl-backlog-size` to +/// 16 KiB; the floor also protects `ServerConfig`'s derived `Default` (0) +/// from allocating an evict-everything backlog. +pub const MIN_REPL_BACKLOG_SIZE: usize = 16 * 1024; + pub struct ReplicationState { pub role: ReplicationRole, /// Primary replication ID (40-char hex). Survives restarts -- persisted to disk. @@ -33,6 +40,11 @@ pub struct ReplicationState { /// `try_enforce_readonly` can avoid taking the surrounding RwLock. /// Invariant: written by `set_role()` whenever `role` changes. pub is_replica_mirror: Arc, + /// Per-shard backlog capacity in bytes (`--repl-backlog-size`), the single + /// source of truth read by every allocation site and by INFO. Set once at + /// startup via [`set_backlog_capacity`](Self::set_backlog_capacity); never + /// resizes already-allocated backlogs. + pub backlog_capacity: usize, } pub enum ReplicationRole { @@ -68,9 +80,18 @@ impl ReplicationState { .map(|_| Arc::new(parking_lot::Mutex::new(None))) .collect(), is_replica_mirror: Arc::new(AtomicBool::new(false)), + backlog_capacity: DEFAULT_REPL_BACKLOG_SIZE, } } + /// Set the per-shard backlog capacity from config, clamped to + /// [`MIN_REPL_BACKLOG_SIZE`] (Redis clamps `repl-backlog-size` the same + /// way). Call before any replica handshake; it does not resize backlogs + /// that were already allocated. + pub fn set_backlog_capacity(&mut self, capacity: usize) { + self.backlog_capacity = capacity.max(MIN_REPL_BACKLOG_SIZE); + } + /// Set `role` and update the lock-free `is_replica_mirror` atomically. /// /// Single owner of the invariant that the mirror tracks `role`. All @@ -87,13 +108,14 @@ impl ReplicationState { /// Allocate the per-shard backlog if not already allocated. Idempotent. /// Called when a replica handshake begins (REPLCONF or PSYNC arrival) so /// subsequent writes on the shard's event loop start being captured for - /// partial resync. Capacity defaults to 1 MiB per shard. - pub fn ensure_backlogs_allocated(&self, capacity: usize) { + /// partial resync. Capacity comes from `self.backlog_capacity` + /// (`--repl-backlog-size`, default 1 MiB per shard). + pub fn ensure_backlogs_allocated(&self) { for slot in &self.per_shard_backlogs { let mut guard = slot.lock(); if guard.is_none() { *guard = Some(crate::replication::backlog::ReplicationBacklog::new( - capacity, + self.backlog_capacity, )); } } @@ -361,6 +383,38 @@ mod tests { assert!(matches!(state.role, ReplicationRole::Master)); } + #[test] + fn test_backlog_capacity_default_and_clamp() { + let mut state = ReplicationState::new(1, generate_repl_id(), ZEROED_ID.to_string()); + assert_eq!(state.backlog_capacity, DEFAULT_REPL_BACKLOG_SIZE); + + // Configured capacity is honored… + state.set_backlog_capacity(64 * 1024); + assert_eq!(state.backlog_capacity, 64 * 1024); + + // …but clamped to the Redis floor (also guards derived-Default 0). + state.set_backlog_capacity(1024); + assert_eq!(state.backlog_capacity, MIN_REPL_BACKLOG_SIZE); + state.set_backlog_capacity(0); + assert_eq!(state.backlog_capacity, MIN_REPL_BACKLOG_SIZE); + } + + #[test] + fn test_ensure_backlogs_use_configured_capacity() { + let mut state = ReplicationState::new(2, generate_repl_id(), ZEROED_ID.to_string()); + state.set_backlog_capacity(MIN_REPL_BACKLOG_SIZE); + state.ensure_backlogs_allocated(); + // Overflow one backlog past the configured capacity: eviction must + // kick in exactly at the configured size, proving the capacity plumb. + let slot = &state.per_shard_backlogs[0]; + let mut guard = slot.lock(); + let backlog = guard.as_mut().expect("allocated"); + let data = vec![0xAAu8; MIN_REPL_BACKLOG_SIZE + 100]; + backlog.append(&data); + assert_eq!(backlog.start_offset(), 100, "evicts past configured cap"); + assert_eq!(backlog.end_offset(), (MIN_REPL_BACKLOG_SIZE + 100) as u64); + } + #[test] fn test_is_replica_mirror_default_false() { let state = ReplicationState::new(1, generate_repl_id(), ZEROED_ID.to_string()); diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index d2e3bdd65..22dcd2ae5 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -511,7 +511,7 @@ pub(super) fn try_handle_replconf( if let Some(ref rs) = ctx.repl_state { if let Ok(g) = rs.read() { if matches!(g.role, crate::replication::state::ReplicationRole::Master) { - g.ensure_backlogs_allocated(1024 * 1024); + g.ensure_backlogs_allocated(); } } } @@ -584,7 +584,7 @@ pub(super) fn try_handle_psync( return None; } if let Some(g) = g { - g.ensure_backlogs_allocated(1024 * 1024); + g.ensure_backlogs_allocated(); } } let repl_id = match &cmd_args[0] { diff --git a/src/server/embedded.rs b/src/server/embedded.rs index c33b93958..e011ce5a8 100644 --- a/src/server/embedded.rs +++ b/src/server/embedded.rs @@ -200,9 +200,12 @@ pub async fn run_embedded( // event loop still expects a populated state object. let (repl_id, repl_id2) = crate::replication::state::load_replication_state(std::path::Path::new(&config.dir)); - let repl_state = Arc::new(std::sync::RwLock::new( - crate::replication::state::ReplicationState::new(num_shards, repl_id, repl_id2), - )); + let repl_state = { + let mut rs = + crate::replication::state::ReplicationState::new(num_shards, repl_id, repl_id2); + rs.set_backlog_capacity(config.repl_backlog_size); + Arc::new(std::sync::RwLock::new(rs)) + }; crate::admin::metrics_setup::set_global_repl_state(repl_state.clone()); // ACL table (loads aclfile if configured; default no-op otherwise). diff --git a/src/server/listener.rs b/src/server/listener.rs index 6bfcdbaec..81c405691 100644 --- a/src/server/listener.rs +++ b/src/server/listener.rs @@ -186,12 +186,14 @@ pub async fn run_with_shutdown( // Create replication state -- load persisted repl_id or generate new one. let (repl_id, repl_id2) = crate::replication::state::load_replication_state(std::path::Path::new(&config.dir)); - let repl_state = Arc::new(RwLock::new( - crate::replication::state::ReplicationState::new( + let repl_state = { + let mut rs = crate::replication::state::ReplicationState::new( 1, // non-sharded mode uses 1 shard repl_id, repl_id2, - ), - )); + ); + rs.set_backlog_capacity(config.repl_backlog_size); + Arc::new(RwLock::new(rs)) + }; // Register repl_state globally for INFO command queries. crate::admin::metrics_setup::set_global_repl_state(repl_state.clone()); diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index e50cbbeb8..c23b3cef6 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -456,9 +456,12 @@ pub enum ShardMessage { /// Register a connected replica's per-shard sender channel with this shard. /// Called once per shard per replica when a new replica connection is established. /// The shard adds `tx` to its replica_txs list for WAL fan-out. + /// `backlog_capacity` (`--repl-backlog-size`) sizes the lazy backlog + /// fallback-init so it can't diverge from the handshake-path allocation. RegisterReplica { replica_id: u64, tx: channel::MpscSender, + backlog_capacity: usize, }, /// Remove a replica's sender channel from this shard's fan-out list. /// Called when a replica disconnects or REPLICAOF NO ONE is executed. diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 51e20bc48..787e73dce 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2472,14 +2472,20 @@ pub(crate) fn handle_shard_message_shared( ShardMessage::Shutdown => { info!("Received shutdown via SPSC"); } - ShardMessage::RegisterReplica { replica_id, tx } => { + ShardMessage::RegisterReplica { + replica_id, + tx, + backlog_capacity, + } => { // Lazy-init replication backlog on first replica registration (saves 1MB/shard). // The backlog is shared with PSYNC handlers via Arc>> on // ReplicationState — see ReplicationState::ensure_backlogs_allocated for the - // earlier allocation point triggered by REPLCONF. + // earlier allocation point triggered by REPLCONF. Capacity is carried + // in the message (from `--repl-backlog-size`) so this fallback can't + // silently diverge from the handshake-path allocation. let mut guard = repl_backlog.lock(); if guard.is_none() { - *guard = Some(ReplicationBacklog::new(1024 * 1024)); + *guard = Some(ReplicationBacklog::new(backlog_capacity)); } drop(guard); replica_txs.push((replica_id, tx)); diff --git a/tests/observability_resident_bytes.rs b/tests/observability_resident_bytes.rs index 92f98bdd5..677c25192 100644 --- a/tests/observability_resident_bytes.rs +++ b/tests/observability_resident_bytes.rs @@ -124,8 +124,9 @@ fn replication_state_backlog_resident_bytes_lazy_init() { "uninitialized backlogs must report 0" ); - // Allocate backlogs (simulates replica handshake). - state.ensure_backlogs_allocated(1024); + // Allocate backlogs (simulates replica handshake); capacity comes from + // state.backlog_capacity (--repl-backlog-size plumb). + state.ensure_backlogs_allocated(); let after = state.backlog_resident_bytes(); assert!( diff --git a/tests/replication_hardening.rs b/tests/replication_hardening.rs index e48b3a0ba..1e6e9f16e 100644 --- a/tests/replication_hardening.rs +++ b/tests/replication_hardening.rs @@ -12,19 +12,38 @@ use std::process::{Command, Stdio}; use std::thread; use std::time::Duration; -fn start_moon(port: u16, dir: &str, extra: &[&str]) -> std::process::Child { - Command::new("./target/release/moon") - .args( - [ - &["--port", &port.to_string(), "--shards", "1", "--dir", dir][..], - extra, - ] - .concat(), - ) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("Failed to start moon") +fn start_moon(port: u16, dir: &str, extra: &[&str]) -> Guard { + Guard( + Command::new("./target/release/moon") + .args( + [ + &["--port", &port.to_string(), "--shards", "1", "--dir", dir][..], + extra, + ] + .concat(), + ) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("Failed to start moon"), + ) +} + +/// Kill-on-drop server guard. +/// +/// Teardown is SIGKILL-based, NOT `SHUTDOWN NOSAVE`: moon has no working +/// SHUTDOWN command on the production path (the dispatch arm is an error stub +/// and the connection handlers never intercept it — see task #27), so a +/// graceful-shutdown cleanup blocks `Child::wait` forever. Kill-on-drop also +/// survives assert panics, so a failed test no longer leaks servers that hold +/// ports and wedge later tests. (Repo harness rule: always kill -9 moon.) +struct Guard(std::process::Child); + +impl Drop for Guard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } } fn send_cmd(addr: &str, cmd: &str) -> String { @@ -102,7 +121,7 @@ mod tests { let master_dir = tempfile::tempdir().unwrap(); let replica_dir = tempfile::tempdir().unwrap(); - let mut master = start_moon(16600, master_dir.path().to_str().unwrap(), &[]); + let _master = start_moon(16600, master_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); // Write initial data @@ -127,15 +146,15 @@ mod tests { // Kill replica // SAFETY: `child.id()` returns a valid PID for a process we just spawned. // SIGKILL is always valid. We check the return code for robustness. - let ret = unsafe { libc::kill(replica.id() as i32, libc::SIGKILL) }; + let ret = unsafe { libc::kill(replica.0.id() as i32, libc::SIGKILL) }; assert_eq!(ret, 0, "libc::kill failed"); - let _ = replica.wait(); + let _ = replica.0.wait(); // Write more data while replica is down (within backlog) write_keys("127.0.0.1:16600", "new", 50); // Restart replica — should partial resync - let mut replica2 = start_moon(16601, replica_dir.path().to_str().unwrap(), &[]); + let _replica2 = start_moon(16601, replica_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); send_cmd("127.0.0.1:16601", "REPLICAOF 127.0.0.1 16600"); thread::sleep(Duration::from_secs(3)); @@ -143,12 +162,6 @@ mod tests { let final_size = dbsize("127.0.0.1:16601"); let master_size = dbsize("127.0.0.1:16600"); - // Cleanup - send_cmd("127.0.0.1:16600", "SHUTDOWN NOSAVE"); - send_cmd("127.0.0.1:16601", "SHUTDOWN NOSAVE"); - let _ = master.wait(); - let _ = replica2.wait(); - assert_eq!( final_size, master_size, "Replica should match master after partial resync: replica={}, master={}", @@ -163,7 +176,7 @@ mod tests { let master_dir = tempfile::tempdir().unwrap(); let replica_dir = tempfile::tempdir().unwrap(); - let mut master = start_moon(16610, master_dir.path().to_str().unwrap(), &[]); + let _master = start_moon(16610, master_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); write_keys("127.0.0.1:16610", "kill_test", 200); @@ -176,15 +189,15 @@ mod tests { // Kill replica with SIGKILL // SAFETY: `child.id()` returns a valid PID for a process we just spawned. // SIGKILL is always valid. We check the return code for robustness. - let ret = unsafe { libc::kill(replica.id() as i32, libc::SIGKILL) }; + let ret = unsafe { libc::kill(replica.0.id() as i32, libc::SIGKILL) }; assert_eq!(ret, 0, "libc::kill failed"); - let _ = replica.wait(); + let _ = replica.0.wait(); // Write more data write_keys("127.0.0.1:16610", "post_kill", 100); // Restart replica - let mut replica2 = start_moon(16611, replica_dir.path().to_str().unwrap(), &[]); + let _replica2 = start_moon(16611, replica_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); send_cmd("127.0.0.1:16611", "REPLICAOF 127.0.0.1 16610"); thread::sleep(Duration::from_secs(3)); @@ -192,11 +205,6 @@ mod tests { let master_size = dbsize("127.0.0.1:16610"); let replica_size = dbsize("127.0.0.1:16611"); - send_cmd("127.0.0.1:16610", "SHUTDOWN NOSAVE"); - send_cmd("127.0.0.1:16611", "SHUTDOWN NOSAVE"); - let _ = master.wait(); - let _ = replica2.wait(); - assert_eq!( replica_size, master_size, "Replica should match master after kill-restart: replica={}, master={}", @@ -211,12 +219,12 @@ mod tests { let master_dir = tempfile::tempdir().unwrap(); let replica_dir = tempfile::tempdir().unwrap(); - let mut master = start_moon(16620, master_dir.path().to_str().unwrap(), &[]); + let _master = start_moon(16620, master_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); write_keys("127.0.0.1:16620", "promo", 100); - let mut replica = start_moon(16621, replica_dir.path().to_str().unwrap(), &[]); + let _replica = start_moon(16621, replica_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); send_cmd("127.0.0.1:16621", "REPLICAOF 127.0.0.1 16620"); thread::sleep(Duration::from_secs(2)); @@ -236,11 +244,6 @@ mod tests { get_result.contains("promoted_value"), "Promoted replica should accept writes" ); - - send_cmd("127.0.0.1:16620", "SHUTDOWN NOSAVE"); - send_cmd("127.0.0.1:16621", "SHUTDOWN NOSAVE"); - let _ = master.wait(); - let _ = replica.wait(); } /// G17: Full resync when replica offset falls outside backlog window. @@ -254,7 +257,7 @@ mod tests { let replica_dir = tempfile::tempdir().unwrap(); // Start master with a tiny replication backlog (1KB) - let mut master = start_moon( + let _master = start_moon( 16630, master_dir.path().to_str().unwrap(), &["--repl-backlog-size", "1024"], @@ -280,15 +283,15 @@ mod tests { // Disconnect replica // SAFETY: `child.id()` returns a valid PID for a process we just spawned. // SIGKILL is always valid. We check the return code for robustness. - let ret = unsafe { libc::kill(replica.id() as i32, libc::SIGKILL) }; + let ret = unsafe { libc::kill(replica.0.id() as i32, libc::SIGKILL) }; assert_eq!(ret, 0, "libc::kill failed"); - let _ = replica.wait(); + let _ = replica.0.wait(); // Write enough to overflow the 1KB backlog write_keys("127.0.0.1:16630", "overflow", 500); // Reconnect replica — should trigger full resync - let mut replica2 = start_moon(16631, replica_dir.path().to_str().unwrap(), &[]); + let _replica2 = start_moon(16631, replica_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); send_cmd("127.0.0.1:16631", "REPLICAOF 127.0.0.1 16630"); thread::sleep(Duration::from_secs(4)); @@ -296,12 +299,6 @@ mod tests { let master_size = dbsize("127.0.0.1:16630"); let replica_size = dbsize("127.0.0.1:16631"); - // Cleanup - send_cmd("127.0.0.1:16630", "SHUTDOWN NOSAVE"); - send_cmd("127.0.0.1:16631", "SHUTDOWN NOSAVE"); - let _ = master.wait(); - let _ = replica2.wait(); - assert_eq!( replica_size, master_size, "Replica should match master after full resync: replica={}, master={}", @@ -319,13 +316,13 @@ mod tests { let master_dir = tempfile::tempdir().unwrap(); let replica_dir = tempfile::tempdir().unwrap(); - let mut master = start_moon(16640, master_dir.path().to_str().unwrap(), &[]); + let _master = start_moon(16640, master_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); // Initial data write_keys("127.0.0.1:16640", "partition", 100); - let mut replica = start_moon(16641, replica_dir.path().to_str().unwrap(), &[]); + let _replica = start_moon(16641, replica_dir.path().to_str().unwrap(), &[]); thread::sleep(Duration::from_millis(500)); send_cmd("127.0.0.1:16641", "REPLICAOF 127.0.0.1 16640"); thread::sleep(Duration::from_secs(2)); @@ -351,12 +348,6 @@ mod tests { let master_size = dbsize("127.0.0.1:16640"); let replica_size = dbsize("127.0.0.1:16641"); - // Cleanup - send_cmd("127.0.0.1:16640", "SHUTDOWN NOSAVE"); - send_cmd("127.0.0.1:16641", "SHUTDOWN NOSAVE"); - let _ = master.wait(); - let _ = replica.wait(); - assert_eq!( replica_size, master_size, "Replica should catch up after partition: replica={}, master={}", From ac698e6af3d9d571c048454ecc7293abc3acedc4 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 01:14:40 +0700 Subject: [PATCH 3/5] =?UTF-8?q?feat(replication):=20v0.7=20R0.5=20?= =?UTF-8?q?=E2=80=94=20vector/text=20index-plane=20sync=20(defs=20+=20cont?= =?UTF-8?q?ents)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before this change a replica synchronized only the KV keyspace: FT index definitions never left the master (FT.CREATE/FT.DROPINDEX/FT.CONFIG are connection-layer commands that never reach wal_append_and_fanout), and the replica's apply path ran generic dispatch only, skipping the master's auto-index parity hooks. A replica answered FT._LIST empty and FT.SEARCH with errors even while its hashes matched the master byte-for-byte. Snapshot leg: - The FULLRESYNC RDB now carries the master's vector + text index DEFINITIONS as moon-private RDB AUX fields (opcode 0xFA, keys moon-vector-defs / moon-text-defs), reusing the sidecar codecs (serialize_index_metas_v5 / serialize_text_index_metas). Standard RDB loaders skip unknown AUX keys, so the snapshot stays Redis-compatible and the replication offset math is untouched. - On load the replica drops ALL local indexes (full resync = authoritative replace), installs the master's definitions, and backfills them by rescanning matching HASH keys — the same "restart semantics" rescan restart recovery performs (event_loop.rs phase 2). Live leg: - New ShardMessage::ReplicateVerbatim fans a pre-serialized command into the replication plane only (backlog + offset + replica streams; WAL/AOF deliberately None — defs are durable via the sidecars, an AOF copy would double-apply on recovery). - The monoio connection layer pushes successful FT.CREATE / FT.DROPINDEX / FT.CONFIG SET through it (single-shard; multi-shard FT.* replication rides the R2 broadcast redesign; the tokio-sharded master already rejects PSYNC). - The replica applies FT.* through the same ft_create/ft_dropindex/ ft_config handlers (live FT.CREATE = no backfill, master parity), and runs the master's index-parity hooks after every applied KV write: HSET auto-index, DEL/UNLINK tombstone, HDEL vector-field tombstone, FLUSHDB/FLUSHALL content clear. Tests (red/green TDD): - New black-box e2e replica_syncs_vector_index_defs_and_contents (confirmed RED at step 1 before the fix): snapshot defs + KNN backfill, live HSET indexing, live DEL tombstone, live FT.CREATE streaming, FLUSHALL clears contents keeps defs. - Unit tests for the RDB aux carrier: round-trip, loader-skips-aux, foreign/truncated-buffer rejection (no panic on any prefix cut). - No regression: 41 replication unit tests, 2/2 streaming e2e, 5/5 hardening suite, clippy clean on monoio + tokio feature sets. Refs: task #24 (v0.7 R0.5), stacked on R0 (faed291b, e346d0e0) author: Tin Dang --- CHANGELOG.md | 32 +++ src/persistence/redis_rdb.rs | 104 ++++++++++ src/replication/apply.rs | 291 ++++++++++++++++++++++++++- src/replication/master.rs | 39 +++- src/server/conn/handler_monoio/ft.rs | 38 ++++ src/shard/dispatch.rs | 9 + src/shard/spsc_handler.rs | 20 ++ tests/replication_streaming.rs | 221 ++++++++++++++++++++ 8 files changed, 745 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d3ac1102..4db60a4de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,38 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 post-ACL, so the H-3 deniability guarantee is unchanged). Re-greens all 5 `client_tracking_invalidation` black-box tests on monoio. +### Added — vector/text index-plane replication sync (v0.7 R0.5) + +- **Before this change a replica synchronized only the KV keyspace** — FT index + definitions never left the master (FT.CREATE/FT.DROPINDEX/FT.CONFIG are + handled at the connection layer and never reached the replication fanout), + and the replica's `apply` path ran generic dispatch only, skipping the + master's auto-index parity hooks. A replica answered `FT._LIST` with nothing + and `FT.SEARCH` with errors even while its hashes matched the master. +- **Snapshot leg:** the FULLRESYNC RDB now carries the master's vector + text + index *definitions* as moon-private RDB AUX fields (opcode `0xFA`, keys + `moon-vector-defs` / `moon-text-defs`), reusing the sidecar codecs + (`serialize_index_metas_v5`, `serialize_text_index_metas`). Standard RDB + loaders skip AUX fields, so the snapshot stays Redis-tool-compatible. On + load the replica drops all local indexes (full resync = authoritative + replace), installs the master's definitions, and backfills them by + rescanning matching HASH keys — the same "restart semantics" rescan restart + recovery performs. +- **Live leg:** successful FT.CREATE / FT.DROPINDEX / FT.CONFIG SET on the + master now fan out verbatim to the backlog + connected replicas via a new + `ShardMessage::ReplicateVerbatim` (offset-accounted like any replicated + write; durability remains the sidecar's job — no WAL/AOF leg). The replica + applies them through the same `ft_create`/`ft_dropindex`/`ft_config` + handlers, and runs the master's index-parity hooks after every applied KV + write (HSET auto-index, DEL/UNLINK tombstone, HDEL vector-field tombstone, + FLUSHDB/FLUSHALL content clear) so replica indexes track replica keyspace. +- New black-box acceptance test (`replica_syncs_vector_index_defs_and_contents`) + covering snapshot defs + backfill, live HSET indexing, live DEL tombstoning, + live FT.CREATE streaming, and FLUSHALL clear-contents-keep-defs semantics. +- **Scope:** single-shard master (matches R0); multi-shard FT.* replication + rides the R2 broadcast redesign. Graph-plane replication is a separate + v0.7 workstream. + ### Added — replica now applies the replication stream end-to-end (v0.7 R0) - **Foundational fix**: before this change a `REPLICAOF` replica completed the diff --git a/src/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs index bdad5fa93..d2a1f4fa2 100644 --- a/src/persistence/redis_rdb.rs +++ b/src/persistence/redis_rdb.rs @@ -432,7 +432,32 @@ pub fn write_rdb(databases: &[Database], buf: &mut Vec) { /// avoiding `Database: Clone` requirements. Used by PSYNC full-resync where /// databases are held behind `RwLockReadGuard`s and cannot be cloned or moved. pub fn write_rdb_refs(databases: &[&Database], buf: &mut Vec) { + write_rdb_refs_with_moon_aux(databases, &[], buf); +} + +/// Moon replication aux key: FULLRESYNC snapshots carry the vector index +/// DEFINITIONS (serialized `index_persist` v5 sidecar bytes) inside the RDB as +/// a private AUX field, so defs arrive atomically with the keyspace snapshot +/// and never perturb the replication offset math. Foreign RDB parsers (and +/// older moons) skip unknown aux keys by design. +pub const MOON_AUX_VECTOR_DEFS: &[u8] = b"moon-vector-defs"; +/// Moon replication aux key: text index definitions +/// (`text::index_persist::serialize_text_index_metas` bytes). See +/// [`MOON_AUX_VECTOR_DEFS`]. +pub const MOON_AUX_TEXT_DEFS: &[u8] = b"moon-text-defs"; + +/// `write_rdb_refs` plus moon-private AUX fields written immediately after +/// the standard header aux block (before any SELECTDB), which is what lets +/// [`read_moon_aux`] find them with a bounded header walk. +pub fn write_rdb_refs_with_moon_aux( + databases: &[&Database], + moon_aux: &[(&[u8], &[u8])], + buf: &mut Vec, +) { write_rdb_header(buf); + for (key, value) in moon_aux { + write_aux(buf, key, value); + } let now_ms = current_time_ms(); @@ -487,6 +512,38 @@ pub fn save(databases: &[Database], path: &Path) -> anyhow::Result<()> { // Public API: read // --------------------------------------------------------------------------- +/// Extract one moon-private AUX value from an RDB buffer produced by +/// [`write_rdb_refs_with_moon_aux`]. +/// +/// Walks the header AUX block only — every aux field (standard + moon) +/// precedes the first SELECTDB by construction, so the walk is bounded and +/// never has to understand entry encodings. Returns `None` for a foreign or +/// truncated buffer, an unknown key, or any parse hiccup — the RDB loader +/// proper is where malformed data gets reported. +pub fn read_moon_aux(data: &[u8], key: &[u8]) -> Option> { + if data.len() < 9 || &data[..5] != REDIS_RDB_MAGIC { + return None; + } + let mut cursor = Cursor::new(data); + cursor.set_position(9); // magic + version + loop { + let mut opcode = [0u8; 1]; + if cursor.read_exact(&mut opcode).is_err() { + return None; + } + if opcode[0] != RDB_OPCODE_AUX { + // First non-AUX opcode ends the header block — moon aux, if + // present, would have appeared by now. + return None; + } + let k = read_redis_string(&mut cursor).ok()?; + let v = read_redis_string(&mut cursor).ok()?; + if k == key { + return Some(v); + } + } +} + /// Load an RDB file in Redis format into the provided databases. /// /// Verifies magic bytes, version, and CRC64 checksum. @@ -675,6 +732,53 @@ fn read_rdb_entry( mod tests { use super::*; + #[test] + fn moon_aux_round_trip_and_loader_skips_it() { + let db = Database::new(); + let refs = [&db]; + let mut buf = Vec::new(); + write_rdb_refs_with_moon_aux( + &refs, + &[ + (MOON_AUX_VECTOR_DEFS, b"vec-blob\x00\x01"), + (MOON_AUX_TEXT_DEFS, b"text-blob"), + ], + &mut buf, + ); + assert_eq!( + read_moon_aux(&buf, MOON_AUX_VECTOR_DEFS).as_deref(), + Some(&b"vec-blob\x00\x01"[..]) + ); + assert_eq!( + read_moon_aux(&buf, MOON_AUX_TEXT_DEFS).as_deref(), + Some(&b"text-blob"[..]) + ); + assert_eq!(read_moon_aux(&buf, b"moon-unknown"), None); + // The standard loader must skip moon aux fields untouched. + let mut dbs = vec![Database::new()]; + let loaded = load_rdb(&mut dbs, &buf).expect("aux-carrying RDB must load"); + assert_eq!(loaded, 0); + } + + #[test] + fn read_moon_aux_rejects_foreign_and_truncated_buffers() { + assert_eq!(read_moon_aux(b"", MOON_AUX_VECTOR_DEFS), None); + assert_eq!(read_moon_aux(b"REDIS", MOON_AUX_VECTOR_DEFS), None); + assert_eq!(read_moon_aux(b"NOTRDB0010....", MOON_AUX_VECTOR_DEFS), None); + // Plain RDB with no moon aux: header walk ends at first non-AUX opcode. + let db = Database::new(); + let mut buf = Vec::new(); + write_rdb_refs(&[&db], &mut buf); + assert_eq!(read_moon_aux(&buf, MOON_AUX_VECTOR_DEFS), None); + // Truncated mid-aux must not panic. + let db2 = Database::new(); + let mut buf2 = Vec::new(); + write_rdb_refs_with_moon_aux(&[&db2], &[(MOON_AUX_VECTOR_DEFS, b"blob")], &mut buf2); + for cut in 9..buf2.len().min(40) { + let _ = read_moon_aux(&buf2[..cut], MOON_AUX_VECTOR_DEFS); + } + } + #[test] fn test_crc64_jones_polynomial() { // Known test vector: CRC64 of "123456789" with Jones polynomial diff --git a/src/replication/apply.rs b/src/replication/apply.rs index d995d48a7..b64f84d00 100644 --- a/src/replication/apply.rs +++ b/src/replication/apply.rs @@ -174,6 +174,16 @@ pub(crate) fn apply_local(rc: &ReplCommand) -> bool { ); } + // FT.* index-definition commands (v0.7 R0.5) are streamed verbatim by + // the master's connection layer (`ShardMessage::ReplicateVerbatim`) — + // generic `dispatch()` does not know them. Route through the same + // handlers the master's connection layer uses. + if cmd.len() > 3 && cmd[..3].eq_ignore_ascii_case(b"FT.") { + let resp = apply_ft(s, cmd, args, db_idx); + warn_on_error(cmd, &resp); + return; + } + // MOVE / cross-db COPY touch two databases at once and are intercepted // BEFORE generic dispatch on the master (see `spsc_two_db`). Generic // `dispatch()` cannot apply them — it returns an error for MOVE and @@ -188,19 +198,106 @@ pub(crate) fn apply_local(rc: &ReplCommand) -> bool { // COPY with no DB clause / same-db COPY: fall through to dispatch. } - let db = &mut s.databases[db_idx]; - // Replica applies off the shard's periodic clock tick; refresh directly - // so command-relative expiries (EXPIRE/SETEX) compute against real time. - db.refresh_now(); - let mut selected = db_idx; - let resp = match cmd_dispatch(db, cmd, args, &mut selected, db_count) { - DispatchResult::Response(f) | DispatchResult::Quit(f) => f, + let resp = { + let db = &mut s.databases[db_idx]; + // Replica applies off the shard's periodic clock tick; refresh + // directly so command-relative expiries (EXPIRE/SETEX) compute + // against real time. + db.refresh_now(); + let mut selected = db_idx; + match cmd_dispatch(db, cmd, args, &mut selected, db_count) { + DispatchResult::Response(f) | DispatchResult::Quit(f) => f, + } }; + // Index-plane parity (v0.7 R0.5): mirror of the master's + // connection-layer hook block — every dispatch path that applies KV + // writes MUST run these, or the replica's FT indexes silently diverge + // from its own keyspace (wire-parity requirement, see + // `auto_delete_vectors` docs). + if !matches!(resp, Frame::Error(_)) { + apply_index_parity_hooks(s, cmd, args, db_idx as u8); + } warn_on_error(cmd, &resp); }) .is_some() } +/// Apply a replicated FT.* index-definition command (FT.CREATE / FT.DROPINDEX +/// / FT.CONFIG SET) through the same handlers the master's connection layer +/// uses. Deliberately NO keyspace backfill on live FT.CREATE — master parity: +/// FT.CREATE never indexes pre-existing keys outside restart recovery (and, +/// on a replica, the snapshot-load path below). +fn apply_ft( + s: &mut crate::shard::slice::ShardSlice, + cmd: &[u8], + args: &[Frame], + db_idx: usize, +) -> Frame { + use crate::command::vector_search::{ft_admin, ft_config, ft_create}; + let db_index = db_idx as u8; + if cmd.eq_ignore_ascii_case(b"FT.CREATE") { + ft_create::ft_create(&mut s.vector_store, &mut s.text_store, args, db_index) + } else if cmd.eq_ignore_ascii_case(b"FT.DROPINDEX") { + ft_admin::ft_dropindex( + &mut s.vector_store, + &mut s.text_store, + s.databases.get_mut(db_idx), + args, + db_index, + ) + } else if cmd.eq_ignore_ascii_case(b"FT.CONFIG") { + ft_config::ft_config(&mut s.vector_store, &mut s.text_store, args, db_index) + } else { + // Only def mutations are streamed; any other FT.* here is unexpected — + // skip quietly rather than fabricate a divergence warning. + tracing::debug!( + "replica apply: ignoring non-replicable FT command {}", + String::from_utf8_lossy(cmd) + ); + Frame::Null + } +} + +/// Mirror of the master's connection-layer index-parity block +/// (`handler_monoio/mod.rs`, "HSET auto-index" onwards): HSET feeds the +/// auto-indexer, DEL/UNLINK tombstone, HDEL of a vector field tombstones, +/// FLUSHDB/FLUSHALL clear index contents (definitions survive). +fn apply_index_parity_hooks( + s: &mut crate::shard::slice::ShardSlice, + cmd: &[u8], + args: &[Frame], + db_index: u8, +) { + use crate::shard::spsc_handler as hooks; + if cmd.eq_ignore_ascii_case(b"HSET") { + if let Some(key) = args + .first() + .and_then(crate::server::connection::extract_bytes) + { + // Return value (txn vector-intents) is only meaningful inside an + // active CrossStoreTxn; replica apply has none. + let _ = hooks::auto_index_hset_public( + &mut s.vector_store, + &mut s.text_store, + key.as_ref(), + args, + db_index, + ); + } + } else if cmd.eq_ignore_ascii_case(b"DEL") || cmd.eq_ignore_ascii_case(b"UNLINK") { + hooks::auto_delete_vectors(&mut s.vector_store, args, db_index); + } else if cmd.eq_ignore_ascii_case(b"HDEL") { + hooks::auto_hdel_vectors(&mut s.vector_store, args, db_index); + } else if cmd.eq_ignore_ascii_case(b"FLUSHDB") || cmd.eq_ignore_ascii_case(b"FLUSHALL") { + hooks::auto_flush_indexes( + &mut s.vector_store, + &mut s.text_store, + cmd.eq_ignore_ascii_case(b"FLUSHDB"), + db_index, + ); + } +} + /// A replicated write that fails to apply is a silent-divergence risk — log it /// loudly instead of dropping it on the floor. (Read-only errors cannot occur /// here: apply bypasses the connection-layer read-only guard by construction.) @@ -259,11 +356,18 @@ fn apply_two_db( /// Returns the number of keys loaded, or an error if this thread has no /// `ShardSlice` or the RDB is malformed. pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { + use crate::persistence::redis_rdb; + // Moon-private aux fields (written by the master right after the RDB + // header) carry the FT index DEFINITIONS; standard RDB loaders skip them. + let vec_defs = redis_rdb::read_moon_aux(rdb, redis_rdb::MOON_AUX_VECTOR_DEFS); + let text_defs = redis_rdb::read_moon_aux(rdb, redis_rdb::MOON_AUX_TEXT_DEFS); match crate::shard::slice::try_with_shard(|s| { for db in s.databases.iter_mut() { db.clear(); } - crate::persistence::redis_rdb::load_rdb(&mut s.databases, rdb) + let loaded = redis_rdb::load_rdb(&mut s.databases, rdb)?; + install_snapshot_index_defs(s, vec_defs.as_deref(), text_defs.as_deref()); + Ok(loaded) }) { Some(r) => r, None => Err(anyhow::anyhow!( @@ -272,6 +376,177 @@ pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { } } +/// Full resync = authoritative replace: drop every replica-local FT index, +/// install the master's definitions, then backfill them from the just-loaded +/// keyspace. Backfill here is "restart semantics" — the same matching-hash +/// rescan restart recovery performs (`event_loop.rs`, "Auto-reindex existing +/// HASH keys"); live FT.CREATE apply deliberately does NOT backfill. +fn install_snapshot_index_defs( + s: &mut crate::shard::slice::ShardSlice, + vec_defs: Option<&[u8]>, + text_defs: Option<&[u8]>, +) { + let names: Vec = s.vector_store.index_names().into_iter().cloned().collect(); + for n in &names { + s.vector_store.drop_index(n); + } + for n in s.text_store.index_names() { + s.text_store.drop_index(&n); + } + + // (db_index, key_prefix) pairs feeding the backfill scan below. + let mut prefixes: Vec<(u8, bytes::Bytes)> = Vec::new(); + + if let Some(blob) = vec_defs { + match crate::vector::index_persist::deserialize_index_metas_with_weights(blob) { + Ok(pairs) => { + for (meta, weight) in pairs { + for p in &meta.key_prefixes { + prefixes.push((meta.db_index, p.clone())); + } + let name = meta.name.clone(); + if let Err(e) = s.vector_store.create_index(meta) { + tracing::warn!( + "replica snapshot: failed to create vector index '{}': {}", + String::from_utf8_lossy(&name), + e + ); + continue; + } + if weight != 1.0 + && let Some(idx) = s.vector_store.get_index_mut(&name) + { + idx.set_compaction_weight(weight); + } + } + } + Err(e) => tracing::warn!( + "replica snapshot: vector index defs unreadable ({}); vector indexes NOT replicated", + e + ), + } + } + + #[cfg(not(feature = "text-index"))] + if text_defs.is_some() { + tracing::warn!( + "replica snapshot: master streamed text index defs but this build lacks the \ + text-index feature; text indexes NOT replicated" + ); + } + #[cfg(feature = "text-index")] + if let Some(blob) = text_defs { + match crate::text::index_persist::deserialize_text_index_metas(blob) { + Ok(metas) => { + for meta in metas { + for p in &meta.key_prefixes { + prefixes.push((meta.db_index, p.clone())); + } + let mut text_index = crate::text::store::TextIndex::new( + meta.name.clone(), + meta.key_prefixes.clone(), + meta.text_fields.clone(), + meta.bm25_config, + ); + // Carry the master's db binding forward (WS5a) — same as + // the restart-recovery restore path. + text_index.db_index = meta.db_index; + if let Err(e) = s.text_store.create_index(meta.name.clone(), text_index) { + tracing::warn!( + "replica snapshot: failed to create text index '{}': {}", + String::from_utf8_lossy(&meta.name), + e + ); + } + } + } + Err(e) => tracing::warn!( + "replica snapshot: text index defs unreadable ({}); text indexes NOT replicated", + e + ), + } + } + + if prefixes.is_empty() { + return; + } + let db_count = s.databases.len(); + let mut backfilled = 0usize; + for db_idx in 0..db_count { + let wanted: Vec<&bytes::Bytes> = prefixes + .iter() + .filter(|(d, _)| *d as usize == db_idx) + .map(|(_, p)| p) + .collect(); + if wanted.is_empty() { + continue; + } + let matching = collect_matching_hash_args(&s.databases[db_idx], |key| { + wanted.iter().any(|p| key.starts_with(&p[..])) + }); + for (key, args) in &matching { + let _ = crate::shard::spsc_handler::auto_index_hset_public( + &mut s.vector_store, + &mut s.text_store, + key, + args, + db_idx as u8, + ); + } + backfilled += matching.len(); + } + if backfilled > 0 { + tracing::info!( + "replica snapshot: backfilled {} hash key(s) into replicated FT indexes", + backfilled + ); + } +} + +/// Collect `(key, HSET-shaped args)` for every HASH key in `db` matching +/// `key_matches` — args are `[key, field, value, ...]`, ready for +/// `auto_index_hset_public`. Same extraction as restart recovery's rescan +/// closure in `event_loop.rs`. +fn collect_matching_hash_args( + db: &crate::storage::Database, + key_matches: impl Fn(&[u8]) -> bool, +) -> Vec<(Vec, Vec)> { + use crate::storage::compact_value::RedisValueRef; + let mut matching = Vec::new(); + for (key, entry) in db.data().iter() { + let key_bytes = key.as_bytes(); + if !key_matches(key_bytes) { + continue; + } + let mut args = Vec::new(); + args.push(Frame::BulkString(bytes::Bytes::copy_from_slice(key_bytes))); + match entry.as_redis_value() { + RedisValueRef::Hash(map) => { + for (field, value) in map.iter() { + args.push(Frame::BulkString(bytes::Bytes::copy_from_slice(field))); + args.push(Frame::BulkString(bytes::Bytes::copy_from_slice(value))); + } + } + RedisValueRef::HashListpack(lp) => { + let entries: Vec<_> = lp.iter().collect(); + let mut j = 0; + while j + 1 < entries.len() { + args.push(Frame::BulkString(bytes::Bytes::from(entries[j].as_bytes()))); + args.push(Frame::BulkString(bytes::Bytes::from( + entries[j + 1].as_bytes(), + ))); + j += 2; + } + } + _ => continue, + } + if args.len() > 1 { + matching.push((key_bytes.to_vec(), args)); + } + } + matching +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/replication/master.rs b/src/replication/master.rs index 1afd3c405..e8472ab8e 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -647,7 +647,44 @@ pub async fn handle_psync_inline_single_shard( // Shard 0 is this thread's shard — use the thread-local slice. crate::shard::slice::with_shard(|s| { let refs: Vec<&crate::storage::Database> = s.databases.iter().collect(); - crate::persistence::redis_rdb::write_rdb_refs(&refs, &mut rdb_buf); + // v0.7 R0.5: carry vector/text index DEFINITIONS inside the + // snapshot as moon-private RDB aux fields (reusing the + // sidecar codecs), so a fresh replica can recreate the + // indexes and backfill matching hashes after loading the + // keyspace. Contents then stay in sync via the live stream. + let vec_defs = { + let pairs = s.vector_store.collect_index_metas_with_weights(); + if pairs.is_empty() { + None + } else { + Some(crate::vector::index_persist::serialize_index_metas_v5( + &pairs, + )) + } + }; + let text_defs = { + let metas = s.text_store.collect_index_metas(); + if metas.is_empty() { + None + } else { + Some(crate::text::index_persist::serialize_text_index_metas( + &metas, + )) + } + }; + let mut moon_aux: Vec<(&[u8], &[u8])> = Vec::new(); + if let Some(ref v) = vec_defs { + moon_aux + .push((crate::persistence::redis_rdb::MOON_AUX_VECTOR_DEFS, &v[..])); + } + if let Some(ref t) = text_defs { + moon_aux.push((crate::persistence::redis_rdb::MOON_AUX_TEXT_DEFS, &t[..])); + } + crate::persistence::redis_rdb::write_rdb_refs_with_moon_aux( + &refs, + &moon_aux, + &mut rdb_buf, + ); }); } let header = format!("${}\r\n", rdb_buf.len()); diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index a343c4147..e2000017c 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/src/server/conn/handler_monoio/ft.rs @@ -11,6 +11,22 @@ use crate::server::conn::core::{ConnectionContext, ConnectionState}; use crate::server::conn::shared::resolve_ft_search_as_of_lsn; use crate::workspace::strip_workspace_prefix_from_response; +/// True for the FT.* commands whose SUCCESS mutates index definitions and +/// therefore must fan out to replicas (v0.7 R0.5): FT.CREATE, FT.DROPINDEX, +/// and FT.CONFIG SET (GET is a read). Content mutations (HSET/DEL/…) already +/// replicate via the KV stream; FT.COMPACT is a local optimization. +fn is_replicated_ft_def_mutation(cmd: &[u8], cmd_args: &[Frame]) -> bool { + if cmd.eq_ignore_ascii_case(b"FT.CREATE") || cmd.eq_ignore_ascii_case(b"FT.DROPINDEX") { + return true; + } + if cmd.eq_ignore_ascii_case(b"FT.CONFIG") { + if let Some(Frame::BulkString(sub) | Frame::SimpleString(sub)) = cmd_args.first() { + return sub.eq_ignore_ascii_case(b"SET"); + } + } + false +} + /// Handle FT.* commands. Returns `true` if the command was consumed. /// /// Caller should `continue` the frame loop when this returns `true`. @@ -763,6 +779,28 @@ pub(super) async fn try_handle_ft_command( Frame::Error(Bytes::from_static(b"ERR unknown FT.* command")) } }); + // v0.7 R0.5: index-DEFINITION mutations must reach replicas. FT.* + // executes here at the connection layer (never crosses the SPSC write + // path), so on success fan the ORIGINAL command bytes into the + // replication plane via ReplicateVerbatim — exact parity, no + // reconstruction. Single-shard only (multi-shard FT.* replication + // rides the R2 broadcast redesign). The replica applies it through + // the same ft_create/ft_dropindex/ft_config handlers. + if !matches!(response, Frame::Error(_)) && is_replicated_ft_def_mutation(cmd, cmd_args) { + use ringbuf::traits::Producer; + let serialized = crate::persistence::aof::serialize_command(frame); + let mut producers = ctx.dispatch_tx.borrow_mut(); + if let Some(prod) = producers.get_mut(0) { + let msg = + crate::shard::dispatch::ShardMessage::ReplicateVerbatim { bytes: serialized }; + if prod.try_push(msg).is_err() { + tracing::warn!( + "replication: SPSC full, {} not fanned to replicas (replicas must full-resync)", + String::from_utf8_lossy(cmd) + ); + } + } + } let mut response = response; if let Some(ws_id) = conn.workspace_id.as_ref() { strip_workspace_prefix_from_response(ws_id, cmd, &mut response); diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index c23b3cef6..3bd466191 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -466,6 +466,15 @@ pub enum ShardMessage { /// Remove a replica's sender channel from this shard's fan-out list. /// Called when a replica disconnects or REPLICAOF NO ONE is executed. UnregisterReplica { replica_id: u64 }, + /// Fan a pre-serialized RESP command verbatim into the replication plane + /// (backlog + live replica streams + offset) WITHOUT touching WAL/AOF. + /// + /// For connection-layer commands that never cross the SPSC write path but + /// must replicate — FT.CREATE / FT.DROPINDEX / FT.CONFIG SET (v0.7 R0.5): + /// their durability is the vector/text sidecar, not the AOF, so only the + /// replication legs of `wal_append_and_fanout` apply. No-ops when no + /// replica has ever attached (no backlog, no replica_txs). + ReplicateVerbatim { bytes: bytes::Bytes }, /// Register a CDC subscriber with this shard's fan-out registry (C3b-2). /// /// The connection handler creates a bounded channel, ships the sender diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 787e73dce..f4e614988 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2493,6 +2493,26 @@ pub(crate) fn handle_shard_message_shared( ShardMessage::UnregisterReplica { replica_id } => { replica_txs.retain(|(id, _)| *id != replica_id); } + ShardMessage::ReplicateVerbatim { bytes } => { + // Replication legs only: backlog + offset + live replica fan-out. + // WAL writer / AOF pool are deliberately None — the commands routed + // here (FT.CREATE / FT.DROPINDEX / FT.CONFIG SET) are durable via + // the vector/text sidecars, and an AOF copy would double-apply on + // recovery. `wal_fanout_has_work` no-ops the whole call when no + // replica has ever attached. + let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND; + let _ = wal_append_and_fanout( + &bytes, + &mut None, + repl_backlog, + replica_txs, + repl_state, + shard_id, + None, + false, + &mut aof_budget, + ); + } ShardMessage::MigrateConnection(_) => { // MigrateConnection is collected by drain_spsc_shared into pending_migrations, // not dispatched through handle_shard_message_shared. diff --git a/tests/replication_streaming.rs b/tests/replication_streaming.rs index ebff2b667..d3930541f 100644 --- a/tests/replication_streaming.rs +++ b/tests/replication_streaming.rs @@ -186,6 +186,64 @@ impl Drop for Killer { } } +/// Send one command as a RESP array (binary-safe — vector blobs contain +/// arbitrary bytes) and return the whole reply, lossy-decoded, read until the +/// server goes quiet. FT.SEARCH replies are nested arrays; tests only need +/// substring assertions, not structured parsing. +fn send_resp(addr: &str, parts: &[&[u8]]) -> String { + let Ok(mut stream) = TcpStream::connect(addr) else { + return String::new(); + }; + stream + .set_read_timeout(Some(Duration::from_millis(500))) + .ok(); + let mut req = Vec::new(); + req.extend_from_slice(format!("*{}\r\n", parts.len()).as_bytes()); + for p in parts { + req.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + req.extend_from_slice(p); + req.extend_from_slice(b"\r\n"); + } + if stream.write_all(&req).is_err() { + return String::new(); + } + stream.flush().ok(); + let mut out = Vec::new(); + let mut buf = [0u8; 4096]; + loop { + match Read::read(&mut stream, &mut buf) { + Ok(0) | Err(_) => break, // closed or quiet for 500ms — reply done + Ok(n) => out.extend_from_slice(&buf[..n]), + } + } + String::from_utf8_lossy(&out).into_owned() +} + +/// Little-endian FP32 blob for a 4-dim vector. +fn vec4(a: f32, b: f32, c: f32, d: f32) -> Vec { + let mut v = Vec::with_capacity(16); + for f in [a, b, c, d] { + v.extend_from_slice(&f.to_le_bytes()); + } + v +} + +/// FT.SEARCH KNN over `idx` for the query vector, returning the raw reply text. +fn knn_search(addr: &str, idx: &str, query: &[u8]) -> String { + send_resp( + addr, + &[ + b"FT.SEARCH", + idx.as_bytes(), + b"*=>[KNN 4 @vec $q]", + b"PARAMS", + b"2", + b"q", + query, + ], + ) +} + /// REPL-STREAM-01: replica applies the initial snapshot AND subsequent live /// writes streamed from the master. #[test] @@ -298,3 +356,166 @@ fn replica_applies_snapshot_and_live_stream() { "cross-db COPY did not land in replica db1" ); } + +/// REPL-STREAM-02 (v0.7 R0.5): the vector/text index plane replicates — +/// definitions AND contents. +/// +/// Before R0.5 only the KV plane replicated: the replica applied HSET into the +/// hash but never fed `auto_index_hset` (FT.SEARCH on the replica returned +/// nothing), never tombstoned on DEL (ghosts), never cleared on FLUSHALL, and +/// FT.CREATE was not streamed at all (FT._LIST on the replica was empty). +#[test] +#[ignore] +fn replica_syncs_vector_index_defs_and_contents() { + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + + let master_addr = "127.0.0.1:16710"; + let replica_addr = "127.0.0.1:16711"; + + let _master = Killer(start_moon(16710, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + + // Pre-attach state: index definition + one indexed document. Both must + // reach the replica via the FULLRESYNC snapshot (defs) + backfill (docs). + let created = send_resp( + master_addr, + &[ + b"FT.CREATE", + b"repidx", + b"ON", + b"HASH", + b"PREFIX", + b"1", + b"v:", + b"SCHEMA", + b"vec", + b"VECTOR", + b"HNSW", + b"6", + b"DIM", + b"4", + b"TYPE", + b"FLOAT32", + b"DISTANCE_METRIC", + b"L2", + ], + ); + assert!(created.contains("OK"), "FT.CREATE failed: {created}"); + let v1 = vec4(1.0, 0.0, 0.0, 0.0); + send_resp(master_addr, &[b"HSET", b"v:1", b"vec", &v1]); + // Sanity: master itself must find it. + let q = vec4(1.0, 0.0, 0.0, 0.0); + assert!( + knn_search(master_addr, "repidx", &q).contains("v:1"), + "master should find v:1 before attach" + ); + + let _replica = Killer(start_moon(16711, replica_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + send_cmd(replica_addr, &format!("REPLICAOF 127.0.0.1 {}", 16710)); + + // 1. Index DEFINITION must arrive with the snapshot. + let def_synced = wait_until(Duration::from_secs(10), || { + send_resp(replica_addr, &[b"FT._LIST"]).contains("repidx") + }); + assert!( + def_synced, + "replica FT._LIST never listed repidx (index defs not in snapshot): {}", + send_resp(replica_addr, &[b"FT._LIST"]) + ); + + // 2. Pre-attach CONTENT must be searchable (snapshot backfill). + let content_synced = wait_until(Duration::from_secs(10), || { + knn_search(replica_addr, "repidx", &q).contains("v:1") + }); + assert!( + content_synced, + "replica never indexed snapshot doc v:1: {}", + knn_search(replica_addr, "repidx", &q) + ); + + // 3. Live HSET must be indexed on the replica (apply-side auto-index hook). + let v2 = vec4(0.0, 1.0, 0.0, 0.0); + send_resp(master_addr, &[b"HSET", b"v:2", b"vec", &v2]); + let live_indexed = wait_until(Duration::from_secs(10), || { + knn_search(replica_addr, "repidx", &q).contains("v:2") + }); + assert!( + live_indexed, + "replica never indexed live-streamed v:2: {}", + knn_search(replica_addr, "repidx", &q) + ); + + // 4. Live DEL must tombstone on the replica (apply-side delete hook). + send_cmd(master_addr, "DEL v:1"); + let deleted = wait_until(Duration::from_secs(10), || { + !knn_search(replica_addr, "repidx", &q).contains("v:1") + }); + assert!( + deleted, + "replica still returns deleted v:1 (ghost): {}", + knn_search(replica_addr, "repidx", &q) + ); + + // 5. Live FT.CREATE must stream (def created AFTER attach). + let created2 = send_resp( + master_addr, + &[ + b"FT.CREATE", + b"repidx2", + b"ON", + b"HASH", + b"PREFIX", + b"1", + b"w:", + b"SCHEMA", + b"vec", + b"VECTOR", + b"HNSW", + b"6", + b"DIM", + b"4", + b"TYPE", + b"FLOAT32", + b"DISTANCE_METRIC", + b"L2", + ], + ); + assert!( + created2.contains("OK"), + "second FT.CREATE failed: {created2}" + ); + let live_def = wait_until(Duration::from_secs(10), || { + send_resp(replica_addr, &[b"FT._LIST"]).contains("repidx2") + }); + assert!( + live_def, + "replica FT._LIST never listed live-created repidx2: {}", + send_resp(replica_addr, &[b"FT._LIST"]) + ); + + // 6. FLUSHALL clears index CONTENTS on the replica but keeps definitions. + send_cmd(master_addr, "FLUSHALL"); + let flushed = wait_until(Duration::from_secs(10), || { + !knn_search(replica_addr, "repidx", &q).contains("v:2") + }); + assert!( + flushed, + "replica still returns flushed v:2 (ghost): {}", + knn_search(replica_addr, "repidx", &q) + ); + let list_after_flush = send_resp(replica_addr, &[b"FT._LIST"]); + assert!( + list_after_flush.contains("repidx") && list_after_flush.contains("repidx2"), + "FLUSHALL must keep index definitions on the replica: {list_after_flush}" + ); +} From db9c916b44b58b1bc7141a765565679f8f6660db Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 01:51:53 +0700 Subject: [PATCH 4/5] fix(replication): close PSYNC attach races found by adversarial review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes for the C1/M1 findings from the R0/R0.5 adversarial review, plus two of its hygiene items. C1 — registration-bounded catch-up (silent replica gap): the PSYNC task read backlog catch-up bytes and only then registered the replica with the event loop. Any write drained in between (KV Execute or the new ReplicateVerbatim FT.* fanout) reached neither the RDB, nor the catch-up read, nor live fan-out — the master answered +OK while the new replica silently never learned about it. The master now registers FIRST; RegisterReplica carries a reply channel and the event loop answers with the shard offset at which live fan-out begins (read synchronously between drains, so it is exact). Catch-up then sends precisely [snapshot_offset, registration_offset): no gap, and bytes at or above the registration offset arrive only via the replica channel (the read is truncated, so nothing is delivered twice). C1b — atomic snapshot capture: snapshot_offset was read at fn entry, with the +FULLRESYNC write awaited before the RDB capture. A write interleaving there landed inside the RDB AND above the advertised offset, so catch-up re-delivered it — double-applying non-idempotent commands (INCR) on the replica. The offset read and RDB capture now share one synchronous stretch with no .await between them. Also: backlog eviction during catch-up now aborts the sync loudly (send_backlog_range bails; the replica retries a fresh full resync) instead of the old silent `if let Some` skip. M1 — a full resync whose RDB carries no moon index-def aux (pre-R0.5 master, def-serialization failure) silently wiped the replica's local FT indexes. The authoritative replace now logs a warning with the dropped counts when there is no replacement. Hygiene (review m2/m3): read_moon_aux validates the RDB version bytes like load_rdb (a foreign version would misalign the fixed 9-byte header walk); the FT.* fanout hook checks replication_fanout_active (replica registered or backlog allocated) before paying the serialize + SPSC round trip — the backlog is allocated during the PSYNC handshake before any snapshot capture, so a mutation racing a first-ever attach still fans out. The multi-shard RegisterReplica paths pass registered: None (fire-and-forget, superseded by the R2 PrepareReplicaSync redesign). Tests: new race-guard e2e replica_attach_races_live_ft_create — 30 FT.CREATEs from a side thread racing a mid-stream replica attach, exact FT._LIST parity required (3/3 green). No regression: 3/3 streaming e2e, 5/5 hardening, 40 replication + 27 redis_rdb unit tests, clippy clean on monoio + tokio feature sets. Refs: task #24 follow-up; review findings C1/M1/m2/m3 author: Tin Dang --- CHANGELOG.md | 26 +++ src/persistence/redis_rdb.rs | 5 +- src/replication/apply.rs | 21 ++- src/replication/master.rs | 234 +++++++++++++++++++-------- src/replication/state.rs | 12 ++ src/server/conn/handler_monoio/ft.rs | 21 ++- src/shard/dispatch.rs | 8 + src/shard/spsc_handler.rs | 13 ++ tests/replication_streaming.rs | 88 ++++++++++ 9 files changed, 359 insertions(+), 69 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4db60a4de..a5bc007c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 post-ACL, so the H-3 deniability guarantee is unchanged). Re-greens all 5 `client_tracking_invalidation` black-box tests on monoio. +### Fixed — PSYNC attach races closed (adversarial-review findings on R0/R0.5) + +- **Registration-bounded catch-up:** the master now registers the replica with + the event loop BEFORE reading backlog catch-up bytes, and the event loop + replies with the exact offset where live fan-out begins + (`RegisterReplica.registered` reply channel). Catch-up sends exactly + `[snapshot_offset, registration_offset)` — previously a write drained + between the catch-up read and registration reached neither the RDB, the + catch-up, nor the live stream: a silent, unlogged replica gap (worst for + FT.* def mutations, whose fanout always crosses the SPSC queue). +- **Atomic snapshot capture:** the FULLRESYNC snapshot offset is read in the + same synchronous stretch as the RDB capture (no `.await` between), closing + the inverse race where a write landed inside the RDB *and* above the + advertised offset — double-applying non-idempotent commands (INCR) on the + replica. +- **Backlog eviction during catch-up now aborts the sync loudly** (replica + retries a fresh full resync) instead of silently skipping the missing bytes. +- A replica whose full resync carries NO index definitions (pre-R0.5 master + or def-serialization failure) now logs a warning when the authoritative + replace drops local indexes with no replacement. +- New race-guard e2e `replica_attach_races_live_ft_create` (30 FT.CREATEs + racing a mid-stream attach, exact index-list parity required). +- Hygiene: `read_moon_aux` validates RDB version bytes like `load_rdb`; the + FT.* fanout hook skips the serialize+SPSC round trip until a replica or + backlog exists. + ### Added — vector/text index-plane replication sync (v0.7 R0.5) - **Before this change a replica synchronized only the KV keyspace** — FT index diff --git a/src/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs index d2a1f4fa2..08b50a087 100644 --- a/src/persistence/redis_rdb.rs +++ b/src/persistence/redis_rdb.rs @@ -521,7 +521,10 @@ pub fn save(databases: &[Database], path: &Path) -> anyhow::Result<()> { /// truncated buffer, an unknown key, or any parse hiccup — the RDB loader /// proper is where malformed data gets reported. pub fn read_moon_aux(data: &[u8], key: &[u8]) -> Option> { - if data.len() < 9 || &data[..5] != REDIS_RDB_MAGIC { + // Version bytes are checked too (matching `load_rdb`): the fixed + // `set_position(9)` below is only valid for a 4-byte version field, and a + // foreign version would silently misalign the opcode walk. + if data.len() < 9 || &data[..5] != REDIS_RDB_MAGIC || &data[5..9] != REDIS_RDB_VERSION { return None; } let mut cursor = Cursor::new(data); diff --git a/src/replication/apply.rs b/src/replication/apply.rs index b64f84d00..5cb48a76f 100644 --- a/src/replication/apply.rs +++ b/src/replication/apply.rs @@ -387,12 +387,31 @@ fn install_snapshot_index_defs( text_defs: Option<&[u8]>, ) { let names: Vec = s.vector_store.index_names().into_iter().cloned().collect(); + let local_text_names = s.text_store.index_names(); + let (local_vec, local_text) = (names.len(), local_text_names.len()); for n in &names { s.vector_store.drop_index(n); } - for n in s.text_store.index_names() { + for n in local_text_names { s.text_store.drop_index(&n); } + // The authoritative drop above is silent when the master streams no defs + // at all (pre-R0.5 master, or def serialization failed) — but wiping a + // replica's local indexes with zero replacement is exactly the scenario an + // operator needs to hear about (mixed-version rolling upgrade, REPLICAOF + // pointed at the wrong host). + if vec_defs.is_none() && text_defs.is_none() { + if local_vec + local_text > 0 { + tracing::warn!( + "replica full resync: dropped {} local vector and {} local text index(es); \ + the master's snapshot carried NO index definitions (pre-R0.5 master or \ + def-serialization failure) — FT indexes are gone on this replica", + local_vec, + local_text + ); + } + return; + } // (db_index, key_prefix) pairs feeding the backfill scan below. let mut prefixes: Vec<(u8, bytes::Bytes)> = Vec::new(); diff --git a/src/replication/master.rs b/src/replication/master.rs index e8472ab8e..6f33db7b3 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -442,6 +442,10 @@ async fn register_replica_with_shards( replica_id, tx, backlog_capacity, + // Fire-and-forget: the multi-shard register paths are superseded + // by the R2 PrepareReplicaSync redesign; the offset-reply catch-up + // protocol is wired on the single-shard inline path only. + registered: None, }; let _ = prod.try_push(msg); } @@ -529,6 +533,10 @@ async fn register_replica_with_shards( replica_id, tx, backlog_capacity, + // Fire-and-forget: the multi-shard register paths are superseded + // by the R2 PrepareReplicaSync redesign; the offset-reply catch-up + // protocol is wired on the single-shard inline path only. + registered: None, }; let _ = prod.try_push(msg); } @@ -599,7 +607,10 @@ pub async fn handle_psync_inline_single_shard( ) -> anyhow::Result<()> { use monoio::io::AsyncWriteRentExt; - let (repl_id, repl_id2, current_offset, backlog_slot) = { + // The snapshot offset is NOT read here: FullResync re-reads it inside the + // same synchronous stretch as the RDB capture (see below) so no write can + // slip between the two. + let (repl_id, repl_id2, backlog_slot) = { let rs = repl_state .read() .map_err(|_| anyhow::anyhow!("lock poisoned"))?; @@ -608,12 +619,7 @@ pub async fn handle_psync_inline_single_shard( .first() .cloned() .ok_or_else(|| anyhow::anyhow!("backlog slot missing"))?; - ( - rs.repl_id.clone(), - rs.repl_id2.clone(), - rs.total_offset(), - slot, - ) + (rs.repl_id.clone(), rs.repl_id2.clone(), slot) }; // Decide full vs partial resync against the single-shard backlog. @@ -633,17 +639,24 @@ pub async fn handle_psync_inline_single_shard( match decision { PsyncDecision::FullResync => { - let snapshot_offset = current_offset; - let response = format!("+FULLRESYNC {} {}\r\n", repl_id, snapshot_offset); - let (wr, _) = stream.write_all(response.into_bytes()).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - - // Generate RDB inline by reading all databases on shard 0. + // Snapshot-offset read and RDB capture share ONE synchronous + // stretch (no `.await` between them): tasks on this thread are + // cooperatively scheduled, so nothing can advance the offset or + // mutate the keyspace in between. Reading the offset at fn entry + // (before the FULLRESYNC line was written) let a write land both + // inside the RDB AND above snapshot_offset — re-delivered via + // catch-up, double-applying non-idempotent commands (INCR). + // + // The RDB is generated inline by reading all databases on shard 0. // Hold read guards across the synchronous write to avoid any // Clone requirement on Database (the type intentionally is not // Clone — its internal DashTable + FT/graph indices are large). let mut rdb_buf: Vec = Vec::new(); - { + let snapshot_offset = { + let off = repl_state + .read() + .map(|g| g.total_offset()) + .map_err(|_| anyhow::anyhow!("lock poisoned"))?; // Shard 0 is this thread's shard — use the thread-local slice. crate::shard::slice::with_shard(|s| { let refs: Vec<&crate::storage::Database> = s.databases.iter().collect(); @@ -686,7 +699,11 @@ pub async fn handle_psync_inline_single_shard( &mut rdb_buf, ); }); - } + off + }; + let response = format!("+FULLRESYNC {} {}\r\n", repl_id, snapshot_offset); + let (wr, _) = stream.write_all(response.into_bytes()).await; + wr.map_err(|e| anyhow::anyhow!(e))?; let header = format!("${}\r\n", rdb_buf.len()); let (wr, _) = stream.write_all(header.into_bytes()).await; wr.map_err(|e| anyhow::anyhow!(e))?; @@ -696,15 +713,22 @@ pub async fn handle_psync_inline_single_shard( // string with \r\n during diskless full resync; the next bytes are // backlog/replication stream. Match that wire format. - // Stream any backlog bytes appended between snapshot capture and now. - if let Some(bytes) = backlog_bytes_from(&backlog_slot, snapshot_offset) { - if !bytes.is_empty() { - let (wr, _) = stream.write_all(bytes).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - } - } - - register_replica_inline_single_shard(replica_addr, stream, repl_state, dispatch_tx) + // Register FIRST, then catch up to exactly the registration + // offset. The event loop replies with the offset at which live + // fan-out to this replica begins; every byte below it comes from + // the backlog read, every byte at or above it arrives on the + // replica channel. Reading the backlog BEFORE registering (the + // old order) left a window where a write drained in between + // reached neither leg — a silent, unlogged replica gap. + let reg = push_register_replica_inline(&repl_state, &dispatch_tx)?; + let reg_offset = reg + .reg_rx + .recv_async() + .await + .map_err(|_| anyhow::anyhow!("event loop dropped registration reply"))?; + send_backlog_range(&mut stream, &backlog_slot, snapshot_offset, reg_offset).await?; + + drain_replica_inline_single_shard(reg, replica_addr, stream, repl_state, dispatch_tx) .await?; } PsyncDecision::PartialResync { from_offset } => { @@ -712,35 +736,87 @@ pub async fn handle_psync_inline_single_shard( let (wr, _) = stream.write_all(response.into_bytes()).await; wr.map_err(|e| anyhow::anyhow!(e))?; - if let Some(bytes) = backlog_bytes_from(&backlog_slot, from_offset) { - if !bytes.is_empty() { - let (wr, _) = stream.write_all(bytes).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - } - } - register_replica_inline_single_shard(replica_addr, stream, repl_state, dispatch_tx) + // Same register-then-catch-up ordering as the FullResync arm. + let reg = push_register_replica_inline(&repl_state, &dispatch_tx)?; + let reg_offset = reg + .reg_rx + .recv_async() + .await + .map_err(|_| anyhow::anyhow!("event loop dropped registration reply"))?; + send_backlog_range(&mut stream, &backlog_slot, from_offset, reg_offset).await?; + + drain_replica_inline_single_shard(reg, replica_addr, stream, repl_state, dispatch_tx) .await?; } } Ok(()) } -/// Single-shard inline replica registration. -/// -/// Creates an MPSC channel, pushes `RegisterReplica` onto shard 0's SPSC -/// producer so the event loop picks up the tx into its local `replica_txs` -/// (the authority for live write fan-out), and also records the replica in -/// `ReplicationState.replicas` for WAIT / INFO bookkeeping. Then drains the -/// channel onto the replica's socket until the peer disconnects. +/// Send backlog bytes `[from, to)` to the replica, or fail LOUDLY if the +/// backlog can no longer serve that range (evicted mid-sync). Aborting drops +/// the connection so the replica retries with a fresh full resync — strictly +/// better than the silent gap the old `if let Some(...)` skip produced. #[cfg(feature = "runtime-monoio")] -#[allow(clippy::await_holding_refcell_ref)] -async fn register_replica_inline_single_shard( - addr: std::net::SocketAddr, - stream: monoio::net::TcpStream, - repl_state: Arc>, - dispatch_tx: Rc>>>, +async fn send_backlog_range( + stream: &mut monoio::net::TcpStream, + backlog_slot: &SharedBacklog, + from: u64, + to: u64, ) -> anyhow::Result<()> { use monoio::io::AsyncWriteRentExt; + + if to <= from { + return Ok(()); + } + let need = (to - from) as usize; + let bytes = backlog_bytes_from(backlog_slot, from).ok_or_else(|| { + anyhow::anyhow!( + "replication backlog evicted during catch-up ({}..{}); aborting sync so the \ + replica retries a fresh full resync", + from, + to + ) + })?; + if bytes.len() < need { + // The event loop appended [from, to) before replying with `to`, so a + // shorter read means the head of the range was evicted. + anyhow::bail!( + "replication backlog short read during catch-up (have {} bytes, need {}); \ + aborting sync so the replica retries a fresh full resync", + bytes.len(), + need + ); + } + // Bytes past `to` are already queued on the replica channel by live + // fan-out — truncate to avoid delivering them twice. + let (wr, _) = stream.write_all(bytes[..need].to_vec()).await; + wr.map_err(|e| anyhow::anyhow!(e))?; + Ok(()) +} + +/// Everything the PSYNC task holds between pushing `RegisterReplica` and +/// draining the replica channel: the id, the receive half of the live +/// fan-out channel, its keep-alive tx (for WAIT/INFO bookkeeping), and the +/// registration-offset reply receiver. +#[cfg(feature = "runtime-monoio")] +struct InlineReplicaRegistration { + replica_id: u64, + tx: crate::runtime::channel::MpscSender, + rx: crate::runtime::channel::MpscReceiver, + reg_rx: crate::runtime::channel::MpscReceiver, +} + +/// Push `RegisterReplica` onto shard 0's SPSC so the event loop captures the +/// tx into its local `replica_txs` Vec — the sole authority used by +/// `wal_append_and_fanout` for live write streaming. The message carries a +/// reply channel; the event loop answers with the shard offset at which live +/// fan-out begins, which the caller uses to bound its backlog catch-up read +/// (see `handle_psync_inline_single_shard`). +#[cfg(feature = "runtime-monoio")] +fn push_register_replica_inline( + repl_state: &Arc>, + dispatch_tx: &Rc>>>, +) -> anyhow::Result { use ringbuf::traits::Producer; use std::sync::atomic::Ordering; @@ -748,32 +824,58 @@ async fn register_replica_inline_single_shard( let replica_id = NEXT_REPLICA_ID.fetch_add(1, Ordering::Relaxed); let (tx, rx) = crate::runtime::channel::mpsc_bounded::(1024); + let (reg_tx, reg_rx) = crate::runtime::channel::mpsc_bounded::(1); - // Push RegisterReplica onto shard 0's SPSC so the event loop captures the - // tx into its local replica_txs Vec — the sole authority used by - // wal_append_and_fanout for live write streaming. - { - // `--repl-backlog-size`, carried in RegisterReplica for the lazy fallback-init. - let backlog_capacity = repl_state - .read() - .map(|g| g.backlog_capacity) - .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); - let mut prods = dispatch_tx.borrow_mut(); - if let Some(prod) = prods.get_mut(0) { - let msg = crate::shard::dispatch::ShardMessage::RegisterReplica { - replica_id, - tx: tx.clone(), - backlog_capacity, - }; - if prod.try_push(msg).is_err() { - anyhow::bail!("failed to push RegisterReplica onto shard 0 SPSC"); - } - } else { - anyhow::bail!("shard 0 producer missing"); + // `--repl-backlog-size`, carried in RegisterReplica for the lazy fallback-init. + let backlog_capacity = repl_state + .read() + .map(|g| g.backlog_capacity) + .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); + let mut prods = dispatch_tx.borrow_mut(); + if let Some(prod) = prods.get_mut(0) { + let msg = crate::shard::dispatch::ShardMessage::RegisterReplica { + replica_id, + tx: tx.clone(), + backlog_capacity, + registered: Some(reg_tx), + }; + if prod.try_push(msg).is_err() { + anyhow::bail!("failed to push RegisterReplica onto shard 0 SPSC"); } + } else { + anyhow::bail!("shard 0 producer missing"); } + Ok(InlineReplicaRegistration { + replica_id, + tx, + rx, + reg_rx, + }) +} + +/// Single-shard inline replica drain: record the replica in +/// `ReplicationState.replicas` for WAIT / INFO bookkeeping, then pump the +/// live fan-out channel onto the replica's socket until the peer disconnects. +#[cfg(feature = "runtime-monoio")] +#[allow(clippy::await_holding_refcell_ref)] +async fn drain_replica_inline_single_shard( + reg: InlineReplicaRegistration, + addr: std::net::SocketAddr, + stream: monoio::net::TcpStream, + repl_state: Arc>, + dispatch_tx: Rc>>>, +) -> anyhow::Result<()> { + use monoio::io::AsyncWriteRentExt; + use ringbuf::traits::Producer; + + let InlineReplicaRegistration { + replica_id, + tx, + rx, + reg_rx: _, + } = reg; - // Also bookkeeping for WAIT/INFO. + // Bookkeeping for WAIT/INFO. let replica_info = ReplicaInfo { id: replica_id, addr, diff --git a/src/replication/state.rs b/src/replication/state.rs index 5b65b5487..f068e4657 100644 --- a/src/replication/state.rs +++ b/src/replication/state.rs @@ -238,6 +238,18 @@ impl OffsetHandle { pub fn increment_shard_offset(&self, shard_id: usize, delta: u64) { let _ = self.issue_lsn(shard_id, delta); } + + /// Current offset of one shard. Used by the `RegisterReplica` reply to + /// tell the PSYNC task exactly where live fan-out begins, so its backlog + /// catch-up read covers `[snapshot_offset, this)` with no gap and no + /// overlap. + #[inline] + pub fn shard_offset(&self, shard_id: usize) -> u64 { + self.shard_offsets + .get(shard_id) + .map(|o| o.load(Ordering::Relaxed)) + .unwrap_or(0) + } } const ZEROED_ID: &str = "0000000000000000000000000000000000000000"; diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index e2000017c..1000c3820 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/src/server/conn/handler_monoio/ft.rs @@ -27,6 +27,22 @@ fn is_replicated_ft_def_mutation(cmd: &[u8], cmd_args: &[Frame]) -> bool { false } +/// True once the replication plane has anything to feed: a registered replica +/// or an allocated backlog (a PSYNC handshake allocates the backlog before any +/// snapshot is captured, so an FT.* mutation racing a first-ever attach still +/// fans out). Gates the serialize+SPSC round trip so servers that never +/// replicate pay nothing — mirrors `wal_fanout_has_work` on the KV path. +fn replication_fanout_active(ctx: &ConnectionContext) -> bool { + ctx.repl_state.as_ref().is_some_and(|rs| { + rs.read().is_ok_and(|g| { + !g.replicas.is_empty() + || g.per_shard_backlogs + .first() + .is_some_and(|slot| slot.lock().is_some()) + }) + }) +} + /// Handle FT.* commands. Returns `true` if the command was consumed. /// /// Caller should `continue` the frame loop when this returns `true`. @@ -786,7 +802,10 @@ pub(super) async fn try_handle_ft_command( // reconstruction. Single-shard only (multi-shard FT.* replication // rides the R2 broadcast redesign). The replica applies it through // the same ft_create/ft_dropindex/ft_config handlers. - if !matches!(response, Frame::Error(_)) && is_replicated_ft_def_mutation(cmd, cmd_args) { + if !matches!(response, Frame::Error(_)) + && is_replicated_ft_def_mutation(cmd, cmd_args) + && replication_fanout_active(ctx) + { use ringbuf::traits::Producer; let serialized = crate::persistence::aof::serialize_command(frame); let mut producers = ctx.dispatch_tx.borrow_mut(); diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index 3bd466191..77b7f6728 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -462,6 +462,14 @@ pub enum ShardMessage { replica_id: u64, tx: channel::MpscSender, backlog_capacity: usize, + /// When set, the event loop replies with the shard's replication + /// offset AT registration — the exact point where live fan-out to + /// `tx` begins. The PSYNC task sends backlog catch-up bytes strictly + /// below this offset, closing the race where a write drained between + /// the catch-up read and registration reached neither leg (silent + /// replica gap). `None` = legacy fire-and-forget registration (the + /// multi-shard paths, redesigned in R2). + registered: Option>, }, /// Remove a replica's sender channel from this shard's fan-out list. /// Called when a replica disconnects or REPLICAOF NO ONE is executed. diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index f4e614988..199abc109 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2476,6 +2476,7 @@ pub(crate) fn handle_shard_message_shared( replica_id, tx, backlog_capacity, + registered, } => { // Lazy-init replication backlog on first replica registration (saves 1MB/shard). // The backlog is shared with PSYNC handlers via Arc>> on @@ -2489,6 +2490,18 @@ pub(crate) fn handle_shard_message_shared( } drop(guard); replica_txs.push((replica_id, tx)); + // Reply with the offset at which live fan-out begins. This runs + // synchronously between drains, so every fanout message queued + // BEFORE this registration has already advanced the offset, and + // every one after it will reach `tx` — the PSYNC task's catch-up + // read below this offset is therefore gap-free and overlap-free. + if let Some(reg_tx) = registered { + let offset = repl_state + .as_ref() + .map(|h| h.shard_offset(shard_id)) + .unwrap_or(0); + let _ = reg_tx.send(offset); + } } ShardMessage::UnregisterReplica { replica_id } => { replica_txs.retain(|(id, _)| *id != replica_id); diff --git a/tests/replication_streaming.rs b/tests/replication_streaming.rs index d3930541f..99f6643e5 100644 --- a/tests/replication_streaming.rs +++ b/tests/replication_streaming.rs @@ -519,3 +519,91 @@ fn replica_syncs_vector_index_defs_and_contents() { "FLUSHALL must keep index definitions on the replica: {list_after_flush}" ); } + +/// REPL-STREAM-03 (C1 regression guard): FT.CREATE traffic racing a replica +/// attach must never lose a definition. The master's PSYNC path registers the +/// replica BEFORE the backlog catch-up read and bounds that read to the +/// registration offset; before that fix, a def-mutation drained between the +/// catch-up read and registration reached neither the RDB, the catch-up, nor +/// the live stream — a silent gap. This test hammers FT.CREATE from a side +/// thread while the replica attaches mid-stream, then requires exact +/// index-list parity. Probabilistic per run, deterministic across CI history: +/// any hit is a real ordering regression. +#[test] +#[ignore] +fn replica_attach_races_live_ft_create() { + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + + let master_addr = "127.0.0.1:16720"; + let replica_addr = "127.0.0.1:16721"; + + let _master = Killer(start_moon(16720, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + + // Side thread: create rc_idx_0..rc_idx_29 with tiny gaps so creations + // interleave with every phase of the attach (handshake, RDB write, + // catch-up, registration). + const N: usize = 30; + let writer = std::thread::spawn(move || { + for i in 0..N { + let name = format!("rc_idx_{i}"); + let prefix = format!("rc{i}:"); + let created = send_resp( + "127.0.0.1:16720", + &[ + b"FT.CREATE", + name.as_bytes(), + b"ON", + b"HASH", + b"PREFIX", + b"1", + prefix.as_bytes(), + b"SCHEMA", + b"vec", + b"VECTOR", + b"HNSW", + b"6", + b"DIM", + b"4", + b"TYPE", + b"FLOAT32", + b"DISTANCE_METRIC", + b"L2", + ], + ); + assert!(created.contains("OK"), "FT.CREATE {name} failed: {created}"); + std::thread::sleep(Duration::from_millis(20)); + } + }); + + // Attach the replica mid-stream (~1/4 of the creations done). + std::thread::sleep(Duration::from_millis(150)); + let _replica = Killer(start_moon(16721, replica_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + let attach = send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16720"); + assert!(attach.starts_with("+OK"), "REPLICAOF failed: {attach}"); + + writer.join().expect("FT.CREATE writer thread panicked"); + + // Every index the master knows must reach the replica — via RDB aux, + // catch-up, or live stream; which leg is timing-dependent, parity is not. + let synced = wait_until(Duration::from_secs(15), || { + let list = send_resp(replica_addr, &[b"FT._LIST"]); + (0..N).all(|i| list.contains(&format!("rc_idx_{i}"))) + }); + let master_list = send_resp(master_addr, &[b"FT._LIST"]); + let replica_list = send_resp(replica_addr, &[b"FT._LIST"]); + assert!( + synced, + "replica lost FT.CREATE(s) during attach race.\nmaster: {master_list}\nreplica: {replica_list}" + ); +} From 1ea05f908b4cf46e2f4205117938ed5c8d0eac21 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 07:18:09 +0700 Subject: [PATCH 5/5] fix(test): add repl_backlog_size to tokio-gated ServerConfig initializers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #278 CI (Check + Check macOS) failed with E0063: the new `repl_backlog_size` ServerConfig field (--repl-backlog-size, default 1 MiB) was missing from four struct literals that only compile under --features runtime-tokio,jemalloc — invisible to default-feature builds: - tests/mq_integration.rs - tests/txn_kv_wiring.rs - tests/workspace_integration.rs (x2) All four now set `repl_backlog_size: 1024 * 1024` (the CLI default). Verified: `cargo check --tests --no-default-features --features runtime-tokio,jemalloc` and default-feature `cargo check --tests` both clean. author: Tin Dang --- tests/mq_integration.rs | 1 + tests/txn_kv_wiring.rs | 1 + tests/workspace_integration.rs | 2 ++ 3 files changed, 4 insertions(+) diff --git a/tests/mq_integration.rs b/tests/mq_integration.rs index 83015b94e..6f844d921 100644 --- a/tests/mq_integration.rs +++ b/tests/mq_integration.rs @@ -42,6 +42,7 @@ async fn start_mq_server(num_shards: usize) -> (u16, CancellationToken) { bind: "127.0.0.1".to_string(), port, tcp_backlog: 1024, + repl_backlog_size: 1024 * 1024, databases: 16, requirepass: None, db_maxmemory: Vec::new(), diff --git a/tests/txn_kv_wiring.rs b/tests/txn_kv_wiring.rs index 9ddd9e7ca..1ea1f88bb 100644 --- a/tests/txn_kv_wiring.rs +++ b/tests/txn_kv_wiring.rs @@ -48,6 +48,7 @@ async fn start_txn_server(num_shards: usize, persistence_dir: &str) -> (u16, Can bind: "127.0.0.1".to_string(), port, tcp_backlog: 1024, + repl_backlog_size: 1024 * 1024, databases: 16, requirepass: None, db_maxmemory: Vec::new(), diff --git a/tests/workspace_integration.rs b/tests/workspace_integration.rs index 906e793e1..8a6e7108e 100644 --- a/tests/workspace_integration.rs +++ b/tests/workspace_integration.rs @@ -35,6 +35,7 @@ async fn start_workspace_server(num_shards: usize) -> (u16, CancellationToken) { bind: "127.0.0.1".to_string(), port, tcp_backlog: 1024, + repl_backlog_size: 1024 * 1024, databases: 16, requirepass: None, appendonly: "no".to_string(), @@ -275,6 +276,7 @@ async fn start_workspace_server_with_auth( bind: "127.0.0.1".to_string(), port, tcp_backlog: 1024, + repl_backlog_size: 1024 * 1024, databases: 16, requirepass: Some(password), appendonly: "no".to_string(),