Durable Work

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

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