Data

@wrnexus/queue

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

bun add @wrnexus/queue@0.2.15

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

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

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 is generated from the exact published 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

Copy-ready examples taken from this package's published documentation.

Example 1

bun add @wrnexus/queue

Example 2

function createQueue(options?: QueueOptions): Queue;

Example 3

type JobHandler<T = unknown> = (job: Job<T>) => void | Promise<void>;

Example 4

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
}