fix(replication): replicas apply streamed SWAPDB, emitted exactly once (#386)

CHANGELOG.md

@@ -7,7 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Fixed
-- **Cross-shard fan-out no longer drops messages silently on a full SPSC ring
+- **Replicas now apply streamed `SWAPDB` (#386), and the record reaches the
+  wire exactly once per client call.** Two stacked defects: (1) the replica's
+  apply path had no SWAPDB intercept — generic dispatch hard-errors ("must be
+  issued at the connection handler level") and the error was only logged, so
+  every streamed SWAPDB silently no-op'd and the replica served pre-swap data
+  for both databases until a full resync; (2) a multi-shard master emitted the
+  record once per REMOTE shard leg and never for the coordinator's own leg —
+  against today's single merged replica stream that means N−1 swaps, a net
+  no-op whenever N−1 is even (e.g. `--shards 3`). The coordinator now emits
+  the replication record exactly once, after the durability gate and the
+  local swap (an aborted SWAPDB can never ship to replicas); remote SPSC legs
+  keep their per-shard AOF/WAL writes but stay off the replication plane; the
+  tokio single-shard handler emits it too. Replicas apply it with the same
+  slice-split swap as WAL replay, skipping (with a warning) indexes outside
+  their own `--databases` range instead of poisoning the stream.
   (c10k E1/E3), and cross-shard reply awaits are bounded (E4).** PUBLISH
   fan-out (immediate, batched, and EXEC-queued) and SCRIPT LOAD propagation
   used a single `try_push` — a transiently-full ring lost the message with no

src/replication/apply.rs

@@ -367,6 +367,17 @@ pub(crate) fn apply_local(
             return apply_temporal_invalidate(s, cmd, args);
         }
 
+        // SWAPDB (#386): the wire carries exactly ONE record per client
+        // SWAPDB (emitted by the master's coordinator after its durability
+        // gate; remote SPSC legs write AOF/WAL only). Generic `dispatch()`
+        // hard-errors on SWAPDB ("must be issued at the connection handler
+        // level") and `warn_on_error` only logs — so without this intercept
+        // every streamed SWAPDB silently no-ops on the replica.
+        if cmd.eq_ignore_ascii_case(b"SWAPDB") {
+            apply_swapdb(cmd, args, &mut s.databases);
+            return true;
+        }
+
         // MOVE / cross-db COPY touch two databases at once and are intercepted
         // BEFORE generic dispatch on the master (see `spsc_two_db`). Generic
         // `dispatch()` cannot apply them — it returns an error for MOVE and
@@ -786,6 +797,39 @@ fn warn_on_error(cmd: &[u8], resp: &Frame) {
     }
 }
 
+/// Apply a streamed `SWAPDB a b` (#386) against the replica's full database
+/// slice — same slice-split swap as the WAL replay intercept
+/// (`persistence/replay.rs`). Out-of-range / same-index / malformed args skip
+/// with a warn: the replica must never poison its stream over an index the
+/// master accepted (e.g. a replica configured with fewer `--databases`), it
+/// just can't honor it.
+fn apply_swapdb(cmd: &[u8], args: &[Frame], databases: &mut [crate::storage::Database]) {
+    let parse_idx = |f: &Frame| match f {
+        Frame::BulkString(b) => std::str::from_utf8(b).ok()?.parse::<usize>().ok(),
+        Frame::Integer(n) => usize::try_from(*n).ok(),
+        _ => None,
+    };
+    match (
+        args.first().and_then(parse_idx),
+        args.get(1).and_then(parse_idx),
+    ) {
+        (Some(a), Some(b)) if a != b && a < databases.len() && b < databases.len() => {
+            let (lo, hi) = if a < b { (a, b) } else { (b, a) };
+            // Split the slice to get two non-overlapping mutable references.
+            let (left, right) = databases.split_at_mut(lo + 1);
+            std::mem::swap(&mut left[lo], &mut right[hi - lo - 1]);
+        }
+        (Some(a), Some(b)) if a == b => {} // same-index: no-op, matches Redis
+        _ => {
+            tracing::warn!(
+                "replication apply: skipping {} with unusable args (out of range for {} local dbs)",
+                String::from_utf8_lossy(cmd),
+                databases.len()
+            );
+        }
+    }
+}
+
 /// Apply `MOVE` / cross-db `COPY ... DB n` on the replica using the same core
 /// helpers as the master's two-db intercept. Returns `None` for a same-db /
 /// no-`DB`-clause COPY (caller falls through to generic dispatch), `Some(resp)`
@@ -1483,6 +1527,74 @@ mod tests {
         assert_eq!(db, 3);
     }
 
+    // ── #386: streamed SWAPDB apply ─────────────────────────────────────
+
+    /// Marker key in db `i` so a swap is observable.
+    fn dbs_with_markers(n: usize) -> Vec<crate::storage::Database> {
+        (0..n)
+            .map(|i| {
+                let mut db = crate::storage::Database::new();
+                db.set(
+                    Bytes::copy_from_slice(format!("marker:{i}").as_bytes()),
+                    crate::storage::Entry::new_string(Bytes::copy_from_slice(
+                        format!("from-db-{i}").as_bytes(),
+                    )),
+                );
+                db
+            })
+            .collect()
+    }
+
+    fn has_marker(db: &mut crate::storage::Database, origin: usize) -> bool {
+        db.get(format!("marker:{origin}").as_bytes()).is_some()
+    }
+
+    #[test]
+    fn apply_swapdb_swaps_databases() {
+        let mut dbs = dbs_with_markers(4);
+        let args = [
+            Frame::BulkString(Bytes::from_static(b"0")),
+            Frame::BulkString(Bytes::from_static(b"2")),
+        ];
+        apply_swapdb(b"SWAPDB", &args, &mut dbs);
+        assert!(has_marker(&mut dbs[0], 2), "db0 must now hold db2's data");
+        assert!(has_marker(&mut dbs[2], 0), "db2 must now hold db0's data");
+        assert!(has_marker(&mut dbs[1], 1), "db1 untouched");
+        assert!(has_marker(&mut dbs[3], 3), "db3 untouched");
+    }
+
+    #[test]
+    fn apply_swapdb_integer_args_and_reversed_order() {
+        let mut dbs = dbs_with_markers(3);
+        // Integer frames + b > a ordering must both work.
+        let args = [Frame::Integer(2), Frame::Integer(1)];
+        apply_swapdb(b"SWAPDB", &args, &mut dbs);
+        assert!(has_marker(&mut dbs[1], 2));
+        assert!(has_marker(&mut dbs[2], 1));
+    }
+
+    #[test]
+    fn apply_swapdb_out_of_range_and_same_index_are_noops() {
+        let mut dbs = dbs_with_markers(2);
+        // Out of range for this replica's db_count — skip, don't panic.
+        let oor = [Frame::Integer(0), Frame::Integer(9)];
+        apply_swapdb(b"SWAPDB", &oor, &mut dbs);
+        assert!(has_marker(&mut dbs[0], 0));
+        // Same index — no-op.
+        let same = [Frame::Integer(1), Frame::Integer(1)];
+        apply_swapdb(b"SWAPDB", &same, &mut dbs);
+        assert!(has_marker(&mut dbs[1], 1));
+        // Malformed (missing / non-numeric args) — skip, don't panic.
+        apply_swapdb(b"SWAPDB", &[], &mut dbs);
+        let junk = [
+            Frame::BulkString(Bytes::from_static(b"x")),
+            Frame::Integer(1),
+        ];
+        apply_swapdb(b"SWAPDB", &junk, &mut dbs);
+        assert!(has_marker(&mut dbs[0], 0));
+        assert!(has_marker(&mut dbs[1], 1));
+    }
+
     #[test]
     fn integer_select_arg_parses() {
         // SELECT sent with an integer arg instead of bulk string.

src/server/conn/handler_single.rs

@@ -865,29 +865,27 @@ pub async fn handle_connection(
                                         //   - appendfsync=everysec/no → fire-and-forget (fast)
                                         // On any Err the caller aborts and leaves both DBs
                                         // untouched, preserving atomicity from the WAL's perspective.
+                                        let mut a_buf = itoa::Buffer::new();
+                                        let mut b_buf = itoa::Buffer::new();
+                                        let wal_frame = Frame::Array(crate::framevec![
+                                            Frame::BulkString(Bytes::from_static(b"SWAPDB")),
+                                            Frame::BulkString(Bytes::copy_from_slice(
+                                                a_buf.format(a).as_bytes()
+                                            )),
+                                            Frame::BulkString(Bytes::copy_from_slice(
+                                                b_buf.format(b).as_bytes()
+                                            )),
+                                        ]);
+                                        let serialized =
+                                            crate::persistence::aof::serialize_command(&wal_frame);
                                         let wal_ok = if let Some(ref pool) = aof_pool {
-                                            let mut a_buf = itoa::Buffer::new();
-                                            let mut b_buf = itoa::Buffer::new();
-                                            let wal_frame = Frame::Array(crate::framevec![
-                                                Frame::BulkString(Bytes::from_static(b"SWAPDB")),
-                                                Frame::BulkString(Bytes::copy_from_slice(
-                                                    a_buf.format(a).as_bytes()
-                                                )),
-                                                Frame::BulkString(Bytes::copy_from_slice(
-                                                    b_buf.format(b).as_bytes()
-                                                )),
-                                            ]);
-                                            let serialized =
-                                                crate::persistence::aof::serialize_command(
-                                                    &wal_frame,
-                                                );
                                             // Single-shard mode — shard_id = 0.
                                             let lsn = crate::persistence::aof::AofWriterPool::issue_append_lsn(&repl_state, 0, serialized.len());
                                             // task #35: SWAPDB affects both `a` and
                                             // `b` — no single db context applies;
                                             // pass 0 (writer may emit a harmless
                                             // redundant SELECT 0).
-                                            pool.try_send_append_durable(0, lsn, 0, serialized)
+                                            pool.try_send_append_durable(0, lsn, 0, serialized.clone())
                                                 .await
                                                 .is_ok()
                                         } else {
@@ -904,6 +902,14 @@ pub async fn handle_connection(
                                             let mut guard_lo = db[lo].write();
                                             let mut guard_hi = db[hi].write();
                                             std::mem::swap(&mut *guard_lo, &mut *guard_hi);
+                                            drop(guard_hi);
+                                            drop(guard_lo);
+                                            // #386 — replication plane, exactly once per
+                                            // client SWAPDB, AFTER the durability gate and
+                                            // the swap itself (mirrors the coordinator leg).
+                                            crate::replication::state::record_local_write_global(
+                                                0, serialized,
+                                            );
                                             Frame::SimpleString(Bytes::from_static(b"OK"))
                                         }
                                     }

src/shard/coordinator.rs

@@ -2907,10 +2907,10 @@ pub async fn coordinate_swapdb(
     aof_pool: Option<&Arc<crate::persistence::aof::AofWriterPool>>,
     repl_state: ReplStateRef<'_>,
 ) -> Frame {
-    debug_assert!(
-        num_shards > 1,
-        "coordinate_swapdb is the multi-shard path (single-shard SWAPDB lives in handler_single.rs)"
-    );
+    // Serves num_shards >= 1: the tokio single-shard SWAPDB lives in
+    // handler_single.rs, but the monoio handler routes ALL shard counts here
+    // (at shards=1 the remote loop is simply empty).
+    debug_assert!(num_shards >= 1);
     // Local shard first: durable append BEFORE the swap AND before any
     // remote dispatch. SWAPDB has no command-level rollback; anything that
     // can fail must fail while NOTHING in the cluster has mutated —
@@ -2949,7 +2949,10 @@ pub async fn coordinate_swapdb(
                 my_shard,
                 serialized.len(),
             );
-            match pool.send_append_group(my_shard, lsn, 0, serialized).await {
+            match pool
+                .send_append_group(my_shard, lsn, 0, serialized.clone())
+                .await
+            {
                 Ok(needs_barrier) => {
                     if needs_barrier && pool.fsync_barrier(my_shard).await.is_err() {
                         return Frame::Error(bytes::Bytes::from_static(
@@ -2972,6 +2975,18 @@ pub async fn coordinate_swapdb(
                 s.databases.swap(a, b);
             }
         });
+
+        // #386 — replication plane, exactly once per client SWAPDB. Today's
+        // replica applies the merged wire as ONE stream, so the record must
+        // appear on it exactly once: the coordinator emits it here, AFTER
+        // the durability gate (an abort above never reaches this line, so a
+        // failed SWAPDB can never ship to replicas) and after the local
+        // swap; the remote legs' SPSC arms write AOF/WAL only. Safe on both
+        // runtimes: this runs on the shard's own OS thread (monoio shard
+        // thread / tokio per-shard LocalSet), whose event loop drains
+        // `self_msg`. When #406 lands per-shard demuxed replicas this must
+        // flip to per-shard emission.
+        crate::replication::state::record_local_write_global(my_shard, serialized);
     }
 
     // ChannelMesh has no self-send slot (target_index panics when my_id == target_id).

src/shard/spsc_handler.rs

@@ -2449,7 +2449,7 @@ pub(crate) fn handle_shard_message_shared(
             // WAL-before-swap: emit the SWAPDB record so that crash-recovery
             // replay can re-apply the swap in the correct order.  The record
             // is written even when wal_writer/wal_writer are None (the
-            // fast-path in wal_append_and_fanout will skip it cheaply).
+            // fast-path skips it cheaply).
             //
             // Serialise "SWAPDB <a> <b>" without heap allocation on the number
             // formatting (itoa writes into a stack buffer).
@@ -2463,25 +2463,40 @@ pub(crate) fn handle_shard_message_shared(
                 crate::protocol::Frame::BulkString(bytes::Bytes::copy_from_slice(b_str.as_bytes())),
             ]);
             let serialized = aof::serialize_command(&wal_frame);
+            // #386 exactly-once wire contract: remote legs write ONLY the
+            // durability planes (WAL v3 + this shard's AOF — per-shard
+            // recovery replays each shard's own record). The REPLICATION
+            // plane record is emitted exactly once, by the coordinator's
+            // local leg (`coordinate_swapdb`), because today's replica
+            // applies the merged wire as one stream: per-remote-leg emission
+            // made the replica swap N-1 times (net no-op at odd shard
+            // counts). When #406 lands per-shard demuxed replicas this must
+            // flip back to per-shard emission.
+            //
             // No per-client response frame exists here (coordinator broadcast,
             // reply is `()`): an AOF-append loss is already counted +
             // error!-logged inside the pool, so the result is discarded.
-            let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
-            let _ = wal_append_and_fanout(
-                &serialized,
+            if wal_kv_log {
+                if let Some(w) = wal_writer {
+                    w.append(
+                        crate::persistence::wal_v3::record::WalRecordType::Command,
+                        &serialized,
+                    );
+                }
+            }
+            if let Some(pool) = aof_pool {
+                let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
                 // task #35: SWAPDB affects both `a` and `b` — no single db
                 // context applies; pass 0 (writer may emit a harmless
                 // redundant SELECT 0 if last_db was already non-zero).
-                0,
-                wal_writer,
-                repl_backlog,
-                replica_txs,
-                repl_state,
-                shard_id,
-                aof_pool, // FIX-W1-2
-                wal_kv_log,
-                &mut aof_budget,
-            );
+                let _ = pool.send_append_bounded_blocking(
+                    shard_id,
+                    0,
+                    0,
+                    bytes::Bytes::copy_from_slice(&serialized),
+                    &mut aof_budget,
+                );
+            }
 
             // Perform the in-place swap via ShardSlice (thread-local, no locks needed).
             crate::shard::slice::with_shard(|s| {