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

CHANGELOG.md

@@ -6,6 +6,36 @@ 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 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
+  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`.
+
 ### Fixed — cold-tier (disk offload) correctness & reliability (PR #TBD)
 
 - **DEL/UNLINK/FLUSHALL now reach the cold tier** — deleting a key whose value

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;
     }
@@ -1318,6 +1328,7 @@ pub(super) async fn try_handle_blocking<S: monoio::io::AsyncWriteRent>(
     conn: &mut ConnectionState,
     ctx: &ConnectionContext,
     responses: &mut Vec<Frame>,
+    local_leg_write_idxs: &mut Vec<usize>,
     codec: &mut crate::server::codec::RespCodec,
     write_buf: &mut bytes::BytesMut,
     stream: &mut S,
@@ -1343,6 +1354,17 @@ pub(super) async fn try_handle_blocking<S: monoio::io::AsyncWriteRent>(
         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);

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();
 
@@ -751,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);
                     }
@@ -819,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,
@@ -919,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,
@@ -940,8 +956,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 +1824,21 @@ pub(crate) async fn handle_connection_sharded_monoio<
             }
         }
 
+        // 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 {
             monoio::time::sleep(std::time::Duration::from_millis(auth_delay_ms)).await;

src/server/conn/handler_monoio/pubsub.rs

@@ -124,6 +124,7 @@ pub(super) async fn try_handle_subscribe_entry<S: monoio::io::AsyncWriteRent>(
     ctx: &super::super::core::ConnectionContext,
     peer_addr: &str,
     responses: &mut Vec<Frame>,
+    local_leg_write_idxs: &mut Vec<usize>,
     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<S: monoio::io::AsyncWriteRent>(
     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);