diff --git a/README.md b/README.md index fc3cf1c..fab3243 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@ # @falcondev-oss/workflow -Simple type-safe queue worker with durable execution using Redis. +Durable, type-safe queue workers on Redis. Workflows are plain async functions whose `step`s are +memoized in Redis, so a retried job replays completed steps instead of re-running them. ## Installation @@ -8,12 +9,26 @@ Simple type-safe queue worker with durable execution using Redis. npm install @falcondev-oss/workflow ``` +Requires Redis (any version with Lua scripting) and Node 24. + ## Usage +Workflows live in a `WorkflowNamespace`, which owns the Redis connection, the cross-workflow +concurrency cap, and the shared option defaults. + ```ts -const workflow = new Workflow({ +import { createRedis, WorkflowNamespace } from '@falcondev-oss/workflow' +import { z } from 'zod' + +const namespace = new WorkflowNamespace({ + id: 'my-app', + redis: await createRedis({ url: process.env.REDIS_URL }), + logger: console, +}) + +const workflow = namespace.createWorkflow({ id: 'example-workflow', - input: z.object({ + schema: z.object({ timezone: z.string().default('UTC'), name: z.string(), }), @@ -24,42 +39,112 @@ const workflow = new Workflow({ await step.wait('wait a lil', 60_000) - const isEngaged = await step.do('check engagement', () => { - return Math.random() > 0.5 - }) - + const isEngaged = await step.do('check engagement', () => Math.random() > 0.5) if (!isEngaged) return { engagementLevel: 'low' } await step.do('send tips', () => { console.log(`Here are some tips to get started, ${input.name}!`) }) - await step.wait('wait feedback', 3000) - - await step.do('send survey', () => { - console.log(`Hi ${input.name}, please take our survey!`) - }) - - return { - engagementLevel: 'high', - } + return { engagementLevel: 'high' } }, }) -// Start worker +// Start a worker for this process await workflow.work() -// Run workflow -const job = await workflow.run({ - name: 'John Doe', - timezone: 'America/New_York', -}) +// Enqueue a run +const job = await workflow.run({ name: 'John Doe', timezone: 'America/New_York' }) -// Wait for completion +// Wait for completion (works from a pure producer too — no worker needed) const result = await job.wait() console.log(result.engagementLevel) ``` +### Steps + +- `step.do(name, fn)` — run once, memoize the result. Replayed from Redis on a retry. +- `step.wait(name, ms)` — durable sleep; remaining time is computed from the persisted start. +- `step.waitUntil(name, date)` — the same, to an absolute time. + +Steps nest: the callback receives its own `step` scoped under the parent's name. + +### Scheduling + +```ts +await workflow.run(input) // now +await workflow.runIn(input, 60_000) // in 60s +await workflow.runAt(input, new Date('2030-01-01')) // at a time + +// Cron, keyed by (workflow, scheduleId) — upserting the same id replaces in place +await workflow.upsertSchedule('nightly', { + pattern: '0 3 * * *', + input, + tz: 'Europe/Berlin', +}) +await workflow.getSchedules() +await workflow.removeSchedule('nightly') +``` + +### Ordering, priority and concurrency + +Jobs sharing a `groupId` run one at a time, in enqueue order. Everything else runs in parallel up +to the concurrency caps. + +```ts +namespace.createWorkflow({ + id: 'per-user', + schema: z.object({ userId: z.string() }), + getGroupId: (input) => input.userId, // serialize per user + queueOptions: { concurrency: 10, groupConcurrency: 1 }, + workerOptions: { concurrency: 4, maxAttempts: 3 }, + jobOptions: { priority: 1 }, // 0…2^21-1, higher runs first + run: async () => {}, +}) +``` + +### Options + +`WorkflowNamespace` options are shared defaults — each is shallow-merged under the matching +per-workflow override. + +| Option | Default | Description | +| --------------- | ---------- | ------------------------------------------------------- | +| `id` | — | Namespace id; scopes the cross-workflow concurrency cap | +| `redis` | new client | Shared connection, owned by the namespace | +| `prefix` | `wf` | Global key prefix | +| `concurrency` | unlimited | Ceiling across all workflows in the namespace | +| `logger` | none | Inherited by every workflow, queue and worker | +| `autoClose` | `true` | Close (drain workers, disconnect) on `SIGINT`/`SIGTERM` | +| `queueOptions` | — | Defaults for every workflow's queue | +| `workerOptions` | — | Defaults for every workflow's worker | +| `jobOptions` | — | Defaults for every enqueued job | + +Shutdown is handled at the namespace: workers drain in-flight jobs, then the connections close. +Set `autoClose: false` and call `await namespace.close()` yourself to own it. + +### Failures and retries + +A throwing handler is retried up to `maxAttempts` with an exponential backoff (`expBackoff()`), +then dead-lettered. Throw `NonRecoverableError` to dead-letter immediately, skipping the +remaining budget — the library does this itself when a job's stored payload no longer validates +against the workflow `schema` (a job enqueued before a schema change), since a retry would only +re-read the same payload. + +### Metrics + +`workflow.getMetrics()` returns point-in-time `{ active, waiting, delayed }` depths. Pass an +OpenTelemetry meter to export them as gauges: + +```ts +const namespace = new WorkflowNamespace({ + id: 'my-app', + workerOptions: { metrics: { meter, prefix: 'my_app' } }, +}) +``` + +Spans are emitted for producers, workers and each step via the global OpenTelemetry tracer. + ## Inspiration - https://x.com/imsh4yy/status/1984073526605967785?s=46 diff --git a/package.json b/package.json index 76aa570..e1dc947 100644 --- a/package.json +++ b/package.json @@ -3,7 +3,7 @@ "type": "module", "version": "0.12.0", "packageManager": "pnpm@10.28.2", - "description": "Simple type-safe queue worker with durable execution using Redis.", + "description": "Durable, type-safe queue workers on Redis with memoized steps, cron schedules, group ordering, OpenTelemetry tracing.", "license": "MIT", "repository": "github:falcondev-oss/workflow", "bugs": { diff --git a/src/index.ts b/src/index.ts index 5546bcd..2eed76e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,3 +1,10 @@ -export { expBackoff, JobAlreadyExistsError, ResultExpiredError, TimeoutError } from './queue' -export { createRedis, Settings } from './settings' +export { + expBackoff, + JobAlreadyExistsError, + NonRecoverableError, + ResultExpiredError, + TimeoutError, +} from './queue' +export type { WorkflowLogger } from './queue' +export { createRedis } from './settings' export * from './workflow' diff --git a/src/queue/errors.ts b/src/queue/errors.ts index 568340f..8a489cb 100644 --- a/src/queue/errors.ts +++ b/src/queue/errors.ts @@ -12,6 +12,18 @@ export class ResultExpiredError extends Error { } } +/** + * Thrown by a handler for a failure that retrying cannot fix — the job is dead-lettered + * immediately, skipping the remaining `maxAttempts` budget. Only the thrown error itself is + * checked; wrapping one in another error's `cause` does not skip retries. + */ +export class NonRecoverableError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options) + this.name = 'NonRecoverableError' + } +} + export class TimeoutError extends Error { constructor(jobId: string) { super(`Timed out waiting for job: ${jobId}`) diff --git a/src/queue/index.ts b/src/queue/index.ts index 7a93733..75eb9cd 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -1,6 +1,11 @@ export { expBackoff } from './backoff' export type { ExpBackoffOptions } from './backoff' -export { JobAlreadyExistsError, ResultExpiredError, TimeoutError } from './errors' +export { + JobAlreadyExistsError, + NonRecoverableError, + ResultExpiredError, + TimeoutError, +} from './errors' export { Namespace } from './namespace' export { Queue } from './queue' export type { @@ -14,6 +19,7 @@ export type { ScheduleOptions, WaitOptions, WorkerOptions, + WorkflowLogger, } from './types' export { Worker } from './worker' export type { WorkerHandler } from './worker' diff --git a/src/queue/namespace.ts b/src/queue/namespace.ts index 6778a0e..f3a02d7 100644 --- a/src/queue/namespace.ts +++ b/src/queue/namespace.ts @@ -1,6 +1,6 @@ import type Redis from 'ioredis' import type { QueueRedis } from './scripts' -import type { NamespaceOptions, QueueOptions } from './types' +import type { NamespaceOptions, QueueOptions, WorkflowLogger } from './types' import { Queue } from './queue' import { registerScripts, UNLIMITED } from './scripts' @@ -14,6 +14,7 @@ export class Namespace { readonly prefix: string readonly concurrency: number readonly redis: QueueRedis + readonly logger?: WorkflowLogger private readonly subscriber: Redis private readonly subscriberReady: Promise @@ -24,6 +25,7 @@ export class Namespace { this.id = opts.id this.prefix = opts.prefix ?? 'wf' this.concurrency = opts.concurrency ?? UNLIMITED + this.logger = opts.logger this.redis = registerScripts(opts.redis) this.subscriber = opts.redis.duplicate() diff --git a/src/queue/queue.ts b/src/queue/queue.ts index ca49670..30424c7 100644 --- a/src/queue/queue.ts +++ b/src/queue/queue.ts @@ -8,6 +8,7 @@ import type { ScheduleOptions, WaitOptions, WorkerOptions, + WorkflowLogger, } from './types' import { randomUUID } from 'node:crypto' import { JobAlreadyExistsError, ResultExpiredError, TimeoutError } from './errors' @@ -43,6 +44,10 @@ export class Queue { return this.ns.prefix } + get logger(): WorkflowLogger | undefined { + return this.ns.logger + } + /** Enqueue an immediate job. Throws `JobAlreadyExistsError` on a live id collision. */ async add(data: string, opts?: AddOptions): Promise<{ id: string; groupId: string }> { const id = opts?.jobId ?? randomUUID() diff --git a/src/queue/scripts.ts b/src/queue/scripts.ts index cd25ace..ccaa482 100644 --- a/src/queue/scripts.ts +++ b/src/queue/scripts.ts @@ -341,10 +341,11 @@ return 1 * claim). Else `HINCRBY attempts 1` (the single attempt counter, never in JS; `stalledCount` * is untouched). Retryable (`attempts < maxAttempts`) → set `state=delayed`, `releaseActive` * (frees ALL concurrency slots + kicks both wake lists during backoff), and `ZADD delayed` at - * the JS-computed `runAt` score so `reserve` promotes it when due. Exhausted → `finalizeFailed`. + * the JS-computed `runAt` score so `reserve` promotes it when due. Exhausted, or `noRetry=1` + * (a `NonRecoverableError` — retrying cannot fix it) → `finalizeFailed`. * * Returns 0 (stale token no-op), 1 (terminal dead-letter), or 2 (requeued for retry). - * ARGV: prefix, wfId, jobId, token, reason, stack, runAt, resultTtl, groupCap, keepFailed + * ARGV: prefix, wfId, jobId, token, reason, stack, runAt, resultTtl, groupCap, keepFailed, noRetry */ const FAIL = ` ${MAINTAIN_GROUP} @@ -360,6 +361,7 @@ local runAt = tonumber(ARGV[7]) local resultTtl = tonumber(ARGV[8]) local groupCap = tonumber(ARGV[9]) local keepFailed = tonumber(ARGV[10]) +local noRetry = tonumber(ARGV[11]) local wf = prefix .. ":" .. wfId local jobKey = wf .. ":j:" .. jobId @@ -371,7 +373,7 @@ end local attempts = redis.call("HINCRBY", jobKey, "attempts", 1) local maxAttempts = tonumber(redis.call("HGET", jobKey, "maxAttempts")) -if attempts < maxAttempts then +if noRetry == 0 and attempts < maxAttempts then redis.call("HSET", jobKey, "state", "delayed", "runAt", runAt) releaseActive(prefix, wfId, jobId, groupCap) redis.call("ZADD", wf .. ":delayed", runAt, jobId) diff --git a/src/queue/types.ts b/src/queue/types.ts index 338c48b..20a3be6 100644 --- a/src/queue/types.ts +++ b/src/queue/types.ts @@ -1,5 +1,13 @@ import type Redis from 'ioredis' +export type WorkflowLogger = { + info?: (...data: any[]) => void + success?: (...data: any[]) => void + error?: (...data: any[]) => void + debug?: (...data: any[]) => void + warn?: (...data: any[]) => void +} + export interface NamespaceOptions { /** Namespace id — scopes the cross-workflow concurrency cap. */ id: string @@ -9,6 +17,8 @@ export interface NamespaceOptions { redis: Redis /** Global key prefix. Default: `wf`. */ prefix?: string + /** Logger inherited by every queue and worker of this namespace. Default: no logging. */ + logger?: WorkflowLogger } export interface QueueOptions { @@ -67,9 +77,9 @@ export interface WorkerOptions { backoff?: (attempt: number) => number /** Max retained failed jobs (count-trimmed by `finishedOn`). Default: 100. */ keepFailed?: number - /** Called with a worker-internal/unexpected error (best-effort). Defaults to `Settings.logger`. */ + /** Called with a worker-internal/unexpected error (best-effort). Defaults to the namespace logger. */ onError?: (error: unknown) => void - /** Called with the job + error each time a handler throws and the job fails (best-effort). Defaults to `Settings.logger`. */ + /** Called with the job + error each time a handler throws and the job fails (best-effort). Defaults to the namespace logger. */ onFailed?: (job: ReservedJob, error: unknown) => void } diff --git a/src/queue/worker.ts b/src/queue/worker.ts index 45bc766..dbba116 100644 --- a/src/queue/worker.ts +++ b/src/queue/worker.ts @@ -3,8 +3,8 @@ import type { Queue } from './queue' import type { QueueRedis } from './scripts' import type { JobContext, ReservedJob, WorkerOptions } from './types' import { randomUUID } from 'node:crypto' -import { Settings } from '../settings' import { expBackoff } from './backoff' +import { NonRecoverableError } from './errors' import { localTimeZone, nextRunMs } from './schedule' export type WorkerHandler = (job: ReservedJob, ctx: JobContext) => Promise | string @@ -65,8 +65,8 @@ export class Worker { this.promoteBatchSize = opts?.promoteBatchSize ?? 500 this.backoff = opts?.backoff ?? expBackoff() this.keepFailed = opts?.keepFailed ?? 100 - this.onError = opts?.onError ?? ((error) => Settings.logger?.error?.(error)) - this.onFailed = opts?.onFailed ?? ((_job, error) => Settings.logger?.error?.(error)) + this.onError = opts?.onError ?? ((error) => queue.logger?.error?.(error)) + this.onFailed = opts?.onFailed ?? ((_job, error) => queue.logger?.error?.(error)) this.redis = queue.redis this.blockingRedis = queue.redis.duplicate() @@ -304,6 +304,7 @@ export class Worker { this.queue.resultTtl, this.queue.groupConcurrency, this.keepFailed, + err instanceof NonRecoverableError ? 1 : 0, ) } catch (failErr) { this.onError(failErr) diff --git a/src/settings.ts b/src/settings.ts index 610c02e..26b260a 100644 --- a/src/settings.ts +++ b/src/settings.ts @@ -1,22 +1,7 @@ -import type { Meter } from '@opentelemetry/api' import type { RedisOptions } from 'ioredis' import { createSingletonPromise } from '@antfu/utils' import IORedis from 'ioredis' -export type WorkflowLogger = { - info?: (...data: any[]) => void - success?: (...data: any[]) => void - error?: (...data: any[]) => void - debug?: (...data: any[]) => void - warn?: (...data: any[]) => void -} - -export const Settings = { - defaultConnection: undefined as (() => Promise | IORedis) | undefined, - logger: undefined as WorkflowLogger | undefined, - metrics: undefined as { meter: Meter; prefix: string } | undefined, -} - const defaultRedisOptions: RedisOptions = { lazyConnect: true, maxRetriesPerRequest: null, @@ -24,9 +9,8 @@ const defaultRedisOptions: RedisOptions = { enableOfflineQueue: false, } +/** Fallback connection for namespaces created without an explicit `redis`, shared process-wide. */ export const defaultRedisConnection = createSingletonPromise(async () => { - if (Settings.defaultConnection) return Settings.defaultConnection() - const redis = new IORedis(defaultRedisOptions) await redis.connect() return redis diff --git a/src/step.ts b/src/step.ts index d0873a2..31cf56b 100644 --- a/src/step.ts +++ b/src/step.ts @@ -2,7 +2,6 @@ import type { Span } from '@opentelemetry/api' import type { WorkflowQueueInternal } from './types' import { setTimeout } from 'node:timers/promises' import { deserialize, serialize } from './serializer' -import { Settings } from './settings' import { runWithTracing } from './tracer' export type WorkflowStepData = @@ -23,6 +22,7 @@ export class WorkflowStep { private stepNamePrefix private signal private stepPromises + private logger constructor(opts: { queue: WorkflowQueueInternal @@ -38,6 +38,7 @@ export class WorkflowStep { this.stepNamePrefix = opts.stepNamePrefix ? `${opts.stepNamePrefix}|` : '' this.signal = opts.signal this.stepPromises = opts.stepPromises + this.logger = opts.queue.logger } private addNamePrefix(name: string) { @@ -51,7 +52,7 @@ export class WorkflowStep { // attempt, so a job-level retry replays it from cache and only re-runs the failed step. const stepData = await this.getStepData('do', name) if (stepData && 'result' in stepData) { - Settings.logger?.debug?.( + this.logger?.debug?.( `[${this.workflowId}/${this.workflowJobId}] Step '${name}' already completed, returning cached result`, ) return stepData.result as R @@ -60,7 +61,7 @@ export class WorkflowStep { // Cooperative cancellation: bail before starting new work if the claim was lost. this.signal.throwIfAborted() - Settings.logger?.debug?.(`[${this.workflowId}/${this.workflowJobId}] Running step '${name}'`) + this.logger?.debug?.(`[${this.workflowId}/${this.workflowJobId}] Running step '${name}'`) const promise = runWithTracing( `workflow-worker/${this.workflowId}/step/${name}`, { @@ -85,9 +86,7 @@ export class WorkflowStep { await this.updateStepData(name, { type: 'do', result }) - Settings.logger?.debug?.( - `[${this.workflowId}/${this.workflowJobId}] Completed step '${name}'`, - ) + this.logger?.debug?.(`[${this.workflowId}/${this.workflowJobId}] Completed step '${name}'`) return result }, @@ -121,7 +120,7 @@ export class WorkflowStep { }, }, async () => { - Settings.logger?.debug?.( + this.logger?.debug?.( `[${this.workflowId}/${this.workflowJobId}] Waiting in step '${name}' for ${durationMs} ms`, ) // Durable sleep survives resume: remaining time is computed from the persisted start. diff --git a/src/workflow.ts b/src/workflow.ts index 429bf49..871f0de 100644 --- a/src/workflow.ts +++ b/src/workflow.ts @@ -2,14 +2,21 @@ import type { Meter, Span } from '@opentelemetry/api' import type { StandardSchemaV1 } from '@standard-schema/spec' import type Redis from 'ioredis' import type { IsUnknown } from 'type-fest' -import type { AddOptions, QueueOptions, ReservedJob, Worker, WorkerOptions } from './queue' +import type { + AddOptions, + QueueOptions, + ReservedJob, + Worker, + WorkerOptions, + WorkflowLogger, +} from './queue' import type { WorkflowJobPayloadInternal, WorkflowQueueInternal } from './types' import { context, propagation, ROOT_CONTEXT, SpanKind } from '@opentelemetry/api' import { asyncExitHook } from 'exit-hook' import { WorkflowJob } from './job' -import { Namespace } from './queue' +import { Namespace, NonRecoverableError } from './queue' import { deserialize, serialize } from './serializer' -import { defaultRedisConnection, Settings } from './settings' +import { defaultRedisConnection } from './settings' import { WorkflowStep } from './step' import { runWithTracing } from './tracer' @@ -32,6 +39,13 @@ export interface WorkflowNamespaceOptions { concurrency?: number redis?: Redis prefix?: string + /** Logger inherited by every workflow, queue and worker of this namespace. Default: none. */ + logger?: WorkflowLogger + /** + * Close the namespace (draining workers, disconnecting redis) on `SIGINT`/`SIGTERM`/exit. + * Default: true. + */ + autoClose?: boolean queueOptions?: WorkflowQueueOptions workerOptions?: WorkflowWorkerOptions jobOptions?: WorkflowJobRunOptions @@ -67,11 +81,14 @@ export interface WorkflowScheduleOptions { */ export class WorkflowNamespace { readonly id: string + readonly logger?: WorkflowLogger private readonly opts: WorkflowNamespaceOptions private namespace?: Promise + private unregisterExitHook?: () => void constructor(opts: WorkflowNamespaceOptions) { this.id = opts.id + this.logger = opts.logger this.opts = opts } @@ -84,9 +101,19 @@ export class WorkflowNamespace { concurrency: this.opts.concurrency, redis, prefix: this.opts.prefix, + logger: this.opts.logger, }) - // One namespace-level exit hook for the redis disconnect (cascades to queues/workers). - asyncExitHook(async () => namespace.close(), { wait: 10_000 }) + // One namespace-level exit hook: drains every worker and disconnects redis on a + // process signal. Opt out with `autoClose: false` to own shutdown yourself. + if (this.opts.autoClose ?? true) { + this.unregisterExitHook = asyncExitHook( + async (signal) => { + this.logger?.info?.(`[${this.id}] Received ${signal}, closing namespace...`) + await namespace.close() + }, + { wait: 10_000 }, + ) + } return namespace })() } @@ -107,6 +134,9 @@ export class WorkflowNamespace { /** Top-level cascade: closes every queue/worker and disconnects the shared connections. */ async close(): Promise { if (!this.namespace) return + // Drop the exit hook first: an explicit close means a later signal must not close again. + this.unregisterExitHook?.() + this.unregisterExitHook = undefined const namespace = await this.namespace await namespace.close() } @@ -124,6 +154,10 @@ export class Workflow { this.id = opts.id } + private get logger() { + return this.ns.logger + } + private async getQueue(): Promise { if (!this.queue) { this.queue = (async () => { @@ -145,12 +179,22 @@ export class Workflow { const worker = queue.worker( async (job: ReservedJob, ctx) => { - Settings.logger?.info?.(`[${this.id}] Processing job ${job.id}`) + this.logger?.info?.(`[${this.id}] Processing job ${job.id}`) const deserializedData = deserialize(job.data) const parsedData = this.opts.schema && (await this.opts.schema['~standard'].validate(deserializedData.input)) - if (parsedData?.issues) throw new Error(`Invalid workflow input`) + if (parsedData?.issues) { + // Stored payload no longer matches the schema — typically a job enqueued before a + // schema change. Retrying re-reads the same payload, so this can never succeed. + this.logger?.warn?.( + `[${this.id}] Job ${job.id} data does not match the workflow schema (stale payload from an older schema version?):`, + parsedData.issues, + ) + throw new NonRecoverableError(`Invalid workflow input for job ${job.id}`, { + cause: parsedData.issues, + }) + } return runWithTracing( `workflow-worker/${this.id}`, @@ -178,13 +222,13 @@ export class Workflow { span, }) - Settings.logger?.success?.( + this.logger?.success?.( `[${this.id}] Completed job ${job.id} in ${(performance.now() - start).toFixed(2)} ms`, ) return serialize(result) } catch (err) { if (stepPromises.size > 0) { - Settings.logger?.warn?.( + this.logger?.warn?.( `[${this.id}] Job failed but there are still ${stepPromises.size} running step(s), waiting for them to finish. Be careful when using 'Promise.all([step0, step1, ...])', as running steps are not canceled when one of them fails.`, ) await Promise.allSettled(stepPromises) @@ -202,25 +246,15 @@ export class Workflow { ...workerOpts, onFailed: workerOpts.onFailed ?? - ((job, error) => Settings.logger?.error?.(`[${this.id}] Job ${job.id} failed:`, error)), + ((job, error) => this.logger?.error?.(`[${this.id}] Job ${job.id} failed:`, error)), onError: - workerOpts.onError ?? - ((error) => Settings.logger?.error?.(`[${this.id}] Job error:`, error)), + workerOpts.onError ?? ((error) => this.logger?.error?.(`[${this.id}] Job error:`, error)), }, ) - Settings.logger?.info?.(`[${this.id}] Worker started`) + this.logger?.info?.(`[${this.id}] Worker started`) - const metricsOpts = metrics ?? Settings.metrics - if (metricsOpts) this.setupMetrics(queue, metricsOpts) - - asyncExitHook( - async (signal) => { - Settings.logger?.info?.(`[${this.id}] Received ${signal}, shutting down worker...`) - await worker.close() - }, - { wait: 10_000 }, - ) + if (metrics) this.setupMetrics(queue, metrics) return worker } @@ -314,6 +348,12 @@ export class Workflow { return queue.getSchedules() } + /** Point-in-time queue-depth gauges (`active`/`waiting`/`delayed`) for this workflow. */ + async getMetrics() { + const queue = await this.getQueue() + return queue.getMetrics() + } + private setupMetrics( queue: WorkflowQueueInternal, { meter, prefix }: { meter: Meter; prefix: string }, @@ -338,7 +378,7 @@ export class Workflow { observableResult.observe(waitingJobsGauge, waiting, attributes) observableResult.observe(delayedJobsGauge, delayed, attributes) } catch (err) { - Settings.logger?.error?.('Error collecting workflow metrics:', err) + this.logger?.error?.('Error collecting workflow metrics:', err) } }, [activeJobsGauge, waitingJobsGauge, delayedJobsGauge], diff --git a/tests/queue.test.ts b/tests/queue.test.ts index 696b2dd..15edd2c 100644 --- a/tests/queue.test.ts +++ b/tests/queue.test.ts @@ -8,6 +8,7 @@ import { expBackoff, JobAlreadyExistsError, Namespace, + NonRecoverableError, ResultExpiredError, TimeoutError, } from '../src/queue' @@ -891,6 +892,33 @@ test('maxAttempts exhaustion dead-letters to the failed ZSET', async () => { } }) +test('nonRecoverableError dead-letters immediately, skipping the remaining maxAttempts budget', async () => { + const prefix = randomUUID() + const wfId = randomUUID() + const ns = new Namespace({ id: randomUUID(), redis: await connect(), prefix }) + const queue = ns.queue({ id: wfId }) + + let calls = 0 + queue.worker(() => { + calls++ + throw new NonRecoverableError('bad payload') + }) + + try { + const { id } = await queue.add('x', { maxAttempts: 5 }) + await expect(queue.wait(id)).rejects.toThrow('bad payload') + + expect(calls).toBe(1) + expect(await redis.zscore(`${prefix}:${wfId}:failed`, id)).not.toBeNull() + expect(await redis.zscore(`${prefix}:${wfId}:delayed`, id)).toBeNull() + const hash = await redis.hgetall(`${prefix}:${wfId}:j:${id}`) + expect(hash.state).toBe('failed') + expect(hash.attempts).toBe('1') // burned one of five, then went terminal + } finally { + await ns.close() + } +}) + test('keepFailed count-trims the failed ZSET and DELs evicted job hashes', async () => { const prefix = randomUUID() const wfId = randomUUID() diff --git a/tests/workflow.test.ts b/tests/workflow.test.ts index 25f889c..3522c5b 100644 --- a/tests/workflow.test.ts +++ b/tests/workflow.test.ts @@ -3,20 +3,22 @@ import { randomUUID } from 'node:crypto' import { sleep } from '@antfu/utils' import { type } from 'arktype' import { beforeAll, describe, expect, test, vi } from 'vitest' -import { createRedis, ResultExpiredError, Settings, TimeoutError, WorkflowNamespace } from '../src' - -beforeAll(() => { - Settings.logger = console - Settings.defaultConnection = async () => - createRedis({ - host: 'localhost', - port: Number(process.env.REDIS_PORT), - }) +import { createRedis, ResultExpiredError, TimeoutError, WorkflowNamespace } from '../src' + +let sharedRedis: Awaited> + +beforeAll(async () => { + sharedRedis = await connect() }) /** Mint a fresh namespace so each test is key-isolated by its random workflow/namespace ids. */ function namespace() { - return new WorkflowNamespace({ id: randomUUID() }) + return new WorkflowNamespace({ + id: randomUUID(), + redis: sharedRedis, + logger: console, + autoClose: false, + }) } async function connect() { @@ -126,7 +128,13 @@ describe('wait', () => { const redis = await connect() const prefix = randomUUID() const wfId = randomUUID() - const ns = new WorkflowNamespace({ id: randomUUID(), redis, prefix }) + const ns = new WorkflowNamespace({ + id: randomUUID(), + redis, + prefix, + logger: console, + autoClose: false, + }) const workflow = ns.createWorkflow({ id: wfId, run: async () => 'x' }) try { @@ -302,6 +310,42 @@ describe('groups', () => { }) }) +test('job data that no longer matches the schema warns and fails without retrying', async () => { + const logger = { ...console, warn: vi.fn() } + const ns = new WorkflowNamespace({ + id: randomUUID(), + redis: sharedRedis, + logger, + autoClose: false, + }) + const wfId = randomUUID() + + // Enqueue under the old (permissive) schema, then work it under the new (stricter) one — + // exactly the shape of a job left in the queue across a schema change. + // A retry would re-read the same stored payload, so the budget must go unused. + const oldVersion = ns.createWorkflow({ + id: wfId, + run: async () => 'ok', + jobOptions: { maxAttempts: 5 }, + }) + const job = await oldVersion.run({ name: 123 }) + + const handler = vi.fn() + const newVersion = ns.createWorkflow({ + id: wfId, + schema: type({ name: 'string' }), + run: handler, + }) + await newVersion.work() + + await expect(job.wait(5000)).rejects.toThrow('Invalid workflow input') + expect(handler).not.toHaveBeenCalled() + expect(logger.warn).toHaveBeenCalledWith( + expect.stringContaining('does not match the workflow schema'), + expect.anything(), + ) +}) + test('numeric priority — higher runs first', async () => { const handler = vi.fn() const workflow = namespace().createWorkflow({ @@ -402,7 +446,13 @@ describe('durable sleep', () => { const redis = await connect() const prefix = randomUUID() const wfId = randomUUID() - const ns = new WorkflowNamespace({ id: randomUUID(), redis, prefix }) + const ns = new WorkflowNamespace({ + id: randomUUID(), + redis, + prefix, + logger: console, + autoClose: false, + }) const started = makeGate() let aborted = false let committed = false