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

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

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

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

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

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.

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

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

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

// 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).

// 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 — omit it to edit only options.

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.

// 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 }).

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.

The following contract is a typescript defintion 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.

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

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

interface FlowJob {
  ref: string;
  name: string;
  data?: object;
  options?: Omit<JobInsert, 'data'>;
  dependsOn?: string[];
}

Returns a map of ref to created job id.

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

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.

    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
    }

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.

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.

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.

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.

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.

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

await boss.deleteQueuedJobs('email-send')

deleteStoredJobs(name)

Deletes all jobs in completed, failed, and cancelled state in a queue.

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.

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

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.

await boss.cancel('email-send', [jobId1, jobId2])

resume(name, id, options)

Resumes a cancelled job.

await boss.resume('email-send', jobId)

resume(name, [ids], options)

Resumes a set of cancelled jobs.

await boss.resume('email-send', [jobId1, jobId2])

retry(name, id, options)

Retries a failed job.

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

retry(name, [ids], options)

Retries a set of failed jobs.

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

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.

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

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], options)

Fails a set of active jobs.

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.

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.

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

// 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 })

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

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.

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