@kingsleyweb/bun-jobs

v3.0.0-rc.2

Published

18 hours ago

Background work for Bun: run JS/TS files on a schedule or on demand (child process, Worker or in-process) and queue/process jobs across processes and services, over memory, file, Redis, SQL (postgres/mysql/mariadb/sqlite) and MongoDB drivers

0High
0Medium
0Low

kingsloob1

@kingsleyweb/bun-jobs

Background work for Bun, built on @kingsleyweb/bun-common. Two subsystems share one storage contract and one required namespace:

Queue (BunQueue / BunQueueWorker). Queue and process many small jobs across processes and services. It supports priorities, delays, retries with backoff, per-attempt timeouts, stalled-job recovery, repeatable jobs, debounce and throttle, cluster-wide rate and concurrency limits, dead letters, flows (parents waiting on children), job logs, retention and events.
Runner (BunRunner). Run one JS/TS file on a schedule (cron with optional seconds, an interval, or a one-shot) or on demand. It runs in a child process, a Worker or in-process. Single-run mode holds a cluster-wide lock and queues extra triggers. A stuck run is killed with a close, then SIGTERM, then SIGKILL.
Drivers: memory, file, SQL (Bun.sql: SQLite, Postgres, MySQL, MariaDB), MongoDB and Redis, all behind one contract. Producers, consumers and runners in different processes can share a backend.

BunJobs ties them together for one service. It sets a namespace and a backend once, and every queue, worker and runner is derived from it. It also adds a registry of named jobs with a fluent builder that reads schedules written in words ("every 2 weeks starting next monday").

Requires Bun ≥ 1.4.2. The Redis and SQL drivers use RedisClient and Bun.sql APIs that 1.4.2 is the first release to declare.

Bun runs the shipped TypeScript source (main is lib/index.ts), so there is no build step on your side, and a runner's child-process and Worker bootstraps run straight from lib/runner/bootstrap/. Your type checker reads the built declarations in dts/ (types), so your compiler options never apply to this package's source. mongodb and chrono-node stay optional: no shipped declaration imports either.

Status

The package is being assembled in phases. What has landed on this branch, and is documented below:

the runner, with BunRunnerManager and per-run captured logs
the queue and worker, including repeatable jobs, debounce and throttle, limits, dead letters, flows, where attempts run (target) and job logs
the BunJobs context, the job registry and the builder with dates in words
JobsNotifier, one event stream per namespace
remote control from any process: pausing, stopping, starting and reconfiguring workers, and overriding a runner's executor and overlap settings
analytics: per-second and per-minute series of every queue's jobs, each runner's runs and durations, and each worker's jobs and busyness, served by the management API
the memory, file, SQL, MongoDB and Redis drivers, with schema sync for SQL and MongoDB

The generated API reference lands with the final phase. Until then this README, the JSDoc on every export, and the option tours are the reference.

Installation

bun add @kingsleyweb/bun-jobs

The memory, file, SQL and Redis drivers use only Bun's built-in clients. The remaining features rely on optional peer dependencies, which you install only if you use them:

| Peer | Range | Needed for | |---|---|---| | mongodb | >=6 | the MongoDB driver. Imported only when that driver connects. Needs MongoDB 4.2 or later, with a raised open-file limit. | | chrono-node | >=2.7.0 <3 | reading dates in words ("tomorrow at 9am", "every 2 weeks starting 1st december"). Durations ("5 minutes") and cron do not need it. You can also supply your own dateParser. | | @types/bun | >=1.4.2 | types. The package ships raw .ts, so your project compiles it. |

bun add mongodb        # MongoDB driver
bun add chrono-node    # dates in words

Quick start

A queue and a worker

import { BunQueue, BunQueueWorker, MemoryDriver } from "@kingsleyweb/bun-jobs";

interface Email {
  to: string;
}

// The memory driver keeps jobs inside the instance, so producer and worker
// must share one. On any other backend each side can build its own.
const driver = new MemoryDriver();
const namespace = "shop";

const emails = new BunQueue<Email, string>("emails", { namespace, driver });

const worker = new BunQueueWorker<Email, string>(
  "emails",
  async (job, ctx) => {
    await ctx.log(`sending to ${job.data.to}`);
    return `sent to ${job.data.to}`;
  },
  { namespace, driver, concurrency: 5 },
);

worker.on("completed", (job, result) => {
  console.log(job.id, result);
});
void worker.run(); // resolves once the worker is closed

await emails.add("welcome", { to: "[email protected]" }, { attempts: 3 });

The job registry

import { BunJobs } from "@kingsleyweb/bun-jobs";

const jobs = new BunJobs({
  namespace: "shop",
  driver: { type: "redis", url: "redis://localhost:6379" },
});

jobs.define<{ to: string }>(
  "sendEmail",
  async (job) => {
    await deliver(job.data.to);
  },
  { attempts: 5 },
);

await jobs.start({ concurrency: 2 }); // one worker for every defined name

await jobs.now("sendEmail", { to: "[email protected]" });
await jobs.run("sendEmail", { to: "[email protected]" }).in("5 minutes").start();
await jobs.schedule("sendEmail", { to: "[email protected]" }).every("1 day").start();

await jobs.close(); // stops workers and runners, closes the driver it built

A runner

// jobs/cleanup.ts
import { defineHandler } from "@kingsleyweb/bun-jobs";

export default defineHandler<{ days: number }, number>(async (ctx) => {
  ctx.logger.info("cleaning", { days: ctx.args.days });
  ctx.progress(50);
  return 42; // stored in the run's history
});

import { BunRunner } from "@kingsleyweb/bun-jobs";

const runner = new BunRunner({
  id: "cleanup",
  namespace: "shop",
  file: new URL("./jobs/cleanup.ts", import.meta.url),
  schedule: "0 */10 * * * *", // six fields: every ten minutes, at second 0
  args: { days: 30 },
});

runner.on("finished", (run, result) => {
  console.log(run.runId, result);
});
await runner.start();

await runner.trigger({ args: { days: 7 } }); // { outcome: "started", runId }

Concepts

