SQL (database)

The db driver 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 an atomic lease, 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. To weigh it against the other drivers, see the comparison.

It ships inside @adonis-agora/durable — @adonisjs/lucid is an optional peer, imported lazily only when you select the db transport. The transport tables come from a migration published when you configure the package:

node ace configure @adonis-agora/durable
node ace migration:run

configure publishes four migrations: the durable-store migration, the transport-tables migration (create_durable_transport_tables, which creates the four db-transport channel tables), the worker-advertisement migration (create_durable_transport_worker_keys, which creates durable_transport_worker_keys), and the messages migration (create_durable_transport_messages, which creates durable_transport_messages for remote workflows and tenant pods). If you don't use the db transport, delete create_durable_transport_tables and create_durable_transport_messages before running them. Keep create_durable_transport_worker_keys if you run the queue transport with Lucid installed — its default worker advertisement lives in the same table.

Wiring it upCopy Anchor Link

transports.db() rides your app's default Lucid connection — the driver resolves the db service for you when it starts:

import { defineConfig, transports } from '@adonis-agora/durable'

export default defineConfig({
  transport: 'db',
  transports: {
    db: transports.db(), // pass { connection: 'durable' } to target a non-default connection
  },
})

The idiomatic worker is the bundled ace command. It boots the app from the same config/durable.ts and serves the step handlers under app/steps:

node ace durable:work

Outside that command you can construct the transport yourself and register handlers by name. Do it after the app has booted (Lucid's db service is undefined before that):

import db from '@adonisjs/lucid/services/db'
import { DbTransport } from '@adonis-agora/durable'

const transport = new DbTransport({ db }) // pass { partition } to isolate this pool
transport.handle('pipeline:extract', async (input) => extract(input))

Each handler name maps to a routing token — the handler name with : replaced by -, suffixed @<partition> when the instance sets a partition other than 'default' — and one DbTransport instance claims rows for every token it registered, in one claim loop. On the engine side the loops start when the engine registers its result, heartbeat, and control handlers; on the worker side the first handle() starts the task loop. start() is an explicit no-op convenience that reopens the pollers and waits for the schema; stop()/close() stop every loop and wait up to closeTimeoutMs (5 seconds) for in-flight work, so a claimed task mid-handler gets to write its result. Anything still unfinished after that is handed back for another worker to pick up at once. Neither closes the shared Lucid Database.

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. But it adds zero new infrastructure — no Redis, no SQS — which is often the right trade.

Concurrency and idle pollingCopy Anchor Link

One DbTransport runs one task loop for every handler name it serves, and by default that loop runs one task at a time. Raise concurrency to run several at once:

db: transports.db({
  concurrency: 8,        // up to 8 task rows in flight per worker process
  resultConcurrency: 1024, // the engine holds up to 1024 results at once (at least 512)
}),

The slots share the one poll loop. Each poll claims only as many rows as there are free slots (never more than batchSize), so a busy worker never holds rows an idle sibling could run, and when every slot is busy the loop waits for one to free instead of polling. Every claim the worker holds is renewed (see the claim protocol), and stop()/close() waits within its closeTimeoutMs budget for every running slot. Fleet-wide parallelism for a step is concurrency × worker processes; use a flow-control queue for a cap across the fleet.

concurrency: 'adaptive', or { mode: 'adaptive', min, max, start, tickMs, ramCeilingPct, cpuCeilingPct }, lets the worker tune its own limit with the controller the queue and BullMQ transports use, with the same fields and defaults. It grows the limit while steps saturate it at a steady latency (doubling until the first move down, then ⌊√limit⌋ a tick), and shrinks it when steps start to queue, on errors or stalls, at the CPU ceiling, or when RSS reaches ramCeilingPct. The task loop reads the live limit on every poll. A lower limit lets running tasks finish and claims nothing new until in-flight work is under it.

Idle task, heartbeat, control, step-event and tenant-protocol loops back off: after each empty poll the sleep doubles, up to idleBackoffMax × pollIntervalMs (800ms by default), and the first row found resets it. The cost is up to that much extra pickup latency for the first task after a quiet spell; idleBackoffMax: 1 restores the fixed cadence. The results, decisions and run-reply loops never back off, because each is on a waiting run's or call's critical path.

The consumers setting applies here too. A deferred process (a node ace command or REPL under the default 'auto') starts no task, result, heartbeat or control loop, writes no liveness key, and claims nothing. It still consumes the message rows addressed to it alone (its run replies, tenant events and decisions), because no other process could take them.

PriorityCopy Anchor Link

A dispatch's priority (from ctx.step(..., { priority }); higher is more urgent) is stored on the task row, and workers claim higher priority first, then oldest. A dispatch without a priority is stored as 0, "normal": a positive priority runs ahead of it, a negative one behind it. The queue transport places an unprioritised job in the same middle position. The value is rounded and clamped to a 32-bit integer. Unlike the queue transport's 1–10 broker scale, the full range is kept, so priority: 7 still runs ahead of priority: 6.

Ordering is best-effort, like on the broker transports. It decides which row a free slot claims next, but it never interrupts a running step, and rows claimed in the same poll start together.

Worker healthCopy Anchor Link

Every worker process beats a liveness key for each step name it serves — durable-worker-heartbeat:<token>:<instanceId> (durable-<namespace>-… in a non-default namespace) in durable_transport_worker_keys, refreshed every workerHeartbeatIntervalMs (10s) and live for 35s. Each key carries the worker's WorkerStatus: concurrency mode and live limit, in-flight count, RSS, CPU, throughput, p95 and the last adaptive adjustment. A process with deferred consumers beats nothing until it starts consuming.

From those keys and the tasks table, DbTransport implements groupHealth, so engine.workerHealth(), the dashboard's /api/workers and /api/health and the Telescope worker panel report db-transport groups:

• depth: every task row of the group in this namespace, pending and claimed — rows waiting for a worker plus rows a worker is running (or held when it died, until they are reclaimed). The database can count both, so this is closer to BullMQ's waiting + active than the queue transport's pending-only depth.
• liveWorkers: the group's unexpired liveness keys, with each worker's instanceId, lastBeatAt and status. A worker that dies drops out within 35s.

depth > 0 with no live worker is the alert state. listWorkerGroups() and listWorkerDescriptors(token) read the same worker advertisement table, so the engine also knows which tokens have a live worker for dispatch routing.

Liveness of a claimed task is its lease, not the liveness key: a task row whose claim stopped being renewed for leaseMs is treated as abandoned and reclaimed by the next poll (see the claim protocol). The durable_transport_heartbeats table is a different thing — the per-step liveness pings that timeoutMs steps send to the engine.

Poison tasksCopy Anchor Link

A task row whose claim lapses (its worker crashed, or blocked its event loop past leaseMs) is reclaimed by the next poll, and each takeover adds one to the row's reclaims count. A row reclaimed more than maxStalledCount times (default 3, as on the queue transport) is one that kills or wedges every worker it reaches, so the worker that claims it next does not run it. It deletes the row and publishes a failed result for the step, remote step worker failed: claim lapsed more than N times, marked retryable. The engine treats that like any failed attempt: it re-dispatches the step while it has retries left (a fresh row, counted from zero), then fails the step into the workflow, where a try/catch around the step sees a RemoteStepError. Before this bound existed, such a row was re-delivered forever.

A remote workflow turn row past the bound is deleted and reported to onError without a decision, as on bullmq and queue: a turn has no step to fail, so the run waits for the turn's liveness window (remoteAdvanceSilenceMs, or the executor's timeoutMs) to re-drive it.

