Jobs

Background jobs run work that outlives the request that started it (an AI call, an export). A procedure enqueues and returns { jobId } at once; a worker in any process runs the job with retries and backoff; status goes out on the pub/sub, so the client subscribes instead of polling. For the concepts, see Features → Background jobs.

import { createJobs, createInMemoryJobStore, createPgJobStore, followJob } from 'dendra'
import { useJob } from 'dendra/react'

const jobs = createJobs({ store: createPgJobStore(pool), pubsub, context: () => ({ db }) })
 
const summarize = jobs.define('summarize', {
  input: s.object({ docId: s.string() }),
  key: (input) => input.docId, // one queued-or-running job per doc
  run: async ({ input, ctx, job }) => {
    await job.progress({ step: 'loading' })
    return { summary: await ctx.db.summarize(input.docId, { signal: job.signal }) }
  },
})
 
jobs.start() // in every process that should run jobs
 
export const appRouter = {
  summarize: {
    start: authed.input(s.object({ docId: s.string() })).background(summarize), // → { jobId }
    watch: authed.input(s.object({ jobId: s.string() }))
      .stream(({ input, signal }) => summarize.watch(input.jobId, { signal })),
  },
}

.background(job, map?) is a route builder method; see Router. It validates the input, runs the middlewares, and enqueues. A job’s ctx comes from context(), never from the request, so use map to pass what the job needs from the call (tenant, user id) in its input. Check that the caller may see a job in the watch procedure; watch itself doesn’t check ownership.

Defining and running jobs

createJobs(opts)

Creates the job system: a registry of named jobs and a worker. Every process that enqueues or runs jobs calls it with the same store and pub/sub.

function createJobs<Ctx>(opts: JobsOptions<Ctx>): Jobs<Ctx>

Jobs are named, not tied to a procedure path, so a queued job still finds its handler after a deploy renames a route.

JobsOptions

Jobs

What createJobs returns.

A worker only claims jobs whose names are defined in its process. Each loop it enqueues due schedules (if the store supports them), claims one job whose name has free concurrency, and goes straight on to the next; when there’s nothing to claim it waits pollMs or until woken.

define throws if the name is already defined, if timeoutMs isn’t a positive safe integer, if onInterrupt isn’t 'retry' or 'fail', or if timeoutMs or onInterrupt is set and the store doesn’t advertise capabilities.executionPolicies.

start() after stop() does nothing: a stopped Jobs doesn’t restart. Create a new one.

JobOptions

Retries. When run throws, the error goes to onError. If attempts remain, the job goes back to "queued" with the error and runs again after the backoff (1 s, 2 s, 4 s, … up to 60 s by default); otherwise it’s "failed". The snapshot’s error is { code, message }: a DendraError keeps its code and message, anything else becomes INTERNAL: Internal error (the full error goes to onError).

Keys. With key, enqueueing while a job with the same name and key is "queued" or "running" creates nothing and returns the existing jobId. Once that job succeeds or fails, the key is free again. A key of null or undefined never dedupes.

Interruptions. These follow onInterrupt:

On a deadline or lost lease, job.signal aborts. A handler that ignores the signal keeps its concurrency slot until it actually returns, but its late progress and result are ignored. These errors have code INTERNAL.

Job

What define returns. Pass it to .background() or call it directly.

enqueue and schedule throw a DendraError BAD_REQUEST (status 400, with the schema issues as details) when the input is invalid. watch throws NOT_FOUND when there’s no such job. watch subscribes before reading the store, and drops events whose version it has already seen, so each state appears once and in order.

const { jobId } = await summarize.enqueue({ docId: 'd1' }, { delayMs: 5_000 })
for await (const snap of summarize.watch(jobId)) console.log(snap.status, snap.progress)

JobHandle

The job argument run receives.

JobSnapshot

What a job looks like from outside: get, watch, and the pub/sub events. O is the job’s result type.