Namespace. Every queue, worker and runner requires one. The same queue name or runner id in two namespaces refers to two separate things, so services sharing a backend cannot collide. BunJobs makes the namespace structural: set it once, and everything derived from the context uses it.
Queue. A queue is just a name inside a namespace, with no server and no registration step. Producers and workers find each other by naming the same namespace and queue on the same backend.
Driver. Where state lives. You can pass either an instance, which is shared and never closed by whoever borrowed it, or a config, which is built and closed by the object it was given to. A spawned child process can only receive a config.
Job states. A job moves through these states:
- waiting or delayed, then active;
- from active, to completed, to failed (an attempt failed and a retry is scheduled), or to dead (attempts exhausted, or unrecoverable);
- a flow parent sits in waiting-children until its children settle.
Maintenance needs no leader, and comes in two halves. Liveness — promoting delayed jobs, recovering stalled ones and healing flows — is every worker's concern and cannot be turned off: each is the only thing that moves a particular kind of stuck job, so a queue whose one worker skipped them would strand all three for ever. Housekeeping — pruning expired results, healing repeat series, sweeping stale debounce and throttle windows and the leftovers of workers that are gone — is on by default and is what maintenance: false opts out of; a worker that skips it costs the queue tidiness, not progress. Each operation in both halves is idempotent, so no single process is load-bearing. A worker's heartbeat record says whether it does the housekeeping half, as sweeps.
Maintenance is leased, not shared. Promoting delayed jobs is every worker's own business — it is what bounds its next wake-up — and so is redelivering a flow event it recorded. Everything else repairs the queue: recovering stalled jobs, pruning expired results, healing repeat series and flows, and sweeping stale debounce, throttle and worker-control entries. One worker per queue does those — whichever holds that cadence's sweep lease — so twenty workers make one pass rather than twenty. Every pass is idempotent and atomic at the driver, so the lease only stops duplicate work, never loss: a holder that is parked or closed hands over at once, one whose process dies is replaced within three cadences (90 s for the stalled sweeps at the defaults, three minutes for the minute ones), and no pass is ever skipped. A driver without queue state has no lease and every worker sweeps.
The two leases are contended for by different sets of workers, and that is deliberate. The stalled lease is contended for by every worker, including one with maintenance: false — contending for it is how a worker takes part in recovery, and it costs one queue-state read per stalledInterval. If an opt-out worker skipped it, a queue whose only worker sets maintenance: false would never recover a stalled job: the liveness bug the two halves exist to prevent, reached through the lease instead of the timer. The minute lease is contended for only by workers that do the housekeeping. An opt-out worker arms no minute timer, so a minute lease it won would be one it never swept under — strictly worse than standing aside. So maintenance: false costs one queue-state read per stalled interval and none per minute.
The holder's settings govern the sweeps, and the lease settles on the fastest sweeper. stalledInterval and maxStalledCount are read from the worker holding the lease, so a fleet whose workers are configured differently is swept on that worker's terms — and that worker is the one sweeping most often, not whoever won the startup race. A worker takes the lease from a holder whose stalledInterval is more than twice its own, so the queue is swept at the shortest stalledInterval among its workers: one deliberately set to 100 ms governs the queue even when a default 30 s one started first. Equal and near-equal cadences never take it from each other, and a slower worker can never take it back, so it settles once and stays. A lease is written for two of the holder's cadences, so that is also what a hand-over waits for — and it tightens with the fastest sweeper's interval, not the slowest. Only the stalled lease is ever taken this way: every worker runs the minute sweeps at the same cadence, so that lease has nothing to compare and stays with whoever took it.
When a delayed job runs. A worker promotes due delayed and retrying jobs whenever it runs out of work, and ends its idle wait when the next one is due; its promotion sweep also runs every pollInterval, at least once a second, whatever else the worker is doing. Between promotions a worker remembers when the next scheduled job is due, and does not ask the backend again before then. Anything this process schedules — added with a delay, rescheduled, or failed into a retry, through any queue, job or worker sharing the driver instance — clears that memory at once; a job another process (or another driver instance) schedules does not.
That memory saves reads, not time. Every case is bounded by the same sweep: a job that comes due earlier than the remembered time runs at most one sweep interval (a second by default) late, whoever scheduled it, and the same bound already applied to a delayed job added while a worker was waiting. Measured across the backends, a job scheduled by another instance and due immediately ran 472–512 ms late, and one scheduled through the worker's own instance 7–698 ms late: the same bound either way.

Queues and adding jobs

const mail = new BunQueue<Mail, void, "welcome" | "digest">("mail", {
  namespace: "account",
  driver: { type: "sql", url: "postgres://user:pass@db/jobs" },
  defaultJobOptions: { attempts: 3 },
});

const job = await mail.add("welcome", { userId: 7 }, { priority: -1, delay: 5_000 });
job.wasAdded; // false when `jobId` already existed

await mail.addBulk([
  { name: "welcome", data: { userId: 8 } },
  { name: "digest", data: { userId: 9 }, opts: { runAt: new Date("2026-12-01") } },
]);

The third type parameter narrows the job names that add accepts.

Examples:

Job options

Queue defaultJobOptions are merged under the options of each add(). The built-in defaults are exported as DEFAULT_JOB_OPTIONS.

| Option | Type | Default | Meaning | |---|---|---|---| | jobId | string | fresh id | The job's id, and also its idempotency key. Adding an existing id returns the stored job untouched (wasAdded: false) and emits duplicate. Checked — see What an id may be. | | priority | number | 0 | Lower runs first; ties break FIFO. Clamped to ±1,048,576. | | delay | number | 0 | Milliseconds before the job may run. | | runAt | Date \| number | | An absolute time the job may run. Takes precedence over delay. | | attempts | number | 1 | Total attempts, including the first. Must be a whole number ≥ 1. | | backoff | number \| JobBackoffOptions | exponential from 1s, max 5 min, jitter 0.1 | Delay between attempts. See Retries and backoff. | | timeout | number | 0 (none) | Per-attempt timeout in ms. The attempt's ctx.signal is aborted, and it fails with JobTimeoutError. | | removeOnComplete | Retention | { ttl: 86_400_000 } | How long a completed job is kept (see below). | | removeOnFail | Retention | false | The same, for a dead job. | | keepStacktraces | number | 5 | How many failure stack traces the record keeps. | | keepLogs | number | 1000 | How many log lines the job keeps, newest last. 0 keeps every line. | | deadLetter | string | | The queue, in the same namespace, that receives a copy of the job when it dies. Takes precedence over the worker's deadLetterQueue. | | debounce | { id, ttl } | | Keep one pending job per id. See Debounce and throttle. | | throttle | { id, ttl } | | At most one job per id per window. | | repeat | RepeatOptions | | Makes this a repeatable job. See Scheduling. | | ignoreFailure | boolean | false | For a flow child: if it fails for good, its parent carries on anyway. |

