Queue Operations

Each poll-loop tick reaps stuck leases, claims due jobs up to the spare concurrency, re-validates their payloads, and runs the handlers — completing, retrying with backoff, or dead-lettering as each one resolves. A worker that dies mid-job simply lets the lease expire; another worker's reapStuck returns the job to pending.

import { createQueueWorker } from '@absolutejs/queue';

const worker = createQueueWorker({
  concurrency: 8,       // default 5
  leaseMs: 30_000,      // default 30s — stuck-lease reap window
  onError: (error, job) => {
    console.error('[queue]', job?.id, job?.kind, error);
  },
  pollIntervalMs: 1000, // default 1s between ticks
  registry,
  store,
  tracerProvider,       // optional — one OTel span per run
  workerId: 'worker-1', // default crypto.randomUUID()
});

worker.start();         // begin polling
await worker.runOnce(); // or drive a single tick yourself (tests, cron)
worker.drain();         // refuse new claims; in-flight handlers finish
await worker.stop();    // halt the loop; waits for active handlers

drain() is "stop accepting new work," not "halt the worker" — the poll loop keeps running so stuck-lease reaps continue, in-flight handlers finish, and no new jobs are claimed. stop() halts the loop and waits for active handlers, so the pair gives the worker a clean SIGTERM story.

The Elysia queue() plugin runs an in-process worker by default. Pass runWorker: false to keep only the enqueue surface in your HTTP layer and run the worker separately — the recommended split for prod: one HTTP fleet, one worker fleet, both sharing the durable store. runQueueWorker(options) is the standalone-entrypoint wrapper — it starts the worker and wires process signals to stop() followed by a clean exit.

// http.ts — HTTP fleet: enqueue surface only, no in-process worker.
const app = new Elysia().use(queue({ registry, runWorker: false, store }));

// worker.ts — worker fleet, scaled separately: bun run worker.ts
import { runQueueWorker } from '@absolutejs/queue';

runQueueWorker({
  registry,
  signals: ['SIGINT', 'SIGTERM'], // default — each wires stop() → exit(0)
  store,
});

createQueueRoutes({ prefix, store }) exposes the optional JobStore methods as HTTP endpoints — useful for an internal admin dashboard. Endpoints whose store method is missing respond 501 rather than failing at mount time.

import { createQueueRoutes } from '@absolutejs/queue';

const adminQueue = createQueueRoutes({ prefix: '/queue', store });

// The routes ship with no built-in auth — mount them behind your own
// guard (or keep them off the public listener entirely):
new Elysia()
  .onBeforeHandle(({ headers, status }) => {
    if (!isAdmin(headers)) return status(401);
  })
  .use(adminQueue);

The in-memory store and the Postgres adapter implement the full optional set, so every route lights up. The Redis adapter (0.0.1) adds get and countByStatus on top of the worker contract — list / cancel / retry respond 501 until a later release — and its countByStatus counts only pending and claimed jobs.

worker.metrics() returns a point-in-time snapshot plus cumulative counters since createQueueWorker(). Scrape it on a 30s interval — it pairs well with @absolutejs/metering for per-worker cost and throughput attribution.

Pass a tracerProvider (any @opentelemetry/api-compatible provider — see @absolutejs/telemetry for the type shape) and every job run is wrapped in a queue.runJob span carrying these attributes — when no provider is set, tracing is a zero-allocation noop:

On failure the handler's exception is recorded on the span and its status becomes ERROR — for retryable and dead-lettered outcomes alike. Each attempt gets its own span on its own run, with the attempt number in abs.job.attempt. Pair with @absolutejs/audit's recordQueueError helper for an audit-log trail of every failed job:

import { recordQueueError } from '@absolutejs/audit';

createQueueWorker({
  onError: recordQueueError(audit),
  registry,
  store,
});
// → 'queue.error' events with jobKind, attempts, and maxAttempts in
//   metadata and the job id as the audit target