Overview

A transport is how a dispatched ctx.step reaches a step handler and how the result comes back. It's a pluggable driver (dispatch, onResult, onHeartbeat), independent of the state store — mix any transport with any store.

Transports are config-driven drivers: list the ones you use under transports in config/durable.ts, built with the transports factory, and pick the active one by name with transport. Each driver lazily imports its optional peer (@adonisjs/queue, @adonisjs/lucid, bullmq + ioredis) only when selected, so nothing extra is loaded for the drivers you don't use. Only the selected entry is built: a process runs exactly one transport.

The driversCopy Anchor Link

Selecting a transport in configCopy Anchor Link

transport names a key of the transports map; that driver is what the engine dispatches every remote step over. Omit transport (or pick memory) for the in-process default.

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

export default defineConfig({
  transport: 'queue',
  transports: {
    memory: transports.memory(),
    // `connection` names an adapter from config/queue.ts; `db` uses a Lucid connection.
    queue: transports.queue({ connection: 'redis' }),
    db: transports.db(),
  },
})

Switching transports is a one-line change to transport — the workflow code never changes.

Single-process production — eventEmitterCopy Anchor Link

transports.eventEmitter() runs real durable workflows with nothing else to deploy: no broker, no Redis, no extra database. Step handlers under app/steps are served in this same process, and a dispatch travels to them over a Node EventEmitter:

export default defineConfig({
  transport: 'event-emitter',
  transports: {
    'event-emitter': transports.eventEmitter(),
  },
  store: 'lucid',
  stores: { lucid: stores.lucid() },
})

Pair it with the Lucid store and the durability guarantees are the real ones: a crash mid-run resumes from the last checkpoint, because the checkpoints are in Postgres even though the steps never left the process.

What you do not get is a second process serving your steps: a ctx.step dispatched over eventEmitter is handled only by the process that dispatched it. A dispatched step with no registered handler is ignored silently (another subscriber on the same emitter might own it), so the run waits on it. Lifecycle events and cancellation reach subscribers in this process with no control plane at all; they do not fan out across pods.

Several replicas can still share eventEmitter + a shared Lucid store when every step is a ctx.localStep: leases, checkpoints, signals and cancellation all go through the store, so two replicas never execute the same run. What you lose without a control plane is cross-pod fan-out — live-tail, onCancel listeners, schedule pause and a waitForRun (or a blocking start()) that waits on a run resumed by another pod. Add controlPlanes.redis for those, and move to queue, db or bullmq once steps must run in another process. See Pick a topology.

In-process for tests — memoryCopy Anchor Link

transports.memory() is the default when config/durable.ts sets no transport, and it is what createTestEngine() uses. It routes a dispatch straight into the handler registered for that name, so a test never waits on a poll interval.

1@Workflow({ name: 'process-doc', version: '1' })2export default class ProcessDocumentWorkflow {3  constructor(4    private intake: IntakeSteps,5    private render: RenderSteps,6  ) {}7 8  async run(ctx: WorkflowCtx, doc: Document) {9    const clean = await ctx.step(this.intake.validate, doc)10    const pdf = await ctx.step(this.render.toPdf, clean)11    const summary = await ctx.step<Summary>('python:enrich', pdf)12    return { pdf, summary }13  }14}

Transport pools and failoverCopy Anchor Link

A single instance can dispatch over an ordered pool of named transports and fail over between them; a step pins one with ctx.step(step, input, { transport: id }). In config/durable.ts, only transport is built by default (the rest of transports is a menu); list the others to build into the pool in transportPool, in failover order. Each one gets the same consumer deferral, namespace and shutdown as transport:

export default defineConfig({
  transport: 'queue',
  transportPool: ['db'], // built alongside 'queue'; the engine falls over to it on a dispatch error
  transports: {
    queue: transports.queue({ connection: 'redis' }),
    db: transports.db(),
  },
})

When you construct a WorkflowEngine yourself, pass transports: [{ id, transport }, …] instead of transport. Either way the engine dispatches each step on the first transport and fails over to the next on a dispatch error, throwing only when every one fails. A step pins one by id, tried first with the rest as failover:

