@wrnexus/queue

A background job queue with delays, retries + exponential backoff, recurring jobs, and concurrent workers.

Part of the WRNexusJS framework — an SSR-first, Bun-native full-stack web framework.

Overview

@wrnexus/queue is a server-side durable job queue. You register named workers, enqueue jobs (optionally delayed or recurring), and let the queue poll and run them on a timer — with per-job retry limits and doubling backoff between attempts. defineQueue persists to .wrnexus/queue.sqlite by default, while low-level createQueue remains an intentionally in-memory primitive. Reach for it when you need to defer work (emails, webhooks, cleanup) off the request path without a heavyweight external broker. Tests can drive it deterministically via drain().

bun add @wrnexus/queue

Private package — the machine must be authenticated to the wrnexus npm org

(a read token in ~/.npmrc). Requires Bun (Node is not supported).

API

The package exports a single factory plus its supporting types.

createQueue(options?): Queue

Creates a new queue instance.

function createQueue(options?: QueueOptions): Queue;

QueueOptions

Queue

The object returned by createQueue.

AddOptions

JobHandler<T>

type JobHandler<T = unknown> = (
  job: Job<T>,
  context: { signal: AbortSignal },
) => void | Promise<void>;

Job<T>

interface Job<T = unknown> {
  id: string; // e.g. "job_1"
  name: string;
  data: T;
  attempts: number;
  maxAttempts: number;
  runAt: number; // epoch ms; job runs when now ≥ runAt
  repeat?: number; // if set, re-enqueue this many ms after each success
}

Usage

Application queues belong in app/queues/*.ts and default-export defineQueue(...). Configure persistence once in wrnexus.config.ts:

export default {
  queue: { storage: "database", databaseName: "default" },
  // Or omit queue entirely for .wrnexus/queue.sqlite.
};

Use storage: "memory" only for disposable work. A selected database must exist; WRNexusJS fails startup/first use with a clear error instead of silently falling back. Each job may declare success(data, job) and failed(data, error, job) lifecycle hooks.

Register workers, enqueue jobs, then start the poller:

import { createQueue } from "@wrnexus/queue";

const queue = createQueue({ maxAttempts: 3, backoffMs: 1000 });

// Register a worker for the "email" job name.
queue.process<{ to: string }>("email", async (job) => {
  await send(job.data.to);
});

// Enqueue a delayed job with up to 3 attempts.
await queue.add("email", { to: "a@b.com" }, { delayMs: 5000, maxAttempts: 3 });

queue.start(); // begin polling; queue.stop() to halt

Application queues with defineQueue

WRNexusJS applications can place typed queue modules in app/queues. The dev and production servers discover them, start them after runtime initialization, and shut them down gracefully:

import { defineQueue } from "@wrnexus/queue";

export default defineQueue({
  name: "email",
  jobs: {
    send: {
      maxAttempts: 3,
      idempotency: ({ messageId }: { messageId: number }) => `message:${messageId}`,
      run: async ({ messageId }) => sendMessage(messageId),
    },
  },
});

Callers get a typed producer and inspection helpers:

const job = await email.send.add({ messageId: 42 });
await email.send.status(job.id);
await email.send.cancel(job.id);
await email.send.retry(job.id);

createDurableQueue also exposes start(), stop(), isRunning(), health() and get() so applications do not need to build their own polling middleware.

SQLite

Use sqliteQueueStore(db) from @wrnexus/queue/sqlite with a WRNexusJS-compatible SQLite client. installSqliteQueueSchema(db) installs the queue table and its due-job and idempotency indexes.

Use context.signal in network/database calls so forced shutdown and active cancellation finish promptly. For process termination, prefer await queue.shutdown(); use { force: true } only after your grace period.

Durable queue

createDurableQueue({ store }) retains jobs until their handler succeeds and supports atomic leases when a driver implements QueueStore.claim. It exposes the same cancellation/shutdown behavior plus list, failed, and retry. The included memoryQueueStore() is useful for tests; production Redis/SQL drivers should make claim() atomic to prevent two workers executing one job.

Recurring jobs

Pass repeat to re-enqueue a job a fixed interval after each successful run:

queue.process("heartbeat", async () => ping());
await queue.add("heartbeat", {}, { repeat: 60_000 }); // runs ~every minute

Handling permanent failures

When a job's attempts reaches maxAttempts, it is dropped and onFailed fires instead of retrying:

const queue = createQueue({
  onFailed: (job, error) => {
    console.error(`job ${job.id} (${job.name}) gave up`, error);
  },
});

Deterministic testing

Instead of start(), inject a clock and drive the queue with drain():

let clock = 0;
const queue = createQueue({ now: () => clock });

queue.process("task", async () => {
  /* ... */
});
await queue.add("task", {}, { delayMs: 5000 });

clock = 5000;
const ran = await queue.drain(); // => 1

Durable workflows and approvals

createWorkflowEngine(store) executes dependency-ordered steps and persists every transition, result, progress update, failure, cancellation, and approval record. Approval steps pause safely and can resume after a process restart because the snapshot lives in the supplied WorkflowStore.

const workflow = defineDurableWorkflow({
  name: "publish-report",
  steps: [
    { name: "build", run: buildReport },
    { name: "approve", dependsOn: ["build"], approval: true, run: (report) => report },
    { name: "publish", dependsOn: ["approve"], run: publishReport },
  ],
});

const run = await engine.start(workflow, input);
await engine.approve(workflow, run.id, "approve", currentUser.id);

Use memoryWorkflowStore() for tests. Production stores implement the small get, put, and list contract using the same transactional database or durable service as the application.

Retry & backoff behavior

• On a thrown handler error, the job is retried while attempts < maxAttempts.
• The next runAt is set to now + backoffMs * 2^(attempts - 1) (exponential

backoff): with backoffMs: 1000 the delays are 1s, 2s, 4s, …

• A job whose worker name has no registered handler stays queued until one is

registered (it is not counted as runnable by drain).

• drain is re-entrant-safe: overlapping calls are skipped while one is running.

Requirements / Notes

• Bun-only runtime (Node is not supported), consistent with the rest of the

WRNexusJS framework. The queue itself relies only on standard timers (setInterval/clearInterval) and has no runtime dependencies.

• The default store is in-process, so queued jobs do not survive a restart; a

pluggable driver is intended for backing it with Redis/SQL for durability.

• Works alongside @wrnexus/core for offloading work from the request path.