What it is for
Work that is too long to redo:
- a multi-day data import that a deploy would otherwise kill
- a job that stays
running forever because the worker that held it is gone - a transient error at hour nine throwing away the first eight
- two workers driving one job over a single cursor
Install
yarn add @fullstackhouse/open-mercato-durable-work pg-boss
The transport client is yours to choose and is an optional peer: pg-boss for the default transport, or bullmq and ioredis for the Redis one. Nothing is installed for the transport you do not run.
Still 0.x — pin the exact version in production. If you also run @fullstackhouse/open-mercato-data-sync-durable, it releases in lockstep with this package at the same version; upgrade the two together.
Exactly one copy of this package may end up in the tree. It is a peer of everything that builds on it, and that is not tidiness: it exports a process-wide registry, so a second copy would mean job kinds register into one while the worker reads the other — silently.
src/modules.ts:
{ id: 'durable_work', from: '@fullstackhouse/open-mercato-durable-work' },
Then yarn generate && yarn db:migrate, and run the worker as its own process:
yarn mercato durable_work worker
Its own process on purpose: a slice can run for minutes, and hosting that inside the web process means a deploy either kills work mid-batch or waits out a slice.
Configuration
| Variable | Default | What it does |
|---|
DURABLE_WORK_TRANSPORT | pgboss | pgboss, bullmq or memory |
DURABLE_WORK_REDIS_URL | QUEUE_REDIS_URL | BullMQ only |
DURABLE_WORK_PGBOSS_SCHEMA | durable_work_boss | keeps pg-boss's tables out of public |
DURABLE_WORK_TICK_MS | 15000 | how often the reconciler runs |
DURABLE_WORK_DRAIN_TIMEOUT_MS | 30000 | how long a SIGTERM waits for slices to hand back |
pgboss is the default because it needs nothing but the database you already have, and it is the only transport that can enqueue a delivery inside your own transaction. Use bullmq if the app already runs Redis. memory is for development and is refused in production.
Declaring work
import { registry } from '@fullstackhouse/open-mercato-durable-work'
registry.register({
kind: 'catalog.reindex',
queue: 'durable-work.catalog',
// A job nobody declares idempotent is parked for a human rather than re-run automatically.
orphanPolicy: 'redrive',
async step(ctx) {
let done = ctx.checkpoint?.done ?? 0
while (done < total) {
// Stop at a boundary when the budget is spent or the process is shutting down.
if (ctx.shouldYield()) return 'budget'
await ctx.fencedWrite(async (tx) => {
// Rolls back if the lease was lost, so this cannot outlive the right to write it.
})
done += 1
await ctx.checkpoint_({ done }) // resume point, and a committed unit of work
}
return 'drained'
},
// Mirrors the terminal state onto your own row, in the same transaction.
async onTransition(job, scope, tx) { /* … */ return { matched: 1 } },
async onRedrive(job, scope, tx) { /* … */ return { matched: 1 } },
})
Throw TransientError (or anything unrecognised) to retry, TerminalError to stop, and UnrecoverableError to stop and require { force: true } before it can run again.
Lease settings
