jobs

Some work should not keep a request waiting: emailing an invoice, generating a PDF, syncing orders, pruning expired sessions at 3am. @kitr/jobs hands that work to a background worker. You define a job with a typed payload, dispatch it from a controller, and the worker runs it with retries, backoff, a timeout and an optional schedule.

It runs on the server only. The built-in queue lives in memory, which suits one process; for several servers, or jobs that must survive a restart, use a shared driver (Share the queue between processes). Work that must finish before the response, such as charging a card, should be called directly.

The examples follow Acme's app at app.acme.com, which generates a monthly invoice for each team, emails it, and cleans up old data every night.

Install

kitr add jobs

Everything kitr add changes

What kitr add does, in order:

• Installs @kitr/jobs. Jobs require the server package, so kitr add also adds server (and its logger) when the project does not have them yet.
• Adds '@kitr/jobs' to packages and () => import('@kitr/jobs/provider') to providers in kitrrc.ts.
• Writes config/jobs.ts (shown in Configure) and regenerates config/index.ts.
• Writes an example job, server/jobs/example.ts, exporting exampleJob.
• When you answer no to the worker question, adds a worker script (kitr jobs:work) to package.json.
• Adds a ## kitr: jobs section to AGENTS.md telling coding agents to put background work in server/jobs/ and never setTimeout work in a request, plus a line about the separate worker when there is one.

No kit includes jobs; add it to any kit with a server platform.

kitr remove jobs reverses every step above. Your jobs in server/jobs/ stay, because you may have edited them.

kitr add jobs asks one question. Agents and CI answer it with a flag:

kitr add jobs --no-worker

With --no-worker, config/jobs.ts gets worker: false and the worker runs in its own process, as in Run the worker in its own process.

Configure

kitr add jobs writes this file:

// config/jobs.ts
import { defineConfig } from '@kitr/jobs'
import { exampleJob } from '#server/jobs/example'

export default defineConfig({
  // Every job the worker runs (kitr make:job <name> writes one in server/jobs/).
  jobs: [exampleJob],
  // In-memory queue by default: fine for one process. For several, pass a shared driver.
  concurrency: 5,
  worker: true, // false when a separate worker process runs the jobs
})

Per-job options (retries, backoff, timeout, schedule) live on the job itself; see Define a job.

Dispatch your first job

Generate a job:

kitr make:job generate-invoice

That writes server/jobs/generate-invoice.ts. Edit it to build a team's invoice for a month:

// server/jobs/generate-invoice.ts
import { defineJob } from '@kitr/jobs'
import { z } from 'zod'

export const generateInvoiceInput = z.object({
  teamId: z.string(),
  month: z.string().regex(/^\d{4}-\d{2}$/),
})

export const generateInvoice = defineJob('generate_invoice', {
  schema: generateInvoiceInput,
  retries: 3,
  async run({ teamId, month }, { attempt }) {
    // Load the team's usage, render the PDF, store it under storage/invoices/.
    return { invoiceNumber: `INV-${month}-${teamId}`, attempt }
  },
})

List it in config/jobs.ts, next to the example:

// config/jobs.ts
import { defineConfig } from '@kitr/jobs'
import { exampleJob } from '#server/jobs/example'
import { generateInvoice } from '#server/jobs/generate_invoice'

export default defineConfig({
  jobs: [exampleJob, generateInvoice],
  concurrency: 5,
  worker: true,
})

Dispatch it from a controller. dispatch() checks the payload, queues the job and resolves with its id straight away:

// server/controllers/invoices-controller.ts
import type { HttpContext } from '@kitr/api'
import { useJobs } from '@kitr/jobs'
import { generateInvoice } from '#server/jobs/generate_invoice'

export default class InvoicesController {
  async store({ app, params, response }: HttpContext<{ teamId: string }>) {
    const jobId = await useJobs(app).dispatch(generateInvoice, {
      teamId: params.teamId,
      month: '2026-10',
    })
    return response.json({ jobId }, { status: 202 })
    // => { "jobId": "3f0c9a52-6b1e-4d7a-9a51-0e2f8c1d7b44" }
  }
}

Register the route with api.post('/teams/:teamId/invoices', [InvoicesController, 'store']) in server/routes.ts. The request answers at once; within half a second (pollInterval) the worker picks the job up.

The payload is typed from the schema. A wrong value at run time, such as month: 'October', makes dispatch() throw a VALIDATION error before anything is queued; in a controller that becomes a 422.

