165 lines
6.4 KiB
Markdown
165 lines
6.4 KiB
Markdown
# @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
|
|
|
|
```bash
|
|
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.
|
|
|
|
```ts
|
|
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>`
|
|
|
|
```ts
|
|
type JobHandler<T = unknown> = (job: Job<T>) => void | Promise<void>;
|
|
```
|
|
|
|
#### `Job<T>`
|
|
|
|
```ts
|
|
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:
|
|
|
|
```ts
|
|
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:
|
|
|
|
```ts
|
|
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:
|
|
|
|
```ts
|
|
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()`:
|
|
|
|
```ts
|
|
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.
|