BullMQ / Redis

The bullmq driver sends steps over BullMQ on Redis. Its wire format matches the NestJS nestjs-durable BullMQ transport and the Python durable-worker byte for byte, so it is the driver to pick when an Adonis engine and a NestJS or Python worker share one Redis. Cross-ecosystem interop covers the polyglot side. This page covers the transport itself. To weigh it against the other drivers, see the comparison.

The driver ships inside @adonis-agora/durable. bullmq and ioredis are optional peers, loaded only when bullmq is the selected transport, so install them in the app that uses it:

npm i bullmq ioredis

How it worksCopy Anchor Link

• ctx.step dispatches a task to a per-name queue, <P>-tasks-<token>. The routing token is the step name with : replaced by - (BullMQ does not allow : in a queue name), plus @<partition> when a partition other than 'default' is set.
• A worker that registered that step name consumes the queue, runs the handler, and adds the result to the shared <P>-results queue.
• The engine-side instance consumes <P>-results and checkpoints each result.
• Heartbeats from long steps (timeoutMs) travel on the <P>-heartbeat pub/sub channel. Workers advertise themselves with TTL'd Redis keys.

P is the effective prefix: prefix (default durable), with -<namespace> appended for a non-default namespace.

Run one engine-side process, which consumes results, and one or more worker processes. Each worker starts a dedicated BullMQ Worker for every step name it serves:

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

export default defineConfig({
  transport: 'bullmq',
  transports: {
    bullmq: transports.bullmq({ connection: { host: '127.0.0.1', port: 6379 } }),
  },
  // The BullMQ transport does not include a control plane: add one for multi-replica cancel + live-tail.
  controlPlane: controlPlanes.redis({ connection: 'main' }),
  store: 'lucid',
  stores: { lucid: stores.lucid() },
})

node ace durable:work

durable:work boots the app from the same config/durable.ts and serves the step handlers the app registers. Handlers under app/steps are discovered automatically.

The consumers setting in config/durable.ts works on bullmq as on the queue driver. A BullMQ Worker starts fetching as soon as it is built, so a deferred process builds none: its task, results, decisions, step-events, start-run and run-request workers are created only when durable:work calls engine.startConsumers(). With the default 'auto', a node ace console or REPL process is a pure producer — it dispatches and publishes, and never claims a job. 'never' makes every process a producer (for web fleets); 'always' opts a process in. Pub/sub subscriptions — heartbeats, run replies, tenant events — are broadcast and start regardless.

A deferred process also stamps no liveness key: a step name's <P>-worker-heartbeat:<token>:<instanceId> starts with its BullMQ worker, so a node ace process that registers app/steps never shows up as a live worker in groupHealth or the dashboard.

ConnectionCopy Anchor Link

connection is required. It is passed directly to BullMQ and ioredis, and accepts:

• ioredis ConnectionOptions (host, port, password, db, sentinels, ...),
• a redis:// or rediss:// URL — host (default 127.0.0.1), port (default 6379), URL-decoded username and password, the db from the first path segment, and TLS (tls: {}) for rediss:. Query parameters are refused, and so are other schemes and a db that isn't a non-negative integer, all at boot; pass ConnectionOptions for anything else — or
• an existing ioredis Redis instance.

An options object can also carry a url ({ url: 'redis://…', keepAlive: 10_000 }). The transport reads that URL into host, port, credentials, db and TLS the same way, so the pub/sub, heartbeat and liveness clients reach the same Redis as the queues. A key set beside the url wins over the one the URL implies, so { url: 'rediss://…', tls: { ca } } keeps its CA. ioredis on its own ignores url.

BullMQ workers need maxRetriesPerRequest: null. When you pass an options object, the transport sets that for its worker connections. A Redis instance you pass in is used as-is, so configure it yourself. Pub/sub, heartbeat and liveness clients are duplicate()s of a passed instance, or fresh clients with a 10s keep-alive.

A missing or empty connection throws at boot. Without that check, ioredis would silently connect to its own default, 127.0.0.1:6379, instead of your Redis.

ConcurrencyCopy Anchor Link

