pg-boss

Jobs ​

Creating jobs ​

send() ​

Creates a new job and returns the job id.

NOTE

send() may resolve a null instead of a job id when opting into features such as unique jobs, throttling, or when using non-standard queue policies.

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

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, retryDelay * (2 ** Math.Min(16, retryCount) / 2 + 2 ** Math.Min(16, retryCount) / 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 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 and <= 86400 (24 hours)

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 inheritied for each job, unless they are overridden.

Connection options

  • db, object

    Instead of using pg-boss's default adapter, you can use your own, as long as it implements the following interface (the same as the pg module).

    ts
    interface Db {
  executeSql(text: string, values: any[]): Promise<{ rows: any[] }>;
}

    pg-boss ships with built-in adapters for popular ORMs. See ORM Transaction Adapters for details.

Deferred jobs

  • startAfter int, string, or Date

    • int: seconds to delay starting the job

    • string: Start after an ISO 8601 date time string, or after a relative interval

    • Date: Start after a Date object

      Default: 0

    A string that begins with an ISO 8601 calendar date (YYYY-MM-DD) is read as a date time. If it carries an explicit zone designator (Z or an offset such as +05:30) it resolves to that exact instant. If it carries none, it is interpreted as UTC, so the instant does not depend on the database's time zone.

    StringResolves to
    '2027-01-01T08:00:00Z'2027-01-01 08:00 UTC
    '2027-01-01T08:00:00.123Z'2027-01-01 08:00:00.123 UTC
    '2027-01-01T08:00:00+00:00'2027-01-01 08:00 UTC
    '2027-01-01T08:00:00+0000'2027-01-01 08:00 UTC
    '2027-01-01T08:00:00+00'2027-01-01 08:00 UTC
    '2027-01-01T13:30:00+05:30'2027-01-01 08:00 UTC
    '2027-01-01T00:00:00-08:00'2027-01-01 08:00 UTC
    '20270101T080000Z' (basic format)2027-01-01 08:00 UTC
    '2027-01-01T08:00:00'2027-01-01 08:00 UTC
    '2027-01-01T08:00'2027-01-01 08:00 UTC
    '2027-01-01 08:00:00'2027-01-01 08:00 UTC
    '2027-01-01'2027-01-01 00:00 UTC

    A string naming a time zone instead of an offset ('2027-01-01 08:00:00 America/New_York') is left for the database to resolve, and observes that zone's rules including daylight saving.

    Any other string is read as a relative delay from now, using Postgres interval syntax, so '5 minutes', '1 hour', 'PT1H' and '90' (bare seconds) are all valid. A string that begins with a calendar date but is not a valid date time ('2027-13-45') is an error rather than a delay.

    Note that the ISO 8601 year-month and ordinal date formats are not recognized as date times. They are valid Postgres intervals, so they are read as very long delays rather than rejected:

    StringRead as
    '2027-01'a delay of 2027 years 1 month
    '2027-001'a delay of 2027 years 1 month
    js
    await boss.send('email-reminder', { userId: 123 }, { startAfter: '2027-01-01T08:00:00Z' })
await boss.send('email-reminder', { userId: 123 }, { startAfter: '2027-01-01T13:30:00+05:30' })
await boss.send('email-reminder', { userId: 123 }, { startAfter: '1 hour' })

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 8601 date time or a relative interval | Date

See startAfter for how a string is interpreted.

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'))

// same instant, expressed with an offset instead of Z
await boss.sendAfter('email-reminder', { userId: 123 }, null, '2027-01-01T13:30:00+05:30')

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<UpdateResponse>: { 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', retries: 0 }, { 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, so 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. However, remember that on a key_strict_fifo queue, singletonKey is required to insert.

Returns a Promise<UpsertResponse>: { 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.

Concurrent upsert() calls with the same singletonKey agree on one job: at most one of them inserts, and the others update that job, waiting for the first one's transaction to finish if they have to. This holds on every queue policy and inside a transaction passed as db. It applies between upserts only. A send() or insert() with the same key still adds a job of its own wherever the queue policy allows several queued jobs per key (standard, singleton and key_strict_fifo), and a job brought back with resume() queues beside a newer upserted one rather than replacing it.

On CockroachDB the later upsert fails with a serialization error (40001) instead of waiting. pg-boss runs it again, up to 3 attempts in all, when the transaction is its own. Inside a db you pass, the error reaches you to retry.

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, Job[], 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.

Each job is an object of this shape:

ts
interface JobInsert<T = object> {
  id?: string;
  data?: T;
  priority?: number;
  retryLimit?: number;
  retryDelay?: number;
  retryBackoff?: boolean;
  retryDelayMax?: number;
  startAfter?: number | string | Date;
  singletonKey?: string;
  singletonSeconds?: number;
  expireInSeconds?: number;
  deleteAfterSeconds?: number;
  retentionSeconds?: number;
  heartbeatSeconds?: number;
  group?: { id: string; tier?: string };
  deadLetter?: string;
}

Each field works like the send() option of the same name, and a startAfter string is interpreted exactly as it is in send().

Returns a Promise<string[] | null>. Without returnId: true it always resolves to null. With returnId: true it resolves to the ids of the jobs that were inserted, or to null when no jobs were inserted, including when the array passed in is empty. It never resolves to an empty array. Add ?? [], as in the example below, to always get an array back.

Behind the scenes, insert() is a single INSERT statement. An error on any job, such as a value of the wrong type, rolls back the whole batch. A job that conflicts with an existing one is skipped, and the rest of the batch is still inserted. Conflicts are a duplicate id, a second job in the same singletonSeconds throttle slot, or a second queued job for the same singletonKey on a queue whose policy allows only one (short, stately or exclusive). The returned ids can then be fewer than the jobs passed in, and they are not guaranteed to line up with the input by position. If you need to align the input jobs with the output ids, you should set each job's id in the input array. Use flow() when the batch must be all or nothing, since it rolls back if any job is skipped. With a custom db, the insert commits or rolls back with your transaction.

js
const ids = 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 pg-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<JobInsert, 'data'>;
  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. When its last dependency is resolved, a dependent job's startAfter moves up to that moment unless it is already later, so it reads as the time the job could first run, and its wait in getQueueStats() counts from then.

Unblocking happens off the completion hot path: a background resolver wakes shortly after a parent completes (see flowIntervalSeconds in the constructor options) 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() 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.

A call made while another pass is in flight waits for it to finish, and stop() waits for the call. After stop(), it resolves without running.

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, deprecated, ignored since 12.30.0

      Jobs are always fetched in priority order. This option existed to skip the priority sort for throughput; the fetch index is now ordered to match the fetch, so there is no sort to skip. Setting it false was measured roughly 180x slower than leaving it alone, because no index leads with created_on. Emits a Node DeprecationWarning (code PGBOSS_DEP_FETCH_SORT) once per option per instance, and will be rejected in the next major. Run with --trace-deprecation to find the call site.

    • orderByCreatedOn, bool, deprecated, ignored since 12.30.0

      Jobs are always fetched in creation order. Same reasoning: the fetch index now provides that order directly, so disabling it saved nothing measurable.

    • 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.

      js
      interface JobWithMetadata<T = object> {
  id: string;
  name: string;
  data: T;
  priority: number;
  state: 'created' | 'retry' | 'active' | 'completed' | 'cancelled' | 'failed';
  retryLimit: number;
  retryCount: number;
  retryDelay: number;
  retryBackoff: boolean;
  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,
  deadLetter: string,
  policy: string,
  output: object,
  sourceName: string | null,
  sourceId: string | null,
  sourceCreatedOn: Date | null,
  sourceRetryCount: number | null,
  sourceOutput: object | null,
  sourceRootId: string | null
}

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, sourceRetryCount is how many retries it consumed before being dead-lettered, and sourceOutput is the original job's output when it failed, usually its error. The dead-lettered job's own output starts empty, since it is a new job that has not run. sourceId only names the previous hop, which after a redrive is the redriven copy. On the first failure, sourceRootId will be the same as sourceId. These are null for jobs that were not dead-lettered.

After a redrive, sourceRootId will be carried onto every dead-lettered and redriven job in the future, so the whole history of a job can be found from its original id.

Notes

The following example shows how to fetch and delete up to 20 jobs.

js
const QUEUE = 'email-daily-digest'
const emailer = require('./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. Accepts a fetched job in place of its id, which only deletes the attempt that was fetched (see passing jobs instead of ids).

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)

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 (with sourceRootId still pointing at the job send() returned), a reset retry count, cleared output, and the destination queue's current retry, retention, policy, expiration, heartbeat, and deadLetter configuration. Per-job overrides passed to the original send() (such as expireInSeconds or retryLimit) are not restored, since the queue's configuration wins. The job's priority, singletonKey, and group are carried through, so a redriven job keeps its ordering weight and stays subject to its group concurrency limits. Only jobs that are not currently being processed (still in the created/retry state) are moved.

Jobs that were part of a flow cannot be recovered with redrive(). A dead-lettered parent never reaches the completed state, so its dependents stay blocked, and the re-created job has a new id that the existing dependency rows do not point at. Redriving such a job runs it again standalone; the original flow does not resume.

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.
  • data: only redrive jobs whose payload contains this object, matched the same way as findJobs().
  • createdBefore: only redrive jobs that arrived in the dead letter queue before this Date. Pass the same value to every call of a loop to drain a fixed set: jobs dead-lettered while it runs are never swept in.
  • ids: only redrive these jobs, by their id in the dead letter 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.

A job whose re-created copy the destination refuses, because its short, stately or exclusive policy already holds a job with the same singletonKey (or one earlier in the same batch), is not lost: it stays in the dead letter queue in the failed state, with an output saying why (reason: 'redrive_conflict', plus the destination, policy, singletonKey and a message). Retry it once the job it collided with has finished and it becomes a redrive candidate again; otherwise the dead letter queue's deleteAfterSeconds removes it like any other failed job.

In a dead letter queue with the key_strict_fifo policy, only the oldest collision per singletonKey is failed. Jobs with a key that an active, retrying or failed job holds are not redrive candidates until that job is resolved, the same rule fetch follows.

Jobs a dead letter queue's own workers have already failed are never redriven; only jobs still waiting there are candidates. The return value counts only the jobs re-created, so a call can return 0 while waiting jobs remain, if every one of them collided; those are now failed and the next call moves on.

js
// drain a dead letter queue back to its source queues, 500 at a time, stopping at the
// jobs that were already there when it started
const createdBefore = new Date()
let left
do {
  await boss.redrive('email-dlq', { limit: 500, createdBefore })
  left = await boss.previewRedrive('email-dlq', { createdBefore })
} while (left.total > left.unroutable)

previewRedrive(name, options) ​

Reports what redrive() would do with the same options, without moving anything. Takes every redrive() filter, and uses the same matching, so the numbers agree with what a redrive would move at that moment.

It reads at most limit matching jobs of each kind rather than counting them all:

  • limit: most jobs to read of each kind, those a redrive would move and those it would leave in place, from 1 to 100,000 (default 1000). A count that reaches it means at least that many.

Returns { total, destinations, unroutable }, each counting the jobs read:

  • total: the matching jobs read, routable or not.
  • destinations: { name, count } for each queue the matching jobs would land in, largest first. Without destination this is the fan-out back to each source queue.
  • unroutable: matching jobs a redrive would leave in place, because they have no recorded source queue and no destination was given, or their source queue no longer exists.

A destination with a singleton or short policy can still drop jobs that collide at redrive time; the preview cannot see those collisions ahead of time.

js
const { total, destinations, unroutable } = await boss.previewRedrive('email-dlq', {
  data: { tenant: 'acme' },
  createdBefore: new Date(Date.now() - 24 * 60 * 60 * 1000)
})

deleteQueuedJobs(name) ​

Deletes all queued jobs in a queue, and resolves to how many it deleted.

Jobs are deleted in batches, each in its own transaction, so the delete is not atomic: if it fails partway, the batches before the failure stay deleted, and a job sent while it runs may or may not be deleted.

js
await boss.deleteQueuedJobs('email-send')

deleteStoredJobs(name) ​

Deletes all jobs in completed, failed, and cancelled state in a queue, and resolves to how many it deleted. Deletes in batches, as deleteQueuedJobs() does.

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.

Resolves to how many jobs it deleted, deleting in batches as deleteQueuedJobs() does. A partitioned queue, or every queue when no name is given, is emptied with a single TRUNCATE instead, which reports no count, so it resolves to null; its cached counts in getQueue() are zeroed at the same time. After any other delete, including deleteQueuedJobs() and deleteStoredJobs(), totalCount drops by the jobs deleted at once, and the other cached counts catch up at the next monitor pass.

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. Accepts a fetched job in place of its id, which only cancels the attempt that was fetched (see passing jobs instead of ids).

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. Its startAfter moves up to the time it was resumed unless it is already later, so its wait in getQueueStats() counts from then.

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. Its startAfter moves up to the time it was retried unless it is already later, so its wait in getQueueStats() counts from then.

js
await boss.retry('email-send', jobId)

retry(name, [ids], options) ​

Retries a set of failed jobs.

js
// requeue 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 ​

Passing jobs instead of ids ​

complete(), fail(), touch(), cancel() and deleteJob() accept the jobs fetch() returned in place of their ids, or any object with the job's id and retryCount. The call then only applies to the attempt that was fetched. If the claim lapsed in the meantime (the job expired, or missed its heartbeat, and has been retried by another worker), the job is left alone and reported as not affected, rather than settling the newer attempt with this one's outcome.

js
const [job] = await boss.fetch('report-generation')

const report = await generateReport(job.data)

// { affected: 0 } if the job was retried elsewhere while this report was generated
const { affected } = await boss.complete('report-generation', job, { reportUrl: report.url })

A plain id applies to whatever attempt currently holds the job, which is what an operator acting on a job wants. work() always settles and refreshes by attempt.

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, { 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], 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)
} catch (err) {
  // stored in the job's output and eligible for retry per the queue config
  await boss.fail('email-send', job, err)
}

fail(name, [ids], options) ​

Fails a set of active jobs.

js
const jobs = await boss.fetch('email-send', { batchSize: 10 })

await boss.fail('email-send', jobs, { 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)
}, 5000)