Only task rows are bounded. A result, heartbeat, control or message row whose engine dies while handling it is re-delivered after the lease, without a count.

Closing a worker hands its claims backCopy Anchor Link

Only a claim that lapses counts. When a worker shuts down cleanly (close()/stop(), which durable:work calls on SIGTERM), it waits up to closeTimeoutMs for in-flight work. Then it clears the claim on every row it still holds: running tasks, and any result, heartbeat or control batch it was still handling. It clears a claim only where claimed_by is still its own. The next poll on any worker claims those rows as fresh, at once rather than after leaseMs, and reclaims does not go up, so a long step that spans several deploys is never failed as poison. A worker that dies without closing (SIGKILL, OOM, a lost node) cannot hand anything back. Its rows wait out the lease, and each such takeover counts.

A handler still running when its row is handed back keeps running in the closing process, since it cannot be aborted. Another worker may run the same step meanwhile. The closing worker then drops its own result and reports it to onError, so steps should be idempotent. Raise closeTimeoutMs if your steps usually finish within a longer drain.

Remote workflows and tenant podsCopy Anchor Link

The db transport also carries the two protocols that used to need bullmq or queue:

• Remote workflow turns. A workflow registered with engine.registerRemote(...) and a RemoteWorkflowExecutor advances one turn at a time on a worker that serves its body (handleWorkflow, which a WorkerRuntime calls for every registerWorkflow). A turn is a task row of kind workflow on the workflow's routing token, next to the steps of the same name: it is claimed by priority, runs in one of the worker's concurrency slots, and is renewed, handed back on close and bounded by maxStalledCount like a step. The worker writes the turn's decision as a row of durable_transport_messages addressed to the engine instance that dispatched the turn. Any replica could apply it (the engine applies a decision by run id, against the taskId recorded on the run), so the addressing only keeps the decision with an instance known to be live. If that instance dies before taking the row, the turn's liveness window (remoteAdvanceSilenceMs) re-drives it. Streamed step events (a long turn's local steps, checkpointed live) are rows any engine instance may take.
• Store-less tenant pods. A pod's run starts and read/control requests are rows any control-plane instance may take. Each carries the requesting instance's id, and the reply is a row addressed to that instance only, so two pods never receive or take each other's replies. Live-tail events are written once per pod subscribed to the event's tenant, addressed to that pod; pods of other tenants receive nothing.

A pod on the db transport reaches the same database as the control plane by definition, so it needs no durable store but does need the transport tables. Point both at the same connection:

export default defineConfig({
  role: 'control-plane',
  transport: 'db',
  transports: { db: transports.db() },
  store: 'lucid',
  stores: { lucid: stores.lucid() },
  verifyTenant: hmacTenantVerifier(process.env.DURABLE_TENANT_SECRET!),
})

export default defineConfig({
  role: 'tenant',
  transport: 'db',
  // The transport's `partition` decides which `<name>@acme-corp` rows the pod claims.
  transports: { db: transports.db({ partition: 'acme-corp' }) },
  partition: 'acme-corp',
  tenant: { token: process.env.DURABLE_TENANT_TOKEN },
})

The pod writes its own worker advertisement and live-tail subscriptions to durable_transport_worker_keys, as any db worker does — nothing is relayed through the control plane:

• <prefix>-inbox:<instanceId> says an instance is consuming the rows addressed to it (a requester waiting for replies, a subscriber, an engine waiting for decisions).
• <prefix>-tenant-subscriber:<tenant>:<instanceId> says it wants that tenant's events. A publisher reads these (reusing the read for tenantSubscribersCacheMs) and writes one row per live subscriber, in one INSERT.

Both keys live for inboxTtlMs and are re-written every third of it. A clean close() removes them and deletes the rows still addressed to the closing instance. Rows any instance may take (start-run, run-request, step events) are handed back on close like task rows.

How this compares with the other transports carrying the same protocol:

• Latency. Rows are polled. The reply loop and the decision loop never back off (one pollIntervalMs), like the results loop: each is on the critical path of a waiting call or turn. The run-request, start-run, step-event and tenant-event loops back off when idle, like the task loops (up to idleBackoffMax × pollIntervalMs); a burst resets them. A proxied call therefore costs up to one backed-off control-plane poll plus one reply poll after a quiet spell. Set idleBackoffMax: 1 on the control plane to keep it at two polls.
• Subscribing. A subscription is visible to publishers once its key is written plus tenantSubscribersCacheMs, so events published in the first second or so after subscribing can be missed, as on queue. bullmq pub/sub misses only events published before the subscription.
• Order. Events from one control-plane instance reach a pod in the order they were published. Events from several instances are not ordered against each other, on any transport.
• Cost. Each event is one row per subscribed pod of its tenant. Each consuming instance re-writes one or two small keys every third of inboxTtlMs. Every engine polls the step-event table, and every standalone or control-plane engine polls the start-run and run-request rows — with idle backoff, a few cheap indexed queries per second.
• Trust. Anything with access to the database can subscribe to any tenant's events, as anything with access to the broker can on bullmq and queue. Requests are checked against verifyTenant on the control plane; subscriptions are not. A standalone or control-plane engine on db starts the run-request responder, as on bullmq and queue. A control-plane engine logs a warning at boot unless verifyTenant is set; a standalone engine boots quietly.
• Interop. The rows are this package's own wire. Python and NestJS workers speak bullmq only.

Without the messages table (autoCreate: false and no create_durable_transport_messages migration), the transport reports it once through onError the first time something uses the protocol, and run starts, requests and remote turns fail at once instead of timing out. A deployment that never uses the protocol hears nothing.

The run queue lives in the same table. Without it every run start falls back to the pending-run poll (the run is still executed, one poll interval later), and the engine reports the failed enqueues at most once a minute.

Options — transports.db(config)Copy Anchor Link

transports.db(config) returns the lazy driver thunk used in config/durable.ts; the underlying class is DbTransport, also exported from @adonis-agora/durable for constructing a worker-side instance directly. @adonis-agora/durable also exports TRANSPORT_TABLES, createDurableTransportTables(db), and dropDurableTransportTables(db) for managing the schema yourself.

Constructing DbTransport directly takes the same options with two differences: the Lucid Database goes in db, and the connection name is connectionName (not connection). It also accepts one option transports.db() does not forward:

The tablesCopy Anchor Link

The published migrations create six tables (default names):

durable_transport_tasks carries step_id (<runId>:<seq>, the primary key and step identity), run_id, seq, name (the handler key), grp (the routing token workers claim by), namespace, input, traceparent, context, transport, attempt, reclaims (how many times a lapsed claim on the row was taken over), priority (the dispatch priority rows are claimed by, 0 when absent), kind (step, or workflow for a remote workflow turn whose WorkflowTask is the input), plus the claim columns claimed_by/claimed_at and created_at. The results table mirrors it with status (completed or failed), output, error, events, and started_at (when the worker picked the task up — what powers queue-wait timing in the dashboard). Heartbeats, control and message rows use an auto-increment id as their key; control and message rows carry the whole message as a JSON payload. A message row also has kind (decision, stepEvent, startRun, runRequest, runReply, tenantEvent), recipient ('' for a row any consumer of its kind may take, else the instance id it is addressed to), and reply_to (the requester of a start-run or run-request).

Every channel table has a namespace column (default 'default') and an index leading with it; every claim is scoped to the instance's namespace.

Upgrading existing tablesCopy Anchor Link

Columns added after the first release are back-filled on existing tables by createDurableTransportTables, and so by autoCreate on first use. No new migration is needed:

autoCreate also creates durable_transport_messages when it is missing.

With autoCreate: false, add the columns yourself in a migration of your own (table.integer('reclaims').notNullable().defaultTo(0) and table.integer('priority').notNullable().defaultTo(0)). Until you do, tasks still run: without reclaims, takeovers are not counted and the poison bound is off; without priority, dispatch priority is ignored and rows are claimed oldest first. The transport reports each missing column once through onError. For kind and the messages table, publish and run the messages migration: re-run node ace configure @adonis-agora/durable (or copy create_durable_transport_messages from the package's stubs/database/migrations), keep only the new create_durable_transport_messages file, and run node ace migration:run. It creates the table and adds kind to an existing tasks table. Without them, steps run as before and only remote workflows and tenant pods are off.

The worker-key table is a key/value board, not a message channel: key (primary key), value, and expires_at. Each write upserts its key and sweeps expired ones; it has no namespace column because the keys embed the namespace.

Column typesCopy Anchor Link

The schema is built with Lucid's schema builder (Knex), so the same DDL runs on SQLite, Postgres, and MySQL:

There is no dependency on a dialect's native JSON or date type, and no dialect-specific locking clause.

The claim protocolCopy Anchor Link

A worker claims rows atomically, runs each task outside the lock, then records the result — portable across SQLite, Postgres, and MySQL without FOR UPDATE SKIP LOCKED:

1. Claim — select candidate rows for the served tokens that are unclaimed or whose lease expired (claimed_at IS NULL OR claimed_at < now − leaseMs), highest priority first and then oldest, bounded by the free slots; then conditional UPDATEs stamp a unique per-round token into claimed_by/claimed_at (and add one to reclaims on a row whose lapsed claim it takes over); then select back the rows that match this round's token. Two instances can never claim the same row because the token is unique per claim round.
2. Run — re-check that the round still holds the row, then execute the handler for name with the parsed input.
3. Record — in one transaction, delete the task row where claimed_by is still this round's token and, only if that deleted it, insert the result row (idempotent on step_id).

While rows are held, the worker renews every claim it holds — running or not — every leaseMs / 3, with one UPDATE per table matched on its claim tokens. A slow but alive worker therefore keeps all its rows for as long as it needs them; only a worker that stopped renewing (crashed, lost the database, or blocked its event loop past the lease) has its rows reclaimed. If that happens to a live worker, it finds the claim gone when it finishes, drops its result instead of writing a second one, and reports the lost claim to onError. The step may then have run twice, so steps should still be idempotent.

The engine side mirrors the protocol on the results table: claim up to 256 rows a round, with rounds pipelined so at least 512 results (or resultConcurrency) are in the engine's hands at once, deliver each result, then delete each round's delivered rows in one WHERE IN statement, again only where claimed_by is still its token. The batch is renewed while it is being delivered, so a slow settle never hands the rest of the batch to another engine. Rows are deleted only after their handler ran. Heartbeats and control messages are drained the same way, keyed on id.