Jobs

@ultimabase/jobs runs code outside the request that asked for it. A run is a row of the database, written in the write's transaction: it exists if and only if the write committed, a crash or a restart does not lose it, and it is retried until it succeeds. A job can also run every few minutes, or on a cron pattern in a time zone. Every server runs the jobs, and each run on one server at a time.

npm install @ultimabase/jobs

A job takes a name, the input its runs take, and the function that runs them:

// The calls, whose video room closes in a job ten minutes after they end.
import { defineJob } from '@ultimabase/jobs'
import { defineModel } from 'ultimabase'
import { z } from 'zod'
import { closeVideoRoom } from './services.ts'

export const closeRoom = defineJob({
	name: 'closeRoom',
	// Validated when the run is enqueued, and stored with it.
	input: z.object({ room: z.string() }),
	// A run may run twice, after a crash: `id`, the run's, is the key that
	// makes the service close the room once.
	run: ({ input, id }) => {
		closeVideoRoom(input.room, { key: id })
	},
})

The full example, with its tests: examples/jobs.

Enqueue a runCopy Anchor Link

enqueue validates the input, then writes the run in the current transaction, or alone outside one. From an afterUpdate hook, the run exists if the update commits, and never after a rollback. after delays it, in milliseconds:

export const Call = defineModel({
	name: 'Call',
	fields: (f) => ({
		room: f.string(),
		endedAt: f.dateTime({ optional: true }),
	}),
}).with({
	hooks: {
		// In the write's transaction: the run exists if the update commits.
		afterUpdate: async ({ record, previous }) => {
			if (previous.endedAt !== null || record.endedAt === null) return
			// In milliseconds: the room stays open ten minutes, to rejoin.
			await closeRoom.enqueue({ room: record.room }, { after: 10 * 60 * 1000 })
		},
	},
})

A run due now starts as its transaction commits, on the server that wrote it; a delayed run starts within five seconds of its due time, on whichever server finds it first. An invalid input is refused with VALIDATION_FAILED, as a function's, and no run is written.

A *Commit hook runs once too, but in memory: a crash before it ends loses it, and it never runs again. A job is for what must happen, even after a crash.

Once or moreCopy Anchor Link

A run that throws runs again 10 seconds later, then twice as long each time. After ten attempts, about 2 h 50 min, its row turns failed and is kept, with its last error, for someone to read.

A run may also run twice: a server that crashes mid-run leaves it to another five minutes later. A job that calls another service passes id, the run's, as its idempotency key, or checks the state it changes. A run is never cancelled: the job reads what is still true, as whether the call is still over, when it runs.

Recurring jobsCopy Anchor Link

A job with every runs every few milliseconds from the start; one with cron runs on a five-field pattern, minute, hour, day of the month, month and day of the week, in timeZone, UTC by default:

// Every hour from the start, in milliseconds.
export const purgeCalls = defineJob({
	name: 'purgeCalls',
	every: 60 * 60 * 1000,
	run: async () => {
		const before = new Date(Date.now() - 30 * day)
		await Call.deleteMany(
			{ where: { endedAt: { lt: before } } },
			{ as: master },
		)
	},
})

// At 3:00 each night in Paris, daylight saving included; UTC by default.
export const nightlyReport = defineJob({
	name: 'nightlyReport',
	cron: '0 3 * * *',
	timeZone: 'Europe/Paris',
	run: async () => {
		const since = new Date(Date.now() - day)
		const ended = await Call.count(
			{ where: { endedAt: { gte: since } } },
			{ as: master },
		)
		outbox.push(`Calls ended yesterday: ${ended}.`)
	},
})

A recurring job takes no input and is never enqueued. Each has one row, holding its next due time, so each occurrence runs on one server. While no server ran, missed occurrences run once, not once each. A failure is logged at error, and the next occurrence runs as planned.

Across daylight saving, a time the clocks skip runs at the first time after the gap, and a time they repeat runs once.

The pluginCopy Anchor Link

jobsPlugin() takes every job the application runs:

export const app = defineApp({
	// Kept in memory, so the example needs no database.
	database: memory(),
	// Listens on PORT, or 3000, and answers /health.
	plugins: [
		httpPlugin(),
		jobsPlugin({ jobs: [closeRoom, purgeCalls, nightlyReport] }),
	],
	models: [Call],
})

At start, it writes the rows of new recurring jobs, re-times those whose schedule changed, and deletes those of jobs no longer declared. Stopping, a server waits for the runs it started.

In a testCopy Anchor Link

runJobs(), from @ultimabase/testing, runs every run due now, and the runs they enqueue, as the servers would. The clock decides what is due: vi.useFakeTimers({ toFake: ['Date'] }) moves the date alone, before the application starts, so the recurring jobs are due from it:

// Before the application starts: its recurring jobs are due from now.
beforeEach(() => {
	// Fakes the date alone, so the timers still run.
	vi.useFakeTimers({ toFake: ['Date'] })
	vi.setSystemTime(new Date('2027-01-04T08:00:00Z'))
})
afterEach(() => {
	vi.useRealTimers()
})

testApplication(app)

A delayed run is due once the test moves the clock past its delay:

const call = await Call.create({ data: { room: 'standup' } }, { as: master })
await Call.update(
	{ where: { id: call.id }, data: { endedAt: new Date() } },
	{ as: master },
)
// Runs what is due now, and what those runs enqueue: nothing yet.
await runJobs()
expect(closedRooms).toEqual([])

vi.setSystemTime(Date.now() + 10 * minute)
await runJobs()
expect(closedRooms).toEqual(['standup'])

And a recurring job, once the clock reaches its time:

await Call.create(
	{ data: { room: 'standup', endedAt: new Date() } },
	{ as: master },
)
// 3:00 in Paris, in winter.
vi.setSystemTime(new Date('2027-01-05T02:00:00Z'))
await runJobs()
expect(outbox).toEqual(['Calls ended yesterday: 1.'])

ReferenceCopy Anchor Link

defineJob(definition)Copy Anchor Link

A job takes one of input, every and cron, or none. With input or none, it has enqueue(input?, options?):

jobsPlugin(options)Copy Anchor Link

It adds the model JobsRun, which no transport exposes: one row for each pending or failed run, and one for each recurring job.

runJobs()Copy Anchor Link

Runs, during a test, every run due now and the runs they enqueue, then resolves.

ErrorsCopy Anchor Link

LimitsCopy Anchor Link

• A run starts within five seconds of its due time, each server polling the database; a run due now starts as its write commits.
• A server runs 20 runs at a time, and holds each for five minutes: a run longer than that may start again on another server.
• Each run runs at least once, not exactly once: its id is the idempotency key.
• A run's input is stored as JSON: it holds what JSON holds, a date as its ISO string.
• cron takes five fields, no seconds.