task-queue

A real async job queue: named handlers, a sliding concurrency limit, per-task retries with exponential backoff + jitter, a dead-letter list, priority ordering, and graceful drain — with an event emitter API.

Honest about what it is: this is an in-memory queue. State does not survive a process restart. If you need durability, either persist yourself via the store observer hook, or reach for a broker-backed queue (SQS, BullMQ+Redis, etc). This library's job is to be the best in-process concurrency/retry engine, not to replace a message broker.

Install

npm install task-queue

Quickstart

import { TaskQueue } from "task-queue";

const queue = new TaskQueue({ concurrency: 5, maxRetries: 3 });

queue.handle("send-email", async (payload) => {
  await sendEmail(payload.to, payload.body);
});

queue.on("task_done", (task) => console.log("done:", task.id));
queue.on("dead_letter", (entry) => console.error("gave up:", entry.task.id, entry.error));

queue.enqueue("send-email", { to: "[email protected]", body: "hi" }, { priority: 5 });

await queue.drain(); // wait for everything in-flight/queued to settle

API

new TaskQueue(options?)

| Option | Default | Description | |---|---|---| | concurrency | 5 | Max tasks running at once. A finished task immediately frees a slot for the next one — sliding, not batched. | | maxRetries | 3 | Default max retry attempts before dead-lettering. | | baseRetryDelay | 200 | Base backoff delay (ms). | | retryDecay | 2 | Backoff multiplier per attempt. | | maxRetryDelay | 10000 | Backoff ceiling (ms). | | jitter | 0.3 | Jitter fraction (0–1) randomized into each retry delay. | | store | — | Optional TaskStore observer for persistence/metrics (see below). |

Methods

• handle(type, handler) — register the async handler for a task type.
• enqueue(type, payload, options?) — options: { priority?, maxRetries? }. Returns the task id. Throws if the queue is draining.
• drain(): Promise<void> — stop accepting new enqueues, resolve once all queued + in-flight (+ pending retries) work has settled.
• pendingCount / runningCount — current queue depth.
• deadLetterList — array of { task, error, failedAt }.
• on(event, listener) / off(event, listener)

Events

• task_done(task)
• task_failed(task, error, willRetry)
• dead_letter(entry) — fired once a task exhausts its retries.
• drain() — fired when drain() resolves.

TaskStore (pluggable, optional)

interface TaskStore {
  onEnqueued?(task): void | Promise<void>;
  onCompleted?(task): void | Promise<void>;
  onFailed?(task, error, willRetry): void | Promise<void>;
  onDeadLettered?(entry): void | Promise<void>;
}

These are observer hooks, not a source of truth — the queue's actual scheduling state always lives in memory. Use them to mirror state into a database if you need to survive a restart.

Design notes

Concurrency is "sliding" rather than "batched": a completed task immediately calls back into the scheduler to pull the next pending task, instead of waiting for a whole batch of N to finish before starting the next N. That keeps throughput high when task durations vary. Retries use the same exponential-backoff-with-jitter shape you'd want for any distributed retry (jitter avoids synchronized retry storms), scoped per-task rather than per-queue so one task's retry schedule doesn't throttle unrelated work.

Sponsored by Ferrow

Part of the ferrow-toolkit collection · Sponsored by Ferrow