From 788fa031d35acf80a4d3c30bbaa4b24b4c09fca6 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 15:59:07 +0700 Subject: [PATCH 1/5] =?UTF-8?q?feat(replication):=20R2=20multi-shard=20mas?= =?UTF-8?q?ter=20PSYNC=20=E2=80=94=20merged=20RDB=20+=20per-record=20SELEC?= =?UTF-8?q?T=20framing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Masters running --shards N now serve full replication to a single-shard replica (task #20, RFC 1B). Previously PSYNC on a multi-shard master was rejected outright. Design: - ShardMessage::PrepareReplicaSync fans to every shard (self queue for the accepting shard, SPSC mesh + notifier for the rest). Each shard's arm serializes its keyspace slice to an RDB BODY, captures its shard offset, and registers the replica's live channel in ONE synchronous stretch on its own thread — per shard there is no window between "in the snapshot" and "streamed live", so no backlog catch-up leg exists and non-idempotent commands cannot double-apply. - The PSYNC task stitches bodies into ONE valid Redis-format RDB (redis_rdb::write_rdb_merged) and replies `+FULLRESYNC ` + one $ bulk — the replica's existing R0 loader is unchanged. Index defs ride once; graph content is sharded so the snapshot carries one moon-graph-store aux entry PER shard, and the replica imports all of them (install_graph_store_many / read_moon_aux_all). - Merged-wire db context: N shard threads feed one replica socket, so per-shard SELECT tracking cannot work. On multi-shard masters every db-scoped record is fused with its own `SELECT ` prefix as ONE record (one channel send, one backlog append pair, one offset advance) in both the cross-shard path (wal_append_and_fanout) and the local leg (record_local_write_db) — no interleave can split a SELECT from the write it frames. Gated on the replica-attach hint; single-shard masters keep the emit-on-change tracking. - Partial resync degrades to full: a single scalar offset cannot map back onto N per-shard backlogs. - R1 pieces carry over unchanged: overflow-kick (shared kicked flag across all N fan-out entries), REPLCONF ACK reader, WAIT (summed snapshot offset keeps total_offset - base == bytes-on-wire, so ACK math stays exact). Verification (red/green): - NEW tests/replication_multishard.rs — 6 e2e over real processes, written first and failing on main: 2/4/8-shard full resync + live convergence with INCR exactness, interleaved multi-db writers (5k keys x 2 dbs, leak asserts both directions), per-shard graph snapshot import, raw-handshake partial->full degradation with merged-RDB REDIS-magic assert. - NEW unit merged_multishard_rdb_round_trip (repeated SELECTDB sections + repeated graph aux entries load as one keyspace, CRC valid). - Regression: replication_streaming 7/7, replication_hardening 3/3, replication_graph 5/5, aof_multidb_kill9 4/4, lib 4115 (monoio) + 3316 (tokio), clippy -D warnings on both matrices, fmt clean. author: Tin Dang --- CHANGELOG.md | 41 ++ src/persistence/redis_rdb.rs | 127 ++++- src/replication/apply.rs | 23 +- src/replication/graph_sync.rs | 32 +- src/replication/master.rs | 205 +++++++- src/replication/state.rs | 9 + src/server/conn/handler_monoio/dispatch.rs | 13 +- src/server/conn/handler_monoio/ft.rs | 22 + src/shard/conn_accept.rs | 35 +- src/shard/dispatch.rs | 57 +++ src/shard/spsc_handler.rs | 122 ++++- tests/replication_multishard.rs | 528 +++++++++++++++++++++ 12 files changed, 1178 insertions(+), 36 deletions(-) create mode 100644 tests/replication_multishard.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index b3a6c910..a68e973f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added — R2: multi-shard master PSYNC (task #20, RFC 1B) + +A master running `--shards N` now serves full replication to a single-shard +replica — previously PSYNC was rejected with `-ERR PSYNC across multiple +shards is not yet supported`. + +- **Per-shard atomic snapshot legs.** A new `ShardMessage::PrepareReplicaSync` + fans out to every shard (own shard via the self queue, the rest over the + SPSC mesh). Each shard serializes its keyspace slice to an RDB *body*, + captures its replication offset, and registers the replica's live channel + in ONE synchronous stretch on its own thread — per shard, nothing can land + between "inside the snapshot" and "streamed live", so there is no backlog + catch-up leg and non-idempotent commands (INCR) can never double-apply. +- **One merged Redis-format RDB.** The PSYNC task stitches the per-shard + bodies into a single valid RDB (`redis_rdb::write_rdb_merged`: header + + bodies + EOF/CRC64) and answers `+FULLRESYNC <Σ shard offsets>` + with one `$` bulk — the replica's existing R0 loader needs no changes. + Index definitions ride once; graph content is sharded, so the snapshot + carries one `moon-graph-store` aux entry per shard and the replica imports + all of them (`install_graph_store_many`, `read_moon_aux_all`). +- **Per-record SELECT framing on the merged wire.** N shard threads feed one + replica socket, so a shared "current db" context cannot exist: on + multi-shard masters every db-scoped record is fused with its own + `SELECT ` prefix (single channel send / backlog append pair / offset + advance) — no cross-shard interleave can split a SELECT from the write it + frames. Gated on the replica-attach hint; single-shard masters keep the + cheaper emit-on-change tracking. +- **Partial resync degrades to full.** A replica's single scalar offset + cannot be mapped back onto N per-shard backlogs, so a multi-shard master + answers every PSYNC (any replid/offset) with `+FULLRESYNC`. +- Overflow-kick (task #35), `REPLCONF ACK`, and `WAIT` all carry over: the + summed snapshot offset keeps `total_offset - base == bytes on wire`, so + WAIT/ACK math stays exact on multi-shard masters. +- New e2e suite `tests/replication_multishard.rs`: 2/4/8-shard full resync + + live-stream convergence with INCR exactness, interleaved multi-db writers + with db-leak asserts, per-shard graph snapshot import, and the + partial→full degradation handshake. +- Known limitation (unchanged from R1): master-side PSYNC requires + `runtime-monoio` (the default); the tokio build has no master-side PSYNC + intercept. Multi-shard *replicas* remain unsupported (`--shards 1`). + ### Fixed — consistency/durability defects caught by the R1 gates (task #35) Three pre-existing data-integrity bugs surfaced by the new load/kill-9 gates diff --git a/src/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs index 5003fbfd..72bb5630 100644 --- a/src/persistence/redis_rdb.rs +++ b/src/persistence/redis_rdb.rs @@ -462,7 +462,17 @@ pub fn write_rdb_refs_with_moon_aux( for (key, value) in moon_aux { write_aux(buf, key, value); } + write_rdb_body_refs(databases, buf); + write_rdb_footer(buf); +} +/// Body-only writer: per-DB SELECTDB/RESIZEDB sections and entries — NO +/// header, aux, EOF, or CRC. Building block for [`write_rdb_merged`]: a +/// multi-shard master has each shard serialize its own keyspace slice with +/// this, then stitches the bodies into ONE valid RDB. Repeated SELECTDB +/// opcodes for the same db index (one per shard) are valid RDB — loaders +/// treat SELECTDB as "switch current db" and accumulate entries. +pub fn write_rdb_body_refs(databases: &[&Database], buf: &mut Vec) { let now_ms = current_time_ms(); for (db_idx, db) in databases.iter().enumerate() { @@ -486,7 +496,7 @@ pub fn write_rdb_refs_with_moon_aux( buf.push(RDB_OPCODE_SELECTDB); write_length(buf, db_idx as u64); - // RESIZEDB + // RESIZEDB (per-shard slice size — a presize hint only, safe to repeat) buf.push(RDB_OPCODE_RESIZEDB); write_length(buf, live.len() as u64); write_length(buf, expires_count as u64); @@ -495,7 +505,21 @@ pub fn write_rdb_refs_with_moon_aux( write_rdb_entry(buf, key.as_bytes(), entry, base_ts); } } +} +/// Assemble ONE valid Redis-format RDB from pre-serialized per-shard bodies +/// (each produced by [`write_rdb_body_refs`]): header + moon aux fields + +/// concatenated bodies + EOF/CRC footer. The multi-shard PSYNC full resync +/// (R2, task #20) sends this as a single `$` bulk so a single-shard +/// replica loads it with the unchanged R0 path. +pub fn write_rdb_merged(moon_aux: &[(&[u8], &[u8])], bodies: &[Vec], buf: &mut Vec) { + write_rdb_header(buf); + for (key, value) in moon_aux { + write_aux(buf, key, value); + } + for body in bodies { + buf.extend_from_slice(body); + } write_rdb_footer(buf); } @@ -551,6 +575,39 @@ pub fn read_moon_aux(data: &[u8], key: &[u8]) -> Option> { } } +/// Like [`read_moon_aux`] but collects EVERY occurrence of `key` in the +/// header aux block, in write order. A merged multi-shard snapshot +/// ([`write_rdb_merged`]) carries one `moon-graph-store` aux entry PER shard — +/// graph content is sharded, so the replica must import all of them, not just +/// the first. Returns an empty Vec for foreign/truncated buffers or when the +/// key never appears. +pub fn read_moon_aux_all(data: &[u8], key: &[u8]) -> Vec> { + let mut out = Vec::new(); + if data.len() < 9 || &data[..5] != REDIS_RDB_MAGIC || &data[5..9] != REDIS_RDB_VERSION { + return out; + } + 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 out; + } + if opcode[0] != RDB_OPCODE_AUX { + return out; + } + let Ok(k) = read_redis_string(&mut cursor) else { + return out; + }; + let Ok(v) = read_redis_string(&mut cursor) else { + return out; + }; + if k == key { + out.push(v); + } + } +} + /// Load an RDB file in Redis format into the provided databases. /// /// Verifies magic bytes, version, and CRC64 checksum. @@ -767,6 +824,74 @@ mod tests { assert_eq!(loaded, 0); } + /// R2 (task #20): a merged multi-shard snapshot — per-shard bodies with + /// REPEATED SELECTDB sections for the same db — must load as one keyspace, + /// and repeated graph aux entries must all be readable in write order. + #[test] + fn merged_multishard_rdb_round_trip() { + // "Shard 0": keys in db0 and db2. "Shard 1": different keys, same dbs. + let mk = |pairs: &[(usize, &str, &str)]| { + let mut dbs = vec![Database::new(), Database::new(), Database::new()]; + for (db_idx, k, v) in pairs { + dbs[*db_idx].set( + Bytes::copy_from_slice(k.as_bytes()), + Entry::new_string(Bytes::copy_from_slice(v.as_bytes())), + ); + } + dbs + }; + let shard0 = mk(&[(0, "a", "1"), (2, "c", "3")]); + let shard1 = mk(&[(0, "b", "2"), (2, "d", "4")]); + + let mut body0 = Vec::new(); + write_rdb_body_refs(&shard0.iter().collect::>(), &mut body0); + let mut body1 = Vec::new(); + write_rdb_body_refs(&shard1.iter().collect::>(), &mut body1); + + let mut merged = Vec::new(); + write_rdb_merged( + &[ + (MOON_AUX_VECTOR_DEFS, b"vec-defs"), + (MOON_AUX_GRAPH_STORE, b"graph-shard-0"), + (MOON_AUX_GRAPH_STORE, b"graph-shard-1"), + ], + &[body0, body1], + &mut merged, + ); + + // Single-occurrence reader still finds the first entry of each key. + assert_eq!( + read_moon_aux(&merged, MOON_AUX_VECTOR_DEFS).as_deref(), + Some(&b"vec-defs"[..]) + ); + // All-occurrences reader returns every shard's graph blob in order. + assert_eq!( + read_moon_aux_all(&merged, MOON_AUX_GRAPH_STORE), + vec![b"graph-shard-0".to_vec(), b"graph-shard-1".to_vec()] + ); + assert!(read_moon_aux_all(&merged, b"moon-unknown").is_empty()); + + // The merged blob is ONE valid RDB (magic/version/CRC) whose repeated + // SELECTDB sections accumulate into a single keyspace. + let mut loaded = vec![Database::new(), Database::new(), Database::new()]; + let n = load_rdb(&mut loaded, &merged).expect("merged RDB must load"); + assert_eq!(n, 4); + let get = |db: &Database, k: &str| { + db.data() + .iter() + .find(|(key, _)| key.as_bytes() == k.as_bytes()) + .map(|(_, e)| match e.as_redis_value() { + RedisValueRef::String(s) => s.to_vec(), + _ => panic!("expected string"), + }) + }; + assert_eq!(get(&loaded[0], "a").as_deref(), Some(&b"1"[..])); + assert_eq!(get(&loaded[0], "b").as_deref(), Some(&b"2"[..])); + assert_eq!(get(&loaded[2], "c").as_deref(), Some(&b"3"[..])); + assert_eq!(get(&loaded[2], "d").as_deref(), Some(&b"4"[..])); + assert!(get(&loaded[1], "a").is_none()); + } + #[test] fn read_moon_aux_rejects_foreign_and_truncated_buffers() { assert_eq!(read_moon_aux(b"", MOON_AUX_VECTOR_DEFS), None); diff --git a/src/replication/apply.rs b/src/replication/apply.rs index 5277d16d..98647bd0 100644 --- a/src/replication/apply.rs +++ b/src/replication/apply.rs @@ -460,8 +460,11 @@ pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { // 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); + // R2 (task #20): a multi-shard master's merged snapshot carries one + // graph-store aux entry PER shard (graph content is sharded) — collect + // them all; a single-shard snapshot yields exactly one. #[cfg(feature = "graph")] - let graph_blob = redis_rdb::read_moon_aux(rdb, redis_rdb::MOON_AUX_GRAPH_STORE); + let graph_blobs = redis_rdb::read_moon_aux_all(rdb, redis_rdb::MOON_AUX_GRAPH_STORE); match crate::shard::slice::try_with_shard(|s| { for db in s.databases.iter_mut() { db.clear(); @@ -472,13 +475,19 @@ pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { // (authoritative replace — an EMPTY blob drops replica-local graphs; // an ABSENT aux means a pre-graph-sync master, warn-and-keep). #[cfg(feature = "graph")] - match graph_blob.as_deref() { - Some(blob) => { - match crate::replication::graph_sync::install_graph_store(&mut s.graph_store, blob) - { + match &graph_blobs[..] { + blobs if !blobs.is_empty() => { + match crate::replication::graph_sync::install_graph_store_many( + &mut s.graph_store, + blobs, + ) { Some(n) => { if n > 0 { - tracing::info!("replica snapshot: installed {} graph(s)", n); + tracing::info!( + "replica snapshot: installed {} graph(s) from {} shard blob(s)", + n, + blobs.len() + ); } } None => { @@ -488,7 +497,7 @@ pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { } } } - None => { + _ => { if s.graph_store.graph_count() > 0 { tracing::warn!( "replica snapshot carried no graph-store aux but {} local graph(s) \ diff --git a/src/replication/graph_sync.rs b/src/replication/graph_sync.rs index 2ad85521..7261a5ca 100644 --- a/src/replication/graph_sync.rs +++ b/src/replication/graph_sync.rs @@ -183,15 +183,39 @@ pub fn export_graph_store(store: &mut GraphStore) -> Vec { /// (store is left in whatever partial state was reached — the caller aborts /// the sync and the replica retries with a fresh full resync). pub fn install_graph_store(store: &mut GraphStore, blob: &[u8]) -> Option { - let mut cur = Cursor { data: blob, pos: 0 }; - if cur.u8()? != FORMAT_VERSION { - return None; + drop_all_local_graphs(store); + install_graphs_additive(store, blob) +} + +/// R2 (task #20): install a MULTI-SHARD snapshot's graph blobs — one per +/// master shard (`read_moon_aux_all` order). Authoritative replace happens +/// ONCE, then every blob installs additively. Graph names are disjoint across +/// blobs (each graph lives on exactly one master shard); a duplicate name is +/// malformed input and fails the install (`None`). +pub fn install_graph_store_many(store: &mut GraphStore, blobs: &[Vec]) -> Option { + drop_all_local_graphs(store); + let mut total = 0usize; + for blob in blobs { + total += install_graphs_additive(store, blob)?; } - // Authoritative replace: drop everything local first. + Some(total) +} + +/// Authoritative replace leg shared by both install entry points. +fn drop_all_local_graphs(store: &mut GraphStore) { let local: Vec = store.list_graphs().into_iter().cloned().collect(); for name in local { let _ = store.drop_graph(&name); } +} + +/// Decode one export blob and create its graphs on top of whatever the store +/// already holds. Callers handle the drop-local leg. +fn install_graphs_additive(store: &mut GraphStore, blob: &[u8]) -> Option { + let mut cur = Cursor { data: blob, pos: 0 }; + if cur.u8()? != FORMAT_VERSION { + return None; + } let graph_count = cur.u32()? as usize; for _ in 0..graph_count { diff --git a/src/replication/master.rs b/src/replication/master.rs index 760a6ae5..7c609db3 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -799,6 +799,199 @@ pub async fn handle_psync_inline_single_shard( Ok(()) } +/// R2 (task #20): multi-shard master full resync — RFC 1B. +/// +/// Every multi-shard PSYNC is answered with a FULL resync: the replica's +/// single scalar offset cannot be mapped back onto N per-shard backlogs, so +/// `+CONTINUE` is never offered (the client's requested replid/offset are +/// accepted but ignored). Flow: +/// +/// 1. Fan a [`ShardMessage::PrepareReplicaSync`] to every shard — its own +/// via the self queue (the SPSC mesh has no self-loop), the rest over +/// `dispatch_tx` + notifier. Each shard's arm snapshots its keyspace +/// slice to an RDB *body*, captures its shard offset, and registers the +/// replica's live channel — all in ONE synchronous stretch, so per shard +/// nothing can land between "in the snapshot" and "streamed live". +/// 2. Stitch the bodies into ONE Redis-format RDB (`write_rdb_merged`) — +/// index definitions once, one graph blob PER shard — and send +/// `+FULLRESYNC <Σ shard offsets>` + the `$` bulk. A +/// single-shard replica loads it through the unchanged R0 path. +/// 3. Drain the merged live channel onto the socket (same drain + ACK +/// reader + overflow-kick loop as the single-shard path). +/// +/// The summed offset is consistent even though shards capture at different +/// times: each shard's live records begin exactly at its own captured offset, +/// so bytes-on-wire past the FULLRESYNC base always equal +/// `total_offset() - base` — which keeps WAIT/ACK math exact. +#[cfg(feature = "runtime-monoio")] +#[allow(clippy::too_many_arguments)] +pub async fn handle_psync_inline_multi_shard( + mut stream: monoio::net::TcpStream, + repl_state: Arc>, + replica_addr: std::net::SocketAddr, + dispatch_tx: Rc>>>, + spsc_notifiers: Vec>, + self_shard_id: usize, + num_shards: usize, +) -> anyhow::Result<()> { + use monoio::io::AsyncWriteRentExt; + use ringbuf::traits::Producer; + + let (repl_id, backlog_capacity) = { + let rs = repl_state + .read() + .map_err(|_| anyhow::anyhow!("lock poisoned"))?; + (rs.repl_id.clone(), rs.backlog_capacity) + }; + + let replica_id = next_replica_id(); + // One merged live channel: every shard's fan-out entry holds a clone of + // `tx`; the drain loop below pumps `rx` onto the socket. Capacity choice + // matches the single-shard path (task #35) — shared across all shards. + let (tx, rx) = crate::runtime::channel::mpsc_bounded::(16384); + let kicked = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + + // Fan out PrepareReplicaSync; replies are bounded(1) channels usable + // across threads. + let mut reply_rxs = Vec::with_capacity(num_shards); + for shard in 0..num_shards { + let (reply_tx, reply_rx) = + crate::runtime::channel::mpsc_bounded::(1); + let mut msg = crate::shard::dispatch::ShardMessage::PrepareReplicaSync(Box::new( + crate::shard::dispatch::PrepareReplicaSyncPayload { + replica_id, + tx: tx.clone(), + kicked: kicked.clone(), + backlog_capacity, + reply_tx, + }, + )); + if shard == self_shard_id { + crate::shard::self_msg::push(msg); + } else { + let idx = crate::shard::mesh::ChannelMesh::target_index(self_shard_id, shard); + // The SPSC ring can be transiently full under load — bounded retry, + // then abort loudly (the replica reconnects and retries the sync). + let mut attempts = 0u32; + loop { + let res = { dispatch_tx.borrow_mut()[idx].try_push(msg) }; + match res { + Ok(()) => { + spsc_notifiers[shard].notify_one(); + break; + } + Err(back) => { + msg = back; + attempts += 1; + if attempts > 5_000 { + anyhow::bail!( + "shard {} SPSC full for >5s during PSYNC fan-out; aborting sync", + shard + ); + } + monoio::time::sleep(std::time::Duration::from_millis(1)).await; + } + } + } + } + reply_rxs.push((shard, reply_rx)); + } + + // Collect every shard's leg. A dropped reply means that shard could not + // prepare (or we raced shutdown) — abort; already-registered shards clean + // up passively when `rx` drops (their next fan-out send sees Disconnected). + let mut bodies: Vec> = Vec::with_capacity(num_shards); + let mut snapshot_offset: u64 = 0; + let mut vector_defs: Option> = None; + let mut text_defs: Option> = None; + #[cfg(feature = "graph")] + let mut graph_blobs: Vec> = Vec::with_capacity(num_shards); + for (shard, reply_rx) in reply_rxs { + let prepared = reply_rx + .recv_async() + .await + .map_err(|_| anyhow::anyhow!("shard {} dropped its PrepareReplicaSync reply", shard))?; + snapshot_offset += prepared.shard_offset; + // Index definitions are keyspace-global and identical on every shard — + // keep the first non-empty copy. + if vector_defs.is_none() { + vector_defs = prepared.vector_defs; + } + if text_defs.is_none() { + text_defs = prepared.text_defs; + } + #[cfg(feature = "graph")] + graph_blobs.push(prepared.graph_blob); + bodies.push(prepared.rdb_body); + } + + // Stitch ONE valid Redis-format RDB. Graph content is sharded: one aux + // entry per shard, imported in order by the replica (`read_moon_aux_all`). + let mut moon_aux: Vec<(&[u8], &[u8])> = Vec::new(); + if let Some(v) = &vector_defs { + moon_aux.push((crate::persistence::redis_rdb::MOON_AUX_VECTOR_DEFS, &v[..])); + } + if let Some(t) = &text_defs { + moon_aux.push((crate::persistence::redis_rdb::MOON_AUX_TEXT_DEFS, &t[..])); + } + #[cfg(feature = "graph")] + for blob in &graph_blobs { + moon_aux.push(( + crate::persistence::redis_rdb::MOON_AUX_GRAPH_STORE, + &blob[..], + )); + } + let mut rdb_buf: Vec = Vec::new(); + crate::persistence::redis_rdb::write_rdb_merged(&moon_aux, &bodies, &mut rdb_buf); + info!( + replica_id, + num_shards, + snapshot_offset, + rdb_bytes = rdb_buf.len(), + "multi-shard full resync prepared" + ); + + 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))?; + let (wr, _) = stream.write_all(rdb_buf).await; + wr.map_err(|e| anyhow::anyhow!(e))?; + // No backlog catch-up leg: each shard's registration IS its snapshot + // point (same synchronous stretch), so live fan-out already covers every + // byte past `snapshot_offset`. + + let reg = InlineReplicaRegistration { + replica_id, + tx, + rx, + // The multi-shard path has no registration-offset reply channel — + // offsets arrived in the PrepareReplicaSync replies. + reg_rx: crate::runtime::channel::mpsc_bounded::(1).1, + kicked, + }; + let drain_result = + drain_replica_inline_single_shard(reg, replica_addr, stream, repl_state).await; + // Best-effort explicit unregister on the REMOTE shards (the drain already + // self-queued UnregisterReplica for this shard). A full ring is fine — + // dropping `rx` above already flipped every sender to Disconnected, which + // the next fan-out send prunes. + for shard in 0..num_shards { + if shard == self_shard_id { + continue; + } + let idx = crate::shard::mesh::ChannelMesh::target_index(self_shard_id, shard); + let pushed = dispatch_tx.borrow_mut()[idx] + .try_push(crate::shard::dispatch::ShardMessage::UnregisterReplica { replica_id }); + if pushed.is_ok() { + spsc_notifiers[shard].notify_one(); + } + } + drain_result +} + /// 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 @@ -864,14 +1057,18 @@ struct InlineReplicaRegistration { /// 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 next_replica_id() -> u64 { + use std::sync::atomic::Ordering; + static NEXT_REPLICA_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + NEXT_REPLICA_ID.fetch_add(1, Ordering::Relaxed) +} + #[cfg(feature = "runtime-monoio")] fn push_register_replica_inline( repl_state: &Arc>, ) -> anyhow::Result { - use std::sync::atomic::Ordering; - - static NEXT_REPLICA_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); - let replica_id = NEXT_REPLICA_ID.fetch_add(1, Ordering::Relaxed); + let replica_id = next_replica_id(); // 16384 records (task #35): 1024 overflowed within one pipelined burst on // the same host — every overflow now KICKS the replica into a resync, so diff --git a/src/replication/state.rs b/src/replication/state.rs index 3ec31636..c3cbf737 100644 --- a/src/replication/state.rs +++ b/src/replication/state.rs @@ -276,6 +276,15 @@ impl OffsetHandle { .map(|o| o.load(Ordering::Relaxed)) .unwrap_or(0) } + + /// Number of shards this handle tracks. R2 (task #20): `> 1` switches the + /// replica-stream serialization to per-record `SELECT` framing — N shard + /// threads feed ONE replica wire, so a shared "current db" context cannot + /// exist and every db-scoped record must carry its own. + #[inline] + pub fn num_shards(&self) -> usize { + self.shard_offsets.len() + } } const ZEROED_ID: &str = "0000000000000000000000000000000000000000"; diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 706e1e9a..7d53ec14 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -541,10 +541,9 @@ pub(super) fn try_handle_cdc_read( /// loop and returns the stream so the master replication driver can take over. /// /// Returns `None` for non-PSYNC commands. -/// Returns `Some((..))` only when num_shards == 1 (the supported topology). -/// For multi-shard topologies, pushes a clear error and returns `None` -/// (consumed via `responses`); the caller treats it like any other command -/// reply and continues — the replica will see the error and give up. +/// Returns `Some((..))` for every accepted PSYNC — the accept loop routes the +/// hijacked stream to the single-shard inline handler or, at num_shards > 1, +/// to the R2 multi-shard handler (`handle_psync_inline_multi_shard`). pub(super) fn try_handle_psync( cmd: &[u8], cmd_args: &[Frame], @@ -560,12 +559,6 @@ pub(super) fn try_handle_psync( ))); return None; } - if ctx.num_shards != 1 { - responses.push(Frame::Error(Bytes::from_static( - b"ERR PSYNC across multiple shards is not yet supported (use --shards 1 on the master)", - ))); - return None; - } let Some(ref rs) = ctx.repl_state else { responses.push(Frame::Error(Bytes::from_static( b"ERR replication is not enabled on this server", diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index 524a8851..89aa3f15 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/src/server/conn/handler_monoio/ft.rs @@ -103,6 +103,28 @@ pub(super) fn record_local_write(ctx: &ConnectionContext, bytes: Bytes) { /// db-agnostic record can never silently strand a stale context for the next /// KV write. pub(super) fn record_local_write_db(ctx: &ConnectionContext, db: usize, bytes: Bytes) { + // R2 (task #20): multi-shard masters merge N shard streams onto one + // replica wire, so the per-shard `stream_db` context tracking below is + // meaningless there — another shard may have moved the wire's db context + // between any two of this shard's records. Instead EVERY db-scoped record + // is framed with its own `SELECT ` prefix, fused into ONE record so + // no cross-shard interleave can split them. Gated on the fanout hint: a + // multi-shard master that never had a replica attach pays nothing, and + // the hint flips (process-global, then re-asserted by each shard's + // `PrepareReplicaSync` arm BEFORE that shard's snapshot offset is + // captured) before any of this shard's records can reach a wire. + if ctx.num_shards > 1 { + if crate::replication::state::fanout_hint_active() { + let select = serialize_select(db); + let mut combined = Vec::with_capacity(select.len() + bytes.len()); + combined.extend_from_slice(&select); + combined.extend_from_slice(&bytes); + record_local_write(ctx, Bytes::from(combined)); + } else { + record_local_write(ctx, bytes); + } + return; + } let needs_select = ctx.repl_state.as_ref().is_some_and(|rs| { rs.read().is_ok_and(|g| { g.stream_db.get(ctx.shard_id).is_some_and(|slot| { diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index 79850b18..eec250c9 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -757,6 +757,12 @@ pub(crate) fn spawn_monoio_connection( _hijacked_psync = true; let repl_state_clone = conn_ctx.repl_state.clone(); let shard_databases_clone = conn_ctx.shard_databases.clone(); + // R2 (task #20): the multi-shard handler fans + // PrepareReplicaSync over the SPSC mesh. + let psync_num_shards = conn_ctx.num_shards; + let psync_dispatch_tx = conn_ctx.dispatch_tx.clone(); + let psync_notifiers = conn_ctx.spsc_notifiers.clone(); + let psync_shard_id = conn_ctx.shard_id; let parsed_addr: std::net::SocketAddr = hp_peer .parse() .unwrap_or_else(|_| std::net::SocketAddr::from(([0, 0, 0, 0], 0))); @@ -764,14 +770,27 @@ pub(crate) fn spawn_monoio_connection( let client_offset_v = *client_offset; monoio::spawn(async move { if let Some(rs) = repl_state_clone { - if let Err(e) = crate::replication::master::handle_psync_inline_single_shard( - &client_repl_id_owned, - client_offset_v, - stream, - rs, - shard_databases_clone, - parsed_addr, - ).await { + let res = if psync_num_shards > 1 { + crate::replication::master::handle_psync_inline_multi_shard( + stream, + rs, + parsed_addr, + psync_dispatch_tx, + psync_notifiers, + psync_shard_id, + psync_num_shards, + ).await + } else { + crate::replication::master::handle_psync_inline_single_shard( + &client_repl_id_owned, + client_offset_v, + stream, + rs, + shard_databases_clone, + parsed_addr, + ).await + }; + if let Err(e) = res { tracing::warn!("PSYNC handler exited: {}", e); } } else { diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index c2d67a3e..827d6e4d 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -422,6 +422,47 @@ pub type ReplicaFanout = ( std::sync::Arc, ); +/// Payload of [`ShardMessage::PrepareReplicaSync`] (R2 multi-shard PSYNC). +pub struct PrepareReplicaSyncPayload { + /// Master-side replica id — same id on every shard's fan-out entry. + pub replica_id: u64, + /// The replica's ONE live channel: every shard pushes its records here + /// and the PSYNC drain task pumps them onto the socket (merged wire). + pub tx: channel::MpscSender, + /// Shared overflow disconnect signal — any shard's fan-out overflow + /// kicks the replica into a resync (see `RegisterReplica::kicked`). + pub kicked: std::sync::Arc, + /// `--repl-backlog-size`, for the lazy backlog fallback-init. + pub backlog_capacity: usize, + /// Reply channel (bounded 1). Cross-thread capable — remote shards send + /// their prepared leg back to the PSYNC connection task. + pub reply_tx: channel::MpscSender, +} + +/// One shard's prepared full-resync leg — the reply to +/// [`ShardMessage::PrepareReplicaSync`]. +pub struct PreparedShardSync { + /// RDB *body* (SELECTDB/RESIZEDB/entry sections only) of this shard's + /// keyspace slice — see `redis_rdb::write_rdb_body_refs`. The PSYNC task + /// stitches all shards' bodies into one valid RDB via + /// `redis_rdb::write_rdb_merged`. + pub rdb_body: Vec, + /// This shard's replication offset at capture. Live fan-out to the + /// replica begins exactly here (same synchronous stretch). + pub shard_offset: u64, + /// Vector index definitions (index_persist v5 sidecar bytes) — defs are + /// keyspace-global and identical on every shard; the stitcher uses shard + /// 0's copy. `None` when no vector indexes exist. + pub vector_defs: Option>, + /// Text index definitions; same convention as `vector_defs`. + pub text_defs: Option>, + /// This shard's graph-store snapshot blob. Graph CONTENT is sharded, so + /// the stitcher writes one `moon-graph-store` aux entry PER shard and the + /// replica imports all of them (`read_moon_aux_all`). + #[cfg(feature = "graph")] + pub graph_blob: Vec, +} + /// Messages sent to a shard via SPSC channels from the connection layer /// or from other shards for cross-shard operations. pub enum ShardMessage { @@ -503,6 +544,22 @@ pub enum ShardMessage { /// Remove a replica's sender channel from this shard's fan-out list. /// Called when a replica disconnects or REPLICAOF NO ONE is executed. UnregisterReplica { replica_id: u64 }, + /// R2 (task #20): one shard's leg of a MULTI-SHARD full resync. + /// + /// The PSYNC connection task fans this to every shard (its own via the + /// self queue, the rest over the SPSC mesh). Each shard's arm runs the + /// whole leg in ONE synchronous stretch on its own thread — serialize its + /// keyspace slice to an RDB *body*, capture its shard replication offset, + /// and register `(replica_id, tx, kicked)` for live fan-out — so per + /// shard there is no window between "state captured" and "live stream + /// begins": every mutation is either inside the RDB body (offset already + /// counted below the captured offset) or delivered live. No backlog + /// catch-up leg exists on this path, and the `+FULLRESYNC` offset is the + /// sum of the per-shard captured offsets. + /// + /// Boxed: the payload carries channels + capacity fields well past the + /// enum's inline size budget. + PrepareReplicaSync(Box), /// Deliver an already-RECORDED local write to the live replica streams. /// /// The producing thread (`replication::record_local_write`) has ALREADY diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index f5d572d4..5332cfb0 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2584,6 +2584,96 @@ pub(crate) fn handle_shard_message_shared( ShardMessage::UnregisterReplica { replica_id } => { replica_txs.retain(|(id, _, _)| *id != replica_id); } + ShardMessage::PrepareReplicaSync(payload) => { + // R2 (task #20): this shard's leg of a multi-shard full resync. + // The ENTIRE leg — RDB body serialization, offset capture, live + // fan-out registration — runs in this one synchronous stretch on + // the shard's own thread, so no mutation can slip between "inside + // the snapshot" and "delivered live" (the same atomicity argument + // as `handle_psync_inline_single_shard`, applied per shard). + let crate::shard::dispatch::PrepareReplicaSyncPayload { + replica_id, + tx, + kicked, + backlog_capacity, + reply_tx, + } = *payload; + crate::replication::state::mark_fanout_active(); + // Lazy backlog init (offset accounting parity with RegisterReplica; + // the backlog itself is not replayed on this path — multi-shard + // PSYNC always answers with a full resync). + { + let mut guard = repl_backlog.lock(); + if guard.is_none() { + let offset = repl_state + .as_ref() + .map(|h| h.shard_offset(shard_id)) + .unwrap_or(0); + *guard = Some(ReplicationBacklog::new_at(backlog_capacity, offset)); + } + } + let started = std::time::Instant::now(); + let mut rdb_body: Vec = Vec::new(); + let mut vector_defs: Option> = None; + let mut text_defs: Option> = None; + #[cfg(feature = "graph")] + let mut graph_blob: Vec = Vec::new(); + crate::shard::slice::with_shard(|s| { + let refs: Vec<&crate::storage::Database> = s.databases.iter().collect(); + crate::persistence::redis_rdb::write_rdb_body_refs(&refs, &mut rdb_body); + // Index DEFINITIONS ride as moon aux (same as the single-shard + // path); contents stay in sync via the live stream + backfill. + let pairs = s.vector_store.collect_index_metas_with_weights(); + if !pairs.is_empty() { + vector_defs = Some(crate::vector::index_persist::serialize_index_metas_v5( + &pairs, + )); + } + let metas = s.text_store.collect_index_metas(); + if !metas.is_empty() { + text_defs = Some(crate::text::index_persist::serialize_text_index_metas( + &metas, + )); + } + #[cfg(feature = "graph")] + { + graph_blob = + crate::replication::graph_sync::export_graph_store(&mut s.graph_store); + } + }); + let shard_offset = repl_state + .as_ref() + .map(|h| h.shard_offset(shard_id)) + .unwrap_or(0); + replica_txs.push((replica_id, tx, kicked)); + tracing::debug!( + shard_id, + replica_id, + body_bytes = rdb_body.len(), + shard_offset, + elapsed_ms = started.elapsed().as_millis() as u64, + "prepared multi-shard replica sync leg" + ); + let prepared = crate::shard::dispatch::PreparedShardSync { + rdb_body, + shard_offset, + vector_defs, + text_defs, + #[cfg(feature = "graph")] + graph_blob, + }; + if reply_tx.try_send(prepared).is_err() { + // The PSYNC task is gone (replica dropped mid-handshake) — + // undo the registration so this shard doesn't fan out to a + // channel nobody drains. + replica_txs.retain(|(id, _, _)| *id != replica_id); + tracing::warn!( + shard_id, + replica_id, + "PrepareReplicaSync reply dropped — replica disconnected before sync" + ); + } + } ShardMessage::ReplicaLiveFanout { bytes } => { // Live-delivery leg ONLY: backlog append + offset advance already // happened synchronously at write time on this same thread @@ -3471,6 +3561,20 @@ pub(crate) fn wal_append_and_fanout( ); } } + // R2 (task #20): on a multi-shard master every db-scoped record on the + // replica wire carries its OWN `SELECT ` prefix. N shard threads feed + // one merged wire, so a shared "current db" context cannot exist — and + // the prefix+payload must travel as ONE record (one channel send, one + // backlog append pair, one offset advance) so no cross-shard interleave + // can split a SELECT from the write it frames. Single-shard masters keep + // the emit-on-change tracking in `record_local_write_db` (this fan-out + // leg only fires for cross-shard dispatch, which needs num_shards > 1). + let select_prefix: Option = + if !replica_txs.is_empty() && repl_state.as_ref().is_some_and(|h| h.num_shards() > 1) { + Some(crate::persistence::aof::serialize_select_record(db)) + } else { + None + }; // 2. Replication backlog (in-memory circular buffer for partial resync). // // The backlog is shared via Arc>> with PSYNC handlers. @@ -3480,6 +3584,9 @@ pub(crate) fn wal_append_and_fanout( // acquire per WAL flush (typically once per 1ms tick batch, NOT per write). let mut guard = repl_backlog.lock(); if let Some(backlog) = guard.as_mut() { + if let Some(prefix) = &select_prefix { + backlog.append(prefix); + } backlog.append(data); } drop(guard); @@ -3487,13 +3594,24 @@ pub(crate) fn wal_append_and_fanout( // QW3 (2026-06 review finding 1.4): `repl_state` is a lock-free // OffsetHandle cloned out of `RwLock` once at shard // startup — the per-write advance no longer read-locks the RwLock. + // The SELECT prefix counts too: offset accounting must equal the bytes + // the replica receives, or WAIT/ACK math diverges. if let Some(offsets) = repl_state { - offsets.increment_shard_offset(shard_id, data.len() as u64); + let prefix_len = select_prefix.as_ref().map_or(0, |p| p.len()); + offsets.increment_shard_offset(shard_id, (prefix_len + data.len()) as u64); } // 4. Fan-out to replica sender tasks (non-blocking: a replica whose // channel is FULL is kicked to resync — see `fanout_send_or_kick`). if !replica_txs.is_empty() { - let bytes = bytes::Bytes::copy_from_slice(data); + let bytes = match &select_prefix { + Some(prefix) => { + let mut combined = Vec::with_capacity(prefix.len() + data.len()); + combined.extend_from_slice(prefix); + combined.extend_from_slice(data); + bytes::Bytes::from(combined) + } + None => bytes::Bytes::copy_from_slice(data), + }; fanout_send_or_kick(replica_txs, &bytes); } // 5. Per-shard AOF pool (FIX-W1-2): route to the owning shard's writer. diff --git a/tests/replication_multishard.rs b/tests/replication_multishard.rs new file mode 100644 index 00000000..df7d8bec --- /dev/null +++ b/tests/replication_multishard.rs @@ -0,0 +1,528 @@ +//! R2 acceptance: MULTI-SHARD master PSYNC (task #20, RFC 1B). +//! +//! A master running `--shards N` (N > 1) must serve a full resync to a +//! single-shard replica: one merged Redis-format RDB snapshot followed by the +//! merged live command stream from all N shards. Before R2 the master answered +//! `-ERR PSYNC across multiple shards is not yet supported`. +//! +//! Black-box tests over real `moon` processes; `#[ignore]`d like the other +//! replication suites: +//! +//! ```text +//! MOON_BIN=./target/release/moon \ +//! cargo test --test replication_multishard -- --ignored --nocapture +//! ``` + +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_shards(port: u16, dir: &str, shards: usize) -> Child { + Command::new(moon_bin()) + .args([ + "--port", + &port.to_string(), + "--shards", + &shards.to_string(), + "--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); + read_one_reply(&mut reader) +} + +/// Read exactly one RESP reply from `reader`, returned as text (bulk bodies +/// inline; `$-1` nil yields an empty string). +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(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + let mut last = String::new(); + for cmd in cmds { + stream + .write_all(format!("{}\r\n", cmd).as_bytes()) + .expect("write"); + stream.flush().ok(); + last = read_one_reply(&mut reader); + } + last +} + +fn get_in_db(addr: &str, db: usize, key: &str) -> Option { + let sel = format!("SELECT {}", db); + let out = send_seq(addr, &[&sel, &format!("GET {}", key)]); + if out.is_empty() { None } else { Some(out) } +} + +fn dbsize_in_db(addr: &str, db: usize) -> i64 { + let sel = format!("SELECT {}", db); + let out = send_seq(addr, &[&sel, "DBSIZE"]); + out.strip_prefix(':') + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(-1) +} + +fn wait_until bool>(timeout: Duration, f: F) -> bool { + let deadline = std::time::Instant::now() + timeout; + while std::time::Instant::now() < deadline { + if f() { + return true; + } + thread::sleep(Duration::from_millis(100)); + } + false +} + +fn await_ready(addr: &str) { + assert!( + wait_until(Duration::from_secs(15), || send_cmd(addr, "PING") + .starts_with("+PONG")), + "server at {} did not become ready", + addr + ); +} + +struct Guard(Vec); +impl Drop for Guard { + fn drop(&mut self) { + for c in &mut self.0 { + let _ = c.kill(); + let _ = c.wait(); + } + } +} + +/// Core R2 scenario, parameterized on the master's shard count: +/// 1. Load keys into db0 and db2 on an N-shard master (keys spread across +/// all shards by hash). +/// 2. Attach a single-shard replica → full resync must deliver EVERYTHING. +/// 3. Write more keys (including INCR — non-idempotent, catches any +/// double-delivery between snapshot and live stream) → replica converges. +/// 4. WAIT 1 must observe the acked replica. +fn run_multishard_master_scenario(shards: usize, master_port: u16, replica_port: u16) { + let mdir = tempfile::tempdir().expect("mdir"); + let rdir = tempfile::tempdir().expect("rdir"); + let master = start_moon_shards(master_port, mdir.path().to_str().unwrap(), shards); + let replica = start_moon_shards(replica_port, rdir.path().to_str().unwrap(), 1); + let _guard = Guard(vec![master, replica]); + let m = format!("127.0.0.1:{}", master_port); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&m); + await_ready(&r); + + // Pre-sync dataset: 200 keys in db0, 60 in db2, plus a counter INCR'd to 7. + { + let mut stream = TcpStream::connect(&m).expect("connect master"); + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + let mut run = |cmd: String| { + stream.write_all(format!("{}\r\n", cmd).as_bytes()).unwrap(); + stream.flush().ok(); + read_one_reply(&mut reader) + }; + for i in 0..200 { + assert!( + run(format!("SET pre:{} v{}", i, i)).starts_with("+OK"), + "SET pre:{}", + i + ); + } + for _ in 0..7 { + run("INCR pre:counter".to_string()); + } + assert!(run("SELECT 2".to_string()).starts_with("+OK")); + for i in 0..60 { + assert!( + run(format!("SET d2:{} w{}", i, i)).starts_with("+OK"), + "SET d2:{}", + i + ); + } + } + assert_eq!(dbsize_in_db(&m, 0), 201, "master db0 baseline"); + assert_eq!(dbsize_in_db(&m, 2), 60, "master db2 baseline"); + + // Attach the replica. Before R2 the master answered + // `-ERR PSYNC across multiple shards is not yet supported` and the replica + // stayed empty forever. + let ro = send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)); + assert!(ro.starts_with("+OK"), "REPLICAOF failed: {}", ro); + + assert!( + wait_until(Duration::from_secs(30), || { + dbsize_in_db(&r, 0) == 201 && dbsize_in_db(&r, 2) == 60 + }), + "replica did not receive the {}-shard master's full snapshot: db0={} (want 201) db2={} (want 60)", + shards, + dbsize_in_db(&r, 0), + dbsize_in_db(&r, 2) + ); + // Spot-check values in both dbs + the non-idempotent counter. + assert_eq!(get_in_db(&r, 0, "pre:42").as_deref(), Some("v42")); + assert_eq!(get_in_db(&r, 0, "pre:counter").as_deref(), Some("7")); + assert_eq!(get_in_db(&r, 2, "d2:13").as_deref(), Some("w13")); + + // Live stream: more writes across dbs and shards, incl. INCRs. + { + let mut stream = TcpStream::connect(&m).expect("connect master"); + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + let mut run = |cmd: String| { + stream.write_all(format!("{}\r\n", cmd).as_bytes()).unwrap(); + stream.flush().ok(); + read_one_reply(&mut reader) + }; + for i in 0..150 { + assert!(run(format!("SET live:{} L{}", i, i)).starts_with("+OK")); + } + for _ in 0..5 { + run("INCR pre:counter".to_string()); + } + assert!(run("SELECT 2".to_string()).starts_with("+OK")); + for i in 0..40 { + assert!(run(format!("SET live2:{} M{}", i, i)).starts_with("+OK")); + } + } + assert!( + wait_until(Duration::from_secs(30), || { + dbsize_in_db(&r, 0) == 351 && dbsize_in_db(&r, 2) == 100 + }), + "replica did not converge on the live stream: db0={} (want 351) db2={} (want 100)", + dbsize_in_db(&r, 0), + dbsize_in_db(&r, 2) + ); + assert_eq!(get_in_db(&r, 0, "live:149").as_deref(), Some("L149")); + assert_eq!(get_in_db(&r, 0, "pre:counter").as_deref(), Some("12")); + assert_eq!(get_in_db(&r, 2, "live2:39").as_deref(), Some("M39")); + + // WAIT must see the acked replica (R1 plumbing on a multi-shard master). + let w = send_cmd(&m, "WAIT 1 3000"); + assert_eq!(w.trim(), ":1", "WAIT on multi-shard master: {}", w); +} + +#[test] +#[ignore] +fn multishard_master_full_resync_2shards() { + run_multishard_master_scenario(2, 17021, 17022); +} + +#[test] +#[ignore] +fn multishard_master_full_resync_4shards() { + run_multishard_master_scenario(4, 17031, 17032); +} + +#[test] +#[ignore] +fn multishard_master_full_resync_8shards() { + run_multishard_master_scenario(8, 17041, 17042); +} + +/// Interleaved multi-db writers against a 4-shard master: two connections pin +/// different dbs and hammer pipelined SETs concurrently WHILE the replica is +/// attached. On a merged multi-shard wire the per-record `SELECT` framing must +/// keep every write in its own db — any cross-shard interleave that splits a +/// SELECT from its payload lands writes in the wrong db (leak asserts catch +/// it). +#[test] +#[ignore] +fn multishard_master_interleaved_multidb_live_stream() { + let shards = 4; + let (master_port, replica_port) = (17051, 17052); + let mdir = tempfile::tempdir().expect("mdir"); + let rdir = tempfile::tempdir().expect("rdir"); + let master = start_moon_shards(master_port, mdir.path().to_str().unwrap(), shards); + let replica = start_moon_shards(replica_port, rdir.path().to_str().unwrap(), 1); + let _guard = Guard(vec![master, replica]); + let m = format!("127.0.0.1:{}", master_port); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&m); + await_ready(&r); + + let ro = send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)); + assert!(ro.starts_with("+OK"), "REPLICAOF failed: {}", ro); + assert!( + wait_until(Duration::from_secs(15), || send_cmd(&r, "INFO replication") + .contains("master_link_status:up")), + "replica link did not come up" + ); + + const PER_DB: usize = 5000; + let m0 = m.clone(); + let t0 = thread::spawn(move || { + let mut stream = TcpStream::connect(&m0).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(10))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + // db 0 writer, pipelined bursts of 100. + for burst in 0..(PER_DB / 100) { + let mut buf = String::new(); + for i in 0..100 { + buf.push_str(&format!("SET a:{} x{}\r\n", burst * 100 + i, i)); + } + stream.write_all(buf.as_bytes()).unwrap(); + stream.flush().ok(); + for _ in 0..100 { + read_one_reply(&mut reader); + } + } + }); + let m2 = m.clone(); + let t2 = thread::spawn(move || { + let mut stream = TcpStream::connect(&m2).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(10))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + stream.write_all(b"SELECT 2\r\n").unwrap(); + read_one_reply(&mut reader); + for burst in 0..(PER_DB / 100) { + let mut buf = String::new(); + for i in 0..100 { + buf.push_str(&format!("SET b:{} y{}\r\n", burst * 100 + i, i)); + } + stream.write_all(buf.as_bytes()).unwrap(); + stream.flush().ok(); + for _ in 0..100 { + read_one_reply(&mut reader); + } + } + }); + t0.join().expect("db0 writer"); + t2.join().expect("db2 writer"); + assert_eq!(dbsize_in_db(&m, 0), PER_DB as i64, "master db0"); + assert_eq!(dbsize_in_db(&m, 2), PER_DB as i64, "master db2"); + + assert!( + wait_until(Duration::from_secs(60), || { + dbsize_in_db(&r, 0) == PER_DB as i64 && dbsize_in_db(&r, 2) == PER_DB as i64 + }), + "replica diverged under interleaved multi-db load: db0={} db2={} (want {} each)", + dbsize_in_db(&r, 0), + dbsize_in_db(&r, 2), + PER_DB + ); + // Leak checks: a misapplied SELECT would put a:* keys in db2 or b:* in db0. + assert_eq!(get_in_db(&r, 0, "a:4999").as_deref(), Some("x99")); + assert_eq!(get_in_db(&r, 2, "b:4999").as_deref(), Some("y99")); + assert!(get_in_db(&r, 0, "b:0").is_none(), "db2 key leaked into db0"); + assert!(get_in_db(&r, 2, "a:0").is_none(), "db0 key leaked into db2"); +} + +fn send_resp(addr: &str, parts: &[&str]) -> String { + let Ok(mut stream) = TcpStream::connect(addr) else { + return String::new(); + }; + stream + .set_read_timeout(Some(Duration::from_millis(500))) + .ok(); + let mut out = format!("*{}\r\n", parts.len()).into_bytes(); + for p in parts { + out.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + out.extend_from_slice(p.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + if stream.write_all(&out).is_err() { + return String::new(); + } + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + let deadline = std::time::Instant::now() + Duration::from_millis(600); + while std::time::Instant::now() < deadline { + match stream.read(&mut chunk) { + Ok(0) => break, + Ok(n) => buf.extend_from_slice(&chunk[..n]), + Err(_) => { + if !buf.is_empty() { + break; + } + } + } + } + String::from_utf8_lossy(&buf).into_owned() +} + +/// Graph content is SHARDED: a merged multi-shard snapshot carries one +/// graph-store aux blob per shard, and the replica must import ALL of them +/// (`install_graph_store_many`) — importing only the first/last blob loses +/// every graph living on the other shards. +#[test] +#[ignore] +fn multishard_master_graph_snapshot_all_shards() { + let shards = 4; + let (master_port, replica_port) = (17071, 17072); + let mdir = tempfile::tempdir().expect("mdir"); + let rdir = tempfile::tempdir().expect("rdir"); + let master = start_moon_shards(master_port, mdir.path().to_str().unwrap(), shards); + let replica = start_moon_shards(replica_port, rdir.path().to_str().unwrap(), 1); + let _guard = Guard(vec![master, replica]); + let m = format!("127.0.0.1:{}", master_port); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&m); + await_ready(&r); + + // Enough graphs that hashing spreads them across all 4 shards. + let graphs = ["ga", "gb", "gc", "gd", "ge", "gf", "gg", "gh"]; + for (i, g) in graphs.iter().enumerate() { + assert!( + send_cmd(&m, &format!("GRAPH.CREATE {}", g)).contains("OK"), + "GRAPH.CREATE {}", + g + ); + for j in 0..=i { + let reply = send_resp( + &m, + &["GRAPH.ADDNODE", g, "Person", "name", &format!("p{}", j)], + ); + assert!(reply.starts_with(':'), "ADDNODE {} p{}: {}", g, j, reply); + } + } + + // Attach AFTER the writes — everything must arrive via the merged + // snapshot (per-shard graph aux blobs), not the live stream. + assert!(send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)).starts_with("+OK")); + assert!( + wait_until(Duration::from_secs(30), || { + let list = send_resp(&r, &["GRAPH.LIST"]); + graphs.iter().all(|g| list.contains(g)) + }), + "replica GRAPH.LIST missing graphs after snapshot: {}", + send_resp(&r, &["GRAPH.LIST"]) + ); + // Node counts survive per graph (graph i has i+1 nodes). + for (i, g) in graphs.iter().enumerate() { + let want = format!(":{}", i + 1); + assert!( + wait_until(Duration::from_secs(10), || { + send_resp(&r, &["GRAPH.QUERY", g, "MATCH (n:Person) RETURN count(n)"]) + .contains(&want) + }), + "replica graph {} node count != {}: {}", + g, + i + 1, + send_resp(&r, &["GRAPH.QUERY", g, "MATCH (n:Person) RETURN count(n)"]) + ); + } +} + +/// A multi-shard master must answer ANY resumable PSYNC with +FULLRESYNC (a +/// single total offset cannot be mapped back onto N per-shard backlogs), and +/// the payload must be ONE merged RDB bulk. +#[test] +#[ignore] +fn multishard_master_partial_resync_degrades_to_full() { + let (master_port,) = (17061,); + let mdir = tempfile::tempdir().expect("mdir"); + let master = start_moon_shards(master_port, mdir.path().to_str().unwrap(), 4); + let _guard = Guard(vec![master]); + let m = format!("127.0.0.1:{}", master_port); + await_ready(&m); + for i in 0..50 { + assert!(send_cmd(&m, &format!("SET k:{} v", i)).starts_with("+OK")); + } + + // Learn the master's replid. + let info = send_cmd(&m, "INFO replication"); + let replid = info + .lines() + .find_map(|l| l.strip_prefix("master_replid:")) + .map(|s| s.trim().to_string()) + .expect("master_replid in INFO"); + + // Speak the handshake by hand and ask to RESUME at offset 10 — the master + // must refuse to CONTINUE and issue a full resync instead. + let mut stream = TcpStream::connect(&m).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + stream.write_all(b"PING\r\n").unwrap(); + read_one_reply(&mut reader); + stream + .write_all(b"REPLCONF listening-port 17062\r\n") + .unwrap(); + read_one_reply(&mut reader); + stream + .write_all(format!("PSYNC {} 10\r\n", replid).as_bytes()) + .unwrap(); + stream.flush().ok(); + let mut line = String::new(); + reader.read_line(&mut line).expect("psync reply"); + assert!( + line.starts_with("+FULLRESYNC"), + "multi-shard master must degrade partial resync to FULLRESYNC, got: {}", + line.trim_end() + ); + // Next line: one merged RDB bulk header `$` with a REDIS magic body. + line.clear(); + reader.read_line(&mut line).expect("rdb header"); + let len: usize = line + .trim_start_matches('$') + .trim() + .parse() + .unwrap_or_else(|_| panic!("expected $ RDB header, got: {}", line.trim_end())); + let mut magic = vec![0u8; 5]; + reader.read_exact(&mut magic).expect("rdb magic"); + assert_eq!(&magic, b"REDIS", "merged snapshot must be Redis-format RDB"); + assert!(len > 9, "suspiciously small RDB ({} bytes)", len); +} From bbe779c24acd375024f903ead16534322fcd1913 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 16:07:39 +0700 Subject: [PATCH 2/5] docs(replication): document self-queue-first drain dependency in PrepareReplicaSync arm author: Tin Dang --- src/shard/spsc_handler.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 5332cfb0..f3863975 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2591,6 +2591,13 @@ pub(crate) fn handle_shard_message_shared( // the shard's own thread, so no mutation can slip between "inside // the snapshot" and "delivered live" (the same atomicity argument // as `handle_psync_inline_single_shard`, applied per shard). + // + // This additionally leans on the self-queue-FIRST drain order + // (see `drain_spsc_shared`): a local write visible to this body + // capture pushed its `ReplicaLiveFanout` BEFORE this arm could + // drain, and the self queue drains first — so that fan-out + // message no-ops against the not-yet-registered replica instead + // of double-delivering a record that is already in the body. let crate::shard::dispatch::PrepareReplicaSyncPayload { replica_id, tx, From 58574783df1aa7e3f439040f795eca2fc576354b Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 17:32:52 +0700 Subject: [PATCH 3/5] =?UTF-8?q?feat(replication):=20R3=20=E2=80=94=20tokio?= =?UTF-8?q?=20PSYNC=20clear=20error=20+=20topology=20docs=20refresh?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RFC v0.2-R3 (2A + 4B), riding the R2 PR: - A runtime-tokio master now answers PSYNC with `-ERR PSYNC requires runtime-monoio on the master (this build runs runtime-tokio)` instead of falling through to generic dispatch — an attaching replica's log states WHY the sync failed. Wired in both tokio paths (handler_sharded intercept + handler_single inline arm) and verified live against real tokio binaries at shards=1 and shards=2. - Docs refreshed for the R2 topology: - docs/guides/clustering.md: v0.1.x "master must run --shards 1" warning replaced with the v0.7 deployment shape (any-N master / --shards 1 replicas, merged-RDB semantics, partial->full at N>1, WS/MQ plane caveat). - README.md: replication bullets updated (also fixes the stale claim that replicas could run --shards N — the replica task refuses N != 1). - docs/PRODUCTION-CONTRACT.md: REPL-MULTISHARD-01 and WAIT-01 flipped to done with R2/R1 evidence. - CHANGELOG [Unreleased] amended accordingly. author: Tin Dang --- CHANGELOG.md | 9 ++++-- README.md | 4 +-- docs/PRODUCTION-CONTRACT.md | 4 +-- docs/guides/clustering.md | 34 ++++++++++++--------- src/server/conn/handler_sharded/dispatch.rs | 13 ++++++++ src/server/conn/handler_sharded/mod.rs | 5 +++ src/server/conn/handler_single.rs | 8 +++++ 7 files changed, 57 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a68e973f..af948ed3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,8 +44,13 @@ shards is not yet supported`. with db-leak asserts, per-shard graph snapshot import, and the partial→full degradation handshake. - Known limitation (unchanged from R1): master-side PSYNC requires - `runtime-monoio` (the default); the tokio build has no master-side PSYNC - intercept. Multi-shard *replicas* remain unsupported (`--shards 1`). + `runtime-monoio` (the default). A `runtime-tokio` master now answers PSYNC + with a clear `-ERR PSYNC requires runtime-monoio on the master` instead of + an unknown-command reply (RFC R3/2A). Multi-shard *replicas* remain + unsupported (`--shards 1`). +- Docs refreshed for the new topology: `docs/guides/clustering.md` deployment + shape, `README.md` replication bullets, and `docs/PRODUCTION-CONTRACT.md` + rows REPL-MULTISHARD-01 + WAIT-01 flipped to ✅ with evidence. ### Fixed — consistency/durability defects caught by the R1 gates (task #35) diff --git a/README.md b/README.md index 29d43baf..c0f96317 100644 --- a/README.md +++ b/README.md @@ -325,7 +325,7 @@ GA exit criteria) and [`docs/OPERATOR-GUIDE.md`](docs/OPERATOR-GUIDE.md) **Recommended for production today** - **Single-node deployments** — Linux aarch64 (Tier 1) or x86_64 (Tier 2), `--shards N` master. -- **Read replication** — `--shards 1` master with any `--shards N` replica topology (single-shard PSYNC2, wired since v0.1.10). +- **Read replication** — any `--shards N` master with `--shards 1` replicas (PSYNC2; multi-shard masters send one merged RDB + merged live stream, wired in v0.7). `WAIT` reflects real replica ACKs. - **AI workloads** — vector, BM25, GraphRAG, semantic cache, hybrid retrieval. All in-core, all RDB/WAL durable, crash-recovery validated. - **Cache + feature store** — honest durability modes (`always`/`everysec`/`no`), forkless snapshots, tiered NVMe offload under `maxmemory`. - **Crash recovery** — 100% survived across 7 persistence configs and 5K-key SIGKILL workloads (RDB v2 + WAL v3 + multi-part AOF + cold tier). @@ -333,7 +333,7 @@ GA exit criteria) and [`docs/OPERATOR-GUIDE.md`](docs/OPERATOR-GUIDE.md) **Not yet GA — avoid for production** - **Multi-node clustering** (16K-slot gossip, MOVED/ASK, failover) — protocol code exists but **PSYNC2 atomic slot migration is not soak-tested**. Scheduled for a v0.2.x follow-up. -- **Multi-shard master PSYNC** — single-shard only today ([RFC](.planning/rfcs/multi-shard-replication-design.md)). +- **Multi-shard replicas** — replicas run `--shards 1`; scale reads with more replicas, not replica shards. - **`CDC.SUBSCRIBE` push channel** and **zero-snapshot PITR (P3c)** — `CDC.READ` polling is ready; push/live-LSN are deferred. - **GPU vector acceleration** (`gpu-cuda`) — kernel scaffold only. - **Performance SLOs** in [`docs/PRODUCTION-CONTRACT.md`](docs/PRODUCTION-CONTRACT.md) are `[provisional]` until the 24-h HDR-histogram rig validates them. Treat the benchmarks above as point-in-time measurements, not committed SLOs. diff --git a/docs/PRODUCTION-CONTRACT.md b/docs/PRODUCTION-CONTRACT.md index abe459b6..0d346ab7 100644 --- a/docs/PRODUCTION-CONTRACT.md +++ b/docs/PRODUCTION-CONTRACT.md @@ -119,8 +119,8 @@ per the CI gate · Blocking `—` = tracked but never blocks a tag (see Out of S | Status | ID | Item | Evidence | Blocking | |---|---|---|---|---| | ✅ | REPL-01 | Single-shard-master PSYNC2 (full + partial resync), replica promotion | `src/replication/`, `tests/replication_hardening.rs`, `tests/replication_test.rs`, run by `integration-tests.yml` job `Replication Tests` | GA | -| ⬜ | REPL-MULTISHARD-01 | Multi-shard master replication (a `--shards N>1` master can be replicated at all) | Explicitly rejected at the wire: `src/server/conn/handler_monoio/dispatch.rs:564` — `"ERR PSYNC across multiple shards is not yet supported (use --shards 1 on the master)"`. Design exists (`.planning/rfcs/multi-shard-replication-design.md`); implementation is ROADMAP v0.7.0 workstream R1. | GA | -| ⬜ | WAIT-01 | `WAIT` reflects real replica ACK state | Stub — always returns `Frame::Integer(0)` on the sharded dispatch path (`src/command/mod.rs`, `WAIT` arm, comment: *"WAIT: no replication — always 0"*); `REPLCONF ACK` is parsed on the master socket but never consumed into replica ack-offset state. ROADMAP v0.7.0 workstream R2. | GA | +| ✅ | REPL-MULTISHARD-01 | Multi-shard master replication (a `--shards N>1` master can be replicated at all) | R2 (task #20): `ShardMessage::PrepareReplicaSync` per-shard atomic snapshot legs + merged Redis-format RDB + per-record SELECT framing on the merged wire. monoio only; replicas run `--shards 1`; partial resync degrades to full at N>1. `tests/replication_multishard.rs` (2/4/8-shard resync, interleaved multi-db parity, graph, partial→full). | GA | +| ✅ | WAIT-01 | `WAIT` reflects real replica ACK state | R1 (task #19, PR #282): replica 1s `REPLCONF ACK` ticker on the split PSYNC socket; master `ack_read_loop` + `drain_ack_offsets` record into `ReplicaInfo.ack_offsets`; connection-layer `try_handle_wait` blocks until ACK ≥ target or timeout. `wait_returns_acked_replica_count` e2e; exact on multi-shard masters too (summed snapshot offset). | GA | | ⬜ | KEYSPACE-NOTIF-01 | `notify-keyspace-events` keyspace notifications | No implementation found in `src/`. ROADMAP v0.7.0 workstream R5. | GA | | ⬜ | MONITOR-01 | `MONITOR` command | No implementation found in `src/command/`. ROADMAP v0.7.0 workstream R5. | GA | | ⬜ | XSHARD-READ-01 | Lock-free cross-shard read path (retire the shardslice waiver) | Waiver **expires 2026-08-01** per `RELEASES.md` v0.6.0 entry and ROADMAP §5; L4 redesign (`tmp/MULTISHARD-REDESIGN.md`) unstarted. ROADMAP v0.7.0 workstream R4. | GA | diff --git a/docs/guides/clustering.md b/docs/guides/clustering.md index a4aea2c3..a6767de3 100644 --- a/docs/guides/clustering.md +++ b/docs/guides/clustering.md @@ -11,21 +11,27 @@ Moon supports Redis-compatible replication and cluster mode for high availabilit Moon implements PSYNC2-compatible replication with per-shard WAL streaming and partial resync support. -!!! warning - **v0.1.x limitation — master must run `--shards 1`.** - - PSYNC on a multi-shard master (`--shards N` where N > 1) currently returns `-ERR PSYNC across multiple shards is not yet supported (use --shards 1 on the master)`. The master's N shard-local databases cannot yet be serialized into a single consistent RDB stream for a replica to consume. - - **Supported deployment shape for v0.1.x:** - - **Master:** `--shards 1` (single-core writer, ~1–1.5 M ops/s ceiling) - - **Replicas:** any `--shards N` (multi-core read scaling is unaffected) - - **Multi-shard master replication is scheduled for v0.2** (see `.planning/rfcs/multi-shard-replication-design.md`). If you need a multi-core master today, run without replication; if you need replication, accept the single-shard master ceiling. - - **Observability caveats (v0.1.x):** - - `WAIT` returns 0 until the master parses `REPLCONF ACK ` (v0.2 scope). +!!! info + **Supported deployment shape (v0.7):** + + - **Master:** any `--shards N` (multi-core writer). Multi-shard masters serve + a full resync as ONE merged Redis-format RDB followed by the merged live + stream from all shards; every record carries its own `SELECT` framing, so + multi-db workloads replicate exactly. Requires the default `runtime-monoio` + build — a `runtime-tokio` master answers PSYNC with + `-ERR PSYNC requires runtime-monoio on the master`. + - **Replicas:** `--shards 1` each (scale reads by adding replicas, not + replica shards). A multi-shard replica refuses to start replication. + - **Partial resync:** supported on single-shard masters (backlog window); + a multi-shard master answers every reconnect with a full resync (a single + scalar offset cannot be mapped back onto N per-shard backlogs). + + **Observability:** + - `WAIT N timeout` reflects real replica ACKs (1s `REPLCONF ACK` cadence). + - `master_link_status` in `INFO replication` reflects the handshake state — use it to detect a failed REPLICAOF. - `CLIENT LIST TYPE replica` has no predicate yet; returns all clients. - - `master_link_status` in `INFO replication` correctly reflects the handshake state — use it to detect a failed REPLICAOF. + - WS.\*/MQ.\* planes are **not replicated yet** (the master logs one warning + when a replica is attached); vector/text/graph planes replicate fully. ### Set up a replica diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 593dc98b..a86bb31a 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -713,3 +713,16 @@ pub(super) fn try_handle_replconf( responses.push(crate::command::connection::replconf(cmd_args)); true } + +/// RFC v0.2-R3 (2A): master-side PSYNC is monoio-only — the tokio runtime has +/// no connection-hijack path. Answer with a clear error instead of the +/// generic unknown-command reply so an attaching replica's log says WHY. +pub(super) fn try_handle_psync_unsupported(cmd: &[u8], responses: &mut Vec) -> bool { + if !cmd.eq_ignore_ascii_case(b"PSYNC") { + return false; + } + responses.push(Frame::Error(bytes::Bytes::from_static( + b"ERR PSYNC requires runtime-monoio on the master (this build runs runtime-tokio)", + ))); + true +} diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 13ad4537..d7ebedb5 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -861,6 +861,11 @@ pub(crate) async fn handle_connection_sharded_inner< continue; } + // --- PSYNC (unsupported on tokio; clear error, R3/2A) --- + if dispatch::try_handle_psync_unsupported(cmd, &mut responses) { + continue; + } + // --- CDC.READ --- if dispatch::try_handle_cdc_read(cmd, cmd_args, &mut responses) { continue; diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 7d65555b..1b80b7f6 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -859,6 +859,14 @@ pub async fn handle_connection( continue; } + // --- PSYNC (unsupported on tokio; clear error, R3/2A) --- + if cmd.eq_ignore_ascii_case(b"PSYNC") { + responses.push(Frame::Error(Bytes::from_static( + b"ERR PSYNC requires runtime-monoio on the master (this build runs runtime-tokio)", + ))); + continue; + } + // --- WAIT --- if cmd.eq_ignore_ascii_case(b"WAIT") { // WAIT numreplicas timeout From 8e56d23a2c1f9e910cc4d1527f09b076513b9dab Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 18:56:14 +0700 Subject: [PATCH 4/5] fix(replication): exactly-once live fanout (offset cut) + replica-task epoch cancellation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Attach-under-write stress testing of R2 (#283) surfaced three defects. All are fixed here before the PR merges; none shipped in a release. 1) REPLICAOF leaked the previous replica task (pre-existing, dominant signal). `REPLICAOF host port` spawned a fresh run_replica_task without stopping the old one, and `REPLICAOF NO ONE` only flipped the role: the old task kept streaming AND applying. Each NO-ONE → re-attach cycle stacked one more live applier — replica INCR counters ran ~25-35% ABOVE the master (reproduced at shards=1 and shards=4). Fix: process-global REPLICA_TASK_EPOCH ticket. StartReplication and PromoteToMaster bump the generation; a superseded task exits at the reconnect-loop top, before snapshot load, and before applying any parsed chunk — it can never mutate the keyspace again. 2) Snapshot-vs-live exactly-once now rests on an offset cut, not on FIFO placement. Two review rounds found opposite failure modes for placement schemes on the PSYNC connection's own shard: - queued snapshot leg (drain-time capture): a local write between push and drain was in the body AND live-sent behind the registration → double-apply (reproduced, +35%); - inline capture + queued registration: a same-cycle SPSC execute was neither in the body (applied after capture) nor live-sent (execute_batch direct-send runs before other_messages processes the registration) — with its bytes in the backlog and the offset advanced: silent loss + permanent replica ACK lag. Fix: every fan-out entry records cut = shard offset at body capture; every live record carries its per-shard end_offset; delivery requires end_offset > cut. Registration placement no longer matters, so ALL shards — including the PSYNC connection's own — use the same PrepareReplicaSync arm (the special-cased inline self-leg is gone). The cut/end_offset axis is the PER-SHARD counter, never the master offset (seed_master_offset diverges the two after AOF recovery). 3) Same-key wire ordering. SPSC-dispatched writes sent to replicas directly from the execute arm while local handler writes deferred through the self queue — a later-offset write could reach the wire before an earlier-offset same-key write, replaying out of the master's serialization order (analysis finding; not reproduced in ~10 black-box attempts). ALL live sends now flow through the self-queue ReplicaLiveFanout arm: per-shard wire order == FIFO order == offset order by construction. Tests (red/green: 1 and 2 reproduced red before their fixes): - tests/replication_multishard.rs +3: attach_under_write_no_double_apply (4-shard, 5x detach/re-attach under 4-writer pipelined INCR load, exact per-counter parity), singleshard_master_attach_under_write_control (same at shards=1 — the discriminating control that proved the task leak pre-existing), same_key_write_order_parity (12 conns APPEND-race 32 shared keys through both write paths; replica must byte-equal master). - Full regression: multishard 9/9, streaming 7/7, hardening 5/5, graph 5/5, aof_multidb_kill9 4/4, lib 4115 (monoio) + 3316 (tokio), clippy -D warnings both matrices, fmt — macOS and Linux VM. Mechanical: ShardMessage::RegisterReplica boxed (RegisterReplicaPayload; the extra offset fields pushed it past the 64-byte cap), ReplicaFanout tuple → struct with cut, increment_shard_offset returns the per-shard post-advance offset. Test-harness hardening surfaced by the Linux VM sweep (task #18 class): - replication_hardening now honors MOON_BIN (hardcoded ./target/release/moon exec'd the wrong-platform binary on the shared macOS/Linux checkout — all 5 tests failed to connect on the VM); - aof_multidb_kill9 wait_ready treats a connection RESET during moon's bootstrap→per-shard SO_REUSEPORT listener handover as "not ready yet" instead of panicking (3/4 VM failures, reproduced with main's binary too — environment-timing flake, not a code regression). Refs: task #20, task #36, PR #283 author: Tin Dang --- CHANGELOG.md | 40 +++ src/replication/master.rs | 238 +++++++++++----- src/replication/replica.rs | 69 +++++ src/replication/state.rs | 27 +- src/server/conn/handler_monoio/dispatch.rs | 10 + src/server/conn/handler_monoio/ft.rs | 11 +- src/server/conn/handler_sharded/dispatch.rs | 10 + src/server/conn/handler_single.rs | 4 + src/shard/dispatch.rs | 119 +++++--- src/shard/spsc_handler.rs | 149 +++++++--- tests/aof_multidb_kill9.rs | 29 +- tests/replication_hardening.rs | 10 +- tests/replication_multishard.rs | 301 ++++++++++++++++++++ 13 files changed, 840 insertions(+), 177 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index af948ed3..59688762 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -52,6 +52,46 @@ shards is not yet supported`. shape, `README.md` replication bullets, and `docs/PRODUCTION-CONTRACT.md` rows REPL-MULTISHARD-01 + WAIT-01 flipped to ✅ with evidence. +### Fixed — live-fanout exactly-once redesign + replica-task leak (task #20 follow-up) + +Attach-under-write stress testing of R2 surfaced three defects; all fixed +before release (none shipped): + +- **`REPLICAOF` leaked the previous replica task — every re-attach stacked + one more live applier.** `REPLICAOF host port` spawned a fresh + `run_replica_task` without stopping the old one, and `REPLICAOF NO ONE` + only flipped the role state: the old task kept its master link open and + kept APPLYING the stream. After NO-ONE → re-attach cycles the replica ran + INCR counters ~25-35% ABOVE the master (reproduced at shards=1 AND + shards=4 — pre-existing, not an R2 defect). Replica tasks now carry a + process-global epoch ticket; a new `REPLICAOF` target or `NO ONE` bumps + the generation and superseded tasks exit before their next connect, + before loading a snapshot, and before applying any parsed chunk. +- **Snapshot-vs-live exactly-once is now offset-cut based, not + FIFO-placement based.** Two adversarial-review rounds found opposite + failure modes for placement schemes (a queued self-shard snapshot leg + double-delivered local writes; an inline-captured one lost same-cycle + cross-shard writes — neither in the body nor live-sent, with the offset + advanced: permanent replica lag). Every fan-out entry now records + `cut = ` and every live record carries its + per-shard `end_offset`; delivery requires `end_offset > cut`, making + correctness independent of where the registration lands in the drain + FIFO. All shards (including the PSYNC connection's own) use the same + `PrepareReplicaSync` arm. +- **Same-key wire ordering.** Cross-shard (SPSC-dispatched) writes used to + send to replicas directly from the execute arm while local handler writes + deferred through the self queue — a later-offset write could reach the + wire before an earlier-offset write to the same key, replaying same-key + writes out of the master's order on the replica (found by analysis; not + reproduced in ~10 black-box runs). ALL live replica sends now flow + through the self-queue `ReplicaLiveFanout` arm, so per-shard wire order + equals offset order by construction. +- New e2e regressions: `attach_under_write_no_double_apply` (4-shard + + single-shard control, 5× detach/re-attach under pipelined INCR load, + exact per-counter parity) and `same_key_write_order_parity` (12 + connections APPEND-race the same 32 keys through both write paths; + replica strings must byte-equal the master). + ### Fixed — consistency/durability defects caught by the R1 gates (task #35) Three pre-existing data-integrity bugs surfaced by the new load/kill-9 gates diff --git a/src/replication/master.rs b/src/replication/master.rs index 7c609db3..1866ab77 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -438,22 +438,27 @@ 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, - // Legacy multi-shard drain loops do not poll the kick flag - // (superseded by the R2 redesign); overflow still stops - // queueing via the fan-out's retain. - kicked: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), - 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, - // Cross-shard registration: the target shard's offset is - // owned by its own thread — the arm reads it at drain. - push_offset: None, - }; + let msg = crate::shard::dispatch::ShardMessage::RegisterReplica(Box::new( + crate::shard::dispatch::RegisterReplicaPayload { + replica_id, + tx, + // Legacy multi-shard drain loops do not poll the kick flag + // (superseded by the R2 redesign); overflow still stops + // queueing via the fan-out's retain. + kicked: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + 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, + // Cross-shard registration: the target shard's offset is + // owned by its own thread — the arm reads it at drain. + push_offset: None, + // No snapshot body was captured on this shard's thread — + // the arm's drain-time offset is the correct cut. + cut: None, + }, + )); let _ = prod.try_push(msg); } @@ -536,22 +541,27 @@ 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, - // Legacy multi-shard drain loops do not poll the kick flag - // (superseded by the R2 redesign); overflow still stops - // queueing via the fan-out's retain. - kicked: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), - 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, - // Cross-shard registration: the target shard's offset is - // owned by its own thread — the arm reads it at drain. - push_offset: None, - }; + let msg = crate::shard::dispatch::ShardMessage::RegisterReplica(Box::new( + crate::shard::dispatch::RegisterReplicaPayload { + replica_id, + tx, + // Legacy multi-shard drain loops do not poll the kick flag + // (superseded by the R2 redesign); overflow still stops + // queueing via the fan-out's retain. + kicked: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + 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, + // Cross-shard registration: the target shard's offset is + // owned by its own thread — the arm reads it at drain. + push_offset: None, + // No snapshot body was captured on this shard's thread — + // the arm's drain-time offset is the correct cut. + cut: None, + }, + )); let _ = prod.try_push(msg); } @@ -851,8 +861,24 @@ pub async fn handle_psync_inline_multi_shard( let (tx, rx) = crate::runtime::channel::mpsc_bounded::(16384); let kicked = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); - // Fan out PrepareReplicaSync; replies are bounded(1) channels usable - // across threads. + // ── One uniform leg per shard: PrepareReplicaSync — the self shard via + // the thread-local self queue (the SPSC mesh has no self-loop), remote + // shards over the mesh + notifier. Each arm captures its RDB body, reads + // its shard offset, and registers the replica's fan-out entry with + // `cut = ` in ONE synchronous stretch on its own thread. + // + // Exactly-once no longer depends on WHERE the registration lands in the + // drain FIFO (two adversarial-review rounds found opposite failure modes + // for FIFO-placement schemes): every live record is delivered through + // `ReplicaLiveFanout` messages carrying the record's per-shard + // `end_offset`, and delivery is filtered per replica by `end_offset > + // cut`. A write applied before the arm's capture is inside the body and + // at/below the cut (its queued fan-out message no-ops); a write applied + // after it carries a higher end_offset and is delivered live exactly + // once. Wire order per shard equals the self-queue FIFO order equals + // offset order, so same-key writes replay in the master's order. + let mut vector_defs: Option> = None; + let mut text_defs: Option> = None; let mut reply_rxs = Vec::with_capacity(num_shards); for shard in 0..num_shards { let (reply_tx, reply_rx) = @@ -867,50 +893,68 @@ pub async fn handle_psync_inline_multi_shard( }, )); if shard == self_shard_id { + // Self queue push is infallible; the event loop drains it on its + // next cycle while this task awaits the reply below. crate::shard::self_msg::push(msg); - } else { - let idx = crate::shard::mesh::ChannelMesh::target_index(self_shard_id, shard); - // The SPSC ring can be transiently full under load — bounded retry, - // then abort loudly (the replica reconnects and retries the sync). - let mut attempts = 0u32; - loop { - let res = { dispatch_tx.borrow_mut()[idx].try_push(msg) }; - match res { - Ok(()) => { - spsc_notifiers[shard].notify_one(); - break; - } - Err(back) => { - msg = back; - attempts += 1; - if attempts > 5_000 { - anyhow::bail!( - "shard {} SPSC full for >5s during PSYNC fan-out; aborting sync", - shard - ); - } - monoio::time::sleep(std::time::Duration::from_millis(1)).await; + reply_rxs.push((shard, reply_rx)); + continue; + } + let idx = crate::shard::mesh::ChannelMesh::target_index(self_shard_id, shard); + // The SPSC ring can be transiently full under load — bounded retry, + // then abort loudly (the replica reconnects and retries the sync). + let mut attempts = 0u32; + loop { + let res = { dispatch_tx.borrow_mut()[idx].try_push(msg) }; + match res { + Ok(()) => { + spsc_notifiers[shard].notify_one(); + break; + } + Err(back) => { + msg = back; + attempts += 1; + if attempts > 5_000 { + unregister_replica_all_shards( + replica_id, + &dispatch_tx, + &spsc_notifiers, + self_shard_id, + num_shards, + ); + anyhow::bail!( + "shard {} SPSC full for >5s during PSYNC fan-out; aborting sync", + shard + ); } + monoio::time::sleep(std::time::Duration::from_millis(1)).await; } } } reply_rxs.push((shard, reply_rx)); } - // Collect every shard's leg. A dropped reply means that shard could not - // prepare (or we raced shutdown) — abort; already-registered shards clean - // up passively when `rx` drops (their next fan-out send sees Disconnected). + // Collect every leg. A dropped reply means that shard could not prepare + // (or we raced shutdown) — abort and explicitly unregister everywhere + // (review P2: passive Disconnected pruning only fires on a shard's NEXT + // write, which may never come). let mut bodies: Vec> = Vec::with_capacity(num_shards); let mut snapshot_offset: u64 = 0; - let mut vector_defs: Option> = None; - let mut text_defs: Option> = None; #[cfg(feature = "graph")] let mut graph_blobs: Vec> = Vec::with_capacity(num_shards); for (shard, reply_rx) in reply_rxs { - let prepared = reply_rx - .recv_async() - .await - .map_err(|_| anyhow::anyhow!("shard {} dropped its PrepareReplicaSync reply", shard))?; + let prepared = match reply_rx.recv_async().await { + Ok(p) => p, + Err(_) => { + unregister_replica_all_shards( + replica_id, + &dispatch_tx, + &spsc_notifiers, + self_shard_id, + num_shards, + ); + anyhow::bail!("shard {} dropped its PrepareReplicaSync reply", shard); + } + }; snapshot_offset += prepared.shard_offset; // Index definitions are keyspace-global and identical on every shard — // keep the first non-empty copy. @@ -978,6 +1022,35 @@ pub async fn handle_psync_inline_multi_shard( // self-queued UnregisterReplica for this shard). A full ring is fine — // dropping `rx` above already flipped every sender to Disconnected, which // the next fan-out send prunes. + unregister_replica_all_shards( + replica_id, + &dispatch_tx, + &spsc_notifiers, + self_shard_id, + num_shards, + ); + drain_result +} + +/// Best-effort `UnregisterReplica` to every shard: the self shard via the +/// self queue, remote shards via the mesh (a full ring is tolerated — the +/// passive Disconnected prune covers it on that shard's next write). Used on +/// multi-shard PSYNC abort paths and after the drain loop exits, so a shard +/// that never sees another write doesn't hold a dead fan-out entry forever +/// (review P2). +#[cfg(feature = "runtime-monoio")] +fn unregister_replica_all_shards( + replica_id: u64, + dispatch_tx: &Rc>>>, + spsc_notifiers: &[std::sync::Arc], + self_shard_id: usize, + num_shards: usize, +) { + use ringbuf::traits::Producer; + + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::UnregisterReplica { + replica_id, + }); for shard in 0..num_shards { if shard == self_shard_id { continue; @@ -989,7 +1062,6 @@ pub async fn handle_psync_inline_multi_shard( spsc_notifiers[shard].notify_one(); } } - drain_result } /// Send backlog bytes `[from, to)` to the replica, or fail LOUDLY if the @@ -1097,18 +1169,28 @@ fn push_register_replica_inline( // also delivers it live: double-applied on the replica. The push-time // offset keeps catch-up and live delivery disjoint for every interleave // (see `RegisterReplica::push_offset`). - let push_offset = repl_state - .read() - .map(|g| g.total_offset()) - .map_err(|_| anyhow::anyhow!("replication state lock poisoned"))?; - crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::RegisterReplica { - replica_id, - tx: tx.clone(), - kicked: kicked.clone(), - backlog_capacity, - registered: Some(reg_tx), - push_offset: Some(push_offset), - }); + let (push_offset, push_shard_offset) = { + let g = repl_state + .read() + .map_err(|_| anyhow::anyhow!("replication state lock poisoned"))?; + // Master-axis offset for the catch-up reply protocol, PER-SHARD-axis + // offset for the fan-out cut — the two counters diverge after + // `seed_master_offset` (AOF recovery) and must never be mixed. This + // path only runs at shards=1 (multi-shard PSYNC routes through + // `handle_psync_inline_multi_shard`), so shard 0 is THE shard. + (g.total_offset(), g.shard_offset(0)) + }; + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::RegisterReplica( + Box::new(crate::shard::dispatch::RegisterReplicaPayload { + replica_id, + tx: tx.clone(), + kicked: kicked.clone(), + backlog_capacity, + registered: Some(reg_tx), + push_offset: Some(push_offset), + cut: Some(push_shard_offset), + }), + )); Ok(InlineReplicaRegistration { replica_id, tx, diff --git a/src/replication/replica.rs b/src/replication/replica.rs index 7a95ae14..4167a405 100644 --- a/src/replication/replica.rs +++ b/src/replication/replica.rs @@ -18,6 +18,35 @@ use tracing::{info, warn}; use crate::replication::handshake::ReplicaHandshakeState; use crate::replication::state::{ReplicationRole, ReplicationState, save_replication_state}; +/// Process-global generation counter for replica tasks (attach-under-write +/// P0, found while testing R2): `REPLICAOF host port` used to spawn a fresh +/// `run_replica_task` WITHOUT stopping the previous one, and `REPLICAOF NO +/// ONE` only flipped the role state — the old task kept its master link open +/// and kept APPLYING the stream. After a NO-ONE → re-attach cycle, two (then +/// three, ...) live tasks each applied every record: replica INCR counters +/// ran ~25-35% ABOVE the master under write load (reproduced at shards=1 and +/// shards=4 — pre-existing, not an R2 defect). +/// +/// Every spawn bumps the epoch and hands the task its ticket; `REPLICAOF NO +/// ONE` bumps it too. A task whose ticket no longer matches exits before its +/// next connect, before applying a snapshot, and before applying any parsed +/// chunk — a superseded task can never mutate the keyspace again (its parked +/// socket read wakes on the next master byte or link close and hits the +/// pre-apply check). +static REPLICA_TASK_EPOCH: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + +/// Bump the generation (new REPLICAOF target, or NO ONE) and return the new +/// ticket to hand to a freshly spawned task. +pub fn bump_replica_task_epoch() -> u64 { + REPLICA_TASK_EPOCH.fetch_add(1, Ordering::AcqRel) + 1 +} + +/// True when `epoch` is no longer the live generation — the owning task must +/// stop without touching local state. +fn superseded(epoch: u64) -> bool { + REPLICA_TASK_EPOCH.load(Ordering::Acquire) != epoch +} + /// Configuration for the replica outbound connection task. pub struct ReplicaTaskConfig { pub master_host: String, @@ -26,6 +55,9 @@ pub struct ReplicaTaskConfig { pub num_shards: usize, pub persistence_dir: Option, pub listening_port: u16, + /// Generation ticket from [`bump_replica_task_epoch`] — the task exits + /// as soon as a newer generation exists. + pub epoch: u64, /// Logical-db context of the replication stream, preserved ACROSS /// reconnects (HIGH-2, task #22): a `+CONTINUE` partial resync replays /// backlog bytes that only contain `SELECT` at db CHANGES — if the stream @@ -63,6 +95,10 @@ pub async fn run_replica_task(cfg: ReplicaTaskConfig) { const MAX_BACKOFF_MS: u64 = 30_000; loop { + if superseded(cfg.epoch) { + info!("Replica: task superseded (epoch {}), exiting", cfg.epoch); + return; + } info!("Replica: connecting to master at {}", addr); match TcpStream::connect(&addr).await { Ok(stream) => { @@ -84,6 +120,12 @@ pub async fn run_replica_task(cfg: ReplicaTaskConfig) { } } + // A superseded task must not clobber the successor's handshake state. + if superseded(cfg.epoch) { + info!("Replica: task superseded (epoch {}), exiting", cfg.epoch); + return; + } + // Update handshake state to Disconnected in ReplicationState if let Ok(mut rs) = cfg.repl_state.write() { if let ReplicationRole::Replica { ref mut state, .. } = rs.role { @@ -220,6 +262,9 @@ async fn run_handshake_and_stream( // 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. + if superseded(cfg.epoch) { + anyhow::bail!("replica task superseded before snapshot load"); + } for shard_id in 0..cfg.num_shards { let rdb_bytes = read_rdb_bulk(&mut stream).await?; match crate::replication::apply::load_snapshot(&rdb_bytes) { @@ -323,6 +368,11 @@ async fn stream_commands_read_loop( if n == 0 { return Err(anyhow::anyhow!("Master closed connection")); } + // Superseded tasks must never apply another byte — checked after the + // parked read wakes, before any parse/apply. + if superseded(cfg.epoch) { + anyhow::bail!("replica task superseded — dropping stream unapplied"); + } // Parse every complete RESP command in the buffer and apply it to the // local shard. The replication offset advances by CONSUMED bytes (whole @@ -380,6 +430,10 @@ pub async fn run_replica_task(cfg: ReplicaTaskConfig) { const MAX_BACKOFF_MS: u64 = 30_000; loop { + if superseded(cfg.epoch) { + info!("Replica: task superseded (epoch {}), exiting", cfg.epoch); + return; + } info!("Replica: connecting to master at {}", addr); match monoio::net::TcpStream::connect(addr).await { Ok(stream) => { @@ -401,6 +455,12 @@ pub async fn run_replica_task(cfg: ReplicaTaskConfig) { } } + // A superseded task must not clobber the successor's handshake state. + if superseded(cfg.epoch) { + info!("Replica: task superseded (epoch {}), exiting", cfg.epoch); + return; + } + if let Ok(mut rs) = cfg.repl_state.write() { if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::Disconnected; @@ -533,6 +593,9 @@ async fn run_handshake_and_stream( // 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. + if superseded(cfg.epoch) { + anyhow::bail!("replica task superseded before snapshot load"); + } for shard_id in 0..cfg.num_shards { let rdb_bytes = read_rdb_bulk(&mut stream).await?; match crate::replication::apply::load_snapshot(&rdb_bytes) { @@ -645,6 +708,12 @@ async fn stream_commands_read_loop( } buf.extend_from_slice(&tmp[..n]); + // Superseded tasks must never apply another byte — checked after the + // parked read wakes, before any parse/apply. + if superseded(cfg.epoch) { + anyhow::bail!("replica task superseded — dropping stream unapplied"); + } + // 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. diff --git a/src/replication/state.rs b/src/replication/state.rs index c3cbf737..b7e8bbd1 100644 --- a/src/replication/state.rs +++ b/src/replication/state.rs @@ -163,8 +163,19 @@ impl ReplicationState { /// Increment the offset for the given shard by delta bytes. /// Also adds delta to master_repl_offset. - pub fn increment_shard_offset(&self, shard_id: usize, delta: u64) { - let _ = self.issue_lsn(shard_id, delta); + /// + /// Returns the PER-SHARD offset after the advance — the record's + /// `end_offset` on the live-fanout wire, compared against each replica's + /// per-shard snapshot cut (`ReplicaFanout::cut`). Deliberately NOT the + /// master offset: `seed_master_offset` (AOF recovery) advances only the + /// master counter, so the two axes diverge and must never be mixed. + pub fn increment_shard_offset(&self, shard_id: usize, delta: u64) -> u64 { + if shard_id >= self.shard_offsets.len() { + return 0; + } + let prev = self.shard_offsets[shard_id].fetch_add(delta, Ordering::Relaxed); + self.master_repl_offset.fetch_add(delta, Ordering::Relaxed); + prev + delta } /// Atomically issue an LSN for a write and advance per-shard + @@ -259,10 +270,16 @@ impl OffsetHandle { self.master_repl_offset.fetch_add(delta, Ordering::Relaxed) } - /// See [`ReplicationState::increment_shard_offset`]. + /// See [`ReplicationState::increment_shard_offset`] — returns the + /// per-shard offset after the advance (the record's fan-out `end_offset`). #[inline] - pub fn increment_shard_offset(&self, shard_id: usize, delta: u64) { - let _ = self.issue_lsn(shard_id, delta); + pub fn increment_shard_offset(&self, shard_id: usize, delta: u64) -> u64 { + if shard_id >= self.shard_offsets.len() { + return 0; + } + let prev = self.shard_offsets[shard_id].fetch_add(delta, Ordering::Relaxed); + self.master_repl_offset.fetch_add(delta, Ordering::Relaxed); + prev + delta } /// Current offset of one shard. Used by the `RegisterReplica` reply to diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 7d53ec14..4d61e7e6 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -464,6 +464,11 @@ pub(super) fn try_handle_replicaof( }); } let rs_clone = Arc::clone(rs); + // Bump the task generation FIRST: any previously spawned + // replica task (old REPLICAOF target) sees itself + // superseded and exits instead of double-applying the + // stream alongside the new task. + let epoch = crate::replication::replica::bump_replica_task_epoch(); let cfg = crate::replication::replica::ReplicaTaskConfig { master_host: host, master_port: port, @@ -471,12 +476,17 @@ pub(super) fn try_handle_replicaof( num_shards: ctx.num_shards, persistence_dir: None, listening_port: 0, + epoch, stream_db: std::sync::atomic::AtomicUsize::new(0), }; monoio::spawn(crate::replication::replica::run_replica_task(cfg)); } ReplicaofAction::PromoteToMaster => { use crate::replication::state::generate_repl_id; + // Kill the running replica task — flipping the role alone + // left it streaming + applying forever (each NO ONE → + // re-attach cycle stacked one more live applier). + let _ = crate::replication::replica::bump_replica_task_epoch(); if let Ok(mut rs_guard) = rs.write() { rs_guard.repl_id2 = rs_guard.repl_id.clone(); rs_guard.repl_id = generate_repl_id(); diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index 89aa3f15..8b5b9aca 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/src/server/conn/handler_monoio/ft.rs @@ -76,6 +76,10 @@ pub(super) fn replication_fanout_active(ctx: &ConnectionContext) -> bool { /// ⚠ Monoio shard threads only (pushes to `shard::self_msg`) — callers are /// all inside `handler_monoio`, which is `runtime-monoio`-gated. pub(super) fn record_local_write(ctx: &ConnectionContext, bytes: Bytes) { + // Per-shard offset AFTER this record — the fan-out arm compares it + // against each replica's snapshot cut (`ReplicaFanout::cut`) so a record + // already inside a FULLRESYNC body is never live-delivered again. + let mut end_offset = u64::MAX; if let Some(rs) = ctx.repl_state.as_ref() { if let Ok(g) = rs.read() { if let Some(slot) = g.per_shard_backlogs.get(ctx.shard_id) { @@ -83,10 +87,13 @@ pub(super) fn record_local_write(ctx: &ConnectionContext, bytes: Bytes) { backlog.append(&bytes); } } - g.increment_shard_offset(ctx.shard_id, bytes.len() as u64); + end_offset = g.increment_shard_offset(ctx.shard_id, bytes.len() as u64); } } - crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { bytes }); + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { + bytes, + end_offset, + }); } /// Db-aware variant of [`record_local_write`] (HIGH-2, task #22): prepends a diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index a86bb31a..62245fdd 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -281,6 +281,11 @@ pub(super) fn try_handle_replicaof( }); } let rs_clone = Arc::clone(rs); + // Bump the task generation FIRST: any previously spawned + // replica task (old REPLICAOF target) sees itself + // superseded and exits instead of double-applying the + // stream alongside the new task. + let epoch = crate::replication::replica::bump_replica_task_epoch(); let cfg = crate::replication::replica::ReplicaTaskConfig { master_host: host, master_port: port, @@ -288,12 +293,17 @@ pub(super) fn try_handle_replicaof( num_shards: ctx.num_shards, persistence_dir: None, listening_port: 0, + epoch, stream_db: std::sync::atomic::AtomicUsize::new(0), }; tokio::task::spawn_local(crate::replication::replica::run_replica_task(cfg)); } ReplicaofAction::PromoteToMaster => { use crate::replication::state::generate_repl_id; + // Kill the running replica task — flipping the role alone + // left it streaming + applying forever (each NO ONE → + // re-attach cycle stacked one more live applier). + let _ = crate::replication::replica::bump_replica_task_epoch(); if let Ok(mut rs_guard) = rs.write() { rs_guard.repl_id2 = rs_guard.repl_id.clone(); rs_guard.repl_id = generate_repl_id(); diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 1b80b7f6..538a11c7 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -838,6 +838,10 @@ pub async fn handle_connection( } ReplicaofAction::PromoteToMaster => { use crate::replication::state::generate_repl_id; + // handler_single spawns no replica task itself, but + // bump the generation anyway so any task spawned by + // another handler path stops applying. + let _ = crate::replication::replica::bump_replica_task_epoch(); if let Ok(mut rs_guard) = rs.write() { rs_guard.repl_id2 = rs_guard.repl_id.clone(); rs_guard.repl_id = generate_repl_id(); diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index 827d6e4d..61086728 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -411,16 +411,69 @@ pub struct CdcSubscribePayload { pub from_lsn: u64, } -/// One live replica fan-out endpoint held by a shard thread: -/// `(replica_id, live channel sender, kicked flag)`. -/// -/// `kicked` is the overflow disconnect signal — see -/// [`ShardMessage::RegisterReplica::kicked`]. -pub type ReplicaFanout = ( - u64, - channel::MpscSender, - std::sync::Arc, -); +/// One live replica fan-out endpoint held by a shard thread. +pub struct ReplicaFanout { + /// Master-side replica id (see [`ShardMessage::RegisterReplica`]). + pub replica_id: u64, + /// Live channel onto the replica's socket drain task. + pub tx: channel::MpscSender, + /// Overflow disconnect signal — see [`ShardMessage::RegisterReplica::kicked`]. + pub kicked: std::sync::Arc, + /// Exactly-once cut line: this shard's replication offset at the moment + /// the replica's snapshot body was captured. A live record is delivered + /// iff its `end_offset > cut` — records at or below the cut are already + /// inside the snapshot body (their mutation and offset advance happened + /// before the capture), so sending them again would double-apply + /// non-idempotent commands on the replica. This makes correctness + /// independent of WHERE the registration lands relative to a record's + /// fan-out message in the drain FIFO. + pub cut: u64, +} + +/// Payload of [`ShardMessage::RegisterReplica`]. +pub struct RegisterReplicaPayload { + pub replica_id: u64, + pub tx: channel::MpscSender, + /// Set by the shard fan-out when this replica's bounded channel is + /// FULL: a record that cannot be queued would otherwise be silently + /// dropped, permanently diverging the replica while + /// `master_link_status` stays "up" (task #35 — observed 2k of 40k + /// keys delivered). The drain task polls the flag and disconnects, + /// converting silent divergence into a loud PSYNC resync (Redis + /// parity: output-buffer-limit disconnects). + pub kicked: std::sync::Arc, + /// `--repl-backlog-size`, sizes the lazy backlog fallback-init so it + /// can't diverge from the handshake-path allocation. + pub 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). + pub registered: Option>, + /// Live-fanout start offset captured by the pusher AT PUSH TIME, on + /// the shard's own thread (same-thread self-queue pushes only; `None` + /// for the cross-shard legacy registrations, where the arm replies + /// with the offset at drain). Same-thread pushes MUST set this: local + /// writes advance the shard offset synchronously at write time + /// (`record_local_write`), so an offset read at DRAIN could include a + /// write whose `ReplicaLiveFanout` message is queued BEHIND this + /// registration — the catch-up range would cover it AND the fan-out + /// message would deliver it live: double-applied on the replica. + pub push_offset: Option, + /// PER-SHARD exactly-once cut for the fan-out entry + /// ([`ReplicaFanout::cut`]): live records are delivered iff their + /// per-shard `end_offset` exceeds this. `Some` = captured by the + /// pusher in the same synchronous stretch as its snapshot body + /// (single-shard inline path); `None` = the arm reads the shard + /// offset at drain time (legacy cross-shard registrations, where + /// no snapshot body was captured on this shard's thread). + /// NOTE: per-shard axis, NOT the master offset — `push_offset` + /// stays on the master axis for the catch-up reply. + pub cut: Option, +} /// Payload of [`ShardMessage::PrepareReplicaSync`] (R2 multi-shard PSYNC). pub struct PrepareReplicaSyncPayload { @@ -507,40 +560,10 @@ pub enum ShardMessage { BlockCancel { wait_id: u64 }, /// 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 `(id, tx, kicked)` 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, - /// Set by the shard fan-out when this replica's bounded channel is - /// FULL: a record that cannot be queued would otherwise be silently - /// dropped, permanently diverging the replica while - /// `master_link_status` stays "up" (task #35 — observed 2k of 40k - /// keys delivered). The drain task polls the flag and disconnects, - /// converting silent divergence into a loud PSYNC resync (Redis - /// parity: output-buffer-limit disconnects). - kicked: std::sync::Arc, - 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>, - /// Live-fanout start offset captured by the pusher AT PUSH TIME, on - /// the shard's own thread (same-thread self-queue pushes only; `None` - /// for the cross-shard legacy registrations, where the arm replies - /// with the offset at drain). Same-thread pushes MUST set this: local - /// writes advance the shard offset synchronously at write time - /// (`record_local_write`), so an offset read at DRAIN could include a - /// write whose `ReplicaLiveFanout` message is queued BEHIND this - /// registration — the catch-up range would cover it AND the fan-out - /// message would deliver it live: double-applied on the replica. - push_offset: Option, - }, + /// The shard adds a [`ReplicaFanout`] entry to its replica_txs list for + /// WAL fan-out. Boxed: the payload (channels + three offset fields) is + /// past the enum's 64-byte cap. + RegisterReplica(Box), /// 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 }, @@ -570,7 +593,13 @@ pub enum ShardMessage { /// double-applying non-idempotent commands on the replica). This message /// carries ONLY the remaining leg: `try_send` to each registered /// replica's sender channel. Same-thread self-queue only. - ReplicaLiveFanout { bytes: bytes::Bytes }, + ReplicaLiveFanout { + bytes: bytes::Bytes, + /// This shard's replication offset AFTER the record's advance — + /// compared against each [`ReplicaFanout::cut`] at delivery time. + /// `u64::MAX` when no offset handle exists (send unconditionally). + end_offset: u64, + }, /// 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 f3863975..bc4e05d4 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2528,14 +2528,16 @@ pub(crate) fn handle_shard_message_shared( ShardMessage::Shutdown => { info!("Received shutdown via SPSC"); } - ShardMessage::RegisterReplica { - replica_id, - tx, - kicked, - backlog_capacity, - registered, - push_offset, - } => { + ShardMessage::RegisterReplica(payload) => { + let crate::shard::dispatch::RegisterReplicaPayload { + replica_id, + tx, + kicked, + backlog_capacity, + registered, + push_offset, + cut, + } = *payload; // 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 @@ -2555,7 +2557,23 @@ pub(crate) fn handle_shard_message_shared( *guard = Some(ReplicationBacklog::new_at(backlog_capacity, offset)); } drop(guard); - replica_txs.push((replica_id, tx, kicked)); + // Exactly-once cut (per-shard axis): pusher-captured when the + // registration rode with an inline snapshot capture, else the + // shard offset now — every record already fanned out (or whose + // fan-out message precedes this registration in the FIFO) is at + // or below it, so the filtered delivery can't double-send. + let entry_cut = cut.unwrap_or_else(|| { + repl_state + .as_ref() + .map(|h| h.shard_offset(shard_id)) + .unwrap_or(0) + }); + replica_txs.push(crate::shard::dispatch::ReplicaFanout { + replica_id, + tx, + kicked, + cut: entry_cut, + }); // Reply with the offset at which live fan-out begins. For // same-thread self-queue registrations this is `push_offset`, // captured AT PUSH TIME: local writes advance the offset @@ -2582,7 +2600,7 @@ pub(crate) fn handle_shard_message_shared( } } ShardMessage::UnregisterReplica { replica_id } => { - replica_txs.retain(|(id, _, _)| *id != replica_id); + replica_txs.retain(|r| r.replica_id != replica_id); } ShardMessage::PrepareReplicaSync(payload) => { // R2 (task #20): this shard's leg of a multi-shard full resync. @@ -2652,7 +2670,17 @@ pub(crate) fn handle_shard_message_shared( .as_ref() .map(|h| h.shard_offset(shard_id)) .unwrap_or(0); - replica_txs.push((replica_id, tx, kicked)); + // `cut = shard_offset` is the exactly-once line: every mutation + // in the body captured above has already advanced the counter to + // at most this value, so its (possibly still-queued) fan-out + // message is filtered out at delivery; anything applied later + // carries a higher end_offset and is delivered live exactly once. + replica_txs.push(crate::shard::dispatch::ReplicaFanout { + replica_id, + tx, + kicked, + cut: shard_offset, + }); tracing::debug!( shard_id, replica_id, @@ -2673,7 +2701,7 @@ pub(crate) fn handle_shard_message_shared( // The PSYNC task is gone (replica dropped mid-handshake) — // undo the registration so this shard doesn't fan out to a // channel nobody drains. - replica_txs.retain(|(id, _, _)| *id != replica_id); + replica_txs.retain(|r| r.replica_id != replica_id); tracing::warn!( shard_id, replica_id, @@ -2681,13 +2709,20 @@ pub(crate) fn handle_shard_message_shared( ); } } - ShardMessage::ReplicaLiveFanout { bytes } => { + ShardMessage::ReplicaLiveFanout { bytes, end_offset } => { // Live-delivery leg ONLY: backlog append + offset advance already - // happened synchronously at write time on this same thread - // (`record_local_write`) — doing either again here would double- - // count. A replica whose channel is FULL is KICKED (task #35): - // skipping the record would silently and permanently diverge it. - fanout_send_or_kick(replica_txs, &bytes); + // happened synchronously at the write's own execution point on + // this same thread (`record_local_write` for handler-local + // writes, `wal_append_and_fanout` for SPSC-dispatched ones) — + // doing either again here would double-count. ALL live replica + // sends flow through this single arm so the wire order per shard + // equals the self-queue FIFO order equals the offset order — a + // direct send from the execute path would let a LATER-offset + // cross-shard write overtake an earlier local write's queued + // fan-out message, reordering same-key writes on the replica. + // A replica whose channel is FULL is KICKED (task #35): skipping + // the record would silently and permanently diverge it. + fanout_send_or_kick(replica_txs, &bytes, end_offset); } ShardMessage::MigrateConnection(_) => { // MigrateConnection is collected by drain_spsc_shared into pending_migrations, @@ -3504,20 +3539,29 @@ pub(crate) fn wal_fanout_has_work( pub(crate) fn fanout_send_or_kick( replica_txs: &mut Vec, bytes: &bytes::Bytes, + end_offset: u64, ) { - replica_txs.retain(|(id, tx, kicked)| match tx.try_send(bytes.clone()) { - Ok(()) => true, - Err(flume::TrySendError::Full(_)) => { - kicked.store(true, std::sync::atomic::Ordering::Release); - tracing::warn!( - replica_id = id, - "replica live fan-out channel FULL — kicking replica to force a \ - resync (a skipped record would silently diverge it forever)" - ); - false + replica_txs.retain(|r| { + // Exactly-once cut: a record at or below the replica's snapshot cut + // is already inside its FULLRESYNC body — delivering it live would + // double-apply non-idempotent commands (INCR/LPUSH) on the replica. + if end_offset <= r.cut { + return true; + } + match r.tx.try_send(bytes.clone()) { + Ok(()) => true, + Err(flume::TrySendError::Full(_)) => { + r.kicked.store(true, std::sync::atomic::Ordering::Release); + tracing::warn!( + replica_id = r.replica_id, + "replica live fan-out channel FULL — kicking replica to force a \ + resync (a skipped record would silently diverge it forever)" + ); + false + } + // Drain task already gone; just stop queueing. + Err(flume::TrySendError::Disconnected(_)) => false, } - // Drain task already gone; just stop queueing. - Err(flume::TrySendError::Disconnected(_)) => false, }); } @@ -3603,12 +3647,30 @@ pub(crate) fn wal_append_and_fanout( // startup — the per-write advance no longer read-locks the RwLock. // The SELECT prefix counts too: offset accounting must equal the bytes // the replica receives, or WAIT/ACK math diverges. - if let Some(offsets) = repl_state { + let end_offset = if let Some(offsets) = repl_state { let prefix_len = select_prefix.as_ref().map_or(0, |p| p.len()); - offsets.increment_shard_offset(shard_id, (prefix_len + data.len()) as u64); - } - // 4. Fan-out to replica sender tasks (non-blocking: a replica whose - // channel is FULL is kicked to resync — see `fanout_send_or_kick`). + offsets.increment_shard_offset(shard_id, (prefix_len + data.len()) as u64) + } else { + // No offset handle — no cut accounting possible; deliver + // unconditionally (replicas can't exist without repl_state in + // practice, this keeps the degenerate path fail-open). + u64::MAX + }; + // 4. Fan-out to replica sender tasks — DEFERRED through the self queue + // (`ReplicaLiveFanout`), never sent directly from here. Two reasons + // (R2 exactly-once redesign, task #20): + // - Ordering: local handler writes already queue their delivery as + // self-queue messages; a direct send here would put this (later- + // offset) record on the wire BEFORE their (earlier-offset) queued + // bytes — the replica would apply same-key writes out of the + // master's order. + // - Registration cut: a replica registration queued behind this + // drain cycle (self-shard PSYNC leg) would miss a direct send + // entirely — the record is past its snapshot body but not in + // `replica_txs` yet: lost, with the offset advanced (permanent + // replica lag). Deferring one cycle guarantees delivery lands + // after the registration; the per-replica `cut` filter keeps it + // exactly-once. if !replica_txs.is_empty() { let bytes = match &select_prefix { Some(prefix) => { @@ -3619,7 +3681,10 @@ pub(crate) fn wal_append_and_fanout( } None => bytes::Bytes::copy_from_slice(data), }; - fanout_send_or_kick(replica_txs, &bytes); + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { + bytes, + end_offset, + }); } // 5. Per-shard AOF pool (FIX-W1-2): route to the owning shard's writer. // Bounded-blocking (`send_append_bounded_blocking`) because this function @@ -3704,11 +3769,13 @@ mod wal_append_tests { let backlog: SharedBacklog = std::sync::Arc::new(parking_lot::Mutex::new(Some(ReplicationBacklog::new(1024)))); let (tx, _rx) = crate::runtime::channel::mpsc_unbounded::(); - let mut replica_txs: Vec = vec![( - 1u64, - tx, - std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), - )]; + let mut replica_txs: Vec = + vec![crate::shard::dispatch::ReplicaFanout { + replica_id: 1, + tx, + kicked: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + cut: 0, + }]; wal_append_and_fanout( b"hello", diff --git a/tests/aof_multidb_kill9.rs b/tests/aof_multidb_kill9.rs index 77aaec70..3e499f65 100644 --- a/tests/aof_multidb_kill9.rs +++ b/tests/aof_multidb_kill9.rs @@ -100,15 +100,36 @@ impl Conn { } reply.trim_end().to_string() } + + /// Fallible variant for readiness polling: any I/O error (including a + /// RESET on a connection the kernel queued into the bootstrap listener's + /// backlog during moon's bootstrap→per-shard SO_REUSEPORT listener + /// handover — observed on the Linux VM, task #18 flake class) is a + /// "not ready yet", never a panic. + fn try_cmd(&mut self, line: &str) -> Option { + self.reader + .get_mut() + .write_all(format!("{}\r\n", line).as_bytes()) + .ok()?; + let mut reply = String::new(); + self.reader.read_line(&mut reply).ok()?; + Some(reply.trim_end().to_string()) + } } fn wait_ready(addr: &str) { for _ in 0..100 { if let Ok(stream) = TcpStream::connect(addr) { - drop(stream); - let mut c = Conn::connect(addr); - if c.cmd("PING") == "+PONG" { - return; + if stream + .set_read_timeout(Some(Duration::from_secs(5))) + .is_ok() + { + let mut c = Conn { + reader: BufReader::new(stream), + }; + if c.try_cmd("PING").as_deref() == Some("+PONG") { + return; + } } } std::thread::sleep(Duration::from_millis(200)); diff --git a/tests/replication_hardening.rs b/tests/replication_hardening.rs index 1e6e9f16..d54be146 100644 --- a/tests/replication_hardening.rs +++ b/tests/replication_hardening.rs @@ -4,7 +4,9 @@ //! replica kill-restart, and replica promotion paths. //! //! Run: cargo test --test replication_hardening -- --ignored -//! Requires: built moon binary at ./target/release/moon +//! Requires: a built moon binary — set MOON_BIN, or default +//! ./target/release/moon (⚠ on a shared macOS/Linux checkout the default may +//! be the other platform's binary; always pin MOON_BIN, repo harness rule). use std::io::{BufRead, BufReader, Write}; use std::net::TcpStream; @@ -12,9 +14,13 @@ use std::process::{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, extra: &[&str]) -> Guard { Guard( - Command::new("./target/release/moon") + Command::new(moon_bin()) .args( [ &["--port", &port.to_string(), "--shards", "1", "--dir", dir][..], diff --git a/tests/replication_multishard.rs b/tests/replication_multishard.rs index df7d8bec..97276356 100644 --- a/tests/replication_multishard.rs +++ b/tests/replication_multishard.rs @@ -467,6 +467,220 @@ fn multishard_master_graph_snapshot_all_shards() { } } +/// Adversarial-review P0 regression (attach-under-write race): a local write +/// on the ACCEPTING shard that lands between the PSYNC task queueing its +/// self-shard snapshot leg and the event loop draining it is visible to the +/// snapshot body (mutation + offset already applied) while its live fan-out +/// message sits BEHIND the snapshot leg in the same FIFO — so it was +/// delivered twice (in the RDB and again live), double-applying INCR. +/// +/// Hammer counters continuously WHILE the replica attaches; every counter +/// must match the master exactly after convergence. Repeated attaches widen +/// the race window. +#[test] +#[ignore] +fn multishard_master_attach_under_write_no_double_apply() { + let shards = 4; + let (master_port, replica_port) = (17081, 17082); + let mdir = tempfile::tempdir().expect("mdir"); + let master = start_moon_shards(master_port, mdir.path().to_str().unwrap(), shards); + let mut guard = Guard(vec![master]); + let m = format!("127.0.0.1:{}", master_port); + await_ready(&m); + + const COUNTERS: usize = 64; + let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let mut writers = Vec::new(); + for w in 0..4 { + let m = m.clone(); + let stop = stop.clone(); + writers.push(thread::spawn(move || { + let mut stream = TcpStream::connect(&m).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(10))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + while !stop.load(std::sync::atomic::Ordering::Relaxed) { + let mut buf = String::new(); + for i in 0..COUNTERS / 4 { + buf.push_str(&format!("INCR cnt:{}\r\n", w * (COUNTERS / 4) + i)); + } + stream.write_all(buf.as_bytes()).unwrap(); + stream.flush().ok(); + for _ in 0..COUNTERS / 4 { + read_one_reply(&mut reader); + } + } + })); + } + + // Attach (and re-attach) replicas mid-load: each fresh attach runs the + // full multi-shard snapshot fan-out while writes race it. + let rdir = tempfile::tempdir().expect("rdir"); + let replica = start_moon_shards(replica_port, rdir.path().to_str().unwrap(), 1); + guard.0.push(replica); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&r); + for _ in 0..5 { + assert!(send_cmd(&r, "REPLICAOF NO ONE").starts_with("+OK")); + thread::sleep(Duration::from_millis(50)); + assert!(send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)).starts_with("+OK")); + assert!( + wait_until(Duration::from_secs(15), || send_cmd(&r, "INFO replication") + .contains("master_link_status:up")), + "replica link did not come up during attach-under-write" + ); + thread::sleep(Duration::from_millis(300)); + } + + stop.store(true, std::sync::atomic::Ordering::Relaxed); + for w in writers { + w.join().expect("writer"); + } + + // Convergence, then exact per-counter parity. A double-applied INCR + // shows as replica > master for that counter. + let master_vals: Vec = (0..COUNTERS) + .map(|i| { + get_in_db(&m, 0, &format!("cnt:{}", i)) + .and_then(|v| v.parse().ok()) + .unwrap_or(-1) + }) + .collect(); + assert!( + wait_until(Duration::from_secs(30), || { + (0..COUNTERS).all(|i| { + get_in_db(&r, 0, &format!("cnt:{}", i)).and_then(|v| v.parse().ok()) + == Some(master_vals[i]) + }) + }), + "replica counters diverged after attach-under-write: {:?}", + (0..COUNTERS) + .filter_map(|i| { + let rv: i64 = get_in_db(&r, 0, &format!("cnt:{}", i)) + .and_then(|v| v.parse().ok()) + .unwrap_or(-2); + (rv != master_vals[i]).then_some((i, master_vals[i], rv)) + }) + .collect::>() + ); +} + +/// R2 exactly-once redesign regression (D2, same-key wire ordering): before +/// the unified fan-out, a cross-shard (SPSC-dispatched) write was sent to the +/// replica DIRECTLY from the execute arm while a local handler write's +/// delivery sat queued as a self-queue message — so a later-offset write +/// could hit the wire before an earlier-offset write to the SAME key on the +/// same shard. The replica applied them in arrival order and finished with +/// the loser: permanent same-key divergence with byte-exact offsets (WAIT +/// and DBSIZE both look healthy). +/// +/// Four writers on distinct connections APPEND distinguishable tokens to the +/// SAME key set while a replica is attached. APPEND is order-sensitive: ONE +/// reordered pair anywhere in the stream leaves the strings permanently +/// different ("..ab.." vs "..ba.."), so this catches even a single mid-stream +/// swap — a SET-based last-write-wins check only sees a race on the very +/// last pair. Replica must byte-equal the master on every key after quiesce. +#[test] +#[ignore] +fn multishard_master_same_key_write_order_parity() { + let shards = 4; + let (master_port, replica_port) = (17091, 17092); + let mdir = tempfile::tempdir().expect("mdir"); + let master = start_moon_shards(master_port, mdir.path().to_str().unwrap(), shards); + let mut guard = Guard(vec![master]); + let m = format!("127.0.0.1:{}", master_port); + await_ready(&m); + + let rdir = tempfile::tempdir().expect("rdir"); + let replica = start_moon_shards(replica_port, rdir.path().to_str().unwrap(), 1); + guard.0.push(replica); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&r); + assert!(send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)).starts_with("+OK")); + assert!( + wait_until(Duration::from_secs(15), || send_cmd(&r, "INFO replication") + .contains("master_link_status:up")), + "replica link did not come up during ordered-write load" + ); + + const KEYS: usize = 32; + const BURSTS: u64 = 400; + let mut writers = Vec::new(); + // 12 connections: SO_REUSEPORT placement is kernel-hashed, so a handful + // of conns can all land on one shard — enough conns makes mixed + // local + SPSC traffic per key near-certain. + for w in 0..12 { + let m = m.clone(); + writers.push(thread::spawn(move || { + let mut stream = TcpStream::connect(&m).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(10))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + for seq in 0..BURSTS { + let mut buf = String::new(); + // Every writer APPENDs to every key — same-key races between + // connections homed on different shards exercise both the + // local and the SPSC-dispatched write path on each shard. + for k in 0..KEYS { + let tok = format!("w{}:{};", w, seq); + buf.push_str(&format!( + "*3\r\n$6\r\nAPPEND\r\n${}\r\nokey:{}\r\n${}\r\n{}\r\n", + format!("okey:{}", k).len(), + k, + tok.len(), + tok + )); + } + stream.write_all(buf.as_bytes()).unwrap(); + stream.flush().ok(); + for _ in 0..KEYS { + read_one_reply(&mut reader); + } + } + })); + } + + for w in writers { + w.join().expect("writer"); + } + + let master_vals: Vec = (0..KEYS) + .map(|k| get_in_db(&m, 0, &format!("okey:{}", k)).unwrap_or_default()) + .collect(); + assert!( + master_vals.iter().all(|v| !v.is_empty()), + "master lost keys?!" + ); + assert!( + wait_until(Duration::from_secs(30), || { + (0..KEYS).all(|k| { + get_in_db(&r, 0, &format!("okey:{}", k)).as_deref() == Some(&master_vals[k]) + }) + }), + "replica strings diverged (same-key write reorder): {:?}", + (0..KEYS) + .filter_map(|k| { + let rv = get_in_db(&r, 0, &format!("okey:{}", k)).unwrap_or_default(); + (rv != master_vals[k]).then(|| { + // Print the first divergent window, not multi-KB strings. + let mv = &master_vals[k]; + let d = mv + .bytes() + .zip(rv.bytes()) + .position(|(a, b)| a != b) + .unwrap_or(mv.len().min(rv.len())); + let lo = d.saturating_sub(20); + ( + k, + mv.get(lo..(d + 20).min(mv.len())).unwrap_or("").to_string(), + rv.get(lo..(d + 20).min(rv.len())).unwrap_or("").to_string(), + mv.len(), + rv.len(), + ) + }) + }) + .collect::>() + ); +} + /// A multi-shard master must answer ANY resumable PSYNC with +FULLRESYNC (a /// single total offset cannot be mapped back onto N per-shard backlogs), and /// the payload must be ONE merged RDB bulk. @@ -526,3 +740,90 @@ fn multishard_master_partial_resync_degrades_to_full() { assert_eq!(&magic, b"REDIS", "merged snapshot must be Redis-format RDB"); assert!(len > 9, "suspiciously small RDB ({} bytes)", len); } +/// CONTROL: same scenario, single-shard master (R0/R1 path untouched by R2). +#[test] +#[ignore] +fn singleshard_master_attach_under_write_control() { + let shards = 1; + let (master_port, replica_port) = (17085, 17086); + let mdir = tempfile::tempdir().expect("mdir"); + let master = start_moon_shards(master_port, mdir.path().to_str().unwrap(), shards); + let mut guard = Guard(vec![master]); + let m = format!("127.0.0.1:{}", master_port); + await_ready(&m); + + const COUNTERS: usize = 64; + let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let mut writers = Vec::new(); + for w in 0..4 { + let m = m.clone(); + let stop = stop.clone(); + writers.push(thread::spawn(move || { + let mut stream = TcpStream::connect(&m).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(10))).ok(); + let mut reader = BufReader::new(stream.try_clone().expect("clone")); + while !stop.load(std::sync::atomic::Ordering::Relaxed) { + let mut buf = String::new(); + for i in 0..COUNTERS / 4 { + buf.push_str(&format!("INCR cnt:{}\r\n", w * (COUNTERS / 4) + i)); + } + stream.write_all(buf.as_bytes()).unwrap(); + stream.flush().ok(); + for _ in 0..COUNTERS / 4 { + read_one_reply(&mut reader); + } + } + })); + } + + // Attach (and re-attach) replicas mid-load: each fresh attach runs the + // full multi-shard snapshot fan-out while writes race it. + let rdir = tempfile::tempdir().expect("rdir"); + let replica = start_moon_shards(replica_port, rdir.path().to_str().unwrap(), 1); + guard.0.push(replica); + let r = format!("127.0.0.1:{}", replica_port); + await_ready(&r); + for _ in 0..5 { + assert!(send_cmd(&r, "REPLICAOF NO ONE").starts_with("+OK")); + thread::sleep(Duration::from_millis(50)); + assert!(send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {}", master_port)).starts_with("+OK")); + assert!( + wait_until(Duration::from_secs(15), || send_cmd(&r, "INFO replication") + .contains("master_link_status:up")), + "replica link did not come up during attach-under-write" + ); + thread::sleep(Duration::from_millis(300)); + } + + stop.store(true, std::sync::atomic::Ordering::Relaxed); + for w in writers { + w.join().expect("writer"); + } + + // Convergence, then exact per-counter parity. A double-applied INCR + // shows as replica > master for that counter. + let master_vals: Vec = (0..COUNTERS) + .map(|i| { + get_in_db(&m, 0, &format!("cnt:{}", i)) + .and_then(|v| v.parse().ok()) + .unwrap_or(-1) + }) + .collect(); + assert!( + wait_until(Duration::from_secs(30), || { + (0..COUNTERS).all(|i| { + get_in_db(&r, 0, &format!("cnt:{}", i)).and_then(|v| v.parse().ok()) + == Some(master_vals[i]) + }) + }), + "replica counters diverged after attach-under-write: {:?}", + (0..COUNTERS) + .filter_map(|i| { + let rv: i64 = get_in_db(&r, 0, &format!("cnt:{}", i)) + .and_then(|v| v.parse().ok()) + .unwrap_or(-2); + (rv != master_vals[i]).then_some((i, master_vals[i], rv)) + }) + .collect::>() + ); +} From 6839aeca7ae644239b25c71e081636b76fb0f2cb Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 19:02:01 +0700 Subject: [PATCH 5/5] fix(replication): bounded PSYNC prepare wait + unregister on transfer failure (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit round on PR #283: - PrepareReplicaSync reply collection is now bounded (30s timeout per leg): a wedged shard can no longer park the PSYNC task — and the fan-out registrations it holds — forever. On expiry the sync aborts with explicit unregister everywhere; the replica reconnects and retries. - Socket-write failures during the +FULLRESYNC/RDB transfer also unregister on every shard instead of leaving the entries to the passive next-write Disconnected prune. - docs/guides/clustering.md quotes the full tokio PSYNC error text. - CHANGELOG documents the 16K pre-RDB live-buffer limitation (overflow during a very large snapshot under sustained writes KICKS the replica loudly into a retry — never silent divergence). Verified: multishard 9/9, clippy -D warnings, fmt. Refs: task #20, PR #283 author: Tin Dang --- CHANGELOG.md | 6 ++++ docs/guides/clustering.md | 2 +- src/replication/master.rs | 74 ++++++++++++++++++++++++++++----------- 3 files changed, 60 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 59688762..feb1ff59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,12 @@ shards is not yet supported`. with a clear `-ERR PSYNC requires runtime-monoio on the master` instead of an unknown-command reply (RFC R3/2A). Multi-shard *replicas* remain unsupported (`--shards 1`). +- Known limitation: during snapshot preparation + transfer the live stream + buffers in the replica's 16,384-record channel. A very large keyspace + under sustained heavy write load can overflow it mid-attach — the replica + is then KICKED (loud) and retries the sync; it never diverges silently. + Attach such deployments during a write lull, or raise the buffer if this + becomes a practical constraint. - Docs refreshed for the new topology: `docs/guides/clustering.md` deployment shape, `README.md` replication bullets, and `docs/PRODUCTION-CONTRACT.md` rows REPL-MULTISHARD-01 + WAIT-01 flipped to ✅ with evidence. diff --git a/docs/guides/clustering.md b/docs/guides/clustering.md index a6767de3..757588ed 100644 --- a/docs/guides/clustering.md +++ b/docs/guides/clustering.md @@ -19,7 +19,7 @@ Moon implements PSYNC2-compatible replication with per-shard WAL streaming and p stream from all shards; every record carries its own `SELECT` framing, so multi-db workloads replicate exactly. Requires the default `runtime-monoio` build — a `runtime-tokio` master answers PSYNC with - `-ERR PSYNC requires runtime-monoio on the master`. + `-ERR PSYNC requires runtime-monoio on the master (this build runs runtime-tokio)`. - **Replicas:** `--shards 1` each (scale reads by adding replicas, not replica shards). A multi-shard replica refuses to start replication. - **Partial resync:** supported on single-shard masters (backlog window); diff --git a/src/replication/master.rs b/src/replication/master.rs index 1866ab77..9b5d9649 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -942,19 +942,34 @@ pub async fn handle_psync_inline_multi_shard( #[cfg(feature = "graph")] let mut graph_blobs: Vec> = Vec::with_capacity(num_shards); for (shard, reply_rx) in reply_rxs { - let prepared = match reply_rx.recv_async().await { - Ok(p) => p, - Err(_) => { - unregister_replica_all_shards( - replica_id, - &dispatch_tx, - &spsc_notifiers, - self_shard_id, - num_shards, - ); - anyhow::bail!("shard {} dropped its PrepareReplicaSync reply", shard); - } - }; + // Bounded wait (review): a wedged shard must not park this task — + // and its registrations — forever. 30s is far past any observed + // body-serialization time; on expiry the replica reconnects and + // retries the sync. + let prepared = + match monoio::time::timeout(std::time::Duration::from_secs(30), reply_rx.recv_async()) + .await + { + Ok(Ok(p)) => p, + timeout_or_dropped => { + unregister_replica_all_shards( + replica_id, + &dispatch_tx, + &spsc_notifiers, + self_shard_id, + num_shards, + ); + anyhow::bail!( + "shard {} PrepareReplicaSync reply {} — aborting sync", + shard, + if timeout_or_dropped.is_err() { + "timed out after 30s" + } else { + "dropped" + } + ); + } + }; snapshot_offset += prepared.shard_offset; // Index definitions are keyspace-global and identical on every shard — // keep the first non-empty copy. @@ -995,14 +1010,31 @@ pub async fn handle_psync_inline_multi_shard( "multi-shard full resync prepared" ); - 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))?; - let (wr, _) = stream.write_all(rdb_buf).await; - wr.map_err(|e| anyhow::anyhow!(e))?; + // Socket-write failures (replica died mid-transfer) must ALSO unregister + // everywhere — otherwise the fan-out entries linger until each shard's + // next write passively prunes them (review). + let sent: anyhow::Result<()> = async { + 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))?; + let (wr, _) = stream.write_all(rdb_buf).await; + wr.map_err(|e| anyhow::anyhow!(e))?; + Ok(()) + } + .await; + if let Err(e) = sent { + unregister_replica_all_shards( + replica_id, + &dispatch_tx, + &spsc_notifiers, + self_shard_id, + num_shards, + ); + return Err(e); + } // No backlog catch-up leg: each shard's registration IS its snapshot // point (same synchronous stretch), so live fan-out already covers every // byte past `snapshot_offset`.