Move a report out of the request
Accept the work quickly, persist its identity and let a registered worker produce the result. Keep access and progress in the application.
On this page
Source-backed MarkdownSet up the queue and storage
Use Installation’s independent app, tools and dedicated test SQL credentials. The lab creates QueuedJob and FailedJob tables plus a temporary storage root, then removes them. Production apps apply committed migrations and use storage shared by producer/worker processes.
Copy the complete example
This recipe reuses the maintained SQL report lab rather than a second version of the same job. All Queue examples are shipped with the package.
mkdir -p examples
cp -R node_modules/@db3.ai/app/src/queue/examples/. examples/Run it
Expect the wrong queue to be idle, the first missing-source attempt released, the last attempt failed, a linked replay and one report file containing Report: Three notes ready. The final queue is empty and the failed audit record remains.
npx tsx examples/runQueueReports.tsGive the work a stable identity
The job constructor validates its durable data when dispatching and when restoring. Use a report revision identity, not a request object, database connection, access token or submitted filesystem path.
The handler reads source bytes and replaces a stable result. Running it twice produces one output. That is a property of this operation, not a claim that queues execute exactly once. Email, billing and remote API calls need their own idempotency keys and recovery design.
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`);
}
}
Keep the request boundary small
A real request first authenticates, authorizes the report and validates its inputs. Save application-owned report status/ownership, then dispatch the job to the same named queue the worker consumes. Return an accepted response with the application report ID; a queue ID is operational metadata, not read permission.
If saving the report and dispatching must be atomic, use an application outbox or another explicit transactional handoff. A normal model transaction does not include file storage or every queue driver. This framework does not supply a general outbox API today.
The lab exercises producer-to-worker durability without an HTTP endpoint. The API guide supplies the request-validation and owner-scope pattern; an HTTP report-progress UI remains a follow-up.
Boot the worker with the same registrations
Register WriteReportJob in every worker boot and consume reports, not default. The producer is closed before a new worker App restores the saved job, so in-memory registration in the request is not mistaken for durability.
For a long-running worker, use startWorker() with process supervision and stop/drain before shutdown. A one-step call returning no job means none is ready on that queue now; it does not mean delayed work is complete.
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));
}
Repair the cause, then replay
The first run has missing source bytes. Queue records ordinary failure and retries according to its policy. The lab writes the missing source, calls retryFailed() and checks linkage to the retained failure record.
Keep result reads scoped to the application owner. Queue history, raw payloads and errors are privileged operations; do not expose them as an unauthenticated progress endpoint.
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 the test
Change the immutable report ID and source contents. Keep wrong-queue, failure, linked replay, one-output and empty-queue assertions. Repeat with your production driver in release verification.
npx vitest run tests/queue/runQueueReports.test.ts
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --skipLibCheck examples/*.tsCoverage and next steps
Tested: real SQL handoff across App instances, registration, queue selection, failure/repair/replay, worker drain and repeatable file output. Explained: HTTP ownership, report status, outbox, shared storage and production worker supervision.
Persists a report job, closes the producer, restores it in a fresh worker, repairs failure and repeats without duplicate output.
The guide test passes against the real framework components.packages/app/src/queue/tests/examples/runQueueReports.test.tsThis test command requires the framework repository. Use the walkthrough commands in an installed application.
Environment: Disposable SQL and local files; no Redis, external provider or HTTP server.