diff --git a/CHANGELOG.md b/CHANGELOG.md index 48d077804..93dc158eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -125,6 +125,59 @@ 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 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/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..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, @@ -1655,6 +1697,66 @@ 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", + )); + } + } + // 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(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", + )), + } +} + /// Coordinate KEYS across all shards. /// /// Dispatches KEYS command to every shard, collects and merges results. @@ -3111,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/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..4822498d9 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,176 @@ 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; + + // 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 { + 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..e10a942b5 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 { @@ -646,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 new file mode 100644 index 000000000..f7d10aaa4 --- /dev/null +++ b/tests/script_key_routing.rs @@ -0,0 +1,280 @@ +//! 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_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 \ + 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." + ); + // 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!( + !r.contains("CROSSSLOT"), + "same-tag keys cannot cross slots; CROSSSLOT here means routing is \ + hashing something other than the tag — got {r:?}" + ); +}