fix(server): sharded MULTI/EXEC safety, durability & owner-routing (Phase A+B)

CHANGELOG.md

@@ -6,6 +6,72 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added — sharded MULTI/EXEC routes a single-owner-shard body to its owner (Phase B, PR #TBD)
+
+- **`src/shard/{dispatch,spsc_handler,coordinator}.rs`,
+  `src/server/conn/{handler_sharded,handler_monoio}/write.rs`**: with keys routed
+  per-key to their owning shard but MULTI/EXEC executing on the connection's
+  (random, SO_REUSEPORT) accept shard, a hash-tagged `{tag}` transaction only
+  worked from the ~1/N connections that happened to land on the owning shard —
+  Phase A rejected the rest with `CROSSSLOT`. Phase B adds a boxed
+  `ShardMessage::TxnExecute` (kept within the 64-byte cache-line cap like
+  `MqCommand`) and a `coordinator::execute_txn_on_owner` hop: when
+  `analyze_txn_locality` proves every key is owned by ONE shard that isn't the
+  accept shard, the whole body is routed there, executed atomically on the
+  owner's slice, and each write persisted to the OWNER's AOF/WAL via the same
+  `wal_append_and_fanout` path as normal cross-shard writes. PUBLISH fan-out is
+  deferred back to the originating connection (returned in the reply) so it keeps
+  the normal scatter path; under `appendfsync=always` the originator issues one
+  `fsync_barrier` to the owner before acking (H1-BARRIER parity), and owner-side
+  append backpressure surfaces `AOF_APPEND_LOST_ERR`. Genuinely multi-shard
+  bodies (`CrossShard`) stay rejected — a shared-nothing engine can't commit
+  across shards atomically. Red/green: `tests/sharded_multi_exec_routing.rs`
+  (hash-tagged txn succeeds from every connection + writes visible; routed writes
+  survive kill-9 + restart via the owner's AOF; multi-shard span still CROSSSLOT).
+  WATCH in sharded mode remains an unimplemented follow-up.
+
+### Fixed — sharded MULTI/EXEC writes are now persisted (were lost on restart) (PR #TBD)
+
+- **`src/server/conn/{shared,handler_sharded/write,handler_monoio/write}.rs`**:
+  `execute_transaction_sharded` — the MULTI/EXEC executor used by the monoio
+  handler at **every** shard count (including `--shards 1`) and by the tokio
+  sharded handler at `--shards ≥ 2` — appended **nothing** to the AOF. Every
+  transactional write was silently lost on restart under `appendonly=yes`, while
+  an identical write issued outside MULTI survived. The executor now returns the
+  serialized AOF bytes for each successful write (mirroring the single-shard
+  tokio `execute_transaction`), and a shared `persist_txn_aof` helper appends
+  them to the owning shard's writer via the normal group-commit path, issuing one
+  `fsync_barrier` under `appendfsync=always` before EXEC is acked. On a barrier
+  failure EXEC returns `AOF_FSYNC_ERR` and suppresses any queued PUBLISH fan-out,
+  rather than acking a durability it can't guarantee. `try_handle_multi_exec` is
+  now `async` in both sharded handlers. Red/green:
+  `tests/sharded_multi_exec_durability.rs` pins *EXEC-committed write survives
+  kill-9 + restart, exactly like a non-MULTI write* (fails on the pre-fix path:
+  the txn key returns nil after restart while the plain control key survives).
+
+### Security — sharded MULTI/EXEC no longer silently misplaces cross-shard writes (Phase A, PR #TBD)
+
+- **`src/server/conn/{shared,handler_sharded/write,handler_monoio/write}.rs`**:
+  `execute_transaction_sharded` runs the whole queued body on the connection's
+  OWN shard with no per-key routing, so at `--shards ≥ 2` a key owned by another
+  shard was silently written to / read from the wrong shard's table — EXEC
+  reported success while the data diverged (silent lost updates; other
+  connections routing to the true owner saw nothing). Hash tags did **not** help:
+  they co-locate keys with each other but not with the (random, SO_REUSEPORT)
+  execution shard, and migration is disabled during MULTI. A new
+  `analyze_txn_locality` classifies the queued body via the command-metadata key
+  specs + `key_to_shard`; EXEC now **rejects** with `CROSSSLOT` any transaction
+  whose keys aren't all owned by the executing shard, instead of corrupting.
+  Invariant restored: *EXEC-success ⇒ every write is visible to other
+  connections; CROSSSLOT ⇒ nothing was written.* No effect at `--shards 1`.
+  Red/green: `tests/sharded_multi_exec_locality.rs` (silent-divergence invariant,
+  multi-shard span rejection, single-shard unaffected) + `analyze_txn_locality`
+  unit tests. Phase B (above) now routes single-owner-shard bodies to their owner
+  instead of rejecting them, so only genuinely cross-shard bodies still get
+  CROSSSLOT. Note found in passing (unrelated, out of scope): a *solo* GET sent
+  mid-MULTI on the monoio single-shard handler executes immediately instead of
+  queueing.
+
 ### Docs — tuning guide: vector bulk load & compaction (PR #TBD)
 
 - `docs/guides/tuning.md`: new "Vector bulk load and compaction" section — documents

src/server/conn/handler_monoio/mod.rs

@@ -1006,7 +1006,9 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 ctx,
                 &mut responses,
                 &mut exec_publishes,
-            ) {
+            )
+            .await
+            {
                 // C2: PUBLISH queued inside MULTI fans out only now — after the
                 // transaction body has been applied — and its placeholder in the
                 // EXEC reply array is patched with the real receiver count.

src/server/conn/handler_monoio/write.rs

@@ -562,7 +562,12 @@ async fn mq_hop_or_local(
 }
 
 /// Handle MULTI/EXEC/DISCARD commands. Returns `true` if consumed.
-pub(super) fn try_handle_multi_exec(
+///
+/// `async` because EXEC now persists the transaction body to the shard AOF via
+/// the same group-commit path as normal writes (previously this path logged
+/// nothing, so every transactional write was lost on restart — including at
+/// `--shards 1` under the monoio TopLevel writer).
+pub(super) async fn try_handle_multi_exec(
     cmd: &[u8],
     conn: &mut ConnectionState,
     ctx: &ConnectionContext,
@@ -591,14 +596,143 @@ pub(super) fn try_handle_multi_exec(
             responses.push(Frame::Error(Bytes::from_static(b"ERR EXEC without MULTI")));
         } else {
             conn.in_multi = false;
-            let result = execute_transaction_sharded(
+            // The body runs on THIS shard with no per-key routing, so a
+            // foreign-owned key would be silently misplaced. Classify locality:
+            //  - CrossShard: genuinely spans shards — a shared-nothing engine
+            //    can't commit it atomically, so reject with CROSSSLOT.
+            //  - SingleShard(other): every key is owned by ONE remote shard —
+            //    Phase B routes the whole body there for atomic execution + AOF
+            //    on the owner (instead of the Phase-A CROSSSLOT rejection).
+            //  - Keyless / SingleShard(self): fall through to local execution.
+            if ctx.num_shards > 1 {
+                match crate::server::conn::shared::analyze_txn_locality(
+                    &conn.command_queue,
+                    ctx.num_shards,
+                ) {
+                    crate::server::conn::shared::TxnLocality::SingleShard(s)
+                        if s != ctx.shard_id =>
+                    {
+                        let commands: Vec<Frame> = conn.command_queue.to_vec();
+                        conn.command_queue.clear();
+                        // Keep a copy of the body for CLIENT TRACKING
+                        // invalidation ONLY when tracking is active — the routed
+                        // reply carries results, not the command frames, and the
+                        // owner does not invalidate (the tracking table is
+                        // process-global; invalidation is issued originating-side).
+                        let tracking_cmds =
+                            crate::tracking::tracking_active().then(|| commands.clone());
+                        let reply = crate::shard::coordinator::execute_txn_on_owner(
+                            s,
+                            ctx.shard_id,
+                            conn.selected_db,
+                            commands,
+                            &ctx.dispatch_tx,
+                            &ctx.spsc_notifiers,
+                        )
+                        .await;
+                        match reply {
+                            Some(r) => {
+                                if r.append_lost {
+                                    exec_publishes.clear();
+                                    responses.push(Frame::Error(Bytes::from_static(
+                                        crate::shard::spsc_handler::AOF_APPEND_LOST_ERR,
+                                    )));
+                                    return true;
+                                }
+                                // appendfsync=always: confirm the owner's queued
+                                // appends are on disk before acking (H1-BARRIER
+                                // parity for normal cross-shard writes). No-op
+                                // under everysec/no.
+                                if r.wrote {
+                                    if let Some(ref pool) = ctx.aof_pool {
+                                        if pool.fsync_barrier(s).await.is_err() {
+                                            exec_publishes.clear();
+                                            responses.push(Frame::Error(Bytes::from_static(
+                                                crate::persistence::aof::AOF_FSYNC_ERR,
+                                            )));
+                                            return true;
+                                        }
+                                    }
+                                }
+                                // CLIENT TRACKING: invalidate every key written by
+                                // the routed body, same as the local EXEC path
+                                // (which the early return would otherwise skip).
+                                if let Some(cmds) = tracking_cmds.as_ref() {
+                                    if let Frame::Array(ref txn_results) = r.result {
+                                        for (i, cmd_frame) in cmds.iter().enumerate() {
+                                            if i >= txn_results.len()
+                                                || matches!(txn_results[i], Frame::Error(_))
+                                            {
+                                                continue;
+                                            }
+                                            if let Some((c, a)) =
+                                                crate::server::conn::util::extract_command(
+                                                    cmd_frame,
+                                                )
+                                            {
+                                                crate::tracking::invalidation::invalidate_after_write(
+                                                    &ctx.tracking_table,
+                                                    c,
+                                                    a,
+                                                    conn.client_id,
+                                                );
+                                            }
+                                        }
+                                    }
+                                }
+                                // Adopt the owner's deferred PUBLISH fan-out; the
+                                // caller's post-EXEC loop patches placeholders +
+                                // scatters from this (originating) shard.
+                                exec_publishes.extend(r.exec_publishes);
+                                responses.push(r.result);
+                            }
+                            None => {
+                                responses.push(Frame::Error(Bytes::from_static(
+                                    b"CROSSSLOT MULTI/EXEC owner shard unavailable; retry",
+                                )));
+                            }
+                        }
+                        return true;
+                    }
+                    crate::server::conn::shared::TxnLocality::CrossShard => {
+                        conn.command_queue.clear();
+                        responses.push(Frame::Error(Bytes::from_static(
+                            b"CROSSSLOT Keys in MULTI/EXEC don't hash to the same shard",
+                        )));
+                        return true;
+                    }
+                    _ => {}
+                }
+            }
+            let (result, aof_entries) = execute_transaction_sharded(
                 &ctx.shard_databases,
                 ctx.shard_id,
                 &conn.command_queue,
                 conn.selected_db,
                 &ctx.cached_clock,
                 exec_publishes,
             );
+            // DURABILITY: append every successful write in the body to THIS
+            // shard's AOF via the same group-commit path as normal writes, then
+            // issue ONE fsync barrier under appendfsync=always before acking.
+            // All keys are local here (Phase A rejected foreign-owned bodies),
+            // so ctx.shard_id is the correct AOF target. On barrier failure we
+            // surface AOF_FSYNC_ERR instead of a false EXEC success — parity
+            // with the normal write path.
+            if crate::server::conn::shared::persist_txn_aof(ctx, aof_entries)
+                .await
+                .is_err()
+            {
+                conn.command_queue.clear();
+                // Durability could not be guaranteed: report the error and
+                // suppress any queued PUBLISH fan-out — the client sees EXEC
+                // fail, so it must not observe the txn's pub/sub side effects.
+                exec_publishes.clear();
+                responses.push(Frame::Error(Bytes::from_static(
+                    crate::persistence::aof::AOF_FSYNC_ERR,
+                )));
+                return true;
+            }
             // CLIENT TRACKING: invalidate keys written inside the txn, same as
             // the normal write path (EXEC previously bypassed this). Self-gated
             // on tracking_active(); must run before command_queue is cleared.

src/server/conn/handler_sharded/mod.rs

@@ -927,7 +927,7 @@ pub(crate) async fn handle_connection_sharded_inner<
 
                     // --- MULTI / EXEC_CMD / DISCARD ---
                     let mut exec_publishes: Vec<(usize, Bytes, Bytes)> = Vec::new();
-                    if write::try_handle_multi_exec(cmd, &mut conn, ctx, &mut responses, &mut exec_publishes) {
+                    if write::try_handle_multi_exec(cmd, &mut conn, ctx, &mut responses, &mut exec_publishes).await {
                         // C2: PUBLISH queued inside MULTI fans out only now — after the
                         // transaction body has been applied — and its placeholder in the
                         // EXEC reply array is patched with the real receiver count.