Queues

Durable, at-least-once job queues on embedded SQLite: leases, exponential backoff, priorities, delays, rate limits, batch consumers, and a dead-letter queue.

Basics#

emails.ts

const emails = app.queue<{ to: string; subject: string }>("emails", {
  concurrency: 10,
  retries: 5,
});

emails.process(async (job) => {
  // job: { id, queue, data, attempt, maxAttempts }
  await smtp.send(job.data);
  // resolve = ack (job deleted) · throw = nack (retry → DLQ)
});

await app.start();
await emails.push({ to: "a@b.co", subject: "Hi" });

Jobs are persisted before push() returns. A claimed job holds a lease; if the worker dies without acknowledging, the sweeper returns it to pending (or dead-letters it if the budget is spent). Delivery is at-least-once — design handlers to be idempotent.

Queue options#

Option	Type	Default	Description
concurrency	number \| { limit, key }	10	Max handler invocations in flight. A number caps the whole queue; { limit, key: (data) => string } caps in-flight per key — e.g. 1 per user/tenant.
retries	number	2	Retries after the first failed attempt — total attempts = `retries + 1`.
backoff	{ delay?, maxDelay? }	{ delay: "1s", maxDelay: "60s" }	Exponential backoff between retries (delay × 2^attempt, capped, ±25% jitter).
lease	Duration	"30s"	Crash-redelivery horizon. Renewed automatically at lease/3 while the handler runs.
poll	Duration	"250ms"	Poll interval for work pushed by other processes. Same-process pushes wake the dispatcher instantly.
batch	number	32	Max jobs claimed per storage round-trip.
rateLimit	{ max, per }	—	Token-bucket cap on job starts. See below.
maxPending	number	—	Backpressure: push() / pushBulk() throw QueueFullError once this many jobs are pending. Best-effort admission control; omit for unbounded.
schema	StandardSchemaV1<T>	—	Validated on push. zod 3.24+, valibot, arktype…

Durations everywhere accept number (ms) or strings: "250ms", "30s", "5m", "2h", "1d".

Pushing jobs#

// Single push — returns the job id.
const id = await emails.push({ to, subject });

// Bulk push — one transaction, one boundary crossing.
const ids = await emails.pushBulk(items);

// Options per push:
await emails.push(data, {
  delay: "5m",        // not claimable before now + 5m
  priority: 3,        // higher runs first (default 0)
  retries: 1,         // override the queue's retry budget for this job
});

Producer-only processes

// Producer-only process: define the queue, never call .process().
// Consumers run elsewhere against the same store.
const emails = app.queue("emails");
await app.start();
await emails.push({ to, subject });

A queue without a processor never claims — useful for web servers that only enqueue while dedicated workers consume.

Processing#

