From 866e2f40c140d122c21e397fba81711d5cc50ed9 Mon Sep 17 00:00:00 2001 From: jotarios Date: Tue, 19 May 2026 00:38:04 -0300 Subject: [PATCH 1/2] fix(redis): guard short/oversized delayed-member decode in Lua MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PROMOTE_SCRIPT and SCHEDULE_REPEATABLE_SCRIPT decoded the slice-3 length-prefixed delayed-ZSET member with string.byte(member, 1, 4) followed by name_len arithmetic and no bounds check. A member shorter than the 4-byte prefix made string.byte return nil; arithmetic on nil aborted the whole EVAL *before* the per-member ZREM, so the poison member was never removed and every promote tick re-read it and re-errored — delayed-job promotion for the queue was permanently wedged, also starving valid jobs scored at/after the poison. Mirror the existing Rust decoder (delayed_member::decode_delayed_member) in Lua: reject members shorter than the prefix and reject a declared name_len that overruns the buffer. In PROMOTE_SCRIPT a malformed member is ZREM'd (cleansed) and skipped; payloads uses its own dense counter so a skipped member can't punch a nil hole that truncates the reply for valid members after it. In SCHEDULE_REPEATABLE_SCRIPT a malformed member on the fire branch is dropped without counting toward fired/fired_now (it can't be dispatched, so it must not burn a fire slot against the repeatable limit); the loop still advances and the rest of the batch fires normally. Well-formed members hit the identical command sequence as before — no behavior change, no API/config change. Tests: new live-Redis regression poison_member_does_not_wedge_promotion (valid job still promoted, poison cleansed, ZSET drains, no wedge) plus in-crate structural guards so a future edit can't strip the check back out. Existing delayed/promoter/repeatable suites unchanged and green. --- chasquimq/src/redis/commands.rs | 133 +++++++++++++++++++++++++++----- chasquimq/tests/delayed.rs | 112 +++++++++++++++++++++++++++ 2 files changed, 226 insertions(+), 19 deletions(-) diff --git a/chasquimq/src/redis/commands.rs b/chasquimq/src/redis/commands.rs index 9941a46..f9da0ae 100644 --- a/chasquimq/src/redis/commands.rs +++ b/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"); + } +} diff --git a/chasquimq/tests/delayed.rs b/chasquimq/tests/delayed.rs index 2fcef40..e7ee4b5 100644 --- a/chasquimq/tests/delayed.rs +++ b/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 = 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(); +} From 28cfb29b2c165292b0e075fb069566c4e7edb693 Mon Sep 17 00:00:00 2001 From: jotarios Date: Tue, 19 May 2026 00:46:36 -0300 Subject: [PATCH 2/2] docs(history): record delayed-member Lua decode hardening (PR #143) --- docs/history.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/docs/history.md b/docs/history.md index 1a81d8b..29c7682 100644 --- a/docs/history.md +++ b/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]`): `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:}` 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.