concurrency sets how many tasks each of this instance's per-name task queues runs at once (the BullMQ Worker concurrency). It accepts three forms:

• A number fixes the limit. The default is 1. A non-finite or sub-1 value falls back to 1, and a fractional value is rounded down. A fixed limit runs no control loop. The number also applies to the decisions, step-events, start-run and run-request consumers.
• 'adaptive' lets an AdaptiveController move the limit at runtime, using the defaults below.
• { mode: 'adaptive', ... } is adaptive mode with overrides.

Every tickMs, an adaptive worker compares the steps' median latency (p50) with their no-queuing baseline, and checks the memory and CPU it is using. The baseline tracks the fastest step of the window, and p50 always sits above it, so the latency gradient allows for that: it is (2 × baseline + 5 ms) / p50, capped at 1. A step of 1–2 ms that sometimes takes 5 ms reads as noise, not as queuing. The first matching signal moves the limit:

Until the limit first moves down, growth works like slow start: the worker re-decides it after each limit's worth of settled steps instead of waiting for the next tick, so a worker of fast steps reaches its working limit within milliseconds rather than after many ticks. Only growth is decided early; every move down waits for the tick. Once the limit has moved down, growth happens on the tick, by ⌊√limit⌋ (at least 1).

bullmq: transports.bullmq({
  connection: { host: '127.0.0.1', port: 6379 },
  concurrency: { mode: 'adaptive', min: 2, max: 64, start: 8, ramCeilingPct: 80, cpuCeilingPct: 90 },
}),

A non-finite or out-of-range value falls back to its default. A non-finite or non-positive cpuCeilingPct turns the CPU ceiling off.

The limit stays within [min, max]. It applies to every task queue of the instance at once, and a task queue registered later starts at the current limit. Only step durations are measured. A workflow turn on the same queue counts as in-flight work and as progress, but its duration is not added to the latency window. In adaptive mode, the decisions, step-events, start-run and run-request consumers run at 1.

Total parallelism across a fleet is concurrency × replicas. In both modes, the worker heartbeat carries the live limit, in-flight count, RSS, CPU, throughput, p95 and (in adaptive mode) the last adjustment with its reason. The dashboard's worker cards display these. Throughput is steps per minute over the trailing minute, so it decays while the worker is idle and drops to 0 after a minute with no completed step. This is the same controller as the NestJS BullMQ transport's concurrency: 'adaptive'.

The transport's concurrency is separate from a flow-control queue's concurrency. That one limits what the engine admits. This one limits how many admitted steps a worker runs at once.

Result concurrencyCopy Anchor Link

concurrency applies to the task queues a worker consumes. The engine side consumes the shared <P>-results queue, and holds at least 512 results at once per instance, or resultConcurrency when that is higher. A non-finite or sub-1 value is ignored, and a fractional value is rounded down.

The engine settles one run's results that are in hand together, so the results of a wide fan land a few hundred per checkpoint write rather than one at a time. Raise it past 512 only on instances that consume results of many runs at once:

bullmq: transports.bullmq({
  connection: { host: '127.0.0.1', port: 6379 },
  resultConcurrency: 2048, // more results in hand at once, across runs
}),

Raising it never gives a run two concurrent turns: the run lease admits one executor, and replay is positional. The fan's join barrier resumes the parent once per settled wave rather than once per result. For the engine-side timer batch (timerBatchSize) that goes with wide fans, see Wide fans: results and timers.

Result deliveryCopy Anchor Link

If the engine's result handler throws on a result job (for example, a store timeout at peak load), the job is retried up to 5 times with exponential backoff starting at 500 ms. Retrying is safe because the engine settles a step only while its checkpoint is still pending. A result that still fails is kept in the results queue for 24 hours, and the transport reports that it gave up on it (with the step and run ids) to onError, or to console.error when no onError is set.

PriorityCopy Anchor Link

A per-call priority on ctx.step follows the engine's convention: higher wins. BullMQ's job priority works the other way, with lower numbers running first. The transport converts between them (clamp(round(1_048_576 - priority), 1, 2_097_151)), so "higher means more urgent" holds end to end. The NestJS and Python sides use the same mapping. Leave priority unset to keep the default FIFO ordering.

