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
• Install MariaDB locally

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.

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.

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.

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<string, unknown> {
	/** 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<WriteReportData> {
	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<void> {
		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.

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

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

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.

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.

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<string, unknown> {
	/** Stable application-owned report identifier. */
	reportId: string;
}

/**
 * Example application job restored from its JSON-safe queue payload.
 */
export class GenerateReportJob extends QueueableJob<GenerateReportJobData> {
	/**
	 * 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<void> {
		app().log.info({
			reportId: this.data.reportId,
		}, 'Generating report');
	}
}

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.

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<QueueJobId> {
	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<QueueProcessResult | null> {
	return app().queue.workNextJob();
}

{
	"jobId": 1,
	"status": "succeeded",
	"completedJobs": 1,
	"remainingJobs": 0
}

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.

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<QueueJobId> {
	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<QueueJobId[]> {
	return app().queue.batch(
		reportIds.map(reportId => new GenerateReportJob({ reportId })),
		{
			queue: 'reports',
		},
	);
}

{
	"pipeline": {
		"initiallyQueued": 1,
		"completedJobs": 2
	},
	"batch": {
		"jobIds": [
			3,
			4
		],
		"queuedJobs": 2,
		"queue": "reports"
	}
}

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.

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<QueueJobId> {
	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<QueueJobId> {
	return app().queue.retryFailed(failedJobId, {
		queue: 'reports',
	});
}

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

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

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.

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.

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.

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
• Schedule recurring reports
• Problem-solving recipe catalogue