diff --git a/CHANGELOG.md b/CHANGELOG.md index e05f8ea4..a5bc007c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,117 @@ 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 + 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 + 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 1db28a82..b19b9b1c 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 07bf945d..36e683b1 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/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs index bdad5fa9..08b50a08 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,41 @@ 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> { + // 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); + 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 +735,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 new file mode 100644 index 00000000..5cb48a76 --- /dev/null +++ b/src/replication/apply.rs @@ -0,0 +1,707 @@ +//! 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 + ); + } + + // 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 + // 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 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.) +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 { + 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(); + } + 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!( + "replica snapshot load: no ShardSlice on this thread" + )), + } +} + +/// 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(); + 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 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(); + + 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::*; + 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/handshake.rs b/src/replication/handshake.rs index 52d0f89a..434122fc 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 608741a5..6f33db7b 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,15 @@ 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, + // 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); } @@ -498,6 +512,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 +529,15 @@ 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, + // 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); } @@ -579,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"))?; @@ -588,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. @@ -613,23 +639,71 @@ 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(); - 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, + ); }); - } + 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))?; @@ -639,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 } => { @@ -655,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; @@ -691,26 +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); - - // 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. - { - 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(), - }; - if prod.try_push(msg).is_err() { - anyhow::bail!("failed to push RegisterReplica onto shard 0 SPSC"); - } - } else { - anyhow::bail!("shard 0 producer missing"); + let (reg_tx, reg_rx) = crate::runtime::channel::mpsc_bounded::(1); + + // `--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/mod.rs b/src/replication/mod.rs index 65fbb9e7..bc63f290 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 843262e1..157bfa3c 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/src/replication/state.rs b/src/replication/state.rs index f75ec32d..f068e465 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, )); } } @@ -216,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"; @@ -361,6 +395,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 d2e3bdd6..22dcd2ae 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/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index a343c414..1000c382 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/src/server/conn/handler_monoio/ft.rs @@ -11,6 +11,38 @@ 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 +} + +/// 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`. @@ -763,6 +795,31 @@ 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) + && replication_fanout_active(ctx) + { + 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/server/embedded.rs b/src/server/embedded.rs index c33b9395..e011ce5a 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 6bfcdbae..81c40569 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 e50cbbeb..77b7f672 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -456,13 +456,33 @@ 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, + /// 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. 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 51e20bc4..199abc10 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2472,21 +2472,60 @@ 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, + registered, + } => { // 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)); + // 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); } + 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/mq_integration.rs b/tests/mq_integration.rs index 83015b94..6f844d92 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/observability_resident_bytes.rs b/tests/observability_resident_bytes.rs index 92f98bdd..677c2519 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 e48b3a0b..1e6e9f16 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={}", diff --git a/tests/replication_streaming.rs b/tests/replication_streaming.rs new file mode 100644 index 00000000..99f6643e --- /dev/null +++ b/tests/replication_streaming.rs @@ -0,0 +1,609 @@ +//! 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(); + } +} + +/// 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] +#[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" + ); +} + +/// 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}" + ); +} + +/// 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}" + ); +} diff --git a/tests/txn_kv_wiring.rs b/tests/txn_kv_wiring.rs index 9ddd9e7c..1ea1f88b 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 906e793e..8a6e7108 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(),