Namespace, partition and prefixCopy Anchor Link

• prefix (default durable) is the root of every queue, channel and key name.
• namespace separates every name this instance touches, so one Redis can host several isolated deployments. Unset or 'default' keeps the names identical to the un-namespaced scheme. Any other value inserts -<namespace> after the prefix. If you don't set it here, the transport takes the engine's namespace from config/durable.ts. A value set explicitly on the transport takes precedence.
• partition adds @<partition> to every per-name task queue this instance subscribes to. Several worker pools can then serve the same step name on one Redis without their tasks mixing.

An engine and its workers must agree on all three, or they end up on different queues. See Tenancy for what each axis actually partitions.

Crash recoveryCopy Anchor Link

If a worker crashes or stalls mid-step, it never produces a StepResult, and the step's checkpoint would stay pending. The transport handles this case automatically:

• BullMQ's own stalled-job check, run every 30 s by a peer worker on the same queue (or by the crashed worker once it restarts), finds the job whose 30 s lock lapsed. The first stall puts the job back in the queue to run again; a job that stalls more than BullMQ's maxStalledCount (default 1, so on its second stall) is marked failed. With the defaults a crashed worker's step is re-delivered in about 30–60 s.
• Every task worker listens for failed. When a task job ends failed, the listener publishes a synthetic, retryable failed StepResult for it on <P>-results. The engine then settles the checkpoint as failed, and its normal durable retry re-dispatches the step.
• Task jobs are enqueued with removeOnFail: { age: 86400 }, so a failed job's payload stays in Redis for 24 hours. That gives the bridge time to read the runId / seq / stepId it needs before BullMQ removes the job.

A handler's own error does not go through this path. The worker catches every handler throw and publishes a failed StepResult itself, so the BullMQ job succeeds and nothing is published twice. A failed workflow-turn job carries no step identity, so the bridge skips it. The engine re-drives that turn on recovery instead.

The bridge lives in the consumer, not in the queue. A worker written in another SDK consumes the same queues, and only gets this behaviour if that SDK implements its own failed listener.

Two gaps remain. The engine's opt-in remoteRedispatchMs self-heal covers both (see Lost remote dispatch):

• If the only worker for a queue crashes and never restarts, no peer's stalled check runs, so nothing settles the checkpoint.
• If the consuming SDK has no failure bridge, the job's final failure is never turned into a result.

engine.redispatchPending(runId) is the manual fix for a run that is already stuck.

Wedged steps (stepTimeoutMs)Copy Anchor Link

A third gap: a handler that never returns. An await that never settles (a dead connection swallowed by a wrapper stream, an HTTP call with no timeout) holds its BullMQ job for good. BullMQ renews the job lock on a timer, whether or not the handler makes progress, so the job never stalls and no peer reclaims it. The run waits until someone kills the worker.

stepTimeoutMs puts a wall-clock ceiling on each step handler a worker runs. A step still running at the ceiling publishes a retryable failed StepResult, so the engine's durable retry re-dispatches it, and the transport stops waiting on the handler. The handler's promise cannot be cancelled from outside: it is abandoned, and anything it does later is ignored.

bullmq: transports.bullmq({
  connection: { host: '127.0.0.1', port: 6379 },
  stepTimeoutMs: 15 * 60_000, // well above the slowest legitimate step
}),

It is off by default. Pick a value comfortably above your slowest legitimate step: a step that is slow but healthy and crosses the ceiling is retried, and its first attempt keeps running in the background. Workflow turns are not covered. This is a per-worker safety net; the per-step timeoutMs on ctx.step is the engine-side limit, enforced through heartbeats.

Worker livenessCopy Anchor Link

Each served routing token gets a liveness key, <P>-worker-heartbeat:<token>:<instanceId>. Its value is { ts, status }, where status is the live concurrency snapshot. Every token is refreshed on one shared 10s interval, and the key has a 35s TTL. A worker that stops refreshing disappears from the registry when its key expires.