Retention (Retention = boolean | number | { count?, ttl? }) takes these forms:

true removes the job immediately;
false keeps it forever;
a number keeps that many jobs. On the SQL driver a count of 20 or more is enforced once every tenth of it (at most every 500 settles) per process, per queue and state, rather than on every settle, so up to that many extra jobs can be kept between sweeps (removeOnComplete: 1000 keeps at most about 1,099 per process). A smaller count is exact;
{ count, ttl } does both.

A ttl is enforced by the workers' housekeeping: every minute the worker holding the queue's minute-sweep lease — which only a worker with maintenance on ever contends for — sweeps expired jobs in batches of 100 while batches come back full, up to 5,000 jobs or 500 ms per tick, and runs a catch-up pass a second later while a backlog remains. On Redis the sweep also reaches expired jobs queued behind long-lived ones, resuming from a saved cursor.

A flow child is never removed before its parent has recorded its outcome.

What an id may be

Breaking change. Ids you choose are now checked, and a bad one throws ConfigError at the call that supplied it rather than being written. This applies to jobId, a flow child's jobId, debounce.id / throttle.id, and repeat.key (see below), checked before anything is written.

An id is rejected when it is:

| Rejected | Why | |---|---| | empty | an id is an identity, and "" names nothing | | longer than 191 characters | MySQL and MariaDB store ids as VARCHAR(191) and used to truncate silently, merging two jobs into one — after which neither id resolved | | a control character (C0 U+0000–U+001F, NUL included; DEL and C1 U+007F–U+009F) | PostgreSQL rejects NUL outright while every other driver accepted it, so the same id worked on one backend and failed on another | | starting with . | . and .. are directories, and a dotfile hides from the tools people use to inspect a queue's directory | | not well-formed UTF-16 (a lone surrogate) | it has no UTF-8 spelling, so a backend storing UTF-8 replaces it with U+FFFD — two different ids then collide — or refuses it |

Everything else is allowed, punctuation and unicode included — :, |, @, *, ,, spaces, emoji, and / all pass. The rule is a denylist on purpose: this package builds its own ids by joining text with : and |, and an ordinary cron series contributes spaces, *, , and @, so anything tidier would reject the library's own ids.

Slashes are allowed deliberately. It is tempting to refuse them because the file driver turns an id into a file name, but encodeName escapes every character, so a slash was never a path hazard there — what actually bounds a file name is the encoded byte length, which that driver checks itself (see below). Tenant-scoped ids like tenant/7 work, and the management API URL-encodes them (GET /queues/mail/jobs/tenant%2F7).

The cap is one number for every driver, set by the tightest of them. 191 characters is itself the widest a utf8mb4 id column can be while the composite claim index stays inside InnoDB's 3,072-byte key limit (191 × 4 × 4 + 12 = 3,068), so it is a ceiling rather than a preference.

Ids and names this package derives — a dead-letter copy's id, a repeat occurrence's, a debounced job's, a window pointer's name, a generated series key — are shortened to fit rather than refused, so a long but legal id never makes a later step throw. They are fitted to the tightest store: 191 characters and the file driver's 201-byte encoded budget below. Shortening keeps the start and appends ~ and a hash of the whole, so two long names stay distinct, and it is deterministic, so every process derives the same one.

One more limit, on the file driver only. The character cap does not bound a file name. encodeName expands per character — 9 bytes for a character outside Latin such as 漢, 12 for an emoji — so 120 perfectly legal characters can encode to over 1,000 against a budget of 201 (NAME_MAX 255, less 54 for the marker and temp-file suffixes). The file driver refuses such an id — and a queue-state name of yours, from setQueueState — before writing anything, naming the encoded byte size and the limit. Lowercase letters and digits encode to one byte each, so an id of those alone never comes near it; an uppercase letter takes two (the encoding is case-proof), so 101 capitals already exceed it. Repeat keys are exempt: the file driver fits their file names itself.

On MySQL and MariaDB, namespaces and queue names are limited to 191 characters — the width of those columns — although the shared check allows 200. A longer one throws ConfigError instead of being truncated, as it used to be. A long queue name also leaves less room in the 191-character key under which that driver stores a queue's state and repeat entries: names that no longer fit are fitted as above, and a queue name that leaves no room at all is refused with ConfigError.

Reserved queue-state names

Breaking change. Debounce and throttle pointers are now stored under __win: — __win:debounce:<id> and __win:throttle:<id> — and that prefix is reserved: setQueueState throws ConfigError for any name beginning with it. The library's own writes carry a private token that no option you can pass reproduces, so the reservation cannot be bypassed from outside.

This fixes a data-loss bug. Pointers used to be named debounce:<id>, a name an application could choose too, and the window sweep deleted whatever it found under it — so a debounce:-prefixed entry of your own silently vanished at the next sweep.

Migration: wait one window. Pointers written by an earlier version are orphaned under the old names. Nothing reads them, and each expires by its own TTL, so they clear themselves once every window that was open at upgrade has passed. Debounced adds during that time open a fresh window rather than joining the orphan, which at worst means one extra job per id.

Repeat keys you choose are namespaced

A repeat.key you supply is stored namespaced, as k:<your key>, so one job can no longer pass another job's generated key and take that series over. Generated keys — <name>|<schedule>|<start> — are unchanged, except that one longer than 191 characters is now shortened to fit (it is stored in a column of that width on MySQL and MariaDB). Such a series registers again under the shortened key on its next add, and the old one stays until removed.

