feat: persistent job progress + logs#148
Merged
Merged
Conversation
Add `progress_key(queue, job_id)` returning `{chasqui:<q>}:progress:<id>`
and `log_key(queue, job_id)` returning `{chasqui:<q>}:log:<id>`. Mirrors
the existing per-queue key style and uses the same `{chasqui:<queue>}`
hash tag so the new keys co-locate on a single Redis Cluster slot with
the stream, delayed ZSET, and result key.
Re-exported via `producer` alongside the existing key helpers so
external admin tooling (CLI, observability) can construct the keys
without poking into a private module.
Unit tests pin the hash-tag invariant — a future helper that forgets
the `{chasqui:...}` braces would silently shard a queue's keyspace
across Cluster slots and break the engine's multi-key Lua.
Add `JobHandle`: the per-handler progress + log surface attached to `Job<T>::handle` by the consumer immediately before the user handler runs. Two methods: - `update_progress(n)` — clamp `n` to `0..=100`, SET the per-job progress key with TTL `result_ttl_secs` (plain ASCII decimal string so shims can read with `parseInt` / `int(str(...))` without a msgpack dependency). Warn-once per handle for out-of-range values so a hot-loop handler doesn't flood the operator log. - `log(line)` — XADD one entry under field `line` with `MAXLEN ~ <log_max_stream_len>`, pipelined with an XLEN that the call returns. Lines exceeding `log_max_line_bytes` are truncated on a UTF-8 char boundary with a `[…truncated]` marker appended; warn-once per handle on first truncation. `Job<T>::handle: Option<Arc<JobHandle>>` is `#[serde(skip)]`, identical to the load-bearing pattern on `Job::name`: the handle is a runtime-only Redis-connection backref, never part of the on-wire payload. `None` on read-only / synthesized `Job` instances. Connection budget is borrowed via a shared `fred::clients::Pool` rather than opening a client per worker; the consumer wires the pool in (next commit). Event emission on progress write is pre-plumbed but unwired in this commit — `EventsWriter::emit_progress` lands next. Integration tests against a real Redis pin: progress writes the key with `EX`, clamps `>100` to 100, log appends in order and returns XLEN, oversize lines truncate on a UTF-8 boundary with the marker, and empty lines still bump XLEN.
Add `EventsWriter::emit_progress(id, name, progress)`. Same shape as `emit_active` / `emit_completed`: `e=progress`, `id=<id>`, `progress=<u8 as decimal string>`, `ts=<unix_ms>`, plus `n=<name>` when the job has a dispatch name (the field is omitted on the wire for unnamed jobs — matches the existing convention). Wire it into `JobHandle::update_progress` behind `events_progress_enabled`. SET-first-then-emit ordering: the persisted progress key is the source of truth, so a best-effort event-emit failure can never leave subscribers ahead of (or behind) what an introspector would read. Tests pin: two emit calls produce two XRANGE-visible `progress` events with the right `progress` / `n` / `ts` fields, and a disabled writer short-circuits the XADD entirely.
Open a small shared `Pool` (`(concurrency / 8).max(2).min(8)`) at
`Consumer::run` and thread it through `WorkerWiring` so every dispatched
job gets a fresh `JobHandle` with the same pool — no client-per-worker.
The handle is attached to `dispatched.job.handle` *before* the user
handler runs and the worker emits the `active` event, so a handler that
calls `update_progress` mid-execution writes to Redis through the
shared pool without contending with the reader / ack / retry hot paths.
The retry path's cloned `Job` snapshot clears `handle = None` so a
retried delivery picks up a fresh per-attempt handle (it would in any
case, since the consumer rebuilds one for the next dispatch — explicit
here for clarity).
Pre-bake `log_max_stream_len = 1000`, `log_max_line_bytes = 4096`, and
`events_progress_enabled = true` directly in `WorkerWiring`; the
follow-up config slice swaps these for fields on `ConsumerConfig`.
Integration test: a handler that calls `h.update_progress(50)` produces
`{chasqui:<q>}:progress:<id> = "50"` after the worker completes.
Add `JobInfo::progress: Option<u8>` populated by `get_job` across all five state branches (Waiting / Active / Delayed / Failed / Completed). The progress GET is issued from the same `Pool` connection as the matched branch's last probe so it adds no sequential RTT on the admin path. Pagination paths (`get_jobs`) leave `progress = None` — they fetch many entries and the per-id key would explode the round-trip count; callers that need per-job progress in a list should follow up with `get_job(id)`. The introspector tolerates forward-compat shape skew at the boundary: non-decimal payloads and values outside `u8` range collapse to `None` rather than erroring. Values that parse as `u8` but exceed 100 clamp to 100, mirroring the engine's write-side clamp. Add `Introspector::get_job_logs(id, start, end, asc) -> (Vec<String>, u64)`: - `start` / `end` are inclusive entry offsets in the requested order. - Negative `start` is "this many from the end" (translated via XLEN), matching BullMQ's `getLogs` convention. - `end = -1` means "to end". - `asc` selects XRANGE vs XREVRANGE. - Returns the captured `line` field values plus the current XLEN. - Empty stream returns `(vec![], 0)`. Tests pin: progress surfaces in JobInfo after a handle write, the field is `None` when never written, asc / desc / window / negative- start pagination behave correctly, and an empty log returns `(empty, 0)`. Unit tests pin the progress decoder against shape skew (clamp `>100`, reject overflow, accept bytes/integer forms).
`Producer::remove(id, group)` already DEL'd the result key as the final
step; extend it to also UNLINK the per-job progress key and the per-job
log stream in the same pipelined round trip. UNLINK is async-reclaim so
a multi-MB log stream never stalls the call; the result key stays on
DEL so the `RemovalReport::result` field's "the result key was removed"
semantic is preserved (the multi-key UNLINK count couldn't be
attributed back to a single key).
All three keys share the `{chasqui:<queue>}` hash tag, so the pipeline
lands on a single Cluster slot.
`obliterate` was already correct via `SCAN MATCH {chasqui:<q>}:*`
(progress and log keys carry the queue tag), but pin that with a test
so a future change to the key shape or to obliterate's pattern can't
silently regress.
…ss_enabled Add three fields to `ConsumerConfig`: - `log_max_stream_len: u64` (default `1000`) — `MAXLEN ~` cap applied by `JobHandle::log` to every per-job log stream. - `log_max_line_bytes: usize` (default `4096`) — per-line byte cap; oversize lines are truncated on a UTF-8 boundary with the `"[…truncated]"` marker. - `events_progress_enabled: bool` (default `true`) — gates only the `e=progress` events-stream `XADD`; the persisted progress key is always written so a handler that opts out of the events flood still surfaces progress via the introspector. `ConsumerConfig::validate()` rejects `log_max_stream_len < 16` with `Error::Config`; `Consumer::run` calls it at startup so the misconfig surfaces immediately rather than as a silent data-loss bug after the first `JobHandle::log` call (anything below the minimum and Redis's `MAXLEN ~` rounding can leave the stream effectively empty between writes). Wire the three fields through `WorkerWiring` so per-dispatch `JobHandle` construction reads them — replaces the temporary hard-coded defaults plumbed in when the worker first started attaching handles. Unit tests pin the defaults, the at-minimum validate-OK boundary, and the below-minimum reject.
Pure rustfmt — no behavioral changes. The earlier commits in this slice introduced new code that wasn't rustfmt-clean; this commit brings the slice up to the CI gate's `cargo fmt --all -- --check`.
Surface the engine's per-handler progress + log API on the Node shim:
- NAPI `Job` becomes a `#[napi]` class carrying the engine's
`Arc<JobHandle>` opaquely across the FFI boundary. `updateProgress(n)`
and `log(line)` are async instance methods that call into the handle
attached by the consumer's worker dispatch. Getters keep the existing
`(job.id, job.name, job.payload, job.createdAtMs, job.attempt)` JS
access shape, so high-level shim consumers do not need to change.
- NAPI `JobInfo` gains optional `progress: number` populated by
`getJob` (the inspector pipelines a GET on the per-job progress key
into the matched-state probe). `Introspector.getJobLogs(id, start?,
end?, asc?) -> { logs, count }` wraps the engine's bounded XRANGE
path; pagination + negative-`start` semantics match BullMQ.
- NAPI `ConsumerOpts` plumbs the three new engine knobs:
`logMaxLen`, `logMaxLineBytes`, `eventsProgressEnabled`.
- TypeScript `Job.updateProgress`/`log` call the native bridge and
guard read-only Jobs (those returned by `Queue.getJob`/`getJobs`
have no live handle) with a clear `Job.updateProgress() requires
the Job be passed to your Worker handler` error.
- `Queue.getJobLogs(jobId, start=0, end=-1, asc=true)` exposes the
introspector's log read; `Queue.getJob`/`getJobs` thread the new
`progress` field through `nativeInfoToJob`.
- `QueueEvents` dispatches a new `progress` event branch onto both
the broadcast channel and the per-id `progress:<jobId>` channel.
- `Worker` lazily spawns a single internal `QueueEvents` subscriber
on the first `on('drained' | 'progress', ...)` (was `drained`
only); the `progress` forwarder looks the live `Job` up by id in
the `inflight` map populated for the duration of each processor
invocation and re-emits as `(job, progress)`, matching BullMQ's
shape. `WorkerOptions` gains `logMaxLen`, `logMaxLineBytes`,
`eventsProgressEnabled` plumbed through to `NativeConsumerOpts`.
Vitest covers: in-handler `updateProgress` persists and surfaces via
`Queue.getJob().progress`; `QueueEvents` per-id + broadcast
`progress:<id>` payload; Worker EE `progress` event fires once per
call with `(job, n)`; `Queue.getJobLogs` round-trip + 10-line
pagination; oversize-progress clamp to 100; read-only Job guard
throws on `Queue.getJob().updateProgress()` and `.log()`.
BREAKING CHANGE: `JobProgress` TypeScript type narrows from
`number | object` to `number`. The engine stores progress as a Redis
STRING (ASCII decimal `u8`, clamped to `0..=100`); the previous
`object` arm was a non-functional carryover from the BullMQ-shaped
type. Numeric-progress callers are unaffected; object-progress
callers must encode their state separately (e.g. via `Job.log`).
Surface the engine's per-handler progress + log API on the Python shim, mirroring the Node slice (70516be): - PyO3 `_Job` carries the engine's `Arc<JobHandle>` opaquely across the FFI boundary. `update_progress(n)` and `log(line)` are async instance methods that delegate to the handle attached by the consumer's worker dispatch. Jobs without a backref (none today on the PyO3 surface) raise a clear "read-only Job" `RuntimeError`. - PyO3 `Introspector.get_job_logs(id, start, end, asc) -> (list[str], int)` wraps the engine's bounded XRANGE / XREVRANGE path; negative `start` and `end = -1` semantics match BullMQ. `JobInfo` dict gains `progress: Optional[int]` populated by `get_job` (pipelined into the matched-state probe). - PyO3 `Consumer` plumbs three new engine knobs through the constructor: `log_max_stream_len`, `log_max_line_bytes`, `events_progress_enabled`. - High-level `Job` dataclass gains `progress: Optional[int]` plus `async update_progress(n)` / `async log(line) -> int`. Read-only guard mirrors the Node shim's message; `_handle` is set by the Worker's native bridge before the user handler sees the Job. - `Queue.get_job_logs(job_id, *, start=0, end=-1, asc=True)` exposes the introspector's log read; `Queue.get_job` / `Queue.get_jobs` thread the new `progress` field through `_native_info_to_job`. - `QueueEvents` adds `progress` to the per-id event set and `progress` to the numeric-coerced field set, so `events.on('progress:<id>', cb)` and `events.on('progress', cb)` receive `{ jobId, name, progress: int }` payloads with the same shape as the Node shim. - `Worker` lazily spawns a single internal `QueueEvents` subscriber on the first `on('drained' | 'progress', ...)` (was `drained` only); the `progress` forwarder looks the live `Job` up by id in the `_inflight` map populated for the duration of each handler invocation and re-emits as `(job, n)`. `WorkerOptions` gains `log_max_stream_len`, `log_max_line_bytes`, `events_progress_enabled` plumbed through to the native consumer kwargs. Pytest covers: in-handler `update_progress` persists and surfaces via `Queue.get_job().progress`; `QueueEvents` per-id + broadcast `progress:<id>` payload; Worker EE `progress` event fires once per call with `(job, n)`; `Queue.get_job_logs` round-trip + 10-line pagination; out-of-range `end` clamps without exception; read-only Job guard raises on `Queue.get_job().update_progress()` and `.log()`.
Three sibling scenarios — progress-throughput-1, progress-throughput-10,
progress-throughput-100 — measure the cost of JobHandle::update_progress
called K∈{1,10,100} times per job from inside the handler. Concurrency
matches worker-concurrent (100) so headline jobs/sec is apples-to-apples
with the no-progress baseline. Handler does nothing else; progress
values cycle 0..=100 to keep the engine clamp off the hot path.
New CLI flag --progress-events-enabled (default false) toggles the
best-effort e=progress event XADD so users can isolate the SET+TTL
write cost from the events-stream pressure.
Add bidirectional cross-shim wire-contract tests for the per-handler
progress + log API. Phase 9 enqueues from Node, runs a Python worker
that calls `update_progress(75)` and appends two log lines, then reads
back via the Node `Queue.getJob(id).progress` + `Queue.getJobLogs(id)`
introspector surface. Phase 10 is the symmetric Python-produces /
Node-consumes / Python-verifies direction.
Pinning a numeric progress value (75) plus a fixed two-line log
sequence exercises both pieces of the new wire format in one round
trip: the ASCII-decimal `u8` STRING at `{chasqui:<q>}:progress:<id>`
and the bounded log Stream at `{chasqui:<q>}:log:<id>`. A regression
that drops/mistypes either surface on either shim surfaces here.
Mechanics reuse the existing fixture pattern:
- `worker.{py,ts}` takes a new `EMIT_PROGRESS=1` env var that triggers
the handler-side progress + log calls (pairs with the existing
`STORE_RESULT=1` so the completed jobs stay discoverable by the
introspector's result-key probe).
- New `verify_progress.{py,ts}` mirror the existing `verify_results.*`
shape: read ids from `JOB_IDS_FILE`, poll `getJob` / `getJobLogs`
for each, assert deep-equality with `EXPECT_PROGRESS` +
`EXPECT_LOGS`.
`COUNT=5` per phase keeps the new scenarios cheap; the matrix already
runs eight existing phases.
JobHandle::log() pipelined XADD + XLEN with a MAXLEN cap on the entry count but never set an EXPIRE on the stream key itself. A job that logged at least once would leave the per-job log stream in Redis indefinitely — only obliterate(queue) or remove(job_id) would reap it. Add EXPIRE log_key result_ttl_secs as a third pipelined command (same round trip, no perf regression) so the log key disappears alongside the result key. Update the docstring that incorrectly claimed TTL behavior. Add an integration test (log_key_has_ttl_matching_result_ttl_secs) asserting PTTL is bounded by result_ttl_secs * 1000 after a single log() call.
remove(id) was extended to UNLINK the per-job progress key and log
stream alongside the primary surfaces, but the bulk clean() paths
(clean_stream, clean_delayed, clean_completed) were missed. Every
job clean() drops from the stream / delayed ZSET / DLQ / result-key
set would leave its progress key + log stream behind, only reaped
by obliterate(queue).
Introduce unlink_progress_and_log(pool, queue_name, &job_ids) that
mirrors the UNLINK pipeline tail from remove(id) and call it from
clean() after the per-state helper returns. One round trip per
batch; the keys share the {chasqui:<queue>} hash tag so the
pipeline is Cluster-correct. Best-effort — a Redis error warns
but does not revert the primary cleanup the caller already did.
Add clean_purges_progress_and_log_keys: plants progress + log keys
for three waiting jobs, calls clean(Waiting), asserts both per-job
key types are gone for every removed id.
Introspector::get_job_logs translated negative start / end indices against the current XLEN using plain integer addition. A caller passing i64::MIN would panic in debug builds (`attempt to add with overflow`) and silently wrap in release. Switch both branches to saturating_add so an extreme negative offset simply clamps to the window edge. Add get_job_logs_handles_extreme_negative_offsets covering both sides — i64::MIN as start (clamps to 0, returns every entry) and i64::MIN as end (saturates below 0, returns an empty window) — proving no panic on either path.
5 repeats × scale=5, drop-slowest. Engine *was* touched (JobHandle attached per dispatch + progress field on JobInfo), so the host-load gate forfeits the contention explanation; every delta vs the 1.0 baseline lands inside one baseline stddev: queue-add-bulk 184,295 jobs/s (-2.4%, stddev 4,909) worker-concurrent 110,726 jobs/s (-1.1%, stddev 4,246) queue-add 16,340 jobs/s (+6.3%, stddev 846) worker-generic 9,810 jobs/s (+3.1%, noisy / direction-only) Host: contended Mac (load avg 14-16, vs 1.8-4.3 on baseline); redis:8.6.2 on loopback. JobHandle is per-dispatch but carries only Arc<str> clones + a shared pool reference (zero per-job allocation, zero extra round trip when the handler never calls update_progress / log).
Sync every doc surface touched by the persistent-progress + per-job-
log slice.
- README.md: feature comparison rows for persistent progress and
per-job log stream.
- docs/engine.md: new 'Progress and logs' section (Redis keys,
TTL contract, three new ConsumerConfig fields, shared pool
budget, read-only Job guard, progress event); extend the
maintenance section to note remove/clean/obliterate reap the
new keys.
- docs/history.md: new slice entry covering what shipped, key
design decisions, the three post-implementation bug fixes
(log key TTL, clean() leak, i64::MIN saturating arithmetic),
and the no-regression bench result.
- chasquimq-node/README.md, chasquimq-py/README.md: symmetric
'Progress and logs' sections; flag the JobProgress TS-types
narrowing as the only breaking change.
- site/src/content/docs/reference/node-api.md: rewrite
Job.updateProgress (was a no-op stub); add Job.log,
Queue.getJobLogs; extend WorkerOptions with logMaxLen /
logMaxLineBytes / eventsProgressEnabled; add progress + per-id
progress:<jobId> to QueueEvents events; narrow JobProgress
type.
- site/src/content/docs/reference/python-api.md: same on the
Python side (Job.update_progress, Job.log, Queue.get_job_logs,
Worker kwargs, Worker + QueueEvents progress event).
- site/src/content/docs/reference/rust-api.md: add JobHandle to
Job types; document Introspector::get_job_logs and the new
JobInfo.progress field; update Job<T> to show the handle
field.
- site/src/content/docs/reference/options.md: new 'Progress and
logs' section for the three ConsumerConfig fields.
- site/src/content/docs/reference/wire-format.md: add the two
new keys (progress STRING, log STREAM) and the progress
events-stream field.
- site/src/content/docs/concepts/events-and-listeners.md:
promote progress from no-op to live; document the
events_progress_enabled toggle for high-rate handlers;
extend the lazy-subscriber-lifecycle note to cover progress.
No site/astro.config.mjs change — reference + concepts sidebars
are autogenerated; no new pages were added.
jotarios
added a commit
that referenced
this pull request
May 26, 2026
…e tests Two tests in progress-and-log.test.ts rely on a `waitFor(..., 10_000)` to observe per-id and Worker EE progress events. On a contended GitHub Actions runner, the spawned QueueEvents subscriber's connect+block latency plus the 3-event delivery window can exceed 10s, surfacing as a `waitFor timed out` failure even when the engine emitted the events promptly. Local repro: 18 consecutive runs all pass. CI: failed on attempt 1 of run 26425401120 (the run that includes the max_stalled_attempts=2 default bump). The test path is byte-identical to the PR #148 lazy-spawn code; the slice's worker.ts diff is purely additive (adds `stalled` to the newListener filter and a new forwarder; the `progress` forwarder is untouched). Bump both `waitFor(..., 10_000)` calls to 25_000 and the matching per-it timeouts from 15s to 30s. Headroom for slow runners; local pass time (~1s) is unaffected.
jotarios
added a commit
that referenced
this pull request
Jun 1, 2026
…#149) * feat(engine): stalled-job detector with DlqReason::Stalled Adds a leader-elected background task (`StalledDetector`) spawned alongside the existing promoter / scheduler. Every tick, the leader runs `XPENDING ... IDLE - + <batch>` against the consumer group's PEL, pipelined-XRANGEs each entry's envelope to recover `(job_id, name)`, then invokes `STALLED_SCAN_SCRIPT` which atomically: - INCRs each entry's per-job stall counter (`{chasqui:<queue>}:stalls:<job_id>`). - EXPIREs the counter at `idle_threshold_ms * max_stalled_attempts * 2` (sliding TTL — counter survives between ticks for genuinely stalled jobs). - If `n < max_stalled_attempts`: records the INCR for `e=stalled`. - Else: gates the entry out of the PEL via XACKDEL. On gate-held, Rust sends to the existing `dlq_tx` channel with the new `DlqReason::Stalled` variant. The relocator does the IDMP XADD + `e=dlq` emit. The DLQ wire shape is unchanged. The reader's CLAIM-on-read path is untouched — the detector never issues `XCLAIM`, never bumps `delivery_count`. Letting one subsystem own redelivery and another own stall observability is the key race isolation. `JOB_OK_SCRIPT` and `REPLAY_DLQ_SCRIPT` both DEL the per-job stall counter in the same Lua call so a one-off stall followed by success / replay starts a fresh streak. `REPLAY_DLQ_SCRIPT`'s ARGV layout widened from triples (id, payload, name) to quads (+ job_id) — single-deploy atomic upgrade per the engine-script contract. New / extended config: - `StalledDetectorConfig` mirrors the `SchedulerConfig` shape. Defaults: tick=30s, idle=30s, max_stalled_attempts=1 (matches BullMQ's `maxStalledCount`), scan_batch=256, lock_ttl=5s. - `ConsumerConfig::stalled_detector_enabled` (default true) and `ConsumerConfig::stalled_detector` for embedded use; the embedded spawn overrides `tick_interval_ms` + `idle_threshold_ms` from `claim_min_idle_ms` so the per-crash counting invariant (`tick == idle == claim_min_idle`) is automatic. Validation rejects: - `max_stalled_attempts == 0` (would relocate every observed entry on first scan). - `scan_batch == 0`. - `tick_interval_ms < idle_threshold_ms` (faster ticks break per-crash counting — 5s tick on 30s idle INCRs 6x per crash). `JobInfo.stalled_count: Option<u32>` populated only for Active-state lookups (the counter is `DEL`'d on terminal transitions, so probing elsewhere always returns nil). Observability: - `DlqReason::Stalled` (wire string `"stalled"`). - `MetricsSink::stalled_tick(StalledTick { scanned, incremented, relocated })` — default-noop method, backwards-compatible. - `e=stalled` event on the events stream (`id`, `attempt`, `prev=active`, `n` opt). New `leader_task` module extracts the shared script-load / transient-classifier / value-coercion helpers the new detector needs; promoter and scheduler retain their own copies (zero-diff this slice — consolidation is a follow-up refactor). * feat(engine): pre-acked DLQ relocate + stalled integration tests The stalled detector's STALLED_SCAN_SCRIPT XACKDEL's the source entry out of the PEL at the threshold-hit branch so the relocate decision is atomic with the counter INCR. Without a matching change to the relocator, RELOCATE_DLQ_SCRIPT's own XACKDEL gate then returned 0 ("gate lost") and silently skipped the DLQ XADD — the entry was acked but never written to the DLQ. Fix: add `DlqRelocate::pre_acked` and a sibling RELOCATE_DLQ_PRE_ACKED_SCRIPT that skips the XACKDEL gate and goes straight to the IDMP-XADD path (the IDMP marker is the only dedup guard needed on this path; the upstream XACKDEL already excluded concurrent CLAIM / manual-ack from racing). Lazy SCRIPT LOAD on first pre-acked entry so the common (retry-exhausted-only) path pays no extra startup round trip. Integration tests cover: - `detector_relocates_after_max_stalled_attempts` — produce → hang → relocate as `Stalled` (embedded path, ~10s wall clock). - `detector_no_false_positives_under_healthy_load` — handler completes well under `idle_threshold_ms`; zero INCR, zero relocate. - `detector_disabled_skips_spawn` — `stalled_detector_enabled = false` means the lock key never appears in Redis and no `stalled_tick` ever fires. - `job_ok_script_acks_without_stall_counter` — common path (no counter ever written) still acks fine; DEL of a non-existent counter is not an error. - `replay_dlq_quad_argv_shape_works` — regression for the slice-12 ARGV-shape change on REPLAY_DLQ_SCRIPT. - `stalled_event_payload_shape` — wire fields: `e=stalled`, `id`, `attempt`, `prev=active`, `ts`. Promotes `DlqRelocate` / `DlqRelocatorConfig` / `run_relocator` to `pub` so the standalone-detector path (and integration tests) can wire the relocator without reaching into pub(crate) internals. * feat(node)!: stalled-detector wiring + maxStalledCount routing fix BREAKING CHANGE: `WorkerOptions.maxStalledCount` now routes to engine `max_stalled_attempts` (the slice-12 stalled-detector ceiling — stall cycles before DLQ-as-stalled), not `max_attempts` (total handler attempts before DLQ-as-retries_exhausted). Pre-v1.4 the shim mapped this field to `max_attempts` with a `?? 3` fallback that masked the engine's real default of `25`; users who set `maxStalledCount` on its own will now see different behavior. Migration: replace `maxStalledCount: N` with `maxAttempts: N` to preserve the old "cap total attempts at N" semantic. A one-time `WARN [chasquimq]` log fires per `(queueName)` per process when `maxStalledCount` is set without `maxAttempts`, so upgraders see the change loudly at runtime. New surfaces: - `WorkerOptions.maxAttempts` — canonical name for total handler attempts before DLQ-as-retries_exhausted. Routes to engine `max_attempts`. Undefined now reaches the engine literally so the default (25) applies. - `WorkerOptions.stalledDetectorEnabled` — toggle the embedded detector. Default `true`. - `WorkerOptions.stalledInterval` — override the detector tick (ms). Default inherits from the engine's `claim_min_idle_ms`. - `Worker.on('stalled', (jobId, prev) => ...)` — fires when the detector INCRs the per-job counter under threshold. `prev` is always `'active'` (every stalled entry was PEL-resident when the detector saw it). Cross-process scope: every worker on the queue receives the event, not just the one holding the entry. Lazily spawns the internal QueueEvents subscriber on first attach. - `QueueEvents` parses the `e=stalled` entry and emits both the broadcast and per-id channels. NAPI: adds `maxStalledAttempts`, `stalledDetectorEnabled`, `stalledDetectorTickMs`, `stalledDetectorIdleThresholdMs`, and `stalledDetectorScanBatch` to `ConsumerOpts`. Tests: - New `worker-stalled.test.ts` pins the wiring (listener attach, warn-once dedup, routing fix proof). - Existing tests that used `maxStalledCount` as a retry cap migrate to `maxAttempts` (the new canonical name). * feat(py): stalled-detector wiring + max_stalled_attempts field Adds the Python-shim surface for the slice-12 stalled-job detector. No deprecation needed (unlike the Node shim) — the Python `Worker(max_attempts=...)` field already routed correctly to engine `max_attempts`, so no pre-slice user could have been mis-using a mis-named field. New `Worker(...)` keyword args: - `max_stalled_attempts: Optional[int]` — routes to engine `stalled_detector.max_stalled_attempts`. Default `1` (matches the engine default). - `stalled_detector_enabled: bool = True` — toggle the embedded detector. - `stalled_interval_ms: Optional[int]` — override the detector tick. (The embedded spawn overrides this from `claim_min_idle_ms` to preserve the per-crash counting invariant; setting this only matters in the rarely-used standalone path.) - `stalled_detector_scan_batch: Optional[int]` — XPENDING ... IDLE cap per tick. New event listener: - `worker.on('stalled', cb)` — callback receives `(job_id: str, prev: str)`. `prev` is always `'active'`. Cross-process scope (the detector is leader-elected per queue, so any worker on the queue receives the event regardless of which worker held the stalled entry). Lazily spawns the internal `QueueEvents` subscriber on first attach, mirroring the `drained` / `progress` shape. `QueueEvents`: `stalled` joins the per-id-channels set so a future `Job.wait_until_stalled` (or equivalent) can target a single job without paying the broadcast dispatch cost. PyO3 bridge: extends the `Consumer.__init__` signature with the new keyword args; updates `_native.pyi` to match. Test coverage in `tests/test_stalled_detector.py`: - `test_max_stalled_attempts_routes_to_stalled_detector` — pins that the new field routes to the right engine knob (an always-failing handler respects `max_attempts`, not `max_stalled_attempts`). - `test_stalled_listener_wires_up_lazily` — `worker.on('stalled', cb)` spawns the internal subscriber without crashing. - `test_stalled_detector_disabled_via_constructor` — confirms the detector lock key never appears when `stalled_detector_enabled=False`. * docs: stalled-job detector + maxStalledCount routing fix - `docs/engine.md`: new "Stalled-job detector" section + extends the DLQ-tooling reason enum to include `stalled`. - `docs/history.md`: slice changelog entry (architecture, engine/Node/Python changes, breaking-change call-out, tests). - `chasquimq-node/README.md` + `chasquimq-py/README.md`: symmetric "Stalled-job detection" sections with the BullMQ-shaped event payload + tuning table; Node call-out for the v1.4 routing fix. - `site/src/content/docs/concepts/dlq-and-recovery.md`: adds `stalled` to the DLQ reason table + a new "Stalled jobs" subsection explaining the distinction between `retries_exhausted` (handler-failure loops) and `stalled` (worker-crash loops). - `site/src/content/docs/concepts/events-and-listeners.md`: promotes `stalled` out of the no-op list, documents the `(jobId, prev='active')` payload + cross-process scope, adds `stalled` to the per-id-channel set. - `site/src/content/docs/reference/wire-format.md`: new `{chasqui:<q>}:stalls:<jobId>` and `{chasqui:<q>}:stalled:lock` keys + `e=stalled` event row + `stalled` DLQ reason + pre-acked relocate path note. - `site/src/content/docs/reference/options.md`: new "Stalled-job detection" section; calls out the `maxStalledCount` routing fix on the retries-and-backoff row. - `site/src/content/docs/reference/node-api.md`: extends `WorkerOptions` with `maxAttempts`, `stalledDetectorEnabled`, `stalledInterval`; documents the `maxStalledCount` breaking change inline; promotes `stalled` to the live-events table. - `site/src/content/docs/reference/python-api.md`: extends `Worker(...)` with `max_stalled_attempts`, `stalled_detector_enabled`, `stalled_interval_ms`, `stalled_detector_scan_batch`; promotes `stalled` to the live-events table. - `CLAUDE.md`: adds the slice to the orientation timeline. * chore(engine): satisfy clippy on slice-12 additions - Reflow `relocate_once` doc comment so the bullet wraps within the rust-1.95 lazy-continuation lint's tolerance. - Switch `std::io::Error::new(ErrorKind::Other, "poison")` to `std::io::Error::other("poison")` in the replay-quad test. - Annotate the standalone `seed_pending_entry` test helper with `#[allow(dead_code)]` — it's reserved for the standalone-detector test variant deferred to a follow-up slice. * fix(engine)!: default ConsumerConfig.max_attempts to 25 for FFI parity The engine default was `3`, but the Python kwarg default is `25` and the Node shim docs / deprecation-warning copy describe the engine default as `25`. Result: a Node user who relied on "no maxAttempts → engine default" got `3` retries, a Python user got `25`, and a Rust user got `3` despite the v1.4 site copy advertising parity. Bump the engine to `25` so all three surfaces match. Update three site pages that previously claimed `3`. Existing Node/Python tests that set `max_attempts` explicitly are unaffected; engine tests that pinned the value via the default (none found) would also be unaffected. BREAKING CHANGE: `ConsumerConfig::default().max_attempts` is now `25` (was `3`). Rust users who relied on the implicit `3` must set `max_attempts: 3` explicitly. * fix(node,py): plumb stalled_count through shim Job surface The engine ships `JobInfo.stalled_count` as an Active-only probe (see `chasquimq/src/introspect.rs`) and both shim READMEs document the field on the `Job` value class. But the FFI bridges and `Job` classes never carried it through — `Queue.getJob().stalledCount` was always `undefined` on Node, `Queue.get_job().stalled_count` always `None` on Python. A documented field that silently never populates is worse than a missing field. Plumb end-to-end: - Node: add `stalled_count: Option<u32>` to napi `JobInfo`, set on `engine_info_into_napi`, expose `stalledCount?: number` on the `Job` class (with the same Active-only doc), copy through in the `Queue` synthesis path, include in `toJSON`. - Python: emit `stalled_count` in the introspector dict, add to the `Job` dataclass (excluded from repr/eq to mirror `progress`), copy through in `Queue._native_info_to_job`. - New test `test_job_stalled_count_active_only.py`: pins None for Waiting jobs and a positive integer for an Active hung job once the embedded detector has ticked. * test(engine): assert stall-counter DEL on ack and replay The two existing slice-12 tests asserted the scripts succeed but never asserted the counter was actually DEL'd — a regression that silently left the counter behind would still have passed. Strengthen both: - `job_ok_script_acks_without_stall_counter`: have the handler block until the test seeds a stall counter under the in-flight job's id, then release the handler. After the JOB_OK_SCRIPT ack, EXISTS on the stall key must return 0. The original "DEL of a missing key must not error" invariant is preserved by the other 4 jobs that ack with no counter ever written. - `replay_dlq_quad_argv_shape_works`: capture the dlq'd job id in the handler, manually INCR a stall counter under that id (no live detector to seed it naturally because the test runs with stalled_detector_enabled=false), call replay_dlq, then EXISTS on the stall key must return 0. Both invariants are load-bearing: a replayed entry that inherits a sliding-TTL counter from its previous incarnation would trip threshold on the next detector tick. * fix(engine): bump stalled detector lock_ttl to outlive tick interval Default `tick_interval_ms` is 30s but default `lock_ttl_secs` was 5s. The detector sleeps for one full tick between scans, so the sleeping leader's lock expired 25 seconds before wakeup — every tick a replica took the lock, then the original leader reacquired on wake. Result: leader thrashing in any multi-replica deployment. Bump `StalledDetectorConfig::lock_ttl_secs` default to `90` (3× the default tick) so the sleeping leader stays leader. Promoter (tick 100ms, lock 5s) and scheduler (tick 1s, lock 5s) defaults already satisfy the `lock_ttl >> tick` invariant — this slice's detector was the inverted case. Document the relationship on the field doc + the engine.md section. * fix(node): warn-once on maxStalledCount is process-scoped, not per-queue The slice-12 dedup `Set` was keyed on `queueName`, so a multi-queue deployment that set `maxStalledCount` on each Worker would print N WARN lines — one per queue. The doc copy promised "fires once per process"; reality was once per queue. - Drop the per-queue `Set` and switch to a single process-global latch. Stash it on `globalThis` under a unique `Symbol.for` key so multiple loaded copies of `worker.ts` (which happens under Vitest's isolated test runner) still share one latch — matching the "once per process" semantics a real user app gets for free via the single-instance CommonJS module cache. - Tweak the message wording: include the first triggering queue name as "first seen on queue ..." (since it's no longer the only one) and call out "fires once per process" so a multi-queue operator reading logs knows missing copies are by design. - Add a test-only `__resetMaxStalledCountWarnLatchForTests()` export (underscore-prefixed so it doesn't leak through the package index) for per-test latch resets. - Add a 6th case to `worker-stalled.test.ts`: construct 4 Workers on 4 different queues, each with `maxStalledCount` set on its own, assert exactly 1 `[chasquimq]` warning fired total. Pins the process-scoped invariant the doc claims. * fix(engine): honor user-supplied stalled tick/idle on embedded spawn `Consumer::spawn_stalled_detector` unconditionally overrode `stalled_detector.tick_interval_ms` and `idle_threshold_ms` from `claim_min_idle_ms`. Result: the Node `stalledInterval` option and the Python `stalled_interval_ms` kwarg were silently dead letters — the FFI shims forwarded user-supplied values, the engine threw them away at spawn time. Switch to a default-sentinel check: inherit `claim_min_idle_ms` only when the field is at its `StalledDetectorConfig::default()` value (`30_000`). An explicit non-default value flows through verbatim, and `validate()` enforces `tick >= idle` so the operator owns the lockstep when they opt in. New test `embedded_spawn_honors_user_supplied_tick_and_idle` pins the invariant: set tick+idle=200ms with claim_min_idle_ms=5000ms, assert at least 5 detector ticks fire within 15s (would be 1–2 under the old silent override). Doc updates on `ConsumerConfig::stalled_detector`, `StalledDetectorConfig::tick_interval_ms` / `idle_threshold_ms`, the Node `stalledInterval` JSDoc, and the napi `ConsumerOpts` field docs spell out the default-only inherit rule explicitly. * test(engine): isolate per-job CLAIM-reclaim test from embedded stalled detector `per_job_override_honored_after_claim_reclaim` pins the dispatch-time per-job override gate on the reader's CLAIM-reclaim path. The test runs with `claim_min_idle_ms = 250` so the PEL entry becomes CLAIMable quickly. Since slice 12 (`feat(engine): stalled-job detector with DlqReason::Stalled`) `Consumer::run` auto-spawns an embedded `StalledDetector` per `(queue, group)`. The embedded spawn inherits `tick == idle == claim_min_idle_ms = 250`; combined with the default `max_stalled_attempts = 1`, the detector's first scan after the PEL entry passes the 250ms idle threshold INCRs the stall counter to 1, crosses threshold immediately, XACKDEL-gates the entry out of the PEL, and relocates to the DLQ as `DlqReason::Stalled` — before consumer B's reader can CLAIM-redeliver it. The 15s `wait_until(dlq_len >= 1)` returns satisfied (via the stalled-relocate, not via the retry-exhaust path the assertion is checking), and the handler-call assertion fails with `observed 0` (detector wins) or `observed 1..=4` (detector wins mid-retry). CI run 26424121303 reproduced the failure on both attempt 1 and attempt 2. Local 20-run loop reproduced at ~5–20% rate; 30/30 pass after the fix. Disable the embedded detector on both consumers in this test — it isn't about stalled detection (the `stalled_detection.rs` integration tests already cover that path with a tolerant race-aware wait window). Detector defaults (`stalled_detector_enabled = true`, `max_stalled_attempts = 1`) are deliberately left alone here — they mirror BullMQ's `maxStalledCount` default and the slice author's documented intent. The interaction with short `claim_min_idle_ms` is a design surface worth surfacing to operators in docs/options rather than silently flipping defaults from this PR. * fix(engine): default max_stalled_attempts to 2 to avoid racing CLAIM-on-read With max_stalled_attempts = 1, the first idle PEL scan that finds an entry beyond the threshold routes it to DLQ-Stalled — racing the reader's CLAIM-on-read recovery path. Bumping to 2 requires a second consecutive idle observation (~tick_interval_ms apart) before relocating, eliminating the race for operators with short claim_min_idle_ms. Slightly slower DLQ-Stalled routing for genuinely crashed workers (one extra tick_interval_ms of latency). BullMQ's `maxStalledCount = 1` means "fail after 1 stall observation"; ours of `2` means "fail after 2 consecutive stall observations" — comparable safety stance. Doc surfaces synced: docs/engine.md, docs/history.md (slice note), both shim READMEs (option tables + quickstart), and the Starlight reference/concepts pages (options.md, node-api.md, python-api.md, dlq-and-recovery.md). NAPI field doc on consumer.rs / worker.ts updated; index.d.ts regenerates on build. * test(engine): re-enable embedded detector in per_job_override_honored_after_claim_reclaim The max_stalled_attempts default bump to 2 eliminated the CLAIM-vs-detector race this workaround papered over: with two consecutive idle observations now required before DLQ-Stalled routing, consumer B's reader has a full extra tick to CLAIM-reclaim the entry before the detector can act on it. The test now passes 30/30 in a loop with the default-on embedded detector enabled — drop the stalled_detector_enabled = false workaround introduced in c7454e6. * test(node): bump waitFor timeouts on progress-and-log timing-sensitive tests Two tests in progress-and-log.test.ts rely on a `waitFor(..., 10_000)` to observe per-id and Worker EE progress events. On a contended GitHub Actions runner, the spawned QueueEvents subscriber's connect+block latency plus the 3-event delivery window can exceed 10s, surfacing as a `waitFor timed out` failure even when the engine emitted the events promptly. Local repro: 18 consecutive runs all pass. CI: failed on attempt 1 of run 26425401120 (the run that includes the max_stalled_attempts=2 default bump). The test path is byte-identical to the PR #148 lazy-spawn code; the slice's worker.ts diff is purely additive (adds `stalled` to the newListener filter and a new forwarder; the `progress` forwarder is untouched). Bump both `waitFor(..., 10_000)` calls to 25_000 and the matching per-it timeouts from 15s to 30s. Headroom for slow runners; local pass time (~1s) is unaffected.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Job.updateProgress(n)) with cross-processprogressevent on Worker EE + per-job channel on QueueEvents.Job.log(line)+Queue.getJobLogs(id, start?, end?, asc?)).JobInfo.progressfield returned by introspection.log_max_stream_len(default 1000),log_max_line_bytes(default 4096),events_progress_enabled(default true).progress_throughputCriterion bench scenario.Design highlights
{chasqui:<q>}:progress:<id>STRING,{chasqui:<q>}:log:<id>STREAM) — envelope (msgpackJob<T>) is untouched, wire-format-stable.JobHandlebackref attached only during worker dispatch;Queue.getJob(...)-returned Jobs throw a clear read-only error.result_ttl_secsand are reaped byremove()andclean().Bench
No regression vs
benchmarks/chasquimq-1.0.mdbaseline (host load 14-16 vs baseline 1.8-4.3):queue-add: 16,340 jobs/s (+6.3%)queue-add-bulk: 184,295 jobs/s (-2.4%, within stddev)worker-concurrent: 110,726 jobs/s (-1.1%, within stddev)Full report:
benchmarks/progress-log-slice.md.Breaking changes
JobProgressnarrowed fromnumber | objecttonumber. No runtime impact.Test plan
cargo test --workspace -- --include-ignored)pnpm testinchasquimq-node/)pytestinchasquimq-py/)