feat(replication): R2 multi-shard master PSYNC — merged RDB + per-record SELECT framing

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 <replid> <Σ shard offsets>`
+  with one `$<len>` 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 <db>` 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 = <shard offset at body capture>` 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

README.md

@@ -325,15 +325,15 @@ 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).
 
 **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.

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 |

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 <offset>` (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
 

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<u8>) {
     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 `$<len>` bulk so a single-shard
+/// replica loads it with the unchanged R0 path.
+pub fn write_rdb_merged(moon_aux: &[(&[u8], &[u8])], bodies: &[Vec<u8>], buf: &mut Vec<u8>) {
+    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<Vec<u8>> {
     }
 }
 
+/// 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<Vec<u8>> {
+    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::<Vec<_>>(), &mut body0);
+        let mut body1 = Vec::new();
+        write_rdb_body_refs(&shard1.iter().collect::<Vec<_>>(), &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);