A key you supply is checked by the same rules as an id (above), at add() and before anything is written, with one tighter bound: at most 189 characters, since it is stored with the two-character k: in front.

For almost everyone this is invisible. The prefix is hidden everywhere it would surface: job.repeatKey, listRepeatables()[].key, the repeatScheduled event payload, and the occurrence id (repeat:<key>:<runAt>). removeRepeatable() takes either spelling, and a series that is listed as the key you pass wins over one merely stored that way: with series keyed nightly and k:nightly, listed as exactly that, removeRepeatable("k:nightly") removes the second.

The one exception, and the reason for it: if your key itself contains |, the prefix stays visible. A generated key always contains |, so a key like report|every:60000| could be either — and hiding the prefix would make two genuinely different series display identically. That is not merely confusing: the occurrence id is built from the key and is the idempotency key for scheduling, so two series that displayed alike would derive the same occurrence id at the same instant and silently merge into one job, losing one series' run. Keys without | can never collide with a generated key, which is why they are safe to show bare.

Migration: a series with a custom key re-registers under the new name on its next add, and the old series stays until you remove it. listRepeatables() shows both; removeRepeatable(key) accepts either spelling — the key as listed, or the one you originally supplied — so removing the stale one is await queue.removeRepeatable("<your key>") against the old, unprefixed entry.

Queue options

| Option | Type | Default | Meaning | |---|---|---|---| | namespace | string | required | The namespace the queue belongs to. | | driver | JobsDriver \| DriverConfig | a new memory driver | Where jobs live. A config is built and closed here; an instance is shared. | | logger | LoggerLike | no-op | See Logging. | | defaultJobOptions | JobOptions | | Merged under every add(). | | jobDefaultsRefreshInterval | number | 1000 | How long the queue's stored job defaults are trusted before being re-read, in ms: how long a saved change takes to reach this producer, plus one read. 0 reads them on every add. See Queue job defaults. | | subscribe | boolean | false | Re-emit events published by other processes, so a producer can watch jobs a worker elsewhere runs. Costs a subscription. | | publish | boolean | value of subscribe | Publish this queue's events for other processes. | | publishGate | () => Promise<void> | | Awaited before each publish. BunJobs sets it so events are not lost while its notifiers subscribe. | | dateParser | DateParser | chrono-node | Reads dates written in words. See Dates in words. |

Queue methods

Every method connects the driver on first use. After close(), calls throw QueueClosedError.

| Method | What it does | |---|---| | add(name, data, opts?) | Adds a job, and returns a Job. | | addBulk([{ name, data, opts? }]) | Adds several jobs in one call. Entries with repeat are added one at a time. | | addFlow(node) | Adds a job together with the jobs it waits on. See Flows. | | schedule(name, data?) / run(...), now(name, data?, opts?), create(name, data?) | The registry's builder verbs, bound for this queue. See Sending a job to another queue. | | getJob(id) | Returns one job, or null. | | list(state \| states, { offset, limit = 100, order = "asc", sort = "natural" }) | Returns jobs in the given state or states. Also takes name, search, and the worker and finish-time filters workerKey, workerId, finishedFrom and finishedTo. See Who ran a job. sort: "createdAt" orders by creation time on memory, SQL and MongoDB; see sorting by creation time. | | walk(state \| states, { limit, order, after? }) | A page read with a keyset cursor — the same filters as list(), plus after, and answering { jobs, offset, next }. This is how a list is walked; list() and page() are how it is sampled. next is null exactly when the walk is complete. Measured on a draining queue, an asc offset walk lost 20 of the 59 jobs that never left the list and the cursor walk lost none; see Pagination. Refused for active alone, whose key every lock renewal rewrites. | | count() / count(state) | Returns counts for every state, or for one. | | getDemand({ cap = 10_000 }) | How much work is there for a worker right now: waiting, due delayed/retrying jobs, stalled jobs, active, live workers, paused, and demand/outstanding. Bounded reads: no job is promoted, recovered or touched, though counting the live workers removes expired worker records, as listWorkers() does. See Demand. | | countAdded({ from, to }) | Of the jobs added in [from, to), how many are in each state now. Memory, SQL and MongoDB only; see Jobs added in a range. | | update(id, { data?, priority?, runAt?, onlyIn? }) | Patches a stored job. runAt moves only a waiting or delayed job. onlyIn makes the change conditional on the job's state. | | remove(id) | Removes a job. Refused while the job is active. | | retry(id, { resetAttempts = true }) | Returns a finished job to the queue. | | retryJobs(ids, opts?) | Retries several jobs, and returns the ids that went. | | retryAll(state, { name?, reason?, filter?, limit?, resetAttempts? }) | Re-drives every matching dead, failed or completed job, walking the state one page at a time. | | promote(id) | Makes a delayed or retry-pending job claimable now. | | getJobLogs(id, { offset, limit, order }) | Returns a page of a job's log. | | clearJobLogs(id) | Empties a job's log, and answers { status: "cleared", removed }, { status: "active" } (refused, nothing removed) or { status: "missing" }. See Clearing a job's log. | | pause() / resume() / isPaused() | Pauses or resumes claiming for every worker in every process. | | drain({ delayed = false }) | Drops pending jobs and returns the count. It never touches jobs that are running. | | clean(state, { olderThan, limit = 1000 }) | Removes jobs in state older than olderThan ms. | | setLimits(limits \| null) / getLimits() | Cluster-wide limits. See Rate and concurrency limits. | | getJobDefaults() | The queue's job defaults: code, the stored override, effective, overridden, seq. See Queue job defaults. | | setJobDefaults(patch, { expectedSeq? }) / resetJobDefaults({ expectedSeq? }) | Merges into, or clears, the queue's stored job defaults. null clears one key. | | applyJobDefaults({ seq, keys?, states?, limit?, cursor?, dryRun?, includeUnmarked? }) | Rewrites jobs already pending with the stored job defaults, one bounded call of a resumable walk. | | cleanWindows({ limit = 1000 }) | Removes stale debounce and throttle pointers. Workers also do this once a minute. | | listRepeatables() / removeRepeatable(key) | Lists repeat series, each with disabled, or removes one along with its scheduled occurrence. Removing a series clears its disabled flag. It stays removed even when a worker was scheduling its next occurrence at that moment. At most two occurrences still run: one already claimed, and one scheduled at that instant. | | disableRepeatable(key) / enableRepeatable(key) | Stops a series without removing it, or restarts it. See Disabling a series. | | close() | Closes the subscription, and the driver if the queue built it. |