type JobSnapshot<O = unknown> = {
  id: string
  name: string
  version: number // goes up with every change; higher is newer
  attempt: number
  maxAttempts: number
  progress: unknown
} & (
  | { status: 'queued' | 'running'; result?: undefined; error: JobError | null }
  | { status: 'succeeded'; result: O; error: null }
  | { status: 'failed'; result?: undefined; error: JobError }
)

A retrying job is "queued" with the last attempt’s error. Snapshots are published on the topic dendra:job:<id>; wake-ups use dendra:jobs.

JobStatus

type JobStatus = 'queued' | 'running' | 'succeeded' | 'failed'

succeeded and failed are final.

JobError

interface JobError {
  code: ErrorCode
  message: string
}

JobStopOptions

Both must be safe integers with 0 <= graceMs <= timeoutMs, or stop throws.

JobStopResult

process.on('SIGTERM', async () => {
  const stopped = await jobs.stop({ graceMs: 6_000, timeoutMs: 8_000 })
  if (!stopped.drained) console.error('Unfinished jobs', stopped.pending, stopped.errors)
  process.exit(0)
})

Schedules

job.schedule(id, input, calendar) registers a durable daily or weekly schedule in UTC. It needs a store that supports schedules (both built-in stores do; otherwise it throws). The input is validated with the job’s schema, and the key is computed from it.

await sync.schedule('daily-search-sync', { workspaceId: 13 },
  { kind: 'daily', hourUTC: 2, minuteUTC: 0 })
await rankCheck.schedule('weekly-rank-check', { workspaceId: 13 },
  { kind: 'weekly', weekdayUTC: 1, hourUTC: 3, minuteUTC: 0 })
await jobs.unschedule('weekly-rank-check')

• The first run is the next occurrence strictly after now.
• Registering the same id with the same configuration keeps its next occurrence, so restarting a process can’t postpone a missed run. A changed configuration resets it and gets a new revision.
• Workers enqueue each occurrence and advance the schedule atomically, so concurrent workers never create duplicates. Each worker loop handles at most 100 due schedules.
• After an outage, missed occurrences coalesce into one catch-up job, then the next future occurrence. If a job with the same key is still active, the occurrence is consumed without enqueueing.
• Removing a schedule doesn’t cancel jobs already enqueued.

schedule throws if id is empty or the calendar is invalid (RangeError).

JobCalendar

Recurring wall-clock times in UTC. Sunday is weekday 0; Saturday is 6.

type JobCalendar =
  | { kind: 'daily'; hourUTC: number; minuteUTC: number }
  | { kind: 'weekly'; weekdayUTC: number; hourUTC: number; minuteUTC: number }

hourUTC is an integer 0–23, minuteUTC 0–59, weekdayUTC 0–6.

ScheduleRecord

A stored schedule.

NewSchedule

A schedule to store: a ScheduleRecord without revision.

type NewSchedule = Omit<ScheduleRecord, 'revision'>

JobScheduleStore

The schedule methods a store implements to support schedules. Optional for custom stores; advertise it with capabilities.schedules: true.

Stores

createInMemoryJobStore()

A single-process store for development and tests. Jobs and schedules don’t survive a restart, and other processes can’t see them. Supports execution policies and schedules.

function createInMemoryJobStore(): JobStore & JobScheduleStore

createPgJobStore(pool, opts?)

A Postgres store: jobs survive deploys and crashes, and any number of worker processes share the queue. Workers claim with FOR UPDATE SKIP LOCKED, so two never take the same job. It takes a structural pool, so a pg.Pool drops in. Supports execution policies and schedules.

function createPgJobStore(pool: PgQueryable, opts?: { table?: string }): PgJobStore

const store = createPgJobStore(pool)
await store.setup() // or copy store.schema into your migration

Tables. store.schema is a list of additive, idempotent statements. With the default table name they create:

The schema also runs ALTER TABLE … ADD COLUMN IF NOT EXISTS for timeout_ms, on_interrupt and deadline_at, so an existing 0.9 table is upgraded. Apply the updated schema as the database owner before deploying new workers, and stop old workers before using timeoutMs or onInterrupt: 0.9 workers don’t understand them.

