fix(shard): FLUSHALL inside MULTI/EXEC clears every shard, not just one (c10k E2)

CHANGELOG.md

@@ -148,6 +148,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   when authentication is configured and the shard stays on the tokio path.
 
 ### Fixed
+- **`FLUSHALL`/`FLUSHDB` inside `MULTI`/`EXEC` cleared one shard and reported
+  success (c10k E2).** The transactional executor runs the queued body against
+  the LOCAL slice with no fan-out, so a queued flush emptied a single shard of
+  N while `EXEC` still answered `+OK`. Measured at `--shards 4`: 45 of 64 keys
+  survived a transaction that reported the database emptied — a silent wrong
+  answer to a destructive command, typically noticed much later via a non-zero
+  `DBSIZE`. The live (non-`MULTI`) path has broadcast since D-2; the
+  transactional path now does the same. The executor records each flush and the
+  ORIGINATOR fans it out — it cannot fan out itself, being synchronous while
+  the broadcast awaits, and for a routed transaction it runs on the owner
+  shard, where broadcasting from inside that shard's own message loop risks a
+  shard-to-shard wait cycle. A failed leg replaces that entry of the `EXEC`
+  result array with an explicit partial-flush error, so a `+OK` for a flush
+  inside a transaction can be trusted exactly as on the live path. Both
+  handlers and both transaction shapes are covered, and a queued `SELECT`
+  before the flush is honoured. Cross-shard atomicity is unchanged and
+  unchangeable: a concurrent reader can still see shard A flushed before shard
+  B — `MULTI` bounds the report, not the visibility, in a shared-nothing
+  engine.
+
 - **TLS park veto samples `wants_write()` after processing, not before (c10k
   B5).** `Stream::task_park_safe` read `wants_write()` before
   `process_new_packets()`, which can queue outbound bytes and still return

src/server/conn/handler_monoio/write.rs

