# Queue
> Create durable background jobs, run named workers, understand every attempt, compose chains and batches, and recover failures through one complete service guide.
- Package: `@db3.ai/app/queue`
- Canonical page: [https://db3.ai/docs/queue-overview](https://db3.ai/docs/queue-overview)
- Markdown: [https://db3.ai/docs/queue-overview.md](https://db3.ai/docs/queue-overview.md)
- Framework source of truth: `packages/app/src/queue/README.md`
## Workflow
1. **Model durable work** — Put JSON-safe application identity in a QueueableJob.
2. **Dispatch the job** — Choose its queue, delay, and retry policy.
3. **Run workers** — Register jobs and process the matching named queue.
## Start with a real queued report
A queue is useful when the request should finish before the work does. We will queue a report, close the producer, restore it in a fresh worker, fix a missing input and replay the failed job. Repeating it leaves one report, not two.
Complete Installation using Node.js 24, matching preview App/Pure tarballs, `tsx`, TypeScript and Vitest. These packages are not published yet. The isolated lab needs a local MariaDB server and a test account allowed to create and drop only `db3_app_test_*` databases. It never uses the starter’s notes database.
Use the test-only `.env` from Installation. `DATABASE_URL` overrides the separate `DB_*` variables, so unset it unless it deliberately targets your test server. The lab creates and removes its own database and temporary files; do not give it production credentials.
- [Install packages and configure test SQL](https://db3.ai/docs/installation.md#database-labs)
- [Install MariaDB locally](https://db3.ai/docs/starter-app.md#database)
## Copy the example
Run from your independent app directory. The files are shipped with `@db3.ai/app`; you do not need the framework repository. Keep the `outputs/` directory because the consumer test uses its expected result.
### Copy the shipped example
```bash
mkdir -p examples
cp -R node_modules/@db3.ai/app/src/queue/examples/. examples/
```
## Run the report lab
The first attempt is `released`, the second is `failed`. The lab then writes the missing source, replays the failure and drains a real polling worker. Its final result has `reportFiles: 1`, `remainingJobs: 0` and `retainedFailures: 1`.
The deliberate zero-second backoff is for this deterministic lab only. Use a non-zero delay and a bounded retry policy in an application. Nothing is sent to an external provider.
### Run the lab
```bash
npx tsx examples/runQueueReports.ts
```
## Make the job safe to repeat
The payload stores an immutable report revision, not a storage path or secret. The constructor validates that identity when dispatching and restoring. `handle()` reads the source and replaces one report file.
This makes repeated completion useful, but does not promise atomic file replacement or exactly-once side effects. A payment or email needs an application/provider idempotency key. Avoid a read-then-write “already processed” flag without a database constraint or transactional claim.
### examples/WriteReportJob.ts
```typescript
import { QueueableJob } from '@db3.ai/app/queue';
import { app } from '@db3.ai/app/server';
/** Immutable report revision selected and authorized by the application. */
export interface WriteReportData extends Record {
/** Safe storage identifier, including a revision when the source can change. */
reportId: string;
}
/** Rebuilds a local report without appending duplicate output on another attempt. */
export class WriteReportJob extends QueueableJob {
static readonly jobName = 'reports.write.v1';
/**
* Validates both newly dispatched and restored payloads.
*
* @param data - Trusted report identity; never a caller-supplied storage path.
*/
constructor(data: WriteReportData) {
if (!data || typeof data.reportId !== 'string' || !/^[a-z0-9][a-z0-9-]{0,63}$/.test(data.reportId)) {
throw new Error('reportId must contain 1–64 lowercase letters, digits or hyphens, starting with a letter or digit.');
}
super(data);
}
/**
* Reads the immutable source and replaces the report at its stable destination.
*
* Missing source bytes throw normally, allowing Queue to apply its retry policy.
* Replacement is repeatable, not an atomic file/database transaction or a
* substitute for provider idempotency when sending mail or charging money.
*/
async handle(): Promise {
const source = await app().storage.readToString(`sources/${this.data.reportId}.txt`);
await app().storage.write(`reports/${this.data.reportId}.txt`, `Report: ${source.trim()}\n`);
}
}
```
## Dispatch, repair and replay
The producer and worker are separate `App` instances sharing only SQL and configured storage. Dispatching a class auto-registers it in that process only. The worker must register it again.
`Database.install()` is confined to this disposable lab. It is not a production startup instruction. The worker awaits `stopAndDrain()` before the app and test database are closed.
### examples/runQueueReports.ts
```typescript
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { createGeneratedTestDatabase } from '@db3.ai/app/db/test/db';
import { FailedJob, QueuedJob } from '@db3.ai/app/queue';
import { App } from '@db3.ai/app/server';
import { WriteReportJob } from './WriteReportJob';
/**
* Runs durable dispatch, failure repair, replay and duplicate-safe output on SQL.
*
* The producer is closed before a fresh worker App restores persisted work.
* Only this disposable lab uses Database.install(); deployed apps use migrations.
* A zero retry delay keeps the failure demonstration deterministic without sleeps.
*
* @returns Stable observations also asserted by the shipped documentation test.
*/
export async function runQueueReports() {
const root = await mkdtemp(join(tmpdir(), 'db3-queue-guide-'));
try {
const database = await createGeneratedTestDatabase('queue_guide');
try {
const producer = new App({ db: database.db, queue: { driver: 'database', queue: 'reports', queueMonitor: false } });
try {
await producer.db.install(QueuedJob, FailedJob);
await producer.queue.dispatch(new WriteReportJob({ reportId: 'weekly-v1' }), {
maxTries: 2, backoff: { strategy: 'linear', initialSeconds: 0, maxSeconds: 0, jitter: false },
});
} finally {
await producer.close();
}
const consumer = new App({
db: database.db,
queue: { driver: 'database', queue: 'reports', queueMonitor: false },
storage: { disks: { local: { driver: 'local', root } } },
});
try {
consumer.queue.registerJob(WriteReportJob);
const wrongQueueIdle = await consumer.queue.workNextJob('default') === null;
const firstAttempt = await consumer.queue.workNextJob('reports');
const finalAttempt = await consumer.queue.workNextJob('reports');
const [failed] = await consumer.queue.failedJobs(10);
if (!failed) throw new Error('Expected the missing source to produce a persisted terminal failure.');
// Repair the cause before replaying. No queued payload or audit row is edited.
await consumer.storage.write('sources/weekly-v1.txt', 'Three notes ready');
const replacementId = await consumer.queue.retryFailed(failed.id);
const replacement = await QueuedJob.findOrFail(replacementId);
const replayLinked = replacement.payload?.retryOf?.failedJobId === failed.id;
const worker = consumer.queue.startWorker('reports', { force: true, maxJobsPerTick: 1 });
if (!worker) throw new Error('The report worker did not start.');
await worker.stopAndDrain();
await consumer.queue.dispatch(new WriteReportJob({ reportId: 'weekly-v1' }));
const repeated = await consumer.queue.workNextJob('reports');
const files = [];
for await (const file of consumer.storage.list('reports')) files.push(file);
return {
wrongQueueIdle, firstAttempt: firstAttempt?.status, finalAttempt: finalAttempt?.status,
replayLinked, repeated: repeated?.status,
text: await consumer.storage.readToString('reports/weekly-v1.txt'),
reportFiles: files.length, remainingJobs: await QueuedJob.query().count(),
retainedFailures: (await consumer.queue.failedJobs(10)).length,
};
} finally {
await consumer.close();
}
} finally {
await database.destroy();
}
} finally {
await rm(root, { recursive: true, force: true });
}
}
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
console.log(JSON.stringify(await runQueueReports(), null, 2));
}
```
#### Test-backed output
```json
{
"wrongQueueIdle": true,
"firstAttempt": "released",
"finalAttempt": "failed",
"replayLinked": true,
"repeated": "succeeded",
"text": "Report: Three notes ready\n",
"reportFiles": 1,
"remainingJobs": 0,
"retainedFailures": 1
}
```
## Add a worker to the starter
For the migration-first Notes + AI starter, add `QueuedJob` and `FailedJob` from `@db3.ai/app/queue` to the exported `models` array in `server/database/models.ts`, retaining all existing models. Run `npm run db:make:migration -- add_queue_tables`, inspect the generated migration, then run `npm run db:migrate` and `npm run db:check`.
The copied `reportConsole.ts` is a complete database-queue adapter. It loads your app-root `.env`, registers the report job and exposes the framework console. Run this path against your development app database, not the lab’s test-only `.env`.
Its local storage root is `data/queue-reports` or an absolute `REPORT_STORAGE_ROOT`. Both producer and worker must see the same root. Add `data/` and `.env` to `.gitignore`. The local-file example assumes one host or a shared volume; separate hosts need shared storage.
For your own feature, reuse your application bootstrap in the adapter so model registrations, selected driver, storage and configuration cannot drift. Do not start the HTTP server inside a worker.
- [Get the migration-first starter](https://db3.ai/docs/starter-app.md#create)
### examples/reportConsole.ts
```typescript
import 'dotenv/config';
import { resolve } from 'node:path';
import { pathToFileURL } from 'node:url';
import { runQueueConsole, type QueueConsoleOptions } from '@db3.ai/app/queue';
import { App } from '@db3.ai/app/server';
import { WriteReportJob } from './WriteReportJob';
/**
* Wires the public queue console to a bootstrapped application.
*
* @param application - Application with migrated queue tables and report storage.
* @returns CLI configuration; finite commands close the App, workers drain on signals.
*/
export function reportConsoleOptions(application: App): QueueConsoleOptions {
return {
app: () => application,
bootstrap: () => application.queue.registerJob(WriteReportJob),
commands: [{
command: 'queue:failed',
/** Lists bounded operational metadata without exposing payloads or exceptions. */
async run({ app }) {
const failures = await app.queue.failedJobs(25);
console.log(JSON.stringify(failures.map(failure => ({ id: failure.id, queue: failure.queue, job: failure.payload.job, failedAt: failure.failedAt })), null, 2));
},
}, {
command: 'report:seed',
/** Writes only the fixed sample input; never accepts a user-controlled file path. */
async run() {
await application.storage.write('sources/weekly-v1.txt', 'Three notes ready');
console.log('Sample source ready: sources/weekly-v1.txt');
},
}],
};
}
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
const application = new App({
queue: { driver: 'database', queue: 'reports' },
storage: { disks: { local: { driver: 'local', root: resolve(process.env.REPORT_STORAGE_ROOT || 'data/queue-reports') } } },
});
try {
await runQueueConsole(reportConsoleOptions(application));
} catch (error) {
await application.close();
throw error;
}
}
```
## Dispatch and run from the terminal
Run these commands from the app root, after migrating. `report:seed` writes only the fixed demo input; omit it to try a missing-file failure. `queue:failed` is an example-owned command that prints bounded failure metadata. The remaining commands come from `runQueueConsole()`.
Expect a “Dispatched” line, then “Processed” from the one-off worker. The file `data/queue-reports/reports/weekly-v1.txt` contains `Report: Three notes ready`. Without `--once`, leave the worker running in a separate terminal and stop it with Ctrl-C.
`--once` processes at most one available job, not the entire queue. `--max-jobs` is a sequential per-tick bound, not concurrency. Start additional supervised worker processes for parallelism. Queue command names are not globally installed executables.
### Your application, after migrating
```bash
npx tsx examples/reportConsole.ts report:seed
npx tsx examples/reportConsole.ts queue:dispatch reports.write.v1 '{"reportId":"weekly-v1"}' --queue=reports --tries=2
npx tsx examples/reportConsole.ts queue:work --queue=reports --once
npx tsx examples/reportConsole.ts queue:failed
# Continuous worker; Ctrl-C drains and exits.
npx tsx examples/reportConsole.ts queue:work --queue=reports --max-jobs=1
# After repairing a terminal failure, replace 123 with its failed-record id.
npx tsx examples/reportConsole.ts queue:retry 123 --queue=reports
```
## When to use Queue
Use Queue when work must survive the request that created it, may need retries, or should run in a separate worker process. Typical jobs include AI requests, email delivery, file processing, imports, and expensive report generation.
Use Events for immediate in-process reactions. Use Scheduler to decide when recurring work becomes due. Use Flows when several durable steps need explicit orchestration and observable state. Scheduler and Flows may dispatch Queue jobs, but they do not replace the Queue lifecycle.
### QueueableJob and QueueJob
QueueableJob is the application-authored definition: constructor data, serialization, handle logic, and optional failure hooks. QueueJob is the driver-created runtime record: queue identity, driver id, attempt count, and the durable payload envelope.
Keeping these roles separate gives job authors a small new `MyJob(data)` API while preserving the metadata workers need for claiming, retries, lease fencing, monitoring, and results.
### Drivers and named queues
The database driver is the straightforward default and stores active and failed jobs through the application database. The Redis driver keeps the same Queue API while using ready lists, delayed and reserved sets, a job hash, and persisted failure storage.
Named queues separate workloads that need different concurrency or resources. Dispatch to a name such as reports, then run workers for that same name with queue:work --queue=reports. A worker processes only its configured queue.
## Creating jobs
Application jobs extend QueueableJob and own one JSON-safe data object. Store stable identifiers and input values rather than open connections, service instances, request objects, streams, or other process-local state.
Validate durable data in the constructor. The same constructor runs after every rehydration, so corrupt or obsolete payloads fail before application work starts. Most jobs can use the inherited `toJSON()` and `fromJSON()` implementations.
### GenerateReportJob.ts
```typescript
import { app } from '@db3.ai/app/server';
import { QueueableJob } from '@db3.ai/app/queue';
/**
* Durable data required to generate one application report.
*/
export interface GenerateReportJobData extends Record {
/** Stable application-owned report identifier. */
reportId: string;
}
/**
* Example application job restored from its JSON-safe queue payload.
*/
export class GenerateReportJob extends QueueableJob {
/**
* Creates a report job after validating its durable payload.
*
* @param data - Report identity persisted with the queued job.
*/
constructor(data: GenerateReportJobData) {
if (typeof data.reportId !== 'string' || data.reportId.trim() === '') {
throw new Error('GenerateReportJob requires a report id.');
}
super(data);
}
/**
* Generates the report through services resolved from the active application.
*/
async handle(): Promise {
app().log.info({
reportId: this.data.reportId,
}, 'Generating report');
}
}
```
#### What this demonstrates
- The constructor validates the same durable data used after rehydration.
- Only JSON-safe application identity is persisted in the queue payload.
- `handle()` resolves normal services from the active application context.
### Register and dispatch
Register each job class during worker startup. Application code dispatches a constructed instance, so the durable payload and runtime object stay aligned. Dispatch also auto-registers the instance in the current process, but separate worker processes still need startup registration or a jobResolver.
### createAndProcessReportJob.ts
```typescript
import { app } from '@db3.ai/app/server';
import type { QueueJobId, QueueProcessResult } from '@db3.ai/app/queue';
import { GenerateReportJob } from './GenerateReportJob';
/**
* Registers the example job in a worker process.
*/
export function registerGenerateReportJob(): void {
app().queue.registerJob(GenerateReportJob);
}
/**
* Dispatches one report job through the active application queue.
*
* @param reportId - Stable application-owned report identifier.
* @returns Driver-owned queued job identifier.
*/
export async function dispatchGenerateReportJob(reportId: string): Promise {
return app().queue.dispatch(new GenerateReportJob({ reportId }));
}
/**
* Claims and processes the next job from the default queue.
*
* @returns Queue result, or null when no job is available.
*/
export async function processNextReportJob(): Promise {
return app().queue.workNextJob();
}
```
#### Test-backed output
```json
{
"jobId": 1,
"status": "succeeded",
"completedJobs": 1,
"remainingJobs": 0
}
```
#### What this demonstrates
- The worker registers the job class before it claims stored work.
- Dispatch persists the job data and returns the driver-owned id.
- `workNextJob()` restores a fresh instance and returns the attempt status.
### Dispatch options
DispatchOptions can select a named queue, delay initial availability, set maximum attempts, configure backoff, limit the retry window, and attach framework-owned origin metadata. Keep business identifiers inside job data and orchestration correlation inside origin.
Named queues matter operationally: a job dispatched to reports will wait until a reports worker is running. They are a capacity boundary, not merely a label.
## Job lifecycle
Queue serializes the QueueableJob into a payload containing its durable job name, display name, JSON-safe data, retry policy, UUID, optional origin, and any remaining chain. The driver adds its own record id, queue name, availability time, and reservation state.
The job class itself is never persisted. Deploy worker code that can resolve every durable job name currently present in storage before removing or renaming that class.
### Claim and rehydrate
A worker asks its driver for one available job. Claiming increments the attempt count and creates a time-limited reservation. Queue resolves the registered class and calls `fromJSON()` to create a fresh instance from payload.data.
Fresh instances prevent accidental process memory from leaking between attempts. Constructor validation, `handle()`, `onRetry()`, and `onFinalFailure()` operate on the same rehydrated instance for that attempt.
### Leases and heartbeats
Long-running handlers receive periodic lease heartbeats at roughly one third of retryAfterSeconds. Before Queue records success, release, deferral, or failure, it verifies that the same attempt still owns the durable record.
If ownership is lost, Queue returns lease_lost and does not mutate the newer owner. Treat that result as an ambiguity boundary: the handler may already have produced external effects even though this worker cannot commit the queue outcome.
### Attempt outcomes
Success dispatches the next chained job, deletes the completed record, and publishes a succeeded event. An ordinary error releases the job with its persisted backoff when more attempts remain. QueueRetryLaterError defers without consuming the attempt. Exhausted retries move the record into failed storage.
`workNextJob()` returns succeeded, released, deferred, failed, or lease_lost. It returns null only when no job was available on the requested queue.
## Chains and batches
Use one job when one retry boundary is enough. Use a chain when later work is invalid until earlier work succeeds. Use a batch when jobs are independent and parallelism is useful. Use a Flow when branches, nested steps, replay, or an inspectable graph are part of the product requirement.
### dispatchReportWorkflows.ts
```typescript
import { app } from '@db3.ai/app/server';
import type { QueueJobId } from '@db3.ai/app/queue';
import { GenerateReportJob } from './GenerateReportJob';
/**
* Dispatches report jobs as an ordered pipeline.
*
* Each job is persisted only after its predecessor succeeds. A terminal failure
* therefore prevents the remaining reports from being dispatched.
*
* @param reportIds - Report identifiers to process in order.
* @returns Driver-owned identifier for the first queued job.
*/
export async function dispatchReportPipeline(reportIds: string[]): Promise {
if (reportIds.length === 0) {
throw new Error('A report pipeline requires at least one report.');
}
return app().queue.chain(
reportIds.map(reportId => new GenerateReportJob({ reportId })),
{
queue: 'reports',
},
);
}
/**
* Dispatches report jobs independently so workers can process them in parallel.
*
* @param reportIds - Report identifiers that do not depend on one another.
* @returns Driver-owned identifiers for every queued job.
*/
export async function dispatchReportBatch(reportIds: string[]): Promise {
return app().queue.batch(
reportIds.map(reportId => new GenerateReportJob({ reportId })),
{
queue: 'reports',
},
);
}
```
#### Test-backed output
```json
{
"pipeline": {
"initiallyQueued": 1,
"completedJobs": 2
},
"batch": {
"jobIds": [
3,
4
],
"queuedJobs": 2,
"queue": "reports"
}
}
```
#### What this demonstrates
- The chain initially persists one job and dispatches its successor only after success.
- The batch persists both independent jobs immediately on the reports queue.
- Visible aggregate progress still belongs on an application model or Flow.
### Dependent work with chains
`queue.chain()` persists only the first job. Each successful job dispatches the next serialized job onto the same named queue. A released or deferred attempt pauses the chain, and a terminal failure prevents the remaining jobs from being dispatched.
Jobs may also call `chain()` on themselves or override `defaultChain()` when the sequence is intrinsic to that job type. Prefer the queue service call when application orchestration chooses the sequence.
The successor is dispatched before the predecessor is deleted. These are separate operations, so a crash or lost lease between them can duplicate the successor. Chains are not an atomic workflow handoff; make every step repeatable and reconcile important business state.
### Independent work with batches
`queue.batch()` dispatches every job independently and returns every driver-owned id. Workers can process the jobs in any order and failures do not prevent siblings from running.
A batch is dispatch convenience, not a durable batch aggregate. If the product must show total, processing, completed, and failed counts, store that state on an application model or use a Flow when the orchestration itself needs a durable record.
`batch()` pushes jobs sequentially, not in one transaction. If dispatch throws partway through, earlier jobs remain queued. It supplies no all-or-nothing enqueue, completion callback, cancellation or aggregate progress record.
### Register every chained job
All classes that can appear in a chain must be resolvable when a worker reaches them. Register those classes during boot or configure a jobResolver. This is especially important after deployment because chained payloads may outlive the process that dispatched them.
## Retries and failures
When `handle()` throws an ordinary error, Queue releases the job if `maxTries` and `retryUntil` still allow another attempt. The supported strategies are `linear` and `exponential`. For a constant delay, set `initialSeconds` and `maxSeconds` to the same value; there is no `fixed` strategy.
Use jitter when many jobs may fail together, and set retryUntilSeconds when an old result stops being useful even if attempts remain. Application handlers should throw the original meaningful error so failed storage and diagnostics remain useful.
### manageReportRetries.ts
```typescript
import { app } from '@db3.ai/app/server';
import { QueueRetryLaterError, type QueueJobId } from '@db3.ai/app/queue';
import { GenerateReportJob } from './GenerateReportJob';
/**
* Dispatches one report with an explicit ordinary-failure retry policy.
*
* @param reportId - Stable application-owned report identifier.
* @returns Driver-owned queued job identifier.
*/
export async function dispatchReportWithRetryPolicy(reportId: string): Promise {
return app().queue.dispatch(new GenerateReportJob({ reportId }), {
queue: 'reports',
maxTries: 8,
backoff: {
strategy: 'exponential',
initialSeconds: 15,
maxSeconds: 900,
jitter: true,
},
retryUntilSeconds: 21_600,
});
}
/**
* Defers the current job when a provider supplies an explicit retry delay.
*
* Throw this from a QueueableJob handler after detecting provider backpressure.
* Deferral restores the claimed attempt instead of consuming an ordinary retry.
*
* @param delaySeconds - Provider-requested delay before another attempt.
*/
export function retryReportAfterBackpressure(delaySeconds: number): never {
throw new QueueRetryLaterError(delaySeconds, 'The report provider requested backpressure.');
}
/**
* Replays one terminal report failure while preserving its audit record.
*
* @param failedJobId - Driver-owned identifier from failedJobs().
* @returns Driver-owned identifier for the replacement queue job.
*/
export async function retryFailedReportJob(failedJobId: QueueJobId): Promise {
return app().queue.retryFailed(failedJobId, {
queue: 'reports',
});
}
```
#### Test-backed output
```json
{
"policy": {
"queue": "reports",
"maxTries": 8,
"backoff": {
"strategy": "exponential",
"initialSeconds": 15,
"maxSeconds": 900,
"jitter": true
}
},
"backpressure": {
"error": "QueueRetryLaterError",
"delaySeconds": 45
},
"replay": {
"replacementId": 2,
"failedAuditRecords": 1,
"retryOf": {
"failedJobId": "failed_report_01",
"jobUuid": "failed-report-uuid"
}
}
}
```
#### What this demonstrates
- Ordinary failures consume attempts and use the persisted exponential backoff policy.
- QueueRetryLaterError carries provider backpressure without consuming an ordinary attempt.
- Replay creates a replacement job while preserving the original failed audit record.
### Provider backpressure
Throw QueueRetryLaterError when a provider explicitly asks the application to wait, such as a Retry-After response. Queue restores the claimed attempt and defers availability by the requested number of seconds.
Do not use deferral for ordinary bugs or unknown failures. A job that can defer forever needs an application-owned deadline or state check so provider outages do not create immortal work.
### Retry and final-failure hooks
Override `onRetry()` after Queue has safely released an ordinary failed attempt. Override `onFinalFailure()` after the driver has durably recorded terminal failure. Hook errors are reported through onLifecycleError and do not reverse the queue transition.
Use hooks for diagnostics or small application state updates. Keep the main recovery contract in durable application data rather than depending on a hook to make the queue transition valid.
### Inspect and replay failures
`failedJobs()` returns recent driver-owned terminal records without exposing database tables or Redis keys. `retryFailed()` dispatches a new active job with a new UUID and retryOf metadata pointing to the failed record.
The original failed record remains untouched for audit. Replaying is therefore a new attempt chain, not deletion or mutation of history. Confirm the underlying application state still permits the operation before replaying old work.
## Testing and module ownership
Queue owns its contracts, drivers, documentation, examples, behaviour tests, integration tests, fixtures, and test support. Application jobs remain in the consuming app because their payload and handle logic express product behaviour.
The examples on this page are imported from `packages/app/src/queue/examples`. Their displayed outputs live beside the source and are asserted by Queue-owned behaviour tests in the documentation maintenance gate. The website build alone checks source synchronization, not the whole framework runtime.
## Commit records before dispatching work
Do not dispatch a job from an uncommitted model transaction through the default `app().queue` and assume it shares that transaction. It has its own database connection. The worker may run before the data exists, and a rollback will not undo that dispatch.
The simple path is commit, then dispatch. There is still a crash window between those operations. When missing a job is unacceptable, write an application outbox record in the same transaction and publish it with reconciliation and idempotency. There is no built-in `afterCommit()` or transactional outbox API.
- [Scoped model transactions](https://db3.ai/docs/active-record-api.md#record-methods)
## Shut down and deploy workers
Use `stopAndDrain()` before closing the App. It finishes the active tick, which can include several sequential jobs. `stop()` only stops future polling; it does not abort a handler. Give the process supervisor enough grace time for that work.
`App.close()` does not stop queue workers or close a Redis queue driver. When using Redis, construct and retain the `RedisQueueDriver`, inject it into `App` queue options, then drain workers, call `driver.close()` and finally `application.close()`.
Restart workers after deploying new code. Keep stable `jobName` values such as `reports.write.v1`, and support old payloads while stored jobs still reference them. Do not remove old classes before queued and replayable work has been dealt with.
## Choose Redis when you need it
The database driver uses `QueuedJob` and `FailedJob` tables. Redis uses a job hash, ready lists, delayed/reserved sets and failed storage behind the same Queue API. Moving to Redis does not move already queued SQL work.
Use a dedicated namespace per app and environment, authentication/TLS as appropriate, and a persistence and eviction policy that cannot silently discard jobs. A namespace is not an authorization boundary. The current service suite can exercise real Redis; this report lab proves the database path only.
Keep a driver reference so the host can close it. For a console adapter, provide a close callback that closes that driver and then the App after the worker has drained.
### Redis configuration and explicit ownership
```typescript
import { RedisQueueDriver } from '@db3.ai/app/queue';
import { App } from '@db3.ai/app/server';
const driver = new RedisQueueDriver({
url: process.env.QUEUE_REDIS_URL,
keyPrefix: 'my-app:development:queue',
});
const application = new App({ queue: { driver, queue: 'reports' } });
// Register jobs before starting workers. Retain driver for shutdown.
// await worker.stopAndDrain();
// await driver.close();
// await application.close();
```
## When a job does not run
“No jobs available” can mean the wrong queue, a future `availableAt`, an active reservation or genuinely no work. Check the dispatch queue and worker `--queue` first. A delay is earliest availability, not an exact scheduled execution time.
“No handler registered” means the worker bootstrap cannot resolve the durable name. Register the class, deploy matching code, then inspect/replay a terminal failure. Missing `jobs` or `jobs_failed` tables mean queue migrations have not run on this database.
`retryAfterSeconds` is a renewable lease, not a handler timeout. Use explicit timeouts at external API boundaries. A blocked event loop can stop heartbeats. `lease_lost` does not prove the handler did nothing: reconcile side effects before retrying.
A delayed or terminally failed attempt can still produce a successful CLI exit. Inspect the result/logged status, not only the shell exit code. An infrastructure exception can reject `workNextJob()` rather than returning an attempt result.
## Track progress without leaking data
`queue.events.subscribe()` observes lifecycle transitions and returns an unsubscribe function. Listeners and job hooks are awaited after transitions; their failure cannot roll back the transition. Avoid slow observers that hold up workers.
Queue events, stored payloads and failed exceptions can contain sensitive application values. Keep secrets out of payloads and error messages, redact before shipping logs, restrict failure inspection/replay, and define retention. The example CLI lists identities only, not payloads or stack traces.
Successful jobs are removed from active storage. Store user-visible report status and output identity on an application model. `QueueMonitor` is development telemetry, not durable progress, an audit ledger or a substitute for alerts on backlog age and failed jobs.
## Testing
Create a `tests/queue` directory and save the test below as `runQueueReports.test.ts`. It imports the example you copied into `examples/`. Keep the same folder layout so the relative import resolves.
Run from the application root with the development dependencies from Installation. These are consumer tests, not commands that assume a framework checkout. This database lab needs the same test-only SQL credentials when run through Vitest.
### tests/queue/runQueueReports.test.ts
```typescript
import { expect, it } from 'vitest';
import { runQueueReports } from '../../examples/runQueueReports';
import { WriteReportJob } from '../../examples/WriteReportJob';
import expected from '../../examples/outputs/queue-reports.json';
it('restores SQL work in a new App, repairs a failure, drains a worker and avoids duplicate output', async () => {
expect(await runQueueReports()).toEqual(expected);
});
it('rejects unsafe report identities when constructing or restoring a job', () => {
for (const reportId of ['', '../secret', '/absolute', 'A'.repeat(65)]) {
expect(() => new WriteReportJob({ reportId })).toThrow('reportId');
expect(() => WriteReportJob.fromJSON({ reportId })).toThrow('reportId');
}
});
```
## Run and extend your tests
The copied test executes real SQL, local storage and Queue; it does not replace the driver with a fake. It also rejects unsafe report identities. Keep `examples/outputs/queue-reports.json` beside the example.
Change the source text and update the expected result. Add a test for an unregistered worker, a deleted source, a second report revision, or a retry that must not repeat an external effect. Mock only the external provider when you add one.
### Run your copied test and check types
```bash
npx vitest run tests/queue/runQueueReports.test.ts --maxWorkers=1
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --resolveJsonModule --skipLibCheck examples/*.ts
```
## Coverage and next steps
This walkthrough runs SQL persistence across App lifetimes, named queues, ordinary retry/exhaustion, failure inspection and replay, worker drain, repeatable output and invalid payload rejection. The smaller Queue examples on this page separately exercise chains, batches and backpressure using a deterministic test driver; hooks have additional service tests.
The reference lists current service, job, payload, retry, lifecycle and driver contracts. Redis conformance, lease fencing, console and worker behaviour also have service-owned tests; they are not all demonstrated by this lab.
Not provided: hard job cancellation/timeouts, built-in unique jobs, atomic chain handoff, first-class tracked batches, retry-all, automatic pruning or a transactional outbox. Follow-up recipes: provider-backed AI budgets, idempotent email delivery and failure-retention operations. Those remain TODO, not available APIs.
- [Queue API and driver contracts](https://db3.ai/docs/queue-api.md)
- [Schedule recurring reports](https://db3.ai/docs/scheduler.md)
- [Problem-solving recipe catalogue](https://db3.ai/docs/solve-a-problem.md)
## Behavioural verification
Restores persisted SQL work in a new App, repairs and replays a failed report, drains a worker and checks duplicate-safe output.
- Behaviour test: `packages/app/src/queue/tests/examples/runQueueReports.test.ts`
- Repository test command (framework checkout only): `npm run test:service --workspace @platform/app -- queue --maxWorkers=1`
- In an installed application, use the walkthrough commands instead of this repository test.
- Expected outcome: The guide test passes against the real framework components.
- Environment: Node.js 24, MariaDB and a test-only database account. Local files only; no AI key, email or remote provider.
## Related documentation
- [Queue API reference](https://db3.ai/docs/queue-api.md): Current emitted Queue contracts: options, job payloads, attempts, retries, workers, drivers and console integration.
- [Build your first app](https://db3.ai/docs/starter-app.md): Create an account, save a private note and summarise it with AI. Start with working application code you can change.
- [Scheduler](https://db3.ai/docs/scheduler.md): Decide when a daily task is due, claim its occurrence and hand expensive work to Queue. Inspect what happened without hiding schedule definitions in a database.
- [Storage](https://db3.ai/docs/storage.md): Write files to a named disk, stream large payloads and keep paths relative to storage. Add Media when files need durable identities and ownership metadata.
- [Build and operate queued work](https://db3.ai/docs/example-queue.md): Follow the complete imported Queue example set: define durable data, register and process a job, compose chains and batches, configure retry policy, defer backpressure, and replay terminal failures.
## Framework-owned source: `packages/app/src/queue/README.md`
This is the exact source document captured by the documentation build. Use it for detailed API and workflow guidance, subject to the public package exports and behavioural evidence identified above.
````markdown
# Queue
The installed-package chain/batch lab is
[runQueueWorkflows.ts](./examples/runQueueWorkflows.ts). Copy the shipped Queue
examples and run `npx tsx examples/runQueueWorkflows.ts` with dedicated SQL test
credentials. It rejects an empty pipeline, processes one chain successor at a
time and two independent batch jobs, then destroys its generated database.
`GenerateReportJob` only logs an identity; use the queued-report guide for a job
that reads and writes real report files. This lab does not prove atomic chain
handoff or exactly-once side effects.
## Provider backpressure lab
The shipped `examples/runBackpressure.ts` uses a real SQL queue and a controlled
provider readiness flag. It defers twice without spending a try, succeeds after
readiness changes and terminates expired work using an application deadline.
The companion `tests/examples/runBackpressure.test.ts` is consumer-copyable.
The zero-second delay is only for deterministic testing; production adapters
must validate provider delays and apply their own deadline/idempotency policy.
Move work out of a request without losing it when that process exits. Queue
persists JSON-safe payloads, restores a fresh job for each attempt and records
success, retry, deferral or terminal failure.
Import application APIs from `@db3.ai/app/queue`. Keep the job and its business
policy in your app. Queue owns dispatch, leases and retry transitions; Storage
owns bytes; your application owns authorization, progress and idempotency.
## Run a real report
Start with Node.js 24 and an independent ESM app from
[Installation](https://db3.ai/docs/installation). Before npm publication, install
matching App and Pure tarballs supplied by a maintainer. Add `tsx`,
`typescript`, `@types/node` and `vitest` as development dependencies.
The report lab needs local MariaDB and a dedicated test account that may create
and drop only `db3_app_test_*` databases. Do not use production credentials.
Save test-only settings in the app-root `.env`, exclude it from Git, and unset
`DATABASE_URL` unless it intentionally points at this test server:
```dotenv
DB_CONNECTION=mariadb
DB_HOST=127.0.0.1
DB_PORT=3306
DB_USER=your_test_user
DB_PASSWORD=your_test_password
DB_DATABASE=db3_app_test
DB_TEST_DATABASE_PREFIX=db3_app_test
```
Copy the shipped examples into your app and run:
```sh
mkdir -p examples
cp -R node_modules/@db3.ai/app/src/queue/examples/. examples/
npx tsx examples/runQueueReports.ts
```
The lab creates an isolated database and local temporary directory. It removes
both in `finally`; killing the process forcibly can leave temporary resources.
It does not touch the starter's notes database or call an AI/email provider.
Read these actual example sources:
- [WriteReportJob.ts](./examples/WriteReportJob.ts): validate a stable revision
identifier and replace its output instead of appending duplicate content.
- [runQueueReports.ts](./examples/runQueueReports.ts): close the producer, restore
SQL-backed work in a fresh App, repair a missing source and replay a failure.
- [queue-reports.json](./examples/outputs/queue-reports.json): asserted output.
- [reportConsole.ts](./examples/reportConsole.ts): application-owned CLI adapter.
Expected result:
```json
{
"wrongQueueIdle": true,
"firstAttempt": "released",
"finalAttempt": "failed",
"replayLinked": true,
"repeated": "succeeded",
"text": "Report: Three notes ready\n",
"reportFiles": 1,
"remainingJobs": 0,
"retainedFailures": 1
}
```
The missing source deliberately fails twice. The lab fixes the cause, replays
the failed record and drains a polling worker. The original failure remains for
audit. Another dispatch replaces the same report, leaving one output.
Zero-second backoff is used only to keep the test deterministic.
## Run it in the starter
The disposable lab uses `Database.install()`. Your app must use migrations.
In the Notes + AI starter, add `QueuedJob` and `FailedJob` from
`@db3.ai/app/queue` to the existing `models` array in
`server/database/models.ts`, retaining every current model:
```sh
npm run db:make:migration -- add_queue_tables
# Review the generated migration before applying it.
npm run db:migrate
npm run db:check
```
Now use your development app's database settings, not the test-only lab
configuration. The copied console adapter loads `.env`, registers
`WriteReportJob` and uses the database driver. Its local storage root is
`data/queue-reports`, or the absolute `REPORT_STORAGE_ROOT` you supply.
Exclude `data/` from Git. Producer and worker must share that storage.
```sh
npx tsx examples/reportConsole.ts report:seed
npx tsx examples/reportConsole.ts queue:dispatch reports.write.v1 '{"reportId":"weekly-v1"}' --queue=reports --tries=2
npx tsx examples/reportConsole.ts queue:work --queue=reports --once
npx tsx examples/reportConsole.ts queue:failed
```
Expect a dispatch ID, a successful processing line, and
`data/queue-reports/reports/weekly-v1.txt` containing
`Report: Three notes ready`. `report:seed` and `queue:failed` are example-owned
commands. They write fixed sample input and list safe, bounded failure metadata.
For a continuous worker, use a separate terminal:
```sh
npx tsx examples/reportConsole.ts queue:work --queue=reports --max-jobs=1
```
Ctrl-C drains the active tick before closing the App. To replay a repaired
terminal failure, use its failed-record ID, not its original active-job ID:
```sh
npx tsx examples/reportConsole.ts queue:retry 123 --queue=reports
```
Replace `123` with a real ID from `queue:failed`. Repeated replay calls each
create another job; replay is not deduplicated. In your own app, reuse the same
bootstrap/configuration for the HTTP process and worker without starting an HTTP
listener in the worker. These are adapter command names, not global executables.
## Design the payload
Extend `QueueableJob`, validate constructor data and implement `handle()`.
The inherited `toJSON()` and `fromJSON()` cover normal JSON data. Set an explicit
stable `jobName`, such as `reports.write.v1`; class names otherwise become
durable names. Keep old names/payload versions supported while stored work can
still reference them.
Store immutable identities or versioned inputs, not open connections, request
objects, API keys or large file bodies. Load current application data in the
handler and recheck permission/state if access can be revoked after dispatch.
Dispatching a constructed job auto-registers its class in that process only.
Every separate worker must call `registerJob()` for all reachable jobs or supply
a `jobResolver`. `registerHandler()` is available for a stable named function
handler; it does not supply class constructor validation automatically.
## Queue names, timing and capacity
A `reports` worker only claims `reports` jobs. `delaySeconds` sets earliest
availability, not a guaranteed execution time. `workNextJob()` processes at most
one available job and returns `null` when idle. Its statuses are `succeeded`,
`released`, `deferred`, `failed` and `lease_lost`.
`processNextJob()` returns whether any job was processed, not whether it succeeded.
`startWorker()` starts a polling loop. `workerEnabled: false` or
`QUEUE_WORKER=false` disables it unless forced. `maxJobsPerTick` bounds sequential
work per tick; it is not concurrency. Use more supervised worker processes for
parallelism. Polling defaults to 1000 ms and valid intervals are clamped to at
least 100 ms. `queue:work --once` checks once and exits, even if delayed work exists.
## Retries and repair
Ordinary errors consume attempts. Dispatch defaults are three attempts,
exponential backoff starting at 15 seconds, capped at 3600 seconds, with jitter
off unless changed in queue options/environment. The supported strategies are
`linear` and `exponential`. For constant delay set `initialSeconds` and
`maxSeconds` equal; there is no `fixed` strategy.
[manageReportRetries.ts](./examples/manageReportRetries.ts) shows bounded backoff,
provider backpressure and replay. `retryUntilSeconds` bounds scheduling another
ordinary retry; it is not an execution deadline and does not abort a running job.
Throw `QueueRetryLaterError` only for deliberate backpressure. It restores the
attempt and bypasses the ordinary retry window, so the app must own a separate
deadline to prevent immortal work.
`failedJobs(limit)` reads recent failures. `retryFailed(id, options)` creates a
new UUID linked by `payload.retryOf` and preserves the old failure. Repair the
cause and check that the operation is still authorized before replaying. The
replacement receives a fresh retry policy/window; the original deadline is not
a continuing replay cutoff.
`onRetry()` runs after release, `onFinalFailure()` after persisted failure.
Deliberate deferrals call neither. Hook/listener errors go to
`onLifecycleError` and cannot reverse the transition. Use durable application
state for important recovery, not a hook that must never fail.
## Leases are not exactly-once execution
Database and Redis drivers renew reservations during execution, approximately
every third of `retryAfterSeconds` (at least one second). They check ownership
again before the outcome transition. Mutations are fenced by job ID and attempt.
A crash, blocked event loop or lost connection can let another worker claim
the job after expiry. `lease_lost` means this worker cannot safely commit the
queue outcome; the handler may already have performed external work.
`retryAfterSeconds` is a lease duration, not a timeout or cancellation API.
Make side effects idempotent. The report example replaces one stable output for
immutable input, but that is not an atomic file/database transaction. Email and
payments need application/provider idempotency. Avoid a non-transactional
"already done" read followed by an unprotected write.
## Chains and batches
[dispatchReportWorkflows.ts](./examples/dispatchReportWorkflows.ts) demonstrates
`chain()` for ordered work and `batch()` for independent jobs. Register every
job that can appear in a chain, not just its first class.
A chain persists the first job with its remaining steps. Success dispatches the
next step before deleting the predecessor. Those transitions are not atomic: a
crash between them can duplicate the successor. Every step still needs
idempotency and important workflows need reconciliation.
Successors inherit queue, maximum attempts and backoff. They do not inherit the
root's origin, initial delay or ordinary retry deadline. A terminal failure
stops advancement. Replaying its stored failure can resume the remaining chain.
A batch dispatches sequentially and returns IDs; it is not an all-or-nothing
transaction or a tracked batch record. Earlier dispatches remain if a later one
fails. There are no built-in batch counters, cancellation or completion callbacks.
## Transactions and dispatch
The default `app().queue` does not join a surrounding ActiveRecord transaction.
Do not enqueue before commit and assume rollback removes the job or workers
cannot see it. Commit, then dispatch for the simple case.
That still leaves a crash window between commit and dispatch. For work that
must not be missed, write an application outbox in the same transaction and
publish it with idempotency and reconciliation. No built-in `afterCommit()` or
transactional outbox contract exists yet.
## Drivers and shutdown
The database driver needs migrated `QueuedJob` and `FailedJob` tables.
Redis stores jobs, ready/delayed/reserved state and failures under a configurable
prefix. Choosing Redis does not migrate queued SQL work.
For Redis, configure a dedicated app/environment prefix and an appropriate
persistence, eviction and authentication/TLS policy. A prefix is not access
control. Retain the concrete driver you inject:
```ts
import { RedisQueueDriver } from '@db3.ai/app/queue';
import { App } from '@db3.ai/app/server';
const driver = new RedisQueueDriver({
url: process.env.QUEUE_REDIS_URL,
keyPrefix: 'my-app:development:queue',
});
const application = new App({ queue: { driver, queue: 'reports' } });
```
Stop and drain workers first, then `await driver.close()`, then
`await application.close()`. `App.close()` does not stop Queue workers or close
the Redis Queue driver. Adapt a Redis console host's `close()` callback to close
both resources after worker drain.
`stop()` stops future polling; `stopAndDrain()` waits for the current tick,
which can include several jobs. Neither aborts a handler. Give supervisors enough
shutdown grace time and restart workers after code changes.
## Progress, monitoring and privacy
Use `queue.events.subscribe()` for lifecycle observations; it returns an
unsubscribe function. Listeners are awaited after transitions. Slow listeners
delay the worker; exceptions cannot undo already committed state.
Successful jobs leave active storage. Keep user-visible progress/output
identities on application models. `QueueMonitor` is development telemetry, not
durable history or an accounting ledger. Monitor backlog age, retries, terminal
failures and ambiguous lease losses.
Payloads, lifecycle events and stored exceptions can expose sensitive values.
Never queue secrets; redact before exporting telemetry, restrict access to
inspection/replay and define retention. There is no automatic failure pruning.
## Troubleshooting
- Idle worker: check the exact queue name, availability time and reservation.
- Missing handler: fix registration/deployed job name, then inspect and replay.
- Missing SQL table: apply queue migrations against this process's database.
- Missing file: ensure workers share the same storage and immutable source.
- Hanging shutdown: drain the active tick and close any owned Redis driver.
- `failed` status with exit code zero: finite console commands report outcomes
in logs; shell success does not guarantee job success.
- Infrastructure failure: `workNextJob()` can reject before returning a result.
Do not turn an unavailable database into a silent empty queue.
## Test the workflow
The [Queue guide](https://db3.ai/docs/queue-overview#testing) renders the exact
service-owned test. Save it as `tests/queue/runQueueReports.test.ts`, retaining
the example and `outputs/` directory above:
```sh
npx vitest run tests/queue/runQueueReports.test.ts --maxWorkers=1
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --resolveJsonModule --skipLibCheck examples/*.ts
```
It uses real SQL, files and Queue, covering failure/repair/replay, App lifetime,
worker drain, repeated output and invalid identities. Tests fail when SQL is
unavailable rather than skipping this documented path.
For source-repository maintenance:
```sh
npm run test:service --workspace packages/app -- queue --maxWorkers=1
npm run check --workspace packages/app
npm run framework:package:test
```
Older deterministic-driver tests execute the small chain/batch/backpressure
examples. Built-in-driver suites separately exercise SQL/Redis leases and
transitions. Redis tests may skip locally when unavailable; release verification
must set `TEST_REDIS_REQUIRED=1` and supply real infrastructure. A passed SQL
report lab is not proof of Redis operational readiness.
## Reference, coverage and follow-up
The [Queue API reference](https://db3.ai/docs/queue-api) renders freshly emitted
service, job, payload, retry, worker, driver, Redis and console contracts.
Use `@db3.ai/app/queue` imports, not internal declaration paths.
Taught and executed here: SQL persistence, registration, named queues, retries,
terminal history, repair/replay, worker drain, repeatable output and input
validation. Explained with separate tests/reference: leases, Redis, lifecycle
observers, hooks, chains, batches and console adapters.
Not implemented: hard job timeout/cancellation, built-in unique jobs, atomic
chain handoff, first-class tracked batches, retry-all, automatic pruning or a
transactional outbox. Follow-up walkthroughs: shared-storage deployment,
provider-backed AI budgets, idempotent email, and failure-retention operations.
Queue owns its contracts, drivers, examples and tests. It depends on Database,
Logging and App composition; Scheduler and Flows consume its lifecycle.
````
## Guidance for AI tools
Use the documented public import `@db3.ai/app/queue` and its exported types. Prefer the source-backed examples and behavioural outcomes above over invented APIs or source-relative internal imports.