diff --git a/CHANGELOG.md b/CHANGELOG.md index b3a6c910..feb1ff59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,98 @@ 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). 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`). +- 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. + +### 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/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..757588ed 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 (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); + 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/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..9b5d9649 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); } @@ -799,6 +809,293 @@ 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)); + + // ── 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) = + 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 { + // 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); + 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 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; + #[cfg(feature = "graph")] + let mut graph_blobs: Vec> = Vec::with_capacity(num_shards); + for (shard, reply_rx) in reply_rxs { + // 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. + 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" + ); + + // 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`. + + 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. + 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; + } + 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(); + } + } +} + /// 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 +1161,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 @@ -900,18 +1201,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 3ec31636..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 @@ -276,6 +293,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..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(); @@ -541,10 +551,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 +569,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..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 @@ -103,6 +110,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/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 593dc98b..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(); @@ -713,3 +723,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..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(); @@ -859,6 +863,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 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..61086728 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -411,16 +411,110 @@ 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 { + /// 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. @@ -466,43 +560,29 @@ 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 }, + /// 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 @@ -513,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 f5d572d4..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,15 +2600,129 @@ 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::ReplicaLiveFanout { bytes } => { + 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). + // + // 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, + 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); + // `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, + 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(|r| r.replica_id != replica_id); + tracing::warn!( + shard_id, + replica_id, + "PrepareReplicaSync reply dropped — replica disconnected before sync" + ); + } + } + 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, @@ -3407,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, }); } @@ -3471,6 +3612,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 +3635,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,14 +3645,46 @@ 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. - if let Some(offsets) = repl_state { - offsets.increment_shard_offset(shard_id, 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`). + // The SELECT prefix counts too: offset accounting must equal the bytes + // the replica receives, or WAIT/ACK math diverges. + 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) + } 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 = bytes::Bytes::copy_from_slice(data); - fanout_send_or_kick(replica_txs, &bytes); + 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), + }; + 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 @@ -3579,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 new file mode 100644 index 00000000..97276356 --- /dev/null +++ b/tests/replication_multishard.rs @@ -0,0 +1,829 @@ +//! 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)"]) + ); + } +} + +/// 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. +#[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); +} +/// 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::>() + ); +}