// 'db' must be in the pool (transportPool, or transports: [...] on a hand-built engine)
const result = await ctx.step(heavyStep, input, { transport: 'db' })

A pin names a transport by its key in transports (default still means the primary transport). Pinning one the engine didn't build fails the step with an error that lists the transports it has and points at transportPool.

The id of the transport that accepted the dispatch is stamped on the task as task.transport, so a worker consuming several transports replies on the matching one, and the engine registers its result and heartbeat handlers on every transport of the pool. Failover is symmetric. Worker health (groupHealth) and the worker directory are merged across every transport of the pool.

Control plane (separate from the transport)Copy Anchor Link

The transport is point-to-point: it carries a task to one worker and a result back. But some things every engine instance needs regardless of who runs a given run — a dashboard-only process must live-tail a run executing on a worker, and the process actually running a run must learn it was cancelled elsewhere. That cross-instance fan-out is a second channel, the controlPlane, modelled separately from the task transport:

import { controlPlanes } from '@adonis-agora/durable'

export default defineConfig({
  transport: 'queue',
  transports,
  controlPlane: controlPlanes.redis({ connection: 'main' }),
})

Omit it and the engine is local-only: events and cancellation reach subscribers on this instance but don't fan out to other processes (fine for a single-instance app). Selecting a transport never makes it the control plane, even one that implements publishControl / onControl. Supply one and lifecycle events broadcast to every process and a cancel reaches the process running the run. See Control plane for the controlPlanes.redis driver (multi-replica, NestJS-interoperable).

Wide fans: results and timersCopy Anchor Link

A ctx.all fan of thousands of steps sends thousands of results back at once, and two knobs decide how fast an engine-side instance works through them:

• Steps. On bullmq, queue and db, concurrency sets how many steps a worker runs at once (per task queue on bullmq and queue, across the one task loop on db): a number (fixed, with no control loop), or 'adaptive' to let the worker raise and lower it from step latency, errors, memory and CPU; see the BullMQ, queue and db options.
• Results. The bullmq, queue and db transports hand each instance at least 512 results at once (resultConcurrency only raises it); see the BullMQ options and the queue options. db claims result rows 256 a round and deletes each round's in one statement. The engine settles one run's results in hand as a batch, skips the fan's barrier count while more of that fan's results are queued behind the batch, and resumes the workflow once per settled wave, whichever transport carries it. A fan's tasks go out in bulk where the transport can (dispatchMany: BullMQ addBulk, multi-row inserts on db, pushManyOn on queue), and grouping them by queue is linear in the fan's size.
• Timers. Each timer pass loads at most timerBatchSize due runs, most overdue first, and walks them in a shuffled order. It is an engine option in config/durable.ts (default 500), not a transport one; a non-finite or sub-1 value falls back to 500:

export default defineConfig({
  timerBatchSize: 1000,
  // … transport and store
})

See multiple instances for why the batch is shuffled.

Worker advertisementCopy Anchor Link

A thin worker publishes a descriptor (what it serves, including the onError routes of its workflows) and a heartbeat, each with an expiry, and the control plane reads them back for capability-aware routing and to serve workflows by convention. Where they live depends on the transport:

A worker runtime that stops cleanly removes its keys, and so do the queue and db transports for their own per-step-name liveness keys; the bullmq transport's liveness keys expire 35 s after the last beat. A worker that dies drops out when its keys expire.

Wire protocolCopy Anchor Link

Every transport carries the same JSON RemoteTask (dispatch) and StepResult (reply). The task also carries an optional transport (the pool id it was dispatched on) and an optional traceparent (see distributed tracing). The JSON is the same everywhere, but the wire around it is not: only bullmq shares its queue names and job shapes with the NestJS and Python workers. The queue and db layouts are this package's own.

Several engine instances can share one transport safely: an engine settles a result with a conditional write on the step's checkpoint that succeeds only while it is still pending (settleCheckpointsIfPending on the Lucid store), so a duplicate or redelivered result does nothing, and over one store only one engine's write lands.