diff --git a/CHANGELOG.md b/CHANGELOG.md index de170025d..a352a5b9d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/replication/apply.rs b/src/replication/apply.rs index 198c7e3e4..16cd99233 100644 --- a/src/replication/apply.rs +++ b/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::().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 { + (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. diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 6af2ed18b..de0cd7ca9 100644 --- a/src/server/conn/handler_single.rs +++ b/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")) } } diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index c48e37f5e..d88f0d2c7 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -2907,10 +2907,10 @@ pub async fn coordinate_swapdb( aof_pool: Option<&Arc>, 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). diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 949490f64..4d15de862 100644 --- a/src/shard/spsc_handler.rs +++ b/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 " 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| { diff --git a/tests/replication_swapdb.rs b/tests/replication_swapdb.rs new file mode 100644 index 000000000..9230fdd8d --- /dev/null +++ b/tests/replication_swapdb.rs @@ -0,0 +1,300 @@ +//! #386 — streamed SWAPDB must be applied by replicas, exactly once. +//! +//! Two independent defects combine here: +//! +//! 1. **Replica no-op:** `replication/apply.rs::apply_local` had no SWAPDB +//! intercept — the record fell through to generic dispatch, which +//! hard-errors ("SWAPDB must be issued at the connection handler level"), +//! and `warn_on_error` only logs. Every streamed SWAPDB silently no-op'd. +//! +//! 2. **Wire multiplicity:** a multi-shard master's SWAPDB used to reach the +//! replication plane once per REMOTE leg (the SwapDb SPSC arm's +//! `wal_append_and_fanout`) and never for the coordinator's local leg. +//! Today's replica applies the merged wire as ONE stream, record by +//! record — N-1 emissions would swap N-1 times, so the net effect +//! depended on the master's shard-count parity (shards=3 → two swaps → +//! net NO-OP). The wire contract is now: exactly ONE SWAPDB record per +//! client SWAPDB (emitted by the coordinator after its durability gate + +//! local swap); remote legs keep their AOF/WAL writes (per-shard +//! recovery needs them) but stay OFF the replication plane. +//! +//! The multi-shard scenarios run at shards=3 AND shards=4 deliberately: +//! shards=4 (3 remote legs, odd) would accidentally pass under a +//! replica-apply-only fix, while shards=3 (2 remote legs, even) nets to +//! no-op and exposes the multiplicity defect. +//! +//! (When v0.9's #406 lands per-shard demuxed multi-shard replicas, this +//! contract must flip to per-shard emission + per-stream apply — see #386.) + +mod common; + +use std::io::{BufRead, BufReader, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::thread; +use std::time::Duration; + +fn moon_bin() -> std::path::PathBuf { + if let Ok(p) = std::env::var("MOON_BIN") { + return std::path::PathBuf::from(p); + } + std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) +} + +fn start_moon(port: u16, dir: &str, shards: usize, extra: &[&str]) -> Child { + let port_s = port.to_string(); + let shards_s = shards.to_string(); + let mut full: Vec<&str> = vec![ + "--port", + &port_s, + "--shards", + &shards_s, + "--dir", + dir, + "--disk-free-min-pct", + "0", + "--databases", + "4", + ]; + full.extend_from_slice(extra); + Command::new(moon_bin()) + .args(&full) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("Failed to start moon (set MOON_BIN to a built binary)") +} + +struct Guard(Vec); +impl Drop for Guard { + fn drop(&mut self) { + for c in &mut self.0 { + let _ = c.kill(); + let _ = c.wait(); + } + } +} + +fn spawn_into(guard: &mut Guard, dir: &str, shards: usize, extra: &[&str]) -> u16 { + let (child, port) = common::spawn_listening(|port| start_moon(port, dir, shards, extra)); + guard.0.push(child); + port +} + +fn read_one_reply(reader: &mut R) -> String { + let mut line = String::new(); + loop { + line.clear(); + match reader.read_line(&mut line) { + Ok(0) | Err(_) => return String::new(), + Ok(_) => { + let trimmed = line.trim_end_matches("\r\n").trim_end_matches('\n'); + if trimmed.starts_with('+') || trimmed.starts_with('-') || trimmed.starts_with(':') + { + return trimmed.to_string(); + } + if let Some(rest) = trimmed.strip_prefix('$') { + let len: i64 = rest.trim().parse().unwrap_or(-1); + if len < 0 { + return String::new(); // nil + } + let mut buf = vec![0u8; (len as usize) + 2]; + let mut out = String::new(); + if reader.read_exact(&mut buf).is_ok() { + out.push_str(&String::from_utf8_lossy(&buf[..len as usize])); + } + return out; + } + // array/other headers not needed here + } + } + } +} + +/// One connection, several commands in order — SELECT context persists, +/// which per-command connections cannot give us. +fn session_cmds(addr: &str, cmds: &[&str]) -> Vec { + let mut stream = TcpStream::connect(addr).expect("connect"); + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + let mut replies = Vec::with_capacity(cmds.len()); + for cmd in cmds { + stream + .write_all(format!("{cmd}\r\n").as_bytes()) + .expect("write"); + stream.flush().ok(); + let mut reader = BufReader::new(&stream); + replies.push(read_one_reply(&mut reader)); + } + replies +} + +fn send_cmd(addr: &str, cmd: &str) -> String { + session_cmds(addr, &[cmd]).pop().unwrap_or_default() +} + +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 {addr} did not become ready" + ); +} + +fn await_link_up(replica_addr: &str) { + assert!( + wait_until(Duration::from_secs(15), || send_cmd( + replica_addr, + "INFO replication" + ) + .contains("master_link_status:up")), + "replica {replica_addr} link did not come up" + ); +} + +/// GET `key` in logical db `db` on `addr` via one session. +fn get_in_db(addr: &str, db: usize, key: &str) -> String { + session_cmds(addr, &[&format!("SELECT {db}"), &format!("GET {key}")]) + .pop() + .unwrap_or_default() +} + +/// Master (N shards) + replica (1 shard): one client SWAPDB must produce +/// exactly one logical swap on the replica. +fn run_swapdb_replication(master_shards: usize) { + let mdir = tempfile::tempdir().expect("mdir"); + let rdir = tempfile::tempdir().expect("rdir"); + let mut guard = Guard(vec![]); + let master_port = spawn_into( + &mut guard, + mdir.path().to_str().unwrap(), + master_shards, + &["--appendonly", "no"], + ); + let replica_port = spawn_into( + &mut guard, + rdir.path().to_str().unwrap(), + 1, + &["--appendonly", "no"], + ); + let m = format!("127.0.0.1:{master_port}"); + let r = format!("127.0.0.1:{replica_port}"); + await_ready(&m); + await_ready(&r); + + assert!(send_cmd(&r, &format!("REPLICAOF 127.0.0.1 {master_port}")).starts_with("+OK")); + await_link_up(&r); + + // Enough keys in db0 that every master shard owns at least one, plus a + // db1 sentinel — the swap must move ALL of them, including keys owned by + // the coordinator shard (the leg that used to skip the repl plane). + let mut db0_cmds: Vec = vec!["SELECT 0".into()]; + for i in 0..16 { + db0_cmds.push(format!("SET swap:key:{i} before-{i}")); + } + let refs: Vec<&str> = db0_cmds.iter().map(String::as_str).collect(); + for reply in session_cmds(&m, &refs).into_iter().skip(1) { + assert!(reply.starts_with("+OK"), "master SET failed: {reply}"); + } + let one_replies = session_cmds(&m, &["SELECT 1", "SET swap:one only-in-db1"]); + assert!( + one_replies[1].starts_with("+OK"), + "db1 SET: {one_replies:?}" + ); + + // All writes visible on the replica before the swap. + assert!( + wait_until(Duration::from_secs(10), || { + get_in_db(&r, 0, "swap:key:15") == "before-15" + && get_in_db(&r, 1, "swap:one") == "only-in-db1" + }), + "replica did not catch up pre-swap (db0 k15={:?}, db1 one={:?})", + get_in_db(&r, 0, "swap:key:15"), + get_in_db(&r, 1, "swap:one"), + ); + + // The operation under test. + let swap_reply = send_cmd(&m, "SWAPDB 0 1"); + assert!( + swap_reply.starts_with("+OK"), + "[shards={master_shards}] SWAPDB on master failed: {swap_reply}" + ); + + // Master sanity: db0 now holds only the db1 sentinel; db1 holds the 16 keys. + assert_eq!(get_in_db(&m, 0, "swap:one"), "only-in-db1"); + assert_eq!(get_in_db(&m, 1, "swap:key:0"), "before-0"); + assert_eq!( + get_in_db(&m, 0, "swap:key:0"), + "", + "master db0 must not keep swapped key" + ); + + // THE #386 ASSERTION: the replica must converge to the swapped state — + // every db0 key (whatever master shard owned it) now answers from db1, + // and the db1 sentinel answers from db0. A no-op (defect 1, or even + // emission-count under defect 2) leaves keys in their old dbs. + assert!( + wait_until(Duration::from_secs(10), || { + get_in_db(&r, 1, "swap:key:0") == "before-0" + && get_in_db(&r, 1, "swap:key:15") == "before-15" + && get_in_db(&r, 0, "swap:one") == "only-in-db1" + && get_in_db(&r, 0, "swap:key:0").is_empty() + }), + "[shards={master_shards}] replica did not apply SWAPDB exactly once: \ + db1 k0={:?} db1 k15={:?} db0 one={:?} db0 k0={:?}", + get_in_db(&r, 1, "swap:key:0"), + get_in_db(&r, 1, "swap:key:15"), + get_in_db(&r, 0, "swap:one"), + get_in_db(&r, 0, "swap:key:0"), + ); + + // Writes AFTER the swap still replicate into the right (post-swap) dbs. + let post = session_cmds(&m, &["SELECT 0", "SET swap:after post-swap-value"]); + assert!(post[1].starts_with("+OK"), "post-swap SET: {post:?}"); + assert!( + wait_until(Duration::from_secs(10), || get_in_db(&r, 0, "swap:after") + == "post-swap-value"), + "[shards={master_shards}] post-swap write did not replicate into db0" + ); +} + +// `#[ignore]`d like the other replication suites (`replication_streaming.rs`, +// `replication_hardening.rs`): they spawn two real `moon` processes, and +// PSYNC-as-master is monoio-only ("-ERR PSYNC requires runtime-monoio on the +// master"), so they can never pass under the CI tokio job. Run explicitly: +// MOON_BIN=$PWD/target/release/moon cargo test --release \ +// --test replication_swapdb -- --include-ignored + +#[test] +#[ignore] // Requires monoio release binary + real replication link; run explicitly. +fn swapdb_replicates_single_shard_master() { + run_swapdb_replication(1); +} + +/// shards=3: TWO remote legs — under the old per-remote-leg emission a +/// replica-apply-only fix nets to NO-OP (even swap count). The multiplicity +/// half of the defect. +#[test] +#[ignore] // Requires monoio release binary + real replication link; run explicitly. +fn swapdb_replicates_three_shard_master() { + run_swapdb_replication(3); +} + +/// shards=4: THREE remote legs — odd count would accidentally pass an +/// apply-only fix; this leg pins the coordinator-local-leg + exactly-once +/// contract instead (with the fix, exactly one record regardless of shards). +#[test] +#[ignore] // Requires monoio release binary + real replication link; run explicitly. +fn swapdb_replicates_four_shard_master() { + run_swapdb_replication(4); +}