diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 625cbd1b..19180e08 100644 --- a/.github/workflows/ci.yml +++ b/.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__)' diff --git a/CHANGELOG.md b/CHANGELOG.md index 24cf9907..48d07780 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/scripts/client-compat/redis_py/test_acceptance.py b/scripts/client-compat/redis_py/test_acceptance.py index ca44279a..28edc64d 100644 --- a/scripts/client-compat/redis_py/test_acceptance.py +++ b/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() diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 2c8d1b1c..228e71cb 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/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 diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 7e7b9dfe..8c6a2331 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/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 — diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index eb373293..f7252380 100644 --- a/src/server/conn/shared.rs +++ b/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. diff --git a/tests/pipeline_cross_shard_ordering.rs b/tests/pipeline_cross_shard_ordering.rs new file mode 100644 index 00000000..301448ff --- /dev/null +++ b/tests/pipeline_cross_shard_ordering.rs @@ -0,0 +1,680 @@ +//! A pipeline must execute in order — including the commands that reach +//! across shards. (moon#507) +//! +//! Redis executes a pipelined batch in order, so a command must observe every +//! write that acked earlier in its own batch. Moon's sharded pipeline handlers +//! break that: single-key commands whose key lives on another shard are +//! DEFERRED into `remote_groups` and dispatched as one `PipelineBatchSlotted` +//! at the end of the batch ("Phase 2b"), while multi-key and keyless commands +//! execute INLINE, mid-loop. An inline command therefore runs against a shard +//! whose earlier writes have not even been sent yet. +//! +//! Filed as an `MGET`-returns-nulls bug. Measured, it is wider than that and +//! not only a stale read — the write ORDERING inverts: +//! +//! ```text +//! shards=2, 20 trials each, against the unfixed build +//! SET a, MSET a 7/20 MSET's value LOST — the earlier SET lands after it +//! SET, FLUSHALL 10/20 key survives a flush it was told succeeded +//! SET,SET, DEL 6/20 keys survive a DEL that returned success +//! SET,SET, MGET 10/20 the filed symptom +//! SET, SCAN (full) a page taken against a stale shard skips the +//! key permanently — the cursor never comes back +//! SET, DBSIZE 12/20 also KEYS, RANDOMKEY, EXISTS, UNLINK, +//! COPY, BITOP, INFO keyspace +//! SET, TOUCH k 0/20 single-key: correct, and that is the fix's shape +//! SET, TYPE k 0/20 +//! ``` +//! +//! Rate rises with shard count (a key is remote with probability +//! `1 - 1/shards`), so these run at `--shards 4` where a single trial is wrong +//! ~75% of the time — but they still loop, because ONE trial that happens to +//! land its keys locally proves nothing. Every assertion is written so that the +//! FIXED state is the passing state. +//! +//! Raw sockets, because the whole bug is about what is in one TCP write: a +//! client library free to split the batch would hide it. + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +/// Enough shards that a key is remote ~75% of the time, so a run that got +/// lucky on placement is not mistaken for a fix. +const SHARDS: &str = "4"; +/// Distinct key groups per assertion. At p(remote) = 0.75 the chance that all +/// 12 land locally — and vacuously pass — is under 1e-7. +const TRIALS: usize = 12; + +struct Moon { + child: Child, + port: u16, + tmp_dir: std::path::PathBuf, +} + +impl Drop for Moon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_dir_all(&self.tmp_dir); + } +} + +fn spawn_moon(shards: &str) -> Moon { + let bin = std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")); + let (child, port) = common::spawn_listening(|port| { + let tmp_dir = std::env::temp_dir().join(format!("moon-pipeorder-{port}")); + let _ = std::fs::create_dir_all(&tmp_dir); + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + shards, + "--admin-port", + "0", + "--appendonly", + "no", + "--disk-free-min-pct", + "0", + "--dir", + tmp_dir.to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr( + std::fs::File::create(tmp_dir.join("moon.stderr")).expect("create moon stderr log"), + ) + .spawn() + .expect("spawn moon") + }); + let tmp_dir = std::env::temp_dir().join(format!("moon-pipeorder-{port}")); + let mut moon = Moon { + child, + port, + tmp_dir, + }; + let deadline = Instant::now() + Duration::from_secs(30); + while Instant::now() < deadline { + if let Ok(mut c) = TcpStream::connect(("127.0.0.1", moon.port)) { + let _ = c.set_read_timeout(Some(Duration::from_millis(500))); + if c.write_all(b"*1\r\n$4\r\nPING\r\n").is_ok() { + let mut buf = [0u8; 64]; + if let Ok(n) = c.read(&mut buf) + && n > 0 + && buf.starts_with(b"+PONG") + { + return moon; + } + } + } + std::thread::sleep(Duration::from_millis(100)); + } + let status = match moon.child.try_wait() { + Ok(Some(s)) => format!("exited with {s}"), + Ok(None) => "still running but never answered PING".to_string(), + Err(e) => format!("status unavailable: {e}"), + }; + let log = std::fs::read_to_string(moon.tmp_dir.join("moon.stderr")).unwrap_or_default(); + panic!("moon never became ready on port {port} ({status})\n--- stderr ---\n{log}"); +} + +struct Conn { + sock: TcpStream, + /// Bytes read from the socket but not yet consumed by a reply. A pipelined + /// reply stream arrives in arbitrary chunks, so a read can overshoot the + /// replies asked for; keeping the remainder here stops the next call from + /// mistaking it for its own reply. + spill: Vec, +} + +fn encode(parts: &[&str]) -> Vec { + let mut out = format!("*{}\r\n", parts.len()).into_bytes(); + for p in parts { + out.extend_from_slice(format!("${}\r\n{p}\r\n", p.len()).as_bytes()); + } + out +} + +/// Bytes consumed by exactly `want` complete top-level RESP replies at the +/// start of `buf`, or `None` when `buf` does not hold that many yet. +/// +/// This exists because the obvious harness — "read until the socket goes quiet +/// for 250ms" — silently TRUNCATES a reply whenever the server pauses longer +/// than that mid-stream, and then the test reports a wrong VALUE rather than a +/// short READ. The fix under test makes such pauses more likely, not less: +/// every deferral adds a shard dispatch/await boundary inside a single batch's +/// reply stream. Counting frames removes the timing assumption entirely. +/// +/// `pending` counts array elements still outstanding: an item read while +/// `pending > 0` is an ELEMENT of an array already counted, not a reply of its +/// own. Nested arrays work because their children add to the same counter. +fn framed_len(buf: &[u8], want: usize) -> Option { + let mut i = 0usize; + let mut done = 0usize; + let mut pending = 0usize; + while done < want || pending > 0 { + let tag = *buf.get(i)?; + let end = (i..buf.len().checked_sub(1)?).find(|&j| &buf[j..j + 2] == b"\r\n")?; + let line = std::str::from_utf8(&buf[i + 1..end]).ok()?; + i = end + 2; + + if pending > 0 { + pending -= 1; + } else { + done += 1; + } + + match tag { + // Bulk-ish: a length header followed by that many bytes + CRLF. + // A negative length is a null and carries no payload. + b'$' | b'=' | b'!' => { + let n: i64 = line.parse().ok()?; + if n >= 0 { + i = i.checked_add(n as usize + 2)?; + if buf.len() < i { + return None; + } + } + } + // Aggregates. A map's declared length counts PAIRS. + b'*' | b'~' | b'>' => { + let n: i64 = line.parse().ok()?; + if n > 0 { + pending += n as usize; + } + } + b'%' => { + let n: i64 = line.parse().ok()?; + if n > 0 { + pending += (n as usize) * 2; + } + } + // Single-line: +simple, -error, :int, ,double, #bool, (bignum. + _ => {} + } + } + Some(i) +} + +impl Conn { + fn open(port: u16) -> Self { + let sock = TcpStream::connect(("127.0.0.1", port)).expect("connect"); + sock.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + sock.set_write_timeout(Some(Duration::from_secs(5))) + .unwrap(); + Conn { + sock, + spill: Vec::new(), + } + } + + /// Send several commands as ONE write — the whole point of the test. The + /// server must not be able to tell this from any other batch, and must + /// execute it in order. + fn pipeline(&mut self, cmds: &[&[&str]]) -> String { + let mut out = Vec::new(); + for c in cmds { + out.extend_from_slice(&encode(c)); + } + self.sock.write_all(&out).expect("write"); + self.read_replies(cmds.len()) + } + + fn send(&mut self, parts: &[&str]) -> String { + self.sock.write_all(&encode(parts)).expect("write"); + self.read_replies(1) + } + + /// Read until exactly `want` complete top-level replies have arrived. + /// + /// Panics rather than returning short: a truncated read surfacing as a + /// wrong value is the failure mode that would make this suite lie about + /// which defect it caught. + fn read_replies(&mut self, want: usize) -> String { + let deadline = Instant::now() + Duration::from_secs(20); + let mut chunk = [0u8; 65536]; + loop { + if let Some(n) = framed_len(&self.spill, want) { + let reply = String::from_utf8_lossy(&self.spill[..n]).into_owned(); + self.spill.drain(..n); + return reply; + } + if Instant::now() >= deadline { + panic!( + "timed out waiting for {want} replies; got {} bytes: {:?}", + self.spill.len(), + String::from_utf8_lossy(&self.spill) + ); + } + match self.sock.read(&mut chunk) { + Ok(0) => panic!( + "server closed after {} bytes while {want} replies were expected: {:?}", + self.spill.len(), + String::from_utf8_lossy(&self.spill) + ), + Ok(n) => self.spill.extend_from_slice(&chunk[..n]), + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) => {} + Err(e) => panic!("read failed after {} bytes: {e}", self.spill.len()), + } + } + } +} + +/// Cursor from a `SCAN` reply (`*2\r\n$\r\n\r\n`), or +/// `None` when the reply is not a scan page. +fn scan_cursor(reply: &str) -> Option { + let after = reply.split_once("*2\r\n")?.1; + let after = after.strip_prefix('$')?; + let (_len, rest) = after.split_once("\r\n")?; + let (cursor, _) = rest.split_once("\r\n")?; + Some(cursor.to_string()) +} + +/// Run `body` for each of `TRIALS` distinct key groups on a FRESH connection, +/// collecting the trials that came out wrong. +/// +/// Fresh connections on purpose: a connection's own shard decides whether its +/// keys are local, and reusing one would sample a single placement over and +/// over instead of the distribution. +fn each_trial(port: u16, tag: &str, mut body: impl FnMut(&mut Conn, &str) -> Option) { + let mut wrong: Vec = Vec::new(); + for i in 0..TRIALS { + let mut c = Conn::open(port); + // A hash tag co-locates the group on one shard — the idiom CLAUDE.md + // recommends for exactly this kind of multi-key access, and the one + // the bug report found broken. + let key_tag = format!("{{{tag}{i}}}"); + if let Some(detail) = body(&mut c, &key_tag) { + wrong.push(format!(" trial {i} ({key_tag}): {detail}")); + } + } + assert!( + wrong.is_empty(), + "{}/{} trials violated pipeline ordering — a command in the batch did \ + not observe a write that acked earlier in the SAME batch:\n{}", + wrong.len(), + TRIALS, + wrong.join("\n") + ); +} + +// --------------------------------------------------------------------------- +// The filed symptom +// --------------------------------------------------------------------------- + +/// moon#507 as reported: MGET returns nulls for keys its own batch just wrote. +#[test] +fn pco1_mget_sees_writes_from_its_own_batch() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco1", |c, t| { + let (a, b) = (format!("{t}a"), format!("{t}b")); + c.send(&["DEL", &a, &b]); + let r = c.pipeline(&[&["SET", &a, "1"], &["SET", &b, "2"], &["MGET", &a, &b]]); + // Both SETs must ack, or the test is measuring something else. + assert!( + r.starts_with("+OK\r\n+OK\r\n"), + "the SETs themselves failed: {r:?}" + ); + if r.ends_with("*2\r\n$1\r\n1\r\n$1\r\n2\r\n") { + None + } else { + Some(format!("MGET replied {:?}", &r[8..])) + } + }); +} + +/// The same batch with single-key GETs is CORRECT even unfixed — single-key +/// commands are deferred into the same remote batch and stay in order. Asserted +/// so a "fix" that simply broke batching everywhere is not mistaken for one, +/// and so the contrast that located the bug stays in the suite. +#[test] +fn pco2_single_key_gets_were_always_ordered() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco2", |c, t| { + let (a, b) = (format!("{t}a"), format!("{t}b")); + c.send(&["DEL", &a, &b]); + let r = c.pipeline(&[ + &["SET", &a, "1"], + &["SET", &b, "2"], + &["GET", &a], + &["GET", &b], + ]); + if r == "+OK\r\n+OK\r\n$1\r\n1\r\n$1\r\n2\r\n" { + None + } else { + Some(format!("replied {r:?}")) + } + }); +} + +// --------------------------------------------------------------------------- +// Write-ordering inversions — worse than the filed read bug +// --------------------------------------------------------------------------- + +/// `SET a old` then `MSET a new` in one batch: the LAST write must win. Unfixed, +/// the inline MSET runs first and the deferred SET lands on top of it, so the +/// value the client wrote SECOND is silently lost. +#[test] +fn pco3_later_mset_wins_over_earlier_set() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco3", |c, t| { + let (a, b) = (format!("{t}a"), format!("{t}b")); + c.send(&["DEL", &a, &b]); + c.pipeline(&[&["SET", &a, "old"], &["MSET", &a, "new", &b, "2"]]); + let got = c.send(&["GET", &a]); + if got.starts_with("$3\r\nnew") { + None + } else { + Some(format!( + "GET returned {got:?} — the MSET issued AFTER the SET lost to it" + )) + } + }); +} + +/// `SET` then multi-key `DEL` in one batch: the key must be gone. Unfixed, the +/// inline DEL runs before the deferred SET, returns success, and the key is +/// then (re)created by the SET landing afterwards. +#[test] +fn pco4_multi_key_del_removes_writes_from_its_own_batch() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco4", |c, t| { + let (a, b) = (format!("{t}a"), format!("{t}b")); + c.send(&["DEL", &a, &b]); + c.pipeline(&[&["SET", &a, "1"], &["SET", &b, "2"], &["DEL", &a, &b]]); + let got = c.send(&["MGET", &a, &b]); + if got == "*2\r\n$-1\r\n$-1\r\n" { + None + } else { + Some(format!("keys survived their own batch's DEL: {got:?}")) + } + }); +} + +/// `SET` then `FLUSHALL` in one batch: nothing may survive. Unfixed, the flush +/// runs first and the write lands into the freshly-emptied keyspace. +#[test] +fn pco5_flushall_clears_writes_from_its_own_batch() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco5", |c, t| { + let k = format!("{t}k"); + c.pipeline(&[&["SET", &k, "1"], &["FLUSHALL"]]); + let got = c.send(&["GET", &k]); + if got.starts_with("$-1") { + None + } else { + Some(format!("key survived FLUSHALL: {got:?}")) + } + }); +} + +// --------------------------------------------------------------------------- +// Keyless aggregation commands — the class a name-list would have missed +// --------------------------------------------------------------------------- + +/// DBSIZE, KEYS, SCAN and EXISTS all aggregate across shards inline. None of +/// them is a "multi-key command" in the registry sense, which is why the fix is +/// keyed on ROUTABILITY (`extract_primary_key` returning a key) rather than on +/// a list of command names — a list wrote itself around MGET and would have +/// missed every one of these. +#[test] +fn pco6_cross_shard_aggregations_see_their_own_batch() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco6", |c, t| { + let k = format!("{t}k"); + c.send(&["FLUSHALL"]); + + let r = c.pipeline(&[&["SET", &k, "1"], &["DBSIZE"]]); + if !r.ends_with(":1\r\n") { + return Some(format!("DBSIZE after SET replied {:?}", &r[5..])); + } + + let r = c.pipeline(&[&["SET", &k, "1"], &["KEYS", "*"]]); + if !r.contains(k.as_str()) { + return Some(format!("KEYS after SET replied {:?}", &r[5..])); + } + + // SCAN is cursor-paged, and at shards>1 page 0 legitimately covers + // only part of the keyspace — so "the key is not on page 0" is not a + // defect. The ordering property is that a FULL iteration whose first + // page was taken in the same batch as the write still finds it: a page + // taken against a stale shard would skip the key permanently, because + // the cursor never returns to that shard. + let r = c.pipeline(&[&["SET", &k, "1"], &["SCAN", "0"]]); + let mut found = r.contains(k.as_str()); + let mut cursor = scan_cursor(&r).unwrap_or_else(|| "0".to_string()); + let mut pages = 0; + while cursor != "0" && pages < 64 { + let page = c.send(&["SCAN", &cursor]); + found |= page.contains(k.as_str()); + cursor = match scan_cursor(&page) { + Some(next) => next, + None => break, + }; + pages += 1; + } + if !found { + return Some(format!( + "a full SCAN iteration begun in the write's own batch never \ + returned {k}" + )); + } + + let r = c.pipeline(&[&["SET", &k, "1"], &["EXISTS", &k, "nope"]]); + if !r.ends_with(":1\r\n") { + return Some(format!("EXISTS after SET replied {:?}", &r[5..])); + } + None + }); +} + +/// `INFO keyspace` is neither multi-key nor obviously "cross-shard", and it was +/// wrong. It is here because it is what proves the fix cannot be a curated list +/// of command names — nobody writing a list for an MGET bug would have put INFO +/// on it. `extract_primary_key` reports it keyless, so routability catches it +/// for free. +/// +/// `MEMORY USAGE` was also wrong in the same sweep and is deliberately NOT +/// asserted here: it is a DIFFERENT bug (moon#511). It routes by hashing the +/// literal subcommand `"USAGE"` instead of the key, so it answers `$-1` for a +/// key that plainly exists with no pipelining involved at all — 22/24 at +/// `--shards 4` on separate connections and separate batches. Folding it in +/// would make this suite pass or fail for two unrelated reasons. +#[test] +fn pco7_introspection_commands_see_their_own_batch() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco7", |c, t| { + let k = format!("{t}k"); + c.send(&["FLUSHALL"]); + let r = c.pipeline(&[&["SET", &k, "1"], &["INFO", "keyspace"]]); + if r.contains("keys=0") || !r.contains("db0") { + return Some(format!("INFO keyspace after SET replied {:?}", &r[5..])); + } + None + }); +} + +// --------------------------------------------------------------------------- +// The fix must not break the batch it is protecting +// --------------------------------------------------------------------------- + +/// The fix defers the offending command and the unconsumed tail to the next +/// batch iteration, re-encoding the tail back into the read buffer. That +/// machinery must not drop, duplicate, or reorder anything: a long mixed +/// pipeline has to come back with every reply, in order, exactly once. +#[test] +fn pco8_deferred_tail_is_replayed_intact() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco8", |c, t| { + c.send(&["FLUSHALL"]); + let (a, b) = (format!("{t}a"), format!("{t}b")); + + // Interleave the two classes so the batch is split repeatedly: routed + // single-key writes (deferred) against keyless/multi-key commands + // (inline). + let reply = c.pipeline(&[ + &["SET", &a, "1"], + &["DBSIZE"], + &["SET", &b, "2"], + &["MGET", &a, &b], + &["PING"], + &["GET", &a], + &["EXISTS", &a, &b], + &["ECHO", "tail"], + ]); + + let expected = concat!( + "+OK\r\n", // SET a + ":1\r\n", // DBSIZE — sees a + "+OK\r\n", // SET b + "*2\r\n$1\r\n1\r\n$1\r\n2\r\n", // MGET — sees both + "+PONG\r\n", // PING + "$1\r\n1\r\n", // GET a + ":2\r\n", // EXISTS a b + "$4\r\ntail\r\n", // ECHO + ); + if reply == expected { + None + } else { + Some(format!("replied {reply:?}, expected {expected:?}")) + } + }); +} + +/// Inline-protocol commands round-trip through the same defer-and-replay path. +/// The tail is re-encoded with `serialize_resp3`, so a frame that did not +/// arrive as a RESP array is the case most likely to be mangled by it. +#[test] +fn pco9_inline_commands_survive_the_defer_path() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco9", |c, t| { + let k = format!("{t}k"); + c.send(&["DEL", &k]); + // SET (routed, possibly deferred) then an INLINE ping in the SAME write. + let batch = format!( + "*3\r\n$3\r\nSET\r\n${}\r\n{k}\r\n$1\r\n1\r\nPING\r\n", + k.len() + ); + c.sock.write_all(batch.as_bytes()).expect("write"); + let reply = c.read_replies(2); + if reply != "+OK\r\n+PONG\r\n" { + return Some(format!( + "inline PING after a deferred write replied {reply:?}" + )); + } + let got = c.send(&["GET", &k]); + if got == "$1\r\n1\r\n" { + None + } else { + Some(format!("GET after the inline batch replied {got:?}")) + } + }); +} + +// --------------------------------------------------------------------------- +// The class routability alone cannot see +// --------------------------------------------------------------------------- + +/// Some commands are handled INLINE by a `try_handle_*` interceptor before the +/// routing step ever runs, yet have a first argument that `extract_primary_key` +/// would happily hash — so "does it route by its own key?" answers *yes* for +/// them and answers it wrongly. `SWAPDB` (args[0] is a db number) and `EVAL` +/// (args[0] is the script body) are the two that touch real keys. +/// +/// This is the guard for the `is_inline_intercepted` list in +/// `server::conn::shared`: delete the SWAPDB entry and this test fails. Without +/// it the list could rot silently the next time an interceptor is added, which +/// is exactly how moon#507 reached a release. +/// +/// **EVAL is deliberately not asserted here.** It should be — it is the more +/// interesting of the two — but a single-key `EVAL` is currently rejected +/// outright at `--shards >= 2` with `CROSSSLOT Keys in script don't hash to the +/// same slot and shard` (moon#508; measured here at 7/12 trials, and it affects +/// plain `EVAL`, not only the `EVALSHA` form the issue was filed against). A +/// probe that cannot run its command proves nothing about ordering. Restore the +/// EVAL leg when #508 closes. +#[test] +fn pco10_inline_intercepted_commands_see_their_own_batch() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "pco10", |c, t| { + let k = format!("{t}k"); + // SWAPDB moves the whole database out from under a pending write. The + // write acked in db 0, so after the swap db 0 must NOT hold it and db 1 + // must — which can only happen if the write landed BEFORE the swap. + c.send(&["SELECT", "0"]); + c.send(&["FLUSHALL"]); + c.pipeline(&[&["SET", &k, "1"], &["SWAPDB", "0", "1"]]); + let in_db0 = c.send(&["GET", &k]); + c.send(&["SELECT", "1"]); + let in_db1 = c.send(&["GET", &k]); + c.send(&["SELECT", "0"]); + if !in_db0.starts_with("$-1") || !in_db1.starts_with("$1\r\n1") { + return Some(format!( + "after SET+SWAPDB in one batch: db0={in_db0:?} db1={in_db1:?} \ + — the write did not land before the swap" + )); + } + None + }); +} + +/// The reply framer is the one piece of this suite that can make every other +/// case lie — a short read reports a wrong VALUE, not a short read. So it is +/// tested directly, including the boundary where a reply is one byte short. +#[test] +fn pco0_reply_framer_counts_top_level_replies() { + let cases: &[(&str, usize)] = &[ + ("+OK\r\n", 1), + (":42\r\n", 1), + ("-ERR nope\r\n", 1), + ("$3\r\nabc\r\n", 1), + ("$-1\r\n", 1), + ("$0\r\n\r\n", 1), + ("*2\r\n$1\r\na\r\n$1\r\nb\r\n", 1), + ("*-1\r\n", 1), + ("*0\r\n", 1), + // an MGET whose elements are all null — the shape the BUG produced + ("*2\r\n$-1\r\n$-1\r\n", 1), + // nested: SCAN's [cursor, [keys...]] + ("*2\r\n$1\r\n0\r\n*2\r\n$1\r\na\r\n$1\r\nb\r\n", 1), + // a whole pipelined batch: +OK +OK *2 + ("+OK\r\n+OK\r\n*2\r\n$1\r\n1\r\n$1\r\n2\r\n", 3), + ]; + + for (raw, want) in cases { + let bytes = raw.as_bytes(); + + assert_eq!( + framed_len(bytes, *want), + Some(bytes.len()), + "did not frame {raw:?} as {want} complete repl(y|ies)" + ); + + // Every proper prefix must be judged INCOMPLETE. This is the property + // that matters: the old silence-based reader accepted any prefix that + // happened to arrive before a 250ms lull. + for cut in 1..bytes.len() { + assert_eq!( + framed_len(&bytes[..cut], *want), + None, + "accepted a {cut}-byte prefix of {raw:?} as {want} complete repl(y|ies)" + ); + } + + // Trailing bytes belong to the NEXT reply and must not be consumed. + let mut over = bytes.to_vec(); + over.extend_from_slice(b"+NEXT\r\n"); + assert_eq!(framed_len(&over, *want), Some(bytes.len())); + } + + // Asking for more replies than the buffer holds never over-reports. + assert_eq!(framed_len(b"+OK\r\n", 2), None); + assert_eq!(framed_len(b"", 1), None); +}