feat!: namespace-owned configuration and lifecycle

README.md

@@ -1,65 +1,150 @@
 # @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
 
 ```bash
 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(),
   }),
   async run({ input, step }) {
     await step.do('send welcome', () => {
       console.log(`Welcome, ${input.name}! Timezone: ${input.timezone}`)
     })
 
     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

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": {

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'

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}`)

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'

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<unknown>
@@ -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()

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()

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)

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
 }
 

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> | 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)