fix(server): pipeline ordering at --shards >= 2 — inline commands ran ahead of the batch's pending remote writes (#507)

.github/workflows/ci.yml

@@ -486,9 +486,11 @@ jobs:
       # and python3.14-venv is absent. Verified on the runner.
       #
       # Runs at --shards 2 deliberately: single-shard hides the multi-shard
-      # defects this suite found (#507, #508), which are pinned inside it as
-      # expected failures so a fix breaks the run loudly instead of leaving a
-      # stale skip.
+      # defects this suite found (#507, #508). Each is pinned inside the suite
+      # so a fix breaks the run loudly instead of leaving a stale skip — which
+      # is exactly what happened to #507: its pin started failing once the
+      # ordering fix landed and is now a direct assertion (rp7b). #508
+      # (single-key EVAL/EVALSHA rejected CROSSSLOT) is still pinned as a gap.
       - name: redis-py acceptance suite (unmodified client, live server)
         run: |
           python3 -c 'import redis; print("redis-py", redis.__version__)'

CHANGELOG.md

@@ -125,6 +125,67 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   dispatcher — so `COMMAND COUNT` was advertising verbs Moon could not run.
 
 ### Fixed
+- **A pipeline did not execute in order at `--shards >= 2`, and writes were silently lost.**
+  Reported as "MGET in the same pipeline as its SETs returns nulls" (#507). The cause is wider than
+  the symptom: the sharded pipeline handlers DEFER a single-key command whose key lives on another
+  shard into `remote_groups`, dispatching the whole group as one `PipelineBatchSlotted` at the end
+  of the batch — while multi-key and keyless commands execute INLINE, mid-loop. An inline command
+  therefore ran against a shard whose earlier writes in the same batch had not been sent yet.
+
+  Measured at `--shards 2`, 20 trials each, before the fix:
+
+  | in one pipeline batch | wrong | consequence |
+  |---|---|---|
+  | `SET a`, `MSET a` | 7/20 | **the MSET's value is lost** — the earlier SET lands on top of it |
+  | `SET`, `FLUSHALL` | 10/20 | the key survives a flush that returned `+OK` |
+  | `SET`,`SET`, `DEL` | 6/20 | the keys survive a `DEL` that returned success |
+  | `SET`,`SET`, `MGET` | 10/20 | the reported symptom |
+  | `SET`, `DBSIZE` | 12/20 | also `KEYS`, `RANDOMKEY`, `EXISTS`, `UNLINK`, `COPY`, `BITOP`, `INFO keyspace` |
+  | `SET`, `TOUCH k` | 0/20 | single-key: always correct, and the shape of the fix |
+
+  So this was not only the stale read it was filed as — same-key write ordering inverted, which is
+  silent data loss. The rate rises with shard count (a key is remote with probability
+  `1 - 1/shards`).
+
+  The fix defers such a command and the unconsumed batch tail to the next iteration, reusing the
+  mechanism #438 already built for early-flush commands: phase 2 resolves the pending remote replies
+  first, and the tail re-parses with `remote_groups` empty, so it cannot loop. Applied to both
+  sharded handlers (`handler_monoio` and `handler_sharded`); `handler_single` has no deferral and
+  was never affected.
+
+  The predicate is keyed on ROUTABILITY, not on a list of command names: a command routed by its own
+  single key needs no wait, because a key maps to exactly one shard — if that shard is local the key
+  cannot be pending, and if it is remote the command is appended behind the pending ones and the
+  slotted batch preserves order. Everything else waits. A name list written for an MGET bug would
+  not have contained `INFO keyspace` or `RANDOMKEY`, both of which were wrong. The one case
+  routability cannot see — commands intercepted inline BEFORE routing, which still have a key-shaped
+  first argument (`EVAL`, `SWAPDB`, …) — is named explicitly and carries a test that fails if an
+  entry is dropped.
+
+  Deferring is the conservative direction: a command sent down this path unnecessarily is merely
+  executed at the start of the next batch, which is always correct.
+
+  **This costs throughput, and the cost is per interleaving rather than per pipeline** — each
+  deferral is one extra shard dispatch/await boundary (~50µs). Measured at `--shards 2` on one
+  connection, 9 reps, alternating leg order, median; "floor" is the worst within-leg spread, so a
+  delta smaller than its floor resolved nothing:
+
+  | pipeline shape | guard fires | before | after | delta | floor |
+  |---|---|---|---|---|---|
+  | `MGET` after every 2 `SET`s | 64×/flush | 125,885 | 59,889 | **−52.4%** | 5.8% |
+  | 128 `SET`s, then one `MGET` | 1×/flush | 528,764 | 512,686 | −3.0% | 22.2% |
+  | `SET`,`SET`,`GET` (guard never fires) | never | 873,526 | 867,715 | −0.7% | 33.9% |
+
+  A multi-key or keyless command at the END of a pipeline — the shape #507 was filed from, and the
+  shape `redis-py`'s `pipeline()` produces — costs nothing measurable. One interleaved after every
+  pair of writes halves throughput. `redis-benchmark` cannot express any of these shapes (it sends a
+  single command type, so the guard never fires), so this came from a purpose-built harness that
+  refuses to report unless the pre-fix binary actually reproduces the bug first.
+
+  Recovering that cost means letting multi-key commands participate in the slotted batch instead of
+  executing inline — a cross-shard-coordinator change well outside a correctness fix, filed
+  separately.
+
 - **Writes were paying for a memory measurement on every SET.** The `maxmemory` real-footprint
   correction (#478) was computed inside `evict_to_budget`, which runs on the write path, so every
   write performed `open`/`read`/`close` on `/proc/self/statm` *and* an instance-wide accounting sum

scripts/client-compat/redis_py/test_acceptance.py

@@ -239,25 +239,20 @@ def test_rp7_pipeline_without_transaction(self):
         self.assertEqual(out[:2], [True, True])
         self.assertEqual(out[-2:], ["1", "2"])
 
-    def test_rp7b_mget_in_a_pipeline_is_a_known_gap(self):
-        """KNOWN GAP (moon#507), amplified so it cannot pass by luck.
-
-        At `--shards >= 2`, an MGET in the same batch as the SETs that wrote
-        its keys returns nulls — though those SETs acked `+OK` earlier in that
-        same batch and the values are readable the instant the batch ends.
-        Redis executes a pipeline in order, so this is a silent
-        read-your-own-writes violation.
-
-        It fires for roughly HALF of all key groups: whether it happens is
-        decided by which shard owns the keys relative to the connection's own
-        shard. A single trial is therefore a coin flip, which is exactly how an
-        earlier `expectedFailure` version of this test made CI flaky. Twenty
-        independent key groups drop the odds of a spurious pass to ~1e-6, and
-        the assertion is written so that ZERO failures — the state after a fix —
-        breaks the run.
+    def test_rp7b_mget_in_a_pipeline_sees_its_own_batch(self):
+        """moon#507, fixed: MGET must observe the SETs from its own batch.
+
+        This was a KNOWN GAP pinned as an inverted probe (assert that at least
+        one trial is broken). It is now a direct assertion, which is the shape
+        the probe itself asked for when it started failing.
+
+        Still twenty independent key groups rather than one, and for the same
+        reason the probe needed them: whether a group is affected depends on
+        which shard owns its keys relative to the connection's own shard, so a
+        single trial only samples one placement. Twenty makes a regression that
+        reaches even half of placements essentially certain to be caught.
         """
         c = self.client()
-        broken = []
         for i in range(20):
             a, b = f"{{rp7b{i}}}a", f"{{rp7b{i}}}b"
             c.delete(a, b)
@@ -267,19 +262,13 @@ def test_rp7b_mget_in_a_pipeline_is_a_known_gap(self):
                 p.mget(a, b)
                 out = p.execute()
             self.assertEqual(out[:2], [True, True], "the SETs themselves failed")
-            if out[-1] != ["1", "2"]:
-                broken.append((a, out[-1]))
             self.assertEqual(
-                c.mget(a, b), ["1", "2"],
-                f"{a}/{b} are wrong even AFTER the batch — #507 is a visibility "
-                f"bug, not a durability one; this is a different, worse defect",
+                out[-1], ["1", "2"],
+                f"MGET of {a}/{b} did not see the SETs that acked earlier in "
+                f"its OWN pipeline batch — read-your-own-writes violated "
+                f"(moon#507 regressed)",
             )
-        self.assertTrue(
-            broken,
-            "all 20 pipelined MGETs returned the values written in their own "
-            "batch — moon#507 is fixed. Delete this probe and assert the "
-            "correct behaviour directly in rp7.",
-        )
+            self.assertEqual(c.mget(a, b), ["1", "2"])
 
     def test_rp8_multi_exec_transaction(self):
         c = self.client()

src/server/conn/handler_monoio/mod.rs

@@ -1631,6 +1631,23 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 deferred_tail_from = Some(frame_idx - 1);
                 break;
             }
+
+            // #507 pipeline ordering: a command that does NOT route by its own
+            // single key executes inline, against shards whose earlier writes
+            // in this same batch are still sitting in `remote_groups`
+            // undispatched. Reading there returns state the client already
+            // wrote; writing there is overwritten when the pending write lands.
+            // Defer it and the unconsumed tail exactly as #438 does above —
+            // phase 2 resolves the pending replies first, and the tail
+            // re-parses at the top of the next batch with `remote_groups`
+            // empty, so this cannot loop.
+            if !remote_groups.is_empty()
+                && crate::server::conn::shared::must_wait_for_pending_remote(cmd, cmd_args)
+            {
+                frames[frame_idx - 1] = frame;
+                deferred_tail_from = Some(frame_idx - 1);
+                break;
+            }
             // --- Connection-level commands (dispatched to dispatch.rs) ---
             //
             // Length-gated dispatch: each `try_handle_*` starts with a

src/server/conn/handler_sharded/mod.rs

@@ -777,6 +777,24 @@ pub(crate) async fn handle_connection_sharded_inner<
                         break;
                     }
 
+                    // #507 pipeline ordering: a command that does NOT route by
+                    // its own single key executes inline, against shards whose
+                    // earlier writes in this same batch are still sitting in
+                    // `remote_groups` undispatched. Reading there returns state
+                    // the client already wrote; writing there is overwritten
+                    // when the pending write lands. Defer it and the unconsumed
+                    // tail exactly as #438 does above — phase 2 resolves the
+                    // pending replies first, and the tail re-parses at the top
+                    // of the next batch with `remote_groups` empty, so this
+                    // cannot loop.
+                    if !remote_groups.is_empty()
+                        && crate::server::conn::shared::must_wait_for_pending_remote(cmd, cmd_args)
+                    {
+                        batch[frame_idx - 1] = frame;
+                        deferred_tail_from = Some(frame_idx - 1);
+                        break;
+                    }
+
                     // MONITOR feed for the two ACL-EXEMPT commands below.
                     // AUTH and HELLO are intercepted above the ACL gate, hence
                     // above the main feed hook, so they would never be fed —

src/server/conn/shared.rs

@@ -1010,6 +1010,102 @@ pub(crate) fn extract_primary_key<'a>(cmd: &[u8], args: &'a [Frame]) -> Option<&
     }
 }
 
+/// Must this command wait for the batch's already-deferred remote commands to
+/// land before it may execute? (moon#507)
+///
+/// The sharded pipeline handlers DEFER a single-key command whose key lives on
+/// another shard into `remote_groups`, dispatching the whole group as one
+/// `PipelineBatchSlotted` at the end of the batch. Anything that executes
+/// INLINE in the meantime runs against a shard whose earlier writes in the same
+/// batch have not been sent yet — so it reads state the client already wrote,
+/// or writes state the pending command then overwrites. Measured at
+/// `--shards 2`: `SET a` + `MSET a` in one batch lost the MSET's value in 7 of
+/// 20 trials, and `SET` + `FLUSHALL` left the key alive in 10 of 20.
+///
+/// The safe case is narrow and worth stating positively, because it is what
+/// makes the rest of the pipeline fast: a command routed by its OWN single key
+/// needs no wait. A key maps to exactly one shard, so if that shard is local
+/// the key cannot be in `remote_groups` at all, and if it is remote the command
+/// is appended BEHIND the pending ones on the same target and the slotted batch
+/// preserves order. That is why `SET k` + `GET k` and `SET k` + `TYPE k` were
+/// always correct while `MGET` was not.
+///
+/// Everything else waits. Three ways a command fails to be "routed by its own
+/// single key":
+///
+/// 1. it is multi-key (MGET/MSET/DEL/EXISTS/…) — consumed by the cross-shard
+///    coordinator before routing;
+/// 2. it is keyless (`extract_primary_key` → `None`) — DBSIZE, KEYS, SCAN,
+///    RANDOMKEY, FLUSHALL, INFO … all aggregate across shards inline. None of
+///    these is a "multi-key command" in the registry sense, which is why this
+///    predicate is keyed on ROUTABILITY rather than on a list of command names;
+/// 3. it is intercepted inline by a `try_handle_*` handler BEFORE routing runs,
+///    even though it does have an args[0] that `extract_primary_key` would
+///    happily hash. That is the one case routability cannot see, so those
+///    families are named in [`is_inline_intercepted`].
+///
+/// Deferring is conservative: a command wrongly sent down this path is merely
+/// executed at the start of the next batch, which is always correct and costs
+/// one batch boundary. Wrongly calling something SAFE is the direction that
+/// corrupts data, so when in doubt, add it to the wait set.
+pub(crate) fn must_wait_for_pending_remote(cmd: &[u8], args: &[Frame]) -> bool {
+    is_multi_key_command(cmd, args)
+        || is_inline_intercepted(cmd)
+        || extract_primary_key(cmd, args).is_none()
+}
+
+/// Commands handled INLINE by a `try_handle_*` interceptor before the routing
+/// step, and which `extract_primary_key` would nonetheless answer for.
+///
+/// Derived by reading the interceptor chain in `handler_monoio::dispatch` /
+/// `handler_sharded`, not guessed: every other interceptor there guards a
+/// command that `extract_primary_key` already reports keyless (AUTH, HELLO,
+/// CLUSTER, CONFIG, CLIENT, INFO, WAIT, SELECT, KEYS, SCAN, DBSIZE, HOTKEYS,
+/// the persistence verbs …), so those are caught by the keyless arm.
+///
+/// **Adding a new inline interceptor means adding its command here.** A new
+/// interceptor for a command with a key-shaped first argument would silently
+/// re-open moon#507 for that command.
+/// `pco10_inline_intercepted_commands_see_their_own_batch` in
+/// `tests/pipeline_cross_shard_ordering.rs` drives EVAL and SWAPDB — the two
+/// entries that touch real keys — and fails if either is dropped from this
+/// list. It cannot prove the list is COMPLETE against a future interceptor;
+/// that is why the doc above says to err toward waiting.
+fn is_inline_intercepted(cmd: &[u8]) -> bool {
+    // Dotted families first, and deliberately so: a length-keyed match below
+    // would swallow `FT.ALIAS` (8 bytes, 'f') into the FCALL_RO/FUNCTION arm
+    // and answer false for it.
+    const DOTTED: [&[u8]; 4] = [b"FT.", b"GRAPH.", b"CDC.", b"TS."];
+    if DOTTED
+        .iter()
+        .any(|p| cmd.len() > p.len() && cmd[..p.len()].eq_ignore_ascii_case(p))
+    {
+        return true;
+    }
+    let len = cmd.len();
+    if len == 0 {
+        return false;
+    }
+    let b0 = cmd[0] | 0x20;
+    match (len, b0) {
+        // Lua and functions read and write real keys through the interceptor,
+        // never through routing.
+        (4, b'e') => cmd.eq_ignore_ascii_case(b"EVAL"),
+        (7, b'e') => cmd.eq_ignore_ascii_case(b"EVALSHA"),
+        (5, b'f') => cmd.eq_ignore_ascii_case(b"FCALL"),
+        (8, b'f') => cmd.eq_ignore_ascii_case(b"FCALL_RO") || cmd.eq_ignore_ascii_case(b"FUNCTION"),
+        // SWAPDB exchanges whole databases across every shard.
+        // SCRIPT/ACL touch no keyspace data, but they are inline and cost
+        // nothing to serialise behind pending writes.
+        (6, b's') => cmd.eq_ignore_ascii_case(b"SCRIPT") || cmd.eq_ignore_ascii_case(b"SWAPDB"),
+        (3, b'a') => cmd.eq_ignore_ascii_case(b"ACL"),
+        // Container commands for the message-queue and workspace stores.
+        (2, b'm') => cmd.eq_ignore_ascii_case(b"MQ"),
+        (2, b'w') => cmd.eq_ignore_ascii_case(b"WS"),
+        _ => false,
+    }
+}
+
 /// Check if a command is a multi-key command requiring VLL coordination.
 ///
 /// These commands operate on multiple keys that may live on different shards.