--- title: "@octabits-io/queue" summary: "Durable background jobs on the Postgres you already run: Zod-validated payloads on both ends, per-job acking, retries into a dead-letter queue, a declarative queue definition with a scope seam for the handler, transactional enqueue on the caller's own transaction, and a dead-letter audit sink as a contract with an in-memory reference, a plain-SQL store and a Drizzle adapter." install: "pnpm add @octabits-io/queue" availability: "Available from November 2026" version: "0.8.0" license: "MIT" related: - name: "@octabits-io/postgres" url: "https://octabits.io/platform/postgres.md" - name: "@octabits-io/result" url: "https://octabits.io/platform/result.md" canonical_url: "https://octabits.io/platform/queue" --- # `@octabits-io/queue` Durable background jobs on the Postgres you already run. Payloads are Zod-validated when they are enqueued and again when a worker picks them up. Each job in a batch is acked on its own. Failures retry into a dead-letter queue. A declarative definition turns a name, a schema and a handler factory into an enqueuer, a worker and a dead-letter handler. A job can be enqueued on the caller's own transaction, so it exists only if the write that produced it commits. And there is a dead-letter audit sink: a contract with an in-memory reference, a plain-SQL store and a Drizzle adapter. pg-boss is the runtime, behind its own subpath; the logger, the tracer and the meter are structural seams. ```bash npm i @octabits-io/queue zod pg-boss # + drizzle-orm for ./drizzle ``` | Subpath | Runs on | Deps | Contents | |---|---|---|---| | `.` | any runtime | `zod` | the contract: `JobContext`, `JobHandler`, `QueueDomain`, `EnqueueOptions`, the error types, `SCHEMA_BASE_JOB_PAYLOAD` / `SCHEMA_SYSTEM_JOB_PAYLOAD` / `SCHEMA_SCOPED_JOB_PAYLOAD`, `JOB_TRACE_KEY`, `QueueScope` / `QueueScopeFactory`, the monitoring types and error factories, the structural `Logger` / `Tracer` / `Meter` seams, and the dead-letter audit contract (`JobAuditRecord`, `JobAuditStore` with `record` / `list`, `JobAuditPage` / `JobAuditEntry`, `createJobAuditStore`, `createInMemoryJobAuditStore`, `DlqAuditRecord` / `DlqAuditSink`) | | `./pg-boss` | Node, Bun | `pg-boss` (optional peer) | `createBossManager` (lifecycle + monitoring, with a `role` per process: `full`, `consumer`, `producer`, `migrator`), `installPgBossSchema` + `pgBossGrantsSql` (the owner's migration step of a two-role deployment), `createQueueDomain` (queue/worker/DLQ trio), `defineQueue` (the declarative layer) | | `./postgres` | any runtime | none (structural) | `createPostgresJobAuditStore` over any `query(text, params)` executor (`pg`, PGlite, postgres.js), `jobAuditDdl()` | | `./drizzle` | Node, Bun | `drizzle-orm` (optional peer), `@octabits-io/postgres` (peer) | `jobAuditColumns` (timestamps read back as ISO-8601 in UTC) + `createDrizzleJobAuditStore` — the same audit store for a host that already runs Drizzle | The only hard dependency is `@octabits-io/result`. A boundary lint confines pg-boss to `./pg-boss` and drizzle-orm to `./drizzle`, and refuses `@octabits-io/telemetry` and every sibling package. A type check with no Node typings keeps the root and `./postgres` runtime-agnostic. ## When not to use it - **Live updates to a browser.** A job runs once, later, on one worker; it does not reach the people looking at the page. Telling subscribers that something changed is an event — [`@octabits-io/events`](https://octabits.io/platform/events.md), emitted in the same transaction. - **A multi-step workflow.** Steps that depend on each other, durable sleeps, signals, compensation: that is a workflow engine ([`octaflow`](https://octabits.io/platform/octaflow.md)), which runs on the same pg-boss. - **Throughput beyond what your Postgres takes.** Jobs are rows. Thousands a second is fine; a firehose that would dominate the database's write load wants a broker. - **Strict ordering or exactly-once side effects.** Delivery is at-least-once with retries; a handler must be idempotent, and jobs are not processed in enqueue order across workers. - **A database that is not Postgres.** The contract is neutral, but the only runtime is pg-boss, and pg-boss is Postgres. That is the point, not an accident. ## The shape ``` producer Postgres (pg-boss schema) worker process ──────── ───────────────────────── ────────────── enqueue(payload, { db?, parent?, key? }) │ validate against the schema │ ensure queue + DLQ (pool, once) └─ INSERT job ──────────────────▶ ──── fetch batch ──▶ validate again (on `db` when given: handler(job) → Result the caller's transaction; │ failed, retries left ◀── err / throw with `key`: one waiting job per key, upserted) │ retries exhausted ▼ -dlq ─── DLQ handler ─▶ onDlq(scope, job) onDlqAudit(scope, record) ─▶ JobAuditStore ``` ## A queue in forty lines ```ts import { z } from 'zod'; import { ok, err } from '@octabits-io/result'; import { SCHEMA_SCOPED_JOB_PAYLOAD } from '@octabits-io/queue'; import { createBossManager, defineQueue } from '@octabits-io/queue/pg-boss'; import { createPostgresJobAuditStore } from '@octabits-io/queue/postgres'; const SCHEMA_EXPORT_JOB = SCHEMA_SCOPED_JOB_PAYLOAD.extend({ requestId: z.uuid() }); type ExportJob = z.infer; const audit = createPostgresJobAuditStore({ db: pool, scope: { column: 'org_id' } }); export const exportQueue = defineQueue({ name: 'export', // the DLQ is `export-dlq` schema: SCHEMA_EXPORT_JOB, config: { retryLimit: 3, retryDelay: 30, expireInSeconds: 300 }, resolveScopeKey: (data) => data.scopeKey, // binds the DLQ scope + the audit row's scope createHandler: ({ createSystemScope, logger }) => async (job) => { const scope = await createSystemScope(job.data.scopeKey); try { const done = await scope.resolve('exports').run(job.data.requestId); return done.ok ? ok(undefined) : err({ key: 'job_failed', message: done.error.message, jobId: job.id }); } finally { await scope.dispose(); } }, onDlqAudit: audit.onDlqAudit, }); const manager = createBossManager({ connectionString: process.env.DATABASE_URL!, logger }); await manager.start(); const boss = manager.getBoss(); await exportQueue.createWorker({ boss, logger }).startWorker({ createSystemScope }); await exportQueue.createDlqHandler({ boss, createSystemScope, logger }).start(); const { enqueue } = exportQueue.createEnqueuer({ boss }); await enqueue({ scopeKey: 'org_1', requestId: crypto.randomUUID() }); ``` `createSystemScope` is any `(scopeKey?) => Promise<{ resolve(key), dispose() }>` — a DI container's child scope, or a plain object in a test. `logger` is any object with `debug` / `info` / `warn` / `error(message, error?, attributes?)` / `child`. `createQueueDomain` is the layer underneath `defineQueue` when you want the trio without the declarative wrapper. What the queue guarantees, so a handler does not have to: - **Validation both ways.** A bad payload on enqueue returns a `PayloadValidationError` with the Zod issues. A bad payload that reaches a worker anyway — a foreign producer, a corrupt row — goes straight to the DLQ on its first attempt, because a retry cannot fix it. - **Per-job acking.** One failing job in a batch neither fails nor re-runs its batch-mates, and `job.retryCount` is the real count. - **Errors are values.** Every method returns a `Result`; a handler returns `err(...)` to fail a job. A throw is caught and treated the same. - **Permanent failures skip the retries.** A handler that returns `job_failed` with `permanent: true` — a refused credential, an address the provider rejects — goes to the DLQ on that attempt instead of spending `retryLimit` on something a retry cannot fix. - **Queues exist before use.** The queue and its DLQ are created and their settings synced on first use, memoized per domain; a failed ensure is retried, not cached. ## Enqueue on the caller's transaction A job for a write belongs to that write: if the order insert rolls back, the confirmation mail must not go out. Pass the transaction's connection as `db`: ```ts import { fromDrizzle } from 'pg-boss'; import { sql } from 'drizzle-orm'; await db.transaction(async (tx) => { await tx.insert(orders).values(order); const queued = await enqueue({ scopeKey, orderId: order.id }, { db: fromDrizzle(tx, sql) }); if (!queued.ok) throw new Error(queued.error.message); // rolls the order back too }); ``` `db` is structurally pg-boss's `Db` — one method, `executeSql(text, values) → { rows }` — so pg-boss's own adapters (`fromDrizzle`, `fromKysely`, `fromKnex`, `fromPrisma`) fit as-is, and so does a two-line wrapper over a `pg` client inside `BEGIN`: ```ts const client = await pool.connect(); await client.query('BEGIN'); await enqueue(payload, { db: { executeSql: (text, values) => client.query(text, values) } }); await client.query('COMMIT'); ``` Only the job's INSERT runs on that connection. The queue and its DLQ are ensured first, on the queue's own pool and outside the transaction: creating a queue is DDL, a rollback would undo it, and its locks could outlive the send. The ensure is memoized, so a transactional producer pays it once per process. The integration suite proves it against a real Postgres: - A rolled-back transaction leaves no job. - A committed one leaves exactly one. - A rollback on a queue's first-ever enqueue leaves the queue itself in place. On a database with one connection (PGlite), call `ensureQueue()` at boot, before any request can open a transaction and enqueue inside it. The ensure also reads the queue into pg-boss's queue cache, so the send inside the transaction only runs its INSERT. A send that has to look the queue up would wait on the one connection the transaction holds. `pglite-transaction.test.ts` holds this. `ensureQueue()` stays public for a producer that writes jobs some other way. ## One job per record The same work asked for twice while it is still waiting — two requests for one record, a retried HTTP call, a restart that re-sends its start-up work — should stay one job. Pass a key the producer owns: ```ts const queued = await enqueue({ scopeKey, checkId }, { key: `check-${checkId}` }); // queued.value: { jobId, queue, deduplicated } ``` With a key the send is pg-boss's `upsert` by `singletonKey`, one statement pair that is atomic under concurrent producers: | The newest job with that key is… | What happens | `deduplicated` | `jobId` | |---|---|---|---| | none, or `completed` / `cancelled` / `failed` | a new job is inserted | `false` | the new job | | `created` or `retry` | that job takes this payload and keeps its id, position and retry count | `true` | that job | | `active` | left alone — the running attempt keeps the payload it fetched — and a new job is inserted behind it | `false` | the new job | A `singletonKey` on a plain `send` dedupes nothing on a `standard` queue: pg-boss rejects a second job through its per-policy unique indexes (`short`, `exclusive`, `stately`), and a queue's policy is immutable. The upsert works on every policy, so the package never asks you to choose one. A host that created its queue as `key_strict_fifo` wants every job kept and ordered, not folded: enqueue there without `key` and set pg-boss's own `singletonKey` through the boss. The payload is still validated and the trace still rides along; a deduplicated job carries the later request's `traceparent`. The same key is the handle for what a host does later, without a pg-boss id stored beside its record: ```ts const live = await manager.findJobByKey('check', `check-${checkId}`); // ok(JobDetails | null): the newest job that is still created, retry or active, with its state const done = await manager.cancelByKey('check', `check-${checkId}`); // ok({ cancelled }): every created / retry job with that key; an active one is left to finish ``` `cancelByKey` reads the waiting jobs and then cancels them, two statements: a job a worker claims in between is cancelled the way `cancelJob` would cancel it. Inside a caller's transaction (`db`) a keyed enqueue rolls back with it like any other. `dedupe-key.test.ts` holds the table above on PGlite; the integration suite holds the lookup, the cancel and the rollback on a real Postgres. ## Traces across the queue Pass `telemetry: { tracer, meter? }` to `createQueueDomain`, or to `defineQueue`'s `createWorker` / `createEnqueuer`, and every job runs inside a `consumer` span named ` process`, handed to the handler as `job.span`. With a meter, `messaging.process.duration` and `messaging.client.consumed.messages` count every job by queue and `error.type`. `enqueue(payload, { parent })` writes the caller's W3C `traceparent` into the payload under `JOB_TRACE_KEY` (`$traceparent`) — pg-boss has no headers — and the worker strips it before the payload meets the schema, so the job's span is a child of the request that enqueued it. With telemetry on the enqueuer too, the enqueue itself is a `producer` span between the two. `Tracer`, `Span`, `Meter` and `Logger` are structural and declared here. [`@octabits-io/telemetry`](https://octabits.io/platform/telemetry.md) satisfies them unchanged, and a test pins that. A host on the OpenTelemetry SDK adapts in a dozen lines. ## The dead-letter audit sink Every dead-lettered job is logged first, then run through the optional `onDlq(scope, job, data)` hook, then handed to `onDlqAudit(scope, record)`. `JobAuditRecord` is discriminated on `validPayload`: - `true`: the typed `payload`, and `attemptCount` is the retry limit it exhausted — or, for a `permanent` failure (`errorMessage` says so), the attempts it made. - `false`: a schema-invalid payload, dead-lettered on its first attempt, with `rawPayload: unknown` and `attemptCount: 1`. A sink that throws is caught and logged, and the scope is always disposed. `DlqAuditRecord`, `DlqAuditRecordBase` and `DlqAuditSink` are the same types under the names `defineQueue` has always used. A store is the contract plus two primitives, `write(row)` and `read(query)`: ```ts import { createInMemoryJobAuditStore } from '@octabits-io/queue'; import { createPostgresJobAuditStore, jobAuditDdl } from '@octabits-io/queue/postgres'; import { createDrizzleJobAuditStore, jobAuditColumns } from '@octabits-io/queue/drizzle'; jobAuditDdl({ scope: { column: 'org_id' } }); // the table, for your migrations const sqlStore = createPostgresJobAuditStore({ db: pool, scope: { column: 'org_id' } }); const ormStore = createDrizzleJobAuditStore({ db, table: jobAuditLog, scope: { column: 'orgId' } }); ``` `store.record(record)` returns `Result<'recorded' | 'skipped_unscoped'>`. `store.onDlqAudit` is the same write shaped for the sink seam, and it throws on a storage failure: the seam returns `void`, and a swallowed error would be a silently lost audit trail. The scoping rules live in the contract, once: | `scope` | Behaviour | |---|---| | omitted | no scope column stamped; every record written | | `{ column }` | stamped from `record.scopeKey`; a record without one (a system queue that omits `resolveScopeKey`) is skipped → `skipped_unscoped` | | `{ column, value }` | the fixed value on every row; `record.scopeKey` ignored | ### Reading it back `store.list(options?)` returns `Result`: the dead letters, newest first, one page at a time. Read through the store rather than querying the table: the scoping rules, the cursor and the read form of a row are the contract's, the same on every adapter. | Option | Meaning | |---|---| | `queueName` | only rows for this queue | | `scopeKey` | only rows stamped with this value — honoured by a `{ column }` store alone (below) | | `limit` | page size; default 50, clamped to 1..200 | | `before` | keyset cursor: only rows with `id < before` — the previous page's `nextBefore` | A page is `{ items: JobAuditEntry[], nextBefore: number | null }`. `nextBefore` is `null` on the last page; the store fetches one row more than the page to know. A `JobAuditEntry` is the row as written plus what the table assigns: `id` (a number) and `createdAt`. Both timestamps read back in `toISOString()` form (UTC, milliseconds), whether the driver hands over a `Date` or Postgres' text form in any session time zone. `payload` comes back as stored: an object, `null`, or a top-level JSON string that stays a string — the failed job's data exactly as it was enqueued, names and addresses included where the job carried them, so a page that leaves the server goes through the same projection or redaction the job's own records would. `scopeValue` is on an entry only when the store has a scope column. Only `status = 'dead_letter'` rows are read, so a table shared with in-flight job states lists its dead letters alone. The scoping rules for reads mirror the write rules: | `scope` | `list` shows | |---|---| | omitted | every row; `options.scopeKey` ignored | | `{ column }` | the rows stamped with `options.scopeKey` when given, **every row when omitted** | | `{ column, value }` | the rows stamped with `value`, always; `options.scopeKey` ignored | A `{ column }` store cannot tell an operator's view from a scoped request. On a scoped request, pass the key, build a `{ column, value }` store for that request, or query as a role under a row-level-security policy on the scope column. The policy filters whatever the caller passes. A dead-letter page, with the worker's store writing and a store per request reading one scope: ```ts // The worker: one store for every scope, stamping each record's scopeKey. const audit = createPostgresJobAuditStore({ db: pool, scope: { column: 'org_id' } }); // GET /jobs?queue=export&before=812 — a store fixed to the caller's scope, so no option widens it. export async function listDeadLetters(request: Request, scopeKey: string): Promise { const store = createPostgresJobAuditStore({ db: pool, scope: { column: 'org_id', value: scopeKey } }); const params = new URL(request.url).searchParams; const before = params.get('before'); const page = await store.list({ queueName: params.get('queue') ?? undefined, before: before ? Number(before) : undefined }); if (!page.ok) return Response.json({ error: page.error }, { status: 500 }); // Project before it leaves the server: payload and errorMessage are the job's own data. return Response.json({ nextBefore: page.value.nextBefore, items: page.value.items.map(({ id, jobId, queueName, completedAt }) => ({ id, jobId, queueName, completedAt })) }); } ``` A hand-written store implements `list` too. `createJobAuditStore({ scope, write, read })` gives it the rules for free: `read(query)` receives the resolved filter, the page size plus one and the cursor, and returns the rows as the driver hands them over. One conformance suite runs the in-memory, plain-SQL and Drizzle stores against an embedded PGlite. A new adapter passes it or it is not an adapter. `jobAuditDdl()` and `jobAuditColumns` declare the same table column for column. The Drizzle `payload` column is a `jsonb` that does not re-parse a top-level JSON string, so a raw `"73235"` payload stays a string. In-memory stores share one table through `createInMemoryJobAuditStore({ sameTableAs })`, the way a writer and a reader share one in Postgres. ## Two database roles A deployment with one database role starts pg-boss as `role: 'full'` and is done: that process installs the `pgboss` schema, migrates it on an upgrade, consumes, supervises maintenance and runs the cron timekeeper. A deployment with two roles — an owner that runs the migrations once, and an application role that is `NOSUPERUSER NOBYPASSRLS` with no `CREATE` and runs the web processes and the worker (`infra/single-host/init/01-roles.sh`) — splits that across processes by `role`: | `role` | Installs / migrates | Consumes | Maintenance + timekeeper | Runs as | |---|---|---|---|---| | `'full'` | yes, on `start()` | yes | yes | the owner | | `'migrator'` | yes, on `start()`, then waits for the index builds it queued | no | no | the owner, in the migration step | | `'consumer'` | never — `start()` checks and refuses | yes | yes | the application role | | `'producer'` | never — `start()` checks and refuses | no | no | the application role | The order of start-up is the owner's migration step, then the application role's processes: ```ts // 1. The migration step, as the owner (DATABASE_MIGRATION_URL): the schema … const { version } = await installPgBossSchema({ connectionString: ownerUrl }); // 2. … the queues, which are DDL too (a table per queue) … const migrator = createBossManager({ connectionString: ownerUrl, role: 'migrator', logger }); await migrator.start(); for (const queue of [exportQueue, mailQueue]) await queue.createEnqueuer({ boss: migrator.getBoss() }).ensureQueue(); await migrator.stop(); // 3. … and the grants, after the queues, with the owner's SQL client. for (const statement of pgBossGrantsSql('my_app')) await owner.query(statement); // 4. The worker, as the application role (DATABASE_URL): consumes, never migrates. const worker = createBossManager({ connectionString: appUrl, role: 'consumer', logger }); await worker.start(); // 5. The web process, as the application role: sends only. const web = createBossManager({ connectionString: appUrl, role: 'producer', logger }); await web.start(); ``` What each `start()` does and refuses: - **`'migrator'`** installs a missing schema or migrates an outdated one, with pg-boss's own advisory lock against a second migrator. A migration hands its index builds (`CREATE INDEX CONCURRENTLY` on the job tables) to pg-boss's background maintenance, which only a process with migration rights runs — a cluster of consumers would leave them pending forever — so the migrator's `start()` waits until none is pending and fails the step, naming the statements, if one failed (`indexBuildTimeoutMs`, default ten minutes, bounds the wait). It starts no worker, no maintenance sweep and no timekeeper. `installPgBossSchema()` is that start, a read of the version reached and a stop, in one call; a schema already at the current version is left as it is, so the step is idempotent. - **`'consumer'`** consumes and runs maintenance and schedules exactly as `'full'` does. Its `start()` is a check: a schema that is missing or behind fails with `pg-boss cannot start with role 'consumer': pg-boss is not installed. A consumer never installs or migrates the schema — run installPgBossSchema() (or a process with role 'full') as the schema owner against this database first.`, never with `permission denied` halfway through DDL; pg-boss's message stays as the `cause`. - **`'producer'`** sends; the same check, the same error. - **`'full'`** does everything; it is the one role that must run as the owner. `pgBossGrantsSql(appRole, { schema })` is the four statements the application role needs — `USAGE` on the schema, `SELECT, INSERT, UPDATE, DELETE` on all its tables, `USAGE, SELECT` on its sequences, `EXECUTE` on its functions — as an array with no trailing semicolons, so a client that runs one statement per call and one that takes a script are both served; the role and the schema are checked to be plain identifiers. Nothing in it grants `CREATE`. Run it after the queues are created: `ON ALL TABLES` covers the tables that exist when it runs, and a queue is a table of its own, so a release that adds a queue runs the same statements again in its migration step. `apps/starter` grants its app role with the same function (`src/server/app.ts`). `roles.integration.test.ts` proves it against a Postgres with the two roles from the single-host init script: a consumer as the application role before the install fails with the error above; `installPgBossSchema()` as the owner returns the version and leaves no index build pending; a second run changes nothing; after the queues and the grants, a consumer and a producer as the application role — which still has no `CREATE` on `pgboss` — send, work and complete a job and keep a schedule. ## Moved here The queue lived in `@octabits-io/framework/queue` from July to 2026-09-13, and the audit contract in `@octabits-io/postgres/job-audit-store`. Each old path was a named re-export shim for one minor and has been removed: | Was | Is | |---|---| | `@octabits-io/framework/queue` | `@octabits-io/queue` (contract) + `@octabits-io/queue/pg-boss` (runtime) | | `@octabits-io/framework/drizzle/job-audit-store` | `@octabits-io/queue/drizzle` | | `@octabits-io/postgres/job-audit-store` | `@octabits-io/queue` + `@octabits-io/queue/postgres` | ## Stability Pre-1.0: patches never break, and a minor says in the changelog what it breaks. The payload schemas, `JOB_TRACE_KEY` and the DDL helpers are part of the surface. A field a schema adds is a field every producer sends, and a column a DDL helper emits is a column a store writes.