feat(workflow)!: replace groupmq with purpose-built queue module - #1
Merged
Conversation
LouisHaftmann
force-pushed
the
feat/purpose-built-queue
branch
from
July 26, 2026 07:47
094477c to
a5092c4
Compare
…reserve Introduce the purpose-built queue module's happy path end-to-end: Namespace -> Queue -> Worker, string-only immediate `add`, the atomic enqueue/reserve/releaseActive/complete Lua on the §3 key layout (registered via ioredis `defineCommand`), a JS-local worker-cap semaphore with drain-before-block `BRPOP` wake, blocking pub/sub `wait()`, the three-level `close` cascade, and `JobAlreadyExistsError`. Not wired into the workflow lib yet (ticket 11); concurrency caps, delayed/priority surface, retries/fail, stalled recovery, cron, metrics, and step data are later tickets. Gates are present-but-unlimited so those tickets are additive.
Ticket 02 already wired all four caps through enqueue/reserve/releaseActive (namespace SCARD top-gate, workflow ZCARD top-gate, group ready-set membership, worker JS semaphore) with correct defaults and ready-set maintenance on both reserve and release. This adds the ticket-03 acceptance suite proving that wiring, driven through the public Namespace -> Queue -> Worker API: - each cap enforced in isolation (worker / workflow / namespace / group) - group cap 1 serializes; group cap > 1 parallelizes - combined nested gating: a top-gate binds while group + worker have room - a capped group does not stall a runnable sibling (no head-of-line block) - contention invariant: K workers x M jobs across all four levels drain with every active structure empty and no observed cap ever exceeded Tests are property-style, prefix-scoped, K*M small. Verified they go red for the right reason by neutering the Redis cap gates (worker cap stays green as it is JS-side). No source changes needed.
Ticket 02 already packed the score `(PMAX-priority)*2^32 + counter` with the
`pc` counter stamped at enqueue (= promotion time for immediate jobs) and
defaulted `priority` to 0. Ticket 04 adds the missing acceptance criterion:
validate `add({ priority })` is an integer in 0…2^21-1 (an out-of-range or
fractional value would corrupt the packed score / ZSET ordering), throwing a
RangeError. Proves the §6 semantics with tests: higher-runs-first across a
workflow's groups, default-0 after any expedited job, and FIFO-by-promotion
tiebreak among equals (single worker at concurrency 1 for deterministic order).
Per-step data addressed by (jobId, stepName): setStepData is a single atomic HSET into the job's :steps hash, getStepData a single HGET returning null on miss. Opaque strings; no serialization, no Lua, no mutex. Parallel steps write distinct fields without lost updates. The complete script already deletes the :steps hash in the same op that finalizes the job (walking skeleton); tests now assert that seam. The fail-path :steps delete lands with ticket 07.
Schedule a job for a delay (`runIn`) or an absolute time (`runAt`); a due job becomes runnable exactly once with no poller and no idle Redis load. - `enqueue` resolves the effective `runAt` against Redis TIME (`runIn` → `now + runIn`, `runAt` verbatim), parks `runAt > now` in the `delayed` ZSET and routes `runAt <= now` straight into waiting. - Promotion is embedded at the top of `reserve` and reuses a shared `addWaiting` Lua helper (INCR pc + packed score + ready maintenance) also used by `enqueue`, so ready logic cannot drift. Due jobs are promoted in `runAt` order (FIFO-by-due), batch-capped per call. - `reserve` now returns ms-to-next-due on its `empty` branch; the worker uses it as the BRPOP wake timeout (the block is the delayed-job timer) and re-reserves at once when a backlog remains past the promote cap.
…t-claim Add a token-CAS `heartbeat` script that renews the lock PX and the `wf:active` deadline score in lockstep, and one throttled `recoverStalled` script gated by `SET wf:stalled-check NX PX interval`. Detection is a pure deadline-compare (`ZRANGEBYSCORE wf:active 0 now` + `state == "active"` fence); recovery reuses the shared `releaseActive` (group-unblock for free), `addWaiting` (requeue at the stored packed score, front of band), and the `moveToFailed`/`finalizeFailed` dead-letter path once `stalledCount` exceeds `maxStalledCount`. `reserve` now stores the popped packed score on the job hash so recovery can requeue drift-free. The worker runs the heartbeat on a derived `min(lockMs/3, 10s)` timer, triggers the scan from that tick (busy workers) and the wake-loop re-poll (idle workers), and aborts `ctx.signal` + drops the job when a renew returns 0 or errors past `lockMs`. New Worker options `maxStalledCount` (1) and `stalledInterval` (30s).
Add a first-class scheduleId-keyed cron registry to the queue module. `upsertSchedule`/`removeSchedule`/`getSchedules` write on the Queue, storing a `schedules:due` ZSET (member scheduleId, score next-fire ms) plus a `schedule:<id>` hash. Idempotent by construction: both are keyed by scheduleId, so re-upserting replaces in place — never duplicates. Firing is "JS computes next, Lua commits via CAS-on-score": a thin tick folded into the Worker wake loop (no poller — the nearest due schedule is min'd into the BRPOP timeout alongside delayed jobs) reads due schedules, computes croner.nextRun(now), and calls a new `fireSchedule` script that CAS's on `ZSCORE == expectedScore` before enqueuing the occurrence (via a shared `enqueueNow` include factored out of `enqueue`, never a parallel path) and re-arming atomically. CAS = exactly-once across N workers with no distributed lock; JS-before-Lua = crash-safe. Missed runs collapse to one fire (nextRun(now) jumps forward); skip-if-previous-still-running is the default (groupId defaults to scheduleId). Croner is the cron authority (DST-safe IANA tz); tz is captured from the queuer's local zone at register time unless overridden.
Expose `Queue.getMetrics()` returning `{ active, waiting, delayed }` for the
lib's OTel `addBatchObservableCallback` to poll: `active`/`delayed` are O(1)
`ZCARD`s and `waiting` sums the per-group waiting ZSETs over the lightweight
`wf:<id>:groups` membership SET (O(groups), pull-time only). No OTel and no
counters inside the module.
Introduce a shared `maintainGroup` Lua helper that ties both the `ready`
ZSET-of-groups and the `groups` metrics SET to the single `ZCARD groupJobs`
read, and route the three ready-maintenance sites (`addWaiting`, `reserve`'s
pop, `releaseActive`) through it. The SET is a membership set (SADD while a
group has waiting work, SREM when it drains), never an INCR/DECR count, so it
cannot drift from the waiting jobs or over-count.
Add optional `onError` / `onFailed` Worker callbacks defaulting to
`Settings.logger`: `onFailed(job, error)` fires best-effort on every handler
failure (mirroring the old `worker.on('failed')`), `onError` on worker-internal
errors. No pub/sub, no `.on(...)` registry.
The library now runs entirely on the internal `src/queue/` module
(`Namespace → Queue → Worker`). groupmq, p-mutex, p-retry and the
groupmq Lua patch are removed. Step data is per-field superjson in the
`:steps` hash (no read-modify-write mutex); the backbone is string-only
so superjson lives 100% in the lib. wait() blocks on pub/sub and throws;
cancellation is via `ctx.signal`; cron is the schedule registry.
BREAKING CHANGE: the public API changed.
- Entry point is `new WorkflowNamespace({ id, concurrency?, redis?,
prefix?, queueOptions?, workerOptions?, jobOptions? })` then
`ns.createWorkflow({ id, schema?, run, getGroupId?, ...overrides })`.
`new Workflow(...)` is no longer public.
- `priority` is now a raw `number` (higher runs first, default 0); the
`'high' | 'normal'` enum and `orderMs` are gone.
- `run({ repeat })` is removed; use `workflow.upsertSchedule(id, {
pattern, input, ... })` / `removeSchedule` / `getSchedules`.
- `run`/`runIn`/`runAt` are one-shot only.
- `WorkflowJob.wait()` now throws on workflow failure and throws
`TimeoutError` / `ResultExpiredError`; it no longer resolves
`undefined`. Return type is `Promise<Output>`.
- `step.do(..., { retry })` is removed; steps are pure replay
checkpoints (job-level retry replays completed steps).
- Metrics are three gauges (active/waiting/delayed); worker events are
`onError`/`onFailed` callbacks instead of `.on('failed'|'error')`.
Expand tests/workflow.test.ts to exhaustively cover the surviving workflow contract and the intentionally-changed behaviors, driven end-to-end through the public library API. Adds: job-retry step replay (only the failed step re-runs), no step-level retry, durable-sleep resume + signal-aware cancellation with token-safe commit, wait() TimeoutError/ResultExpiredError, numeric-priority enum removal, upsertSchedule idempotency, non-POJO (Date/Map/BigInt/Set) serialization round-trip, and retired-surface guards (keepCompleted / step retry / repeat / removeRepeatingJob).
LouisHaftmann
force-pushed
the
feat/purpose-built-queue
branch
from
July 26, 2026 07:50
a5092c4 to
bb0a306
Compare
LouisHaftmann
commented
Jul 26, 2026
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
Replaces the third-party
groupmqjob queue (and its local Lua patch) with a purpose-built, Redis-backed job queue built as an internal module of@falcondev-oss/workflow(src/queue/), then refactors the library onto it. Deletesgroupmq,p-mutex,p-retry, andpatches/groupmq.patch.Built ticket-by-ticket from the locked design spec (
.scratch/queue-replacement/spec.md), one commit per slice, each landing green.What the new module adds
'high'|'normal'→orderMsfake.runIn/runAt) promoted inside the reserve loop — no poller, RedisTIMEas the one clock.scheduleId-keyed cron registry (Croner) with CAS-on-score exactly-once firing, idempotent upsert (kills theremoveRepeatingJobpatch), skip-missed / skip-if-running.expBackoffdefault), dead-letter +keepFailedretention.recoverStalledscript, token-CAS heartbeat, abort-on-lost-claim.wait()over pub/sub + TTL result record (throws on failure /TimeoutError/ResultExpiredError).active/waiting/delayed) + best-effortonError/onFailedcallbacks.Anti-drift discipline throughout: one shared Lua include (
releaseActive/addWaiting/maintainGroup/finalizeFailed/enqueueNow) reused across reserve/complete/fail/recover/fire, so the groupmq drift-class bugs can't recur. Scripts registered via ioredisdefineCommand(free NOSCRIPT/EVAL fallback).Testing
Namespace → Queue → WorkerAPI driven against a real testcontainers Redis; emergent invariants (no slot-leak/drift,wait()fan-out) reach under the API for raw key-state assertions.testnpm script + a CI job running the full suite on every PR (none existed before).TIMEauthoritative, back-dated scores for time logic).BREAKING CHANGE
new WorkflowNamespace({...}).createWorkflow({...})— the publicnew Workflow(...)is gone.number(default 0); the'high'|'normal'enum is removed.run({ repeat })toworkflow.upsertSchedule(...);removeRepeatingJobis gone.WorkflowJob.wait()now throws on workflow failure / timeout / expired result (previously resolvedundefined).retryandkeepCompletedare removed (steps are replay checkpoints; job-level retry replays steps).🤖 Generated with Claude Code