page wrnexusqueue { seo { title = "@wrnexus/queue" description = "Background jobs with delay, concurrency, retry, and repetition." } view {
W WRNexusJS
Private preview · v0.8.8
Browse documentation
Data · Package reference

@wrnexus/queue

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

v0.8.8Private registryData

Install the package

After WorkRoot approves private registry access, install the release-aligned package:

bun add @wrnexus/queue@0.8.8

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

OptionTypeDefaultDescription
maxAttemptsnumber3Default max attempts per job before it is dead-lettered.
backoffMsnumber1000Base retry backoff in ms; doubles per attempt.
pollMsnumber250Poll interval used once start() is called (ms).
onFailed(job: Job, error: unknown) => void—Called when a job exhausts its attempts.
concurrencynumberunlimitedMaximum jobs claimed by one drain() call.
capacitynumberunlimitedMaximum queued plus active jobs before adds reject.
now() => numberDate.nowClock injection for deterministic tests.

Queue

The object returned by createQueue.

MethodSignatureDescription
addadd<T>(name, data: T, options?: AddOptions): Promise<Job<T>>Enqueue a job under a worker name. Returns the created job.
processprocess<T>(name, handler: JobHandler<T>): voidRegister the worker that runs jobs of the given name.
draindrain(now?: number): Promise<number>Run every job whose runAt ≤ now, once. Returns how many ran.
startstart(): voidBegin polling every pollMs. No-op if already started.
stopstop(): voidStop the poll timer.
shutdownshutdown({ force? }): Promise<void>Stop accepting jobs and await active work; force aborts it.
sizesize(): numberNumber of jobs currently queued.
get/listget(id) / list(name?)Inspect defensive copies of pending jobs.
cancelcancel(id): booleanRemove queued work or abort an active handler.
failedfailed(): Job[]Inspect exhausted jobs in the dead-letter collection.
retryretry(id): Promise<boolean>Reset and requeue a dead-lettered job.

AddOptions

OptionTypeDescription
delayMsnumberDelay before the job becomes runnable (ms).
maxAttemptsnumberMax attempts before dead-lettering. Defaults to the queue's maxAttempts.
repeatnumberRe-enqueue this job this many ms after each successful run (recurring).
prioritynumberHigher values are selected first among due jobs.
idempotencyKeystringReturn the matching pending job instead of enqueueing a duplicate.

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

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

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.

Complete TypeScript API

Generated from the exact installed package declarations.

import { ExecutionContext, Context } from '@wrnexus/core';
import { SubjectContext } from '@wrnexus/rpc';

/**
 * Persistence contract for the durable queue.
 *
 * Distributed drivers should implement `claim()` atomically and exclude leased
 * jobs from `due()` until their lease expires. Calling `put()` must replace the
 * stored record and release any previous lease for that job.
 */
interface QueueStore {
    put(job: Job): Promise<void>;
    get(id: string): Promise<Job | null>;
    remove(id: string): Promise<void>;
    due(now: number, limit: number): Promise<Job[]>;
    list(name?: string): Promise<Job[]>;
    claim?(id: string, worker: string, leaseUntil: number): Promise<boolean>;
}
declare function memoryQueueStore(): QueueStore;
interface DurableQueueOptions {
    store?: QueueStore;
    workerId?: string;
    maxAttempts?: number;
    concurrency?: number;
    capacity?: number;
    leaseMs?: number;
    backoff?: (attempt: number) => number;
    now?: () => number;
    onDeadLetter?: (job: Job, error: unknown) => void | Promise<void>;
    context?: (job: Job, signal: AbortSignal) => ExecutionContext;
}
interface DurableQueue {
    add<T>(name: string, data: T, options?: AddOptions): Promise<Job<T>>;
    process<T>(name: string, handler: JobHandler<T>): void;
    drain(): Promise<number>;
    cancel(id: string): Promise<boolean>;
    shutdown(options?: {
        force?: boolean;
    }): Promise<void>;
    list(name?: string): Promise<Job[]>;
    failed(): Job[];
    retry(id: string): Promise<boolean>;
}
declare function createDurableQueue(options?: DurableQueueOptions): DurableQueue;

interface RedisQueueClient {
    get(key: string): Promise<string | null>;
    set(key: string, value: string, options?: {
        NX?: boolean;
        PX?: number;
    }): Promise<unknown>;
    del(...keys: string[]): Promise<unknown>;
    zadd(key: string, score: number, member: string): Promise<unknown>;
    zrem(key: string, member: string): Promise<unknown>;
    zrangebyscore(key: string, min: number, max: number, options?: {
        limit: [number, number];
    }): Promise<string[]>;
    smembers(key: string): Promise<string[]>;
    sadd(key: string, member: string): Promise<unknown>;
    srem(key: string, member: string): Promise<unknown>;
}
/** Redis-backed queue store using only the common client command surface. */
declare function redisQueueStore(client: RedisQueueClient, prefix?: string): QueueStore;
interface SqlQueueClient {
    query<T = Record<string, unknown>>(sql: string, parameters?: unknown[]): Promise<{
        rows: T[];
    }>;
}
/** PostgreSQL store with atomic SKIP LOCKED leasing and JSON payloads. */
declare function postgresQueueStore(db: SqlQueueClient, table?: string): QueueStore;
declare const POSTGRES_QUEUE_SCHEMA = "CREATE TABLE IF NOT EXISTS wrnexus_jobs (\n  id text PRIMARY KEY, name text NOT NULL, payload jsonb NOT NULL, run_at bigint NOT NULL,\n  priority integer NOT NULL DEFAULT 0, lease_owner text, lease_until bigint\n); CREATE INDEX IF NOT EXISTS wrnexus_jobs_due ON wrnexus_jobs (run_at, priority DESC);";

interface ScheduledJob<T = unknown> {
    name: string;
    data: T;
    everyMs: number;
    options?: AddOptions;
}
interface QueueScheduler {
    start(): void;
    stop(): void;
    tick(now?: number): Promise<number>;
    snapshot(): {
        running: boolean;
        schedules: number;
        nextRuns: Record<string, number>;
    };
}
/** Restart-safe scheduler when used with a durable queue and stable idempotency buckets. */
declare function createQueueScheduler(queue: DurableQueue, schedules: ScheduledJob[], options?: {
    pollMs?: number;
    now?: () => number;
}): QueueScheduler;
declare function addBatch<T>(queue: DurableQueue, name: string, values: T[], options?: AddOptions): Promise<Job<T>[]>;
interface QueueDashboardSnapshot {
    generatedAt: number;
    pending: number;
    failed: number;
    byName: Record<string, number>;
    oldestRunAt?: number;
}
declare function queueDashboardSnapshot(queue: DurableQueue): Promise<QueueDashboardSnapshot>;
declare function renderQueueDashboard(snapshot: QueueDashboardSnapshot): string;
/** Long-running scheduler/worker loop suitable for a dedicated process or container. */
declare function runQueueDaemon(queue: DurableQueue, scheduler: QueueScheduler, options?: {
    signal?: AbortSignal;
    pollMs?: number;
    onError?: (error: unknown) => void;
}): Promise<void>;

type WorkflowStatus = "pending" | "running" | "waiting-approval" | "completed" | "failed" | "cancelled";
interface WorkflowStep$1<I = unknown, O = unknown> {
    name: string;
    dependsOn?: string[];
    approval?: boolean;
    run(input: I, context: WorkflowRunContext): O | Promise<O>;
}
interface WorkflowRunContext {
    workflowId: string;
    step: string;
    results: Readonly<Record<string, unknown>>;
    signal: AbortSignal;
    progress(value: number, message?: string): void;
}
interface WorkflowSnapshot {
    id: string;
    name: string;
    status: WorkflowStatus;
    input: unknown;
    results: Record<string, unknown>;
    completed: string[];
    waitingFor?: string;
    progress: number;
    message?: string;
    error?: string;
    updatedAt: number;
}
interface WorkflowStore {
    get(id: string): Promise<WorkflowSnapshot | null>;
    put(snapshot: WorkflowSnapshot): Promise<void>;
    list(): Promise<WorkflowSnapshot[]>;
}
declare function memoryWorkflowStore(): WorkflowStore;
interface WorkflowDefinition<I = unknown> {
    name: string;
    steps: WorkflowStep$1<any, any>[];
    /** Compile-time input marker; definitions do not store runtime input values. */
    readonly __input?: I;
}
interface WorkflowEngine {
    start<I>(definition: WorkflowDefinition<I>, input: I, id?: string): Promise<WorkflowSnapshot>;
    resume<I>(definition: WorkflowDefinition<I>, id: string): Promise<WorkflowSnapshot>;
    approve<I>(definition: WorkflowDefinition<I>, id: string, step: string, actor: string): Promise<WorkflowSnapshot>;
    cancel(id: string): Promise<boolean>;
    get(id: string): Promise<WorkflowSnapshot | null>;
    list(): Promise<WorkflowSnapshot[]>;
}
declare function createWorkflowEngine(store?: WorkflowStore): WorkflowEngine;
declare function defineDurableWorkflow<I>(definition: WorkflowDefinition<I>): WorkflowDefinition<I>;

interface SubjectEnvelope<T> {
    payload: T;
    identity?: string;
}
interface SubjectJob<T> extends Omit<Job<SubjectEnvelope<T>>, "data"> {
    data: T;
    subject?: SubjectContext;
}
interface SubjectQueue {
    add<T>(ctx: Context, name: string, data: T, options?: AddOptions): Promise<Job<SubjectEnvelope<T>>>;
    process<T>(name: string, handler: (job: SubjectJob<T>, context: {
        signal: AbortSignal;
    }) => void | Promise<void>): void;
}
/** Queue adapter that persists a signed end-user context alongside job data. */
/** Works with both the in-memory Queue and createDurableQueue(). */
declare function subjectQueue(queue: Queue | DurableQueue): SubjectQueue;

/**
 * @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;
    priority: number;
    idempotencyKey?: string;
    createdAt: number;
}
interface JobContext {
    /** Aborted when an active job is cancelled or the queue is force-stopped. */
    signal: AbortSignal;
    /** The same trusted context shape used by HTTP, actions, realtime and webhooks. */
    execution: ExecutionContext;
}
type JobHandler<T = unknown> = (job: Job<T>, context: JobContext) => 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;
    /** Higher-priority jobs run first when multiple jobs are due. */
    priority?: number;
    /** Prevent duplicate queued work with the same stable key. */
    idempotencyKey?: string;
}
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;
    /** Maximum jobs executed in one drain. Default: unlimited. */
    concurrency?: number;
    /** Maximum queued + active jobs. Adds reject once this limit is reached. */
    capacity?: number;
    /** Clock injection (tests). Default Date.now. */
    now?: () => number;
    context?: (job: Job, signal: AbortSignal) => ExecutionContext;
}
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;
    /** Stop accepting work and wait for active handlers (or abort them). */
    shutdown(options?: {
        force?: boolean;
    }): Promise<void>;
    size(): number;
    get(id: string): Job | undefined;
    list(name?: string): Job[];
    cancel(id: string): boolean;
    failed(): Job[];
    retry(id: string): Promise<boolean>;
}
interface JobDefinition<I> {
    name: string;
    options?: Omit<AddOptions, "idempotencyKey">;
    run: JobHandler<I>;
}
declare function defineJob<I>(definition: JobDefinition<I>): JobDefinition<I>;
interface WorkflowStep<I, O> {
    name: string;
    run(input: I): O | Promise<O>;
}
declare function defineWorkflow<T>(name: string, steps: Array<WorkflowStep<any, any>>): {
    name: string;
    steps: WorkflowStep<any, any>[];
    run(input: T): Promise<unknown>;
};
declare function cronToInterval(cron: string): number;
declare function createQueue(options?: QueueOptions): Queue;

export { type AddOptions, type DurableQueue, type DurableQueueOptions, type Job, type JobContext, type JobDefinition, type JobHandler, POSTGRES_QUEUE_SCHEMA, type Queue, type QueueDashboardSnapshot, type QueueOptions, type QueueScheduler, type QueueStore, type RedisQueueClient, type ScheduledJob, type SqlQueueClient, type SubjectJob, type SubjectQueue, type WorkflowDefinition, type WorkflowEngine, type WorkflowRunContext, type WorkflowSnapshot, type WorkflowStatus, type WorkflowStep, type WorkflowStore, addBatch, createDurableQueue, createQueue, createQueueScheduler, createWorkflowEngine, cronToInterval, defineDurableWorkflow, defineJob, defineWorkflow, memoryQueueStore, memoryWorkflowStore, postgresQueueStore, queueDashboardSnapshot, redisQueueStore, renderQueueDashboard, runQueueDaemon, subjectQueue };

Examples

Copy-ready examples from the installed package documentation.

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

Durable workflows and approvals

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);
} }