@samveys/queue — Docs

Background jobs with retries and backoff, a database driver that never hands one job to two workers, a transactional outbox, a sync driver and a fake() for tests.

Define and dispatch a job

import { database, defineDatabaseConfig, sqlite } from '@samveys/database'
import { defineJob, queue } from '@samveys/queue'

export const sendInvoice = defineJob<{ invoiceId: number }>({
  name: 'invoice.send',
  retries: 3, // extra attempts after the first
  backoff: [1_000, 5_000, 30_000], // ms before retry 1, 2, 3 (the last entry repeats)
  timeoutMs: 30_000, // aborts ctx.signal and fails the attempt
  handler: async ({ invoiceId }, ctx) => {
    console.log(`sending invoice ${invoiceId}, attempt ${ctx.attempt}`)
  },
})

export const plugins = [
  database(defineDatabaseConfig({ default: 'main', connections: { main: sqlite({ file: 'app.sqlite' }) } })),
  queue({ jobs: [sendInvoice] }), // driver: 'database' by default
]

queue.dispatch(sendInvoice, { invoiceId: 7 }, { delayMs: 60_000, queue: 'billing' }) stores a row in samveys_jobs. Payloads must be JSON.

Workers

svs queue:work --queue=billing,default   # until SIGTERM/SIGINT; the current job finishes first
svs queue:work --once                    # one job
svs queue:work --max-jobs=100
svs queue:failed                         # jobs that used all their attempts (SVS_E100)
svs queue:retry 12                       # or: svs queue:retry all

Two workers never run the same job

On PostgreSQL, MySQL and MariaDB a worker reserves with SELECT … FOR UPDATE SKIP LOCKED and an UPDATE in one transaction. On SQLite one atomic UPDATE … WHERE id = (SELECT … LIMIT 1) … RETURNING * does it. A reserved job whose worker died becomes available again after visibilityMs (default 5 minutes — keep it above your longest job). Every reservation counts as an attempt, so a job that crashes its worker cannot loop forever.

Transactional outbox

Pass the transaction to dispatch: the job row is written in it. A rollback drops the job, a commit makes it durable — even if the process dies right after the commit.

import type { DatabaseManager } from '@samveys/database'
import { createQueue, databaseQueueDriver, defineJob } from '@samveys/queue'

declare const db: DatabaseManager

const welcome = defineJob<{ userId: number }>({ name: 'user.welcome', handler: () => {} })
const queue = createQueue({ driver: databaseQueueDriver({ database: () => db }) })

export async function register(email: string): Promise<void> {
  await db.transaction(async (trx) => {
    const created = await trx.from('users').insert({ email })
    await queue.dispatch(welcome, { userId: Number(created.insertId) }, { transaction: trx })
    // If anything above throws, the user row and the job are both rolled back.
  })
}

The queue tables must already exist when you dispatch inside a transaction (the transaction holds the connection, so they cannot be created lazily). Create them from a migration or at deploy time with createQueueTables(db).

Testing

import { defineJob, fake } from '@samveys/queue'

const job = defineJob<{ id: number }>({ name: 'report.build', handler: () => {} })
const queue = fake()

await queue.dispatch(job, { id: 3 })
queue.assertDispatched(job, (payload) => payload.id === 3)
queue.assertNotDispatched('other.job')

queue({ driver: 'sync' }) runs jobs inline at dispatch time (with the same retry rules) and throws SVS_E100 when they keep failing.

#Define and dispatch a job

#Workers

#Two workers never run the same job

#Transactional outbox

#Testing

Define and dispatch a job

Workers

Two workers never run the same job

Transactional outbox

Testing