Define a job

defineJob(name, options) returns a job object.

The second argument of run is the JobContext:

Without a schema, a type parameter types the payload: defineJob<{ teamId: string }>('notify_team', { run: ... }). A job with no payload is dispatched with undefined.

Dispatch a job

On the server, useJobs(app).dispatch(job, payload, options?) queues a job and resolves with its id. It works in controllers, actions, services and other jobs:

// server/jobs/close-month.ts
import { defineJob, useJobs } from '@kitr/jobs'
import { generateInvoice } from '#server/jobs/generate_invoice'

/** Fans out one invoice job per team. */
export const closeMonth = defineJob<{ month: string; teamIds: string[] }>('close_month', {
  async run({ month, teamIds }, { app }) {
    const jobs = useJobs(app)
    const ids = await Promise.all(teamIds.map((teamId) => jobs.dispatch(generateInvoice, { teamId, month })))
    return ids.length
  },
})

useJobs(app) is app.container.use('jobs'). Pass ids and small plain values in the payload, not whole records; the job loads fresh records when it runs.

Delay a job

Pass delay in milliseconds, or at for a fixed time:

// server/services/trial-service.ts
import type { KitrApp } from '@kitr/kitr/app'
import { useJobs } from '@kitr/jobs'
import { generateInvoice } from '#server/jobs/generate_invoice'

export async function scheduleFirstInvoice(app: KitrApp, teamId: string) {
  const jobs = useJobs(app)
  // Ten minutes from now.
  await jobs.dispatch(generateInvoice, { teamId, month: '2026-11' }, { delay: 10 * 60_000 })
  // On the first of next month, 09:00 UTC.
  await jobs.dispatch(
    generateInvoice,
    { teamId, month: '2026-11' },
    { at: new Date('2026-12-01T09:00:00Z') },
  )
}

With the memory driver a delayed job is lost if the process restarts before it runs.

Run a job on a schedule

Give a job a schedule and list it in config/jobs.ts. The worker dispatches it; you never call dispatch():

// server/jobs/prune-sessions.ts
import { defineJob } from '@kitr/jobs'

/** Every day at 03:00 server time. */
export const pruneSessions = defineJob('prune_sessions', {
  schedule: '0 3 * * *',
  async run(_payload, { app }) {
    app.container.use('logger').info('pruning expired sessions')
  },
})

// server/jobs/sync-orders.ts
import { defineJob } from '@kitr/jobs'

/** Every five minutes, starting when the worker starts. */
export const syncOrders = defineJob('sync_orders', {
  schedule: { every: '5m' },
  timeout: 4 * 60_000,
  async run(_payload, { signal }) {
    const res = await fetch('https://payments.acme.com/orders?since=5m', { signal })
    return res.status
  },
})

Cron expressions have five fields, read in the server's local time zone:

Each field takes *, a number, a list (1,15), a range (1-5) or a step (*/10, 0-30/10). Names like MON are not supported. When both day of month and weekday are set, either may match, as in standard cron.

Intervals take a number and a unit: 250ms, 30s, 5m, 1.5h, 1d. An interval job runs once when the worker starts, then every interval after.

A bad expression or interval throws at once, when the app boots. Check new expressions with cronMatches():

// scripts/check_cron.ts
import { cronMatches, parseInterval } from '@kitr/jobs'

cronMatches('0 3 * * *', new Date(2026, 10, 7, 3, 0)) // => true
cronMatches('*/15 9-17 * * 1-5', new Date(2026, 10, 7, 9, 30)) // => false, a Saturday
parseInterval('5m') // => 300000

With the memory driver, each worker queues its own copy, so two servers run prune_sessions twice; the Redis driver queues each run once. Cron reads each server's local time, so give every server the same TZ.

Retry a failing job

When run throws or times out, the worker queues the job again after backoff × 2^(attempt - 1) milliseconds, until it has used its retries:

// server/jobs/send-invoice-email.ts
import { defineJob } from '@kitr/jobs'
import { useMailer } from '@kitr/mail'
import { z } from 'zod'
import { invoiceReadyMail } from '#server/mail/invoice_ready'

export const sendInvoiceEmail = defineJob('send_invoice_email', {
  schema: z.object({ teamName: z.string(), invoiceNumber: z.string(), email: z.string() }),
  retries: 5,
  backoff: 10_000,
  async run({ teamName, invoiceNumber, email }, { app }) {
    await useMailer(app).send(invoiceReadyMail, { teamName, invoiceNumber, total: '$480.00' }, { to: email })
  },
})