Queue job defaults

A queue's job options can be changed while it runs: store an override on the queue, and every producer in every process adds under it. The override lives in queue state, beside limits, so every built-in driver can hold one.

await queue.setJobDefaults({
  attempts: 5,
  backoff: { type: "exponential", delay: 1_000, max: 60_000 },
  removeOnComplete: { count: 1_000 },
});
await queue.setJobDefaults({ attempts: null }); // one key back to the code's value
await queue.resetJobDefaults(); // all of them

Precedence, highest first, key by key:

an option passed explicitly on the job's own add();
the stored override;
a define() definition's options, for its name;
the queue's defaultJobOptions;
the built-ins.

The override beats a define() default: only add() wins. Each key is replaced whole — an override's backoff replaces the code's whole object. The editable keys are attempts, backoff, timeout, priority, removeOnComplete, removeOnFail, keepLogs and keepStacktraces (JOB_DEFAULT_KEYS). Each is bounded by JOB_DEFAULTS_BOUNDS: attempts 1–1,000, timeout at most a day, keepLogs at least 1 (a stored override cannot turn on "keep every line"), keepStacktraces at most 100, a retention ttl at most a year. backoff may name only fixed or exponential, which need no strategy registered on the worker. A value outside them is a ConfigError, and nothing is written.