Attach .process() (or .processBatch()) before app.start() — registration is a startup-time act.
The handler resolving acknowledges the job (it's deleted). Throwing schedules a retry; the thrown message is stored as lastError.
job.attempt is 1-based and counts crashes too: if a worker dies mid-job, the redelivered job arrives with the failed attempt already counted.
Priority order is priority DESC, then FIFO within a priority (ids are time-sortable UUIDv7).

Retries & the dead-letter queue#

A job that exhausts its attempts moves to the DLQ instead of being lost:

// Inspect dead jobs (exhausted retry budget):
const dead = await emails.deadJobs();
// → [{ id, queue, data, attempt, lastError, createdAt }]

// Put them back with a fresh retry budget:
await emails.requeueDead();            // all of them
await emails.requeueDead([dead[0].id]); // or specific ids

// …or permanently discard them:
await emails.purgeDead();             // delete all dead jobs, returns the count

// Pause/resume claiming — e.g. during an incident or maintenance:
emails.pause();                       // stop claiming; in-flight jobs finish
emails.isPaused();                    // → true
emails.resume();                      // start claiming again

// Health:
await emails.pendingCount();  // claimable now or scheduled
await emails.activeCount();   // pending + currently running

requeueDead() resets the attempt counter — the job gets a full fresh budget.

Rate limiting#

const webhooks = app.queue("webhooks", {
  concurrency: 20,
  // Token bucket: at most 100 job starts per minute.
  // Burst capacity = max; refill is continuous.
  rateLimit: { max: 100, per: "1m" },
});

The token bucket lives in the Rust dispatcher: it caps job starts, refills continuously, returns unused tokens when a claim comes back short, and sleeps precisely until the next token instead of spinning. Combine with concurrency — rate limits cap throughput, concurrency caps parallelism.

Per-key concurrency#

// At most 1 job per user at a time; different users run in parallel.
const sync = app.queue<{ userId: string }>("sync", {
  concurrency: { limit: 1, key: (data) => data.userId },
});
sync.process(async (job) => { await syncUser(job.data.userId); });

// Pushing computes the key from the data automatically.
await sync.push({ userId: "u_42" });

The object form caps in-flight jobs per key — at most limit jobs sharing key(data) run at once, while different keys run in parallel. Enforced in the store at claim time, so the cap holds across processes and nodes (on Postgres, keyed claims for a queue serialize via an advisory lock to keep the count exact). The key is computed at push from the job data.

Fairness: add fair: trueto round-robin claims across the concurrency-key groups, so one busy tenant can't starve the others — the claim batch takes one job per key before a second from any.

Throttle#

// Smooth starts to at most 5 per second per user (every job runs, paced).
const notify = app.queue<{ userId: string }>("notify", {
  throttle: { key: (d) => d.userId, max: 5, per: "1s" },
});
await notify.push({ userId: "u_42" });

Throttle spreads where debounce drops: each push with a throttle.keyis scheduled after the key's last slot (spacing = per / max), smoothing starts to a steady per-key rate — every job runs, just paced. Enforced at push via a per-key cursor in the store, so the rate holds across processes.

Debounce#

// Collapse a burst: only the last push per key runs, after a quiet window.
const reindex = app.queue<{ docId: string }>("reindex", {
  debounce: { key: (d) => d.docId, window: "5s" },
});
// 100 edits to doc "42" in 5s → one reindex of "42" runs.
await reindex.push({ docId: "42" });

Each push with a debounce.key deletes any still-pending job with that key and reschedules window out — so a burst collapses to a single run once it goes quiet. Enforced at push in the store (atomic delete-then-insert), so it holds across processes. The key is computed from the job data.

Batch consumers#

const indexer = app.queue<Doc>("index", { concurrency: 2 });

// Up to 50 jobs per invocation. All-or-nothing: a throw retries
// every job in the batch (each on its own attempt budget).
indexer.processBatch(
  async (jobs) => {
    await search.bulkIndex(jobs.map((j) => j.data));
  },
  { size: 50 },
);

concurrency counts batch invocations, not individual jobs — the example above can have 100 jobs in flight (2 × 50).
Semantics are all-or-nothing per invocation, but each job tracks its own attempt budget, so a poisoned job eventually dead-letters alone.

Payload validation#

import { z } from "zod";

const Email = z.object({ to: z.string().email(), subject: z.string() });

// Any Standard Schema v1 library works (zod 3.24+, valibot, arktype).
const emails = app.queue("emails", { schema: Email });

await emails.push({ to: "not-an-email" });
// → throws: invalid payload for queue "emails": Invalid email

Crash recovery & shutdown#

Crash: leased jobs whose worker died are swept back to pending after the lease expires (sweep cadence: sweep option, default 5s) — verified by a real SIGKILL harness in CI.
Graceful stop: app.stop({ timeout: “30s” }) stops claiming, drains in-flight handlers, and reports whether the drain completed cleanly.
Signals: SIGINT/SIGTERM handlers are installed by default (handleSignals: false to opt out).

Queue throughput numbers and the methodology behind them are on the benchmarks page.