From ec35cba71dd07616495e56af4ba4ebe9ae81634f Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 18 Jul 2026 16:12:26 +0700 Subject: [PATCH 1/2] =?UTF-8?q?fix(shard):=20multi-shard=20SWAPDB=20?= =?UTF-8?q?=E2=80=94=20durable=20+=20replica-visible=20before=20+OK=20(#13?= =?UTF-8?q?3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit coordinate_swapdb had two defects. Defect 1 (the issue's report): no fsync rendezvous anywhere — the local leg fired its record into the generic wal_append_txs channel and remote legs acked right after enqueueing, so under --appendfsync always the client saw +OK before any shard had fsynced. Defect 2 (found during scoping, empirically confirmed): the local leg's channel drains into WAL v3 ONLY, but for --shards >= 2 --appendonly yes recovery wipes every shard and replays ONLY the per-shard AOF manifest — the coordinator shard's own SWAPDB record never reached the plane recovery reads. A kill-9 after +OK permanently lost the LOCAL shard's half of the swap while remote shards' halves survived: cross-shard keyspace divergence, strictly worse than a fsync-timing gap. The local leg also never reached the replication plane (no backlog append, no offset advance, no live fan-out), unlike remote legs. Fix: - Local leg: durable AOF append via send_append_group (the same v3-5 group-commit primitive persist_local_leg uses) + one fsync_barrier under Always, BEFORE the swap — any failure aborts with no local mutation, mirroring the single-shard handler_single contract. The WAL v3 write is kept (other record types share the channel; it is a documented non-authority for this deployment shape). Replication: record_local_swapdb_repl appends to the per-shard backlog and advances the offset synchronously, then defers the live-replica send via ShardMessage::ReplicaLiveFanout — the same backlog-then-defer contract wal_append_and_fanout and record_local_write_db use — with a SELECT 0 prefix (R2 multi-shard framing; SWAPDB has no single-db context, task #35 precedent). - Remote legs: the SwapDb SPSC arm already enqueues to that shard's AOF writer strictly before acking; the coordinator now issues one fsync_barrier(target) after observing each ack (H1-BARRIER happens-before ordering: barrier queues behind the append in the same FIFO channel, so an acked barrier proves the record durable). Barriers no-op under everysec/no — no cost off the Always policy. - Failure semantics: local durability failure aborts before any mutation; a remote barrier failure lands AFTER that shard's swap (SWAPDB has no rollback) and is reported truthfully as durability-unconfirmed instead of a false +OK. All remote legs are drained even after one leg errors (coordinate_mset pattern). Red/green: crash_133_swapdb_multishard_durability_after_sigkill (2-shard, appendfsync always, distinct db0/db1 content on both shards via hash tags, SWAPDB 0 1, +OK, zero-delay SIGKILL, restart) — RED on unfixed code with the local shard's entire swap half lost (db0 keys kept db0 values), GREEN 8/8 with the fix. Unit tests prove exactly-once backlog append + offset advance with SELECT-0 framing, no-op without repl_state, and inactive-fanout gating. Fixes #133 author: Tin Dang --- CHANGELOG.md | 18 ++ src/server/conn/handler_monoio/dispatch.rs | 2 + src/server/conn/handler_sharded/dispatch.rs | 2 + src/shard/coordinator.rs | 323 ++++++++++++++++++-- tests/crash_matrix_per_shard_aof.rs | 164 ++++++++++ 5 files changed, 487 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b1808215..1051f754 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed +- **Multi-shard SWAPDB is now durable and replica-visible before `+OK` + (#133).** Two defects in `coordinate_swapdb`: (1) no fsync rendezvous — + under `--appendfsync always` the client saw `+OK` before any shard had + fsynced its SWAPDB record; (2) the coordinator shard's own record was + written to WAL v3 only, never the per-shard AOF — and for + `--shards ≥ 2 --appendonly yes` the per-shard AOF manifest is the sole + recovery authority, so a kill-9 after `+OK` permanently lost the LOCAL + shard's half of the swap while remote shards' halves survived + (cross-shard keyspace divergence; reproduced empirically). The local leg + now performs a durable AOF group-commit append + fsync barrier BEFORE + swapping (abort-with-no-mutation on failure) and records the swap on the + replication plane (backlog + offset + live fan-out, `SELECT 0` framing); + the coordinator confirms each remote shard's durability with one + post-ack `fsync_barrier` (H1-BARRIER ordering). A remote barrier failure + after the swap is applied is reported truthfully as + durability-unconfirmed rather than a false `+OK`. + ### Performance - **Shard offload paths are precomputed at shard init — the recurring tick paths no longer allocate (#45).** The 100ms eviction tick, the diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index d17312cd..ad005048 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -1306,6 +1306,8 @@ pub(super) async fn try_handle_swapdb( &ctx.shard_databases, &ctx.dispatch_tx, &ctx.spsc_notifiers, + ctx.aof_pool.as_ref(), + &ctx.repl_state, ) .await; responses.push(response); diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index e4d9e4b5..7a5fc362 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -651,6 +651,8 @@ pub(super) async fn try_handle_swapdb( &ctx.shard_databases, &ctx.dispatch_tx, &ctx.spsc_notifiers, + ctx.aof_pool.as_ref(), + &ctx.repl_state, ) .await; responses.push(response); diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index c3b204d1..83615883 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -2759,22 +2759,135 @@ pub(crate) fn aggregate_doc_freq( (global_df, global_n) } -/// Broadcast SWAPDB to all shards and await acknowledgement from each. +/// Cheap precondition for [`record_local_swapdb_repl`]: skip replication +/// bookkeeping entirely when no replica has ever attached. Ctx-free mirror +/// of `handler_monoio::ft::replication_fanout_active` — the coordinator has +/// no `ConnectionContext` in scope. +fn swapdb_repl_active(repl_state: ReplStateRef<'_>) -> bool { + if !crate::replication::state::fanout_hint_active() { + return false; + } + repl_state.as_ref().is_some_and(|rs| { + let g = rs.read(); + !g.replicas.is_empty() + || g.per_shard_backlogs + .first() + .is_some_and(|slot| slot.lock().is_some()) + }) +} + +/// Record the coordinator LOCAL leg's SWAPDB record on the replication +/// plane: backlog append + per-shard/master offset advance, synchronously +/// (before any `.await`), then defer the live-replica send onto the shard's +/// self-queue (`ShardMessage::ReplicaLiveFanout`) — the same +/// backlog-then-defer contract `wal_append_and_fanout` uses for remote legs +/// and `record_local_write_db` uses for ordinary local single-key writes. +/// +/// `coordinate_swapdb` only runs when `num_shards > 1` (single-shard SWAPDB +/// never reaches this module — see `handler_single.rs`), so every record is +/// framed with its own `SELECT 0` prefix (R2, task #20): a multi-shard +/// master merges N shard streams onto one replica wire, so no shared +/// "current db" context can exist. `db=0` mirrors the remote legs' +/// `wal_append_and_fanout(serialized, 0, ...)` call (task #35: SWAPDB +/// touches both `a` and `b` — no single db context applies). +/// +/// Caller MUST be running on `my_shard`'s own OS thread (pushes to +/// `shard::self_msg`, a `thread_local!` queue) — guaranteed here since +/// `coordinate_swapdb` always executes on the connection's own shard. +fn record_local_swapdb_repl(repl_state: ReplStateRef<'_>, my_shard: usize, data: &Bytes) { + let Some(rs) = repl_state.as_ref() else { + return; + }; + let select = crate::persistence::aof::serialize_select_record(0); + let mut combined = Vec::with_capacity(select.len() + data.len()); + combined.extend_from_slice(&select); + combined.extend_from_slice(data); + let combined = Bytes::from(combined); + + let g = rs.read(); + if let Some(slot) = g.per_shard_backlogs.get(my_shard) { + if let Some(backlog) = slot.lock().as_mut() { + backlog.append(&combined); + } + } + let end_offset = g.increment_shard_offset(my_shard, combined.len() as u64); + drop(g); + crate::shard::self_msg::push(ShardMessage::ReplicaLiveFanout { + bytes: combined, + end_offset, + }); +} + +/// Broadcast SWAPDB to all shards and await acknowledgement AND durability +/// from each (issue #133). /// /// # Flow /// -/// - Local shard: inline swap under ascending-index write locks (no SPSC round-trip). -/// - Remote shards: send `ShardMessage::SwapDb` and collect oneshot replies. +/// - Local shard: durable AOF append (v3-5 group-commit + barrier) BEFORE +/// the swap — mirrors the single-shard `handler_single.rs` SWAPDB +/// contract so the coordinator shard's own record survives a kill-9. +/// - Remote shards: send `ShardMessage::SwapDb` (its SPSC arm durably logs +/// via `wal_append_and_fanout` — WAL v3 + repl backlog/offset/fan-out + +/// AOF — BEFORE swapping and acking), then this function issues ONE +/// `fsync_barrier(target)` per remote shard AFTER observing its ack, +/// confirming the fsync the SPSC arm could not await inline (SPSC arms +/// run synchronously inside the shard event loop — they cannot `.await`). /// -/// All-shard acks are awaited before returning `+OK`. Between the first and -/// last ack a brief window exists where a cross-shard GET may observe the -/// pre-swap state on one shard and post-swap on another. This matches Redis -/// cluster relaxed semantics and is documented as the "brief-skew" acceptance. +/// All-shard acks AND (under `appendfsync=always`) all-shard fsync barriers +/// are awaited before returning `+OK`. Between the first and last ack a +/// brief window exists where a cross-shard GET may observe the pre-swap +/// state on one shard and post-swap on another. This matches Redis cluster +/// relaxed semantics and is documented as the "brief-skew" acceptance. /// -/// # Consistency note +/// # Durability (defect 1: no fsync rendezvous) /// -/// WAL is emitted by each shard's SPSC handler *before* performing the swap, -/// so crash-recovery replay applies them in the correct order. +/// `appendfsync=always` requires every shard's SWAPDB record to be fsynced +/// to disk BEFORE the client observes `+OK`. This function: +/// 1. Appends the LOCAL shard's record via +/// `AofWriterPool::send_append_group` (the same v3-5 group-commit +/// primitive [`persist_local_leg`] uses for MSET/BITOP/COPY/DEL local +/// legs) and, when `Always`, awaits ONE `fsync_barrier(my_shard)` +/// BEFORE performing the swap — a failure aborts with NO mutation +/// applied anywhere on this shard yet. +/// 2. Dispatches `ShardMessage::SwapDb` to every remote shard. +/// 3. After EACH remote ack, issues ONE `fsync_barrier(target)`. Because +/// the remote's `Append` was enqueued into that shard's AOF writer +/// channel strictly before its `reply_tx.send(())` — observed here via +/// `rx.recv().await`, a happens-before edge — and the writer processes +/// that channel in FIFO order regardless of which thread produced each +/// message, the barrier's `AppendSync` is guaranteed to be queued +/// AFTER the remote's `Append`: an acked barrier proves it durable. +/// This is the identical ordering proof the existing H1-BARRIER +/// cross-shard write path already relies on (see +/// `handler_monoio/mod.rs`'s "call fsync_barrier once per target shard +/// AFTER responses are collected"). +/// +/// A remote barrier failure occurs AFTER that shard's swap was already +/// applied — there is no rollback path (SWAPDB has none). The response +/// truthfully reports durability-unconfirmed (mirrors the F2 bounded-wait +/// discipline documented on `try_send_append_durable`) rather than a false +/// `+OK`; the swap itself is NOT undone. `fsync_barrier` internally no-ops +/// under `EverySec`/`No`, so every barrier call below is unconditional and +/// only actually awaits a disk fsync when the policy is `Always`. +/// +/// # Local-leg AOF/recovery gap (defect 2) +/// +/// Before this fix the local leg only wrote to the generic +/// `wal_append_txs` channel (`ShardDatabases::try_wal_append_required`), +/// which the event loop drains into WAL v3 ONLY — never the per-shard AOF +/// writer (`AofWriterPool`). For `--shards >= 2 --appendonly yes`, +/// `main.rs`'s `replay_per_shard` — NOT WAL v3 — is the recovery authority: +/// it unconditionally wipes every shard's databases and replays ONLY from +/// the per-shard AOF manifest (`appendonlydir/shard-N/`). The coordinator +/// shard's own SWAPDB record therefore never reached the plane recovery +/// actually reads: a kill-9 right after `+OK` permanently lost the LOCAL +/// shard's half of the swap on restart while remote shards' halves +/// survived — cross-shard keyspace divergence, worse than a mere fsync +/// gap. The WAL v3 write below is KEPT (harmless — other record types on +/// this channel need it, and it is a documented non-authority for this +/// deployment shape) but is no longer the only durability plane the local +/// leg touches. +#[allow(clippy::too_many_arguments)] pub async fn coordinate_swapdb( a: usize, b: usize, @@ -2783,10 +2896,17 @@ pub async fn coordinate_swapdb( shard_databases: &Arc, dispatch_tx: &Rc>>>, spsc_notifiers: &[Arc], + 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)" + ); // ChannelMesh has no self-send slot (target_index panics when my_id == target_id). // Skip self in the SPSC loop; handle the local shard inline below. let remote_count = num_shards.saturating_sub(1); + let mut targets: Vec = Vec::with_capacity(remote_count); let mut receivers: Vec> = Vec::with_capacity(remote_count); for target in 0..num_shards { @@ -2796,14 +2916,14 @@ pub async fn coordinate_swapdb( let (reply_tx, reply_rx) = channel::oneshot(); let msg = ShardMessage::SwapDb { a, b, reply_tx }; let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + targets.push(target); receivers.push(reply_rx); } - // Local shard: emit WAL via the per-shard append channel, then swap databases. - // This mirrors the spsc_handler SwapDb arm — WAL record BEFORE the swap. - // SWAPDB has no command-level rollback; if persistence is configured and - // the WAL channel rejects the enqueue (full / closed), we MUST NOT perform - // the local swap, otherwise the cluster state diverges from the on-disk log. + // Local shard: durable append BEFORE the swap. SWAPDB has no + // command-level rollback; anything that can fail must fail BEFORE + // mutating this shard's state, otherwise the cluster state diverges + // from the on-disk log. { let mut a_buf = itoa::Buffer::new(); let mut b_buf = itoa::Buffer::new(); @@ -2813,15 +2933,59 @@ pub async fn coordinate_swapdb( Frame::BulkString(Bytes::copy_from_slice(b_buf.format(b).as_bytes())), ]); let serialized = crate::persistence::aof::serialize_command(&wal_frame); + + // WAL v3 — see "Local-leg AOF/recovery gap" doc above: kept for + // parity with other record types on this channel, but no longer the + // sole durability plane. if !shard_databases.try_wal_append_required( my_shard, crate::persistence::wal_v3::record::WalRecordType::Command, - serialized, + serialized.clone(), ) { return Frame::Error(bytes::Bytes::from_static( b"ERR SWAPDB aborted: WAL enqueue failed (persistence backpressure)", )); } + + // AOF pool — the actual multi-shard recovery authority (defect 2). + // Same v3-5 group-commit contract `persist_local_leg` uses for + // MSET/BITOP/COPY/DEL: enqueue fire-and-forget, then ONE barrier + // under `Always` instead of a per-write awaited fsync. + if let Some(pool) = aof_pool { + let repl_active = swapdb_repl_active(repl_state); + let lsn = if repl_active { + // Records backlog + offset synchronously (before any + // .await) and defers only the live-replica send — matches + // the local single-key write contract + // (`record_local_write_db`). lsn=0: the AOF leg must not + // double-advance the offset this already advanced. + record_local_swapdb_repl(repl_state, my_shard, &serialized); + 0 + } else { + crate::persistence::aof::AofWriterPool::issue_append_lsn( + repl_state, + my_shard, + serialized.len(), + ) + }; + match pool.send_append_group(my_shard, lsn, 0, serialized).await { + Ok(needs_barrier) => { + if needs_barrier && pool.fsync_barrier(my_shard).await.is_err() { + return Frame::Error(bytes::Bytes::from_static( + crate::persistence::aof::AOF_FSYNC_ERR, + )); + } + } + Err(_) => { + return Frame::Error(bytes::Bytes::from_static( + crate::persistence::aof::AOF_FSYNC_ERR, + )); + } + } + } + + // Local durability confirmed (or persistence disabled) — apply the + // swap. crate::shard::slice::with_shard(|s| { if a != b { s.databases.swap(a, b); @@ -2829,17 +2993,39 @@ pub async fn coordinate_swapdb( }); } - // Await all-remote-shard acks before returning +OK. - for rx in receivers { + // Await every remote shard's ack, then confirm ITS durability with one + // fsync_barrier (H1-BARRIER pattern) — the SwapDb SPSC arm cannot + // `.await` a barrier itself, so the coordinator closes that gap here, + // after observing the ack (ordering proof in the doc comment above). + // Every leg is drained even after a failure (mirrors `coordinate_mset`): + // a timed-out/closed/unconfirmed leg must not collapse into a false OK, + // but it also must not skip confirming the OTHER shards. + let mut leg_err: Option = None; + for (target, rx) in targets.into_iter().zip(receivers) { match rx.recv().await { - Ok(()) => {} + Ok(()) => { + if let Some(pool) = aof_pool + && pool.fsync_barrier(target).await.is_err() + && leg_err.is_none() + { + leg_err = Some(Frame::Error(bytes::Bytes::from_static( + b"ERR SWAPDB durability unconfirmed on a remote shard \ + (fsync barrier failed after the swap was already applied)", + ))); + } + } Err(_) => { - return Frame::Error(bytes::Bytes::from_static( - b"ERR cross-shard reply channel closed during SWAPDB", - )); + if leg_err.is_none() { + leg_err = Some(Frame::Error(bytes::Bytes::from_static( + b"ERR cross-shard reply channel closed during SWAPDB", + ))); + } } } } + if let Some(err) = leg_err { + return err; + } Frame::SimpleString(bytes::Bytes::from_static(b"OK")) } @@ -2889,6 +3075,99 @@ mod tests { ); } + // ── issue #133: SWAPDB local-leg replication plane ────────────────── + // + // `record_local_swapdb_repl` is the coordinator's ctx-free equivalent of + // `record_local_write_db` for the SWAPDB local leg. These tests exercise + // it directly (no shard mesh / event loop needed) to prove EXACTLY-ONCE + // backlog append + offset advance per call — a double-count here would + // desync `master_repl_offset` from the sum of bytes actually delivered, + // and SWAPDB is explicitly non-idempotent (double-logging `a b` nets a + // no-op swap on replay). + #[test] + fn test_record_local_swapdb_repl_appends_exactly_once_with_select_prefix() { + use crate::replication::state::ReplicationState; + + let repl_state = Arc::new(parking_lot::RwLock::new(ReplicationState::new( + 2, + "a".repeat(40), + "0".repeat(40), + ))); + // Simulate a replica already attached on shard 0 (lazy backlog alloc). + repl_state.read().ensure_backlogs_allocated(); + let repl_state_opt: Option>> = + Some(repl_state.clone()); + + let data = Bytes::from_static(b"*3\r\n$6\r\nSWAPDB\r\n$1\r\n0\r\n$1\r\n1\r\n"); + record_local_swapdb_repl(&repl_state_opt, 0, &data); + + let select = crate::persistence::aof::serialize_select_record(0); + let expected_len = (select.len() + data.len()) as u64; + + let g = repl_state.read(); + // Offset advanced by EXACTLY one SELECT+payload's worth of bytes — + // not zero (would mean the call was a no-op) and not double + // (would mean two advances for one record). + assert_eq!( + g.shard_offset(0), + expected_len, + "shard 0 offset must advance by exactly one SELECT+SWAPDB record" + ); + assert_eq!( + g.total_offset(), + expected_len, + "master_repl_offset must track the single shard-0 advance exactly" + ); + // Shard 1 (not the local leg's shard) must be untouched. + assert_eq!(g.shard_offset(1), 0, "shard 1 offset must be untouched"); + + // Backlog content is SELECT 0 immediately followed by the SWAPDB + // payload, exactly once (no duplication). + let backlog_bytes = g.per_shard_backlogs[0] + .lock() + .as_ref() + .expect("backlog allocated") + .bytes_from(0) + .expect("offset 0 not evicted"); + let mut expected = select.to_vec(); + expected.extend_from_slice(&data); + assert_eq!( + backlog_bytes, expected, + "backlog must contain exactly one SELECT-prefixed SWAPDB record" + ); + } + + #[test] + fn test_record_local_swapdb_repl_noop_when_repl_state_none() { + let repl_state_opt: ReplStateRef<'_> = &None; + let data = Bytes::from_static(b"*3\r\n$6\r\nSWAPDB\r\n$1\r\n0\r\n$1\r\n1\r\n"); + // Must not panic when persistence/replication is fully disabled. + record_local_swapdb_repl(repl_state_opt, 0, &data); + } + + #[test] + fn test_swapdb_repl_active_false_before_any_replica_attaches() { + use crate::replication::state::ReplicationState; + + // A fresh ReplicationState with no backlog allocated and no replicas + // registered must report inactive regardless of the process-global + // fanout hint (which other tests in this binary may have already + // flipped — this assertion only depends on THIS instance's state). + let repl_state = Arc::new(parking_lot::RwLock::new(ReplicationState::new( + 2, + "b".repeat(40), + "0".repeat(40), + ))); + let repl_state_opt: Option>> = + Some(repl_state.clone()); + if !crate::replication::state::fanout_hint_active() { + assert!( + !swapdb_repl_active(&repl_state_opt), + "must be inactive: fanout hint never set AND no backlog allocated" + ); + } + } + // ── R-1: bounded spsc_send backpressure ── // A single-capacity ring at `target_idx` for (my_shard=1, target_shard=0), // since `ChannelMesh::target_index(1, 0) == 0`. diff --git a/tests/crash_matrix_per_shard_aof.rs b/tests/crash_matrix_per_shard_aof.rs index e31f1b60..908f6e7d 100644 --- a/tests/crash_matrix_per_shard_aof.rs +++ b/tests/crash_matrix_per_shard_aof.rs @@ -169,6 +169,71 @@ fn redis_get(port: u16, key: &str) -> Option { } } +/// `SET` scoped to a specific logical db via `redis-cli -n ` (issue #133 +/// SWAPDB acceptance test needs distinct db0/db1 content to prove the swap). +fn redis_set_db(port: u16, db: u8, key: &str, value: &str) { + let out = Command::new("redis-cli") + .args([ + "-p", + &port.to_string(), + "-n", + &db.to_string(), + "SET", + key, + value, + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .expect("redis-cli SET -n"); + let reply = String::from_utf8_lossy(&out.stdout); + assert_eq!( + reply.trim(), + "OK", + "redis-cli -n {} SET {} {} did not reply +OK: {:?} (stderr: {})", + db, + key, + value, + reply, + String::from_utf8_lossy(&out.stderr) + ); +} + +/// `GET` scoped to a specific logical db via `redis-cli -n `. +fn redis_get_db(port: u16, db: u8, key: &str) -> Option { + let out = Command::new("redis-cli") + .args(["-p", &port.to_string(), "-n", &db.to_string(), "GET", key]) + .output() + .expect("redis-cli GET -n"); + if !out.status.success() { + return None; + } + let s = String::from_utf8_lossy(&out.stdout).trim().to_string(); + if s.is_empty() || s == "(nil)" { + None + } else { + Some(s) + } +} + +/// Issue `SWAPDB a b` and return the raw trimmed reply text (so a failing +/// caller can assert on the EXACT +OK / -ERR wording instead of just a bool). +fn redis_swapdb(port: u16, a: u8, b: u8) -> String { + let out = Command::new("redis-cli") + .args([ + "-p", + &port.to_string(), + "SWAPDB", + &a.to_string(), + &b.to_string(), + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .expect("redis-cli SWAPDB"); + String::from_utf8_lossy(&out.stdout).trim().to_string() +} + /// Send N RPUSH commands to `key` in a single pipelined TCP write and drain /// all N integer responses. Useful for double-write detection because /// RPUSH is non-idempotent: N pushes → LLEN N; double-replay → LLEN 2*N. @@ -462,6 +527,105 @@ fn pipeline_batch_no_double_write_after_crash_recovery() { let _ = std::fs::remove_dir_all(&dir); } +/// SWAPDB-133: multi-shard SWAPDB durability acceptance test (issue #133). +/// +/// `coordinate_swapdb` (src/shard/coordinator.rs) broadcasts SWAPDB to every +/// shard. Two defects existed before this fix: +/// +/// 1. No fsync rendezvous: `+OK` was returned as soon as every shard +/// ACKed the swap, without confirming any shard's AOF record was +/// fsynced — a violation of `appendfsync=always`. +/// 2. The LOCAL shard's (the one that accepted the connection) SWAPDB +/// record was enqueued ONLY into the generic WAL v3 channel +/// (`ShardDatabases::try_wal_append_required`), never into the +/// per-shard AOF writer. For `--shards 2 --appendonly yes`, +/// `replay_per_shard` (main.rs) — not WAL v3 — is the recovery +/// authority: it wipes every shard's databases and replays ONLY the +/// per-shard AOF manifest. The local shard's half of the swap was +/// therefore silently lost on any kill-9 restart, while the remote +/// shard's half (logged via `wal_append_and_fanout` inside the SPSC +/// arm) survived — permanent cross-shard keyspace divergence. +/// +/// This test writes distinct, shard-spread content into db0 and db1 (`{a}` +/// / `{b}` hash tags land on different shards — see the +/// PIPELINE-DOUBLE-WRITE docstring above), issues `SWAPDB 0 1` under +/// `--appendfsync always`, SIGKILLs with **no** quiescing sleep (the H1 +/// contract: every `+OK` already saw an fsync), restarts, and asserts the +/// swap is visible on db0 AND db1 for keys owned by BOTH shards. +/// +/// **Red state (commit before this fix):** whichever shard accepted the +/// `SWAPDB` connection (non-deterministic — SO_REUSEPORT) shows the +/// PRE-swap content on that shard's keys after restart (the local leg's +/// record never reached the AOF); the other shard's keys show the swap +/// correctly. Which pair of assertions fails is non-deterministic run to +/// run, but at least one pair fails on every run given a real crash-window +/// (no sleep before SIGKILL). +/// **Green state (post-fix):** every key on every shard reflects the swap. +#[test] +#[ignore] // Requires built release binary + redis-cli; run explicitly. +fn crash_133_swapdb_multishard_durability_after_sigkill() { + let _serial = serialize_server_test(); + let port = unique_port().saturating_add(3); + let dir = unique_dir("swapdb-133"); + std::fs::create_dir_all(&dir).expect("create test dir"); + + // -- Round 1 -------------------------------------------------------- + let mut child = start_moon_with_fsync(port, &dir, "always"); + wait_for_port(port); + + // `{a}` and `{b}` hash-tag to different shards (CRC16 mod 2) — writing + // both in db0 and db1 guarantees the pre-swap content spans both + // shards in both logical databases. + redis_set_db(port, 0, "swap:{a}", "db0-a"); + redis_set_db(port, 0, "swap:{b}", "db0-b"); + redis_set_db(port, 1, "swap:{a}", "db1-a"); + redis_set_db(port, 1, "swap:{b}", "db1-b"); + + let swap_reply = redis_swapdb(port, 0, 1); + assert_eq!( + swap_reply, "OK", + "SWAPDB 0 1 did not reply +OK before the crash: {:?}", + swap_reply + ); + + // NO quiescing sleep — appendfsync=always means the +OK we just saw + // MUST already be durable on disk for every shard. + sigkill(&mut child); + + // -- Round 2 (recovery) --------------------------------------------- + let mut child2 = start_moon_with_fsync(port, &dir, "always"); + wait_for_port(port); + + let got_db0_a = redis_get_db(port, 0, "swap:{a}"); + let got_db0_b = redis_get_db(port, 0, "swap:{b}"); + let got_db1_a = redis_get_db(port, 1, "swap:{a}"); + let got_db1_b = redis_get_db(port, 1, "swap:{b}"); + + sigkill(&mut child2); + + let mut failures: Vec = Vec::new(); + if got_db0_a.as_deref() != Some("db1-a") { + failures.push(format!("db0 swap:{{a}} = {:?} (want \"db1-a\")", got_db0_a)); + } + if got_db0_b.as_deref() != Some("db1-b") { + failures.push(format!("db0 swap:{{b}} = {:?} (want \"db1-b\")", got_db0_b)); + } + if got_db1_a.as_deref() != Some("db0-a") { + failures.push(format!("db1 swap:{{a}} = {:?} (want \"db0-a\")", got_db1_a)); + } + if got_db1_b.as_deref() != Some("db0-b") { + failures.push(format!("db1 swap:{{b}} = {:?} (want \"db0-b\")", got_db1_b)); + } + + assert!( + failures.is_empty(), + "SWAPDB-133: swap not fully durable after kill-9 restart:\n {}", + failures.join("\n ") + ); + + let _ = std::fs::remove_dir_all(&dir); +} + // NOTE on NONOFFLOAD-WAL-V3-FALLBACK (dropped, see recovery.rs / // wal_v3::replay unit tests instead): // From a0878aa1f147c94e1b5f8b290cf58e85528515d7 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 18 Jul 2026 16:46:06 +0700 Subject: [PATCH 2/2] fix(shard): SWAPDB local leg durably persists BEFORE remote dispatch; drop repl-plane recording (#133 review round) Adversarial review of the previous commit refuted two parts of it while confirming the core durability fix sound: - Vector 2 (new hazard, confirmed): record_local_swapdb_repl committed backlog + offset + a queued live fan-out BEFORE the AOF durability gate; a send_append_group/fsync_barrier failure then aborted the local swap but could not retract the already-queued replication record -> the replica applies a swap the master never performed. - Vector 5 (pre-existing, falsifies the "replica-visible" claim): replicas cannot execute streamed SWAPDB at all - replication::apply::apply_local falls through to cmd_dispatch, which hard-errors, and the record silently no-ops with no resync. The new plumbing faithfully delivered bytes the replica cannot apply. - Vector 3 (pre-existing, widened): remote SwapDb messages were dispatched before the local durability gate, so every local abort point left N-1 shards swapped and the coordinator shard not. This commit: - Removes swapdb_repl_active / record_local_swapdb_repl and their unit tests. The local leg no longer touches the replication plane; wiring replica-side SWAPDB application properly (including the remote legs' pre-existing no-op fan-out) is issue #386, with the ordering constraints from the review recorded there. - Reorders coordinate_swapdb: the local durable-append + fsync barrier + local swap now run BEFORE any remote dispatch. Every abort point (WAL backpressure, AOF enqueue failure, fsync failure) fires while the whole cluster is still unmutated - a local abort now leaves all shards consistent instead of N-1 swapped. The confirmed-sound parts are unchanged: durable AOF append on the recovery-authority plane for the local leg (defect 2), post-ack fsync_barrier per remote shard (defect 1, H1-BARRIER ordering), truthful durability-unconfirmed error on post-swap remote barrier failure, lsn=0 PerShard sentinel parity with the remote legs (review Vector 1), no double-replay across WAL v3 + AOF (Vector 4). Refs #133 #386 author: Tin Dang --- CHANGELOG.md | 25 ++-- src/shard/coordinator.rs | 239 ++++++++------------------------------- 2 files changed, 61 insertions(+), 203 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1051f754..43f485db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,22 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Fixed -- **Multi-shard SWAPDB is now durable and replica-visible before `+OK` - (#133).** Two defects in `coordinate_swapdb`: (1) no fsync rendezvous — - under `--appendfsync always` the client saw `+OK` before any shard had +- **Multi-shard SWAPDB is now durable before `+OK` (#133).** Two defects + in `coordinate_swapdb`: (1) no fsync rendezvous — under + `--appendfsync always` the client saw `+OK` before any shard had fsynced its SWAPDB record; (2) the coordinator shard's own record was written to WAL v3 only, never the per-shard AOF — and for `--shards ≥ 2 --appendonly yes` the per-shard AOF manifest is the sole recovery authority, so a kill-9 after `+OK` permanently lost the LOCAL shard's half of the swap while remote shards' halves survived - (cross-shard keyspace divergence; reproduced empirically). The local leg - now performs a durable AOF group-commit append + fsync barrier BEFORE - swapping (abort-with-no-mutation on failure) and records the swap on the - replication plane (backlog + offset + live fan-out, `SELECT 0` framing); - the coordinator confirms each remote shard's durability with one - post-ack `fsync_barrier` (H1-BARRIER ordering). A remote barrier failure - after the swap is applied is reported truthfully as - durability-unconfirmed rather than a false `+OK`. + (cross-shard keyspace divergence; reproduced empirically). The local + leg now performs a durable AOF group-commit append + fsync barrier + BEFORE swapping AND before any remote dispatch (every abort point fires + while the whole cluster is still unmutated — the pre-fix order left + N−1 shards swapped on a local abort); the coordinator confirms each + remote shard's durability with one post-ack `fsync_barrier` + (H1-BARRIER ordering). A remote barrier failure after that shard's + swap is reported truthfully as durability-unconfirmed rather than a + false `+OK`. Replica-side SWAPDB application (streamed SWAPDB records + currently no-op on replicas — a pre-existing gap on the remote legs + too) is deliberately out of scope and tracked separately. ### Performance - **Shard offload paths are precomputed at shard init — the recurring diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index 83615883..1cff1b32 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -2759,73 +2759,31 @@ pub(crate) fn aggregate_doc_freq( (global_df, global_n) } -/// Cheap precondition for [`record_local_swapdb_repl`]: skip replication -/// bookkeeping entirely when no replica has ever attached. Ctx-free mirror -/// of `handler_monoio::ft::replication_fanout_active` — the coordinator has -/// no `ConnectionContext` in scope. -fn swapdb_repl_active(repl_state: ReplStateRef<'_>) -> bool { - if !crate::replication::state::fanout_hint_active() { - return false; - } - repl_state.as_ref().is_some_and(|rs| { - let g = rs.read(); - !g.replicas.is_empty() - || g.per_shard_backlogs - .first() - .is_some_and(|slot| slot.lock().is_some()) - }) -} - -/// Record the coordinator LOCAL leg's SWAPDB record on the replication -/// plane: backlog append + per-shard/master offset advance, synchronously -/// (before any `.await`), then defer the live-replica send onto the shard's -/// self-queue (`ShardMessage::ReplicaLiveFanout`) — the same -/// backlog-then-defer contract `wal_append_and_fanout` uses for remote legs -/// and `record_local_write_db` uses for ordinary local single-key writes. -/// -/// `coordinate_swapdb` only runs when `num_shards > 1` (single-shard SWAPDB -/// never reaches this module — see `handler_single.rs`), so every record is -/// framed with its own `SELECT 0` prefix (R2, task #20): a multi-shard -/// master merges N shard streams onto one replica wire, so no shared -/// "current db" context can exist. `db=0` mirrors the remote legs' -/// `wal_append_and_fanout(serialized, 0, ...)` call (task #35: SWAPDB -/// touches both `a` and `b` — no single db context applies). -/// -/// Caller MUST be running on `my_shard`'s own OS thread (pushes to -/// `shard::self_msg`, a `thread_local!` queue) — guaranteed here since -/// `coordinate_swapdb` always executes on the connection's own shard. -fn record_local_swapdb_repl(repl_state: ReplStateRef<'_>, my_shard: usize, data: &Bytes) { - let Some(rs) = repl_state.as_ref() else { - return; - }; - let select = crate::persistence::aof::serialize_select_record(0); - let mut combined = Vec::with_capacity(select.len() + data.len()); - combined.extend_from_slice(&select); - combined.extend_from_slice(data); - let combined = Bytes::from(combined); - - let g = rs.read(); - if let Some(slot) = g.per_shard_backlogs.get(my_shard) { - if let Some(backlog) = slot.lock().as_mut() { - backlog.append(&combined); - } - } - let end_offset = g.increment_shard_offset(my_shard, combined.len() as u64); - drop(g); - crate::shard::self_msg::push(ShardMessage::ReplicaLiveFanout { - bytes: combined, - end_offset, - }); -} - /// Broadcast SWAPDB to all shards and await acknowledgement AND durability /// from each (issue #133). /// /// # Flow /// -/// - Local shard: durable AOF append (v3-5 group-commit + barrier) BEFORE -/// the swap — mirrors the single-shard `handler_single.rs` SWAPDB -/// contract so the coordinator shard's own record survives a kill-9. +/// - Local shard FIRST: durable AOF append (v3-5 group-commit + barrier) +/// BEFORE the swap AND before any remote dispatch — mirrors the +/// single-shard `handler_single.rs` SWAPDB contract so the coordinator +/// shard's own record survives a kill-9, and (adversarial-review fix) +/// guarantees that every local abort point (WAL backpressure, AOF +/// enqueue failure, fsync failure) fires while the CLUSTER is still +/// untouched — no remote shard has been told to swap yet, so an aborted +/// SWAPDB leaves all shards consistent. +/// +/// NOTE (replication): SWAPDB is deliberately NOT recorded on the live +/// replication plane by this local leg. Replicas cannot currently execute +/// a streamed SWAPDB at all — `replication::apply::apply_local` routes it +/// into `cmd_dispatch`, which hard-errors ("SWAPDB must be issued at the +/// connection handler level") and the record silently no-ops. Fanning the +/// local record out would ship bytes the replica cannot apply, and doing +/// the backlog/offset bookkeeping BEFORE the durability gate can diverge +/// master/replica on an abort. Wiring replica-side SWAPDB application +/// (both this local leg and the remote legs' pre-existing +/// `wal_append_and_fanout` emission, which has the same no-op fate) is +/// tracked as a separate issue. /// - Remote shards: send `ShardMessage::SwapDb` (its SPSC arm durably logs /// via `wal_append_and_fanout` — WAL v3 + repl backlog/offset/fan-out + /// AOF — BEFORE swapping and acking), then this function issues ONE @@ -2903,27 +2861,11 @@ pub async fn coordinate_swapdb( num_shards > 1, "coordinate_swapdb is the multi-shard path (single-shard SWAPDB lives in handler_single.rs)" ); - // ChannelMesh has no self-send slot (target_index panics when my_id == target_id). - // Skip self in the SPSC loop; handle the local shard inline below. - let remote_count = num_shards.saturating_sub(1); - let mut targets: Vec = Vec::with_capacity(remote_count); - let mut receivers: Vec> = Vec::with_capacity(remote_count); - - for target in 0..num_shards { - if target == my_shard { - continue; // handled inline below - } - let (reply_tx, reply_rx) = channel::oneshot(); - let msg = ShardMessage::SwapDb { a, b, reply_tx }; - let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; - targets.push(target); - receivers.push(reply_rx); - } - - // Local shard: durable append BEFORE the swap. SWAPDB has no - // command-level rollback; anything that can fail must fail BEFORE - // mutating this shard's state, otherwise the cluster state diverges - // from the on-disk log. + // 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 — + // dispatching remotes first (the pre-fix order) meant a local abort + // left N-1 shards swapped and this one not. { let mut a_buf = itoa::Buffer::new(); let mut b_buf = itoa::Buffer::new(); @@ -2952,22 +2894,11 @@ pub async fn coordinate_swapdb( // MSET/BITOP/COPY/DEL: enqueue fire-and-forget, then ONE barrier // under `Always` instead of a per-write awaited fsync. if let Some(pool) = aof_pool { - let repl_active = swapdb_repl_active(repl_state); - let lsn = if repl_active { - // Records backlog + offset synchronously (before any - // .await) and defers only the live-replica send — matches - // the local single-key write contract - // (`record_local_write_db`). lsn=0: the AOF leg must not - // double-advance the offset this already advanced. - record_local_swapdb_repl(repl_state, my_shard, &serialized); - 0 - } else { - crate::persistence::aof::AofWriterPool::issue_append_lsn( - repl_state, - my_shard, - serialized.len(), - ) - }; + let lsn = crate::persistence::aof::AofWriterPool::issue_append_lsn( + repl_state, + my_shard, + serialized.len(), + ); match pool.send_append_group(my_shard, lsn, 0, serialized).await { Ok(needs_barrier) => { if needs_barrier && pool.fsync_barrier(my_shard).await.is_err() { @@ -2993,6 +2924,23 @@ pub async fn coordinate_swapdb( }); } + // ChannelMesh has no self-send slot (target_index panics when my_id == target_id). + // Skip self in the SPSC loop; handle the local shard inline below. + let remote_count = num_shards.saturating_sub(1); + let mut targets: Vec = Vec::with_capacity(remote_count); + let mut receivers: Vec> = Vec::with_capacity(remote_count); + + for target in 0..num_shards { + if target == my_shard { + continue; // handled inline below + } + let (reply_tx, reply_rx) = channel::oneshot(); + let msg = ShardMessage::SwapDb { a, b, reply_tx }; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + targets.push(target); + receivers.push(reply_rx); + } + // Await every remote shard's ack, then confirm ITS durability with one // fsync_barrier (H1-BARRIER pattern) — the SwapDb SPSC arm cannot // `.await` a barrier itself, so the coordinator closes that gap here, @@ -3075,99 +3023,6 @@ mod tests { ); } - // ── issue #133: SWAPDB local-leg replication plane ────────────────── - // - // `record_local_swapdb_repl` is the coordinator's ctx-free equivalent of - // `record_local_write_db` for the SWAPDB local leg. These tests exercise - // it directly (no shard mesh / event loop needed) to prove EXACTLY-ONCE - // backlog append + offset advance per call — a double-count here would - // desync `master_repl_offset` from the sum of bytes actually delivered, - // and SWAPDB is explicitly non-idempotent (double-logging `a b` nets a - // no-op swap on replay). - #[test] - fn test_record_local_swapdb_repl_appends_exactly_once_with_select_prefix() { - use crate::replication::state::ReplicationState; - - let repl_state = Arc::new(parking_lot::RwLock::new(ReplicationState::new( - 2, - "a".repeat(40), - "0".repeat(40), - ))); - // Simulate a replica already attached on shard 0 (lazy backlog alloc). - repl_state.read().ensure_backlogs_allocated(); - let repl_state_opt: Option>> = - Some(repl_state.clone()); - - let data = Bytes::from_static(b"*3\r\n$6\r\nSWAPDB\r\n$1\r\n0\r\n$1\r\n1\r\n"); - record_local_swapdb_repl(&repl_state_opt, 0, &data); - - let select = crate::persistence::aof::serialize_select_record(0); - let expected_len = (select.len() + data.len()) as u64; - - let g = repl_state.read(); - // Offset advanced by EXACTLY one SELECT+payload's worth of bytes — - // not zero (would mean the call was a no-op) and not double - // (would mean two advances for one record). - assert_eq!( - g.shard_offset(0), - expected_len, - "shard 0 offset must advance by exactly one SELECT+SWAPDB record" - ); - assert_eq!( - g.total_offset(), - expected_len, - "master_repl_offset must track the single shard-0 advance exactly" - ); - // Shard 1 (not the local leg's shard) must be untouched. - assert_eq!(g.shard_offset(1), 0, "shard 1 offset must be untouched"); - - // Backlog content is SELECT 0 immediately followed by the SWAPDB - // payload, exactly once (no duplication). - let backlog_bytes = g.per_shard_backlogs[0] - .lock() - .as_ref() - .expect("backlog allocated") - .bytes_from(0) - .expect("offset 0 not evicted"); - let mut expected = select.to_vec(); - expected.extend_from_slice(&data); - assert_eq!( - backlog_bytes, expected, - "backlog must contain exactly one SELECT-prefixed SWAPDB record" - ); - } - - #[test] - fn test_record_local_swapdb_repl_noop_when_repl_state_none() { - let repl_state_opt: ReplStateRef<'_> = &None; - let data = Bytes::from_static(b"*3\r\n$6\r\nSWAPDB\r\n$1\r\n0\r\n$1\r\n1\r\n"); - // Must not panic when persistence/replication is fully disabled. - record_local_swapdb_repl(repl_state_opt, 0, &data); - } - - #[test] - fn test_swapdb_repl_active_false_before_any_replica_attaches() { - use crate::replication::state::ReplicationState; - - // A fresh ReplicationState with no backlog allocated and no replicas - // registered must report inactive regardless of the process-global - // fanout hint (which other tests in this binary may have already - // flipped — this assertion only depends on THIS instance's state). - let repl_state = Arc::new(parking_lot::RwLock::new(ReplicationState::new( - 2, - "b".repeat(40), - "0".repeat(40), - ))); - let repl_state_opt: Option>> = - Some(repl_state.clone()); - if !crate::replication::state::fanout_hint_active() { - assert!( - !swapdb_repl_active(&repl_state_opt), - "must be inactive: fanout hint never set AND no backlog allocated" - ); - } - } - // ── R-1: bounded spsc_send backpressure ── // A single-capacity ring at `target_idx` for (my_shard=1, target_shard=0), // since `ChannelMesh::target_index(1, 0) == 0`.