A mail outage is retried after 10s, 20s, 40s, 80s and 160s, then given up. The logger shows each step:

WARN  09:00:10.112 [app] [jobs] send_invoice_email failed (attempt 1), retry in 10000ms
WARN  09:00:20.140 [app] [jobs] send_invoice_email failed (attempt 2), retry in 20000ms
...
ERROR 09:05:10.301 [app] [jobs] send_invoice_email failed after 6 attempts

A payload that fails the schema when the worker picks it up is dropped without a retry and logged as [jobs] <name> dropped: bad payload.

Handle a job that keeps failing

onFailed in config/jobs.ts runs when a job has used its retries or its payload was rejected. Use it to alert someone:

// config/jobs.ts
import { defineConfig } from '@kitr/jobs'
import { generateInvoice } from '#server/jobs/generate_invoice'
import { sendInvoiceEmail } from '#server/jobs/send_invoice_email'

export default defineConfig({
  jobs: [generateInvoice, sendInvoiceEmail],
  onFailed(job, error) {
    console.error(`job ${job.name} (${job.id}) gave up after ${job.attempt} attempts`, error)
  },
})

job is the QueuedJob:

onFailed has no app. To report through monitoring, catch inside run with ctx.app and rethrow so the job still retries:

// server/jobs/generate-invoice.ts
import { defineJob } from '@kitr/jobs'

export const generateInvoice = defineJob<{ teamId: string; month: string }>('generate_invoice', {
  retries: 3,
  async run({ teamId, month }, { app, attempt }) {
    try {
      return { invoiceNumber: `INV-${month}-${teamId}` }
    } catch (error) {
      app.container.use('monitoring').captureException(error, { extra: { teamId, month, attempt } })
      throw error
    }
  },
})

Time out a stuck job

Each attempt gets timeout milliseconds (60 seconds by default). At the limit ctx.signal aborts and the attempt fails and is retried. Pass the signal to anything that accepts one, so the work really stops:

// server/jobs/export-projects.ts
import { defineJob } from '@kitr/jobs'

export const exportProjects = defineJob<{ teamId: string }>('export_projects', {
  timeout: 5 * 60_000,
  retries: 1,
  async run({ teamId }, { signal }) {
    const res = await fetch(`https://exports.acme.com/teams/${teamId}`, { method: 'POST', signal })
    if (!res.ok) throw new Error(`export failed: ${res.status}`)
    return res.status
  },
})

Run a job inline

useJobs(app).run(job, payload) runs a job now, in this process, with its retries and timeout. Nothing is queued:

// scripts/backfill_invoices.ts
import { useJobs } from '@kitr/jobs'
import { generateInvoice } from '#server/jobs/generate_invoice'
import { server } from '#server/main'

await server.boot()
const result = await useJobs(server.app).run(generateInvoice, { teamId: 'team_engines', month: '2026-09' })
// => { invoiceNumber: 'INV-2026-09-team_engines', attempt: 1 }
await server.close()

When every attempt fails, run() throws the last error after the real backoff, and does not call onFailed.

Share the queue between processes

With the memory driver, other servers never see the queue's jobs, and a restart loses them. To share one queue, use createRedisJobsDriver with an ioredis client:

pnpm add ioredis

// config/jobs.ts
import { createRedisJobsDriver, defineConfig } from '@kitr/jobs'
import { Redis } from 'ioredis'
import { generateInvoice } from '#server/jobs/generate_invoice'
import { pruneSessions } from '#server/jobs/prune_sessions'

export default defineConfig({
  jobs: [generateInvoice, pruneSessions],
  driver: createRedisJobsDriver(new Redis(process.env.REDIS_URL ?? 'redis://localhost:6379'), {
    prefix: 'acme-app:jobs:',
    visibilityTimeout: 10 * 60_000,
  }),
})

Add the connection string to .env:

REDIS_URL=redis://localhost:6379

What the driver guarantees:

• Each due job runs on one worker. Twenty invoices dispatched while three workers poll are generated twenty times, not sixty.
• A crashed worker's jobs come back after visibilityTimeout.
• Scheduled jobs queue once per run, through a Redis lock named after the job and the minute.
• Retries keep their id, so onFailed and your logs see one job.

