You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Exhaustive feature + extreme-stress audit (embedded & TCP) found 7 pre-existing
bugs, each fixed with a RED->GREEN reproduction test:
- idempotent re-add of an unfinished active/waiting-children jobId no longer
throws UNIQUE / silently drops (handleCustomId skip; customId twin of #69)
- orphan jobs row collision fixed via INSERT ... ON CONFLICT(id) DO UPDATE
(upsert resets all non-id columns incl. per-execution fields; zero hot-path
cost; also stops one batch collision dropping the whole flush)
- Workflow engine.signal() no longer double-executes post-waitFor steps
(resume only a parked run, once)
- moveToDelayed over TCP: send relative delay (was timestamp->NaN), route via
changeDelay (waiting+active), and persist run_at so the delay survives restart
- deduplication.replace/extend honored in embedded (customId=jobId only;
deduplicationId sourced from customId ?? uniqueKey, #90)
- getMetrics() over TCP reads response.metrics.{totalCompleted,totalFailed}
- webhook SSRF blocks IPv4-mapped/-compatible IPv6 + ULA/link-local/unspecified
Gate: unit 5701/0, TCP suites pass, embedded 36/36 (273/273). Skeptic PASS.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Move / BullMQ-v5 (`queue.ts:502`): `moveJobToCompleted`, `moveJobToFailed`, `moveJobToWait`, `moveJobToDelayed`, `moveJobToWaitingChildren`, `waitJobUntilFinished`.`moveJobToDelayed(id, timestamp)` takes an **absolute** timestamp; embedded routes waiting/active jobs via `changeWaitingDelay`/`changeDelay`, while the TCP path (`jobMove.ts`) sends the `MoveToDelayed` command with a **relative**`delay = max(0, timestamp - now)` (not the raw timestamp) and surfaces a server `ok:false` as a thrown error. Works for both waiting and active jobs.
`ChangeDelay` and `MoveToDelayed` share one manager-level dispatcher (`queueManager.ts:1175`): for jobs already queued (`waiting`/`prioritized`/`delayed`) it calls `changeWaitingDelay` (in-place `runAt` mutation → the job becomes/stays `delayed`); for active (`processing`) jobs it falls back to the two-phase `moveJobToDelayed`. `QueueManager.moveToDelayed` (`queueManager.ts:1171`) delegates to `changeDelay` so the two stay in lock-step. Previously `moveToDelayed` only handled `processing` jobs, so a waiting job was a **silent no-op over TCP/HTTP/MCP** while the embedded SDK special-cased it; routing through `changeDelay` fixes parity for waiting **and** active jobs. Wire field: the public `moveToDelayed(id, timestamp)` API takes an **absolute** timestamp, but `MoveToDelayedCommand` carries a **relative** `delay` (ms) — the TCP client (`jobMove.ts`) converts `delay = max(0, timestamp - now)` before sending, matching the server op and the sibling `ChangeDelay` command (the HTTP route `POST /jobs/:id/move-to-delayed` already posts `{ delay }`).
101
101
102
102
### Eventsemitted
103
103
@@ -156,7 +156,7 @@ On success emits `Removed`. Returns `false` for active/completed/DLQ jobs.
Handles **active** (`processing`) jobs only. Two-phase: remove from the processing shard under `processingLocks[procIdx]` (`:238`), then re-push into the destination shard under `shardLocks[idx]`, resetting `startedAt = null`, `runAt = now + delay`, and calling `incrementQueued` with the temporal flag (`:255`). Emits `Delayed`. Jobs already **in the queue** (`waiting`/`prioritized`/`delayed`) never reach this op — the `QueueManager.moveToDelayed`/`changeDelay` dispatcher routes them to `changeWaitingDelay` (in-place `runAt` update). Like the embedded `changeDelay` path, that in-queue route does **not** emit a `Delayed` event nor bump the O(1) `delayedJobs` aggregate, but `getJobState`/`getJob` correctly report the job as `delayed` from its future `runAt` and it is no longer pullable. Both routes **persist** the new `run_at` via `storage.updateRunAt(jobId, runAt)` (re-deriving `state` from the future timestamp and clearing `started_at`), so the delay survives a restart — without it, recovery would reload the stale on-disk `run_at` (the active path's row would still read `state='active'`) and the job would be immediately pullable again.
>TheclientSDK's `queue.getMetrics('completed'|'failed')` reads this `metrics` payload over TCP — `completed → metrics.totalCompleted`, `failed → metrics.totalFailed` (`client/queue/workers.ts`). It must **not** read `response.stats` (no such key on a `Metrics` reply — that always returned `0`).
-**SSRF protection:**`validateWebhookUrl` (on by default; disabled via `validateUrls: false`) rejects non-http(s) schemes, URLs > 2048 chars, localhost variants, private IPv4 (`10.*`, `172.16–31.*`, `192.168.*`), link-local `169.254.*`, `0.*`, `127.*`, and cloud-metadata hosts (`169.254.169.254`, `metadata.google.internal`, `*.internal`) (`webhookValidation.ts:42`).
178
+
-**SSRF protection:**`validateWebhookUrl` (on by default; disabled via `validateUrls: false`) rejects non-http(s) schemes, URLs > 2048 chars, localhost variants, private IPv4 (`10.*`, `172.16–31.*`, `192.168.*`), link-local `169.254.*`, `0.*`, `127.*`, **IPv4-mapped / IPv4-compatible IPv6** whose embedded IPv4 is loopback/private (`extractMappedIpv4` unwraps the dotted `::ffff:127.0.0.1`, the URL-parser-normalized hex form `[::ffff:7f00:1]`, **and** the deprecated `::`-prefixed compatible form `[::127.0.0.1]`/`[::7f00:1]` before the octet check), **IPv6 ULA `fc00::/7` and link-local `fe80::/10`** plus the unspecified `::` (`checkBlockedIpv6`), and cloud-metadata hosts (`169.254.169.254`, `metadata.google.internal`, `*.internal`) (`webhookValidation.ts:42`).
179
179
-**Dead-event rejection:**`AddWebhook` rejects events not in `WEBHOOK_EVENTS`, so a webhook can't be created against an event that would silently never fire (`monitoring.ts:230`).
180
180
-**Delivery is best-effort / fire-and-forget:** failures are logged and counted but never block job processing; there is no persistent retry queue and webhooks are not persisted to SQLite, so they are lost on restart.
181
181
-**Fixed 10 s per-request timeout** via `AbortSignal.timeout(10000)`; linear (not exponential) inter-attempt backoff.
**waitFor** (`runWaitFor`, `executor.ts:196-236`): if the signal is already present, advance. Otherwise, if a `timeout` is set, it tracks `__waitFor:<event>` start time; if elapsed ≥ timeout it emits `signal:timeout`, marks the wait record failed, sets state `failed`, runs compensation, emits `workflow:failed`, and throws `WaitForSignalError`; else it arms a `setTimeout` via `scheduleTimeoutCheck` for the remaining time. Either way it sets state `waiting`, emits `workflow:waiting`, and throws `WaitForSignalError`. The sentinel is caught in `processStep` (`executor.ts:92`) and turned into `returnnull`, so the step job acks cleanly without triggering compensation.
123
123
124
-
**signal** (`executor.ts:102-115`): loads the execution (throws if not found), clears any pending timeout timer, records `signals[event] =payload`, sets state `running`, persists, emits `signal:received`, and re-enqueues the job at `currentNodeIndex` so `processStep` re-runs the `waitFor` node and advances.
124
+
**signal** (`executor.ts:102-128`): loads the execution (throws if not found), clears any pending timeout timer, records `signals[event] = payload` (idempotent), and emits `signal:received` — **always**, so a signal that lands *before* the run parks is still consumed at the `waitFor` via its `signals[event] !== undefined` gate. The resume is then **gated on state**: only a genuinely-parked run (`state === 'waiting'`) is flipped to `running`, persisted, and re-enqueued at `currentNodeIndex` so `processStep` re-runs the `waitFor` node and advances. For any other state (still running an earlier step, already resumed, completed, or failed) it just persists the recorded signal and returns **without** enqueuing. The `state === 'waiting'` check and the flip to `running` are synchronous (no `await` between them), and `store.update` persists `running` before the first `await this.enqueue`; a second concurrent/duplicate `signal()` therefore reads `running` back from the store and returns early. Duplicate/concurrent signals collapse to a **single resume**, so every post-`waitFor` step runs exactly once.
125
125
126
126
**Compensation** (`runCompensation`, `compensator.ts:19-56`): collects completed steps whose names do **not** start with `__`, reverses them, sets state `compensating` (+ emits `workflow:compensating`), and calls each step's `compensate(ctx)`. For `forEach` records it restores `__item`/`__index` from `loopItem`/`loopIndex`. Compensation handler errors are swallowed so the chain continues. Final state is set to `failed`. Triggered from the generic `processStep` catch (`executor.ts:96`), the `waitFor` timeout path, and recovery.
127
127
@@ -138,7 +138,8 @@ Signal-timeout timers live in an in-memory `Map<execId, timer>` on the executor
138
138
## Edge Cases & Failure Modes
139
139
140
140
- **Step re-execution is NOT idempotent.** `executeStepWithRetry` always re-runs the handler for the node at `currentNodeIndex`; there is no "already completed" short-circuit at the step level. Re-delivery, `recover()` re-enqueue of a `running` execution, or two jobs landing on the same node can run a side-effecting step more than once. The execution-level guard (`processStep` returns `null` for non-`running`/`waiting` executions, `executor.ts:74`) only prevents re-runs of *terminal* executions.
141
-
- **Known signal double-exec race (audit, repro `slowRuns=2`).** When a `waitFor` has a timeout, both the timeout timer and an incoming `signal()` can enqueue a job at the same `currentNodeIndex`; with worker `concurrency>1` two jobs for the same execution may pass the `running`/`waiting` guard concurrently and re-run the post-wait steps. Treat post-`waitFor` steps as needing idempotent side effects. (Tracked in the project audit backlog; unfixed at time of writing.)
141
+
- **Duplicate/concurrent `signal()` no longer double-executes.** `signal()` gates its resume on `state==='waiting'` and does the state check + flip to `running` synchronously, persisting `running` before its first `await` (`executor.ts:121-127`). A second signal for the same parked run — sequential *or* concurrent — reads `running` back from the store and returns after only recording its payload, so the post-`waitFor` steps run exactly once (covered by `test/repro-workflow-signal-double.test.ts`). A signal that arrives *before* the run parks is likewise just recorded and consumed later at the `waitFor`.
142
+
- **Residual timeout-timer vs. `signal()` race (narrow, pre-existing).** `signal()` clears the pending timeout timer, so a signal that wins the race cancels the timeout. But if the timeout timer has *already fired* and enqueued a job at the `waitFor` node and a `signal()` then also enqueues, two jobs for the same node can both pass the `running`/`waiting` guard under `concurrency>1` and advance — there is still no per-node dedup in `processStep`/`runWaitFor`. Treat post-`waitFor` steps as needing idempotent side effects. (Tracked in the project audit backlog.)
142
143
- **In-memory store when `dataPath` is omitted.** `WorkflowStore` opens `dataPath??':memory:'` (`store.ts:67`). In TCP/`connection` mode (no `dataPath`) execution state is in-memory only — it is lost on restart and `recover()` finds nothing. Persistence requires passing `dataPath` (typically with `embedded: true`).
143
144
- **Retry vs. compensation.** A step failing all `retry` attempts throws out of `processStep`, which sets `failed`, persists, emits `workflow:failed`, runs compensation, then re-throws — failing the underlying `wf:step` job. If the queue retries that job, `processStep` short-circuits because the execution is now `failed`, so compensation does not double-run via that path.
144
145
- **Crash mid-compensation** leaves state `compensating`; `recover()` re-runs the whole compensation set, so compensate handlers must be idempotent.
Copy file name to clipboardExpand all lines: docs/src/content/docs/changelog.md
+32Lines changed: 32 additions & 0 deletions
Display the source diff
Display the rich diff
Original file line number
Diff line number
Diff line change
@@ -10,6 +10,38 @@ head:
10
10
11
11
All notable changes to bunqueue are documented here.
12
12
13
+
## [2.8.26] - 2026-07-01
14
+
15
+
A correctness release from an exhaustive feature + extreme-stress audit (every subsystem, embedded **and** TCP). Seven fixes; all pre-existing, none data-loss in normal operation, each shipped with a RED→GREEN reproduction test.
16
+
17
+
### Fixed — idempotent re-add of an unfinished `jobId`/customId (active & waiting-children)
18
+
19
+
Re-adding a job with an existing `jobId` while the prior job was **active** (being processed) or **waiting-children** threw `UNIQUE constraint failed: jobs.id` for durable jobs, or silently dropped the colliding insert (leaving an in-memory duplicate) for buffered jobs — instead of the documented idempotent no-op. `handleCustomId` only handled the still-queued case; it now idempotent-skips for every unfinished state, gated so a **completed** id still recycles into a fresh job (#92). The customId twin of the uniqueKey fix #69.
20
+
21
+
### Fixed — orphan `jobs` row no longer collides on the primary key (durable + buffered)
22
+
23
+
A durable `jobs` row could outlive its in-memory tracking when `obliterate()` (fire-and-forget over TCP) or a write-buffer flush raced an in-flight insert, or when a completed customId job aged out of the 50k `completedJobs` window. Re-adding the same id then hit `UNIQUE constraint failed: jobs.id`. Both insert statements now use `INSERT … ON CONFLICT(id) DO UPDATE` (upsert): a brand-new id is a plain INSERT (zero hot-path cost), an orphan is overwritten in place. The `DO UPDATE SET` resets **all** non-id columns — including `started_at`/`completed_at`/`progress`/`progress_msg`/`last_heartbeat`/`stacktrace` — so a recycled id never inherits a prior life's `progress=100` or stale stacktrace. In the buffered batch path this also stops one stale collision from failing the whole flush and dropping every innocent job batched in the same window.
24
+
25
+
### Fixed — Workflow `engine.signal()` double-executed steps after `waitFor`
26
+
27
+
Two concurrent/duplicate signals (or a signal arriving before the run parked) re-enqueued the current node, so every step after the `waitFor` (e.g. a side-effecting `charge`) ran twice — an exactly-once violation. `signal()` now records the payload always but only resumes a genuinely-parked run (`state === 'waiting'`), flipping to `running` synchronously so duplicate signals collapse to a single resume.
28
+
29
+
### Fixed — `moveToDelayed` was a silent no-op over TCP, and was not durable
30
+
31
+
`Queue.moveJobToDelayed(id, timestamp)` / `job.moveToDelayed(timestamp)` over TCP left a waiting job waiting (no-op) and dropped the delay on an active job (re-queued as `waiting`). The client sent `{ timestamp }` but the command/handler read `delay` (→ `runAt = now + undefined = NaN`), and the server op only handled active jobs. The client now sends the relative `delay`, and `moveToDelayed` routes through `changeDelay` (handles in-queue + active). The new `run_at` is now **persisted** (`storage.updateRunAt`), so the delay survives a restart — previously `moveToDelayed`/`changeDelay` mutated only the in-memory heap and the delay was lost on recovery. Embedded was unaffected by the no-op bug.
32
+
33
+
### Fixed — `deduplication.replace` / `extend` ignored in embedded mode
34
+
35
+
With the documented API `add(name, data, { deduplication: { id, replace: true } })` (no explicit `jobId`), embedded set `customId = deduplication.id`, so `handleCustomId` short-circuited the re-add before the replace/extend strategy ran — the original job survived. The dedup id now rides only on `uniqueKey` (matching TCP); `customId` is set from an explicit `jobId` only. `deduplicationId` on the returned job is sourced from `customId ?? uniqueKey` so it still reflects the requested id (#90).
36
+
37
+
### Fixed — `queue.getMetrics()` over TCP always returned `0`
38
+
39
+
The TCP client read `response.stats.completed` / `.dlq`, but the `Metrics` handler returns `response.metrics.totalCompleted` / `.totalFailed`. The client now reads the correct fields.
40
+
41
+
### Security — webhook SSRF guard now blocks IPv4-mapped/-compatible IPv6 and IPv6 private ranges
42
+
43
+
`http://[::ffff:127.0.0.1]/…`, the deprecated IPv4-compatible `[::127.0.0.1]`, and IPv6 ULA (`fc00::/7`) / link-local (`fe80::/10`) / unspecified (`::`) hosts bypassed the webhook SSRF check (in both dotted and WHATWG hex-normalized forms). The validator now unwraps mapped/compatible addresses and blocks the IPv6 private ranges before delivery.
44
+
13
45
## [2.8.25] - 2026-06-29
14
46
15
47
### Fixed — `finishedOn`/`processedOn` always `undefined` on jobs from list queries (#104)
0 commit comments