Finished jobs stay in the table until you delete them:

DELETE FROM dendra_jobs
WHERE status IN ('succeeded', 'failed') AND updated_at < now() - interval '7 days';

add throws if it can’t enqueue after three tries (a keyed insert that keeps racing with a job that finishes).

PgJobStore

A JobStore plus the JobScheduleStore methods and:

PgQueryable

What createPgJobStore needs from a pool. A pg.Pool matches.

interface PgQueryable {
  query(text: string, values?: unknown[]): Promise<{ rows: unknown[] }>
}

Writing a store

JobStore

The interface a store implements. Every write after claim carries the claim’s token, and does nothing once another worker has taken the job over (its lease ran out).

A store may also implement the JobScheduleStore methods.

JobRecord

A stored job.

NewJob

A job to insert.

interface NewJob {
  id: string
  name: string
  key: string | null
  input: unknown
  maxAttempts: number
  runAt: number
  timeoutMs?: number
  onInterrupt?: 'retry' | 'fail'
}

JobSettlement

How an attempt ended, passed to settle.

type JobSettlement =
  | { status: 'succeeded'; result: unknown }
  | { status: 'failed'; error: JobError }
  // back in the queue: a retry, or (refund) an attempt interrupted by stop()
  | { status: 'queued'; error: JobError | null; runAt: number; refund: boolean }

With refund: true, the store subtracts one from attempt, so the interrupted attempt isn’t counted.

Following a job from the client

A watch procedure streams snapshots until the job settles. These helpers keep that stream open across drops: when it ends early (network error, a WebSocket UNAVAILABLE, a proxy closing it) they open it again after a backoff. A fresh watch starts with the current state, so nothing is missed, and version drops states already seen. Don’t poll or write this loop yourself.

followJob(open, opts?)

Framework-agnostic. open opens the watch stream with the given signal.

function followJob<S extends JobSnapshotLike>(
  open: (signal: AbortSignal) => AsyncIterable<S>,
  opts?: FollowJobOptions,
): JobFollower<S>

const f = followJob((signal) => api.summarize.watch({ jobId }, { signal }))
f.subscribe(({ job, connection }) => render(job, connection))
const final = await f.done // the succeeded or failed snapshot; f.stop() to give up

An error with a string code other than UNAVAILABLE (e.g. NOT_FOUND, UNAUTHORIZED) won’t go away by retrying: following stops and error is set. Errors without a code, UNAVAILABLE, and a stream that ends before a final state are retried.

FollowJobOptions

JobFollower

FollowedJob

JobConnection

type JobConnection = 'idle' | 'connecting' | 'live' | 'reconnecting' | 'ended'

JobSnapshotLike

The part of a snapshot followJob reads. JobSnapshot (and its JSON form on the client) matches.

interface JobSnapshotLike {
  status: string
  version: number
}

useJob(jobId, open, opts?)

From dendra/react. Follows a job by id with followJob and returns its state. It stops when the job settles, jobId changes, or the component unmounts. A null or undefined id returns connection: 'idle'.

function useJob<S extends JobSnapshotLike>(
  jobId: string | null | undefined,
  open: (jobId: string, signal: AbortSignal) => AsyncIterable<S>,
  opts?: Pick<FollowJobOptions, 'backoff' | 'online'>,
): FollowedJob<S>

open may be an inline function; only a new jobId restarts the watch. A new id never shows the previous job’s state.

import { useJob } from 'dendra/react'
 
const { job, connection, error } = useJob(jobId, (id, signal) =>
  api.summarize.watch({ jobId: id }, { signal }),
)
if (error) return <p>{error.message}</p>
if (!job) return <Spinner />
if (job.status === 'failed') return <p>Failed: {job.error.message}</p>
if (job.status === 'succeeded') return <Summary text={job.result.summary} />
return <Progress value={job.progress} reconnecting={connection === 'reconnecting'} />