try {
  await processJob(job)
  await boss.complete('long-running-queue', job)
} 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 result = await boss.touch('long-running-queue', jobs)

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, oldest first.

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.

  • limit, int, default: 1000

    Most jobs to return, from 1 to 100,000.

  • after, string

    The id of the last job of the previous page, to read the next one. Rejects if that job no longer exists.

  • db, object, see notes in send()

Examples

js
// Find by id
const jobs = await boss.findJobs('my-queue', { id: 'abc-123' })

// Find by singletonKey
const jobs = await boss.findJobs('my-queue', { key: 'user-123' })

// Find by data
const jobs = await boss.findJobs('my-queue', { data: { type: 'email' } })

// Find queued jobs only
const jobs = await boss.findJobs('my-queue', { key: 'user-123', queued: true })

// At most 100 jobs from a large queue
const jobs = await boss.findJobs('my-queue', { data: { type: 'email' }, limit: 100 })

// Every job, 500 at a time
let after
for (;;) {
  const page = await boss.findJobs('my-queue', { limit: 500, after })
  if (page.length === 0) break
  after = page.at(-1).id
}

// Combine filters
const jobs = 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, sorted by queue name and id. options.limit sets the most it returns, from 1 to 100,000, and defaults to 1000. Pass the last reference as options.after to read the next page.

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, sorted by queue name and id. options.limit sets the most it returns, from 1 to 100,000, and defaults to 1000. Pass the last reference as options.after to read the next page.

js
const children = await boss.getDependents('process-data', parentJobId)
// [{ name: 'aggregate-results', id: '...' }]