@wrnexus/queue is a server-side in-process 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. The default store lives in memory; the design allows a pluggable driver to back it with Redis/SQL for durability across restarts. 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().

Installation

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`

Option	Type	Default	Description
`maxAttempts`	`number`	`3`	Default max attempts per job before it is dead-lettered.
`backoffMs`	`number`	`1000`	Base retry backoff in ms; doubles per attempt.
`pollMs`	`number`	`250`	Poll interval used once `start()` is called (ms).
`onFailed`	`(job: Job, error: unknown) => void`	—	Called when a job exhausts its attempts.
`now`	`() => number`	`Date.now`	Clock injection for deterministic tests.

`Queue`

The object returned by createQueue.

Method	Signature	Description
`add`	`add<T>(name, data: T, options?: AddOptions): Promise<Job<T>>`	Enqueue a job under a worker name. Returns the created job.
`process`	`process<T>(name, handler: JobHandler<T>): void`	Register the worker that runs jobs of the given name.
`drain`	`drain(now?: number): Promise<number>`	Run every job whose `runAt ≤ now`, once. Returns how many ran.
`start`	`start(): void`	Begin polling every `pollMs`. No-op if already started.
`stop`	`stop(): void`	Stop the poll timer.
`size`	`size(): number`	Number of jobs currently queued.

`AddOptions`

Option	Type	Description
`delayMs`	`number`	Delay before the job becomes runnable (ms).
`maxAttempts`	`number`	Max attempts before dead-lettering. Defaults to the queue's `maxAttempts`.
`repeat`	`number`	Re-enqueue this job this many ms after each successful run (recurring).

`JobHandler<T>`

type JobHandler<T = unknown> = (job: Job<T>) => 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

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

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

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