From b14c8bb9da453ff026f9e01827bf5420a19aeb77 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 16 Aug 2026 23:24:41 +0700 Subject: [PATCH 1/2] fix(scripting): route a script to the shard owning its keys instead of refusing it (#508) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Filed as "EVALSHA of a single-key script fails with CROSSSLOT", with the guess that a 1-element key list was being mis-folded. It is not that. `validate_keys_same_shard` required every key to hash to the shard the CONNECTION happened to occupy. One key cannot cross slots — it just lives somewhere else — so `CROSSSLOT` was standing in for "I cannot run this HERE", and nothing ever asked where it COULD run. Measured at --shards 4, 8 distinct keys on fresh connections: 7/8 CROSSSLOT 1/8 ran (the key that happened to land on the connection's own shard) numkeys=0 always ran, which is why the defect reads as intermittent That breaks redis.lock.Lock.release() — a single-key EVALSHA, and one of the most-used constructs in redis-py. Through the Script.__call__ wrapper the caller sees a NoScriptError and then a cross-slot error, neither of which names the cause. Fix: a script now routes to the shard owning its keys. scripting::route_script_keys -> Local | Remote(shard) | CrossShard coordinator::coordinate_script -> ShardMessage::Execute to that shard spsc_handler -> EVAL/EVALSHA arm on the Execute path A script executes against ONE shard's database, so the correct place to run it is where that data already is. `cmd_dispatch` has no scripting arm (scripts are intercepted at the connection layer), which is why the shard side needs its own. Keys that GENUINELY span shards are still refused — there is no single target for them — and that refusal is now held twice over: the routing decision rejects it before the hop, and validate_keys_same_shard stays as the shard-side backstop, so both must agree before a script could read another shard's (empty) view of a key. Proved by mutation: removing either layer alone keeps skr5 green; removing BOTH turns it red. Lua VMs stay lazy and per-shard, and the runtime shares the same slot conn_accept fills, so a shard first reached by ROUTING (no connection of its own ever landed there) still ends up with exactly one VM. VM creation failure returns an error frame rather than panicking — this now runs on the shard thread, where a panic aborts the process. Both handlers call one shared helper (`shared::route_script_elsewhere`) rather than carrying two routing policies that can drift. Tests: tests/script_key_routing.rs, 5 cases at --shards 4 looping 12 key placements each (a single trial passes by luck ~25% of the time). Covers single-key EVAL, single-key EVALSHA (the reported shape), a script WRITE read back through the normal path (so routing cannot be faked by answering reads from the wrong shard), keyless scripts, and the surviving cross-shard refusal. Green on both runtimes; lib 4646/4646; clippy and fmt clean. The redis-py acceptance pin (rp14b) is converted from an inverted "known gap" probe to a direct assertion, and rp14 is restored to a full acquire+release lock test — the half that was removed while this was open. 19/19 green against the fix binary. One thing this CHANGES rather than fixes, stated plainly: EVAL caches its script only on the shard that ran it, and EVAL (unlike SCRIPT LOAD) never fans out. So a bare EVAL followed by a direct EVALSHA on a key belonging to another shard now answers NOSCRIPT where it previously answered CROSSSLOT. Measured AFTER this fix, --shards 4: one bare EVAL, then EVALSHA of that sha across 12 other keys -> 4 ok, 8 NOSCRIPT. (The pre-fix leg of that comparison was not measured: before routing, the initial EVAL itself was rejected for ~75% of keys, so the scenario did not arise in the same form.) Neither ever worked; NOSCRIPT is the strictly better failure because every client library retries it (redis-py's Script.__call__ re-EVALs and self-heals) whereas CROSSSLOT was unrecoverable. Anything built on register_script/SCRIPT LOAD — including redis.lock.Lock — is unaffected: 12/12. Filed as #515 with a fix sketch; it needs its own review for the per-EVAL sha1 cost and the shutdown-token plumbing. FCALL has the SAME defect plus a second, independent one: FUNCTION LOAD never fans out to other shards (measured 5/8 CROSSSLOT, 3/8 "Function not found", 0/8 succeeded). Routing FCALL without the fan-out would only trade one error for the other at the same rate, so it is filed as #514 rather than half-fixed here. The reply framer added for #507 moved to tests/common so both suites share one implementation. Closes #508 Refs #507, #514, #515 author: Tin Dang --- CHANGELOG.md | 38 +++ .../client-compat/redis_py/test_acceptance.py | 53 ++-- src/scripting/mod.rs | 106 ++++++++ src/server/conn/handler_monoio/dispatch.rs | 18 +- src/server/conn/handler_monoio/mod.rs | 5 +- src/server/conn/handler_sharded/mod.rs | 14 + src/server/conn/shared.rs | 54 ++++ src/shard/coordinator.rs | 48 ++++ src/shard/event_loop.rs | 26 +- src/shard/mod.rs | 2 + src/shard/spsc_handler.rs | 65 +++++ tests/common/mod.rs | 161 +++++++++++ tests/pipeline_cross_shard_ordering.rs | 148 +--------- tests/script_key_routing.rs | 253 ++++++++++++++++++ 14 files changed, 815 insertions(+), 176 deletions(-) create mode 100644 tests/script_key_routing.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 48d077804..88268f7d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -125,6 +125,44 @@ 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 script whose keys all lived on one shard was refused instead of routed (`CROSSSLOT`).** + Reported as "EVALSHA of a single-key script fails with `CROSSSLOT`" (#508), with the guess that a + 1-element key list was being mis-folded. It was not. `validate_keys_same_shard` required every key + to hash to the shard the CONNECTION happened to occupy — so `CROSSSLOT` was standing in for "I + cannot run this *here*", and nothing ever asked where it *could* run. One key cannot cross slots; + it just lives somewhere else. Measured at `--shards 4`: **7 of 8** single-key `EVAL`s rejected, + only the key that happened to land on the connection's own shard running. `numkeys=0` always + worked, which is why the defect read as intermittent rather than total. + + This broke `redis.lock.Lock.release()` — a single-key `EVALSHA`, and one of the most-used + constructs in `redis-py`. Through the `Script.__call__` wrapper the caller saw a `NoScriptError` + followed by a cross-slot error, neither of which names the cause. + + A script now ROUTES to the shard owning its keys (`scripting::route_script_keys` → + `coordinator::coordinate_script` → a shard-side `EVAL`/`EVALSHA` arm on the SPSC `Execute` path), + because a script executes against one shard's database and the correct place to run it is where + that data is. Keys that GENUINELY span shards are still refused — there is no single target for + them — and that refusal is now doubly held: the routing decision rejects it before the hop, and + `validate_keys_same_shard` remains as the shard-side backstop, so the two must agree before a + script can read another shard's (empty) view of a key. Shards build their Lua VM lazily on first + use and share the one slot `conn_accept` fills, so a shard reached only by routing still gets + exactly one VM. + + One thing this changes rather than fixes: `EVAL` caches its script only on the shard that ran it + and, unlike `SCRIPT LOAD`, never fans out — so a bare `EVAL` followed by a direct `EVALSHA` on a + key owned by another shard now answers `NOSCRIPT` where it previously answered `CROSSSLOT`. + Measured after the fix at `--shards 4`: one bare `EVAL`, then `EVALSHA` of that sha across 12 + other keys — 4 ok, 8 `NOSCRIPT`. Neither ever worked, and + `NOSCRIPT` is the better failure because every client library retries it — `redis-py`'s + `Script.__call__` re-issues `EVAL` and self-heals, whereas `CROSSSLOT` was unrecoverable. Anything + built on `register_script`/`SCRIPT LOAD`, including `redis.lock.Lock`, is unaffected (12/12). + Filed as #515. + + Applied to both handlers through one shared helper rather than two implementations that can drift. + `FCALL` has the same defect *plus* a second one — `FUNCTION LOAD` never fans out to other shards, + so routing alone would trade `CROSSSLOT` for `ERR Function not found` at the same rate. Filed as + #514 rather than half-fixed here. + - **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 diff --git a/scripts/client-compat/redis_py/test_acceptance.py b/scripts/client-compat/redis_py/test_acceptance.py index 28edc64d1..330413476 100644 --- a/scripts/client-compat/redis_py/test_acceptance.py +++ b/scripts/client-compat/redis_py/test_acceptance.py @@ -364,11 +364,12 @@ def _await_message(ps, timeout: float = 5.0): # -- library-level constructs ----------------------------------------- - def test_rp14_lock_is_exclusive(self): - """`redis.lock.Lock` acquisition is SET NX PX — a held lock is exclusive. + def test_rp14_lock_acquires_and_releases(self): + """`redis.lock.Lock` end to end — the most-used construct in redis-py. - Release is covered separately: it goes through EVALSHA, which has its - own known gap (moon#508, see below). + Acquisition is SET NX PX; RELEASE is a single-key EVALSHA, which is why + this test was split in half while moon#508 was open. Both halves are + asserted again now that a script routes to the shard owning its key. """ c = self.client() lock = c.lock("rp14", timeout=5, blocking_timeout=2) @@ -377,36 +378,40 @@ def test_rp14_lock_is_exclusive(self): c.lock("rp14", timeout=5, blocking_timeout=0.2).acquire(), "a held lock was acquired twice — SET NX is not exclusive", ) + # The half that moon#508 broke: release() is EVALSHA of a one-key + # script, so it failed for whichever lock names did not hash to the + # connection's own shard. + lock.release() + self.assertTrue( + c.lock("rp14", timeout=5, blocking_timeout=2).acquire(), + "the lock could not be re-acquired after release()", + ) - def test_rp14b_single_key_evalsha_is_a_known_gap(self): - """KNOWN GAP (moon#508), amplified for the same reason as rp7b. + def test_rp14b_single_key_evalsha_runs_on_the_keys_shard(self): + """moon#508, fixed: EVALSHA of a one-key script must run. - At `--shards >= 2`, EVALSHA of a script declaring ONE key is rejected - with `CROSSSLOT Keys in script don't hash to the same slot and shard`. - One key cannot cross slots. `SCRIPT EXISTS` returns `[True]`, so it is - the key-slot check and not a cache miss. + Was an inverted probe (assert at least one of twenty is rejected). It + is a direct assertion now, which is what the probe asked for when it + started failing. - This is what breaks `redis.lock.Lock.release()`, which is a single-key - EVALSHA — so the most-used construct in redis-py fails for about half - of all lock names. Like moon#507 it fires per-KEY (~50% at two shards), - which is why this probe runs twenty distinct keys instead of one. + Still twenty distinct keys rather than one, and for the probe's own + reason: the defect fired per-KEY, decided by which shard owned the key + relative to the connection's, so a single trial only samples one + placement. """ c = self.client() sha = c.script_load("return redis.call('get', KEYS[1])") self.assertEqual(c.script_exists(sha), [True], "the script did not cache") - rejected = [] for i in range(20): key = f"rp14b:{i}" c.set(key, "v") - try: - self.assertEqual(c.evalsha(sha, 1, key), "v") - except redis.exceptions.RedisError as e: - rejected.append((key, str(e))) - self.assertTrue( - rejected, - "all 20 single-key EVALSHA calls succeeded — moon#508 is fixed. " - "Delete this probe and restore the release half of rp14.", - ) + self.assertEqual( + c.evalsha(sha, 1, key), + "v", + f"single-key EVALSHA on {key} did not run — a script whose one " + f"key lives on another shard must be ROUTED there (moon#508 " + f"regressed)", + ) def test_rp15_from_url_and_pool_reuse(self): """`from_url` is how most applications construct a client at all.""" diff --git a/src/scripting/mod.rs b/src/scripting/mod.rs index cffef3009..3019b1d3f 100644 --- a/src/scripting/mod.rs +++ b/src/scripting/mod.rs @@ -180,7 +180,113 @@ pub fn handle_script_subcommand( } } +/// The shard's own Lua VM, plus what is needed to build it on first use. +/// +/// Exists because moon#508's fix ROUTES a script to the shard owning its keys, +/// so a script can now arrive over the SPSC mesh at a shard that has no +/// connection of its own and has therefore never built a VM. Before routing, +/// the VM was only ever created from `conn_accept`, on the connection's shard. +/// +/// The `Rc>>` slot is the SAME one `conn_accept` fills, so a +/// shard still has exactly one VM however it is first reached — whichever path +/// gets there first wins and the other reuses it. +pub struct ShardLuaRuntime { + slot: Rc>>>, + eviction_ctx: bridge::LuaEvictionCtx, + /// Carried here rather than added to the SPSC handler's already very wide + /// signature: both are per-shard constants, which is what this struct is. + num_shards: usize, +} + +impl ShardLuaRuntime { + pub fn new( + slot: Rc>>>, + eviction_ctx: bridge::LuaEvictionCtx, + num_shards: usize, + ) -> Self { + Self { + slot, + eviction_ctx, + num_shards, + } + } + + pub fn num_shards(&self) -> usize { + self.num_shards + } + + /// The shard's VM, built on first use. + /// + /// Returns `None` instead of panicking when the VM cannot be created: this + /// runs on the shard thread, where a panic aborts the whole process, and a + /// malformed-script client must never be able to do that. `conn_accept` + /// still `expect()`s at startup, where failing loudly is right. + pub fn vm(&self) -> Option> { + let mut slot = self.slot.borrow_mut(); + if slot.is_none() { + match setup_lua_vm(self.eviction_ctx.clone()) { + Ok(vm) => *slot = Some(vm), + Err(e) => { + tracing::error!("Lua VM initialization failed on shard thread: {e}"); + return None; + } + } + } + slot.clone() + } +} + +/// Where a script must run, decided by the shard ownership of its keys. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ScriptRoute { + /// Run here: no keys at all, single shard, or every key is local. + Local, + /// Every key lives on this OTHER shard — send the script there. + Remote(usize), + /// The keys span shards. A script executes against ONE shard's database, + /// so this genuinely cannot be served and must be refused. + CrossShard, +} + +/// Decide where a script's keys require it to run (moon#508). +/// +/// Before this existed, [`validate_keys_same_shard`] was the whole policy, and +/// it asked the wrong question: it required every key to hash to the shard the +/// CONNECTION happened to occupy. One key cannot cross slots, but it very +/// easily lives on another shard, so a single-key script was refused with +/// `CROSSSLOT` about `1 - 1/shards` of the time — 7 of 8 measured at +/// `--shards 4`. `CROSSSLOT` was standing in for "I cannot run this *here*", +/// and nothing ever asked where it *could* run. +/// +/// Keyless scripts stay local deliberately: with no key there is nothing to +/// route by, every shard is equally correct, and running in place avoids a +/// pointless hop. That is also why `numkeys=0` always worked and made the +/// defect look intermittent instead of systematic. +pub fn route_script_keys(keys: &[Bytes], shard_id: usize, num_shards: usize) -> ScriptRoute { + if num_shards <= 1 || keys.is_empty() { + return ScriptRoute::Local; + } + use crate::shard::dispatch::key_to_shard; + let target = key_to_shard(&keys[0], num_shards); + if keys[1..] + .iter() + .any(|k| key_to_shard(k, num_shards) != target) + { + return ScriptRoute::CrossShard; + } + if target == shard_id { + ScriptRoute::Local + } else { + ScriptRoute::Remote(target) + } +} + /// Validate that all keys hash to the current shard. Returns Some(error) on violation. +/// +/// Retained as the shard-side backstop AFTER [`route_script_keys`] has already +/// sent the script to the owning shard: at that point every key must be local, +/// so a violation here means the routing decision and the execution site +/// disagree — which would silently read another shard's (empty) view of a key. pub fn validate_keys_same_shard( keys: &[Bytes], shard_id: usize, diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index ab7c738c7..67ab0ffb8 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -204,7 +204,7 @@ pub(super) fn try_handle_cluster( /// `#[inline]`: see `try_handle_cluster` rationale — name check inlines to the /// caller so non-EVALSHA commands cost only a length + byte compare. #[inline] -pub(super) fn try_handle_evalsha( +pub(super) async fn try_handle_evalsha( cmd: &[u8], cmd_args: &[Frame], conn: &ConnectionState, @@ -214,6 +214,13 @@ pub(super) fn try_handle_evalsha( if !cmd.eq_ignore_ascii_case(b"EVALSHA") { return false; } + if let Some(routed) = + crate::server::conn::shared::route_script_elsewhere(cmd, cmd_args, conn.selected_db, ctx) + .await + { + responses.push(routed); + return true; + } let response = crate::shard::slice::with_shard(|s| { let db_count = s.databases.len(); crate::scripting::handle_evalsha( @@ -236,7 +243,7 @@ pub(super) fn try_handle_evalsha( /// `#[inline]`: see `try_handle_cluster` rationale — name check inlines so /// non-matching commands cost only a length + byte compare. #[inline] -pub(super) fn try_handle_eval( +pub(super) async fn try_handle_eval( cmd: &[u8], cmd_args: &[Frame], conn: &ConnectionState, @@ -246,6 +253,13 @@ pub(super) fn try_handle_eval( if !cmd.eq_ignore_ascii_case(b"EVAL") { return false; } + if let Some(routed) = + crate::server::conn::shared::route_script_elsewhere(cmd, cmd_args, conn.selected_db, ctx) + .await + { + responses.push(routed); + return true; + } let response = crate::shard::slice::with_shard(|s| { let db_count = s.databases.len(); let db = &mut s.databases[conn.selected_db]; diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 228e71cb5..a461d9c1c 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1868,11 +1868,12 @@ pub(crate) async fn handle_connection_sharded_monoio< continue; } if cmd_len == 7 - && dispatch::try_handle_evalsha(cmd, cmd_args, &conn, ctx, &mut responses) + && dispatch::try_handle_evalsha(cmd, cmd_args, &conn, ctx, &mut responses).await { continue; } - if cmd_len == 4 && dispatch::try_handle_eval(cmd, cmd_args, &conn, ctx, &mut responses) + if cmd_len == 4 + && dispatch::try_handle_eval(cmd, cmd_args, &conn, ctx, &mut responses).await { continue; } diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 8c6a23314..096e26c4b 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -1097,6 +1097,20 @@ pub(crate) async fn handle_connection_sharded_inner< // --- Lua scripting: EVAL / EVALSHA --- if cmd.eq_ignore_ascii_case(b"EVAL") || cmd.eq_ignore_ascii_case(b"EVALSHA") { + // moon#508: a script whose keys all live on another + // shard runs THERE. Same helper as handler_monoio — + // one routing policy, not two that can drift. + if let Some(routed) = crate::server::conn::shared::route_script_elsewhere( + cmd, + cmd_args, + conn.selected_db, + ctx, + ) + .await + { + responses.push(routed); + continue; + } let db_count = ctx.shard_databases.db_count(); // Unconditional slice path: ShardSlice is always initialized. let response = crate::shard::slice::with_shard_db(conn.selected_db, |db| { diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index f72523807..c7a63c5b9 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -1106,6 +1106,60 @@ fn is_inline_intercepted(cmd: &[u8]) -> bool { } } +/// If this script's keys all live on ANOTHER shard, run it there and return +/// that shard's reply. `None` means "run it here" — no keys, one shard, or the +/// keys are already local. +/// +/// This is the moon#508 fix. Previously any script whose keys were not on the +/// connection's own shard was answered `CROSSSLOT`, which is only true when the +/// keys span shards — a single key never crosses anything, it just lives +/// somewhere else. +/// +/// A genuinely cross-shard key set still returns `CROSSSLOT` here, before the +/// hop: a script executes against one shard's database and cannot reach +/// another's, so there is no target to send it to. +pub(crate) async fn route_script_elsewhere( + cmd: &[u8], + cmd_args: &[Frame], + db_index: usize, + ctx: &crate::server::conn::core::ConnectionContext, +) -> Option { + if ctx.num_shards <= 1 { + return None; + } + // EVALSHA carries a sha where EVAL carries the body, but `parse_eval_args` + // only reads args[1..] for numkeys/keys — so the same parse serves both and + // the sha never has to be resolved just to decide where to run. + let keys = match crate::scripting::parse_eval_args(cmd_args) { + // Malformed args: let the local handler produce the exact error it + // always did rather than inventing a routing-flavoured one here. + Err(_) => return None, + Ok((_script, _numkeys, keys, _argv)) => keys, + }; + match crate::scripting::route_script_keys(&keys, ctx.shard_id, ctx.num_shards) { + crate::scripting::ScriptRoute::Local => None, + crate::scripting::ScriptRoute::CrossShard => Some(Frame::Error(Bytes::from_static( + b"CROSSSLOT Keys in script don't hash to the same slot and shard", + ))), + crate::scripting::ScriptRoute::Remote(target) => { + let mut parts = Vec::with_capacity(cmd_args.len() + 1); + parts.push(Frame::BulkString(Bytes::copy_from_slice(cmd))); + parts.extend_from_slice(cmd_args); + Some( + crate::shard::coordinator::coordinate_script( + std::sync::Arc::new(Frame::Array(parts.into())), + target, + ctx.shard_id, + db_index, + &ctx.dispatch_tx, + &ctx.spsc_notifiers, + ) + .await, + ) + } + } +} + /// 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/src/shard/coordinator.rs b/src/shard/coordinator.rs index 932f93d1e..1e3cc40d6 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -1655,6 +1655,54 @@ async fn coordinate_multi_del_or_exists( Frame::Integer(total_count) } +/// Send a script to the ONE shard that owns all of its keys, and return that +/// shard's reply verbatim (moon#508). +/// +/// Unlike every other coordinator here this is not a fan-out: a script runs +/// against a single shard's database, so there is exactly one correct place +/// for it. `scripting::route_script_keys` has already established that every +/// key maps to `target`; this only moves the call there. +/// +/// The reply is passed through untouched, including errors — a `NOSCRIPT` from +/// the target shard is a real answer about the target's script cache, and +/// rewriting it here would hide a cache fan-out gap. +pub async fn coordinate_script( + command: std::sync::Arc, + target: usize, + my_shard: usize, + db_index: usize, + dispatch_tx: &Rc>>>, + spsc_notifiers: &[Arc], +) -> Frame { + let (reply_tx, reply_rx) = channel::oneshot(); + let msg = ShardMessage::Execute { + db_index, + command, + reply_tx, + }; + // Both failure outcomes mean the script was NEVER executed, so each can be + // reported as a clean reject rather than an ambiguous "maybe it ran". + match spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await { + crate::shard::dispatch::PushOutcome::Pushed => {} + crate::shard::dispatch::PushOutcome::Backpressure => { + return Frame::Error(Bytes::from_static( + b"ERR shard owning the script's keys is not draining; script not executed", + )); + } + crate::shard::dispatch::PushOutcome::Cancelled => { + return Frame::Error(Bytes::from_static( + b"ERR shutting down; script not executed", + )); + } + } + match recv_reply_bounded(reply_rx).await { + Ok(frame) => frame, + Err(_) => Frame::Error(Bytes::from_static( + b"ERR cross-shard reply channel closed during script execution", + )), + } +} + /// Coordinate KEYS across all shards. /// /// Dispatches KEYS command to every shard, collects and merges results. diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index e1e0e6fde..2e7516a39 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -827,6 +827,27 @@ impl super::Shard { .as_ref() .map(|rs| rs.read().is_replica_mirror.clone()); + // moon#508: a script whose keys all live on THIS shard is now routed + // here over the SPSC mesh, so the shard must be able to run one even + // when no connection ever landed on it. Shares `lua_rc` — the same slot + // `conn_accept` lazily fills — so a shard still builds exactly one VM + // whichever path reaches it first. + let shard_lua_rt = crate::scripting::ShardLuaRuntime::new( + lua_rc.clone(), + crate::scripting::bridge::LuaEvictionCtx::new( + shard_databases.clone(), + runtime_config.clone(), + shard_id, + spill_sender.clone(), + spill_file_id.clone(), + disk_offload_dir.clone(), + num_shards, + repl_state.clone(), + aof_pool.as_ref().map(Arc::clone), + ), + num_shards, + ); + // Track last seen snapshot epoch to detect watch channel triggers // Test-only fault injection: delay every non-zero shard's loop start so // integration tests can deterministically exercise the startup window @@ -1445,7 +1466,7 @@ impl super::Shard { &shard_databases, &mut consumers, &pubsub_arc, &blocking_rc, &mut pending_snapshot, &mut snapshot_state, &mut wal_writer, &repl_backlog, &mut replica_txs, - &repl_offsets, shard_id, &script_cache_rc, &cached_clock, + &repl_offsets, shard_id, &script_cache_rc, Some(&shard_lua_rt), &cached_clock, &mut pending_migrations, &mut pending_cdc_subscribes, &mut shard_manifest, @@ -1564,7 +1585,7 @@ impl super::Shard { &shard_databases, &mut consumers, &pubsub_arc, &blocking_rc, &mut pending_snapshot, &mut snapshot_state, &mut wal_writer, &repl_backlog, &mut replica_txs, - &repl_offsets, shard_id, &script_cache_rc, &cached_clock, + &repl_offsets, shard_id, &script_cache_rc, Some(&shard_lua_rt), &cached_clock, &mut pending_migrations, &mut pending_cdc_subscribes, &mut shard_manifest, @@ -2304,6 +2325,7 @@ impl super::Shard { &repl_offsets, shard_id, &script_cache_rc, + Some(&shard_lua_rt), &cached_clock, &mut pending_migrations, &mut pending_cdc_subscribes, diff --git a/src/shard/mod.rs b/src/shard/mod.rs index cc3731538..ac86f0130 100644 --- a/src/shard/mod.rs +++ b/src/shard/mod.rs @@ -462,6 +462,7 @@ mod tests { &None, 0, &script_cache, + None, // no shard Lua runtime in this unit test &clock, &mut Vec::new(), &mut Vec::new(), @@ -527,6 +528,7 @@ mod tests { &None, 0, &script_cache, + None, // no shard Lua runtime in this unit test &clock, &mut Vec::new(), &mut Vec::new(), diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 9e554de4c..5b8a30973 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -129,6 +129,10 @@ pub(crate) fn drain_spsc_shared( repl_state: &Option, shard_id: usize, script_cache: &Rc>, + // moon#508: the shard's Lua VM, so a script ROUTED here (because this + // shard owns its keys) can actually run. `None` in unit tests that + // drive the SPSC path without a shard runtime. + lua_rt: Option<&crate::scripting::ShardLuaRuntime>, cached_clock: &CachedClock, pending_migrations: &mut Vec<( crate::shard::dispatch::MigrateFd, @@ -330,6 +334,7 @@ pub(crate) fn drain_spsc_shared( repl_state, shard_id, script_cache, + lua_rt, cached_clock, shard_manifest, mvcc_prune_margin, @@ -366,6 +371,7 @@ pub(crate) fn drain_spsc_shared( repl_state, shard_id, script_cache, + lua_rt, cached_clock, shard_manifest, mvcc_prune_margin, @@ -416,6 +422,10 @@ pub(crate) fn handle_shard_message_shared( repl_state: &Option, shard_id: usize, script_cache: &Rc>, + // moon#508: the shard's Lua VM, so a script ROUTED here (because this + // shard owns its keys) can actually run. `None` in unit tests that + // drive the SPSC path without a shard runtime. + lua_rt: Option<&crate::scripting::ShardLuaRuntime>, cached_clock: &CachedClock, // P8: optional manifest for VACUUM manifest/WAL passes; None when no persistence_dir. shard_manifest: &mut Option, @@ -460,6 +470,59 @@ pub(crate) fn handle_shard_message_shared( } }; + // EVAL/EVALSHA routed here because THIS shard owns the + // script's keys (moon#508). `cmd_dispatch` has no scripting + // arm — scripts are intercepted at the connection layer — so + // without this a routed script would come back as an unknown + // command. Intercepted before cmd_dispatch for the same reason + // the FT. block below is. + let is_plain_eval = cmd.eq_ignore_ascii_case(b"EVAL"); + if is_plain_eval || cmd.eq_ignore_ascii_case(b"EVALSHA") { + let Some(rt) = lua_rt else { + let _ = reply_tx.send(crate::protocol::Frame::Error( + bytes::Bytes::from_static( + b"ERR scripting is unavailable on this shard", + ), + )); + return; + }; + let Some(vm) = rt.vm() else { + let _ = reply_tx.send(crate::protocol::Frame::Error( + bytes::Bytes::from_static(b"ERR Lua VM initialization failed"), + )); + return; + }; + let frame = crate::shard::slice::with_shard(|s| { + let db_count = s.databases.len(); + let db = &mut s.databases[db_idx]; + if is_plain_eval { + crate::scripting::handle_eval( + &vm, + script_cache, + args, + db, + shard_id, + rt.num_shards(), + db_idx, + db_count, + ) + } else { + crate::scripting::handle_evalsha( + &vm, + script_cache, + args, + db, + shard_id, + rt.num_shards(), + db_idx, + db_count, + ) + } + }); + let _ = reply_tx.send(frame); + return; + } + // FT.* commands route to VectorStore, not the KV Database. // Intercept before cmd_dispatch so the console gateway's // ShardMessage::Execute path reaches the vector handlers. @@ -4519,6 +4582,7 @@ mod drain_cap_tests { &offsets, 0, &script_cache, + None, // no shard Lua runtime in this unit test &clock, &mut migrations, &mut cdc, @@ -4553,6 +4617,7 @@ mod drain_cap_tests { &offsets, 0, &script_cache, + None, // no shard Lua runtime in this unit test &clock, &mut migrations, &mut cdc, diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 841d14fd4..01898e49a 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -24,6 +24,7 @@ #![allow(dead_code)] use std::collections::HashSet; +use std::io::{Read, Write}; use std::net::TcpStream; use std::path::PathBuf; use std::process::Child; @@ -210,3 +211,163 @@ pub fn wait_for_port_down(port: u16) { alive, or SO_REUSEPORT is masking a listener that never exited" ); } + +// --------------------------------------------------------------------------- +// Raw-RESP client +// --------------------------------------------------------------------------- +// +// Suites that must control exactly what goes into one TCP write (pipeline +// ordering, script routing) cannot use a client library: a library is free to +// split or reorder the batch, which is the very thing under test. +// +// The framer here replaced a "read until the socket is quiet for 250ms" +// heuristic. That silently TRUNCATED a reply whenever the server paused longer +// than the window mid-stream, and a short read then surfaced as a wrong VALUE +// — so the suite reported the wrong defect. Do not reintroduce a timing-based +// reader. + +pub struct Conn { + pub 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, +} + +pub 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. +pub 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 { + pub 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. + pub 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()) + } + + pub 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. + pub 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()), + } + } + } +} diff --git a/tests/pipeline_cross_shard_ordering.rs b/tests/pipeline_cross_shard_ordering.rs index 301448ffc..9c380ab6e 100644 --- a/tests/pipeline_cross_shard_ordering.rs +++ b/tests/pipeline_cross_shard_ordering.rs @@ -37,6 +37,8 @@ mod common; +use common::{Conn, framed_len}; + use std::io::{Read, Write}; use std::net::TcpStream; use std::process::{Child, Command, Stdio}; @@ -121,152 +123,6 @@ fn spawn_moon(shards: &str) -> Moon { 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 { diff --git a/tests/script_key_routing.rs b/tests/script_key_routing.rs new file mode 100644 index 000000000..6979df9a1 --- /dev/null +++ b/tests/script_key_routing.rs @@ -0,0 +1,253 @@ +//! A script whose keys all live on ONE shard must run there. (moon#508) +//! +//! Filed as "EVALSHA of a single-key script fails with CROSSSLOT", with the +//! guess that a 1-element key list was being mis-folded. It is not: the check +//! (`scripting::validate_keys_same_shard`) requires every key to hash to the +//! shard the CONNECTION happens to be on. One key cannot cross slots, but it +//! very easily lives on another shard — so `CROSSSLOT` was standing in for +//! "I cannot run this here", and the script never got routed anywhere else. +//! +//! Measured against the unfixed build at `--shards 4`, 8 distinct keys on +//! fresh connections: 7 of 8 rejected. Only the key that happened to land on +//! the connection's own shard ran. `numkeys=0` always worked, which is why the +//! defect reads as intermittent rather than total. +//! +//! This breaks `redis.lock.Lock.release()` — implemented as a single-key +//! EVALSHA — and through redis-py's `Script.__call__` wrapper the caller sees +//! a `NoScriptError` followed by a cross-slot error, neither of which names +//! the real cause. +//! +//! Every assertion here is written so the FIXED state is the passing state, +//! and each loops over key placements: a single trial only samples one +//! placement, and at `--shards 4` a lucky key passes ~25% of the time. + +mod common; + +use common::Conn; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +/// Four shards: a key is remote ~75% of the time, so a build that only ever +/// runs scripts locally cannot pass by luck. +const SHARDS: &str = "4"; +/// Distinct keys per assertion. At p(remote)=0.75 the chance that all 12 land +/// on the connection's own shard — and vacuously pass — is under 1e-7. +const TRIALS: usize = 12; + +/// `return redis.call('get', KEYS[1])` — the shape of every single-key script +/// a lock or cache wrapper issues. +const GET_SCRIPT: &str = "return redis.call('get',KEYS[1])"; + +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-scriptroute-{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-scriptroute-{port}")); + let 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 log = std::fs::read_to_string(moon.tmp_dir.join("moon.stderr")).unwrap_or_default(); + panic!("moon never became ready on port {port}\n--- stderr ---\n{log}"); +} + +/// Run `body` once per key placement on a FRESH connection, collecting the +/// trials that came out wrong. +/// +/// Fresh connections on purpose: which shard a connection lands on is what +/// decides whether its key is "local", so reusing one connection would sample +/// a single placement TRIALS times 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); + let key = format!("{tag}{i}"); + if let Some(why) = body(&mut c, &key) { + wrong.push(format!(" key {key}: {why}")); + } + } + assert!( + wrong.is_empty(), + "{}/{} key placements wrong (moon#508 — a script whose keys all live \ + on one shard must be ROUTED there, not rejected):\n{}", + wrong.len(), + TRIALS, + wrong.join("\n") + ); +} + +/// A single-key EVAL must run wherever its key lives. +#[test] +fn skr1_single_key_eval_runs_on_the_keys_shard() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "skr1", |c, k| { + c.send(&["SET", k, "v"]); + let r = c.send(&["EVAL", GET_SCRIPT, "1", k]); + (r != "$1\r\nv\r\n").then(|| format!("EVAL replied {r:?}, expected the value")) + }); +} + +/// The reported symptom: EVALSHA of a single-key script, the shape +/// `redis.lock.Lock.release()` issues. +#[test] +fn skr2_single_key_evalsha_runs_on_the_keys_shard() { + let m = spawn_moon(SHARDS); + let mut loader = Conn::open(m.port); + let load = loader.send(&["SCRIPT", "LOAD", GET_SCRIPT]); + let sha = load + .rsplit("\r\n") + .find(|s| s.len() == 40) + .unwrap_or_else(|| panic!("SCRIPT LOAD did not return a sha: {load:?}")) + .to_string(); + + each_trial(m.port, "skr2", |c, k| { + c.send(&["SET", k, "v"]); + // SCRIPT EXISTS proves the cache is not the problem: a NOSCRIPT here + // would be a different defect (a load fan-out that missed a shard). + let ex = c.send(&["SCRIPT", "EXISTS", &sha]); + if !ex.contains(":1") { + return Some(format!("SCRIPT EXISTS replied {ex:?} — script not cached")); + } + let r = c.send(&["EVALSHA", &sha, "1", k]); + (r != "$1\r\nv\r\n").then(|| format!("EVALSHA replied {r:?}, expected the value")) + }); +} + +/// A script must be able to WRITE the key it was routed for, not just read it +/// — otherwise routing could be faked by answering reads from the wrong shard. +#[test] +fn skr3_single_key_script_writes_land_on_the_keys_shard() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "skr3", |c, k| { + c.send(&["DEL", k]); + let set = c.send(&[ + "EVAL", + "return redis.call('set',KEYS[1],ARGV[1])", + "1", + k, + "w", + ]); + if !set.contains("OK") { + return Some(format!("script SET replied {set:?}")); + } + // Read it back through the NORMAL path: if the script wrote to the + // wrong shard's database, a plain GET (which routes correctly) misses. + let got = c.send(&["GET", k]); + (got != "$1\r\nw\r\n").then(|| format!("GET after script SET replied {got:?}")) + }); +} + +/// `numkeys=0` has no key to route by and must keep working — it is the case +/// that masked the defect, since it always succeeded. +#[test] +fn skr4_keyless_script_still_runs() { + let m = spawn_moon(SHARDS); + each_trial(m.port, "skr4", |c, _k| { + let r = c.send(&["EVAL", "return 1+1", "0"]); + (r != ":2\r\n").then(|| format!("keyless EVAL replied {r:?}")) + }); +} + +/// The guard must SURVIVE: keys that genuinely span shards still have to be +/// rejected, because a script runs against one shard's database and cannot +/// reach another's. +/// +/// This is the assertion that stops the fix from being "delete the check". +#[test] +fn skr5_genuinely_cross_shard_keys_are_still_rejected() { + let m = spawn_moon(SHARDS); + // Hash tags force co-location, so `{a}k` and `{b}k` land wherever their + // TAG hashes — over many distinct tags, some pairs must differ. + let mut rejected = 0usize; + let mut accepted = 0usize; + for i in 0..24 { + let mut c = Conn::open(m.port); + let k1 = format!("{{x{i}}}one"); + let k2 = format!("{{y{i}}}two"); + c.send(&["SET", &k1, "1"]); + c.send(&["SET", &k2, "2"]); + let r = c.send(&[ + "EVAL", + "return {redis.call('get',KEYS[1]),redis.call('get',KEYS[2])}", + "2", + &k1, + &k2, + ]); + if r.contains("CROSSSLOT") { + rejected += 1; + } else if r.contains('1') && r.contains('2') { + accepted += 1; + } else { + panic!("2-key script replied neither a value pair nor CROSSSLOT: {r:?}"); + } + } + assert!( + rejected > 0, + "no 2-key script was rejected across 24 tag pairs — the cross-shard \ + guard is gone, so a script can now silently read another shard's \ + (empty) view of a key. Fixing #508 must ROUTE same-shard scripts, \ + not delete the check." + ); + assert!( + accepted > 0, + "every 2-key script was rejected across 24 tag pairs — pairs that DO \ + co-locate must still run, or this is just the old bug with extra steps" + ); +} From f01814fff0a84abdb95a7216cca069bd16c51a16 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 17 Aug 2026 00:35:59 +0700 Subject: [PATCH 2/2] fix(scripting): a routed script that gets no reply must not claim it never ran MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of #516 found four issues; three are fixed here and one is filed. 1. coordinator: a reply TIMEOUT was reported as a closed channel. `recv_reply_bounded` returns the same `Err(RecvError)` for two outcomes with opposite retry semantics — the target dropped the sender, and the reply did not arrive within the 30s `XSHARD_REPLY_TIMEOUT`. `coordinate_script` collapsed both into "cross-shard reply channel closed during script execution", which asserts the script never executed. A timeout means the target may still be running it, or may have already applied its writes, so a client told the script never ran will re-send a non-idempotent script that did. This contradicted the PR's own claim that its failure handling was explicit rather than collapsed. `ReplyFailure::{Closed, TimedOut}` now distinguishes them, and the race is split into `recv_reply_within(rx, timeout)` so `recv_reply_bounded` can delegate — every existing call site is unchanged and the race logic is not duplicated. Both arms now report execution status as UNKNOWN: `Closed` too, because a target can drop the sender AFTER applying writes (a panic between apply and reply), so it means "no reply", not "no effect". The timeout records `moon_xshard_reply_timeout_total{kind="script"}`, matching the handler reply paths — without it a wedged owner shard was invisible in metrics. 2. skr5 asserted a probabilistic outcome. `accepted > 0` depended on `{x}`/`{y}` coincidentally co-locating; at 4 shards `P(none of 24 co-locate) = (3/4)^24`, about 1 in 1000. The failure text blamed the fix ("just the old bug with extra steps"), so an unlucky run across four CI platforms would have read as a real regression. Replaced with a `{same}one`/`{same}two` pair — co-located by construction — asserting it both runs and is not CROSSSLOT. The distinct-tag loop still requires `rejected > 0`, plus `rejected + accepted == 24` so every pair stays classified. 3. the shared RESP framer mis-counted RESP3 attribute frames. `|N` fell into the single-line arm and was counted as a top-level reply, so its key/value payload and the reply it decorates were left in the spill buffer and every later frame was read one position out of step. Attributes are now handled before the count, adding N*2 to `pending`. Not reachable today (these suites never send HELLO 3), but the helper documents RESP2+RESP3 and is now shared by two suites. Each of the three is mutation-verified: removing the attribute arm frames `|1\r\n$3\r\nttl\r\n:99\r\n+OK\r\n` at byte 4 instead of 23; collapsing `TimedOut` into `Closed` fails the timeout test; forcing every multi-key script to CrossShard fails the new same-tag leg. 4. NOT fixed here — filed as #517. Script writes skip `cow_intercept` on every path, and under `runtime-tokio` `emit_effect` is compiled out, so they emit no AOF or replication record. Both are real, and both are pre-existing rather than introduced by #516: the routed path is structurally identical to the local path this PR did not touch, and before #516 a script whose keys lived elsewhere was refused outright, so there was no correct behaviour to regress. A persistence change needs replication parity, kill-9 durability and a VM A/B bench, which a routing PR would not subject it to. Verified: fmt clean; clippy clean on default and tokio+jemalloc; tokio full suite 4626 passed / 0 failed. Two monoio full-suite runs hit only known flakes and nothing in the changed code: run 1 `bpv1` (a 30s server-startup timeout, 5/5 in isolation), run 2 `cb12`+`cb15` at cluster_client_bootstrap.rs:213 and :172 — the two panic sites of #505, which a merge-base A/B this session showed failing MORE on base (4/8) than on this branch (3/8). Refs #508, #516, #517 author: Tin Dang --- CHANGELOG.md | 15 +++ src/shard/coordinator.rs | 123 ++++++++++++++++++++++--- tests/common/mod.rs | 13 +++ tests/pipeline_cross_shard_ordering.rs | 13 +++ tests/script_key_routing.rs | 33 ++++++- 5 files changed, 180 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 88268f7d0..93dc158eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -163,6 +163,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 so routing alone would trade `CROSSSLOT` for `ERR Function not found` at the same rate. Filed as #514 rather than half-fixed here. + A routed script that gets no reply no longer claims it did not run. `recv_reply_bounded` returns + the same error for a closed channel and for a 30s reply timeout, and the first cut of this path + reported both as "cross-shard reply channel closed during script execution". Those have opposite + retry semantics: a timeout means the target may still be executing, or may have already applied + its writes, so a client told the script never ran will re-send a non-idempotent script that did. + The two are now distinguished (`ReplyFailure::{Closed, TimedOut}`), both say execution status is + **unknown**, and the timeout records `moon_xshard_reply_timeout_total{kind="script"}` so a wedged + owner shard is visible in metrics like every other cross-shard reply path. + + Not fixed here, and pre-existing rather than introduced: script writes skip `cow_intercept` on + every path, and under `runtime-tokio` `emit_effect` is compiled out entirely, so script writes + emit no AOF or replication record. Routed scripts persist exactly as well as local ones do — + which is the bug. Filed as #517, since a fix needs replication parity, kill-9 durability and a + VM A/B bench of its own. + - **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 diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index 1e3cc40d6..33e36b86d 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -192,29 +192,71 @@ fn run_local( // single constant so the two reply paths can't drift apart. use crate::shard::dispatch::XSHARD_REPLY_TIMEOUT; -/// Await a cross-shard `reply_rx` with a bounded timeout (#11). +/// Why a bounded cross-shard receive produced no reply. /// -/// Returns `Err(RecvError)` on EITHER a closed channel (the target shard -/// dropped the reply sender) OR expiry of [`XSHARD_REPLY_TIMEOUT`] — both mean -/// "no usable reply", so every call site's existing error handling applies -/// unchanged; the point is that neither case can hang the connection forever. -pub(crate) async fn recv_reply_bounded( +/// The distinction is load-bearing for RETRY SAFETY, which is why it is an enum +/// and not a bool: `Closed` and `TimedOut` say different things about whether +/// the command ran. Collapsing them lets a caller report "never executed" for a +/// target that is still executing — and a client that believes a non-idempotent +/// command never ran will happily re-send it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ReplyFailure { + /// The target dropped the reply sender without sending. It may have dropped + /// it *after* applying writes (e.g. a panic between apply and reply), so + /// this is "no reply", NOT "no effect". + Closed, + /// [`XSHARD_REPLY_TIMEOUT`] expired. The target may still be executing, or + /// may have already applied its writes. Execution status is UNKNOWN. + TimedOut, +} + +/// Await `reply_rx` for at most `timeout`, reporting WHICH failure occurred. +/// +/// Split out from [`recv_reply_bounded_reason`] purely so tests can drive both +/// arms without waiting out the real 30s [`XSHARD_REPLY_TIMEOUT`]. +async fn recv_reply_within( reply_rx: channel::OneshotReceiver, -) -> Result { + timeout: std::time::Duration, +) -> Result { use crate::runtime::race::{Arm, race2}; let recv = std::pin::pin!(reply_rx); #[cfg(feature = "runtime-tokio")] - let sleep = std::pin::pin!(tokio::time::sleep(XSHARD_REPLY_TIMEOUT)); + let sleep = std::pin::pin!(tokio::time::sleep(timeout)); #[cfg(feature = "runtime-monoio")] - let sleep = std::pin::pin!(monoio::time::sleep(XSHARD_REPLY_TIMEOUT)); + let sleep = std::pin::pin!(monoio::time::sleep(timeout)); // race2 polls the recv arm first, so a ready reply always wins the tie and // the timer future is dropped un-fired (cheap deregister on both runtimes). match race2(recv, sleep).await { - Arm::First(r) => r, - Arm::Second(()) => Err(channel::RecvError), + Arm::First(Ok(v)) => Ok(v), + Arm::First(Err(_)) => Err(ReplyFailure::Closed), + Arm::Second(()) => Err(ReplyFailure::TimedOut), } } +/// Await a cross-shard `reply_rx` with a bounded timeout (#11), preserving the +/// reason. Prefer this at any call site whose error text makes a claim about +/// whether the command executed. +pub(crate) async fn recv_reply_bounded_reason( + reply_rx: channel::OneshotReceiver, +) -> Result { + recv_reply_within(reply_rx, XSHARD_REPLY_TIMEOUT).await +} + +/// Await a cross-shard `reply_rx` with a bounded timeout (#11). +/// +/// Returns `Err(RecvError)` on EITHER a closed channel (the target shard +/// dropped the reply sender) OR expiry of [`XSHARD_REPLY_TIMEOUT`] — both mean +/// "no usable reply", so every call site's existing error handling applies +/// unchanged; the point is that neither case can hang the connection forever. +/// Use [`recv_reply_bounded_reason`] when the two need telling apart. +pub(crate) async fn recv_reply_bounded( + reply_rx: channel::OneshotReceiver, +) -> Result { + recv_reply_bounded_reason(reply_rx) + .await + .map_err(|_| channel::RecvError) +} + /// Send one full command to a REMOTE shard and await its reply. async fn run_remote( target_shard: usize, @@ -1695,10 +1737,22 @@ pub async fn coordinate_script( )); } } - match recv_reply_bounded(reply_rx).await { + // Past this point the script IS in the target's queue, so no failure here + // can claim it did not run — only that we never heard back. Saying + // otherwise invites a client to re-send a non-idempotent script that + // already applied its writes. + match recv_reply_bounded_reason(reply_rx).await { Ok(frame) => frame, - Err(_) => Frame::Error(Bytes::from_static( - b"ERR cross-shard reply channel closed during script execution", + Err(ReplyFailure::TimedOut) => { + // Mirrors the handler reply paths, which already record this; + // without it a wedged owner shard is invisible in metrics. + crate::admin::metrics_setup::record_xshard_reply_timeout("script"); + Frame::Error(Bytes::from_static( + b"ERR timeout waiting for the shard owning the script's keys; script execution status is unknown", + )) + } + Err(ReplyFailure::Closed) => Frame::Error(Bytes::from_static( + b"ERR cross-shard reply channel closed; script execution status is unknown", )), } } @@ -3159,6 +3213,47 @@ pub async fn coordinate_swapdb( mod tests { use super::*; + // Both arms are driven at a millisecond timeout via `recv_reply_within` + // rather than the real 30s constant. Gated to runtime-tokio because the + // timer arm is the tokio one under this cfg (the monoio sleep needs a + // monoio runtime); the logic under test is runtime-independent. + #[cfg(feature = "runtime-tokio")] + #[tokio::test] + async fn recv_reply_reports_closed_when_sender_is_dropped() { + let (tx, rx) = channel::oneshot::(); + drop(tx); + let got = recv_reply_within(rx, std::time::Duration::from_millis(50)).await; + assert_eq!( + got.err(), + Some(ReplyFailure::Closed), + "a dropped sender is a closed channel, not a timeout" + ); + } + + #[cfg(feature = "runtime-tokio")] + #[tokio::test] + async fn recv_reply_reports_timeout_when_target_never_replies() { + // Hold the sender alive and never send: a wedged owner shard. This is + // the case that must NOT be reported as "closed" — the target may still + // be executing, so the caller cannot claim the command never ran. + let (_tx, rx) = channel::oneshot::(); + let got = recv_reply_within(rx, std::time::Duration::from_millis(20)).await; + assert_eq!( + got.err(), + Some(ReplyFailure::TimedOut), + "a silent-but-open target is a timeout, not a closed channel" + ); + } + + #[cfg(feature = "runtime-tokio")] + #[tokio::test] + async fn recv_reply_returns_the_reply_when_one_arrives() { + let (tx, rx) = channel::oneshot::(); + let _ = tx.send(Frame::SimpleString(Bytes::from_static(b"PONG"))); + let got = recv_reply_within(rx, std::time::Duration::from_millis(50)).await; + assert!(matches!(got, Ok(Frame::SimpleString(ref s)) if s.as_ref() == b"PONG")); + } + #[test] fn test_btreemap_ascending_order() { // BTreeMap guarantees ascending shard order -- VLL deadlock prevention diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 01898e49a..4822498d9 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -266,6 +266,19 @@ pub fn framed_len(buf: &[u8], want: usize) -> Option { let line = std::str::from_utf8(&buf[i + 1..end]).ok()?; i = end + 2; + // RESP3 attribute (`|N`): N key/value pairs of metadata attached to the + // reply that FOLLOWS. It is not a reply of its own, and not an element + // of an enclosing aggregate — so it must consume neither a `done` nor a + // `pending` slot, or the attributed reply is mistaken for the reply + // itself and every later frame is read one position out of step. + if tag == b'|' { + let n: i64 = line.parse().ok()?; + if n > 0 { + pending += (n as usize) * 2; + } + continue; + } + if pending > 0 { pending -= 1; } else { diff --git a/tests/pipeline_cross_shard_ordering.rs b/tests/pipeline_cross_shard_ordering.rs index 9c380ab6e..e10a942b5 100644 --- a/tests/pipeline_cross_shard_ordering.rs +++ b/tests/pipeline_cross_shard_ordering.rs @@ -502,6 +502,19 @@ fn pco0_reply_framer_counts_top_level_replies() { ("*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), + // RESP3 attribute (`|N`): metadata for the reply that FOLLOWS, so the + // pair AND the attributed reply must all be consumed for ONE reply. If + // `|` were counted as a reply of its own, this would frame after the + // header and leave the rest to be misread as the next reply. + ("|1\r\n$3\r\nttl\r\n:99\r\n+OK\r\n", 1), + // attributed reply inside an aggregate: the attribute must not eat an + // element slot either. + ("*2\r\n|1\r\n$3\r\nttl\r\n:99\r\n$1\r\na\r\n$1\r\nb\r\n", 1), + // two attributed replies back to back + ( + "|1\r\n$1\r\nk\r\n:1\r\n+OK\r\n|1\r\n$1\r\nk\r\n:2\r\n:7\r\n", + 2, + ), ]; for (raw, want) in cases { diff --git a/tests/script_key_routing.rs b/tests/script_key_routing.rs index 6979df9a1..f7d10aaa4 100644 --- a/tests/script_key_routing.rs +++ b/tests/script_key_routing.rs @@ -238,6 +238,11 @@ fn skr5_genuinely_cross_shard_keys_are_still_rejected() { panic!("2-key script replied neither a value pair nor CROSSSLOT: {r:?}"); } } + assert_eq!( + rejected + accepted, + 24, + "every pair must be classified as exactly one of ran / CROSSSLOT" + ); assert!( rejected > 0, "no 2-key script was rejected across 24 tag pairs — the cross-shard \ @@ -245,9 +250,31 @@ fn skr5_genuinely_cross_shard_keys_are_still_rejected() { (empty) view of a key. Fixing #508 must ROUTE same-shard scripts, \ not delete the check." ); + // NOT asserted on the loop above: whether any {x}/{y} pair happens to + // co-locate is chance (~1/SHARDS each), so "some pair was accepted" fails + // roughly (1 - 1/SHARDS)^24 of the time — ~1/1000 at 4 shards, which across + // four CI platforms is a flake that would read as a real regression. + // A SHARED tag is co-located by construction, so this leg is deterministic. + let mut c = Conn::open(m.port); + let (k1, k2) = ("{same}one", "{same}two"); + c.send(&["SET", k1, "1"]); + c.send(&["SET", k2, "2"]); + let r = c.send(&[ + "EVAL", + "return {redis.call('get',KEYS[1]),redis.call('get',KEYS[2])}", + "2", + k1, + k2, + ]); + assert!( + r.contains('1') && r.contains('2'), + "a 2-key script whose keys share a hash tag is co-located BY \ + CONSTRUCTION and must run, not be rejected — got {r:?}. Rejecting it \ + is the old bug with extra steps." + ); assert!( - accepted > 0, - "every 2-key script was rejected across 24 tag pairs — pairs that DO \ - co-locate must still run, or this is just the old bug with extra steps" + !r.contains("CROSSSLOT"), + "same-tag keys cannot cross slots; CROSSSLOT here means routing is \ + hashing something other than the tag — got {r:?}" ); }