Data · Preview

@wrnexus/queue

Background jobs with delay, concurrency, retry, and repetition.

Private registry access required

This package is not available from the public npm registry. After WorkRoot approves access and supplies private registry instructions, install the release-aligned package:

bun add @wrnexus/queue@0.2.59

Request preview access. Never put registry tokens in source control.

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 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().

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

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.

Complete TypeScript API

This declaration comes from the exact installed package and lists its exported functions, classes, interfaces, and types.

/**
 * @wrnexus/queue — a background job queue with delays, retries + backoff, and
 * concurrent workers. The default store is in-process; a pluggable driver lets
 * you back it with Redis/SQL for durability across restarts.
 *
 *   const queue = createQueue();
 *   queue.process("email", async (job) => { await send(job.data); });
 *   await queue.add("email", { to: "a@b.com" }, { delayMs: 5000, maxAttempts: 3 });
 *   queue.start();                 // begin polling; queue.stop() to halt
 *
 * Tests can drive it deterministically with `await queue.drain(now)`.
 */
interface Job<T = unknown> {
    id: string;
    name: string;
    data: T;
    attempts: number;
    maxAttempts: number;
    runAt: number;
    /** If set, re-enqueue this job this many ms after each successful run. */
    repeat?: number;
}
type JobHandler<T = unknown> = (job: Job<T>) => void | Promise<void>;
interface AddOptions {
    /** Delay before the job becomes runnable (ms). */
    delayMs?: number;
    /** Max attempts before it's dead-lettered. Default from queue options. */
    maxAttempts?: number;
    /** Re-enqueue this job this many ms after each successful run (recurring). */
    repeat?: number;
}
interface QueueOptions {
    /** Default max attempts per job. Default 3. */
    maxAttempts?: number;
    /** Base retry backoff (ms); doubles per attempt. Default 1000. */
    backoffMs?: number;
    /** Poll interval when started (ms). Default 250. */
    pollMs?: number;
    /** Called when a job exhausts its attempts. */
    onFailed?: (job: Job, error: unknown) => void;
    /** Clock injection (tests). Default Date.now. */
    now?: () => number;
}
interface Queue {
    add<T>(name: string, data: T, options?: AddOptions): Promise<Job<T>>;
    process<T>(name: string, handler: JobHandler<T>): void;
    /** Run every job whose runAt ≤ now, once. Returns how many ran. */
    drain(now?: number): Promise<number>;
    start(): void;
    stop(): void;
    size(): number;
}
declare function createQueue(options?: QueueOptions): Queue;

export { type AddOptions, type Job, type JobHandler, type Queue, type QueueOptions, createQueue };

Examples

Examples are taken from this package's installed documentation and must be evaluated with its requirements and stability notes.

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

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

Handling permanent failures

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

Deterministic testing

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