From 666d32c9df669c52881cc78f0a0d0317c4cef522 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 4 Jul 2026 22:51:06 +0700 Subject: [PATCH 1/4] perf(persistence): route coordinator local-leg persist through group commit under appendfsync=always MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cross-shard coordinator's LOCAL-leg persist (co-located MSET/MSETNX and scattered-MSET local slices, shipped in v3-4 Finding 1) called try_send_append_durable, which under appendfsync=always awaits one fsync ack PER COMMAND bounded by --aof-fsync-timeout-ms (default 2000ms). A pipeline of coordinated writes stacked these serially — the measured 2000-3000ms always far-tail (tmp/V3-4-GCLOUD-BENCH.md), carried as the [HIGH] follow-up into v3-5. Fix: local legs now use the same fire-and-forget-then-barrier contract the remote SPSC legs have used since the H1-BARRIER fix: - New AofWriterPool::send_append_group enqueues under bounded backpressure and returns immediately; Ok(true) means Always policy — the caller owes a barrier. EverySec/No are unchanged (writer loop owns the fsync cadence). - persist_local_leg switches to it and reports needs-barrier up through coordinate_mset/coordinate_msetnx/coordinate_multi_key. - Both connection handlers (monoio + sharded) collect the response indexes of barrier-pending local-leg writes and issue ONE fsync_barrier(local shard) per pipeline batch, BEFORE response serialization — +OK still implies confirmed durability, but a batch of N coordinated writes costs 1 awaited fsync instead of N. - On barrier failure every affected response is overwritten with AOF_FSYNC_ERR — never a false +OK (design-for-failure preserved). Tests: 3 new red-proven pool unit tests (send_append_group must not await the per-write ack; everysec needs no barrier; dead writer errors); v3-4 coordinator_local_leg_durability crash-recovery suite green on the new path (local legs still persist and replay); full lib suite 3656 pass; clippy clean on default + tokio,jemalloc. Absolute tail magnitude needs a GCE run (OrbStack fsync is near-free); tracked in v3-5. author: Tin Dang --- CHANGELOG.md | 15 +++ src/persistence/aof/pool.rs | 110 +++++++++++++++++++++ src/server/conn/handler_monoio/dispatch.rs | 10 ++ src/server/conn/handler_monoio/mod.rs | 32 +++++- src/server/conn/handler_sharded/mod.rs | 28 +++++- src/shard/coordinator.rs | 76 ++++++++------ 6 files changed, 239 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5831ffaf4..de08e4e5f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Performance — coordinator local legs ride group commit under appendfsync=always (PR #TBD) + +- The cross-shard coordinator's LOCAL-leg persist (co-located MSET/MSETNX and + scattered-MSET local slices) awaited one fsync ack **per command**, each + bounded by `--aof-fsync-timeout-ms` (default 2000ms) — a pipeline of + coordinated writes stacked these serially into the measured 2000–3000ms + `always` far-tail. Local legs now enqueue fire-and-forget (bounded + backpressure, same contract as the remote SPSC legs) and the connection + handler confirms them with **one** `fsync_barrier` on the local shard per + pipeline batch, before responses are serialized — so `+OK` still implies + confirmed durability, but a batch of N coordinated writes costs 1 awaited + fsync instead of N. `everysec`/`no` behavior is unchanged. On barrier + failure every affected response is replaced with `MOONERR AOF fsync` + (never a false `+OK`). + ## [0.5.1] — 2026-07-04 ### Fixed diff --git a/src/persistence/aof/pool.rs b/src/persistence/aof/pool.rs index b75196be1..18680b18a 100644 --- a/src/persistence/aof/pool.rs +++ b/src/persistence/aof/pool.rs @@ -282,6 +282,32 @@ impl AofWriterPool { } } + /// Group-commit append for coordinator LOCAL legs (v3-5 local-leg fix). + /// + /// Enqueues the append under bounded backpressure and returns WITHOUT + /// awaiting the per-write fsync ack — under `Always` the old + /// `try_send_append_durable` path serialized one awaited fsync per + /// coordinated command, stacking up to `fsync_timeout` each in a pipeline + /// (the measured 2000–3000ms always-tail). This is the same + /// fire-and-forget-then-barrier contract the cross-shard REMOTE legs + /// already use (SPSC arm append + handler `fsync_barrier`). + /// + /// Returns `Ok(true)` when the caller MUST confirm durability with ONE + /// [`Self::fsync_barrier`]`(shard_id)` before acking the client (`Always`), + /// `Ok(false)` when the writer loop owns the fsync cadence + /// (`EverySec`/`No`), and `Err(_)` when the append never reached the + /// writer — the caller must surface an error frame, never `+OK`. + #[inline] + pub async fn send_append_group( + &self, + shard_id: usize, + lsn: u64, + bytes: Bytes, + ) -> Result { + self.send_append_backpressure(shard_id, lsn, bytes).await?; + Ok(matches!(self.fsync_policy, FsyncPolicy::Always)) + } + /// Durability barrier for cross-shard pipelined writes under /// `appendfsync=always` (H1 fix, C4-FOLD-FIX follow-up). /// @@ -1839,6 +1865,90 @@ mod pool_tests { ); } + // ----------------------------------------------------------------------- + // GC-LOCAL-LEG (v3-5): send_append_group must enqueue WITHOUT awaiting the + // per-write fsync ack, so coordinator local legs ride the batch-end + // fsync_barrier (ONE fsync per pipeline batch) instead of stacking one + // awaited fsync per command (the measured 2000-3000ms always-tail). + // ----------------------------------------------------------------------- + #[test] + fn send_append_group_always_enqueues_without_awaiting_fsync() { + let (tx0, rx0) = channel::mpsc_bounded::(4); + let (tx1, _rx1) = channel::mpsc_bounded::(4); + // Long fsync timeout: if the implementation wrongly awaits the fsync + // ack (nobody acks here), the elapsed assertion below fails loudly. + let pool = AofWriterPool::per_shard_with_policy( + vec![tx0, tx1], + FsyncPolicy::Always, + Duration::from_millis(5000), + ); + + let start = std::time::Instant::now(); + let result = futures::executor::block_on(pool.send_append_group( + 0, + 77, + Bytes::from_static(b"MSET k v"), + )); + + assert_eq!( + result, + Ok(true), + "Always policy must enqueue and report that a barrier is required" + ); + assert!( + start.elapsed() < Duration::from_millis(500), + "send_append_group must NOT await the per-write fsync ack" + ); + // The append must be in the writer channel as a plain Append (no ack + // slot) — durability confirmation belongs to the batch-end barrier. + match rx0.try_recv() { + Ok(AofMessage::Append { lsn, .. }) => assert_eq!(lsn, 77), + other => panic!("expected plain Append in channel, got {:?}", other.is_ok()), + } + } + + #[test] + fn send_append_group_everysec_needs_no_barrier() { + let (tx0, _rx0) = channel::mpsc_bounded::(4); + let (tx1, _rx1) = channel::mpsc_bounded::(4); + let pool = AofWriterPool::per_shard_with_policy( + vec![tx0, tx1], + FsyncPolicy::EverySec, + Duration::ZERO, + ); + let result = futures::executor::block_on(pool.send_append_group( + 0, + 78, + Bytes::from_static(b"MSET k v"), + )); + assert_eq!( + result, + Ok(false), + "EverySec needs no barrier — writer loop owns the 1s fsync cadence" + ); + } + + #[test] + fn send_append_group_dead_writer_returns_err() { + let (tx0, rx0) = channel::mpsc_bounded::(4); + let (tx1, _rx1) = channel::mpsc_bounded::(4); + drop(rx0); // writer gone + let pool = AofWriterPool::per_shard_with_policy( + vec![tx0, tx1], + FsyncPolicy::Always, + Duration::from_millis(50), + ); + let result = futures::executor::block_on(pool.send_append_group( + 0, + 79, + Bytes::from_static(b"MSET k v"), + )); + assert!( + result.is_err(), + "a dead writer must surface as Err so the caller never acks +OK" + ); + } + // ----------------------------------------------------------------------- // H1-BARRIER-1: fsync_barrier under EverySec must be a zero-cost noop — // no message enqueued, channel stays empty, returns Ok(()) immediately. diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index f1c7368cd..95a524e48 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -1199,6 +1199,9 @@ pub(super) async fn try_handle_cross_shard_commands( conn: &ConnectionState, ctx: &ConnectionContext, responses: &mut Vec, + // v3-5 group commit: response indexes of local-leg writes pending the + // batch-end fsync_barrier(ctx.shard_id) (appendfsync=always only). + local_leg_write_idxs: &mut Vec, ) -> bool { if ctx.num_shards <= 1 { return false; @@ -1278,6 +1281,7 @@ pub(super) async fn try_handle_cross_shard_commands( // --- Multi-key commands: MGET, MSET, DEL, UNLINK, EXISTS --- if is_multi_key_command(cmd, cmd_args) { + let mut local_barrier_pending = false; let response = crate::shard::coordinator::coordinate_multi_key( cmd, cmd_args, @@ -1290,9 +1294,15 @@ pub(super) async fn try_handle_cross_shard_commands( &ctx.cached_clock, ctx.aof_pool.as_ref(), &ctx.repl_state, + &mut local_barrier_pending, &(), // monoio: coordinator uses oneshot, not response_pool ) .await; + // A response that is already an error must not be overwritten by a + // barrier failure; only successful writes join the barrier set. + if local_barrier_pending && !matches!(response, Frame::Error(_)) { + local_leg_write_idxs.push(responses.len()); + } responses.push(response); return true; } diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index dc0fa2922..68c2e2e69 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -192,6 +192,10 @@ pub(crate) async fn handle_connection_sharded_monoio< > = HashMap::with_capacity(ctx.num_shards); let mut reply_futures: Vec<(Vec<(usize, Option, Bytes)>, usize)> = Vec::with_capacity(ctx.num_shards); + // v3-5 group commit: response indexes of coordinator LOCAL-leg writes whose + // AOF append was enqueued but not yet fsync-confirmed (appendfsync=always). + // Drained by ONE fsync_barrier(ctx.shard_id) at end of batch. + let mut local_leg_write_idxs: Vec = Vec::new(); // Pre-allocated response slots for zero-allocation cross-shard dispatch // (L3b, tokio parity — handler_sharded/mod.rs). One slot per target shard; @@ -589,6 +593,7 @@ pub(crate) async fn handle_connection_sharded_monoio< let mut should_quit = false; responses.clear(); remote_groups.clear(); + local_leg_write_idxs.clear(); let mut publish_batches: std::collections::HashMap> = std::collections::HashMap::new(); @@ -940,8 +945,15 @@ pub(crate) async fn handle_connection_sharded_monoio< } // --- Cross-shard aggregation commands: KEYS, SCAN, DBSIZE + multi-key --- - if dispatch::try_handle_cross_shard_commands(cmd, cmd_args, &conn, ctx, &mut responses) - .await + if dispatch::try_handle_cross_shard_commands( + cmd, + cmd_args, + &conn, + ctx, + &mut responses, + &mut local_leg_write_idxs, + ) + .await { continue; } @@ -1801,6 +1813,22 @@ pub(crate) async fn handle_connection_sharded_monoio< } } + // v3-5 GROUP-COMMIT BARRIER: coordinator LOCAL legs (MSET/MSETNX) were + // enqueued fire-and-forget into MY shard's AOF writer during dispatch; + // one barrier here confirms every one of them with a single fsync + // instead of the retired per-command awaited fsync (2000ms tail stack). + // Runs BEFORE response serialization — the client never sees +OK for a + // write whose durability was not confirmed. + if !local_leg_write_idxs.is_empty() { + if let Some(ref pool) = ctx.aof_pool { + if pool.fsync_barrier(ctx.shard_id).await.is_err() { + for idx in local_leg_write_idxs.drain(..) { + responses[idx] = Frame::Error(Bytes::from_static(aof::AOF_FSYNC_ERR)); + } + } + } + } + // AUTH rate limiting: delay response to slow down brute-force attacks if auth_delay_ms > 0 { monoio::time::sleep(std::time::Duration::from_millis(auth_delay_ms)).await; diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index a05ddae06..29a209933 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -374,6 +374,9 @@ pub(crate) async fn handle_connection_sharded_inner< } let mut responses: Vec = Vec::with_capacity(batch.len()); + // v3-5 group commit: response indexes of coordinator LOCAL-leg + // writes pending the batch-end fsync_barrier(ctx.shard_id). + let mut local_leg_write_idxs: Vec = Vec::new(); let mut should_quit = false; let mut remote_groups: HashMap, Option, Bytes, usize)>> = HashMap::with_capacity(ctx.num_shards); // Accumulate cross-shard PUBLISH pairs per target shard for batch dispatch @@ -1040,7 +1043,13 @@ pub(crate) async fn handle_connection_sharded_inner< // --- Multi-key commands --- if is_multi_key_command(cmd, cmd_args) { - let response = crate::shard::coordinator::coordinate_multi_key(cmd, cmd_args, ctx.shard_id, ctx.num_shards, conn.selected_db, &ctx.shard_databases, &ctx.dispatch_tx, &ctx.spsc_notifiers, &ctx.cached_clock, ctx.aof_pool.as_ref(), &ctx.repl_state, &()).await; + let mut local_barrier_pending = false; + let response = crate::shard::coordinator::coordinate_multi_key(cmd, cmd_args, ctx.shard_id, ctx.num_shards, conn.selected_db, &ctx.shard_databases, &ctx.dispatch_tx, &ctx.spsc_notifiers, &ctx.cached_clock, ctx.aof_pool.as_ref(), &ctx.repl_state, &mut local_barrier_pending, &()).await; + // Only successful writes join the barrier set — an error + // response must not be overwritten by a barrier failure. + if local_barrier_pending && !matches!(response, Frame::Error(_)) { + local_leg_write_idxs.push(responses.len()); + } responses.push(response); continue; } @@ -1707,6 +1716,23 @@ pub(crate) async fn handle_connection_sharded_inner< } } + // v3-5 GROUP-COMMIT BARRIER: coordinator LOCAL legs (MSET/MSETNX) + // were enqueued fire-and-forget into MY shard's AOF writer during + // dispatch; ONE barrier confirms all of them with a single fsync + // instead of the retired per-command awaited fsync. Runs BEFORE + // response serialization — no +OK without confirmed durability. + if !local_leg_write_idxs.is_empty() { + if let Some(ref pool) = ctx.aof_pool { + if pool.fsync_barrier(ctx.shard_id).await.is_err() { + for idx in local_leg_write_idxs.drain(..) { + responses[idx] = Frame::Error( + Bytes::from_static(aof::AOF_FSYNC_ERR), + ); + } + } + } + } + // Phase 3: Flush accumulated PUBLISH batches as PubSubPublishBatch messages if !publish_batches.is_empty() { let mut batch_slots: Vec<(std::sync::Arc, Vec)> = Vec::new(); diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index ea16bc544..a0ab9a9bc 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -46,6 +46,11 @@ pub async fn coordinate_multi_key( // persistence (tests / no-AOF deployments). aof_pool: Option<&Arc>, repl_state: ReplStateRef<'_>, + // v3-5 group commit: set to true when a local-leg append was enqueued + // under appendfsync=always. The connection handler MUST then issue ONE + // `fsync_barrier(my_shard)` for the batch before acking the client, and + // overwrite this command's response with AOF_FSYNC_ERR on barrier failure. + local_barrier_pending: &mut bool, _response_pool: &(), // placeholder — coordinator uses oneshot internally ) -> Frame { if cmd.eq_ignore_ascii_case(b"MGET") { @@ -73,6 +78,7 @@ pub async fn coordinate_multi_key( cached_clock, aof_pool, repl_state, + local_barrier_pending, _response_pool, ) .await @@ -88,6 +94,7 @@ pub async fn coordinate_multi_key( cached_clock, aof_pool, repl_state, + local_barrier_pending, _response_pool, ) .await @@ -227,11 +234,17 @@ type ReplStateRef<'a> = /// Persist a coordinator LOCAL-leg write to the owning shard's AOF, matching the /// local single-key write contract (the `is_write` block in /// `handler_monoio`/`handler_sharded`): issue an LSN off `repl_state`, then -/// durable-append. Under `appendfsync=always` this awaits the writer's fsync ack; -/// under everysec/no it is fire-and-forget. WAL append is external to -/// `cmd_dispatch`, so the coordinator's in-process local legs (`run_local`, -/// `coordinate_mset` fast path / local slice) MUST call this or their writes are -/// lost on restart while the remote legs (via `wal_append_and_fanout`) survive. +/// group-commit-append. WAL append is external to `cmd_dispatch`, so the +/// coordinator's in-process local legs (`run_local`, `coordinate_mset` fast +/// path / local slice) MUST call this or their writes are lost on restart +/// while the remote legs (via `wal_append_and_fanout`) survive. +/// +/// v3-5 group-commit routing: the append is ENQUEUED (bounded backpressure), +/// never per-write fsync-awaited. `Ok(true)` means `appendfsync=always` — the +/// connection handler MUST issue ONE `fsync_barrier(my_shard)` for the whole +/// pipeline batch before acking the client (the same contract the remote legs +/// use). The old per-write awaited fsync stacked one `fsync_timeout` per +/// coordinated command in a pipeline — the measured 2000–3000ms always-tail. /// /// `serialized` MUST cover only keys OWNED by `my_shard`: /// - co-located command (MSETNX; MSET fast path) → the whole command, @@ -239,25 +252,24 @@ type ReplStateRef<'a> = /// just the local keys (never the full scattered command — `my_shard` does /// not own the remote keys and replay would misapply them on this shard). /// -/// Returns `Err(())` on AOF failure so the caller surfaces `AOF_FSYNC_ERR` -/// instead of a false `+OK` (design-for-failure; matches the handler). +/// Returns `Err(())` when the append never reached the writer so the caller +/// surfaces `AOF_FSYNC_ERR` instead of a false `+OK` (design-for-failure). async fn persist_local_leg( aof_pool: Option<&Arc>, repl_state: ReplStateRef<'_>, my_shard: usize, serialized: Bytes, -) -> Result<(), ()> { - let Some(pool) = aof_pool else { return Ok(()) }; +) -> Result { + let Some(pool) = aof_pool else { + return Ok(false); + }; let lsn = crate::persistence::aof::AofWriterPool::issue_append_lsn( repl_state, my_shard, serialized.len(), ); - match pool - .try_send_append_durable(my_shard, lsn, serialized) - .await - { - Ok(()) => Ok(()), + match pool.send_append_group(my_shard, lsn, serialized).await { + Ok(needs_barrier) => Ok(needs_barrier), Err(_) => Err(()), } } @@ -809,6 +821,7 @@ async fn coordinate_mset( cached_clock: &CachedClock, aof_pool: Option<&Arc>, repl_state: ReplStateRef<'_>, + local_barrier_pending: &mut bool, _response_pool: &(), // placeholder — coordinator uses oneshot internally ) -> Frame { if args.is_empty() || !args.len().is_multiple_of(2) { @@ -850,11 +863,13 @@ async fn coordinate_mset( // owned by my_shard — matching the local single-key write contract. if let Some(pairs) = groups.get(&my_shard) { let serialized = serialize_local_mset(pairs); - if persist_local_leg(aof_pool, repl_state, my_shard, serialized) - .await - .is_err() - { - return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + match persist_local_leg(aof_pool, repl_state, my_shard, serialized).await { + Ok(needs_barrier) => *local_barrier_pending |= needs_barrier, + Err(()) => { + return Frame::Error(Bytes::from_static( + crate::persistence::aof::AOF_FSYNC_ERR, + )); + } } } return resp; @@ -904,11 +919,11 @@ async fn coordinate_mset( // write keys this shard doesn't own). if let Some(pairs) = groups.get(&my_shard) { let serialized = serialize_local_mset(pairs); - if persist_local_leg(aof_pool, repl_state, my_shard, serialized) - .await - .is_err() - { - return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + match persist_local_leg(aof_pool, repl_state, my_shard, serialized).await { + Ok(needs_barrier) => *local_barrier_pending |= needs_barrier, + Err(()) => { + return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + } } } @@ -935,6 +950,7 @@ async fn coordinate_msetnx( cached_clock: &CachedClock, aof_pool: Option<&Arc>, repl_state: ReplStateRef<'_>, + local_barrier_pending: &mut bool, _response_pool: &(), // placeholder — coordinator uses oneshot internally ) -> Frame { if args.is_empty() || !args.len().is_multiple_of(2) { @@ -983,11 +999,13 @@ async fn coordinate_msetnx( if matches!(resp, Frame::Integer(1)) { let serialized = crate::persistence::aof::serialize_command(&Frame::Array(command_parts.into())); - if persist_local_leg(aof_pool, repl_state, my_shard, serialized) - .await - .is_err() - { - return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + match persist_local_leg(aof_pool, repl_state, my_shard, serialized).await { + Ok(needs_barrier) => *local_barrier_pending |= needs_barrier, + Err(()) => { + return Frame::Error(Bytes::from_static( + crate::persistence::aof::AOF_FSYNC_ERR, + )); + } } } resp From d513e8d19f6418809a31a484b3d1370e0c07a9cc Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 4 Jul 2026 23:11:47 +0700 Subject: [PATCH 2/4] fix(persistence): persist BITOP/COPY/DEL/UNLINK coordinator local legs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cross-shard coordinator's in-process legs for BITOP (dest write), COPY (dst write + PEXPIRE TTL restore), and multi-key DEL/UNLINK (both the co-located fast path and the scattered local slice) executed in memory but never appended to the owning shard's AOF — the carried v3-4 follow-up. Failure modes on kill-9 + restart: - DEL/UNLINK: deleted keys RESURRECTED from their seed writes (the seed MSET is in the AOF, the local-leg delete never was). - BITOP/COPY: results whose dest/dst owner == the connection's own shard silently vanished. Remote legs were always durable (MultiExecute -> wal_append_and_fanout). Fix: all four now persist through the same persist_local_leg group-commit path as MSET/MSETNX (previous commit): - New run_on_owner_persist mirrors run_on_owner but appends the command to my_shard's AOF when the owner is local and execution succeeded — used by BITOP's whole-command forward + synthesized DEL/SET dest legs and COPY's whole-command forward + SET/PEXPIRE dst legs. Replay-safe: each persisted command covers only keys owned by my_shard. - coordinate_multi_del_or_exists persists DEL/UNLINK on the co-located fast path (whole command) and the scattered local slice (synthesized over only local keys), skipped when n=0 (nothing removed replays identically without a record). EXISTS/TOUCH stay read-only. - All legs ride the batch-end fsync barrier under appendfsync=always via local_barrier_pending (previous commit's plumbing). Tests: 4 new red-proven crash-recovery tests in coordinator_local_leg_durability.rs (DEL scatter resurrection, UNLINK co-located fast-path resurrection, BITOP dest legs both shapes, COPY dst legs both shapes) — deterministic via per-shard {hash-tags} regardless of which shard the connection lands on; suite 7/7 green post-fix. Full lib suite 3656 pass; clippy clean both feature sets. author: Tin Dang --- CHANGELOG.md | 14 ++ src/shard/coordinator.rs | 161 +++++++++++- tests/coordinator_local_leg_durability.rs | 283 ++++++++++++++++++++++ 3 files changed, 451 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index de08e4e5f..641b5b93a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 failure every affected response is replaced with `MOONERR AOF fsync` (never a false `+OK`). +### Fixed — BITOP/COPY/DEL/UNLINK coordinator local legs now persist (PR #TBD) + +- The cross-shard coordinator's in-process legs for BITOP (dest write), + COPY (dst write + TTL restore), and multi-key DEL/UNLINK (co-located + fast path AND the scattered local slice) executed in memory but never + reached the owning shard's AOF — deleted keys **resurrected** from + their seed writes on restart, and BITOP/COPY results on the + connection's own shard silently vanished (carried v3-4 follow-up; + remote legs were always durable via MultiExecute). All four now + persist through the same `persist_local_leg` group-commit path as + MSET/MSETNX: synthesized over only locally-owned keys, skipped when + nothing was written (DEL of missing keys), and confirmed by the + batch-end fsync barrier under `appendfsync=always`. + ## [0.5.1] — 2026-07-04 ### Fixed diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index a0ab9a9bc..d1313bdf3 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -108,6 +108,9 @@ pub async fn coordinate_multi_key( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, _response_pool, ) .await @@ -121,6 +124,9 @@ pub async fn coordinate_multi_key( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, _response_pool, ) .await @@ -136,6 +142,9 @@ pub async fn coordinate_multi_key( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, _response_pool, ) .await @@ -226,6 +235,68 @@ async fn run_on_owner( } } +/// Like [`run_on_owner`], but for WRITE commands: when the owner is the +/// connection's own shard the command executes in-process, so nothing else +/// persists it — append it to my_shard's AOF via [`persist_local_leg`] +/// (v3-5: BITOP/COPY/DEL/UNLINK carried gap). Remote owners persist on their +/// own shard via MultiExecute → wal_append_and_fanout, exactly as before. +/// +/// `command_parts` MUST be replay-safe against my_shard alone (a full command +/// whose keys are all owned by my_shard, or a synthesized write like +/// `SET dest ` for a scatter BITOP/COPY). +/// +/// Persists only when the local execution did not error; sets +/// `*local_barrier_pending` when the append rides group commit under +/// `appendfsync=always` (the handler owes ONE `fsync_barrier(my_shard)` per +/// batch). Returns `AOF_FSYNC_ERR` when the append never reached the writer. +#[allow(clippy::too_many_arguments)] +async fn run_on_owner_persist( + routing_key: &Bytes, + command_parts: &[Frame], + my_shard: usize, + num_shards: usize, + db_index: usize, + shard_databases: &Arc, + dispatch_tx: &Rc>>>, + spsc_notifiers: &[Arc], + cached_clock: &CachedClock, + aof_pool: Option<&Arc>, + repl_state: ReplStateRef<'_>, + local_barrier_pending: &mut bool, +) -> Frame { + let owner = key_to_shard(routing_key, num_shards); + if owner != my_shard { + let command = Frame::Array(command_parts.to_vec().into()); + return run_remote( + owner, + routing_key, + command, + my_shard, + db_index, + dispatch_tx, + spsc_notifiers, + ) + .await; + } + let (cmd, args) = match command_parts.split_first() { + Some((Frame::BulkString(c), rest)) => (c.clone(), rest), + _ => return Frame::Error(Bytes::from_static(b"ERR invalid command format")), + }; + let resp = run_local(shard_databases, db_index, cached_clock, &cmd, args); + if !matches!(resp, Frame::Error(_)) { + let serialized = crate::persistence::aof::serialize_command(&Frame::Array( + command_parts.to_vec().into(), + )); + match persist_local_leg(aof_pool, repl_state, my_shard, serialized).await { + Ok(needs_barrier) => *local_barrier_pending |= needs_barrier, + Err(()) => { + return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + } + } + } + resp +} + /// Type of the replication-state handle threaded into the coordinator's local /// persistence path (same shape `AofWriterPool::issue_append_lsn` expects). type ReplStateRef<'a> = @@ -312,6 +383,9 @@ async fn coordinate_bitop( dispatch_tx: &Rc>>>, spsc_notifiers: &[Arc], cached_clock: &CachedClock, + aof_pool: Option<&Arc>, + repl_state: ReplStateRef<'_>, + local_barrier_pending: &mut bool, _response_pool: &(), ) -> Frame { // Single-shard server: straight to local dispatch — zero coordinator @@ -357,7 +431,8 @@ async fn coordinate_bitop( let mut parts: Vec = Vec::with_capacity(args.len() + 1); parts.push(bulk_static(b"BITOP")); parts.extend_from_slice(args); - return run_on_owner( + // Write command: the local-owner case must persist (v3-5 carried gap). + return run_on_owner_persist( &dest, &parts, my_shard, @@ -367,6 +442,9 @@ async fn coordinate_bitop( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, ) .await; } @@ -441,7 +519,7 @@ async fn coordinate_bitop( Err(e) => e, Ok(None) => { // All sources empty/missing — dest is deleted, reply 0. - let reply = run_on_owner( + let reply = run_on_owner_persist( &dest, &[bulk_static(b"DEL"), bulk(&dest)], my_shard, @@ -451,6 +529,9 @@ async fn coordinate_bitop( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, ) .await; if let Frame::Error(e) = reply { @@ -460,7 +541,7 @@ async fn coordinate_bitop( } Ok(Some(result)) => { let len = result.len() as i64; - let reply = run_on_owner( + let reply = run_on_owner_persist( &dest, &[ bulk_static(b"SET"), @@ -474,6 +555,9 @@ async fn coordinate_bitop( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, ) .await; if let Frame::Error(e) = reply { @@ -504,6 +588,9 @@ async fn coordinate_copy( dispatch_tx: &Rc>>>, spsc_notifiers: &[Arc], cached_clock: &CachedClock, + aof_pool: Option<&Arc>, + repl_state: ReplStateRef<'_>, + local_barrier_pending: &mut bool, _response_pool: &(), ) -> Frame { // Single-shard server: straight to local dispatch — zero coordinator @@ -537,7 +624,8 @@ async fn coordinate_copy( let mut parts: Vec = Vec::with_capacity(args.len() + 1); parts.push(bulk_static(b"COPY")); parts.extend_from_slice(args); - return run_on_owner( + // Write command: the local-owner case must persist (v3-5 carried gap). + return run_on_owner_persist( &src, &parts, my_shard, @@ -547,6 +635,9 @@ async fn coordinate_copy( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, ) .await; } @@ -603,7 +694,7 @@ async fn coordinate_copy( bulk_static(b"NX"), ] }; - let set_reply = run_on_owner( + let set_reply = run_on_owner_persist( &dst, &set_parts, my_shard, @@ -613,6 +704,9 @@ async fn coordinate_copy( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, ) .await; match set_reply { @@ -623,7 +717,7 @@ async fn coordinate_copy( } if let Some(t) = ttl_ms { let mut ttl_buf = itoa::Buffer::new(); - let reply = run_on_owner( + let reply = run_on_owner_persist( &dst, &[ bulk_static(b"PEXPIRE"), @@ -637,6 +731,9 @@ async fn coordinate_copy( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + local_barrier_pending, ) .await; if let Frame::Error(e) = reply { @@ -1037,9 +1134,15 @@ async fn coordinate_multi_del_or_exists( dispatch_tx: &Rc>>>, spsc_notifiers: &[Arc], cached_clock: &CachedClock, + aof_pool: Option<&Arc>, + repl_state: ReplStateRef<'_>, + local_barrier_pending: &mut bool, _response_pool: &(), // placeholder — coordinator uses oneshot internally ) -> Frame { let cmd_upper = cmd.to_ascii_uppercase(); + // DEL/UNLINK mutate and must persist their in-process legs; EXISTS/TOUCH + // read (TOUCH updates access time only — never AOF-logged, like Redis). + let is_delete = cmd_upper == b"DEL" || cmd_upper == b"UNLINK"; // Group keys by shard in ascending order (BTreeMap = VLL) let mut groups: BTreeMap> = BTreeMap::new(); @@ -1060,10 +1163,30 @@ async fn coordinate_multi_del_or_exists( db.refresh_now_from_cache(cached_clock); cmd_dispatch(db, cmd, args, &mut selected, db_count) }); - return match result { + let resp = match result { DispatchResult::Response(f) => f, DispatchResult::Quit(f) => f, }; + // v3-5 carried gap: the in-process DEL/UNLINK never reached the AOF — + // deleted keys RESURRECTED from the seed writes on restart. Persist + // only when something was actually removed (n=0 replays identically + // without a record). + if is_delete && matches!(resp, Frame::Integer(n) if n > 0) { + let mut parts: Vec = Vec::with_capacity(args.len() + 1); + parts.push(Frame::BulkString(Bytes::from(cmd_upper.clone()))); + parts.extend_from_slice(args); + let serialized = + crate::persistence::aof::serialize_command(&Frame::Array(parts.into())); + match persist_local_leg(aof_pool, repl_state, my_shard, serialized).await { + Ok(needs_barrier) => *local_barrier_pending |= needs_barrier, + Err(()) => { + return Frame::Error(Bytes::from_static( + crate::persistence::aof::AOF_FSYNC_ERR, + )); + } + } + } + return resp; } let mut total_count: i64 = 0; @@ -1078,6 +1201,24 @@ async fn coordinate_multi_del_or_exists( }); if let DispatchResult::Response(Frame::Integer(n)) = result { total_count += n; + // v3-5 carried gap: persist the local slice (synthesized over + // ONLY the keys this shard owns — remote slices persist on + // their owners via MultiExecute). Skip when nothing removed. + if is_delete && n > 0 { + let mut parts: Vec = Vec::with_capacity(key_args.len() + 1); + parts.push(Frame::BulkString(Bytes::from(cmd_upper.clone()))); + parts.extend_from_slice(key_args); + let serialized = + crate::persistence::aof::serialize_command(&Frame::Array(parts.into())); + match persist_local_leg(aof_pool, repl_state, my_shard, serialized).await { + Ok(needs_barrier) => *local_barrier_pending |= needs_barrier, + Err(()) => { + return Frame::Error(Bytes::from_static( + crate::persistence::aof::AOF_FSYNC_ERR, + )); + } + } + } } } else { let (reply_tx, reply_rx) = channel::oneshot(); @@ -2392,6 +2533,7 @@ mod tests { let notifiers: Vec> = Vec::new(); let cached_clock = CachedClock::new(); let response_pool = (); + let mut local_barrier_pending = false; let result = coordinate_mset( &args, 0, @@ -2403,6 +2545,7 @@ mod tests { &cached_clock, None, &None, + &mut local_barrier_pending, &response_pool, ) .await; @@ -2441,6 +2584,7 @@ mod tests { let notifiers: Vec> = Vec::new(); let cached_clock = CachedClock::new(); let response_pool = (); + let mut local_barrier_pending = false; let result = coordinate_multi_del_or_exists( b"DEL", &args, @@ -2451,6 +2595,9 @@ mod tests { &dispatch_tx, ¬ifiers, &cached_clock, + None, + &None, + &mut local_barrier_pending, &response_pool, ) .await; diff --git a/tests/coordinator_local_leg_durability.rs b/tests/coordinator_local_leg_durability.rs index 378a9681c..235a07fcc 100644 --- a/tests/coordinator_local_leg_durability.rs +++ b/tests/coordinator_local_leg_durability.rs @@ -437,6 +437,289 @@ fn mset_scatter_local_slice_persists_across_restart() { // shard runs via `run_on_owner` → `run_local` with no AOF append. // --------------------------------------------------------------------------- +/// Like `assert_all_survive_restart`, but also asserts a set of keys is ABSENT +/// after the restart — for DEL/UNLINK legs, where the pre-fix failure mode is +/// RESURRECTION (the seed MSET is in the AOF, the local-leg DEL never was). +fn assert_state_after_restart( + port: u16, + dir: &std::path::Path, + child1: Child, + present: &[(String, String)], + absent: &[String], + what: &str, +) { + std::thread::sleep(Duration::from_millis(1500)); + sigkill(child1); + + let _guard = ServerGuard(spawn_moon_aof(port, dir, SHARDS)); + wait_ready(port); + + let mut c = Conn::open(port); + let mut missing: Vec = Vec::new(); + for (k, v) in present { + match c.cmd(&["GET", k]) { + Resp::Bulk(Some(got)) if got == v.as_bytes() => {} + _ => missing.push(k.clone()), + } + } + let mut resurrected: Vec = Vec::new(); + for k in absent { + match c.cmd(&["GET", k]) { + Resp::Bulk(None) => {} + _ => resurrected.push(k.clone()), + } + } + assert!( + missing.is_empty() && resurrected.is_empty(), + "{what}: {} kept keys missing {:?}; {} deleted keys RESURRECTED after restart \ + (the co-located leg whose owner == the connection's own shard never hit the AOF): {:?}", + missing.len(), + missing.iter().take(5).collect::>(), + resurrected.len(), + resurrected.iter().take(5).collect::>(), + ); +} + +// --------------------------------------------------------------------------- +// DEL — scatter path (ONE DEL spanning every shard). The slice owned by the +// connection's shard executes via cmd_dispatch in-process with no AOF append; +// pre-fix the deleted keys RESURRECT from the seed MSET on restart. +// --------------------------------------------------------------------------- + +#[test] +fn del_scatter_local_leg_persists_across_restart() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + wait_ready(port); + + let tags = tags_per_shard(SHARDS as usize); + let mut c = Conn::open(port); + + // Seed one co-located group per shard (MSET local leg persists post-v3-4). + let mut kept: Vec<(String, String)> = Vec::new(); + let mut to_delete: Vec = Vec::new(); + for tag in &tags { + let pairs = group_pairs(tag); + let mut argv: Vec<&str> = vec!["MSET"]; + for (k, v) in &pairs { + argv.push(k); + argv.push(v); + } + let resp = c.cmd(&argv); + if is_diskfull(&resp) { + eprintln!("SKIP del_scatter: MOONERR diskfull"); + return; + } + assert_eq!(resp, Resp::Simple("OK".to_string())); + // First key of each group gets deleted; the rest must survive. + to_delete.push(pairs[0].0.clone()); + kept.extend(pairs.into_iter().skip(1)); + } + + // ONE DEL spanning all shards → guaranteed scatter with a local slice. + let mut argv: Vec<&str> = vec!["DEL"]; + for k in &to_delete { + argv.push(k); + } + assert_eq!( + c.cmd(&argv), + Resp::Int(SHARDS as i64), + "scatter DEL should count one key per shard" + ); + drop(c); + + assert_state_after_restart( + port, + dir.path(), + child1, + &kept, + &to_delete, + "DEL scatter local-slice", + ); +} + +// --------------------------------------------------------------------------- +// UNLINK — co-located fast path (all keys of one UNLINK on one shard). The +// group whose owner == the connection's shard takes the `groups.len() == 1` +// fast path (cmd_dispatch in-process, no AOF) → resurrection pre-fix. +// --------------------------------------------------------------------------- + +#[test] +fn unlink_colocated_fastpath_persists_across_restart() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + wait_ready(port); + + let tags = tags_per_shard(SHARDS as usize); + let mut c = Conn::open(port); + + let mut deleted: Vec = Vec::new(); + for tag in &tags { + let pairs = group_pairs(tag); + let mut argv: Vec<&str> = vec!["MSET"]; + for (k, v) in &pairs { + argv.push(k); + argv.push(v); + } + let resp = c.cmd(&argv); + if is_diskfull(&resp) { + eprintln!("SKIP unlink_colocated: MOONERR diskfull"); + return; + } + assert_eq!(resp, Resp::Simple("OK".to_string())); + + // UNLINK the whole co-located group in one command (multi-key, one shard). + let keys: Vec = pairs.iter().map(|(k, _)| k.clone()).collect(); + let mut argv: Vec<&str> = vec!["UNLINK"]; + for k in &keys { + argv.push(k); + } + assert_eq!( + c.cmd(&argv), + Resp::Int(GROUP_SIZE as i64), + "co-located UNLINK on tag {tag} should count the whole group" + ); + deleted.extend(keys); + } + drop(c); + + assert_state_after_restart( + port, + dir.path(), + child1, + &[], + &deleted, + "UNLINK co-located fast-path", + ); +} + +// --------------------------------------------------------------------------- +// BITOP — the dest write leg. Scatter shape: sources on another shard force +// the gather path, then the synthesized `SET dest ` runs on dest's +// owner — in-process and un-persisted when that owner == the connection's +// shard. Fast-path shape: all keys co-located → whole BITOP via run_on_owner. +// --------------------------------------------------------------------------- + +#[test] +fn bitop_dest_local_leg_persists_across_restart() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + wait_ready(port); + + let tags = tags_per_shard(SHARDS as usize); + let n = tags.len(); + let mut c = Conn::open(port); + let mut expected: Vec<(String, String)> = Vec::new(); + + for (s, tag) in tags.iter().enumerate() { + // Scatter shape: sources live on the NEXT shard, dest on shard s. + let src_tag = &tags[(s + 1) % n]; + let src1 = format!("{{{}}}:b1", src_tag); + let src2 = format!("{{{}}}:b2", src_tag); + let dest = format!("{{{}}}:bdest", tag); + let r = c.cmd(&["SET", &src1, "aaa"]); + if is_diskfull(&r) { + eprintln!("SKIP bitop: MOONERR diskfull"); + return; + } + assert_eq!(r, Resp::Simple("OK".to_string())); + assert_eq!( + c.cmd(&["SET", &src2, "bbb"]), + Resp::Simple("OK".to_string()) + ); + assert_eq!( + c.cmd(&["BITOP", "OR", &dest, &src1, &src2]), + Resp::Int(3), + "scatter BITOP OR on dest tag {tag}" + ); + // 'a' | 'b' = 0x61 | 0x62 = 0x63 = 'c' + expected.push((dest, "ccc".to_string())); + + // Fast-path shape: sources AND dest all co-located on shard s. + let fsrc1 = format!("{{{}}}:f1", tag); + let fsrc2 = format!("{{{}}}:f2", tag); + let fdest = format!("{{{}}}:fdest", tag); + assert_eq!( + c.cmd(&["SET", &fsrc1, "aaa"]), + Resp::Simple("OK".to_string()) + ); + assert_eq!( + c.cmd(&["SET", &fsrc2, "bbb"]), + Resp::Simple("OK".to_string()) + ); + assert_eq!( + c.cmd(&["BITOP", "OR", &fdest, &fsrc1, &fsrc2]), + Resp::Int(3), + "co-located BITOP OR on tag {tag}" + ); + expected.push((fdest, "ccc".to_string())); + } + drop(c); + + assert_all_survive_restart(port, dir.path(), child1, &expected, "BITOP dest leg"); +} + +// --------------------------------------------------------------------------- +// COPY — the dst write leg. Cross-shard shape: src on another shard, the +// synthesized `SET dst NX` runs on dst's owner — un-persisted when +// that owner == the connection's shard. Same-owner shape: whole COPY forwards +// via run_on_owner (un-persisted local case). +// --------------------------------------------------------------------------- + +#[test] +fn copy_dst_local_leg_persists_across_restart() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + wait_ready(port); + + let tags = tags_per_shard(SHARDS as usize); + let n = tags.len(); + let mut c = Conn::open(port); + let mut expected: Vec<(String, String)> = Vec::new(); + + for (s, tag) in tags.iter().enumerate() { + // Cross-shard shape: src on the NEXT shard, dst on shard s. + let src_tag = &tags[(s + 1) % n]; + let src = format!("{{{}}}:csrc", src_tag); + let dst = format!("{{{}}}:cdst", tag); + let val = format!("{}-copyval", src_tag); + let r = c.cmd(&["SET", &src, &val]); + if is_diskfull(&r) { + eprintln!("SKIP copy: MOONERR diskfull"); + return; + } + assert_eq!(r, Resp::Simple("OK".to_string())); + assert_eq!( + c.cmd(&["COPY", &src, &dst]), + Resp::Int(1), + "cross-shard COPY to dst tag {tag}" + ); + expected.push((dst, val)); + + // Same-owner shape: src and dst co-located on shard s. + let fsrc = format!("{{{}}}:fsrc", tag); + let fdst = format!("{{{}}}:fdst", tag); + let fval = format!("{}-fcopyval", tag); + assert_eq!( + c.cmd(&["SET", &fsrc, &fval]), + Resp::Simple("OK".to_string()) + ); + assert_eq!( + c.cmd(&["COPY", &fsrc, &fdst]), + Resp::Int(1), + "same-owner COPY on tag {tag}" + ); + expected.push((fdst, fval)); + } + drop(c); + + assert_all_survive_restart(port, dir.path(), child1, &expected, "COPY dst leg"); +} + #[test] fn msetnx_colocated_local_leg_persists_across_restart() { let port = free_port(); From 1fd4ac71e758de5ad2e47b5d85b9884628096f2e Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 4 Jul 2026 23:53:18 +0700 Subject: [PATCH 3/4] test(integration): wait for cluster server bind instead of fixed 100ms sleep The cluster_* tests spawned an in-process cluster server, slept a fixed 100ms, then connected with a no-retry unwrap (integration.rs:242). Under full-suite parallelism the listener thread can take longer to bind -> "Connection refused" flakes (observed repeatedly in CI-parity runs; pass in isolation). Replace the sleep with a bounded connect-poll (10s), the same pattern the binary-spawning suites use. author: Tin Dang --- tests/integration.rs | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/tests/integration.rs b/tests/integration.rs index 161351805..1314eca3a 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -4060,8 +4060,19 @@ async fn start_cluster_server() -> (u16, CancellationToken) { } }); - // Give the server time to bind and start shards - tokio::time::sleep(std::time::Duration::from_millis(100)).await; + // Wait until the listener actually accepts instead of a fixed 100ms sleep — + // under full-suite parallelism the spawned listener thread can take longer + // to bind, and the callers' no-retry connect() then panics on refused. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + match tokio::net::TcpStream::connect(("127.0.0.1", port)).await { + Ok(_) => break, + Err(_) if std::time::Instant::now() < deadline => { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + Err(e) => panic!("cluster server never bound port {port}: {e}"), + } + } (port, token) } From 7e7304aea811da1a9b22d7c4cb6383732303a705 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 5 Jul 2026 00:39:36 +0700 Subject: [PATCH 4/4] fix(persistence): resolve local-leg barrier before early response flushes (PR #213 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings from PR #213 (CodeRabbit + Qodo), all addressed: 1. [CRITICAL, CodeRabbit] The batch-end local-leg barrier could be skipped or corrupted by mid-batch early flushes: - blocking commands (BLPOP...) flush accumulated responses then REPLACE the vec — pending barrier indexes became stale (panic or misattributed AOF_FSYNC_ERR onto the blocking response) and the flushed +OK escaped without confirmed durability; - PSYNC hijack (monoio) and SUBSCRIBE-entry (both runtimes) flush early with the same durability leak. Fix: new shared::resolve_local_leg_barrier(pool, shard, idxs, responses) — always drains, patches AOF_FSYNC_ERR on barrier failure — called at the batch end AND before every early flush (blocking, SUBSCRIBE entry, PSYNC) in both handlers. 2. [Qodo bug] run_on_owner_persist logged no-op writes (COPY SET..NX refusal, DEL of absent dest, PEXPIRE on vanished key) — costing a needless barrier fsync and risking AOF_FSYNC_ERR on a command that wrote nothing. Fix: persist_if predicate per call site (BITOP forward always mutates; DEL only n>0; COPY only :1; SET dst only +OK; PEXPIRE only :1). 3. [Qodo rule] #[allow(clippy::too_many_arguments)] now carries its justification; CHANGELOG latency numbers now state their Linux/GCE measurement context. Tests: coordinator_local_leg_durability 7/7, full lib suite 3656 pass, clippy clean both feature sets. The barrier-failure arm of the early flush paths is not black-box testable (requires an AOF writer dying mid-pipeline); covered by the shared helper's single code path + the existing fsync_barrier unit tests. author: Tin Dang --- CHANGELOG.md | 5 +-- src/server/conn/handler_monoio/dispatch.rs | 12 +++++++ src/server/conn/handler_monoio/mod.rs | 40 +++++++++++++-------- src/server/conn/handler_monoio/pubsub.rs | 10 ++++++ src/server/conn/handler_sharded/mod.rs | 41 +++++++++++++--------- src/server/conn/handler_sharded/pubsub.rs | 10 ++++++ src/server/conn/shared.rs | 36 +++++++++++++++++++ src/shard/coordinator.rs | 21 +++++++++-- 8 files changed, 140 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 641b5b93a..1d97a373e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,8 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - The cross-shard coordinator's LOCAL-leg persist (co-located MSET/MSETNX and scattered-MSET local slices) awaited one fsync ack **per command**, each bounded by `--aof-fsync-timeout-ms` (default 2000ms) — a pipeline of - coordinated writes stacked these serially into the measured 2000–3000ms - `always` far-tail. Local legs now enqueue fire-and-forget (bounded + coordinated writes stacked these serially into the 2000–3000ms `always` + far-tail measured on Linux/GCE (c2d-standard-16, pd-ssd; see the v3-4 + bench notes — OrbStack/macOS fsync is near-free and does not reproduce it). Local legs now enqueue fire-and-forget (bounded backpressure, same contract as the remote SPSC legs) and the connection handler confirms them with **one** `fsync_barrier` on the local shard per pipeline batch, before responses are serialized — so `+OK` still implies diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 95a524e48..336a39682 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -1328,6 +1328,7 @@ pub(super) async fn try_handle_blocking( conn: &mut ConnectionState, ctx: &ConnectionContext, responses: &mut Vec, + local_leg_write_idxs: &mut Vec, codec: &mut crate::server::codec::RespCodec, write_buf: &mut bytes::BytesMut, stream: &mut S, @@ -1353,6 +1354,17 @@ pub(super) async fn try_handle_blocking( return BlockingResult::Queued; } + // Earlier frames in this batch may hold barrier-pending local-leg + // writes — confirm (or fail-loud) them before this early flush, and + // clear the indexes so the batch-end barrier never sees stale ones. + crate::server::conn::shared::resolve_local_leg_barrier( + &ctx.aof_pool, + ctx.shard_id, + local_leg_write_idxs, + responses, + ) + .await; + // Flush accumulated responses before blocking for resp in &*responses { codec.encode_frame(resp, write_buf); diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 68c2e2e69..4e7330369 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -756,6 +756,15 @@ pub(crate) async fn handle_connection_sharded_monoio< if let Some((repl_id, offset)) = dispatch::try_handle_psync(cmd, cmd_args, ctx, &mut responses) { + // Earlier frames in this batch may hold barrier-pending + // local-leg writes — confirm them before this early flush. + crate::server::conn::shared::resolve_local_leg_barrier( + &ctx.aof_pool, + ctx.shard_id, + &mut local_leg_write_idxs, + &mut responses, + ) + .await; for resp in &responses { codec.encode_frame(resp, &mut write_buf); } @@ -824,6 +833,7 @@ pub(crate) async fn handle_connection_sharded_monoio< ctx, &peer_addr, &mut responses, + &mut local_leg_write_idxs, &mut codec, &mut write_buf, &mut stream, @@ -924,6 +934,7 @@ pub(crate) async fn handle_connection_sharded_monoio< &mut conn, ctx, &mut responses, + &mut local_leg_write_idxs, &mut codec, &mut write_buf, &mut stream, @@ -1813,21 +1824,20 @@ pub(crate) async fn handle_connection_sharded_monoio< } } - // v3-5 GROUP-COMMIT BARRIER: coordinator LOCAL legs (MSET/MSETNX) were - // enqueued fire-and-forget into MY shard's AOF writer during dispatch; - // one barrier here confirms every one of them with a single fsync - // instead of the retired per-command awaited fsync (2000ms tail stack). - // Runs BEFORE response serialization — the client never sees +OK for a - // write whose durability was not confirmed. - if !local_leg_write_idxs.is_empty() { - if let Some(ref pool) = ctx.aof_pool { - if pool.fsync_barrier(ctx.shard_id).await.is_err() { - for idx in local_leg_write_idxs.drain(..) { - responses[idx] = Frame::Error(Bytes::from_static(aof::AOF_FSYNC_ERR)); - } - } - } - } + // v3-5 GROUP-COMMIT BARRIER: coordinator LOCAL legs were enqueued + // fire-and-forget into MY shard's AOF writer during dispatch; one + // barrier here confirms every one of them with a single fsync instead + // of the retired per-command awaited fsync (2000ms tail stack). Runs + // BEFORE response serialization — the client never sees +OK for a + // write whose durability was not confirmed. Early-flush paths (PSYNC, + // blocking, SUBSCRIBE) resolve the same barrier before THEIR flushes. + crate::server::conn::shared::resolve_local_leg_barrier( + &ctx.aof_pool, + ctx.shard_id, + &mut local_leg_write_idxs, + &mut responses, + ) + .await; // AUTH rate limiting: delay response to slow down brute-force attacks if auth_delay_ms > 0 { diff --git a/src/server/conn/handler_monoio/pubsub.rs b/src/server/conn/handler_monoio/pubsub.rs index e7ac03f2e..1a2511303 100644 --- a/src/server/conn/handler_monoio/pubsub.rs +++ b/src/server/conn/handler_monoio/pubsub.rs @@ -124,6 +124,7 @@ pub(super) async fn try_handle_subscribe_entry( ctx: &super::super::core::ConnectionContext, peer_addr: &str, responses: &mut Vec, + local_leg_write_idxs: &mut Vec, codec: &mut crate::server::codec::RespCodec, write_buf: &mut bytes::BytesMut, stream: &mut S, @@ -154,6 +155,15 @@ pub(super) async fn try_handle_subscribe_entry( if conn.subscriber_id == 0 { conn.subscriber_id = crate::pubsub::next_subscriber_id(); } + // Earlier frames in this batch may hold barrier-pending local-leg + // writes — confirm (or fail-loud) them before this early flush. + crate::server::conn::shared::resolve_local_leg_barrier( + &ctx.aof_pool, + ctx.shard_id, + local_leg_write_idxs, + responses, + ) + .await; // Flush accumulated responses before entering subscriber mode for resp in &*responses { codec.encode_frame(resp, write_buf); diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 29a209933..ff439aa2c 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -877,6 +877,17 @@ pub(crate) async fn handle_connection_sharded_inner< responses.push(Frame::SimpleString(Bytes::from_static(b"QUEUED"))); continue; } + // Earlier frames in this batch may hold barrier-pending + // local-leg writes — confirm (or fail-loud) them before + // this early flush; the replacement of `responses` below + // would otherwise leave stale indexes (PR #213 review). + crate::server::conn::shared::resolve_local_leg_barrier( + &ctx.aof_pool, + ctx.shard_id, + &mut local_leg_write_idxs, + &mut responses, + ) + .await; write_buf.clear(); for response in responses.iter() { if conn.protocol_version >= 3 { @@ -905,6 +916,7 @@ pub(crate) async fn handle_connection_sharded_inner< if let Some(action) = pubsub::try_handle_subscribe( cmd, cmd_args, &mut stream, &mut write_buf, &mut conn, ctx, &peer_addr, &mut responses, + &mut local_leg_write_idxs, ).await { match action { pubsub::SubscriberAction::Continue => { continue; } @@ -1716,22 +1728,19 @@ pub(crate) async fn handle_connection_sharded_inner< } } - // v3-5 GROUP-COMMIT BARRIER: coordinator LOCAL legs (MSET/MSETNX) - // were enqueued fire-and-forget into MY shard's AOF writer during - // dispatch; ONE barrier confirms all of them with a single fsync - // instead of the retired per-command awaited fsync. Runs BEFORE - // response serialization — no +OK without confirmed durability. - if !local_leg_write_idxs.is_empty() { - if let Some(ref pool) = ctx.aof_pool { - if pool.fsync_barrier(ctx.shard_id).await.is_err() { - for idx in local_leg_write_idxs.drain(..) { - responses[idx] = Frame::Error( - Bytes::from_static(aof::AOF_FSYNC_ERR), - ); - } - } - } - } + // v3-5 GROUP-COMMIT BARRIER: coordinator LOCAL legs were enqueued + // fire-and-forget into MY shard's AOF writer during dispatch; ONE + // barrier confirms all of them with a single fsync instead of the + // retired per-command awaited fsync. Runs BEFORE response + // serialization — no +OK without confirmed durability. Early-flush + // paths (blocking, SUBSCRIBE) resolve the same barrier first. + crate::server::conn::shared::resolve_local_leg_barrier( + &ctx.aof_pool, + ctx.shard_id, + &mut local_leg_write_idxs, + &mut responses, + ) + .await; // Phase 3: Flush accumulated PUBLISH batches as PubSubPublishBatch messages if !publish_batches.is_empty() { diff --git a/src/server/conn/handler_sharded/pubsub.rs b/src/server/conn/handler_sharded/pubsub.rs index 05f6d6c90..f8b8c8095 100644 --- a/src/server/conn/handler_sharded/pubsub.rs +++ b/src/server/conn/handler_sharded/pubsub.rs @@ -269,6 +269,7 @@ pub(super) async fn try_handle_subscribe< ctx: &ConnectionContext, peer_addr: &str, responses: &mut Vec, + local_leg_write_idxs: &mut Vec, ) -> Option { use tokio::io::AsyncWriteExt; @@ -297,6 +298,15 @@ pub(super) async fn try_handle_subscribe< if conn.subscriber_id == 0 { conn.subscriber_id = crate::pubsub::next_subscriber_id(); } + // Earlier frames in this batch may hold barrier-pending local-leg + // writes — confirm (or fail-loud) them before this early flush. + crate::server::conn::shared::resolve_local_leg_barrier( + &ctx.aof_pool, + ctx.shard_id, + local_leg_write_idxs, + responses, + ) + .await; // Flush accumulated responses before entering subscriber mode if !responses.is_empty() { write_buf.clear(); diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index e8ef4fa5d..82685e431 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -525,3 +525,39 @@ mod as_of_tests { } } } + +/// Resolve the pending v3-5 local-leg group-commit barrier, if any. +/// +/// Coordinator local-leg writes (MSET/MSETNX/BITOP/COPY/DEL/UNLINK legs owned +/// by the connection's own shard) enqueue their AOF append fire-and-forget +/// under `appendfsync=always` and record their response index here; ONE +/// `fsync_barrier` per batch confirms them all. This MUST run before ANY +/// flush of `responses` to the client — the batch end, but also the early +/// flushes (blocking commands, SUBSCRIBE entry, PSYNC hijack). Skipping it +/// there would (a) ack a write whose durability was never confirmed and +/// (b) leave stale indexes that panic or misattribute errors when the +/// response vec is replaced (PR #213 review finding). +/// +/// Always drains `idxs`. On barrier failure every recorded response is +/// overwritten with `AOF_FSYNC_ERR` — never a false `+OK`. +pub async fn resolve_local_leg_barrier( + aof_pool: &Option>, + shard_id: usize, + idxs: &mut Vec, + responses: &mut [Frame], +) { + if idxs.is_empty() { + return; + } + if let Some(pool) = aof_pool { + if pool.fsync_barrier(shard_id).await.is_err() { + for idx in idxs.iter() { + if let Some(slot) = responses.get_mut(*idx) { + *slot = + Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + } + } + } + } + idxs.clear(); +} diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index d1313bdf3..6ef9fb6bd 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -245,10 +245,15 @@ async fn run_on_owner( /// whose keys are all owned by my_shard, or a synthesized write like /// `SET dest ` for a scatter BITOP/COPY). /// -/// Persists only when the local execution did not error; sets +/// Persists only when the local execution did not error AND `persist_if` +/// says the response indicates an actual mutation (e.g. `SET ... NX` refusal +/// returns Null and wrote nothing — logging it would cost a needless barrier +/// fsync and could fail a no-op with `AOF_FSYNC_ERR`). Sets /// `*local_barrier_pending` when the append rides group commit under /// `appendfsync=always` (the handler owes ONE `fsync_barrier(my_shard)` per /// batch). Returns `AOF_FSYNC_ERR` when the append never reached the writer. +// Mirrors run_on_owner's routing params + the persistence context; bundling +// them into a struct would obscure the 1:1 correspondence with run_on_owner. #[allow(clippy::too_many_arguments)] async fn run_on_owner_persist( routing_key: &Bytes, @@ -263,6 +268,7 @@ async fn run_on_owner_persist( aof_pool: Option<&Arc>, repl_state: ReplStateRef<'_>, local_barrier_pending: &mut bool, + persist_if: impl Fn(&Frame) -> bool, ) -> Frame { let owner = key_to_shard(routing_key, num_shards); if owner != my_shard { @@ -283,7 +289,7 @@ async fn run_on_owner_persist( _ => return Frame::Error(Bytes::from_static(b"ERR invalid command format")), }; let resp = run_local(shard_databases, db_index, cached_clock, &cmd, args); - if !matches!(resp, Frame::Error(_)) { + if !matches!(resp, Frame::Error(_)) && persist_if(&resp) { let serialized = crate::persistence::aof::serialize_command(&Frame::Array( command_parts.to_vec().into(), )); @@ -432,6 +438,7 @@ async fn coordinate_bitop( parts.push(bulk_static(b"BITOP")); parts.extend_from_slice(args); // Write command: the local-owner case must persist (v3-5 carried gap). + // Any non-error BITOP mutates dest (SET result, or DEL on empty). return run_on_owner_persist( &dest, &parts, @@ -445,6 +452,7 @@ async fn coordinate_bitop( aof_pool, repl_state, local_barrier_pending, + |_| true, ) .await; } @@ -532,6 +540,8 @@ async fn coordinate_bitop( aof_pool, repl_state, local_barrier_pending, + // DEL of an absent dest wrote nothing — skip the no-op record. + |r| matches!(r, Frame::Integer(n) if *n > 0), ) .await; if let Frame::Error(e) = reply { @@ -558,6 +568,7 @@ async fn coordinate_bitop( aof_pool, repl_state, local_barrier_pending, + |_| true, // plain SET always writes on success ) .await; if let Frame::Error(e) = reply { @@ -638,6 +649,8 @@ async fn coordinate_copy( aof_pool, repl_state, local_barrier_pending, + // COPY :0 = refused (dst exists, no REPLACE) — nothing written. + |r| matches!(r, Frame::Integer(1)), ) .await; } @@ -707,6 +720,8 @@ async fn coordinate_copy( aof_pool, repl_state, local_barrier_pending, + // Null = NX refused (dst exists) — nothing was written. + |r| matches!(r, Frame::SimpleString(_)), ) .await; match set_reply { @@ -734,6 +749,8 @@ async fn coordinate_copy( aof_pool, repl_state, local_barrier_pending, + // :0 = key vanished between SET and PEXPIRE — no TTL was set. + |r| matches!(r, Frame::Integer(1)), ) .await; if let Frame::Error(e) = reply {