WRNexusJS/packages/queue/README.md

@wrnexus/queue

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

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

Overview

@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

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

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.