@fullstackhouse/open-mercato-durable-work

Durable, at-least-once background work for Open Mercato.

A job is a row in your database with a lease on it. A worker claims the lease, does a slice of the work, and hands the rest back; if that worker dies, the lease expires and a reconciler repairs the job. Nothing depends on a process remembering anything, which is what makes a deploy, a crash and a network partition the same event.

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

| lease. | Default | What it does | |---|---|---| | ttlMs | 60000 | how long a lease lives without a heartbeat | | sliceBudgetMs | 300000 | when ctx.shouldYield() turns true; a soft limit the step answers at its next boundary | | maxSliceMs | 3 × sliceBudgetMs | the hard deadline on one slice; never below 2 × sliceBudgetMs nor above 2³¹−1 ms; null disables it | | pendingTtlMs | 900000 | how long a pending job may wait for its delivery before it is re-driven |

The budget is a question your step asks; maxSliceMs is enforced on a step that stopped asking — one stuck on a socket or a lock wait that never returns. Past it the slice's signal aborts, and if the step has not returned within a few seconds (five, or the TTL if shorter) the delivery stops waiting for it: the worker slot is freed, every lease write from the abandoned step is refused, the lease expires and the reconciler takes the job — re-driven for an orphanPolicy: 'redrive' kind, parked otherwise, and parked as poison if it hangs every time. Logged as durable_work.slice_overran and durable_work.slice_abandoned, and — if the abandoned step ever settles — durable_work.abandoned_step_settled with its outcome or error. See docs/adr/0008.

Operating

GET    /api/durable_work/jobs           durable_work.view
GET    /api/durable_work/jobs/[id]      durable_work.view
POST   /api/durable_work/jobs/[id]/redrive   durable_work.operate
DELETE /api/durable_work/jobs/[id]      durable_work.operate

A re-drive refuses with a code rather than a generic failure, because the three refusals have different answers: lock_key_held (wait for or cancel the job holding the key), not_redrivable (completed and cancelled jobs are done), unrecoverable_requires_force.

mercato durable_work reconcile runs one repair pass and prints what it repaired.

Sweeps

The reconciler repairs jobs. A row of your own that should be carried by a job and is not — its job was never created, say — is invisible to it. Register a sweep to repair those on the same tick:

registry.registerSweep({
  id: 'catalog.unindexed',
  async run(ctx) {
    // Select at most ctx.batchSize rows `for update skip locked`, honour ctx.tenantId, and
    // settle each one. ctx.start(input, scope) creates a job; call the returned enqueue() once
    // your row locks are released.
    return { acted: 0, errors: 0 }
  },
})

A sweep runs after the job queries on every pass, in every process that owns the tick. A throw counts as one error in the pass's report and stops nothing else. See docs/adr/0007.

What it promises, and what it does not

Promises. No job stays running forever — including one whose step never returns, which is abandoned at lease.maxSliceMs (a kind that sets maxSliceMs: null gives this part up). A worker that lost its lease cannot write. Work resumes from the last committed checkpoint. One live job per lock key per tenant. A job whose slices keep handing back without committing anything is parked as poison after budget.poisonRedrivesWithoutCommit of them, the same as one that keeps orphaning without a commit — a hand-back is progress only when something was committed since the last one.

Does not. Exactly-once execution: delivery is at-least-once, so a process killed between a side effect and its checkpoint will redo that side effect. Make them idempotent, or put them behind onTransition, which runs in the terminal transaction.

MIT. Part of open-mercato-durable.