# bun-boss documentation Generated from https://github.com/khromov/bun-boss (branch main). --- # File: docs/introduction.md # Introduction bun-boss is a job queue that runs on Postgres, PGlite, or SQLite, operated by 1 or more Bun instances. On Postgres, bun-boss uses [SKIP LOCKED](https://www.postgresql.org/docs/current/sql-select.html#SQL-FOR-UPDATE-SHARE), a feature built specifically for message queues to resolve record locking challenges inherent with relational databases. On backends without it (SQLite), an atomic, state-gated claim provides the same guarantee. Either way this provides exactly-once delivery and the safety of guaranteed atomic commits to asynchronous job processing. This will likely cater the most to teams already familiar with the simplicity of relational database semantics and operations (SQL, querying, and backups). It will be especially useful to those already relying on a relational database that want to limit how many systems are required to monitor and support in their architecture. On Postgres and PGlite, bun-boss uses declarative list-based partitioning to expose a single logical `job` table. By default, all queues' jobs will be stored together in a shared table, but this could affect performance if 1 or more of your queues grows significantly or experiences an unexpected backlog. If a queue needs to be scaled out, you can create it with a `partition` option that will create a dedicated physical table within the partitioning hierarchy. This storage strategy should offer a balance between maintenance operations and query plan optimization. According to [the docs](https://www.postgresql.org/docs/current/ddl-partitioning.html#DDL-PARTITIONING-DECLARATIVE-BEST-PRACTICES), Postgres should scale to thousands of queues in a partitioning hierarchy quite well, but the decision on how many dedicated tables to use should be based on your specific needs. If your usage somehow exceeds what Postgres partitioning is capable of (congrats!), consider provisioning queues into separate schemas in the target database. (Postgres and PGlite only — on SQLite the `partition` option is accepted but ignored.) You may use as many Bun instances as desired to connect to the same Postgres database, even running it inside serverless functions if needed. Each instance maintains a client-side connection pool or you can substitute your own database client, limited to the maximum number of connections your database server (or server-side connection pooler) can accept. If you find yourself needing even more connections, bun-boss can easily be used behind your custom web API. ## Job states All jobs start out in the `created` state and become `active` via [`fetch(name, options)`](./api/jobs.md#fetchname-options) or in a polling worker via [`work()`](./api/workers.md#work). In a worker, when your handler function completes, jobs will be marked `completed` automatically unless they are no longer `active` — for example if they were deleted via [`deleteJob(name, id)`](./api/jobs.md#deletejobname-id-options), cancelled, or timed out while the handler was running. Workers started with the `perJobResults` option settle each job themselves instead. If an unhandled error is thrown in your handler, the job will usually enter the `retry` state, and then the `failed` state once all retries have been attempted. Uncompleted jobs may also be assigned to `cancelled` state via [`cancel(name, id)`](./api/jobs.md#cancelname-id-options), where they can be moved back into `created` via [`resume(name, id)`](./api/jobs.md#resumename-id-options). Failed jobs can be retried via [`retry(name, id)`](./api/jobs.md#retryname-id-options). All jobs that are not actively deleted during processing will remain in `completed`, `cancelled` or `failed` state until they are automatically removed. Jobs that are never fetched are also removed once their retention window elapses, without ever reaching a terminal state. --- # File: docs/install.md # Database install bun-boss will automatically create a dedicated schema (`pgboss` is the default name) in the target database. This will require the user in database connection to have the [CREATE](https://www.postgresql.org/docs/current/sql-grant.html) privilege. ```sql GRANT CREATE ON DATABASE db1 TO leastprivuser; ``` If the CREATE privilege is not available or desired, export the schema DDL programmatically with the included [`getConstructionPlans()`](./api/utils.md) utility. It returns the SQL for the current schema version without executing it, so a DBA can review and run the commands manually: ```js import { getConstructionPlans } from 'bun-boss' import fs from 'node:fs' fs.writeFileSync('create-bunboss.sql', getConstructionPlans('pgboss')) ``` Once the schema exists, construct the instance with `migrate: false` so `start()` verifies the schema instead of trying to create it. The runtime user still needs access to the objects the DBA created, otherwise `start()` fails with `permission denied for schema pgboss`. Grant it usage on the schema and DML on its tables: ```sql GRANT USAGE ON SCHEMA pgboss TO leastprivuser; GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA pgboss TO leastprivuser; ALTER DEFAULT PRIVILEGES IN SCHEMA pgboss GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO leastprivuser; ``` `ON ALL TABLES` only covers the tables that exist when you run it, so the `ALTER DEFAULT PRIVILEGES` line is what keeps a queue created later reachable. No CREATE privilege is required at runtime with `migrate: false`, as long as queues are created by the privileged operator. Creating a queue from the application still performs DDL: `createQueue(name, { partition: true })` builds a dedicated table and fails with `permission denied for schema pgboss`, and deleting a partitioned queue requires ownership of that table. > [!NOTE] > When managing schema manually, you will need to monitor future releases for schema changes. > [!WARNING] > Using an existing schema is supported for advanced use cases **but discouraged**, as this opens up the possibility that creation will fail on an object name collision, and it will add more steps to the uninstallation process. # Database uninstall If you need to uninstall bun-boss from a database, just run the following command. ```sql DROP SCHEMA pgboss CASCADE ``` Replace `pgboss` with the name of your schema if you've customized it. The schema name cannot be supplied as a bind parameter — it must be written into the statement. NOTE: If an existing schema was used during installation, created objects will need to be removed manually using the following commands. ```sql DROP TABLE pgboss.version; DROP TABLE pgboss.job_dependency; DROP TABLE pgboss.job_common; DROP TABLE pgboss.job; DROP TYPE pgboss.job_state; DROP TABLE pgboss.schedule; DROP FUNCTION pgboss.create_queue; DROP FUNCTION pgboss.delete_queue; DROP FUNCTION pgboss.job_table_format; DROP FUNCTION pgboss.job_table_run; DROP TABLE pgboss.queue; ``` --- # File: docs/database-backends.md # Database Backends bun-boss runs on stock single-node PostgreSQL by default. It also supports the embedded WASM build [PGlite](https://pglite.dev) and an embedded [SQLite](#sqlite-embedded-via-bunsql) backend. You select one with the `backend` option, which applies all the compatibility behavior that backend needs. ## Backend profiles `backend` is the **only** option you set — it selects the database bun-boss is running against and turns on the right combination of internal compatibility behavior for it: ```typescript import { BunBoss } from 'bun-boss' const boss = new BunBoss({ url: 'postgresql://localhost:5432/pgboss', backend: 'postgres' }) ``` Each backend has a *kind* — `standard` (stock PostgreSQL) or `embedded` (in-process): | `backend` | Kind | What it enables | |-----------|------|-----------------| | `postgres` *(default)* | standard | *(none — full PostgreSQL)* | | `pglite` | embedded | *(none — full PostgreSQL; see [PGlite](#pglite-embedded))* | | `sqlite` | embedded | A different SQL dialect entirely: every compatibility flag plus sqlite-rendered SQL (see [SQLite](#sqlite-embedded-via-bunsql)) | `backend` is the only option you set — bun-boss derives everything above from it, so a deployment can't end up with an inconsistent combination. The rest of this page explains each behavior (and names the internal flag it maps to, for anyone reading the source). ## Table isolation `tableIsolation` controls **where** bun-boss's tables live, independently of the SQL dialect: | `tableIsolation` | Objects look like | Notes | |------------------|-------------------|-------| | `'schema'` *(default on Postgres/PGlite)* | `pgboss.job` | A dedicated Postgres schema; supports partitioning. | | `'prefix'` | `"bunboss.job"` | One quoted identifier per object in the connection's default schema (e.g. `public`), co-located with your own tables. No schema is created; **partitioning is disabled**. | ```typescript // Keep bun-boss's tables in the default schema instead of a separate `pgboss` schema. const boss = new BunBoss({ url: 'postgresql://localhost:5432/appdb', tableIsolation: 'prefix' // creates "bunboss.job", "bunboss.queue", … in the default schema }) ``` The `schema` option still names the namespace in both modes; in prefix mode it becomes the quoted prefix and defaults to `bunboss` (rather than `pgboss`). Everything else on Postgres — `SKIP LOCKED`, LISTEN/NOTIFY, advisory locks, covering indexes — is unaffected; only partitioning is turned off. Because all queues then share one job table, the per-queue [`partition: true`](api/queues.md) option has no effect in prefix mode. **SQLite is always prefix** — it has no schemas — so this is simply how the SQLite backend works, and `tableIsolation: 'schema'` is rejected there. > **Keep namespaces distinct across installs in one database.** The LISTEN/NOTIFY channel and the > maintenance advisory lock are derived from the `schema` **name**, not the isolation mode. Two > bun-boss installs that share the same `schema` name in the same database — for example a > schema-mode `pgboss` and a prefix-mode `schema: 'pgboss'` — would share a NOTIFY channel and lock. > That only causes harmless cross-wakeups and serialized maintenance (never data corruption), but > give co-located installs different `schema` names to avoid it. The defaults (`pgboss` vs > `bunboss`) already differ. > **Upgrading from an earlier SQLite install:** the SQLite default namespace changed from `pgboss` > to `bunboss`. An existing SQLite database whose tables are `"pgboss.job"` must pass > `schema: 'pgboss'` to keep using them; otherwise bun-boss installs a fresh `"bunboss.*"` set. ## Database compatibility The matrix shows which PostgreSQL features each backend supports (✅). Where a feature isn't available (❌), bun-boss automatically switches to the compatible alternative — see the [compatibility flags](#compatibility-flags) below. | Database | Status | `backend` | SKIP LOCKED | Multi-mutation CTEs | Table partitioning | Deferrable constraints | Advisory locks | Covering indexes | LISTEN/NOTIFY | |----------|--------|-----------|:---:|:---:|:---:|:---:|:---:|:---:|:---:| | PostgreSQL | Tested | `postgres` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅² | | PGlite | Tested | `pglite` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅¹ | | SQLite | Tested | `sqlite` | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ¹ PGlite is embedded single-connection PostgreSQL, so LISTEN/NOTIFY works entirely in-process. The `fromPglite` adapter wires it up automatically, so `useListenNotify` works with no extra setup. ² Producer side only through the built-in driver: bun-boss still inlines `pg_notify` into inserts on notify-enabled queues, but Bun's SQL client implements no LISTEN, so the *listener* requires a `db` adapter that implements `listen` — see [No LISTEN/NOTIFY](#no-listennotify). ## Compatibility flags Here's what each behavior does differently from stock PostgreSQL, and the internal flag it maps to in the source. These flags are not user-configurable — `resolveBackend` derives them from the `backend` profile. Today the `sqlite` profile is the one that turns them all on; the `postgres` and `pglite` profiles leave them off. | Capability | Effect | Trade-off | Flag | |------------|--------|-----------|------| | Lock-free fetch | Fetch jobs with an atomic `UPDATE ... RETURNING` (plus a `state < 'active'` recheck) instead of `SELECT FOR UPDATE SKIP LOCKED`. | Under high contention some workers get empty results instead of skipping to unlocked rows. | `noSkipLocked` | | Split-statement writes | Run `complete`, `fail`, and supervisor expiry as split statements inside a transaction rather than a single multi-mutation CTE. | A few extra round-trips per command; negligible for normal workloads. | `noMultiMutationCte` | | Single shared table | Create the job table without `PARTITION BY LIST`. | Per-queue partitioning (`partition: true`) is unavailable; all jobs share one table. | `noTablePartitioning` | | Immediate constraints | Omit `DEFERRABLE INITIALLY DEFERRED` on foreign keys. | Constraints check immediately rather than at commit (no effect on normal operation). | `noDeferrableConstraints` | | Lock-free coordination | Disable `pg_advisory_xact_lock` (used to coordinate schema creation, queue DDL, and supervisor maintenance across instances). | Concurrent instances may occasionally do redundant maintenance — a performance, not correctness, concern. | `noAdvisoryLocks` | | Plain indexes | Omit the `INCLUDE` clause on covering indexes. | Slightly less efficient index-only scans during fetch; minimal for most workloads. | `noCoveringIndexes` | | No LISTEN/NOTIFY | Skip the listener entirely; workers rely on polling. | Wake-up latency only — a NOTIFY is never a correctness requirement. | `noListenNotify` | Lock-free fetch, split-statement writes, lock-free coordination, and polling-only wake-ups are **runtime** behaviors; `noTablePartitioning`, `noDeferrableConstraints`, and `noCoveringIndexes` are **schema** choices applied at install time. ### Why fetch and mutation strategy are tracked separately `noSkipLocked` and `noMultiMutationCte` address two unrelated limitations: - **`noSkipLocked`** is about the *fetch* path. By default bun-boss claims jobs with `SELECT FOR UPDATE SKIP LOCKED`. Where a backend can't rely on `SKIP LOCKED`, bun-boss instead claims jobs with an atomic `UPDATE ... RETURNING` and a `state < 'active'` recheck: ```sql WITH next AS ( SELECT id FROM jobs WHERE name = $name AND state < 'active' AND start_after <= now() ORDER BY priority DESC, created_on, id LIMIT $batchSize ) UPDATE jobs j SET state = 'active', started_on = now(), retry_count = CASE WHEN started_on IS NOT NULL THEN retry_count + 1 ELSE retry_count END FROM next WHERE j.id = next.id AND j.state < 'active' -- recheck for concurrent safety RETURNING j.* ``` See [Andrew Werner's article on distributed work queues](https://dev.to/ajwerner/quick-and-easy-exactly-once-distributed-work-queues-using-serializable-transactions-jdp) for the pattern. The trade-off: under high contention multiple workers' CTEs may select the same candidate rows, all attempt the `UPDATE`, one wins, and the rest receive empty results and poll again. That is acceptable when processing time >> fetch time (typical for job queues). - **`noMultiMutationCte`** is about the *write* path. bun-boss's `complete`, `fail`, and supervisor expiry normally run as a single CTE that mutates more than one table at once (e.g. completing a job and unblocking its flow dependents). Where that isn't available, those operations run as separate statements inside one transaction instead, so a job can't be lost between them. `noSkipLocked` changes two statements: the fetch claim and the `noMultiMutationCte` flow-audit lock (`selectBlockingParents`). Other row locks are unaffected — the flow resolver and `redrive` keep `FOR UPDATE ... SKIP LOCKED`, and the dependency-unblock lock has always been a plain `FOR UPDATE`. ### Testing the runtime toggles `noSkipLocked` and `noMultiMutationCte` are pure runtime behaviors (no schema impact) that work on plain PostgreSQL, so the project exercises them independently of any embedded backend: - **`bun run test:no-skip-locked-no-cte`** — runs the **entire** test suite on Postgres with `NO_SKIP_LOCKED_NO_CTE=true`, which makes `test/testHelper.ts`'s `getConfig()` set the internal `__test__noSkipLockedNoCte` hook for every test (forcing `noSkipLocked` + `noMultiMutationCte` on top of the plain-Postgres schema, since the flags are not publicly configurable). Any new test is automatically exercised against the atomic-`UPDATE` fetch and split-statement write paths, fast and reliably. It runs as its own CI job. `test/noSkipLockedNoCte.test.ts` holds the invariants the general suite cannot express (concurrent-fetch deduplication, `failNoCte`/`completeNoCte` composition inside a caller transaction, and the compatibility-flag construction paths). Those cases force the runtime behavior via `__test__noSkipLockedNoCte`, so they run in every mode. ### Transaction isolation The `state < 'active'` recheck in the `UPDATE` guarantees exactly-once claims under READ COMMITTED (the PostgreSQL default); bun-boss does not set or require SERIALIZABLE isolation. ## Per-database notes Testing status, setup, and caveats for each supported backend. ### Tested: PostgreSQL PostgreSQL is the primary supported database with full feature support. Use standard mode — no special options needed. The built-in driver is [Bun's SQL client](#bunsql-the-built-in-driver). ### PGlite (embedded) [PGlite](https://pglite.dev) is a complete PostgreSQL build packaged as a WASM library that runs embedded in your process — no separate database server. Because PGlite is real PostgreSQL, bun-boss runs against it with **no compatibility flags**: declarative partitioning, deferrable constraints, advisory locks, covering indexes, and `SELECT FOR UPDATE SKIP LOCKED` all work. It is embedded single-connection PostgreSQL, reached through the `@electric-sql/pglite` client rather than the built-in Bun-SQL driver, via the `fromPglite` adapter. #### Usage Install PGlite alongside bun-boss: ```bash bun add @electric-sql/pglite ``` Construct a PGlite instance, wrap it with `fromPglite`, and select the `pglite` backend profile: ```ts import { PGlite } from '@electric-sql/pglite' import { BunBoss, fromPglite } from 'bun-boss' const pglite = new PGlite('./my-app-data') // or new PGlite() for in-memory; 'idb://name' in the browser const boss = new BunBoss({ backend: 'pglite', db: fromPglite(pglite) }) await boss.start() await boss.createQueue('email') await boss.send('email', { to: 'user@example.com' }) const [job] = await boss.fetch('email') // ... do work ... await boss.complete('email', job.id) ``` #### Lifecycle is yours to manage Unlike the built-in driver's connection, bun-boss does **not** open or close the PGlite instance — you own it. Construct it before `boss.start()` and close it after `boss.stop()`: ```ts await boss.stop() await pglite.close() ``` This mirrors the [database adapters](api/adapters.md): bun-boss only issues queries through the object you provide (`executeSql`, plus `withTransaction`/`listen` when it implements them); it never opens or closes it. #### Single-connection considerations PGlite serializes everything through one connection. bun-boss's background loops (maintenance, scheduling, monitoring) and your workers all share that single connection, so queries are processed one at a time. This is fine functionally — PGlite queues requests internally — but you should keep concurrency modest: - There is no benefit to large `batchSize` or many concurrent workers; they cannot run in parallel. - For embedded / local-first / testing workloads (PGlite's sweet spot) this is rarely a constraint. - For high-throughput multi-worker queues, use a server-based PostgreSQL instead. #### Persistence PGlite supports in-memory, IndexedDB (browser), and filesystem persistence — see the [PGlite docs](https://pglite.dev/docs/filesystems). bun-boss treats all of them identically; the job schema and data persist wherever the PGlite instance stores its data directory. ### Bun.SQL (the built-in driver) [Bun's built-in SQL client](https://bun.com/docs/api/sql) is bun-boss's driver for stock PostgreSQL: it is a *driver*, not a backend, so `backend` stays at its default `postgres` and no compatibility flags apply. Passing a connection string or connection options to the `BunBoss` constructor builds an internal `SQL` client wrapped by the `fromBunSql` adapter; passing `db: fromBunSql(sql)` uses a client you own instead. #### Bun 1.3.14 minimum, 1.4 recommended The package floor is **Bun 1.3.14**; prefer **1.4+**. On 1.3.x, a pooled connection can be handed to a waiting query before the `ROLLBACK` clearing its aborted transaction has landed, so an unrelated query can fail with `25P02 current transaction is aborted` whenever a transaction block fails under concurrency — contended maintenance being the realistic trigger. The window is inside Bun's pool and is fixed in 1.4; on 1.3.x the driver detects the aborted state, clears it, and retries, so the failure is masked rather than eliminated. A rarer silent form of the same 1.3.x defect can return an empty result for a committed row under concurrent load — bun-boss bounds it by re-reading a queue lookup that comes back empty before treating the queue as missing, and by requiring two consecutive empty refreshes before it wipes the queue cache. Only 1.4 removes it. Transaction-scoped use (`fromBunSql(tx)` inside `sql.begin()`) never reserves a connection and is unaffected on either version. #### `@types/bun` is an optional peer dependency bun-boss publishes uncompiled TypeScript, so a consumer running `tsc` type-checks bun-boss's own sources as part of its program (`skipLibCheck` does not apply — it only skips `.d.ts`). The driver imports `SQL` from `bun`, so that project needs `@types/bun` installed: `bun add -d @types/bun`, which `bun init` already does. It is declared as an *optional* peer dependency rather than a runtime dependency, so `bun install` never pulls a types package into a production install. #### Usage Bun's SQL client is built in — nothing to install, and nothing to configure beyond the connection: ```ts import { BunBoss } from 'bun-boss' const boss = new BunBoss('postgres://user:pass@localhost:5432/mydb') await boss.start() await boss.createQueue('email') await boss.send('email', { to: 'user@example.com' }) ``` To share a client your application already owns (or to scope a single operation to a `sql.begin()` transaction — see [Database Adapters](api/adapters.md#bun)), wrap it with `fromBunSql`: ```ts import { SQL } from 'bun' import { BunBoss, fromBunSql } from 'bun-boss' const sql = new SQL('postgres://user:pass@localhost:5432/mydb') const boss = new BunBoss({ db: fromBunSql(sql) }) ``` With a bring-your-own client, bun-boss does **not** open or close it — construct it before `boss.start()` and `sql.close()` it after `boss.stop()`. The built-in client's lifecycle is bun-boss's own: it opens on `start()` and closes on `stop()`. #### No LISTEN/NOTIFY Bun's SQL client [does not implement LISTEN or NOTIFY](https://bun.com/docs/api/sql#postgresql-specific-features). The built-in driver therefore exposes no listener, and `useListenNotify: true` emits a `listen_notify_unavailable` warning and continues with polling. Nothing is lost but wake-up latency — a NOTIFY is only ever a hint that makes workers poll sooner, never a correctness requirement. The producer side is unaffected: the `pg_notify` bun-boss inlines into inserts is evaluated by PostgreSQL itself, so a queue can stay opted into `notify` and any listener on another connection (e.g. a `fromPglite`-backed instance, or your own session holding `LISTEN`) can still act on it. #### Bring-your-own clients: keep Bun's default `prepare: true` Bun derives each parameter's wire encoding from the type PostgreSQL reports for that placeholder, which only happens when statements are prepared. Under `prepare: false`, an object bound to an uncast jsonb placeholder is sent in a form PostgreSQL rejects, so that option is not supported. The built-in driver forwards only connection settings to Bun's `SQL` (`src/db.ts` allowlist), so `prepare` passed to the `BunBoss` constructor is silently ignored and the default always applies. The constraint only bites a client you construct yourself and pass via `fromBunSql`. Bun caches prepared statements per connection, keyed by query text. bun-boss generates its SQL per queue table, so that cache grows with the number of partitioned queues — worth watching in `pg_prepared_statements` on deployments with very many of them. #### Multi-statement blocks Schema installs and maintenance run as a single `BEGIN … COMMIT` block, which Bun refuses on a pooled connection. The adapter replays those on a reserved connection automatically — you do not need `max: 1`. ### SQLite (embedded, via Bun.SQL) SQLite is the one supported backend that is **not** a Postgres-compatible engine — it is a different SQL dialect. The `sqlite` profile enables every compatibility flag, and bun-boss renders alternate SQL for it throughout: TEXT ISO-8601 timestamps, TEXT uuids and JSON, a CHECK-constrained state column, `json_each` in place of arrays, and the atomic-`UPDATE` claim in place of row locking. The only supported driver is [Bun's built-in SQL client](https://bun.com/docs/api/sql) opened on a `sqlite://` URL, reached through the `fromBunSqlite` adapter. Requires Bun 1.3.14+ (the package floor; Bun's sqlite support in `SQL`). #### Usage ```ts import { SQL } from 'bun' import { BunBoss, fromBunSqlite } from 'bun-boss' const sql = new SQL('sqlite://app.db') // or 'sqlite://:memory:' const boss = new BunBoss({ backend: 'sqlite', db: fromBunSqlite(sql) }) await boss.start() await boss.createQueue('email') await boss.send('email', { to: 'user@example.com' }) ``` `backend: 'sqlite'` **requires** a `db` adapter — there is no built-in SQLite driver — and **rejects** `url`: the connection lives on the `SQL` instance you hand to `fromBunSqlite`, so `new BunBoss({ backend: 'sqlite', db: fromBunSqlite(sql), url: 'sqlite://app.db' })` throws. The `db` assert fires first, so a literal that omits both is rejected for the missing adapter. Because bun-boss's tables live in the **same database file** as your application's (namespaced by a quoted `"schema.table"` prefix, `"bunboss.table"` by default — see [Table isolation](#table-isolation)), a job enqueued inside a transaction opened through the adapter's `withTransaction` (passed as the operation's `db`) commits atomically with your application writes — see [Database Adapters](api/adapters.md#sqlite-bun) for the pattern. #### Single-process, single logical connection SQLite is a single-writer embedded database. The adapter serializes every statement and every transaction block internally, and it enables `PRAGMA foreign_keys = ON` and a `PRAGMA busy_timeout` on first use. Like PGlite, you own the instance lifecycle — construct the `SQL` instance before `boss.start()` and `sql.close()` it after `boss.stop()`. Running multiple bun-boss **processes** against the same database file is not supported; use worker concurrency within one process instead. #### What is different from the Postgres backends One thing that is *not* different: like every backend, sqlite installs fresh at the current schema version, and upgrading bun-boss against an older install fails with an explicit error rather than migrating in place. - **No LISTEN/NOTIFY**: workers rely on polling (the correctness floor on every backend). - **`findJobs({ data })`** matches shallowly: every top-level key/value in the filter must match; nested objects compare as JSON text rather than by deep containment. - Relative `startAfter` strings (`'5 minutes'`) are parsed by bun-boss rather than the database; the supported grammar is `N unit` sequences (`seconds/minutes/hours/days/weeks`, singular and abbreviated forms accepted, optionally comma-separated) and `HH:MM[:SS]`. - **Flows** verify all-or-nothing creation in code inside a real transaction rather than via Postgres's statement-level error signal. A bring-your-own `IDatabase` that omits `withTransaction` therefore loses flow atomicity, and when a flow fails inside a caller-owned transaction (`{ db }`), the transaction itself stays usable — roll it back rather than committing after a caught flow error. ## Scaling beyond a single table For very high-throughput workloads (thousands of jobs per second), a single job table becomes the bottleneck regardless of the fetch strategy. ### Application-level sharding A more scalable approach is to shard work at the application level across several queues: ```typescript // Each worker claims a shard; shards are separate queues. const workerId = Number(process.env.WORKER_ID ?? 0) // 0, 1, 2, ... const totalWorkers = Number(process.env.TOTAL_WORKERS ?? 1) for (let shard = 0; shard < totalWorkers; shard++) { await boss.createQueue(`my-queue-${shard}`) } // Producers route each job to a shard. await boss.send(`my-queue-${hash(jobKey) % totalWorkers}`, jobData) // Each worker only polls its own shard's queue. await boss.work(`my-queue-${workerId}`, handler) ``` `work()` has no `singletonKey` option — `singletonKey` constrains enqueue-side concurrency for a single queue and cannot partition consumers, so passing it to `work()` is silently ignored and every worker drains every shard. ### When to use alternative systems **Use bun-boss** (database-backed queue) when: - Throughput is under ~10,000 jobs/second (PostgreSQL handles this comfortably) - Processing time >> fetch time (typical for background jobs) - Transactional consistency with your data is required - You want to minimize infrastructure complexity **Consider dedicated message queues** (Kafka, Redis Streams) when: - Sustained throughput exceeds ~50,000 jobs/second - Job processing times are sub-millisecond - Fire-and-forget semantics are acceptable **Throughput reference points:** - PostgreSQL job queues: 7–30k jobs/sec ([benchmarks](https://gist.github.com/chanks/7585810), [Tembo MQ](https://legacy.tembo.io/blog/mq-stack-benchmarking/)) - Kafka: 1–2M messages/sec ([LinkedIn](https://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines), [Honeycomb](https://developer.confluent.io/learn-more/podcasts/handling-2-million-apache-kafka-messages-per-second-at-honeycomb/)) - Redis Streams: 1–7M messages/sec ([benchmarks](https://goatreview.com/building-a-high-performance-message-queue-with-redis-streams/)) ## Known limitations and race conditions These apply when running with `noSkipLocked` (the atomic-UPDATE fetch path). ### Cache staleness bun-boss caches queue metadata (including active singleton keys) with a configurable refresh interval (`queueCacheIntervalSeconds`, default 60s). Under high concurrency: - Two workers may both see stale cache showing no active singletons - Both attempt to claim jobs with the same singleton key - The `state < 'active'` recheck prevents duplicate claims, but one worker receives empty results This is a performance issue, not a correctness issue — no job is processed twice. ### Empty results under contention With `noSkipLocked`, when multiple workers fetch concurrently: 1. All workers' CTEs may select the same candidate jobs (no row locking) 2. All workers attempt the `UPDATE` 3. One succeeds, the others fail the `state < 'active'` recheck 4. Failed workers receive empty results This is the documented trade-off. For job queues where processing time >> fetch time, this is acceptable — workers simply poll again. ### Compatibility notes - All job-level bun-boss features (priorities, groups, singletons, retries, etc.) work on every backend. Two deployment-level features do not: per-queue `partition: true` requires `backend: 'postgres'` or `'pglite'` (it is accepted but has no effect under `noTablePartitioning`), and LISTEN/NOTIFY is unavailable on SQLite. - The atomic-`UPDATE` fetch (`noSkipLocked`) offers no benefit on stock PostgreSQL — under contention workers receive empty results instead of efficiently skipping to unlocked rows — which is why it is only enabled for backends that need it, never on `backend: 'postgres'`. --- # File: docs/api/constructor.md # Constructor ### `new(url)` Passing a string argument to the constructor implies a PostgreSQL connection string, passed to [Bun's SQL client](https://bun.com/docs/api/sql) as its connection URL. ```js const boss = new BunBoss('postgres://user:pass@host:port/database?sslmode=require'); ``` ### `new(options)` The following options can be set as properties in an object for additional configurations. **Connection options** These carry [Bun's SQL client](https://bun.com/docs/api/sql) option names and are forwarded to it verbatim, so its defaults and semantics apply. Every timeout is in **seconds**. Only the options listed below are forwarded. Bun's aliases (`host`, `user`, `pass`, `ssl`) and `prepare`/`bigint` are not accepted and are silently ignored — `prepare` and `bigint` are excluded deliberately, because the adapter's parameter encoding depends on Bun's defaults for both. TypeScript rejects all of them at compile time; JavaScript does not. `application_name`, `schema`, and `db` are bun-boss's own options rather than Bun's, and `application_name` is reshaped into Bun's `connection` object. * **hostname** - string, defaults to "localhost" * **port** - int or string, defaults to 5432 * **tls** - boolean, object, or `Bun.file()` certificate * **database** - string * **username** - string * **password** - string, or a function returning a string or `Promise` (resolved per connection) * **url** - string PostgreSQL connection string, used as an alternative to `hostname`, `port`, `tls`, `database`, `username`, `password`. Don't combine the two: any of those options set alongside `url` overrides the corresponding part of the URL, so `{ url: 'postgres://host/mydb', database: 'other' }` silently connects to `other`. * **path** - string Unix domain socket path, used instead of `hostname` and `port`. * **max** - int, defaults to 10 Maximum number of connections that will be shared by all operations in this instance * **connectionTimeout** - int, defaults to 30 Seconds to wait when establishing a connection. Set to `0` to wait indefinitely. * **idleTimeout** - int, defaults to 0 Seconds a pooled connection may sit idle before it is closed. `0` never closes it. * **maxLifetime** - int, defaults to 0 Seconds a pooled connection may live before it is recycled. `0` never recycles it. * **application_name** - string, defaults to "bunboss" Reported to PostgreSQL as the session's `application_name`. * **db** - object Passing an object named db allows you "bring your own database connection". This option may be beneficial if you'd like to use an existing database service with its own connection pool. Setting this option will bypass the above configuration. The expected interface is an object with an `executeSql` method that allows the following code to run without errors. ```js const text = "select $1 as input" const values = ['arg1'] const { rows } = await executeSql(text, values) assert(rows[0].input === 'arg1') ``` `executeSql` is the only required member. Two optional capabilities extend it: `withTransaction`, without which bun-boss's multi-statement operations (upsert, the split complete/fail/expire, flow resolution) run without a transaction, and `listen`, without which `useListenNotify` (below) degrades to polling. See [Adapters](./adapters.md) for the full interface and the adapters bun-boss ships with. * **schema** - string, defaults to "pgboss" (schema mode) or "bunboss" (prefix mode) The namespace that contains all required storage objects — a Postgres schema in schema mode, or the name folded into each table's quoted identifier in prefix mode (see `tableIsolation` below). Unquoted, only alphanumeric and underscore are allowed, and the name may not start with a number. Quoted (see below), any character is allowed except double quotes, single quotes, percent signs, periods, dollar signs, backslashes and control characters. Either way the limit is <= 50 bytes. To use a name that isn't a legal bare identifier — one containing dashes, or a reserved word — quote it yourself: ```js new BunBoss({ schema: '"My-Schema"' }) ``` The value is used verbatim as an identifier, so the quotes are preserved as written. Note that `MySchema` and `"MySchema"` are different schemas: PostgreSQL folds the unquoted form to `myschema`. Double quotes, single quotes, percent signs, periods, dollar signs, backslashes and control characters are rejected inside a quoted name. The length limit is measured in bytes, since it's possible to use multi-byte characters inside a quoted name. PostgreSQL truncates identifiers past 63 bytes without complaint, which would leave the configured name and the stored name permanently out of sync. Because the two spellings look nearly identical but name different schemas, `start()` refuses to install into a schema when another one differing from it only by case already holds a bun-boss installation, and names the spelling that reaches the existing data. Override with `allowSchemaCaseVariant` (below) if two such installations are genuinely intended. * **tableIsolation** - `'schema' | 'prefix'`, default `'schema'` How bun-boss isolates its tables. * `'schema'` (default on Postgres and PGlite) — a dedicated Postgres schema, so objects are `schema.job`, `schema.queue`, etc. * `'prefix'` — the `schema` name is folded into a single quoted identifier per object (`"schema.job"`) that lives in the connection's default schema (e.g. `public`), alongside your application's own tables. Use this to avoid a separate schema. Prefix mode does not create a schema and **disables table partitioning** (all queues share one job table); everything else — `SKIP LOCKED`, LISTEN/NOTIFY, advisory locks — works as usual on Postgres. The SQLite backend has no schemas, so it always uses prefix mode; setting `tableIsolation: 'schema'` there is rejected. In prefix mode the `schema` default is `"bunboss"` rather than `"pgboss"`. **Operations options** * **supervise**, bool, default true If this is set to false, flows, maintenance, and monitoring operations will be skipped on this instance. This is an advanced use case, and not something you would want to do under normal circumstances. * **schedule**, bool, default true If this is set to false, this instance will not monitor or create scheduled jobs. This is an advanced use case you may want to do for testing or if the clock of the server is skewed and you would like to disable the skew warnings. * **migrate**, bool, default true If this is set to false, this instance will verify the schema during `start()` instead of installing it, skipping any schema mutation. If the schema is missing, `start()` will throw an error and block usage. This is an advanced use case when the configured user account does not have schema mutation privileges. * **useListenNotify**, bool, default false Enables a `LISTEN/NOTIFY` listener so that workers on notify-enabled queues are woken the moment a job is created, instead of waiting out their `pollingIntervalSeconds`. This is a latency optimization layered on top of polling — polling always remains active as a fallback, so jobs are never lost if a notification is missed. See [Low-latency dispatch with LISTEN/NOTIFY](./workers.md#low-latency-dispatch-with-listennotify) for the full picture and the per-queue `notify` option that controls which queues emit notifications. This option requires a `db` adapter that implements `listen` (e.g. `fromPglite`). The built-in driver — Bun's SQL client — implements no LISTEN, so with it bun-boss emits a [`warning`](./events.md#warning) event of type `listen_notify_unavailable` and continues with polling only. The producer side (`pg_notify` inlined into inserts) still fires either way, so a listener on another connection can act on it. The following configuration options should not normally need to be changed, but are still available for special use cases. * **createSchema**, bool, default true If set to false, the `CREATE SCHEMA` statement will not be issued during installation. This may be useful if this privilege is not granted to the role. * **allowSchemaCaseVariant**, bool, default false If set to true, `start()` will install into `schema` even when another schema differing from it only by case already holds a bun-boss installation. The check this disables exists because `schema: 'MySchema'` and `schema: '"MySchema"'` name two different schemas — PostgreSQL folds the unquoted form to `myschema` and stores the quoted one verbatim. Mis-spelling the quoting is not an error on its own: bun-boss simply finds no installation, creates an empty second schema, and every existing job appears to have vanished. Only set this if two installations whose names differ by case are intended. * **superviseIntervalSeconds**, int, default 60 seconds Entry point for how often queues are monitored and maintained. * **maintenanceIntervalSeconds**, int, default 1 day How often maintenance will be run against queue tables to drop queued and completed jobs. * **monitorIntervalSeconds**, int, default 60 seconds How often each queue is monitored for backlogs, expired jobs, and calculating stats. * **queueCacheIntervalSeconds**, int, default 60 seconds How often queue metadata is refreshed in memory. * **flowIntervalSeconds**, int, default 5 seconds How often the background flow resolver runs to unblock dependent jobs (created via [`flow()`](./jobs.md#flowjobs-options)) whose parents have completed. Completing a job no longer unblocks its dependents inline; this resolver handles it shortly after, off the completion hot path. Only runs when `supervise` is enabled. * **cronMonitorIntervalSeconds**, int, default 30 seconds How often schedules are checked so that due scheduled jobs are sent. Must be between 1 and 45 seconds. Only runs when `schedule` is enabled. * **cronWorkerIntervalSeconds**, int, default 5 seconds How often the internal worker that runs scheduled jobs polls. Must be between 1 and 45 seconds. Only runs when `schedule` is enabled. * **clockMonitorIntervalSeconds**, int, default 600 seconds How often this instance compares its clock to the database server's clock, so that the skew can be used as an offset during cron evaluation. Must be between 1 second and 10 minutes. Only runs when `schedule` is enabled. * **warningSlowQuerySeconds**, int, default 30 The threshold, in seconds, above which a monitoring or maintenance query emits a `slow_query` [`warning`](./events.md#warning) event. Applies per instance and must be at least 1. * **warningQueueSize**, int, default 10000 The default number of jobs in the created or retry state a queue may hold before emitting a `queue_backlog` [`warning`](./events.md#warning) event. Applies per instance and must be at least 1. Individual queues can override this with their own [`warningQueueSize`](./queues.md#createqueuename-options) on `createQueue`. * **backend**, string, default `'postgres'` Selects the database bun-boss is running against and applies the compatibility behavior it needs. One of `'postgres'`, `'pglite'`, or `'sqlite'`. ```js const boss = new BunBoss({ backend: 'sqlite', db: fromBunSqlite(sql) }) ``` Based on this setting, the fetch strategy, mutation strategy, and schema shape may be changed. See [Database Backends](../database-backends.md#backend-profiles) for what each backend enables and the [compatibility matrix](../database-backends.md#database-compatibility). --- # File: docs/api/jobs.md # Jobs ## Creating jobs ### `send()` Creates a new job and returns the job id. > [!NOTE] > `send()` will resolve a `null` for job id under some use cases when using unique jobs or throttling (see below). These options are always opt-in on the send side and therefore don't result in a promise rejection. ### `send(name, data, options)` **Arguments** - `name`: string, *required* - `data`: object - `options`: object **General options** * **priority**, int optional priority. Higher numbers have, um, higher priority * **id**, uuid optional id. If not set, a uuid will automatically created * **deadLetter**, string optional dead letter queue for this job, overriding the queue-level `deadLetter`. See [queues](./queues.md). **Retry options** * **retryLimit**, int Default: 2. Number of retries to complete a job. * **retryDelay**, int Default: 0. Delay between retries of failed jobs, in seconds. * **retryBackoff**, bool Default: false. Enables exponential backoff retries based on retryDelay instead of a fixed delay. Sets initial retryDelay to 1 if not set. A simplified function to get the delay between runs is: `retryDelay * 2 ^ retryCount` with some jitter. The full function to determine the backoff delay is `Math.min(retryDelayMax, Math.max(retryDelay, 1) * (2 ** Math.min(16, retryCount + 1) / 2 + 2 ** Math.min(16, retryCount + 1) / 2 * Math.random()))` * **retryDelayMax**, int Default: no limit. Maximum delay between retries of failed jobs, in seconds. Only used when retryBackoff is true. **Heartbeat options** * **heartbeatSeconds**, int Default: none (disabled). Expected heartbeat interval in seconds. Overrides the queue-level `heartbeatSeconds` for this specific job. When set, workers using `work()` will automatically send periodic heartbeats. If no heartbeat is received within this interval, the monitor will fail/retry the job. Must be >= 10. See [Heartbeat vs expiration](./queues.md#heartbeat-vs-expiration) for guidance on when to use this and recommended values. **Expiration options** * **expireInSeconds**, number Default: 15 minutes. How many seconds a job may be in active state before being retried or failed. Must be >=1 **Retention options** * **retentionSeconds**, number Default: 14 days. How many seconds a job may be in created or retry state before it's deleted. Must be >=1 * **deleteAfterSeconds**, int Default: 7 days. How long a job should be retained in the database after it's completed. Set to 0 to never delete completed jobs. All retry, expiration, and retention options can also be set on the queue and will be inherited for each job, unless they are overridden. **Connection options** * **db**, object Instead of using bun-boss's default adapter, you can use your own, as long as it implements the following `executeSql` interface (see [Database Adapters](./adapters.md)). ```ts interface Db { executeSql(text: string, values: any[]): Promise<{ rows: any[] }>; } ``` bun-boss ships with built-in adapters for Bun's SQL client, embedded PGlite, and embedded SQLite. See [Database Adapters](./adapters.md) for details. **Deferred jobs** * **startAfter** int, string, or Date * int: seconds to delay starting the job * string: Start after a UTC Date time string in 8601 format * Date: Start after a Date object Default: 0 **Group options** * **group**, object Assigns a job to a group for use with `groupConcurrency` in `work()`. This allows you to limit how many jobs from the same group can be processed simultaneously. - **id**, string, *required*: The group identifier (e.g., tenant ID, project ID, customer ID) - **tier**, string, *optional*: A tier identifier for tier-based concurrency limits ```js // Assign job to a tenant group await boss.send('process-data', data, { group: { id: 'tenant-123' } }) // Assign job to a group with a tier for tier-based limits await boss.send('process-data', data, { group: { id: 'tenant-456', tier: 'enterprise' } }) ``` **Throttle or debounce jobs** * **singletonSeconds**, int * **singletonNextSlot**, bool * **singletonKey** string Throttling jobs to 'one per time slot'. This option is set on the send side of the API since jobs may or may not be created based on the existence of other jobs. For example, if you set the `singletonSeconds` to 60, then submit 2 jobs within the same minute, only the first job will be accepted and resolve a job id. The second request will resolve a null instead of a job id. Setting `singletonNextSlot` to true will cause the job to be scheduled to run after the current time slot if and when a job is throttled. This option is set to true, for example, when calling the convenience function `sendDebounced()`. As with queue policies, using `singletonKey` will extend throttling to allow one job per key within the time slot. ```js const payload = { email: "billybob@veganplumbing.com", name: "Billy Bob" }; const options = { startAfter: 1, retryLimit: 2 }; const jobId = await boss.send('email-send-welcome', payload, options) console.log(`job ${jobId} submitted`) ``` ### `send({ name, data, options })` This overload supports sending an object with name, data, and options properties. ```js const jobId = await boss.send({ name: 'database-backup', options: { retryLimit: 1 } }) console.log(`job ${jobId} submitted`) ``` ### `sendAfter(name, data, options, value)` Send a job that should start after a number of seconds from now, or after a specific date time. This is a convenience version of `send()` with the `startAfter` option assigned. `value`: int: seconds | string: ISO date string | Date ```js // start in 5 minutes await boss.sendAfter('email-reminder', { userId: 123 }, null, 300) // start at a specific date and time await boss.sendAfter('email-reminder', { userId: 123 }, null, new Date('2027-01-01T08:00:00Z')) ``` ### `sendThrottled(name, data, options, seconds, key)` Only allows one job to be sent to the same queue within a number of seconds. In this case, the first job within the interval is allowed, and all other jobs within the same interval are rejected. This is a convenience version of `send()` with the `singletonSeconds` and `singletonKey` option assigned. The `key` argument is optional. ```js // accept at most 1 job per user per minute; extra sends resolve null const jobId = await boss.sendThrottled('sync-profile', { userId: 123 }, null, 60, `user-${123}`) if (!jobId) { console.log('job was throttled') } ``` ### `sendDebounced(name, data, options, seconds, key)` Like, `sendThrottled()`, but instead of rejecting if a job is already sent in the current interval, it will try to add the job to the next interval if one hasn't already been sent. This is a convenience version of `send()` with the `singletonSeconds`, `singletonKey` and `singletonNextSlot` option assigned. The `key` argument is optional. ```js // coalesce bursts of edits: at most 1 job per document per 30 seconds, // and if the slot is taken, schedule one for the next slot await boss.sendDebounced('reindex-document', { docId: 'doc-1' }, null, 30, 'doc-1') ``` ### `update(name, data, options)` Updates the payload and options of one or more **not-yet-active** jobs (state `created` or `retry`). Jobs that are already `active`, `completed`, or otherwise terminal cannot be updated. If nothing matches, it resolves with an empty `jobs` array and `updated: 0`. Target the job with **exactly one** of: - `options.id` — a single job by id. - `options.singletonKey` — jobs sharing that key. Only the fields you supply are changed, and any option you omit is left at the job's current value. Passing just a new `data` payload with a target replaces the payload without disturbing the job's existing `startAfter`, `priority`, retry settings, etc. Updatable fields are: * `data` * `priority` * `startAfter` * `retryLimit` * `retryDelay` * `retryBackoff` * `retryDelayMax` * `expireInSeconds` * `retentionSeconds` * `deleteAfterSeconds` * `deadLetter` * `heartbeatSeconds` * `group` The job's `singletonKey` and `singletonOn` (throttle slot) are always preserved. To leave the payload unchanged while editing only options, pass `undefined` for `data`; passing `null` clears it. If the updated job ends up runnable (its `startAfter` is now in the past) on a queue with `LISTEN`/`NOTIFY` enabled, a wake-up notification is emitted so idle workers fetch it promptly. Returns a `Promise`: `{ jobs, updated }`, where `jobs` are the affected ids and `updated` is the number of jobs updated (equal to `jobs.length`). ```js // by id await boss.update('email', { to: 'a@b.co' }, { id: jobId }) // by singletonKey await boss.update('article', { articleId: 42, body: '…latest…' }, { singletonKey: 'article-42' }) ``` Because a `singletonKey` is only guaranteed unique per state under the `short` and `stately` policies, several pre-active jobs can share a key (for example under throttling/debouncing, or with a manually-assigned key on a `standard` queue). Use `options.match` to choose which are updated, ordered by `createdOn`: - `newest` (default) — overwrite the most recently created match. - `oldest` — overwrite the earliest created match. - `all` — overwrite every match. `match` is only valid when targeting by `singletonKey`. ### `update({ name, data, options })` This overload supports updating a job with a single object with name, data, and options properties. `data` is optional — omit it to edit only options. ```js await boss.update({ name: 'article', data: { articleId: 42, body: '…latest…' }, options: { singletonKey: 'article-42' } }) ``` ### `upsert(name, data, options)` Update-or-insert one or more **not-yet-active** jobs (state `created` or `retry`). Confused yet? This is more of a special use case and probably shouldn't replace the normal usage of `send()`. Think of `upsert()` as a convenience abstraction over 2 steps: `update()` first, but if no matches were found, then `insert()`. The same options are used here as in `update()`. When matching by `id`, the new job is created with that id. It supports the same `match` option as `update()` when using `singletonKey`. Returns a `Promise`: `{ jobs, updated, inserted }`. On a hit, `updated` reflects the updated job(s) and `inserted` is `0`; on a miss, `inserted` is `1` and `jobs` holds the new id. ```js // ensure exactly one queued "process this article" job carries the latest body await boss.upsert('article', { articleId: 42, body: '…latest…' }, { singletonKey: 'article-42' }) ``` ### `upsert({ name, data, options })` This overload supports upserting a job with a single object with name, data, and options properties, mirroring `update({ name, data, options })`. ```js await boss.upsert({ name: 'article', data: { articleId: 42, body: '…latest…' }, options: { singletonKey: 'article-42' } }) ``` ### `insert(name, JobInsert[], options)` Create multiple jobs in one request with an array of objects. The contract and supported features are slightly different than `send()`, which is why this function is named independently. For example, debouncing is not supported, and it doesn't return job IDs unless spies are enabled or `options.returnId` is set to `true`. The following contract is a TypeScript definition of the expected object. This will likely be enhanced later with more support for deferral and retention by an offset. For now, calculate any desired timestamps for these features before insertion. ```ts interface JobInsert { id?: string, data?: T; priority?: number; retryLimit?: number; retryDelay?: number; retryBackoff?: boolean; retryDelayMax?: number; startAfter?: number | Date | string; singletonKey?: string; singletonSeconds?: number; expireInSeconds?: number; retentionSeconds?: number; heartbeatSeconds?: number; deleteAfterSeconds?: number; deadLetter?: string; group?: { id: string; tier?: string }; } ``` ```js const [idA, idB] = await boss.insert('etl', [ { data: { step: 'extract' } }, { data: { step: 'transform' } } ], { returnId: true }) ``` ## Flows ### `flow(jobs, options)` Create a set of jobs and their dependencies atomically in one transaction. Use `flow()` when jobs depend on other jobs. Dependencies are not configured on `send()` or `insert()` because creating jobs and dependencies in separate calls can race with job completion. Atomicity is handled by bun-boss when it owns the database connection. If you pass a custom `db` in `options`, wrap the call in your own transaction if you need the job and dependency inserts to commit or roll back together. The method accepts a flat array of jobs in any order. Each job has a local `ref`, and dependent jobs reference parent refs with `dependsOn`. ```ts interface FlowJob { ref: string; name: string; data?: object; options?: Omit; dependsOn?: string[]; } ``` Returns a map of `ref` to created job id. ```js const flow = await boss.flow([ { ref: 'extract-a', name: 'extract', data: { file: '1.csv' } }, { ref: 'extract-b', name: 'extract', data: { file: '2.csv' } }, { ref: 'load', name: 'load', data: { output: 'report.csv' }, dependsOn: ['extract-a', 'extract-b'] } ]) const loadJobId = flow.load ``` Dependent jobs are created in a `blocked` state and won't be eligible for fetching until all parent jobs have completed. If a parent job fails or is cancelled, the child remains blocked. You can explicitly `cancel` or `fail` the blocked child if needed. When a dependent job uses `startAfter`, both conditions must be met: all dependencies completed and `startAfter` has passed. Unblocking happens off the completion hot path: a background resolver wakes shortly after a parent completes (see `flowIntervalSeconds` in the [constructor options](./constructor.md)) and unblocks any dependents that are now ready. This keeps completing jobs fast regardless of how many flows exist. The resolver runs when `supervise` is enabled; call [`resolveFlow()`](#resolveflow) to force a pass immediately (e.g. in tests). ### `resolveFlow()` Forces an immediate flow-resolution pass instead of waiting for the next background cycle, unblocking dependents of any parents that have completed. Returns a promise that resolves when the pass finishes. Useful for deterministic tests, or when you have disabled `supervise` and drive maintenance yourself. ```js await boss.complete('extract', parentJobId) // unblock any ready dependents now instead of waiting for the next cycle await boss.resolveFlow() ``` ## Fetching jobs ### `fetch(name, options)` Returns an array of jobs from a queue **Arguments** - `name`: string - `options`: object * `batchSize`, int, *default: 1* Number of jobs to return * `priority`, bool, *default: true* If true, allow jobs with a higher priority to be fetched before jobs with lower or no priority * `orderByCreatedOn`, bool, *default: true* If true, jobs are fetched in the order they were created. Set to false to disable this sorting for improved performance when order doesn't matter. * `includeMetadata`, bool, *default: false* If `true`, all job metadata will be returned on the job object. * `ignoreStartAfter`, bool, *default: false* If `true`, jobs with a `startAfter` timestamp in the future will be fetched. Useful for fetching jobs immediately without waiting for a retry delay. * `minPriority`, int If set, only fetch jobs with a priority greater than or equal to this value. If used together with `maxPriority`, `minPriority` must be less than or equal to `maxPriority`. * `maxPriority`, int If set, only fetch jobs with a priority less than or equal to this value. If used together with `minPriority`, `minPriority` must be less than or equal to `maxPriority`. * `groupConcurrency`, int | object, *default: none* Cap concurrently active jobs per group across all nodes, using the same shape as the `work()` option — a number, or `{ default, tiers }`. See [group concurrency](./workers.md). * `ignoreGroups`, string[], *default: none* Skip jobs whose `group.id` is in this list, so a worker can drain ungrouped or other-group work while a saturated group is handled elsewhere. Fetch-only — there is no `work()` equivalent. **Job metadata** With `includeMetadata: true`, each job is returned with all metadata: ```ts interface JobWithMetadata { id: string; name: string; data: T; priority: number; state: 'created' | 'retry' | 'active' | 'completed' | 'cancelled' | 'failed'; retryLimit: number; retryCount: number; retryDelay: number; retryBackoff: boolean; retryDelayMax?: number; startAfter: Date; startedOn: Date; singletonKey: string | null; singletonOn: Date | null; groupId?: string | null; groupTier?: string | null; expireInSeconds: number; heartbeatSeconds: number | null; heartbeatOn: Date | null; deleteAfterSeconds: number; createdOn: Date; completedOn: Date | null; keepUntil: Date; blocked: boolean, blocking: boolean, pendingDependencies: number, deadLetter: string, policy: string, output: object, sourceName: string | null, sourceId: string | null, sourceCreatedOn: Date | null, sourceRetryCount: number | null } ``` Although `signal` is declared on the exported `Job` type, it is not present on jobs returned by `fetch()`; it is only attached to the job passed to a `work()` handler. When a job is moved into a dead letter queue, the `source*` fields record where it came from: `sourceName` is the queue it originally failed on, `sourceId` is the id of the original job, `sourceCreatedOn` is the original job's creation time (so its true age survives the move), and `sourceRetryCount` is how many retries it consumed before being dead-lettered. These are `null` for jobs that were not dead-lettered. **Notes** The following example shows how to fetch and delete up to 20 jobs. ```js const QUEUE = 'email-daily-digest' import emailer from './emailer.js' const jobs = await boss.fetch(QUEUE, { batchSize: 20 }) await Promise.allSettled(jobs.map(async job => { try { await emailer.send(job.data) await boss.deleteJob(QUEUE, job.id) } catch(err) { await boss.fail(QUEUE, job.id, err) } })) ``` ## Deleting and redriving jobs ### `deleteJob(name, id, options)` Deletes a job by id. > [!NOTE] > Job deletion is offered if desired for a "fetch then delete" workflow similar to SQS. This is not the default behavior for workers so "everything just works" by default, including job throttling and debouncing, which requires jobs to exist to enforce a unique constraint. For example, if you are debouncing a queue to "only allow 1 job per hour", deleting jobs after processing would re-open that time slot, breaking your throttling policy. ```js const [job] = await boss.fetch('email-send') await emailer.send(job.data) await boss.deleteJob('email-send', job.id) ``` ### `deleteJob(name, [ids], options)` Deletes a set of jobs by id. ```js const jobs = await boss.fetch('email-send', { batchSize: 20 }) await boss.deleteJob('email-send', jobs.map(job => job.id)) ``` ### `redrive(name, options)` Moves jobs out of a dead letter queue and re-creates them as fresh jobs on their original source queue. `name` is the dead letter queue to drain. Returns the number of jobs moved. Each job is routed back to the queue it originally failed on (its `sourceName`), so a single dead letter queue that collects from many source queues fans back out correctly. Re-created jobs get a new id, a reset retry count, cleared output, and the destination queue's current retry, retention, and policy configuration. Only jobs that are not currently being processed (still in the `created`/`retry` state) are moved. `options`: - `destination` — override queue to move all matched jobs into, instead of each job's original source queue. Required to redrive jobs that have no recorded source queue (e.g. jobs dead-lettered before this feature existed); such jobs are left in place otherwise. - `sourceName` — only redrive jobs that originated from this source queue. - `limit` — maximum number of jobs to move in this call, oldest first (default `1000`). Loop or schedule repeated calls to drain large dead letter queues at a controlled rate. ```js // drain a dead letter queue back to its source queues, 500 at a time let moved do { moved = await boss.redrive('email-dlq', { limit: 500 }) } while (moved > 0) ``` ### `deleteQueuedJobs(name)` Deletes all queued jobs in a queue. ```js await boss.deleteQueuedJobs('email-send') ``` ### `deleteStoredJobs(name)` Deletes all jobs in completed, failed, and cancelled state in a queue. ```js await boss.deleteStoredJobs('email-send') ``` ### `deleteAllJobs(name?)` Deletes all jobs in a queue, including active jobs. If no queue name is given, jobs are deleted from all queues. ```js // delete everything in one queue await boss.deleteAllJobs('email-send') // delete everything in all queues await boss.deleteAllJobs() ``` ## Cancelling, resuming, and retrying jobs ### `cancel(name, id, options)` Cancels a pending or active job. ```js await boss.cancel('email-send', jobId) ``` ### `cancel(name, [ids], options)` Cancels a set of pending or active jobs. When passing an array of ids, it's possible that the operation may partially succeed based on the state of individual jobs requested. Consider this a best-effort attempt. ```js await boss.cancel('email-send', [jobId1, jobId2]) ``` ### `resume(name, id, options)` Resumes a cancelled job. ```js await boss.resume('email-send', jobId) ``` ### `resume(name, [ids], options)` Resumes a set of cancelled jobs. ```js await boss.resume('email-send', [jobId1, jobId2]) ``` ### `retry(name, id, options)` Retries a failed job. ```js await boss.retry('email-send', jobId) ``` ### `retry(name, [ids], options)` Retries a set of failed jobs. ```js // requeue all failed jobs for another attempt const failed = await boss.findJobs('email-send') const ids = failed.filter(job => job.state === 'failed').map(job => job.id) await boss.retry('email-send', ids) ``` ## Completing and failing jobs ### `complete(name, id, data, options)` Completes an active job. This would likely only be used with `fetch()`. Accepts an optional `data` argument for job output and an optional `options` object. ```js const [job] = await boss.fetch('report-generation') const report = await generateReport(job.data) await boss.complete('report-generation', job.id, { reportUrl: report.url }) ``` **options** * **includeQueued**, bool Default: false. When false (default), only jobs in `active` state can be completed. When true, jobs in `created`, `retry`, or `active` states can be completed. This is useful for completing jobs that haven't been fetched yet, or for marking failed jobs as complete without retrying them. ```js // Complete a job without fetching it first await boss.complete('my-queue', jobId, { result: 'done' }, { includeQueued: true }) ``` * **db**, object, see notes in `send()` The promise will resolve on a successful completion, or reject if the job could not be completed. ### `complete(name, [ids], data, options)` Completes a set of active jobs (or queued jobs when `includeQueued: true` is specified). The promise will resolve on a successful completion, or reject if not all of the requested jobs could not be marked as completed. > [!NOTE] > See comments above on `cancel([ids])` regarding when the promise will resolve or reject because of a batch operation. ### `fail(name, id, data, options)` Marks an active job as failed. The promise will resolve on a successful assignment of failure, or reject if the job could not be marked as failed. ```js const [job] = await boss.fetch('email-send') try { await emailer.send(job.data) await boss.complete('email-send', job.id) } catch (err) { // stored in the job's output and eligible for retry per the queue config await boss.fail('email-send', job.id, err) } ``` ### `fail(name, [ids], data, options)` Fails a set of active jobs. ```js const jobs = await boss.fetch('email-send', { batchSize: 10 }) await boss.fail('email-send', jobs.map(job => job.id), { message: 'smtp outage' }) ``` The promise will resolve on a successful failure state assignment, or reject if not all of the requested jobs could not be marked as failed. > [!NOTE] > See comments above on `cancel([ids])` regarding when the promise will resolve or reject because of a batch operation. ### `touch(name, id, options)` Updates the heartbeat timestamp for an active job, signaling that the worker is still alive. This is useful when using `fetch()` for manual job processing. Workers using `work()` send heartbeats automatically when `heartbeatSeconds` is configured. ```js const [job] = await boss.fetch('long-running-queue') const interval = setInterval(async () => { await boss.touch('long-running-queue', job.id) }, 5000) try { await processJob(job) await boss.complete('long-running-queue', job.id) } finally { clearInterval(interval) } ``` ### `touch(name, [ids], options)` Updates the heartbeat timestamp for a set of active jobs. ```js const jobs = await boss.fetch('long-running-queue', { batchSize: 10 }) const ids = jobs.map(j => j.id) const result = await boss.touch('long-running-queue', ids) ``` ## Finding jobs ### `getJobById(name, id, options)` > [!WARNING] > **Deprecated:** Use `findJobs()` instead. Retrieves a job with all metadata by name and id **options** * **db**, object, see notes in `send()` ### `findJobs(name, options)` Finds jobs in a queue by id, singleton key, and/or data. Returns an array of jobs with all metadata. **Arguments** - `name`: string, *required* - `options`: object **options** * **id**, string Find a job by its id * **key**, string Find jobs by their singletonKey * **data**, object Find jobs where the job data contains the specified key-value pairs (top-level matching only) * **queued**, bool, *default: false* If `true`, only return jobs in queued state (created or retry). If `false`, return jobs in any state. * **db**, object, see notes in `send()` **Examples** ```js // Find by id const byId = await boss.findJobs('my-queue', { id: '9b45b709-910a-4a9f-b57a-9d39da3d4033' }) // Find by singletonKey const byKey = await boss.findJobs('my-queue', { key: 'user-123' }) // Find by data const byData = await boss.findJobs('my-queue', { data: { type: 'email' } }) // Find queued jobs only const queuedOnly = await boss.findJobs('my-queue', { key: 'user-123', queued: true }) // Combine filters const combined = await boss.findJobs('my-queue', { key: 'user-123', data: { type: 'email' }, queued: true }) ``` ## Inspecting dependencies ### `getDependencies(name, id, options)` Returns an array of parent job references that the specified job depends on. ```js const parents = await boss.getDependencies('aggregate-results', jobId) // [{ name: 'process-data', id: '...' }, { name: 'process-data', id: '...' }] ``` ### `getDependents(name, id, options)` Returns an array of child job references that depend on the specified job. ```js const children = await boss.getDependents('process-data', parentJobId) // [{ name: 'aggregate-results', id: '...' }] ``` --- # File: docs/api/workers.md # Workers ### `work()` Adds a new polling worker for a queue and executes the provided callback function when jobs are found. Each call to work() will add a new worker and resolve a unique worker id. Workers can be stopped via `offWork()` all at once by queue name or individually by using the worker id. Worker activity may be monitored by listening to the `wip` event or by polling [`getWipData()`](#getwipdataoptions). The default options for `work()` is 1 job every 2 seconds. ### `work(name, options, handler)` **Arguments** - `name`: string, *required* - `options`: object - `handler`: function(jobs): `Promise`, *required* **Options** * **batchSize**, int, *(default=1)* Same as in [`fetch()`](./jobs.md#fetchname-options) * **includeMetadata**, bool, *(default=false)* Same as in [`fetch()`](./jobs.md#fetchname-options) * **perJobResults**, bool, *(default=false)* Opt in to per-job settlement for batch handlers. By default a batch handler is all-or-nothing: returning completes every job in the batch (and the return value is only stored as `output` when `batchSize` is 1), while throwing fails every job. When `perJobResults` is true, the handler must instead resolve with an array of `JobResult` objects — one per job it processed — and bun-boss settles each job individually, preserving its own output: ```js await boss.work('resize-image', { batchSize: 10, perJobResults: true }, async (jobs) => { return jobs.map(job => { try { const output = resize(job.data) return { id: job.id, status: 'completed', output } } catch (err) { return err.fatal ? { id: job.id, status: 'deadletter', output: err } : { id: job.id, status: 'failed', output: err } } }) }) ``` Each `JobResult` is `{ id, status, output? }` where `id` matches a job from the batch, `status` is `'completed'`, `'failed'`, or `'deadletter'`, and `output` is stored on that job (the completion result, or the failure detail). Notes: - **`deadletter`** fails the job terminally and routes it straight to the queue's configured dead letter queue, bypassing any remaining retries (the `output` travels to the dead letter job). If the queue has no dead letter queue configured, the job simply fails terminally — equivalent to a `failed` job that has exhausted its retries. - Any job in the batch the handler omits from the array is **failed** with a descriptive error so it retries (or dead-letters) per the queue config — a returned result is never assumed. - **Throwing** from the handler still fails the entire batch, exactly as without `perJobResults`. Use the returned array to express per-job failures; reserve throwing for batch-wide errors. - Resolving with anything other than an array is treated as a contract violation and fails the whole batch. * **priority**, bool, *(default=true)* Same as in [`fetch()`](./jobs.md#fetchname-options) * **orderByCreatedOn**, bool, *(default=true)* Same as in [`fetch()`](./jobs.md#fetchname-options) * **ignoreStartAfter**, bool, *(default=false)* Same as in [`fetch()`](./jobs.md#fetchname-options) * **minPriority**, int Same as in [`fetch()`](./jobs.md#fetchname-options) * **maxPriority**, int Same as in [`fetch()`](./jobs.md#fetchname-options) * **pollingIntervalSeconds**, number, *(default=2)* Base interval to check for new jobs, in seconds. Must be >=0.5 (500ms). Used when no faster or slower mode applies: queues without `notify`, or notify-enabled queues when the LISTEN/NOTIFY listener is unavailable. > **Note**: When [LISTEN/NOTIFY](#low-latency-dispatch-with-listennotify) is active for a queue, workers are woken the instant a job is created and polling automatically falls back to the slower `notifyPollingIntervalSeconds` backstop — you don't need to raise `pollingIntervalSeconds` yourself. * **notifyPollingIntervalSeconds**, number, *(default=max(30, pollingIntervalSeconds))* Polling interval used only while [LISTEN/NOTIFY](#low-latency-dispatch-with-listennotify) is active for the queue (the queue has `notify: true` and the instance listener is established). Since NOTIFY wakes workers immediately, polling only needs to run as a slow safety net, so this can be much larger than `pollingIntervalSeconds`. When notify is off or unavailable, `pollingIntervalSeconds` is used instead. Must be >=0.5 (500ms) and never below `pollingIntervalSeconds` — the default is floored at `pollingIntervalSeconds` so raising that above 30 raises this too. * **burstWhenReadyExceeds**, int When the queue's cached `readyCount` (the runnable backlog) exceeds this value, the worker fetches continuously with no delay until it catches up; the first fetch that comes back short ends burst mode. Takes precedence over `notifyPollingIntervalSeconds` and `pollingIntervalSeconds`. Must be an integer >=1. > **Note**: `readyCount` is read from the stats cache, so reaction latency is bounded by the instance-level stats pipeline (`monitorIntervalSeconds` / `superviseIntervalSeconds` / `queueCacheIntervalSeconds`). * **burstWhenBatchFull**, bool, *(default=false)* While each fetch returns a full `batchSize` batch there is clearly more work, so the worker keeps fetching continuously with no delay; the first short fetch ends burst mode. Unlike `burstWhenReadyExceeds` this reacts instantly and needs no cached stats. Ignored when `batchSize` is 1 (every successful fetch would otherwise be "full"). * **burstWhileNonEmpty**, bool, *(default=true, unless `burstWhenReadyExceeds` or `burstWhenBatchFull` is set)* While each fetch settles at least one job there is more work ready, so the worker keeps fetching continuously with no delay; the first fetch that settles nothing — an empty fetch, a failed fetch, or a batch whose handler threw for every job — resumes normal polling. Unlike `burstWhenBatchFull` this works at any `batchSize`, including 1 — the poll interval becomes the idle cadence, not a per-job delay while a backlog drains. Set `false` to restore the pre-0.3 behavior of one fetch per poll interval. > **Note**: Because the other two triggers exist to hold burst mode back, setting either of them opts out of this one — otherwise this would burst on every non-empty fetch and make their thresholds moot. Passing `burstWhileNonEmpty: true` explicitly overrides that opt-out, at which point the other trigger no longer has any effect. Gating on *settled* rather than fetched jobs matters because `retryDelay` defaults to 0: a batch whose handler threw is immediately re-fetchable, and bursting on it would burn every `retryLimit` attempt in milliseconds. * **localConcurrency**, int, *(default=1)* Number of workers to spawn for this queue within the current Bun process. Each worker polls and processes jobs independently, enabling parallel job processing within a single `work()` call. > [!NOTE] > This is a per-node setting. In a distributed deployment with multiple nodes, each node manages its own workers independently. For example, if you have 3 nodes each calling `work()` with `localConcurrency: 5`, you'll have 15 total workers across your cluster. ```js // Create 5 workers that can each process jobs in parallel await boss.work('email-welcome', { localConcurrency: 5 }, async ([job]) => { await sendEmail(job.data) }) ``` * **heartbeatRefreshSeconds**, number Custom interval in seconds at which the worker sends heartbeats for active jobs. Defaults to `heartbeatSeconds / 2` (derived from the job's heartbeat configuration). Should be smaller than `heartbeatSeconds` so a heartbeat lands before the contract deadline. This is a worker-level setting only — it is not available on queue or job configuration. The distinction between `heartbeatSeconds` and `heartbeatRefreshSeconds`: - `heartbeatSeconds` (queue/job level) defines the **contract**: how long before a missing heartbeat is considered a failure - `heartbeatRefreshSeconds` (worker level) controls the **implementation**: how often the worker sends heartbeats to fulfill that contract This option only applies when jobs have `heartbeatSeconds` configured (either on the queue or per-job). Heartbeats are sent automatically by `work()` — no user action is needed unless a custom refresh interval is desired. When using `fetch()` for manual processing, call `touch()` directly instead. ```js // Queue configured with 60s heartbeat, worker sends heartbeats every 10s await boss.work('video-processing', { heartbeatRefreshSeconds: 10 }, async ([job]) => { await processVideo(job.data) }) ``` * **groupConcurrency**, int | object Limits how many jobs from the same group can be processed simultaneously **globally across all nodes**. This is enforced via database queries. Can be specified as: - A simple number: `groupConcurrency: 2` - limits all groups to 2 concurrent jobs globally - An object with tier-based limits: ```js groupConcurrency: { default: 1, // Default limit for groups without a tier tiers: { enterprise: 5, // Enterprise tier can have 5 concurrent jobs pro: 2 // Pro tier can have 2 concurrent jobs } } ``` Jobs are assigned to groups using the `group` option in `send()`. Jobs without a group are not limited by groupConcurrency. > [!WARNING] > The `groupConcurrency` limit is enforced globally across all nodes by tracking active jobs in the database. However, due to the optimistic locking nature of job fetching, there may be brief moments where the limit is slightly exceeded during race conditions when multiple workers fetch jobs simultaneously. ```js // Limit each tenant to 2 concurrent jobs globally across all nodes await boss.work('process-data', { localConcurrency: 10, groupConcurrency: 2 }, async ([job]) => { await processData(job.data) }) ``` #### Understanding concurrency options The two concurrency options work together to control job processing at different levels: | Option | Scope | Tracking | Use case | | - | - | - | - | | `localConcurrency` | Per-node | N/A (worker count) | Control total parallel processing capacity per node | | `groupConcurrency` | Global, per-group | Database | Coordinate group limits across distributed nodes | **Key relationships:** - `localConcurrency` sets the maximum number of jobs a single node can process simultaneously (limited by worker count) - `groupConcurrency` can exceed `localConcurrency` because it's a global limit across all nodes **Example: Multi-node deployment** ```js // 3 nodes, each running: await boss.work('process-tenant-data', { localConcurrency: 5, // Each node has 5 workers (15 total across cluster) groupConcurrency: 10 // Max 10 jobs from same tenant globally }, handler) ``` In this setup: - Each node can process up to 5 jobs simultaneously (limited by `localConcurrency`) - Across all 3 nodes, at most 10 jobs from the same group/tenant can be active (enforced by `groupConcurrency` via DB) - This ensures predictable load on external resources (APIs, databases) per tenant **Handler function** `handler` should return a promise (Usually this is an `async` function). If the `handler` returns a value or an object, it will be stored in the `output` property. If an unhandled error occurs in a handler, `fail()` will automatically be called for the jobs, storing the error in the `output` property, making the job or jobs available for retry. > [!TIP] > By default this is all-or-nothing across the batch. To complete and fail individual jobs within a batch — each with its own `output` — enable the **perJobResults** option above. The jobs argument is an array of jobs with the following properties. | Prop | Type | | | - | - | -| |`id`| string, uuid | |`name`| string | |`data`| object | |`expireInSeconds`| number | How many seconds the job may stay active before being retried or failed | |`heartbeatSeconds`| number \| null | Heartbeat interval configured for this job, or null if not configured | |`signal`| AbortSignal | |`groupId`| string \| null | Group identifier from the `group` option in `send()`, or null if the job has no group | |`groupTier`| string \| null | Group tier from the `group` option in `send()`, or null if no tier was set | An example of a worker that checks for a job every 10 seconds. ```js await boss.work('email-welcome', { pollingIntervalSeconds: 10 }, ([ job ]) => myEmailService.sendWelcomeEmail(job.data)) ``` An example of a worker that returns a maximum of 5 jobs in a batch. ```js await boss.work('email-welcome', { batchSize: 5 }, (jobs) => myEmailService.sendWelcomeEmails(jobs.map(job => job.data))) ``` ### Low-latency dispatch with LISTEN/NOTIFY By default, workers fetch new jobs by polling on their `pollingIntervalSeconds`, so a freshly created job waits up to one interval before it is picked up. bun-boss can optionally use Postgres [`LISTEN/NOTIFY`](https://www.postgresql.org/docs/current/sql-notify.html) to wake workers the instant a job is created, cutting dispatch latency to milliseconds. This is an **opt-in optimization on top of polling, not a replacement for it.** Polling always keeps running as a safety net, so jobs are never lost if a notification is missed (for example during a brief connection drop). A notification is only ever a hint that tells a worker to fetch now instead of waiting — the normal locking fetch, queue policies, and concurrency limits are unchanged. **Enabling it requires two opt-ins:** 1. Start the instance with `useListenNotify: true` and a `db` adapter that implements `listen` — e.g. `fromPglite`. The built-in driver (Bun's SQL client) implements no LISTEN, so with it the listener cannot be established and bun-boss continues with polling. 2. Mark each queue that should emit notifications with the `notify: true` option on `createQueue()` (or `updateQueue()`). ```js import { PGlite } from '@electric-sql/pglite' import { BunBoss, fromPglite } from 'bun-boss' const pglite = new PGlite() const boss = new BunBoss({ backend: 'pglite', db: fromPglite(pglite), useListenNotify: true }) await boss.start() await boss.createQueue('email-welcome', { notify: true }) // No polling tuning needed — while NOTIFY is active the worker is woken the instant a // job is created and polls only as a slow backstop (notifyPollingIntervalSeconds, default 30s). await boss.work('email-welcome', ([ job ]) => myEmailService.sendWelcomeEmail(job.data) ) // This job is processed almost immediately rather than waiting for the next poll. await boss.send('email-welcome', { to: 'new@user.com' }) ``` **Notes and limitations:** - Only **immediately-available** jobs emit a notification. Future-dated jobs (`startAfter`, `sendAfter()`, throttling/debouncing) and jobs blocked by [flow](./jobs.md#flowjobs-options) dependencies are picked up by polling once they become eligible. - A NOTIFY is emitted transactionally with the insert, so it fires on commit. When you create jobs inside your own transaction via the `db` option, the notification commits atomically with your transaction. - The listener requires a `db` adapter that implements `listen` (e.g. `fromPglite`, or a custom adapter holding a session-pinned connection — **not** one through PgBouncer in transaction or statement pooling mode, which disables `LISTEN/NOTIFY`). The built-in driver implements no LISTEN. When a listener cannot be established, bun-boss emits a [`warning`](./events.md#warning) of type `listen_notify_unavailable` and continues polling only. The producer side still fires either way, so external listeners on other connections can act on it. - The notification channel is namespaced per `schema`, so multiple bun-boss instances (and other services) on the same database do not collide. ### `work(name, handler)` Simplified work() without an options argument ```js await boss.work('email-welcome', ([ job ]) => emailer.sendWelcomeEmail(job.data)) ``` work() with active job deletion ```js const queue = 'email-welcome' await boss.work(queue, async ([ job ]) => { await emailer.sendWelcomeEmail(job.data) await boss.deleteJob(queue, job.id) }) ``` work() with abort signal ```js await boss.work('process-video', async ([ job ]) => { const result = await fetch('https://api.example.com/process', { signal: job.signal }) }) ``` ### `getWipData(options)` Returns a snapshot of all workers in this instance of bun-boss with state `created`, `active`, or `stopping`. This is the same data payload emitted by the `wip` event, but available on-demand without waiting for a job transition. Use this for continuous monitoring of worker utilization — for example, driving metrics or autoscaling signals when jobs are long-running and the `wip` event may not fire frequently enough. **Arguments** - `options`: object *(optional)* **Options** * **includeInternal**, bool, *(default=false)* If true, includes workers for bun-boss internal queues (e.g., scheduling). **Returns**: `WipData[]` ```js // Poll worker utilization every 2 seconds for metrics setInterval(() => { const workers = boss.getWipData() const working = workers.filter(w => w.state === 'active' && w.count > 0).length const idle = workers.filter(w => w.state === 'active' && w.count === 0).length console.log(`working: ${working}, idle: ${idle}`) }, 2000) ``` ### `notifyWorker(id)` Notifies a worker by id to bypass the job polling interval (see `pollingIntervalSeconds`) for this iteration in the loop. ```js const workerId = await boss.work('email-welcome', { pollingIntervalSeconds: 60 }, handler) // a job was just created — tell the worker to fetch now instead of // waiting out the remainder of its polling interval await boss.send('email-welcome', { to: 'new@user.com' }) boss.notifyWorker(workerId) ``` ### `offWork(name, options)` Removes a worker by name or id and stops polling. **Arguments** - name: string - options: object **Options** * **wait**, boolean, *(default=true)* If the promise should wait until current jobs finish * **id**, string Only stop polling by worker id ```js const workerId = await boss.work('email-welcome', handler) // stop all workers for a queue, waiting for active jobs to finish await boss.offWork('email-welcome') // stop a single worker by id without waiting await boss.offWork('email-welcome', { id: workerId, wait: false }) ``` --- # File: docs/api/queues.md # Queues ### `createQueue(name, options?)` Creates a queue. ```js // a basic queue with the default (standard) policy await boss.createQueue('email-send') // a queue with retry and dead letter configuration // (the dead letter queue must exist before it can be referenced) await boss.createQueue('order-processing-dlq') await boss.createQueue('order-processing', { policy: 'singleton', retryLimit: 5, retryDelay: 60, retryBackoff: true, deadLetter: 'order-processing-dlq' }) ``` ```ts type Queue = { name: string; policy?: QueuePolicy; partition?: boolean; deadLetter?: string; warningQueueSize?: number; notify?: boolean; } & QueueOptions ``` Allowed policy values: | Policy | Description | | - | - | | `standard` | (Default) Supports all standard features such as deferral, priority, and throttling | | `short` | Only allows 1 job to be queued, unlimited active. Can be extended with `singletonKey` | | `singleton` | Only allows 1 job to be active, unlimited queued. Can be extended with `singletonKey` | | `stately` | Combination of short and singleton: Only allows 1 job per state, queued and/or active. Can be extended with `singletonKey` | | `exclusive` | Only allows 1 job to be queued or active. Can be extended with `singletonKey` | > [!WARNING] > `stately` queues are special in how retries are handled. By definition, stately queues will not allow multiple jobs to occupy `retry` state. Once a job exists in `retry`, failing another `active` job will bypass the retry mechanism and force the job to `failed`. If this job requires retries, consider a custom retry implementation using a dead letter queue. * **partition**, boolean, default false If set to true, a dedicated table will be created in the partition scheme. This would be more useful for large queues in order to keep it from being a "noisy neighbor". Postgres and PGlite only — backends without table partitioning (SQLite) accept the option but store it as `false` (see [Database backends](../database-backends.md)). * **deadLetter**, string When a job fails after all retries, if the queue has a `deadLetter` property, the job's payload will be copied into that queue, copying the same retention and retry configuration as the original job. The dead-lettered job also records where it came from via the `sourceName`, `sourceId`, `sourceCreatedOn`, and `sourceRetryCount` fields, which power [`redrive()`](./jobs.md#redrivename-options) for moving jobs back to their source queue. * **warningQueueSize**, int How many items can exist in the created or retry state before emitting a warning event. When left unset it is stored as `0`, which falls back to the instance-level `warningQueueSize`, or 10,000 if that is unset as well. * **notify**, boolean, default false When enabled, creating an immediately-available job on this queue emits a Postgres `NOTIFY` so workers wake right away instead of waiting for their next poll. This only has an effect when the instance is started with the [`useListenNotify`](./constructor.md#newoptions) option, which runs the listener. Jobs scheduled for the future (for example via `sendAfter()` or throttling/debouncing) do **not** emit a notification — they are picked up by polling when they mature. See [Workers › Low-latency dispatch with LISTEN/NOTIFY](./workers.md#low-latency-dispatch-with-listennotify). **Retry options** * **retryLimit**, int Default: 2. Number of retries to complete a job. * **retryDelay**, int Default: 0. Delay between retries of failed jobs, in seconds. * **retryBackoff**, bool Default: false. Enables exponential backoff retries based on retryDelay instead of a fixed delay. Sets initial retryDelay to 1 if not set. A simplified function to get the delay between runs is: `retryDelay * 2 ^ retryCount` with some jitter. The full function to determine the backoff delay is `Math.min(retryDelayMax, Math.max(retryDelay, 1) * (2 ** Math.min(16, retryCount + 1) / 2 + 2 ** Math.min(16, retryCount + 1) / 2 * Math.random()))` * **retryDelayMax**, int Default: no limit. Maximum delay between retries of failed jobs, in seconds. Can only be set when `retryBackoff` is `true` — setting it otherwise throws `retryDelayMax can only be set if retryBackoff is true`. **Heartbeat options** * **heartbeatSeconds**, int Default: none (disabled). Expected heartbeat interval in seconds. When set, workers using `work()` will automatically send periodic heartbeats. If no heartbeat is received within this interval, the monitor will fail/retry the job. Must be >= 10. Can be overridden per-job via `send()` options. #### Heartbeat vs expiration Heartbeat and expiration are two independent mechanisms that address different failure modes: - **Expiration** (`expireInSeconds`) is the maximum time a job is allowed to remain active. After this period, the job attempt is considered stale — regardless of whether the worker is alive or dead, the attempt has taken too long and is no longer relevant. Set this to the upper bound of how long the job should ever take. - **Heartbeat** (`heartbeatSeconds`) is a worker liveness check. The worker periodically signals "I'm still alive and working on this job." If the signal stops, it means the worker has died (crash, OOM, network partition, node shutdown) — but the job itself may still be perfectly valid and should be retried on another worker as soon as possible. | | Heartbeat | Expiration | | - | - | - | | **Purpose** | Detect dead workers quickly | Abandon stale job attempts | | **What it means** | The worker stopped responding — the job is still valid, retry it elsewhere | The job has been active too long — this attempt is no longer relevant | | **Failure scenario** | Worker crash, OOM kill, network partition, node shutdown | Infinite loop, deadlock, unresponsive external dependency, or simply exceeding the time budget | | **Detection speed** | Fast (seconds to minutes) | Matches expected job duration | | **Default** | Disabled | 15 minutes | Both mechanisms operate independently and can be used together. When a job fails via either mechanism, it follows the same retry logic (`retryLimit`, `retryDelay`, etc.). **When to use heartbeat:** Long-running jobs where the gap between "worker died" and "job expired" would be unacceptably large. For example, a 2-hour video processing job with `expireInSeconds: 7200` won't be detected as failed until 2 hours after it started, even if the worker crashed immediately. Adding `heartbeatSeconds: 60` means a dead worker is detected within a minute. **When expiration alone is sufficient:** Only when the expiration time is already short enough that waiting for it to trigger a retry is acceptable. In practice, `expireInSeconds` is set conservatively — well above the typical job duration — to account for slowdowns, rate limiting, and transient issues. The default is 15 minutes. This means even a quick task like sending an email could wait 15 minutes before a dead worker is detected via expiration. Heartbeat closes this gap by detecting the dead worker in seconds, regardless of how long the expiration is set. #### Recommended values Set `expireInSeconds` to the maximum time a job should ever take (accounting for worst-case conditions). Set `heartbeatSeconds` based on how quickly you need to detect a dead worker and retry. Actual detection time is `heartbeatSeconds` + up to `monitorIntervalSeconds` (default 60s), since the monitor must run to observe a stale heartbeat. There is no benefit to setting `heartbeatSeconds` below `monitorIntervalSeconds`. | Job type | `expireInSeconds` | `heartbeatSeconds` | Dead worker detected in | | - | - | - | - | | Quick tasks (email, notifications) | 900 (default) | 30-60 | ~1-2 min | | Medium tasks (report generation) | 900-1800 | 30-60 | ~1-2 min | | Long tasks (video processing, ML) | 7200 (2 hr) | 60-300 | ~2-6 min | | Very long tasks (data migration) | 86399 (~24 hr, the maximum) | 300-600 | ~6-11 min | **Expiration options** * **expireInSeconds**, number Default: 15 minutes. How many seconds a job may be in active state before being retried or failed. Must be >=1 and less than 24 hours (86400) **Retention options** * **retentionSeconds**, number Default: 14 days. How many seconds a job may be in created or retry state before it's deleted. Must be >=1 * **deleteAfterSeconds**, int Default: 7 days. How long a job should be retained in the database after it's completed. Set to 0 to never delete completed jobs. * All retry, expiration, and retention options set on the queue will be inherited for each job, unless they are overridden. ### `updateQueue(name, options)` Updates options on an existing queue, with the exception of the `policy` and `partition` settings, which cannot be changed — passing either throws `queue policy cannot be changed after creation` / `queue partitioning cannot be changed after creation`. At least one property must be supplied; an empty options object throws `no properties found to update`. ```js await boss.updateQueue('email-send', { retryLimit: 5, retryDelay: 120 }) ``` ### `deleteQueue(name)` Deletes a queue and all jobs. ```js await boss.deleteQueue('email-send') ``` ### `getQueues(names?)` Returns all queues, or only the named queues when an array of names is provided. The count fields on the result (`queuedCount`, `readyCount`, `activeCount`, ...) are the cached values maintained by the monitor, so they can be stale — they stay at zero until the monitor first runs. Use [`getQueueStats(name, { force: true })`](#getqueuestatsname-options) for a fresh count. ```js const queues = await boss.getQueues() for (const queue of queues) { console.log(`${queue.name}: ${queue.queuedCount} queued, ${queue.activeCount} active`) } ``` ### `getQueue(name)` Returns a queue by name, or `null` if it doesn't exist. ```js const queue = await boss.getQueue('email-send') if (!queue) { await boss.createQueue('email-send') } ``` ### `getQueueStats(name, options?)` Returns the current queue-depth counts as a single-element array (newest first). The one snapshot has the queue `name`, a `capturedOn` timestamp, and these counts: * `queuedCount` — jobs waiting to run, **including** deferred (future-dated) jobs; this drives the queue backlog warning, so dumping a lot of deferred work still trips it * `deferredCount` — jobs scheduled to start in the future (`startAfter` not yet reached) * `readyCount` — jobs ready to be processed now (`queuedCount - deferredCount`); the true runnable backlog * `activeCount` — jobs currently being processed * `failedCount` — failed jobs still retained in the table (bounded by the queue's retention policy, so this is a rolling count of recent failures rather than an all-time total) * `totalCount` — all jobs currently stored for the queue By default the counts are served from the cache on the queue table (refreshed every `monitorIntervalSeconds`), so with the monitor running the value is at most one monitor interval stale. If monitoring is disabled or falling behind, the cached value is served until it is over an hour old, at which point it is recomputed. Pass `{ force: true }` to re-count directly from the job table and update the values on the queue table, but even this option is rate-limited to once a minute, so repeated calls using `force` don't always re-aggregate. ```js const [stats] = await boss.getQueueStats('email-send') console.log(`${stats.readyCount} jobs ready, ${stats.activeCount} active`) // force a fresh count from the job table const [fresh] = await boss.getQueueStats('email-send', { force: true }) ``` --- # File: docs/api/scheduling.md # Scheduling Jobs may be created automatically based on a cron expression. As with other cron-based systems, at least one instance needs to be running for scheduling to work. In order to reduce the amount of evaluations, schedules are checked every 30 seconds, which means the 6-placeholder format should be discouraged in favor of the minute-level precision 5-placeholder format. For example, use this format, which implies "any second during 3:30 am every day" ``` 30 3 * * * ``` but **not** this format which is parsed as "only run exactly at 3:30:30 am every day" ``` 30 30 3 * * * ``` To change how often schedules are checked, you can set the constructor option `cronMonitorIntervalSeconds` (default `30`, range 1-45). To change how often cron jobs are run, you can set the constructor option `cronWorkerIntervalSeconds` (default `5`, range 1-45). In order mitigate clock skew and drift, every 10 minutes the clocks of each instance are compared to the database server's clock. The skew, if any, is stored and used as an offset during cron evaluation to ensure all instances are synchronized. Internally, job throttling options are then used to make sure only 1 job is sent even if multiple instances are running. If needed, the default clock monitoring interval can be adjusted using the constructor option `clockMonitorIntervalSeconds` (default `600`, range 1-600). Additionally, to disable scheduling on an instance completely, use the following in the constructor options. ```js { schedule: false } ``` For more cron documentation and examples see the docs for the [croner package](https://www.npmjs.com/package/croner). ### `schedule(name, cron, data, options)` Schedules a job to be sent to the specified queue based on a cron expression. If the schedule already exists, it's updated to the new cron expression. The queue must already exist; `schedule()` throws `Queue not found` otherwise. **Arguments** - `name`: string, *required* - `cron`: string, *required* - `data`: object - `options`: object `options` supports all properties in `send()` as well as the following additional options. * **tz** An optional time zone name. If not specified, the default is UTC. * **key** An optional unique key when more than one schedule is needed for this queue. Defaults to an empty string. May only contain alphanumeric characters, underscores, hyphens, periods, or forward slashes. For example, the following code will send a job at 3:00am in the US central time zone into the queue `notification-abc`. ```js await boss.schedule('notification-abc', `0 3 * * *`, null, { tz: 'America/Chicago' }) ``` ### `unschedule(name)` Removes the schedule with the default (empty) key for the specified queue name. Schedules created with a `key` are not affected — remove those with [`unschedule(name, key)`](#unschedulename-key). ```js await boss.unschedule('notification-abc') ``` ### `unschedule(name, key)` Removes a schedule by queue name and unique key. ```js // create two schedules on the same queue, then remove just one await boss.schedule('report', '0 6 * * *', { region: 'us' }, { key: 'us' }) await boss.schedule('report', '0 18 * * *', { region: 'eu' }, { key: 'eu' }) await boss.unschedule('report', 'eu') ``` ### `getSchedules()` Returns all scheduled jobs. ```js const schedules = await boss.getSchedules() for (const schedule of schedules) { console.log(`${schedule.name} (${schedule.key}): ${schedule.cron} ${schedule.timezone}`) } ``` ### `getSchedules(name)` Returns all scheduled jobs by queue name. ```js const schedules = await boss.getSchedules('report') ``` ### `getSchedules(name, key)` Returns the schedule for the given queue name and unique key, as a single-element array (empty if none). Since `(name, key)` is the primary key of the schedule table, at most one row is ever returned. ```js const [schedule] = await boss.getSchedules('report', 'eu') ``` --- # File: docs/api/ops.md # Operations ### `start()` Returns the same BunBoss instance used during invocation Prepares the target database and begins job monitoring. ```js await boss.start() await boss.createQueue('hey-there') await boss.send('hey-there', { msg:'this came for you' }) ``` If the required database objects do not exist in the specified database, **`start()` will automatically create them** at the current schema version (with `migrate: false`, `start()` verifies the existing schema instead and throws if it is missing or the wrong version). There is no in-place upgrade from an older installed schema version: if bun-boss finds a schema older than the version this release ships, `start()` throws rather than migrating it in place. On Postgres and PGlite, schema installation is nested within an advisory lock to prevent race conditions during `start()`. Internally, this lock is created using `pg_advisory_xact_lock()` which auto-unlocks at the end of the transaction and doesn't require a persistent session or the need to issue an unlock. The SQLite backend has no advisory locks (do not run multiple bun-boss processes against one database file). One example of how this is useful would be including `start()` inside the bootstrapping of a pod in a ReplicaSet in Kubernetes. Being able to scale up your job processing using a container orchestration tool like k8s is becoming more and more popular, and bun-boss can be dropped into this system without any special startup handling. ### `stop(options)` Stops all background processing, such as maintenance and scheduling, as well as all polling workers started with `work()`. By default, calling `stop()` without any arguments will gracefully wait for all workers to finish processing active jobs before resolving. Emits a `stopped` event if needed. **Arguments** * `options`: object * `graceful`, bool Default: `true`. If `true`, the BunBoss instance will wait for any workers that are currently processing jobs to finish, up to the specified timeout. During this period, new jobs will not be processed, but active jobs will be allowed to finish. * `close`, bool Default: `true`. If the database connection is managed by bun-boss, it will close the connection pool. Use `false` if needed to continue allowing operations such as `send()` and `fetch()`. * `timeout`, int Default: 30000. Maximum time (in milliseconds) to wait for workers to finish job processing before shutting down the BunBoss instance. Values below 1000 are raised to 1000. > [!WARNING] > This option is ignored when `graceful` is set to `false`. ```js // graceful shutdown: wait for active jobs to finish (up to the timeout) await boss.stop() // stop workers but keep the connection pool open for send() and fetch() await boss.stop({ close: false }) // shut down immediately without waiting for active jobs await boss.stop({ graceful: false }) ``` ### `isInstalled()` Utility function to see if bun-boss is installed in the configured database. ```js const installed = await boss.isInstalled() // true ``` ### `schemaVersion()` Utility function to get the database schema version. ```js const version = await boss.schemaVersion() // 1 ``` ### `supervise(name)` Forces an immediate maintenance pass instead of waiting for the next background cycle: it monitors backlog, fails timed-out and heartbeat-stale jobs, deletes jobs past their retention window, and cleans up orphaned job dependencies. Pass a queue name to supervise a single queue, or omit it for all. Useful for deterministic tests, or when you have disabled `supervise` and drive maintenance yourself. ```js await boss.supervise() ``` ### `resolveFlow()` Forces an immediate [flow](./jobs.md#flowjobs-options)-resolution pass, unblocking dependents of any parents that have completed, instead of waiting for the next background cycle. See [`resolveFlow()`](./jobs.md#resolveflow) for details. ### `isMaintaining()` Returns `true` while the background maintenance pass is running. Use it to avoid launching a manual `supervise()` on top of the background one. A manual `supervise()` does not set this flag. ```js const busy = boss.isMaintaining() // false ``` ### `isResolvingFlow()` Returns `true` while a flow-resolution pass is running — including one started by a manual `resolveFlow()`. ### `isCheckingSkew()` Returns `true` while the scheduler's clock-skew check is running. ### `getWipData(options)` Returns the current worker work-in-progress snapshot — the same payload carried by the [`wip`](./events.md#wip) event. See [`getWipData()`](./workers.md#getwipdataoptions) on the Workers page. ### `getDb()` Returns the `Db` instance (the `IDatabase` interface, exported as `Db`) bun-boss is using — the built-in Bun `SQL` driver, or the adapter you passed as the `db` option. Use it to run your own SQL over the same connection via `executeSql(text, values)` instead of opening a second pool. ```js const db = boss.getDb() const { rows } = await db.executeSql('select now()', []) ``` bun-boss owns the built-in driver's lifecycle; a `db` you supplied stays yours to open and close. --- # File: docs/api/events.md # Events Each bun-boss instance is an EventEmitter, and contains the following events. ## `error` The `error` event could be raised during internal processing, such as scheduling and maintenance. Adding a listener to the error event is strongly encouraged because of the default behavior of the `EventEmitter`. Database connection failures are not reported here: the built-in driver (Bun's SQL client) exposes no background-error hook, so a broken connection surfaces as a rejection on the operation that encountered it — typically re-emitted as an `error` by the background component that issued the query. An `error` event with no listener is thrown as an unhandled error. Under Bun it is typically reported on stderr from the async worker loop rather than exiting the process, so an unlistened failure can go unnoticed. Register a listener regardless. Ideally, code similar to the following example would be used after creating your instance, but before `start()` is called. ```js boss.on('error', error => logger.error(error)); ``` The payload is always an `Error` instance. Errors raised while running a worker additionally carry `queue` (the queue name) and `worker` (the worker id) properties, so a handler can attribute the failure: ```js boss.on('error', error => logger.error({ queue: error.queue, worker: error.worker }, error)); ``` ## `warning` During monitoring and maintenance, bun-boss may raise warning events. The payload contains `message` and `data` properties with details about the warning. ```js boss.on('warning', ({ message, data }) => { console.log('bun-boss warning:', message, data); }); ``` ### Warning Types | Type | Description | Data Properties | |------|-------------|-----------------| | `slow_query` | A maintenance query exceeded the slow query threshold | `elapsed` (seconds), `sql`, `values` | | `queue_backlog` | A queue has exceeded its warning threshold | `name`, `queuedCount`, `warningQueueSize` | | `clock_skew` | Database clock is out of sync with application server | `seconds`, `direction` | | `listen_notify_unavailable` | `useListenNotify` is enabled but a `LISTEN/NOTIFY` listener could not be established (an unsupported backend, a `db` adapter without `listen`, or a failed subscribe such as PgBouncer transaction pooling); bun-boss continues with polling only | `type`, and `backend` or `error` depending on the cause | Only `listen_notify_unavailable` carries its type in `data.type`; the other warnings are identified by their `message` text. ## `wip` Emitted at most once every 2 seconds whenever at least one worker has an active job. The payload is an array with one entry per active worker in this instance of bun-boss. Workers that have already stopped, and bun-boss's own internal maintenance workers, are excluded. ```ts const workers = [ { id: 'fc738fb0-1de5-4947-b138-40d6a790749e', workId: 'fc738fb0-1de5-4947-b138-40d6a790749e', name: 'my-queue', options: { pollingInterval: 2000, notifyPollingInterval: 30000 }, state: 'active', count: 1, createdOn: 1620149137015, lastFetchedOn: 1620149137015, lastJobStartedOn: 1620149137015, lastJobEndedOn: null, lastJobDuration: 343, lastError: null, lastErrorOn: null } ] ``` `workId` is the value returned by `work()`. The first worker of a `work()` call uses that same value as its own `id`, so the two match in the single-worker example above. When using `localConcurrency`, multiple worker entries in the array will share the same `workId`, allowing you to correlate them back to a specific `work()` call. ```js const workId = await boss.work('my-queue', { localConcurrency: 5 }, handler) boss.on('wip', workers => { const myWorkers = workers.filter(w => w.workId === workId) const working = myWorkers.filter(w => w.count > 0).length const idle = myWorkers.length - working console.log(`working: ${working}/${myWorkers.length}, idle: ${idle}`) }) ``` ## `stopped` Emitted after `stop()` once all workers have completed their work and maintenance has been shut down. ## `flow` Emitted by the background flow resolver each time it resolves one or more completed blocking parent jobs, unblocking their dependents (created via [`flow()`](./jobs.md#flowjobs-options)). The `resolved` count is the number of parent jobs cleared in that batch, not the number of dependents that became runnable. See `flowIntervalSeconds` in the [constructor options](./constructor.md) for how often the resolver runs. ```js boss.on('flow', event => { console.log(`Resolved ${event.resolved} blocking parent job(s) in ${event.table}`) }) ``` The event payload contains: ```ts const event = { table: 'job_common', // partition table whose blocking parents were resolved resolved: 1 // completed blocking parent jobs resolved in this batch (their dependents were unblocked) } ``` --- # File: docs/api/adapters.md # Database Adapters bun-boss operations such as `send()`, `insert()`, `fetch()`, and `complete()` accept a `db` option that lets you run them inside an existing database transaction. This is how you ensure that job creation (or completion) is atomic with your application's own writes — if the transaction rolls back, so does the job. Each adapter wraps a driver's connection, instance, or transaction object as a bun-boss `Db` (the `executeSql` interface, plus the optional `withTransaction` and `listen` capabilities), so bun-boss can execute its own SQL within your transaction. ```ts interface Db { executeSql(text: string, values?: unknown[]): Promise<{ rows: any[] }>; // Optional. When present, bun-boss runs its multi-statement operations (upsert, the split // complete/fail/expire, flow resolution) inside a real transaction; otherwise they run // sequentially without atomicity. withTransaction?(fn: (tx: Db) => Promise): Promise; // Optional. Enables LISTEN/NOTIFY (`useListenNotify`); only `fromPglite` implements it. listen?(channel: string, onNotification: (payload: string) => void, onReconnect: () => void): Promise; } ``` Only `executeSql` is required. An adapter that omits `withTransaction` still works, but bun-boss falls back to running those multi-statement operations without a transaction — and flow resolution loses atomicity, so a flow that fails part way can leave dependencies partly written (see [Database Backends](../database-backends.md#what-is-different-from-the-postgres-backends)). When a flow fails inside a caller-owned transaction (`{ db }`), the transaction itself stays usable — roll it back rather than committing after a caught flow error. What the shipped adapters implement: | adapter | `executeSql` | `withTransaction` | `listen` | |---|---|---|---| | `fromBunSql` | ✅ | ❌ | ❌ | | `fromBunSqlite` | ✅ | ✅ | ❌ | | `fromPglite` | ✅ | ✅ | ✅ | Write `$1, $2` placeholders in your own SQL on every backend; the adapter translates them to whatever the driver expects. bun-boss ships with `fromBunSql` for Bun's built-in `SQL` client against PostgreSQL, `fromPglite` for embedded PGlite, and `fromBunSqlite` for embedded SQLite through Bun's `SQL` client (see [Database Backends](../database-backends.md)). ## Bun Bun's built-in [`SQL`](https://bun.com/docs/api/sql) client is a driver rather than an ORM, so `fromBunSql` covers both uses: it can back a whole bun-boss instance with a client you own (the built-in driver wraps its own client with this same adapter), and it can scope a single operation to a `sql.begin()` transaction. Bun hands out the same shape for a pool and for a transaction, so one function serves both. ```ts import { SQL } from 'bun' import { BunBoss, fromBunSql } from 'bun-boss' const sql = new SQL('postgres://user:pass@localhost:5432/mydb') // back bun-boss with a client your application already owns const boss = new BunBoss({ db: fromBunSql(sql) }) await boss.start() // or create a job inside your own transaction await sql.begin(async (tx) => { await tx`INSERT INTO orders (item, qty) VALUES (${'widget'}, ${1})` await boss.send('order-processing', { item: 'widget' }, { db: fromBunSql(tx) }) }) ``` Bun talks to real PostgreSQL, so leave `backend` at its default `postgres` — no compatibility flags apply. As with every adapter, the `SQL` client's lifecycle is yours: bun-boss never opens or closes it. `fromBunSql` implements `executeSql` only — it has no `withTransaction`. Backing a whole instance with it therefore runs bun-boss's own multi-statement operations (upsert, the split complete/fail/expire, flow resolution) without a transaction, unlike the built-in driver, which does supply `withTransaction`. Use it for the per-operation `{ db }` case, or accept that trade-off. See [Bun.SQL](../database-backends.md#bunsql-the-built-in-driver) for the driver-level details — LISTEN/NOTIFY, prepared statements, and multi-statement blocks. ## PGlite `fromPglite` adapts a PGlite instance. Like SQLite it backs a whole bun-boss instance (pair it with `backend: 'pglite'`), and bun-boss's tables live in the same in-process database as your application's. It is the only shipped adapter that implements `listen`, so it is the only one that can serve `useListenNotify`. ```ts import { PGlite } from '@electric-sql/pglite' import { BunBoss, fromPglite } from 'bun-boss' const pglite = new PGlite('./my-app-data') const db = fromPglite(pglite) const boss = new BunBoss({ backend: 'pglite', db }) await boss.start() ``` Sharing the database does not by itself make a job atomic with your writes — a plain `send()` outside a transaction commits on its own. To get atomicity, open the transaction through the adapter's `withTransaction` and pass its handle as the operation's `db`: ```ts await db.withTransaction!(async (tx) => { await tx.executeSql('INSERT INTO orders (item, qty) VALUES ($1, $2)', ['widget', 1]) await boss.send('order-processing', { item: 'widget' }, { db: tx }) }) ``` `fromPglite` is typed as the base `Db`, on which `withTransaction` is optional, so TypeScript needs the assertion (or a local check) before the call; `fromBunSqlite` narrows its return type instead, which is why the SQLite sample below needs neither. See [PGlite](../database-backends.md#pglite-embedded) for the instance-level details and limitations. ## SQLite (Bun) `fromBunSqlite` adapts Bun's `SQL` client opened on a `sqlite://` URL. It always backs a whole bun-boss instance (pair it with `backend: 'sqlite'`), and bun-boss's tables live in the same database file as your application's. ```ts import { SQL } from 'bun' import { BunBoss, fromBunSqlite } from 'bun-boss' const sql = new SQL('sqlite://app.db') const db = fromBunSqlite(sql) const boss = new BunBoss({ backend: 'sqlite', db }) await boss.start() ``` Sharing the database file does not by itself make a job atomic with your writes — a plain `send()` outside a transaction commits on its own. To get atomicity, open the transaction through the adapter's `withTransaction` and pass its handle as the operation's `db`: ```ts await db.withTransaction(async (tx) => { await tx.executeSql('INSERT INTO orders (item, qty) VALUES ($1, $2)', ['widget', 1]) await boss.send('order-processing', { item: 'widget' }, { db: tx }) }) ``` Always use `withTransaction` rather than issuing `BEGIN` yourself on the shared `SQL` instance: the adapter serializes its own statements on the single logical connection, but it cannot see a transaction you open directly, and bun-boss's background writes would interleave into it. See [SQLite](../database-backends.md#sqlite-embedded-via-bunsql) for the dialect-level details and limitations. ## Rollback behaviour When the transaction is rolled back (either explicitly or by throwing an error), all bun-boss operations executed through the adapter are rolled back as well. This is the primary reason to use an adapter — to guarantee atomicity between your application writes and job scheduling. --- # File: docs/api/testing.md # Testing bun-boss includes built-in spy support to help write fast, deterministic tests without polling or arbitrary delays. ## Enabling Spies Spies must be explicitly enabled via the `__test__enableSpies` constructor option. This ensures zero overhead in production. ```js const boss = new BunBoss({ url: 'postgres://...', __test__enableSpies: true }) ``` As everywhere else in bun-boss, the queue must already exist — call `await boss.createQueue(name)` before sending or working, since queues are never created implicitly. > [!WARNING] > Calling `getSpy()` without enabling spies will throw an error. ## `getSpy(name)` Returns a spy instance for the specified queue. The spy tracks all job state transitions (created, active, completed, failed) for that queue. Transitions are recorded from the moment spies are enabled, not from the first `getSpy()` call — fetching the spy after a job has already settled still resolves waits for the states it passed through. **Arguments** - `name`: string, queue name **Returns** A spy object with the following interface: ```ts interface JobSpyInterface { clear(): void waitForJob(selector: (data: T) => boolean, state: JobSpyState): Promise> waitForJobWithId(id: string, state: JobSpyState): Promise> } type JobSpyState = 'created' | 'active' | 'completed' | 'failed' interface SpyJob { id: string name: string data: T state: JobSpyState output?: object } ``` Supply the type argument to have `job.data` typed instead of `object`: ```ts const spy = boss.getSpy<{ userId: string }>('my-queue') ``` ### `spy.waitForJob(selector, state)` Waits for a job matching the selector function to reach the specified state. If a job matching the selector criteria was already processed before this method was called, the promise will resolve immediately. **Arguments** - `selector`: function(data) => boolean, filters jobs by their data payload - `state`: string, one of 'created', 'active', 'completed', 'failed' ```js const boss = new BunBoss({ url: process.env.DATABASE_URL, __test__enableSpies: true }) await boss.start() const spy = boss.getSpy('my-queue') // Wait for any job with userId '123' to complete const job = await spy.waitForJob( (data) => data.userId === '123', 'completed' ) console.log(job.output) // handler result ``` ### `spy.waitForJobWithId(id, state)` Waits for a specific job by id to reach the specified state. Like `waitForJob()`, if the job already reached the specified state before this method was called, the promise will resolve immediately. **Arguments** - `id`: string, job id - `state`: string, one of 'created', 'active', 'completed', 'failed' ```js const spy = boss.getSpy('my-queue') const jobId = await boss.send('my-queue', { userId: '123' }) // Wait for this specific job to complete const job = await spy.waitForJobWithId(jobId, 'completed') ``` ### `spy.clear()` Clears all tracked job data from the spy. Useful for resetting state between tests. Any `waitForJob()`/`waitForJobWithId()` promise still pending when `clear()` is called is dropped and will never settle — only call it between tests, never while a wait is outstanding. ```js const spy = boss.getSpy('my-queue') afterEach(() => { spy.clear() }) ``` ## `clearSpies()` Clears all spies and their tracked data across all queues. It also **removes** the spy objects, so a handle obtained before the call is permanently detached and never sees another transition — call `getSpy()` again after clearing rather than reusing a handle held in `beforeAll`. ```js afterEach(() => { boss.clearSpies() }) ``` ## Example Test ```js import { describe, test, expect, beforeAll, afterAll, afterEach } from 'bun:test' import { BunBoss } from 'bun-boss' describe('email notifications', () => { let boss let workerId beforeAll(async () => { boss = new BunBoss({ url: process.env.DATABASE_URL, __test__enableSpies: true }) await boss.start() await boss.createQueue('email-welcome') }) afterAll(async () => { await boss.stop() }) // Stop the test's worker, otherwise it keeps polling and can steal the next test's job afterEach(async () => { if (workerId) await boss.offWork('email-welcome', { id: workerId, wait: true }) workerId = undefined boss.clearSpies() }) test('should send welcome email when user signs up', async () => { const spy = boss.getSpy('email-welcome') // Start the worker workerId = await boss.work('email-welcome', async ([job]) => { await sendEmail(job.data.email, 'Welcome!') return { sent: true } }) // Trigger the action that creates the job await userService.signUp({ email: 'test@example.com' }) // Wait for job to complete - no polling needed const job = await spy.waitForJob( (data) => data.email === 'test@example.com', 'completed' ) expect(job.output).toEqual({ sent: true }) }) test('should handle email failures', async () => { const spy = boss.getSpy('email-welcome') workerId = await boss.work('email-welcome', async () => { throw new Error('SMTP connection failed') }) const jobId = await boss.send('email-welcome', { email: 'test@example.com' }) const job = await spy.waitForJobWithId(jobId, 'failed') expect(job.output.message).toBe('SMTP connection failed') }) }) ``` ## Race Condition Safety The spy is designed to handle race conditions gracefully. You can call `waitForJob()` or `waitForJobWithId()` before or after the job reaches the desired state: ```js const spy = boss.getSpy('my-queue') // This works even if job completes before waitForJob is called const waitPromise = spy.waitForJob((data) => data.id === '123', 'completed') await boss.send('my-queue', { id: '123' }) await boss.work('my-queue', async () => {}) const job = await waitPromise // Resolves correctly ``` ## Tracked States | State | When Tracked | | - | - | | `created` | Job inserted via `send()` or `insert()` | | `active` | Job fetched by a worker and handler started | | `completed` | A worker's handler returned successfully, or `complete()` was called from inside a handler | | `failed` | Handler threw an error and the job's retries were exhausted | A handler that calls `fail()` and then returns normally is recorded as `failed`, not `completed`. Spies observe the worker path only. A job settled outside a handler — `fetch()` followed by `complete()` — reaches `completed` in the database, but the spy never records it and a wait on that state will hang. `retry` and `cancelled` are **not** tracked. `waitForJob()` with an untracked state never resolves and never rejects — TypeScript rejects it via `JobSpyState`, but plain JS will hang. --- # File: docs/api/utils.md # Utility functions The following function is exported from the package and is not required during normal operations, but is intended to assist in schema creation if run-time privileges do not allow schema changes. ```js import { getConstructionPlans } from 'bun-boss' ``` ### `getConstructionPlans(schema?, options?)` **Arguments** - `schema`: string, database schema/namespace name (optional; defaults to `'pgboss'`, or `'bunboss'` in prefix mode) - `options`: object (optional) - `tableIsolation`: `'schema' | 'prefix'` — match the [`tableIsolation`](../database-backends.md#table-isolation) you construct `BunBoss` with. In `'prefix'` mode the DDL creates quoted `"schema.table"` objects in the default schema, skips `CREATE SCHEMA`, and omits partitioning. Returns the SQL commands required for manual creation of the required schema. ```js import fs from 'node:fs' const sql = getConstructionPlans('pgboss') // hand the DDL to a migration tool or a privileged operator fs.writeFileSync('create-bunboss.sql', sql) // prefix mode, into the default schema: const prefixed = getConstructionPlans('bunboss', { tableIsolation: 'prefix' }) ``` ### Constants The package also exports frozen constants so you can reference states, policies, and event names without hardcoding strings. ```js import { states, policies, events } from 'bun-boss' states.completed // 'completed' policies.singleton // a queue policy accepted by createQueue() events.error // 'error' ``` `states` mirrors the job states, `policies` the queue policies accepted by `createQueue()`, and `events` the event names emitted by a `BunBoss` instance. --- # File: docs/sql/job-table.md # Job table If you need to interact with bun-boss outside of Bun, such as other clients or even using triggers within PostgreSQL itself, most functionality is supported when working directly against the internal tables. For example, if you wanted to bulk load jobs and skip calling `send()` or `insert()`, you could use SQL `INSERT` or `COPY` commands. Writing rows directly bypasses the JavaScript layer, which is where `send()` and `insert()` do their defaulting and validation, so the preconditions below become yours to satisfy. The following is the primary job table on the Postgres and PGlite backends. The SQLite backend installs an equivalent table in its own dialect (text timestamps, text ids and json, integer booleans, a CHECK-constrained `state` column instead of an enum, and no partitioning). For manual job creation, the only required column is `name`. All other columns are nullable or have defaults. The queue itself must already exist — `name` is a foreign key to `pgboss.queue`, so create the queue with `createQueue()` or `pgboss.create_queue()` first, otherwise the insert fails with `23503`. Those defaults are the *table's*, not the *queue's*. `send()` and `insert()` copy `policy`, `retry_limit`, `retry_delay`, `expire_seconds` and `deletion_seconds` from the queue row and compute `keep_until` from its retention window; a hand-written `INSERT` silently gets `policy = NULL` and the hardcoded table defaults instead. Set those columns explicitly if the queue is not using the defaults — especially `policy`, since a NULL there makes the row invisible to every policy-enforcing partial index. The `state` column is an enum whose declaration order is significant: several internal queries compare states with `<` and `>` (`state < 'active'` means queued, for example). ```sql CREATE TYPE pgboss.job_state AS ENUM ( 'created', 'retry', 'active', 'completed', 'cancelled', 'failed' ) ``` This is reference DDL — bun-boss creates the table itself during `start()`, so there is no need to run it yourself. ```sql CREATE TABLE pgboss.job ( id uuid not null default gen_random_uuid(), name text not null, priority integer not null default(0), data jsonb, state pgboss.job_state not null default('created'), retry_limit integer not null default(2), retry_count integer not null default(0), retry_delay integer not null default(0), retry_backoff boolean not null default false, retry_delay_max integer, expire_seconds integer not null default (900), deletion_seconds integer not null default (60 * 60 * 24 * 7), singleton_key text, singleton_on timestamp without time zone, group_id text, group_tier text, start_after timestamp with time zone not null default now(), created_on timestamp with time zone not null default now(), started_on timestamp with time zone, completed_on timestamp with time zone, keep_until timestamp with time zone NOT NULL default now() + interval '14 days', output jsonb, dead_letter text, policy text, heartbeat_on timestamp with time zone, heartbeat_seconds int, blocked boolean not null default false, blocking boolean not null default false, pending_dependencies int not null default 0, source_name text, source_id uuid, source_created_on timestamp with time zone, source_retry_count int, CONSTRAINT job_pkey PRIMARY KEY (name, id) ) PARTITION BY LIST (name) ``` ### Constraints and indexes The parent `job` table carries only `job_pkey`; the rest is installed on each partition. Every partition gets `q_fkey` (`name` referencing `pgboss.queue`) and `dlq_fkey` (`dead_letter` referencing `pgboss.queue`), both `ON DELETE RESTRICT DEFERRABLE INITIALLY DEFERRED`, plus eight partial indexes named after the partition (`job_common_i1` through `job_common_i7` and `job_common_i9` on the default partition) that back the queue policies, throttling, fetch, group concurrency, and flow resolution. --- # File: docs/sql/queue-functions.md # Queue functions Queues can be created or deleted from SQL functions. These functions exist on the Postgres and PGlite backends only — SQLite has no stored functions, so `createQueue()` and `deleteQueue()` issue plain statements there instead. Calling them directly bypasses the JavaScript layer, which is where the defaulting, validation, and queue-cache bookkeeping live. In particular, a running instance is not notified: a queue deleted with `pgboss.delete_queue()` may stay in that instance's queue cache until it restarts. A queue created in SQL is picked up, because a cache miss falls through to the database. ### `pgboss.create_queue(queue_name text, options jsonb)` Options are the same as in [`createQueue()`](../api/queues.md#createqueuename-options), with one difference: `policy` has no default here and must always be supplied. `createQueue()` defaults it to `standard` in JavaScript, so omitting it in SQL raises a not-null violation (`23502`) on `queue.policy`. The recognized option keys are `policy`, `retryLimit`, `retryDelay`, `retryBackoff`, `retryDelayMax`, `expireInSeconds`, `retentionSeconds`, `deleteAfterSeconds`, `warningQueueSize`, `deadLetter`, `partition`, `heartbeatSeconds`, and `notify`. The queue name is the separate first argument. Unlike `createQueue()`, this function performs no validation: an unrecognized `policy` is stored as-is, unknown option keys are silently ignored, and a `deadLetter` naming a queue that does not exist fails with a raw foreign key error (`23503`). ### `pgboss.delete_queue(queue_name text)` Deletes a queue, all its jobs, and any schedules attached to it. If the queue was created with `partition: true`, its dedicated partition table is dropped rather than emptied. The queue must exist. Unlike `deleteQueue()`, which is a deliberate no-op for an unknown queue, calling this on a missing queue raises `null values cannot be formatted as an SQL identifier` (`22004`). ### `pgboss.job_table_run(command text, tbl_name text, queue_name text)` Applies a DDL command across the job tables. Both `tbl_name` and `queue_name` default to `NULL`: pass `queue_name` to target that queue's table, `tbl_name` to target a table by name, or neither to apply the command to the common job table and to every queue created with `partition: true`. Write the command against `pgboss.job` and the bare index names (`job_i1`, `job_i2`, and so on); `pgboss.job_table_format(command text, table_name text)` is the helper that rewrites those identifiers for each target table.