SQL (database)

The SQL transport needs no broker. A dispatched step is a row in the same database your durable store already uses: ctx.step inserts a task row, a worker claims it with SELECT … FOR UPDATE SKIP LOCKED, runs the handler, writes a result row, and the engine polls that row to resume the run. The database you already operate is the queue.

pnpm add @dudousxd/nestjs-durable-transport-db

new DbTransport({ executor: mikroOrmExecutor(em) });           // rides your app's connection

const transport = new DbTransport({ executor: typeOrmExecutor(ds), group: 'pipeline.extract' });
transport.handle('pipeline.extract', async (input) => extract(input));

group is the queue token this instance polls — set it to the exact (optionally @<partition>-suffixed) step name it serves; each DbTransport instance polls one queue.

Trade-off vs a real broker: throughput is bounded by polling + row contention — great for workflow/pipeline scale (modest rate, long steps), not for high-fanout firehoses. Requires FOR UPDATE SKIP LOCKED: Postgres 9.5+ or MySQL 8+ (not SQLite).

Worker health

DbTransport doesn't implement groupHealth/listWorkerGroups — the engine's workerHealth() query, and the dashboard/Telescope worker-health panels built on it, return empty for this transport's groups; only a groupHealth-capable transport (BullMQ) returns non-empty results. Liveness here is the lease itself, not a separate heartbeat: a claimed row whose claimed_at is older than leaseMs is treated as abandoned and reclaimed by the next poll (see Claim protocol).

Options

The contract

This is the single source of truth for the table shape and the claim protocol. The TypeScript DbTransport and the Python db_runner both implement against this spec — so a Node engine and a Python worker share the same two tables without drift. Table names are <prefix>_transport_tasks and <prefix>_transport_results (default prefix durable).

Tables

CREATE TABLE durable_transport_tasks (
  step_id    varchar(191) PRIMARY KEY,   -- "<runId>:<seq>", the step identity
  run_id     varchar(191) NOT NULL,
  seq        integer      NOT NULL,
  name       varchar(191) NOT NULL,      -- step name → handler key
  grp        varchar(191) NOT NULL,      -- worker group → which workers poll it
  input      text,                       -- JSON (longtext on MySQL)
  attempt    integer      NOT NULL,
  status     varchar(32)  NOT NULL,      -- 'pending'
  claimed_by varchar(191),               -- lease owner (debug)
  claimed_at bigint,                     -- lease stamp (epoch ms)
  created_at bigint        NOT NULL
);

CREATE TABLE durable_transport_results (
  step_id    varchar(191) PRIMARY KEY,
  run_id     varchar(191) NOT NULL,
  seq        integer      NOT NULL,
  status     varchar(32)  NOT NULL,      -- 'completed' | 'failed'
  output     text,                       -- JSON
  error      text,                       -- JSON
  started_at bigint,                     -- worker pickup (epoch ms) → queue-wait
  claimed_by varchar(191),
  claimed_at bigint,
  created_at bigint        NOT NULL
);

started_at is what makes queue-wait work over this transport: the worker stamps when it picked the task up, the engine subtracts its dispatch time.

Claim protocol

A worker claims a batch atomically, runs each task outside the lock, then records the result:

1. Claim — in one transaction, lease up to batchSize un-leased rows for the group:

   SELECT * FROM durable_transport_tasks
    WHERE grp = ? AND (claimed_at IS NULL OR claimed_at < ?)   -- ? = now − leaseMs
    ORDER BY created_at ASC LIMIT ? FOR UPDATE SKIP LOCKED;
   UPDATE durable_transport_tasks SET claimed_by = ?, claimed_at = ?
    WHERE step_id IN (…);

   SKIP LOCKED is the heart of it: concurrent workers step over each other's locked rows instead of blocking, so a row is never claimed twice. The claimed_at < now − leaseMs clause reclaims a row whose worker crashed mid-flight (crash recovery, in place of heartbeats).
2. Run — execute the handler for name with the parsed input.
3. Record — insert the result row (idempotent: ON CONFLICT DO NOTHING / INSERT IGNORE) and DELETE the task row.

The engine side mirrors it on the results table: claim → deliver to the suspended ctx.step → DELETE. Delivery is at-least-once (a crash between insert-result and delete-task can re-run a handler), so handlers should be idempotent, keyed on step_id.

Dialect notes

Only two things differ between Postgres and MySQL, and both libraries handle them the same way: