fix(redis): harden delayed-member Lua decode against malformed members

chasquimq/src/redis/commands.rs

@@ -28,25 +28,51 @@ use fred::types::Value;
 /// prefix in Lua and re-emits the name as the stream entry's `n` field
 /// when non-empty (matching `xadd_args`'s shape on the immediate path).
 /// See `chasquimq/src/redis/delayed_member.rs` for the encoder.
+///
+/// **Malformed members are cleansed, not fatal.** A member shorter than the
+/// 4-byte length prefix, or whose declared `name_len` runs past the buffer,
+/// is `ZREM`'d and skipped instead of aborting the EVAL — matching the bounds
+/// checks in `decode_delayed_member`. Without this guard a single poison
+/// member permanently wedges promotion for the whole queue (every tick
+/// re-reads it and re-errors before the per-member cleanup runs).
 pub(crate) const PROMOTE_SCRIPT: &str = r#"
 local time = redis.call('TIME')
 local now_ms = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)
 local due = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now_ms, 'LIMIT', 0, tonumber(ARGV[1]))
 local payloads = {}
+local np = 0
 for i, member in ipairs(due) do
   -- Parse the slice-3 length-prefixed member:
   --   [u32_le name_len][name utf8][msgpack payload]
-  local b1, b2, b3, b4 = string.byte(member, 1, 4)
-  local name_len = b1 + (b2 * 256) + (b3 * 65536) + (b4 * 16777216)
-  local name = name_len > 0 and string.sub(member, 5, 4 + name_len) or ''
-  local payload = string.sub(member, 5 + name_len)
-  if name ~= '' then
-    redis.call('XADD', KEYS[2], 'MAXLEN', '~', tonumber(ARGV[2]), '*', 'd', payload, 'n', name)
+  -- A malformed/short member must NOT abort the whole EVAL: string.byte on a
+  -- <4-byte member returns nil, and arithmetic on nil errors out before the
+  -- per-member ZREM below, so the poison member is never removed and every
+  -- subsequent promote tick re-reads + re-errors -> promotion permanently
+  -- wedged for this queue. Mirror the Rust decoder (delayed_member.rs:
+  -- decode_delayed_member): bounds-check first, and on a bad member cleanse
+  -- it with ZREM and skip emitting a job. payloads stays a dense array (own
+  -- counter np, not the loop index i) so a skipped member can't punch a nil
+  -- hole that truncates the reply for valid members after it.
+  if #member < 4 then
+    redis.call('ZREM', KEYS[1], member)
   else
-    redis.call('XADD', KEYS[2], 'MAXLEN', '~', tonumber(ARGV[2]), '*', 'd', payload)
+    local b1, b2, b3, b4 = string.byte(member, 1, 4)
+    local name_len = b1 + (b2 * 256) + (b3 * 65536) + (b4 * 16777216)
+    if name_len > #member - 4 then
+      redis.call('ZREM', KEYS[1], member)
+    else
+      local name = name_len > 0 and string.sub(member, 5, 4 + name_len) or ''
+      local payload = string.sub(member, 5 + name_len)
+      if name ~= '' then
+        redis.call('XADD', KEYS[2], 'MAXLEN', '~', tonumber(ARGV[2]), '*', 'd', payload, 'n', name)
+      else
+        redis.call('XADD', KEYS[2], 'MAXLEN', '~', tonumber(ARGV[2]), '*', 'd', payload)
+      end
+      redis.call('ZREM', KEYS[1], member)
+      np = np + 1
+      payloads[np] = payload
+    end
   end
-  redis.call('ZREM', KEYS[1], member)
-  payloads[i] = payload
 end
 local depth = redis.call('ZCARD', KEYS[1])
 local lag_ms = 0
@@ -450,20 +476,39 @@ while i < fire_count do
   if fire_at_ms <= now_ms then
     -- Split the slice-3 length-prefixed member:
     --   [u32_le name_len][name utf8][msgpack payload]
-    local b1, b2, b3, b4 = string.byte(member, 1, 4)
-    local name_len = b1 + (b2 * 256) + (b3 * 65536) + (b4 * 16777216)
-    local name = name_len > 0 and string.sub(member, 5, 4 + name_len) or ''
-    local payload = string.sub(member, 5 + name_len)
-    if name ~= '' then
-      redis.call('XADD', KEYS[1], 'MAXLEN', '~', max_stream_len, '*', 'd', payload, 'n', name)
-    else
-      redis.call('XADD', KEYS[1], 'MAXLEN', '~', max_stream_len, '*', 'd', payload)
+    -- Bounds-check before string.byte so a malformed/short member can't abort
+    -- the whole EVAL (string.byte on a <4-byte member returns nil; arithmetic
+    -- on nil errors and the whole fire batch aborts). Mirror the Rust decoder
+    -- (delayed_member.rs: decode_delayed_member). A poison member is dropped
+    -- (it can't be dispatched), so it must NOT count toward fired / fired_now
+    -- -- counting it would burn a fire slot against the repeatable limit and
+    -- could prematurely retire the spec. The rest of the batch fires normally.
+    local valid = false
+    local name = ''
+    local payload = nil
+    if #member >= 4 then
+      local b1, b2, b3, b4 = string.byte(member, 1, 4)
+      local name_len = b1 + (b2 * 256) + (b3 * 65536) + (b4 * 16777216)
+      if name_len <= #member - 4 then
+        valid = true
+        name = name_len > 0 and string.sub(member, 5, 4 + name_len) or ''
+        payload = string.sub(member, 5 + name_len)
+      end
+    end
+    if valid then
+      if name ~= '' then
+        redis.call('XADD', KEYS[1], 'MAXLEN', '~', max_stream_len, '*', 'd', payload, 'n', name)
+      else
+        redis.call('XADD', KEYS[1], 'MAXLEN', '~', max_stream_len, '*', 'd', payload)
+      end
+      redis.call('HINCRBY', KEYS[4], 'fired', 1)
+      fired_now = fired_now + 1
     end
   else
     redis.call('ZADD', KEYS[2], fire_at_ms, member)
+    redis.call('HINCRBY', KEYS[4], 'fired', 1)
+    fired_now = fired_now + 1
   end
-  redis.call('HINCRBY', KEYS[4], 'fired', 1)
-  fired_now = fired_now + 1
   i = i + 1
 end
 
@@ -1062,3 +1107,53 @@ pub(crate) fn evalsha_acquire_lock_args(
         Value::from(ttl_secs as i64),
     ]
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Structural regression guard. Both delayed-member decoders previously
+    /// did `string.byte(member, 1, 4)` immediately followed by `name_len`
+    /// arithmetic with no bounds check; a member shorter than the 4-byte
+    /// prefix made `string.byte` return nil, the arithmetic erred on nil, and
+    /// the whole EVAL aborted before any per-member cleanup ran — permanently
+    /// wedging promotion. The end-to-end proof lives in
+    /// `tests/delayed.rs::poison_member_does_not_wedge_promotion` (live
+    /// Redis). This cheap unit check makes sure a future edit can't silently
+    /// strip the guard back out of either script.
+    ///
+    /// The guard mirrors `delayed_member::decode_delayed_member`: reject a
+    /// member shorter than the prefix (`#member < 4` / `#member >= 4`) and
+    /// reject a declared name length that overruns the buffer
+    /// (`name_len > #member - 4` / `name_len <= #member - 4`).
+    fn asserts_bounds_guard(script: &str, what: &str) {
+        assert!(
+            script.contains("string.byte(member, 1, 4)"),
+            "{what}: expected the length-prefixed member decode to still exist"
+        );
+        let has_short_guard = script.contains("#member < 4") || script.contains("#member >= 4");
+        assert!(
+            has_short_guard,
+            "{what}: missing the short-member bounds guard (#member vs 4) — \
+             a <4-byte member will abort the EVAL and wedge promotion"
+        );
+        let has_overrun_guard =
+            script.contains("name_len > #member - 4") || script.contains("name_len <= #member - 4");
+        assert!(
+            has_overrun_guard,
+            "{what}: missing the oversized-name_len guard (name_len vs \
+             #member - 4) — a member whose declared name length overruns \
+             the buffer will mis-slice the payload"
+        );
+    }
+
+    #[test]
+    fn promote_script_guards_malformed_members() {
+        asserts_bounds_guard(PROMOTE_SCRIPT, "PROMOTE_SCRIPT");
+    }
+
+    #[test]
+    fn schedule_repeatable_script_guards_malformed_members() {
+        asserts_bounds_guard(SCHEDULE_REPEATABLE_SCRIPT, "SCHEDULE_REPEATABLE_SCRIPT");
+    }
+}

chasquimq/tests/delayed.rs

@@ -798,3 +798,115 @@ async fn promoter_does_not_release_lock_owned_by_other() {
 
     let _: () = admin.quit().await.unwrap();
 }
+
+/// Regression: a malformed delayed-ZSET member must not permanently wedge
+/// promotion. Pre-fix, the promote Lua did `string.byte(member, 1, 4)` with
+/// no bounds check; a member shorter than the 4-byte length prefix made
+/// `string.byte` return nil, the `name_len` arithmetic erred on nil, and the
+/// whole EVAL aborted *before* the per-member `ZREM` — so the poison member
+/// was never removed and every subsequent promote tick re-read it and
+/// re-errored. Delayed-job promotion for the queue was stuck forever, and
+/// valid jobs scored at/after the poison never moved.
+///
+/// This test ZADDs two poison members (a 2-byte runt and one whose declared
+/// name_len overruns the buffer) alongside a real delayed job, runs the
+/// promoter, and asserts: the valid job is promoted and consumed, both poison
+/// members are cleansed from the ZSET, and the ZSET drains to empty (proving
+/// repeated ticks keep working, i.e. no permanent wedge).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[ignore = "requires REDIS_URL"]
+async fn poison_member_does_not_wedge_promotion() {
+    let admin = admin().await;
+    let queue = "delayed_e18_poison";
+    flush_all(&admin, queue).await;
+
+    let dkey = delayed_key(queue);
+    let past_score: i64 = 1; // epoch-ish, always due
+
+    // Poison #1: 2 bytes — shorter than the 4-byte u32_le name-length prefix.
+    let _: Value = admin
+        .custom(
+            CustomCommand::new_static("ZADD", ClusterHash::FirstKey, false),
+            vec![
+                Value::from(dkey.as_str()),
+                Value::from(past_score),
+                Value::Bytes(vec![0xDE, 0xAD].into()),
+            ],
+        )
+        .await
+        .expect("ZADD poison runt");
+
+    // Poison #2: valid-looking 4-byte prefix declaring name_len = 9999, but
+    // the member only has 2 trailing bytes — name_len overruns the buffer.
+    let _: Value = admin
+        .custom(
+            CustomCommand::new_static("ZADD", ClusterHash::FirstKey, false),
+            vec![
+                Value::from(dkey.as_str()),
+                Value::from(past_score),
+                Value::Bytes(vec![0x0F, 0x27, 0x00, 0x00, 0x41, 0x42].into()),
+            ],
+        )
+        .await
+        .expect("ZADD poison oversized name_len");
+
+    assert_eq!(
+        zcard(&admin, &dkey).await,
+        2,
+        "two poison members planted into the delayed ZSET"
+    );
+
+    // A real delayed job alongside the poison, via the normal producer path.
+    let producer: Producer<Sample> = Producer::connect(&redis_url(), producer_cfg(queue))
+        .await
+        .expect("connect producer");
+
+    let counter = Arc::new(AtomicUsize::new(0));
+    let shutdown_consumer = CancellationToken::new();
+    let h_consumer = spawn_consumer(
+        queue,
+        "c1",
+        false,
+        counter.clone(),
+        shutdown_consumer.clone(),
+    );
+
+    let shutdown_promoter = CancellationToken::new();
+    let promoter = Promoter::new(redis_url(), promoter_cfg(queue, "p1"));
+    let h_promoter = tokio::spawn(promoter.run(shutdown_promoter.clone()));
+
+    producer
+        .add_in(Duration::from_millis(50), Sample { n: 7 })
+        .await
+        .expect("add_in");
+
+    // The valid job must still flow through despite the poison members.
+    wait_until(Duration::from_secs(5), || {
+        let counter = counter.clone();
+        async move { counter.load(Ordering::SeqCst) == 1 }
+    })
+    .await;
+
+    // The poison members must be cleansed (ZREM'd) by the promoter, not
+    // wedge it. The ZSET drains fully — proving repeated ticks keep working.
+    let dkey2 = dkey.clone();
+    wait_until(Duration::from_secs(5), || {
+        let admin = admin.clone();
+        let dkey2 = dkey2.clone();
+        async move { zcard(&admin, &dkey2).await == 0 }
+    })
+    .await;
+
+    assert_eq!(
+        counter.load(Ordering::SeqCst),
+        1,
+        "exactly the one valid job should have been consumed (poison members \
+         are dropped, not dispatched)"
+    );
+
+    shutdown_promoter.cancel();
+    let _ = tokio::time::timeout(Duration::from_secs(5), h_promoter).await;
+    shutdown_consumer.cancel();
+    let _ = tokio::time::timeout(Duration::from_secs(5), h_consumer).await;
+    let _: () = admin.quit().await.unwrap();
+}

docs/history.md

@@ -207,6 +207,25 @@ Bench (same-session A/B, `main` `f4a0abd` vs branch, settled host load ~2.5): `q
 
 Doc surfaces synced: this slice, `docs/engine.md` (new "Pause / resume" section + an operational-notes bullet on the no-TTL durable semantics), `reference/options.md` (`pause_poll_ms`), `reference/{node-api,python-api,rust-api,cli}.md`, a new `concepts/pause-and-resume.md` registered in `site/astro.config.mjs`, both shim READMEs (symmetric "Pausing and resuming" sections), and the root `README.md` feature table.
 
+## Hardened delayed-member Lua decode against malformed members
+
+**Date:** 2026-05-19. **Branch:** `fix/harden-delayed-member-lua-decode`. **PR #143.**
+
+Robustness/correctness hardening of the two Lua scripts in `chasquimq/src/redis/commands.rs` that decode the slice-3 length-prefixed delayed-ZSET member (`[u32_le name_len][name utf8][msgpack Job<T>]`): `PROMOTE_SCRIPT` and `SCHEDULE_REPEATABLE_SCRIPT`.
+
+The bug: both scripts decoded with `string.byte(member, 1, 4)` followed by `name_len` arithmetic and **no bounds check**. On a member shorter than the 4-byte prefix, `string.byte` returns `nil` for the missing positions; arithmetic on `nil` aborts the whole EVAL. In `PROMOTE_SCRIPT` that abort happens *before* the per-member `ZREM`, so the poison member is never removed — every subsequent promote tick re-reads it and re-errors, and delayed-job promotion for that queue is wedged forever (also starving valid jobs scored at/after the poison). Reproduced live against Redis 8.6.2. The Rust decoder (`delayed_member::decode_delayed_member`) already guarded both `len < 4` and `len < 4 + name_len`; the Lua side now mirrors that exactly. No public API change, no config change.
+
+Fix (Lua control flow locked by the engineering review — Redis Lua is 5.1: no `continue`, no `goto`, so the skip is expressed as nested `if/else`):
+
+- **`PROMOTE_SCRIPT`**: bounds-check before `string.byte`. A member with `#member < 4`, or whose decoded `name_len > #member - 4` (the exact inverse of the Rust decoder's `member.len() < 4 + name_len`, empty-payload boundary included), is `ZREM`'d to cleanse it and skipped instead of aborting the batch. `payloads` now uses its own dense counter `np` rather than the loop index `i` — a `nil` hole in a Lua array truncates the RESP reply, which would have silently dropped `didx` side-index cleanup for every valid member promoted after a poison entry. `#due` (return slot 1) is unchanged: it is the length of the `ZRANGEBYSCORE` result, a fixed array length, so it still counts the poison member as scanned/removed-this-tick, which is correct.
+- **`SCHEDULE_REPEATABLE_SCRIPT`**: same guard on the fire branch (decode only happens there; future fires `ZADD` verbatim with no decode, so no guard needed). A poison member is dropped — it cannot be dispatched — and explicitly does **not** `HINCRBY fired` / increment `fired_now`; counting it would burn a fire slot against the repeatable `limit` and could prematurely retire the spec. The loop index advances unconditionally so the fire batch never aborts and never infinite-loops; the rest of the batch fires normally. `{fired_now, removed}` semantics stay accurate. Well-formed members hit the byte-identical command sequence as before.
+
+Tests: new live-Redis regression `chasquimq/tests/delayed.rs::poison_member_does_not_wedge_promotion` (ZADDs a 2-byte runt and an oversized-`name_len` member alongside a real delayed job, runs the promoter, asserts the valid job is promoted+consumed, both poison members are cleansed from the ZSET, and the ZSET drains to empty — proving repeated ticks keep working). Plus two in-crate structural guards (`redis::commands::tests`) so a future edit cannot silently strip the bounds check back out of either script. A repeatable-specific poison test was intentionally skipped: the `Scheduler` builds fire-batch members internally from the spec payload, leaving no clean seam to inject a poison member without large scaffolding, and the decode shape is byte-identical to `PROMOTE_SCRIPT`'s (already covered end-to-end). Full workspace suite green: `cargo test --workspace -- --include-ignored` → 291 passed (34 suites) against live Redis 8.6.2; `cargo fmt --all -- --check` and `cargo clippy --all-targets --workspace -- -D warnings` clean.
+
+Bench (same-host A/B, `worker-delayed-end-to-end`, scale=5, repeats=5, drop-slowest=1): `main` (pre-fix) 17,249 jobs/s vs branch ~16.5k–17.4k jobs/s — within run-to-run stddev (300–790). The added per-member bounds checks are O(1) integer comparisons with zero extra Redis round trips or allocations; no measurable regression. (Absolute numbers differ from the phase-2 705k baseline only because of the documented scale/single-host-contention caveat; the same-host A/B against `main` is the valid comparison.)
+
+Doc surface: changelog only. No user-visible API, flag, config field, env var, error code, or wire-format change, so the READMEs and Starlight reference pages stay as-is per the doc-sync table; this entry is the record.
+
 ## Redis Cluster support (cross-FFI slice)
 
 **The engine was already cluster-correct by prior deliberate design; this slice makes that reachable, documented, and proven.** Two invariants were already in place and verified here: (1) every key for a queue carries the `{chasqui:<queue>}` hash tag — confirmed across **all** key families in `redis/keys.rs` (stream, delayed, dlq, result, dedup `dlid:`, cancel side-index `didx:`, repeat + repeat:spec, promoter/scheduler locks, events, paused), so a queue's whole keyspace is single-slot; (2) every command and multi-key Lua script (`PROMOTE`, `RETRY_RESCHEDULE`, `RELOCATE_DLQ`, `REPLAY_DLQ`, `JOB_OK`, `CANCEL_DELAYED`, `SCHEDULE_*`, locks) is dispatched with `ClusterHash::FirstKey`. fred's `Config::from_url` already auto-detects the `redis-cluster://` / `rediss-cluster://` / `valkey-cluster://` / `valkeys-cluster://` schemes and selects `ServerConfig::Clustered` with `CLUSTER SLOTS` discovery + MOVED/ASK handling — **no engine code change, no feature flag, no config field**. `git diff main -- chasquimq/src` is empty.