getJobDefaults() answers code (this instance's defaultJobOptions over the built-ins), the stored override, effective (what a job passing none of the options gets), overridden, seq and updatedAt. The writes take expectedSeq for a safe read-modify-write: a write that finds the override moved on changes nothing and answers contended: true. A reset stores an empty override rather than deleting it, so seq keeps rising.

Propagation. A producer trusts its read for jobDefaultsRefreshInterval (1,000 ms by default), then reads again before its next add. A producer adding steadily renews that read in the background once three quarters of the interval have passed, so its adds do not wait for it; nothing older than the interval is ever used. So a change reaches every producer within about a second, plus one read, on every driver. The instance that wrote it uses it at once. A warm add reads nothing, and addBulk reads at most once per call. A worker reads the override the same way, to build a repeat series' next occurrence. A job added inside that window keeps the old values.

Explicit options are recorded. Every job added from this version on stores which of the editable options its own add() passed. The management API shows them as opts.explicit, key names in JOB_DEFAULT_KEYS order. A job added before this version has no record, and no explicit.

Rewriting jobs already pending

A save changes only jobs added afterwards. applyJobDefaults() rewrites the backlog with the stored override's values:

const { seq } = await queue.getJobDefaults();
let cursor: string | null = null;
do {
  const step = await queue.applyJobDefaults({ seq, cursor, limit: 1_000 });
  console.log(step.rewritten, step.exhausted);
  cursor = step.next;
} while (cursor !== null);

States. It walks waiting, delayed, failed (retry pending) and waiting-children, in claim order (states narrows them). Never active: its worker already holds its own copy. Never completed or dead.
Keys. It writes only keys the override sets (keys narrows them), and never a key the job's own add() passed: a job with an explicit priority keeps it and still gets the new attempts. A job all of whose changing keys are explicit is counted skippedExplicit.
Older jobs. A job added before the explicit record existed is skipped and counted skippedUnmarked, unless includeUnmarked: true, which treats every option of it as defaulted, including ones its add() may have passed.
attempts below attemptsMade is written, not clamped. The job runs once more, and dies if that attempt fails. Such jobs are counted exhausted, within rewritten.
priority reorders the backlog as the walk goes, keeping FIFO among equals.
Pinned. Each call re-reads the override and throws JobDefaultsChangedError (API: 409 DEFAULTS_CHANGED) when its seq is no longer the one passed, writing nothing. A walk of many calls applies one version, or stops.
Bounded and resumable. One call examines at most limit jobs (default 1,000) and answers next; pass it back as cursor. A cursor names the walk that issued it: one from another namespace or queue, another seq, or other keys or states is a ConfigError (API: 400 INVALID_ARGUMENT), never resumed.
Atomic per job. A job claimed meanwhile is skipped, never half-written: its options, maxAttempts and priority change in one write that re-checks its state.
dryRun: true walks and counts exactly as a real call, and writes nothing.
An override that sets nothing, or keys naming one it does not set, is a ConfigError.

It is irreversible. A rewritten job's earlier values are not kept, and the server never knows the code's values, which live in each producer. So a reset changes new jobs only: the jobs a walk rewrote keep the override's values, and after a reset the override sets nothing, so there is nothing left to apply.

examined can exceed the backlog: a job whose new priority moves it later in the walk is met again, and counted unchanged. On Redis that can reach about twice the backlog. moved counts jobs that left the walked states between a batch's read and its write. It is a lower bound, since drivers that lock or run a batch atomically never see one.

What one rewrite costs, per driver (measured on a loaded development machine; the SQL figures on 20,000 waiting jobs):

| Driver | Unit of work | Per job | Against claims | |---|---|---|---| | memory | the whole call, in one synchronous pass | in-process | atomic: it runs without yielding | | Redis | one Lua script per batch of at most 200 jobs, and at most 1 MiB of stored job blobs | about 15 µs (200 small jobs ≈ 3 ms a script) | a script blocks every client while it runs, so a claim waits at most one script | | Postgres, MySQL, MariaDB | one transaction per 500 jobs, on the claim index (MySQL and MariaDB force it with FORCE INDEX, and fall back when it is missing) | 38–57 µs | row locks, which a claim skips (SKIP LOCKED) rather than waits on | | SQLite | one transaction per 200 jobs | 7–12 µs | the transaction is the database's write lock, so claims wait for each batch | | MongoDB | one find of 500 jobs on the claim index, then one unordered bulkWrite of compare-and-set updates | not measured | nothing locked; a job changed between the read and the write fails its filter and counts moved | | file | each job under its own hold, 16 in flight | about a millisecond or more, growing with the queue (a single job update measured 0.9–5 ms) | a claim finding a job held moves on to the next |

A dry run costs 3–8 µs per job on SQL. A priority change also moves each job it writes in the claim index.

In the management API these are GET, PUT and DELETE /queues/:queue/job-defaults and POST /queues/:queue/job-defaults/apply. Saving (queues.defaults) and applying (queues.applyDefaults) are separate actions, both off by default, so a host can let operators tune defaults for new work without letting them rewrite a backlog. limits.maxApplyDefaults caps one call's limit.

Example: 10-options/job-defaults.ts.

Workers

const worker = new BunQueueWorker<Mail, void>(
  "mail",
  async (job, ctx) => {
    ctx.signal.throwIfAborted(); // aborted on timeout, shutdown or a lost lock
    await job.updateProgress(50);
    await send(job.data);
  },
  { namespace: "account", driver, concurrency: 10, deadLetterQueue: "mail-dead" },
);

await worker.run(); // or `autorun: true`

The processor receives the Job and a ProcessorContext:

| Field | Type | Meaning | |---|---|---| | signal | AbortSignal | Aborted when the attempt times out, the worker closes, or the job's lock is lost. | | logger | Logger | A logger bound to the job's ids. | | workerId | string | The worker's id. | | attempt | number | The attempt number, starting at 1. | | heartbeat() | () => Promise<void> | Extends the lock now, by lockDuration, for a step longer than lockDuration. | | log(line) | (line) => Promise<number> | Appends to the job's persistent log, and returns how many lines it keeps. |

Examples:

Worker options

| Option | Type | Default | Meaning | |---|---|---|---| | namespace | string | required | Must match the producer's namespace. | | driver | JobsDriver \| DriverConfig | a new memory driver | Where jobs live. | | logger | LoggerLike | no-op | Logger. | | id | string | fresh id | Identifies the worker in job records and logs. | | key | string | [service.]queue[.name\|.ordinal], or id when one is given | The worker's stable identity: a remote configuration override is keyed by it, so the override survives restarts and redeploys and reaches every replica. Analytics record under it too. Set it to name a worker something an operator will recognise. | | service | string | BunJobs's service | The service the worker belongs to: reported in its record, and the first segment of the derived key. | | name | string | | What to call the worker within its queue when a service runs more than one there, so they can be configured apart. The last segment of the derived key; prefer it over the ordinal BunJobs assigns, which shifts once a worker's creation becomes conditional. | | concurrency | number | 1 | Jobs processed at once. Can also be set at runtime via worker.concurrency. | | lockDuration | number | 30000 | How long a claim's lock lives. | | heartbeatInterval | number | lockDuration / 3 (min 250) | How often the lock is renewed. | | reportInterval | number | 10000 | How often the worker writes its heartbeat record, which queue.listWorkers() and the management API's worker routes read: once per interval, and on start, pause, resume and a concurrency change, never per job. A record lapses three intervals after its last write. 0 turns reporting off. See Workers. | | stalledInterval | number | 30000 | How often to sweep for jobs whose worker died. | | maxStalledCount | number | 1 | How many stalls a job may have before it is buried. | | pollInterval | number | 1000 | Wait between claim attempts on a polling driver. | | maxBlock | number | 5000 | Longest blocking wait on a blocking driver. | | maintenance | boolean | true | Contend for the queue's housekeeping lease and, while holding it, run those sweeps once a minute: pruning expired results, healing repeat series, and sweeping stale debounce/throttle windows and dead workers' leftovers. false arms no minute timer and never reads that lease. It does not reach liveness — promoting delayed jobs is every worker's own, and every worker contends for the stalled lease that recovers stalled jobs and heals flows, whatever this says (see Maintenance: liveness and housekeeping). Reported on the worker's record as sweeps. | | autorun | boolean | false | Start consuming on construction. | | drainDelay | number | 0 | Milliseconds of quiet before drained is emitted. | | target | WorkerTarget | "in-process" | Where each attempt runs: "in-process", "worker-thread", "child-process", one of those as { kind, … } with its tuning, a ContainerTarget ({ kind: "container", image }, a fresh container per attempt), or a WorkerTargetFactory. See Where attempts run. | | backoffStrategies | BackoffStrategies \| Record<string, BackoffStrategy> | | Named custom backoffs. | | deadLetterQueue | string | | The dead-letter queue for jobs that do not name their own. | | limitsRefreshInterval | number | 1000 | How long stored limits are trusted before being re-read. | | jobDefaultsRefreshInterval | number | 1000 | How long the queue's stored job defaults are trusted before being re-read. The worker reads them only to build a repeat series' next occurrence. See Queue job defaults. | | waitToExit | boolean | true | Keep the process alive while waiting for work. false lets a script exit when its own work is done. The Redis, Postgres and MongoDB clients hold the process on their own, so close the driver in that case. | | publish | boolean | false | Publish job events (active, progress, completed, failed, stalled, and so on) for other processes. | | publishGate | () => Promise<void> | | Awaited before each publish. | | control | boolean \| WorkerControlOptions | false; true for a worker BunJobs builds | Obey pause, resume, stop, start and configuration overrides written by another process, such as the management API. true subscribes where the driver pushes events and polls where it does not (one change counter per queue per driver instance every interval, shared by all its workers there; a worker reads its own instructions only when that counter moves); { enabled, subscribe, interval } overrides either choice, interval defaulting to 2000. The heartbeat re-reads the stored instructions too, so a lost event costs at most one reportInterval. On a driver without queue state the worker reports control.enabled: false and simply runs. | | stopPersistence | "process" \| "key" | "process" | How long a remote stop lasts. "process" records it against this incarnation, so a restart brings the worker back running; "key" also records it against key, so every worker with that key applies it at startup until somebody starts it. | | stopPersistenceOverridable | boolean | false | Whether one instruction may ask for the other stopPersistence. Off, the process rather than the caller decides whether a stop outlives it. On, the worker also obeys a stop recorded against its key at startup (as a persist: "key" stop writes), and any start clears that record. | | metrics | MetricsOptions | everything on | What this worker records for analytics: its jobs completed and failed, under its stable key, and a busyness sample on each heartbeat report. { workers: false } stops both, whatever the driver, and is the first lever for a large fleet. resolution and secondRetentionMs reach only a driver built here from a config. See The metrics option. |

Maintenance: liveness and housekeeping

What a worker does for the queue besides running jobs comes in two halves, and only one of them is yours to switch off.

| Half | What it does | Cadence | Optional? | |---|---|---|---| | Liveness | Promotes due delayed and retrying jobs; recovers jobs whose worker died (stalled); heals flows, so a parent past a child that already finished advances | every pollInterval (at least once a second) and on every idle pass; stalledInterval for the other two | No. Promotion on every worker, always; recovery and flow healing on the holder of the stalled lease, which every worker contends for. | | Housekeeping | Prunes expired results; heals repeat series; sweeps stale debounce/throttle windows and the leftovers of workers that are gone | once a minute | Yes — maintenance: false. Runs on the holder of the minute lease, which only workers that did not opt out contend for. |

Liveness cannot be turned off, because nothing else does it. A delayed retry leaves delayed only when some worker promotes it; a job held by a process that died returns to the queue only when some worker's stalled sweep takes it back; a flow parent whose child finished unrecorded advances only when some worker heals it. A queue whose only worker skipped them stranded all three for ever: in one observed run, a job sat active for forty-five minutes and four jobs were frozen at attempt 1 of 2. maintenance: false used to do exactly that, and no longer does — it now opts out of the housekeeping half alone.

Both halves are leased, so one worker per queue makes each pass rather than all of them making the same one — but the two leases are contended for by different sets of workers, and that is the whole of how the split survives leasing.

The stalled lease covers stalled recovery and the flow heal that rides it. Contending for it is how a worker takes part, so a worker with maintenance: false contends like any other. It costs one queue-state read per stalledInterval, which is the right price for staying part of the queue's liveness. An opt-out worker that stood down from it could never be the holder, so a queue whose only worker sets maintenance: false would never recover a stalled job — the exact failure this split exists to prevent, reached through the lease instead of the timer.

The minute lease covers the housekeeping, and an opt-out worker never reads it. It arms no minute timer, so a minute lease it won would be one it never swept under: strictly worse than standing aside. maintenance: false therefore costs one queue-state read per stalledInterval, and nothing per minute.

That read is not a new cost. Flow healing has been leased under this same entry all along, and every worker already paid one read of it per stalledInterval; the lease now covers stalled recovery as well, so the read buys more. Measured over one second, three opt-out workers at a 100 ms stalledInterval: 33 lease reads and 33 recoverStalled calls before, 34 lease reads and 11 recoverStalled calls after — the same reads, a third of the sweeps.

Housekeeping is safe to leave to others. Nothing waits on it: skipping it costs the queue tidiness — results kept past their ttl, a repeat series whose occurrence went missing, stale window pointers — not progress. So a large fleet may run most of its workers with maintenance: false and leave the sweeps to a few. It is exactly one timer, and a worker that opts out never arms it: no sweep, and no call of theirs.

maintenance stays a boolean on purpose. Under this split a third value for "a dedicated sweeper runs elsewhere" would mean exactly what false means.

Which half a worker does is on its heartbeat record and in the management API as sweeps — true when it arms the housekeeping and contends for its lease, and never a claim about liveness. It reports what the worker is willing to do, not what it did this minute: on a queue with five sweeps: true workers, one of them holds the minute lease and the other four stood down on that pass. That is what a "nobody is sweeping this queue" check wants — a queue with no live true has nobody eligible, which is the condition worth warning about.

The worker's events are:

ready
active
progress
completed
failed, emitted on every failed attempt
retrying
dead
deadLettered
stalled
lockLost
drained
paused, resumed
closing, closed
error

The job events can also be scoped to a job name, such as worker.on("completed:sendEmail", ...).

Retries and backoff

An attempt that throws is retried while attempts remain. Throwing an UnrecoverableJobError sends the job straight to dead, even if attempts are left.

backoff accepts either of two forms:

a number of milliseconds, meaning a fixed delay;
an object { type, delay, factor, max, jitter }.

The built-in type values are fixed (the default when you give an object), exponential, linear, fibonacci, full-jitter and decorrelated-jitter. Any other type is the name of a custom strategy registered on the worker:

jobs.defineBackoff("retryAfter", ({ attempt, error }) =>
  error.message.includes("quota") ? false : attempt * 30_000, // false: stop retrying
);

await jobs.now("sync", data, { attempts: 5, backoff: { type: "retryAfter", max: 300_000 } });

The job stores only the strategy's name, and each consuming process resolves it. A worker that does not know the name falls back to the default backoff and logs a warning. It never loses the job's remaining attempts.

For a worker you construct yourself, pass backoffStrategies. Workers created by BunJobs receive every strategy registered with defineBackoff.

Example: 06-failures/retries-and-backoff.ts.

Rate and concurrency limits

Limits are stored on the queue, so every worker in every process enforces the same numbers. A change reaches running workers within their limitsRefreshInterval.

await queue.setLimits({
  rate: { max: 100, duration: "1 minute" }, // queue-wide start rate
  concurrency: 20, // jobs running at once, across every worker
  names: { sendEmail: { concurrency: 5, rate: { max: 10, duration: 1_000 } } },
});
await queue.setLimits(null); // lift them all

When a name reaches its limit, it is skipped rather than waited behind, so other names keep running.

Enforcement is approximate by design. Concurrency is held as leases that expire lockDuration after a worker stops renewing them. In exchange, no claim waits on a global lock.

The lease is short and kept alive by a heartbeat, not by a long timeout: a claim's lock lives lockDuration (30 s by default) and the worker renews it every lockDuration / 3 (10 s, floored at 250 ms). So a worker that dies — crash, SIGKILL, a machine that disappears — stops renewing at once, its locks lapse 30 s later, and the next stalled sweep, which runs every stalledInterval (30 s), hands its jobs back to the queue. With the defaults that is under a minute from death to retry, and both numbers are yours to shorten.

define(name, handler, { concurrency }) in the registry stores a per-name concurrency the same way. Every built-in driver can store limits.

Examples:

Dead letters

A job that dies can be copied to a dead-letter queue, named by the job's deadLetter option or by the worker's deadLetterQueue. The copy is a DeadLetter with these fields:

queue
id
name
data
failedReason
attemptsMade
diedAt

A job failed from outside its processor with job.fail() is copied only to its own deadLetter queue, since no worker is involved; failed from inside its processor, it is filed like any other death on that worker.

It is added under the original job's name, so a worker on the dead-letter queue can dispatch on it as the original worker did. Its id is derived from the original's, so one death files exactly one letter. The dead job itself is kept or removed according to removeOnFail.

Once the cause is fixed, re-drive a backlog with, for example, queue.retryAll("dead", { reason: /ECONNREFUSED/ }).

Example: 06-failures/dead-letter-queue.ts.

Pause, resume and shutdown

worker.pause({ waitActive? }), worker.resume() and worker.isPaused() affect this worker only.
queue.pause() and queue.resume() affect every worker in every process.
worker.close({ force?, timeout? }) stops claiming and waits for in-flight jobs. A job still running at timeout has its signal aborted, and its lock is left to expire, so another worker recovers it as stalled instead of the job being lost. close() holds the process open until it finishes, so it is safe to await in a SIGTERM handler. A close() that lands while run() is still connecting ends the startup there: nothing is armed, no ready is emitted, and run() resolves.
worker.close({ force: true }) while a graceful close() is under way escalates it — a second SIGTERM, or a shutdown's backstop. From then on it is a forced close at whatever step it had reached: jobs still running are aborted, the target is closed with { force: true } (a built-in child-process or worker-thread target kills its runs at once, even in the middle of its graceful close, instead of at the end of the 4000 ms grace), and nothing a forced close would not wait for is waited for. Both calls resolve once the close has finished; neither rejects because a graceful step was cut short. A force during a forced close, and a close() without force during any close, change nothing, and also resolve once the close has finished — on a worker that never ran as well. A later call's timeout is ignored: force is the way to cut a graceful close short.
A built-in target's close counts a killed run as gone once it has actually stopped: a child once its exit is seen, and a worker-thread run once its thread's close event has fired, since terminate() returns before the thread stops. That wait is bounded at 500 ms. A busy Bun Worker usually stops within tens of milliseconds, but a loaded machine can take it past that (oven-sh/bun#44216). In that case the close resolves anyway and logs one warn naming the run. The thread still stops, only later.
worker.stop({ timeout?, reason? }) parks the worker instead: it stops claiming and running its background passes — liveness and housekeeping alike — drains its jobs in flight, and keeps heartbeating, so worker.start() (or a remote start) can bring it back. run() stays pending while it is parked. timeout abandons the jobs still running after that many milliseconds, which close() also does. worker.state reads running, paused, stopping, stopped or restarting (a moment while a configuration override is applied), and worker.isStopped() is true while stopping or stopped.

Examples:

Controlling workers from another process

jobs.workers.controller(queue) returns a WorkerController, which pauses, resumes, stops and starts the workers of one queue, and overrides their settings, wherever they run. It works the way RunnerController does: each call stores what it asks for in the driver and publishes a worker control event, and the worker applies it when it hears the event, at its next control poll, or at its next heartbeat report if the event is lost. No worker has to be reachable for the call to succeed.

const jobs = new BunJobs({ namespace: "shop", driver, service: "billing" });
jobs.worker("mail", sendMail); // key "billing.mail"

// In any process sharing the driver and namespace:
const mail = jobs.workers.controller("mail");

await mail.pause({ key: "billing.mail" }); // every replica with that key
await mail.resume({ key: "billing.mail" });

const [oldest] = await mail.list();
if (oldest) {
  await mail.stop({ id: oldest.id }, { timeout: 60_000 }); // one process only
  await mail.start({ id: oldest.id });
}

export const stored = await mail.setConfig("billing.mail", { concurrency: 16 });
await mail.resetConfig("billing.mail"); // back to what the code asks for

A worker has two identities, and the two kinds of call use different ones:

Pause, resume, stop and start take a WorkerSelector: { id } for one incarnation, or { key }, which reaches every live worker carrying that stable key. The instruction records the incarnation it was written for, so the process that replaces a worker never applies it: a deployment comes back running. A stop outlives a restart only when it is recorded against the key: on a worker whose stopPersistence is "key", or one sent with persist: "key" to a worker whose stopPersistenceOverridable is on (see Worker options).
Configuration is stored against the stable key alone, so it survives restarts and reaches every replica, including one started tomorrow. The key is [service.]queue[.name|.ordinal] unless you set key; BunJobs's service option supplies the first segment.

| Method | What it does | |---|---| | list() | Every live worker on the queue, oldest first, as WorkerInfo records carrying key, service, state, config and control. Throws NotSupportedError on a driver that keeps neither worker records nor queue state, as queue.listWorkers() does. | | get(id) | One live worker by its incarnation id, or null. | | listConfigs() | Every override stored on the queue, including those whose workers are

Published

Vulnerabilities

Links

Maintainers

Keywords

Readme

@kingsleyweb/bun-jobs

Status

Contents

Installation

Quick start

A queue and a worker

The job registry

A runner

Concepts

Queues and adding jobs

Job options

What an id may be

Reserved queue-state names

Repeat keys you choose are namespaced

Queue options

Queue methods

Queue job defaults

Rewriting jobs already pending

Workers

Worker options

Maintenance: liveness and housekeeping

Retries and backoff

Rate and concurrency limits

Dead letters

Pause, resume and shutdown

Controlling workers from another process