@@ -821,7 +821,21 @@ pub(super) async fn try_handle_multi_exec(
                                 // caller's post-EXEC loop patches placeholders +
                                 // scatters from this (originating) shard.
                                 exec_publishes.extend(r.exec_publishes);
-                                responses.push(r.result);
+                                // c10k E2: the owner flushed only ITS slice.
+                                // Broadcast to the rest from here (the owner
+                                // cannot fan out from inside its own message
+                                // loop) and patch the result on a failed leg.
+                                let mut routed_result = r.result;
+                                crate::shard::coordinator::broadcast_txn_flushes(
+                                    &mut routed_result,
+                                    &r.exec_flushes,
+                                    s,
+                                    ctx.num_shards,
+                                    &ctx.dispatch_tx,
+                                    &ctx.spsc_notifiers,
+                                )
+                                .await;
+                                responses.push(routed_result);
                             }
                             None => {
                                 responses.push(Frame::Error(Bytes::from_static(
@@ -841,13 +855,16 @@ pub(super) async fn try_handle_multi_exec(
                     _ => {}
                 }
             }
-            let (result, aof_entries, graph_records) = execute_transaction_sharded(
+            // c10k E2: a queued FLUSHDB/FLUSHALL clears only this shard.
+            let mut exec_flushes: Vec<(usize, Frame, usize)> = Vec::new();
+            let (mut result, aof_entries, graph_records) = execute_transaction_sharded(
                 &ctx.shard_databases,
                 ctx.shard_id,
                 &conn.command_queue,
                 conn.selected_db,
                 &ctx.cached_clock,
                 exec_publishes,
+                &mut exec_flushes,
             );
             // v0.7 REPLICATION (adversarial-review P0-1): the txn body must
             // reach replicas like any other successful local write. This was
@@ -949,6 +966,20 @@ pub(super) async fn try_handle_multi_exec(
                 }
             }
             conn.command_queue.clear();
+            // c10k E2: the body cleared only THIS shard's slice. Broadcast the
+            // flushes now — after the local body, its AOF and its replication
+            // leg, so the ordering matches the live path — and patch the
+            // result if any leg fails, so a `+OK` for a flush inside a
+            // transaction can be trusted.
+            crate::shard::coordinator::broadcast_txn_flushes(
+                &mut result,
+                &exec_flushes,
+                ctx.shard_id,
+                ctx.num_shards,
+                &ctx.dispatch_tx,
+                &ctx.spsc_notifiers,
+            )
+            .await;
             responses.push(result);
         }
         return true;

src/server/conn/handler_sharded/write.rs

@@ -646,7 +646,21 @@ pub(super) async fn try_handle_multi_exec(
                                 // caller's post-EXEC loop patches placeholders +
                                 // scatters from this (originating) shard.
                                 exec_publishes.extend(r.exec_publishes);
-                                responses.push(r.result);
+                                // c10k E2: the owner flushed only ITS slice.
+                                // Broadcast to the rest from here and patch the
+                                // result on a failed leg (same contract as the
+                                // monoio handler).
+                                let mut routed_result = r.result;
+                                crate::shard::coordinator::broadcast_txn_flushes(
+                                    &mut routed_result,
+                                    &r.exec_flushes,
+                                    s,
+                                    ctx.num_shards,
+                                    &ctx.dispatch_tx,
+                                    &ctx.spsc_notifiers,
+                                )
+                                .await;
+                                responses.push(routed_result);
                             }
                             None => {
                                 responses.push(Frame::Error(Bytes::from_static(
@@ -666,13 +680,16 @@ pub(super) async fn try_handle_multi_exec(
                     _ => {}
                 }
             }
-            let (result, aof_entries, graph_records) = execute_transaction_sharded(
+            // c10k E2: a queued FLUSHDB/FLUSHALL clears only this shard.
+            let mut exec_flushes: Vec<(usize, Frame, usize)> = Vec::new();
+            let (mut result, aof_entries, graph_records) = execute_transaction_sharded(
                 &ctx.shard_databases,
                 ctx.shard_id,
                 &conn.command_queue,
                 conn.selected_db,
                 &ctx.cached_clock,
                 exec_publishes,
+                &mut exec_flushes,
             );
             // task #52: flush the graph-leg wal-v3 records collected by the
             // txn executor. Replication is monoio-only by design (see
@@ -732,6 +749,18 @@ pub(super) async fn try_handle_multi_exec(
                 }
             }
             conn.command_queue.clear();
+            // c10k E2: broadcast the body's flushes to the other shards and
+            // patch the result on a failed leg, so a `+OK` for a flush inside
+            // a transaction can be trusted (same contract as the live path).
+            crate::shard::coordinator::broadcast_txn_flushes(
+                &mut result,
+                &exec_flushes,
+                ctx.shard_id,
+                ctx.num_shards,
+                &ctx.dispatch_tx,
+                &ctx.spsc_notifiers,
+            )
+            .await;
             responses.push(result);
         }
         return true;

src/server/conn/shared.rs

@@ -225,6 +225,7 @@ pub(crate) fn execute_transaction_sharded(
     selected_db: usize,
     cached_clock: &CachedClock,
     exec_publishes: &mut Vec<(usize, Bytes, Bytes)>,
+    exec_flushes: &mut Vec<(usize, Frame, usize)>,
 ) -> (Frame, Vec<(usize, Bytes)>, Vec<(usize, Vec<u8>)>) {
     let db_count = shard_databases.db_count();
 
@@ -403,6 +404,19 @@ pub(crate) fn execute_transaction_sharded(
         if !matches!(response, Frame::Error(_))
             && (cmd.eq_ignore_ascii_case(b"FLUSHDB") || cmd.eq_ignore_ascii_case(b"FLUSHALL"))
         {
+            // c10k E2: this loop clears only the LOCAL slice. The live
+            // (non-MULTI) path fixed the same bug with
+            // `coordinate_flush_broadcast`; a queued flush needs the identical
+            // fan-out or `EXEC` answers +OK having emptied one shard of N.
+            // Recorded rather than performed here for two reasons: this
+            // function is synchronous (the broadcast awaits), and it runs on
+            // the OWNER shard for a routed transaction, where fanning out from
+            // inside the shard's own message loop risks a shard-to-shard wait
+            // cycle. Same deferral contract as `exec_publishes` directly
+            // above: `(result_index, command, db)`, patched by the originator.
+            // `selected` is the per-entry db, so a queued SELECT before the
+            // flush is honoured.
+            exec_flushes.push((results.len(), cmd_frame.clone(), selected));
             crate::shard::slice::with_shard(|s| {
                 crate::shard::spsc_handler::auto_flush_indexes(
                     &mut s.vector_store,

src/shard/coordinator.rs

@@ -1040,6 +1040,54 @@ pub(crate) async fn coordinate_flush_broadcast(
     }
 }
 
+/// Fan out the flushes a MULTI/EXEC body performed locally (c10k E2).
+///
+/// `execute_transaction_sharded` clears only the slice it runs on, so a queued
+/// `FLUSHDB`/`FLUSHALL` empties one shard of N while `EXEC` still answers
+/// `+OK`. Measured at `--shards 4`: 45 of 64 keys survived a transaction that
+/// reported success. The live (non-MULTI) path has broadcast since D-2; this
+/// gives the transactional path the same guarantee.
+///
+/// `exec_shard` is the shard that RAN the body — the owner for a routed
+/// transaction, not necessarily the originator — so it is the one leg already
+/// flushed and correctly skipped.
+///
+/// On a failed leg the corresponding element of the `EXEC` result array is
+/// replaced with the partial-flush error, so a client that sees `+OK` for a
+/// flush inside a transaction can rely on it, exactly as on the live path.
+/// Non-atomic across shards, like every other broadcast here: a concurrent
+/// reader can see shard A flushed before shard B. `MULTI` does not and cannot
+/// change that in a shared-nothing engine — it bounds the report, not the
+/// visibility.
+pub(crate) async fn broadcast_txn_flushes(
+    result: &mut Frame,
+    exec_flushes: &[(usize, Frame, usize)],
+    exec_shard: usize,
+    num_shards: usize,
+    dispatch_tx: &Rc<RefCell<Vec<HeapProd<ShardMessage>>>>,
+    spsc_notifiers: &[Arc<channel::Notify>],
+) {
+    if exec_flushes.is_empty() || num_shards <= 1 {
+        return;
+    }
+    for (result_index, command, db_index) in exec_flushes {
+        if let Err(err) = coordinate_flush_broadcast(
+            command,
+            exec_shard,
+            num_shards,
+            *db_index,
+            dispatch_tx,
+            spsc_notifiers,
+        )
+        .await
+            && let Frame::Array(items) = result
+            && let Some(slot) = items.get_mut(*result_index)
+        {
+            *slot = err;
+        }
+    }
+}
+
 /// Coordinate MGET across shards using VLL pattern.
 ///
 /// Groups keys by shard in a BTreeMap (ascending shard-ID order), executes

src/shard/dispatch.rs

@@ -948,6 +948,15 @@ pub struct TxnExecutePayload {
 pub struct TxnExecReply {
     pub result: crate::protocol::Frame,
     pub exec_publishes: Vec<(usize, Bytes, Bytes)>,
+    /// c10k E2: keyless FLUSHDB/FLUSHALL executed in the body, as
+    /// `(result_index, command, db)`. The owner shard clears only its OWN
+    /// slice, so the ORIGINATOR must broadcast each of these to the remaining
+    /// shards and patch `result[result_index]` with an explicit partial-flush
+    /// error if any leg fails — otherwise EXEC answers +OK having emptied one
+    /// shard of N. Deferred to the originator for the same reason as
+    /// `exec_publishes`: the fan-out awaits, and doing it inside the owner's
+    /// message loop risks a shard-to-shard wait cycle.
+    pub exec_flushes: Vec<(usize, crate::protocol::Frame, usize)>,
     pub wrote: bool,
     /// `true` iff an AOF append could not be enqueued on the owner (bounded
     /// backpressure exhausted) — the originator surfaces `AOF_APPEND_LOST_ERR`

src/shard/spsc_handler.rs

@@ -2722,6 +2722,11 @@ pub(crate) fn handle_shard_message_shared(
                 reply_tx,
             } = *payload;
             let mut exec_publishes: Vec<(usize, bytes::Bytes, bytes::Bytes)> = Vec::new();
+            // c10k E2: a queued FLUSHDB/FLUSHALL clears only THIS shard's
+            // slice. Collect them and hand them back to the originator, which
+            // broadcasts to the other shards (see `TxnExecReply::exec_flushes`
+            // for why the fan-out cannot happen here).
+            let mut exec_flushes: Vec<(usize, crate::protocol::Frame, usize)> = Vec::new();
             let (result, aof_entries, graph_records) =
                 crate::server::conn::shared::execute_transaction_sharded(
                     shard_databases,
@@ -2730,6 +2735,7 @@ pub(crate) fn handle_shard_message_shared(
                     db_index,
                     cached_clock,
                     &mut exec_publishes,
+                    &mut exec_flushes,
                 );
             // task #52: this arm is the CROSS-SHARD EXEC hop (the accepting
             // connection's shard differs from the owner shard, which by
@@ -2783,6 +2789,7 @@ pub(crate) fn handle_shard_message_shared(
             let _ = reply_tx.send(crate::shard::dispatch::TxnExecReply {
                 result,
                 exec_publishes,
+                exec_flushes,
                 wrote,
                 append_lost,
             });

tests/sharded_multi_exec_routing.rs

@@ -59,6 +59,17 @@ impl Moon {
     }
 }
 
+impl Drop for Moon {
+    /// `kill9` only runs on the happy path — a failed assertion unwinds past
+    /// it and strands the server. One such leak ran for four hours before it
+    /// was noticed, holding its port and its data dir the whole time. Killing
+    /// again after `kill9` has already reaped is a harmless no-op error.
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
 /// `durable` selects `appendonly yes` + `appendfsync always` (for the restart
 /// test) vs `appendonly no`. Always `--shards 4`.
 fn moon_args(dir: &std::path::Path, durable: bool) -> Vec<String> {
@@ -366,3 +377,85 @@ fn multi_shard_span_still_rejected() {
     }
     m.kill9();
 }
+
+/// c10k E2: `FLUSHALL` inside MULTI/EXEC must clear EVERY shard, not just the
+/// one the transaction executed on.
+///
+/// The live (non-MULTI) path already broadcasts: a keyless FLUSHDB/FLUSHALL
+/// routed local-only used to clear just its own shard, and `D-2` fixed that
+/// with `coordinate_flush_broadcast`, turning any failed leg into an explicit
+/// partial-flush error rather than a silent `+OK`.
+///
+/// The MULTI/EXEC executor never got the same treatment.
+/// `execute_transaction_sharded` runs the queued body against the LOCAL slice
+/// with no per-key routing and no fan-out, so a queued FLUSHALL clears one
+/// shard of N and `EXEC` still answers `+OK`. At `--shards 4` that leaves
+/// roughly three quarters of the keyspace alive after the client was told the
+/// database was emptied — a silent wrong answer to a destructive command, and
+/// the kind that is only noticed later via a non-zero DBSIZE.
+///
+/// Asserted through DBSIZE rather than per-key GETs so the test fails on ANY
+/// surviving key, not just the ones it thought to name.
+#[test]
+fn flushall_inside_multi_clears_every_shard() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let Some(moon) = spawn_moon_first(dir.path(), false) else {
+        return; // binary missing — skip
+    };
+
+    let mut c = Client::connect(moon.port);
+
+    // Spread keys across all 4 shards. Untagged keys hash-route by full key,
+    // so a spread of distinct names covers every shard with high probability;
+    // 64 makes an all-on-one-shard fluke effectively impossible.
+    for i in 0..64 {
+        let key = format!("e2key:{i}");
+        assert_eq!(
+            c.cmd(&["SET", &key, "v"]),
+            Reply::Simple("OK".into()),
+            "SET {key} must succeed"
+        );
+    }
+    let before = c.cmd(&["DBSIZE"]);
+    assert_eq!(
+        before,
+        Reply::Int(64),
+        "all 64 keys must be visible before the flush (DBSIZE scatter-gathers)"
+    );
+
+    // The whole point: FLUSHALL queued inside a transaction.
+    assert_eq!(c.cmd(&["MULTI"]), Reply::Simple("OK".into()));
+    assert_eq!(
+        c.cmd(&["FLUSHALL"]),
+        Reply::Simple("QUEUED".into()),
+        "FLUSHALL must queue inside MULTI"
+    );
+    let exec = c.cmd(&["EXEC"]);
+
+    // If EXEC reports success it MUST have flushed everything. An explicit
+    // partial-flush error would also be acceptable behaviour (that is what the
+    // live path does on a failed leg) — what is not acceptable is +OK with
+    // keys still present.
+    let claimed_success = matches!(&exec, Reply::Array(v)
+        if v.len() == 1 && matches!(&v[0], Reply::Simple(s) if s == "OK"));
+
+    let after = c.cmd(&["DBSIZE"]);
+    if claimed_success {
+        assert_eq!(
+            after,
+            Reply::Int(0),
+            "EXEC answered +OK for FLUSHALL but keys survive: DBSIZE={after:?}. \
+             A transaction that reports a successful FLUSHALL must have cleared \
+             every shard, not just the one it executed on."
+        );
+    } else {
+        assert!(
+            matches!(&exec, Reply::Array(v) if v.len() == 1 && matches!(&v[0], Reply::Error(_)))
+                || matches!(&exec, Reply::Error(_)),
+            "EXEC must either flush everything or say so explicitly; got {exec:?} \
+             with DBSIZE={after:?}"
+        );
+    }
+
+    moon.kill9();
+}