260 lines
19 KiB
Plaintext
260 lines
19 KiB
Plaintext
page wrnexusqueue {
|
|
seo {
|
|
title = "@wrnexus/queue"
|
|
description = "Background jobs with delay, concurrency, retry, and repetition."
|
|
}
|
|
|
|
view {
|
|
<div class="docs-shell">
|
|
<SkipLink label="Skip to content" href="#main" class="docs-skip-link" />
|
|
<header class="topbar">
|
|
<a class="brand" href="/"><span>W</span> WRNexusJS</a>
|
|
<nav aria-label="Primary"><a href="/getting-started">Get started</a><a href="/packages">Packages</a><a href="https://component.wrnexusjs.dev/">Components</a><a href="/language">Language</a><a href="/architecture">Architecture</a></nav>
|
|
<div class="topbar-actions"><a class="preview-pill" href="/access">Private preview · v0.5.11</a><button data-wire-theme-toggle class="theme-button" aria-label="Toggle color theme" title="Toggle color theme">◐</button></div>
|
|
</header>
|
|
<div class="mobile-doc-nav"><details><summary>Browse documentation</summary><nav><a href="/getting-started">Get started</a><a href="/packages">Packages</a><a href="https://component.wrnexusjs.dev/">Components</a><a href="/language">Language</a><a href="/architecture">Architecture</a><a href="/tutorial">Tutorial</a><a href="/guides/project-structure">Guides</a><a href="/examples">Examples</a><a href="/search">Search</a></nav></details></div>
|
|
<main class="portal-main docs-layout">
|
|
<article id="main" class="documentation prose standalone package-document"><nav class="breadcrumbs" aria-label="Breadcrumb"><a href="/">Home</a><span>/</span><a href="/packages">Packages</a><span>/</span><span aria-current="page">@wrnexus/queue</span></nav><section class="doc-intro"><span class="eyebrow">Data · Package reference</span><h1>@wrnexus/queue</h1><p>Background jobs with delay, concurrency, retry, and repetition.</p><div class="doc-meta"><span>v0.5.11</span><span>Private registry</span><span>Data</span></div><section id="access" class="access-callout"><h2>Install the package</h2><p>After WorkRoot approves private registry access, install the release-aligned package:</p><pre><code>bun add @wrnexus/queue@0.5.11</code><button type="button" class="copy-button" aria-label="Copy installation command">Copy</button></pre><p><a href="/access">Request preview access</a>. Never put registry tokens in source control.</p></section></section><section id="guide"><blockquote>A background job queue with delays, retries + exponential backoff, recurring jobs, and concurrent workers.</blockquote>
|
|
<p>Part of the <strong>WRNexusJS</strong> framework — an SSR-first, Bun-native full-stack web framework.</p>
|
|
<h3 id="overview">Overview</h3>
|
|
<p><code>@wrnexus/queue</code> 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 <code>drain()</code>.</p>
|
|
<pre data-language="bash"><code>bun add @wrnexus/queue</code></pre>
|
|
<blockquote>Private package — the machine must be authenticated to the <code>wrnexus</code> npm org</blockquote>
|
|
<blockquote>(a read token in <code>~/.npmrc</code>). Requires <strong>Bun</strong> (Node is not supported).</blockquote>
|
|
<h3 id="api">API</h3>
|
|
<p>The package exports a single factory plus its supporting types.</p>
|
|
<h4 id="createqueue-options-queue"><code>createQueue(options?): Queue</code></h4>
|
|
<p>Creates a new queue instance.</p>
|
|
<pre data-language="ts"><code>function createQueue(options?: QueueOptions): Queue;</code></pre>
|
|
<h4 id="queueoptions"><code>QueueOptions</code></h4>
|
|
<div class="table-wrap"><table>
|
|
<thead><tr><th>Option</th><th>Type</th><th>Default</th><th>Description</th></tr></thead>
|
|
<tbody><tr><td><code>maxAttempts</code></td><td><code>number</code></td><td><code>3</code></td><td>Default max attempts per job before it is dead-lettered.</td></tr><tr><td><code>backoffMs</code></td><td><code>number</code></td><td><code>1000</code></td><td>Base retry backoff in ms; doubles per attempt.</td></tr><tr><td><code>pollMs</code></td><td><code>number</code></td><td><code>250</code></td><td>Poll interval used once <code>start()</code> is called (ms).</td></tr><tr><td><code>onFailed</code></td><td><code>(job: Job, error: unknown) => void</code></td><td>—</td><td>Called when a job exhausts its attempts.</td></tr><tr><td><code>now</code></td><td><code>() => number</code></td><td><code>Date.now</code></td><td>Clock injection for deterministic tests.</td></tr></tbody></table></div>
|
|
<h4 id="queue"><code>Queue</code></h4>
|
|
<p>The object returned by <code>createQueue</code>.</p>
|
|
<div class="table-wrap"><table>
|
|
<thead><tr><th>Method</th><th>Signature</th><th>Description</th></tr></thead>
|
|
<tbody><tr><td><code>add</code></td><td><code>add<T>(name, data: T, options?: AddOptions): Promise<Job<T>></code></td><td>Enqueue a job under a worker name. Returns the created job.</td></tr><tr><td><code>process</code></td><td><code>process<T>(name, handler: JobHandler<T>): void</code></td><td>Register the worker that runs jobs of the given name.</td></tr><tr><td><code>drain</code></td><td><code>drain(now?: number): Promise<number></code></td><td>Run every job whose <code>runAt ≤ now</code>, once. Returns how many ran.</td></tr><tr><td><code>start</code></td><td><code>start(): void</code></td><td>Begin polling every <code>pollMs</code>. No-op if already started.</td></tr><tr><td><code>stop</code></td><td><code>stop(): void</code></td><td>Stop the poll timer.</td></tr><tr><td><code>size</code></td><td><code>size(): number</code></td><td>Number of jobs currently queued.</td></tr></tbody></table></div>
|
|
<h4 id="addoptions"><code>AddOptions</code></h4>
|
|
<div class="table-wrap"><table>
|
|
<thead><tr><th>Option</th><th>Type</th><th>Description</th></tr></thead>
|
|
<tbody><tr><td><code>delayMs</code></td><td><code>number</code></td><td>Delay before the job becomes runnable (ms).</td></tr><tr><td><code>maxAttempts</code></td><td><code>number</code></td><td>Max attempts before dead-lettering. Defaults to the queue's <code>maxAttempts</code>.</td></tr><tr><td><code>repeat</code></td><td><code>number</code></td><td>Re-enqueue this job this many ms after each successful run (recurring).</td></tr></tbody></table></div>
|
|
<h4 id="jobhandler-t"><code>JobHandler<T></code></h4>
|
|
<pre data-language="ts"><code>type JobHandler<T = unknown> = (job: Job<T>) => void | Promise<void>;</code></pre>
|
|
<h4 id="job-t"><code>Job<T></code></h4>
|
|
<pre data-language="ts"><code>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
|
|
}</code></pre>
|
|
<h3 id="usage">Usage</h3>
|
|
<p>Register workers, enqueue jobs, then start the poller:</p>
|
|
<pre data-language="ts"><code>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</code></pre>
|
|
<h4 id="recurring-jobs">Recurring jobs</h4>
|
|
<p>Pass <code>repeat</code> to re-enqueue a job a fixed interval after each successful run:</p>
|
|
<pre data-language="ts"><code>queue.process("heartbeat", async () => ping());
|
|
await queue.add("heartbeat", {}, { repeat: 60_000 }); // runs ~every minute</code></pre>
|
|
<h4 id="handling-permanent-failures">Handling permanent failures</h4>
|
|
<p>When a job's <code>attempts</code> reaches <code>maxAttempts</code>, it is dropped and <code>onFailed</code> fires instead of retrying:</p>
|
|
<pre data-language="ts"><code>const queue = createQueue({
|
|
onFailed: (job, error) => {
|
|
console.error(`job ${job.id} (${job.name}) gave up`, error);
|
|
},
|
|
});</code></pre>
|
|
<h4 id="deterministic-testing">Deterministic testing</h4>
|
|
<p>Instead of <code>start()</code>, inject a clock and drive the queue with <code>drain()</code>:</p>
|
|
<pre data-language="ts"><code>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</code></pre>
|
|
<h3 id="retry-backoff-behavior">Retry & backoff behavior</h3>
|
|
<ul>
|
|
<li>On a thrown handler error, the job is retried while <code>attempts < maxAttempts</code>.</li>
|
|
<li>The next <code>runAt</code> is set to <code>now + backoffMs * 2^(attempts - 1)</code> (exponential</li>
|
|
<p>backoff): with <code>backoffMs: 1000</code> the delays are 1s, 2s, 4s, …</p>
|
|
<li>A job whose worker name has no registered handler stays queued until one is</li>
|
|
<p>registered (it is not counted as runnable by <code>drain</code>).</p>
|
|
<li><code>drain</code> is re-entrant-safe: overlapping calls are skipped while one is running.</li>
|
|
</ul>
|
|
<h3 id="requirements-notes">Requirements / Notes</h3>
|
|
<ul>
|
|
<li><strong>Bun-only</strong> runtime (Node is not supported), consistent with the rest of the</li>
|
|
<p>WRNexusJS framework. The queue itself relies only on standard timers (<code>setInterval</code>/<code>clearInterval</code>) and has no runtime dependencies.</p>
|
|
<li>The default store is in-process, so queued jobs do not survive a restart; a</li>
|
|
<p>pluggable driver is intended for backing it with Redis/SQL for durability.</p>
|
|
<li>Works alongside <code>@wrnexus/core</code> for offloading work from the request path.</li>
|
|
</ul></section><section id="api" class="api"><h2>Complete TypeScript API</h2><p>Generated from the exact installed package declarations.</p><pre data-language="typescript"><code>/**
|
|
* 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;
|
|
leaseMs?: number;
|
|
backoff?: (attempt: number) => number;
|
|
now?: () => number;
|
|
onDeadLetter?: (job: Job, error: unknown) => void | Promise<void>;
|
|
}
|
|
interface DurableQueue {
|
|
add<T>(name: string, data: T, options?: AddOptions): Promise<Job<T>>;
|
|
process<T>(name: string, handler: JobHandler<T>): void;
|
|
drain(): Promise<number>;
|
|
failed(): Job[];
|
|
retry(id: string): Promise<boolean>;
|
|
}
|
|
declare function createDurableQueue(options?: DurableQueueOptions): DurableQueue;
|
|
|
|
/**
|
|
* @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;
|
|
}
|
|
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;
|
|
/** 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;
|
|
/** 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;
|
|
get(id: string): Job | undefined;
|
|
list(name?: string): Job[];
|
|
cancel(id: string): 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 JobDefinition, type JobHandler, type Queue, type QueueOptions, type QueueStore, type WorkflowStep, createDurableQueue, createQueue, cronToInterval, defineJob, defineWorkflow, memoryQueueStore };
|
|
</code></pre></section><section id="examples" class="examples"><h2>Examples</h2><p>Copy-ready examples from the installed package documentation.</p><div class="example-grid"><article class="example-card"><h3>Register workers, enqueue jobs, then start the poller</h3><pre data-language="ts"><code>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</code></pre></article><article class="example-card"><h3>Recurring jobs</h3><pre data-language="ts"><code>queue.process("heartbeat", async () => ping());
|
|
await queue.add("heartbeat", {}, { repeat: 60_000 }); // runs ~every minute</code></pre></article><article class="example-card"><h3>Handling permanent failures</h3><pre data-language="ts"><code>const queue = createQueue({
|
|
onFailed: (job, error) => {
|
|
console.error(`job ${job.id} (${job.name}) gave up`, error);
|
|
},
|
|
});</code></pre></article><article class="example-card"><h3>Deterministic testing</h3><pre data-language="ts"><code>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</code></pre></article></div></section></article>
|
|
<aside class="on-this-page"><h2>On this page</h2><nav><a class="toc-level-2" href="#guide">Guide</a><a class="toc-level-3" href="#overview">Overview</a><a class="toc-level-3" href="#api">API</a><a class="toc-level-4" href="#createqueue-options-queue">createQueue(options?): Queue</a><a class="toc-level-4" href="#queueoptions">QueueOptions</a><a class="toc-level-4" href="#queue">Queue</a><a class="toc-level-4" href="#addoptions">AddOptions</a><a class="toc-level-4" href="#jobhandler-t">JobHandler<T></a><a class="toc-level-4" href="#job-t">Job<T></a><a class="toc-level-3" href="#usage">Usage</a><a class="toc-level-4" href="#recurring-jobs">Recurring jobs</a><a class="toc-level-4" href="#handling-permanent-failures">Handling permanent failures</a><a class="toc-level-4" href="#deterministic-testing">Deterministic testing</a><a class="toc-level-3" href="#retry-backoff-behavior">Retry & backoff behavior</a><a class="toc-level-3" href="#requirements-notes">Requirements / Notes</a><a class="toc-level-2" href="#api">Complete API</a><a class="toc-level-2" href="#examples">Examples</a></nav></aside>
|
|
</main>
|
|
<footer><div class="footer-brand"><span class="footer-mark" aria-hidden="true">W</span><p><strong>WRNexusJS 0.5.11</strong><span>Complete API documentation generated from installed package declarations.</span></p></div><nav aria-label="Footer"><a href="/packages">All packages</a><a href="/getting-started">Get started</a><a href="/security">Security</a><a href="/support">Support</a><a href="/llms.txt">AI guide</a></nav><p class="footer-meta">Private Developer Preview · Bun-native</p></footer>
|
|
<BackToTop />
|
|
</div>
|
|
}
|
|
}
|