# Build and operate queued work

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

- Package: `@db3.ai/app/queue`
- Canonical page: [https://db3.ai/docs/example-queue](https://db3.ai/docs/example-queue)
- Markdown: [https://db3.ai/docs/example-queue.md](https://db3.ai/docs/example-queue.md)
- Framework source of truth: `packages/app/src/queue/examples`

## Workflow
1. **Define** — Create a validated, JSON-safe QueueableJob.
2. **Compose** — Dispatch one job, a chain, or an independent batch.
3. **Recover** — Apply backoff, defer backpressure, and replay failures.

<a id="setup"></a>

## Run the installed SQL example

Complete Installation’s tarball/development setup and dedicated SQL test account first. Copy the shipped Queue examples below. These are application helper functions, not self-starting commands; the runner constructs the App, installs a disposable schema, registers the job and cleans up.

The job logs a report identity. For real file-backed report generation, start with the Queue guide. The deterministic output panels farther down illustrate older contract tests; the SQL runner gives the actual independent-consumer result.

- [SQL prerequisites](https://db3.ai/docs/installation.md#database-labs)
- [Generate a file-backed report](https://db3.ai/docs/queue-overview.md)

### Copy the shipped example

```bash
mkdir -p examples
cp -R node_modules/@db3.ai/app/src/queue/examples/. examples/
```

<a id="run"></a>

## Compare chain and batch behavior

Run from your independent app root. Expect empty-chain rejection with zero writes, one initial chain job, one remaining successor after its predecessor, two immediate batch jobs, four successful outcomes and zero jobs left.

### Run the lab

```bash
npx tsx examples/runQueueWorkflows.ts
```

<a id="sql-runner"></a>

## Inspect the complete application runner

The bounded calls deliberately process this lab’s known jobs. A production queue needs registered workers and an application idempotency policy; this demonstration does not guarantee atomic chain handoff after a crash.

### examples/runQueueWorkflows.ts

```typescript
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 { registerGenerateReportJob } from './createAndProcessReportJob';
import { dispatchReportBatch, dispatchReportPipeline } from './dispatchReportWorkflows';

/**
 * Compares sequential chains with immediately queued batches using real SQL.
 *
 * The example job only logs a report identity. This lab proves orchestration,
 * not a reporting provider. It creates and destroys its own test database.
 *
 * @returns Stable queue counts, validation and processing observations.
 */
export async function runQueueWorkflows() {
	const database = await createGeneratedTestDatabase('queue_workflows');
	const application = new App({ db: database.db, queue: { queue: 'reports', queueMonitor: false } });
	try {
		application.log.level = 'silent';
		await application.db.install(QueuedJob, FailedJob);
		registerGenerateReportJob();
		let invalidRejected = false;
		try { await dispatchReportPipeline([]); } catch (error) { invalidRejected = error instanceof Error && error.message.includes('at least one report'); }
		const afterInvalid = await QueuedJob.query().count();
		await dispatchReportPipeline(['first', 'second']);
		const chainInitial = await QueuedJob.query().count();
		const first = await application.queue.workNextJob('reports');
		const chainRemaining = await QueuedJob.query().count();
		const second = await application.queue.workNextJob('reports');
		const batch = await dispatchReportBatch(['third', 'fourth']);
		const batchInitial = await QueuedJob.query().count();
		const third = await application.queue.workNextJob('reports');
		const fourth = await application.queue.workNextJob('reports');
		return { invalidRejected, afterInvalid, chainInitial, chainRemaining, batchInitial, batchIds: batch.length, statuses: [first?.status, second?.status, third?.status, fourth?.status], remaining: await QueuedJob.query().count() };
	} finally {
		try { await application.close(); } finally { await database.destroy(); }
	}
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
	console.log(JSON.stringify(await runQueueWorkflows(), null, 2));
}
```

<a id="testing"></a>

## Testing

Create a `tests/queue` directory and save the test below as `runQueueWorkflows.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/runQueueWorkflows.test.ts

```typescript
import { expect, it } from 'vitest';
import { runQueueWorkflows } from '../../examples/runQueueWorkflows';

it('rejects an empty chain, then processes sequential and independent jobs on real SQL', async () => {
	expect(await runQueueWorkflows()).toEqual({
		invalidRejected: true, afterInvalid: 0, chainInitial: 1, chainRemaining: 1,
		batchInitial: 2, batchIds: 2, statuses: ['succeeded', 'succeeded', 'succeeded', 'succeeded'], remaining: 0,
	});
});
```

<a id="run-tests"></a>

## Run and extend the consumer test

Save the exact test in the documented folder, then change the report identities or add one batch job. Keep the invalid-input and final-queue assertions.

### Run your copied test and check types

```bash
npx vitest run tests/queue/runQueueWorkflows.test.ts
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --skipLibCheck examples/*.ts
```

<a id="coverage"></a>

## Coverage and related recovery

This consumer scenario proves real SQL chaining, batch dispatch, queue counts and empty-input recovery. Advanced lease loss, retry state, failure retention and provider backpressure have their own tests and guides. It does not prove Redis conformance or deployment crash safety.

- [Backpressure and deadlines](https://db3.ai/docs/cookbook-retries.md)
- [Detailed Queue API](https://db3.ai/docs/queue-api.md)

<a id="example-job-definition"></a>

## 1. Define the job

The constructor validates the same durable data supplied by application code and restored by workers. `handle()` resolves ordinary services from the active application.

### 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<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');
	}
}
```

#### 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.

<a id="example-dispatch"></a>

## 2. Register, dispatch, and process

The worker process registers the class before it claims persisted work. The dispatching process constructs the job, and `workNextJob()` exposes the detailed attempt outcome.

### 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<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();
}
```

#### 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.

<a id="example-workflows"></a>

## 3. Chain dependent work or batch independent work

The pipeline persists the next report only after success. The batch queues every report immediately on the reports worker pool.

### 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<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',
		},
	);
}
```

#### 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.

<a id="example-recovery"></a>

## 4. Configure and recover retries

The retry example persists exponential backoff, distinguishes explicit provider backpressure, and replays a failed record without deleting its audit history.

### 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<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',
	});
}
```

#### 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.

## Behavioural verification
Rejects an empty chain, then processes sequential and independent jobs on real SQL.
- Behaviour test: `packages/app/src/queue/tests/examples/runQueueWorkflows.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: Disposable SQL; no external report provider.

## Related documentation
- [Queue](https://db3.ai/docs/queue-overview.md): Create durable background jobs, run named workers, understand every attempt, compose chains and batches, and recover failures through one complete service guide.

## 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.