Queues

Some work shouldn't hold up a response: calling another service, resizing an image, sending a hundred emails. Put it in a job, dispatch it, and a worker runs it, retrying when it fails.

Writing a jobCopy Anchor Link

pnpm exec caldera make:job PublishNote

import { defineJob } from '@caldera/queue'
import { z } from 'zod'
import { Note } from '../models/note'

export const PublishNote = defineJob({
  name: 'notes.publish',
  payload: z.object({ noteId: z.string() }),
  tries: 3,
  backoff: [10, 60],
  async handle({ noteId }, ctx) {
    const note = await Note.find(noteId)
    if (!note) return // deleted since it was queued
    const res = await fetch('https://hooks.example.com/notes', {
      method: 'POST',
      body: JSON.stringify({ id: note.id, title: note.title }),
      signal: ctx.signal,
    })
    if (!res.ok) throw new Error(`The webhook answered ${res.status}`)
  },
})

Pass ids, not rows: the job reads the row when it runs, so it sees the latest version, and a deleted row is a quiet return.

ctx has the attempt number, ctx.services (the app's services, like ctx.services.mail), and ctx.signal, which aborts at the timeout. Pass it to fetch and anything else that takes one.

DispatchingCopy Anchor Link

import { dispatch } from '@caldera/queue'

await dispatch(PublishNote, { noteId: note.id })

The payload is typed by the job's schema, and a wrong one is an error here, where the mistake is, not later in a worker.

afterCommit matters when the job reads what the transaction writes. Without it, a fast worker can look for the note before it's committed:

await c.var.db.transaction(async () => {
  const note = await Note.create({ userId, title })
  await dispatch(PublishNote, { noteId: note.id }, { afterCommit: true })
})

When jobs failCopy Anchor Link

A job fails when handle throws or runs out of time. It's tried again after its backoff until it has used its tries. Then failed runs, if the job has one:

async failed({ noteId }, error) {
  console.error(`Couldn't publish note ${noteId}`, error)
},

Inside handle, the job can decide for itself:

if (res.status === 429) ctx.release(30)

ConnectionsCopy Anchor Link

Where jobs go is set in src/config/queue.ts, and QUEUE_CONNECTION picks one:

QStash, for serverless apps without workers, is in @caldera/queue/qstash.

Running a workerCopy Anchor Link

The database and redis connections need a worker running beside the app:

pnpm worker

Processing jobs: notes.publish, notes.report-published
✓ notes.report-published #1
↻ notes.publish #1 retry in 10s: The webhook answered 500

It runs every job in src/jobs. --queue high,default takes queues in that order; --concurrency 4 runs four jobs at once. Restart it when you deploy, so it runs the new code.

On the database connection, jobs that used up their tries stay in the table:

pnpm exec caldera queue:failed
pnpm exec caldera queue:retry <id>   # or --all
pnpm exec caldera queue:forget <id>  # or --all

Unique jobsCopy Anchor Link

Some jobs only need to happen once, however many times they're asked for. With unique, a second dispatch while one is queued or running does nothing:

export const RebuildSearchIndex = defineJob({
  name: 'notes.rebuild-search-index',
  payload: z.object({ userId: z.string() }),
  unique: { key: (p) => p.userId, for: '1h' },
  // …
})

const { dispatched } = await dispatch(RebuildSearchIndex, { userId })
// false: one is already waiting for this user

for is how long the lock lasts if the job never finishes. Unique jobs keep their locks in the cache, so with more than one server, use Redis.

MiddlewareCopy Anchor Link

Middleware wraps a job's runs. Two come with Caldera:

import { rateLimited, withoutOverlapping } from '@caldera/queue'

middleware: [withoutOverlapping({ key: (p) => p.userId })],

Both put the job back to wait, which counts as one of its tries, so give such jobs plenty of tries or a retryUntil.

ChainsCopy Anchor Link

A chain runs jobs one after another. If one fails, the rest don't run:

import { chain } from '@caldera/queue'

await chain([
  ExportNotes.with({ userId }),
  ZipExport.with({ userId }),
  EmailExportLink.with({ userId }),
]).dispatch()

BatchesCopy Anchor Link

A batch runs jobs side by side and tracks them together:

import { batch } from '@caldera/queue'

const notes = await Note.ownedBy(userId).findMany()
const { batchId } = await batch(notes.map((note) => PublishNote.with({ noteId: note.id })))
  .then(ReportPublished.with({ userId }))
  .dispatch()

c.var.queue.batchStatus(batchId) reports progress, for a progress bar: { total, pending, failed, finished }. Batches count in the cache too.

TestingCopy Anchor Link

With sync, the tests' connection, a dispatched job has already run when dispatch returns. To check what was dispatched instead, fake the queue:

const queue = t.fake('queue')

await client.post(`/notes/${note.id}/publish`)

queue.assertDispatched(PublishNote, (p) => p.noteId === note.id)
queue.assertNotDispatched(ReportPublished)

Faked jobs wait until drain(), which runs them for real, with their retries and whatever they dispatch:

const results = await queue.drain()
results.map((r) => r.outcome.status) // ['retry', 'retry', 'failed']

queue.dispatched(PublishNote) lists each dispatch with its payload and delay.