Redis keys

The queue lives under <prefix>queue (a sorted set by runAt), <prefix>reserved and <prefix>data. jobs.driver.size() counts waiting and reserved jobs across every process.

Run the worker in its own process

By default the worker runs inside the web server, starts with it and drains on shutdown. When long jobs should not compete with requests, run a worker process and turn the worker off in the web servers.

// config/jobs.ts
import { createRedisJobsDriver, defineConfig } from '@kitr/jobs'
import { Redis } from 'ioredis'
import { generateInvoice } from '#server/jobs/generate_invoice'
import { pruneSessions } from '#server/jobs/prune_sessions'

export default defineConfig({
  jobs: [generateInvoice, pruneSessions],
  driver: createRedisJobsDriver(new Redis(process.env.REDIS_URL ?? 'redis://localhost:6379')),
  // JOBS_WORKER=off on the web servers: they only queue, the worker process runs jobs.
  worker: process.env.JOBS_WORKER !== 'off',
})

// server/worker.ts
import { useJobs } from '@kitr/jobs'
import { server } from '#server/main'

await server.boot()
const jobs = useJobs(server.app)
jobs.start()
server.logger.info('worker started', { jobs: [...jobs.jobs.keys()] })

for (const signal of ['SIGINT', 'SIGTERM'] as const) {
  process.once(signal, async () => {
    // Waits for running jobs, then runs every provider's shutdown.
    await server.close()
    process.exit(0)
  })
}

Deploy the web servers with JOBS_WORKER=off, and run workers beside them with tsx server/worker.ts or kitr jobs:work. jobs.stop() waits for running jobs, each for at most its timeout, so set your platform's shutdown grace period above your longest job timeout.

Read the queue

The Jobs service exposes its state for health checks:

// server/controllers/admin-jobs-controller.ts
import type { HttpContext } from '@kitr/api'
import { useJobs } from '@kitr/jobs'

export default class AdminJobsController {
  async show({ app }: HttpContext) {
    const jobs = useJobs(app)
    return {
      driver: jobs.driver.name,
      running: jobs.running,
      queued: await jobs.driver.size(),
      jobs: [...jobs.jobs.keys()],
    }
    // => { driver: 'memory', running: true, queued: 0, jobs: ['example', 'generate_invoice'] }
  }
}

Commands

kitr jobs:list reads the defaults when a job sets none (3 retries, a 60,000 ms timeout):

kitr jobs:list
# name              retries  timeout  schedule
# generate_invoice  3        60000
# prune_sessions    3        60000    0 3 * * *

kitr jobs:work boots server.app from the entry's exported server (or app). kitr doctor flags a config/jobs.ts that lists no jobs.

kitr make:job generate-invoice writes server/jobs/generate-invoice.ts:

// server/jobs/generate-invoice.ts
import { defineJob } from '@kitr/jobs'
import { z } from 'zod'

export const generateInvoiceInput = z.object({
  id: z.string(),
})

/** List it in config/jobs.ts, then dispatch with `useJobs(app).dispatch(generateInvoice, { id })`; it runs on the worker with retries. */
export const generateInvoice = defineJob('generate_invoice', {
  schema: generateInvoiceInput,
  retries: 3,
  // schedule: { every: '1h' },
  async run({ id }, ctx) {
    console.info(`generate_invoice ${id} (attempt ${ctx.attempt})`)
  },
})

The generator does not add the job to config/jobs.ts; do that yourself.

Errors

The worker never throws; it logs through the app's logger:

Testing

In a test the worker does not start. Dispatched jobs wait in the memory queue, where you can read them, and jobs.run() runs one inline:

// test/generate-invoice.test.ts
import { useJobs } from '@kitr/jobs'
import jobsProvider from '@kitr/jobs/provider'
import { createTestApp } from '@kitr/testing'
import { describe, expect, it } from 'vitest'
import { generateInvoice } from '#server/jobs/generate_invoice'

