perf(persistence): local-leg group commit + BITOP/COPY/DEL/UNLINK coordinator leg durability

CHANGELOG.md

@@ -6,6 +6,35 @@ 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`).
+
+### 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

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<bool, AofAck> {
+        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::<AofMessage>(4);
+        let (tx1, _rx1) = channel::mpsc_bounded::<AofMessage>(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::<AofMessage>(4);
+        let (tx1, _rx1) = channel::mpsc_bounded::<AofMessage>(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::<AofMessage>(4);
+        let (tx1, _rx1) = channel::mpsc_bounded::<AofMessage>(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.

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<Frame>,
+    // 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<usize>,
 ) -> 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;
     }

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>, 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<usize> = 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<usize, Vec<(usize, Bytes, Bytes)>> =
             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;

src/server/conn/handler_sharded/mod.rs

@@ -374,6 +374,9 @@ pub(crate) async fn handle_connection_sharded_inner<
                 }
 
                 let mut responses: Vec<Frame> = 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<usize> = Vec::new();
                 let mut should_quit = false;
                 let mut remote_groups: HashMap<usize, Vec<(usize, std::sync::Arc<Frame>, Option<Bytes>, 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<crate::shard::dispatch::PubSubResponseSlot>, Vec<usize>)> = Vec::new();