Workers that run the store-less worker runtime also publish a handshake descriptor under <P>-worker-descriptor:<token>:<instanceId>. The engine reads these keys with SCAN (never KEYS) and caches descriptor reads for 5s per token, so routing does not scan Redis on every dispatch. Per-group health, meaning queue depth (waiting + active + delayed + prioritized) plus live workers, comes from the same keys.

instanceId defaults to ts-<hostname>-<pid>. The Python SDK uses py-<hostname>-<pid>, so the fleet view shows which runtime each worker runs.

The transport also reads both keyspaces fleet-wide: readAllWorkerDescriptors() and readAllWorkerHeartbeats() scan every token at once. They back engine.workflowDirectory(), which lists every workflow the live fleet can run.

Subscriber health (pingIntervalMs)Copy Anchor Link

The heartbeat, run-reply and tenant-event channels are Redis pub/sub subscriptions, each on its own connection. A subscriber connection never writes on its own. When a VPN, a NAT gateway or an idle timeout drops its TCP connection without a close, nothing fails: ioredis still reports the connection as ready, while Redis has already dropped the subscription (PUBSUB NUMSUB reads 0). Long-step heartbeats stop resetting step leases, and a tenant pod's run-gateway calls time out, until the process restarts.

The transport runs one shared watchdog that PINGs every subscriber connection each pingIntervalMs (default 30_000). PING is allowed in subscriber mode. A ping that fails, or takes longer than the smaller of 5s and the interval, marks the connection dead: the transport disconnects it, ioredis reconnects, and ioredis's autoResubscribe restores its channels. A connection that is already reconnecting is skipped. 0 or false turns the watchdog off; a negative, non-finite or out-of-range value falls back to the default.

Each subscriber connection also gets an error listener. The first error of each reconnect burst goes to onError, or to console.warn without one, instead of surfacing as an unhandled error event.

A real close (a Redis restart, CLIENT KILL) needs no watchdog: ioredis reconnects and resubscribes by itself. The watchdog is for the silent drop. A RedisControlPlane runs the same watchdog on its control channel, in both its raw-ioredis and controlPlanes.redis() forms (see Control plane).

Beyond stepsCopy Anchor Link

The same transport also carries:

• Workflow turns for workflows whose body runs in another SDK or a worker runtime. A workflow job goes on the workflow's own <P>-tasks-<token> queue (workers tell step jobs and workflow jobs apart by shape), and the worker's decision returns on <P>-decisions. Every engine replica consumes that queue, and whichever takes a decision applies it by run id, only while the run still awaits that turn's taskId (a stale or duplicate decision is dropped). Local-step lifecycle events from those workers come back on <P>-step-events. See Python workers & workflows and Remote workflow turns.
• The store-less tenant protocol. A tenant pod enqueues start requests on <P>-start-run and read/control requests on <P>-run-request. The control plane replies on the <P>-run-reply pub/sub channel and re-publishes each tenant's lifecycle events on <P>-tenant-events-<tenant>.

How the wire lines up lists every channel next to its NestJS and Python counterpart.

ShutdownCopy Anchor Link

closeTimeoutMs (default 5000 ms) limits how long close() waits for a graceful shutdown. 0 skips the wait. Workers close first, concurrently, so a step that is still running can publish its result. The queues close after that. Both phases share one deadline. When the deadline passes, the transport disconnects its pub/sub, heartbeat and liveness connections anyway. Abandoned BullMQ workers may keep running their jobs on their own connections until the process exits. A job that did not finish is recovered through BullMQ's stalled-job handling and the crash-recovery bridge. close() does not delete the process's liveness keys: the worker stays listed for up to 35 s after it stops. The provider calls close() on app shutdown.

Options — transports.bullmq(config)Copy Anchor Link

transports.bullmq(config) returns the lazy driver thunk used in config/durable.ts. The underlying BullMQTransport class (with createBullMQDeps) is also exported from @adonis-agora/durable if you want to build one by hand.

Queue namingCopy Anchor Link

The payloads are the documented RemoteTask / StepResult JSON, with the DTO itself as the job data and no envelope. This is why a non-Node worker on the same queues works with this transport.