describe('generate_invoice', () => {
  it('queues the job with its payload', async () => {
    const app = await createTestApp({
      providers: [jobsProvider],
      config: { jobs: { jobs: [generateInvoice] } },
    })
    const jobs = useJobs(app)

    const id = await jobs.dispatch(generateInvoice, { teamId: 'team_engines', month: '2026-10' })

    expect(jobs.running).toBe(false)
    expect(await jobs.driver.size()).toBe(1)
    const [queued] = await jobs.driver.pull(Date.now(), 10)
    expect(queued).toMatchObject({
      id,
      name: 'generate_invoice',
      payload: { teamId: 'team_engines', month: '2026-10' },
      attempt: 1,
    })
  })

  it('builds the invoice number', async () => {
    const app = await createTestApp({ providers: [jobsProvider] })
    const result = await useJobs(app).run(generateInvoice, {
      teamId: 'team_engines',
      month: '2026-10',
    })
    expect(result).toEqual({ invoiceNumber: 'INV-2026-10-team_engines', attempt: 1 })
  })

  it('rejects a bad month before queueing', async () => {
    const app = await createTestApp({ providers: [jobsProvider] })
    const jobs = useJobs(app)
    await expect(
      jobs.dispatch(generateInvoice, { teamId: 'team_engines', month: 'October' }),
    ).rejects.toMatchObject({ code: 'VALIDATION', status: 422 })
    expect(await jobs.driver.size()).toBe(0)
  })
})

To check a delay, pull at a later time: jobs.driver.pull(Date.now() + 10 * 60_000, 10).

// test/flaky-job.test.ts
import { createJobs, defineJob } from '@kitr/jobs'
import { createTestApp } from '@kitr/testing'
import { expect, it } from 'vitest'

it('retries until the third attempt succeeds', async () => {
  const jobs = createJobs(await createTestApp())
  const attempts: number[] = []
  const flaky = defineJob('sync_orders_flaky', {
    retries: 2,
    backoff: 1,
    run: (_payload, { attempt }) => {
      attempts.push(attempt)
      if (attempt < 3) throw new Error('payments API down')
      return 'synced'
    },
  })
  expect(await jobs.run(flaky, undefined)).toBe('synced')
  expect(attempts).toEqual([1, 2, 3])
})

To test the worker loop, start it with a short poll, wait for the effect, then stop it:

// test/worker.test.ts
import { createJobs, defineJob } from '@kitr/jobs'
import { createTestApp } from '@kitr/testing'
import { expect, it, vi } from 'vitest'

it('runs dispatched jobs on the worker', async () => {
  const done: string[] = []
  const notify = defineJob<string>('notify_team', { run: (teamId) => void done.push(teamId) })
  const jobs = createJobs(await createTestApp(), { jobs: [notify], pollInterval: 5 })
  jobs.start()
  await jobs.dispatch(notify, 'team_engines')
  await vi.waitFor(() => expect(done).toEqual(['team_engines']))
  await jobs.stop()
})

createJobs(app, config) builds the service without the provider, with the same options as config/jobs.ts.

Extend

A driver is where queued jobs wait. The worker calls pull(now, limit) every pollInterval and ack(job) after each run. See JobsDriver and QueuedJob.

The smallest driver that shares a queue is one Redis sorted set scored by runAt. Each job is claimed with ZREM, so two workers never run the same one:

// server/jobs/sorted-set-driver.ts
import { Redis } from 'ioredis'
import type { JobsDriver, QueuedJob } from '@kitr/jobs'

export function createSortedSetDriver(url: string, key = 'acme:jobs'): JobsDriver {
  const redis = new Redis(url)
  return {
    name: 'redis',
    async push(job) {
      await redis.zadd(key, job.runAt, JSON.stringify(job))
    },
    async pull(now, limit) {
      const due = await redis.zrangebyscore(key, 0, now, 'LIMIT', 0, limit)
      const claimed: QueuedJob[] = []
      for (const entry of due) {
        // Only the worker whose ZREM removes the entry runs it.
        if ((await redis.zrem(key, entry)) === 1) claimed.push(JSON.parse(entry) as QueuedJob)
      }
      return claimed
    },
    async size() {
      return redis.zcard(key)
    },
    async clear() {
      await redis.del(key)
    },
  }
}

Register it in config/jobs.ts:

// config/jobs.ts
import { defineConfig } from '@kitr/jobs'
import { generateInvoice } from '#server/jobs/generate_invoice'
import { createSortedSetDriver } from '#server/jobs/sorted_set_driver'

export default defineConfig({
  jobs: [generateInvoice],
  driver: createSortedSetDriver(process.env.REDIS_URL ?? 'redis://localhost:6379'),
})

Keep payloads to plain JSON: a Date comes back as a string. Without ack, a job is lost if the process dies mid-run; without claim, each worker queues its own copy of a scheduled job.

Next steps