BullMQ / Redis

@dudousxd/nestjs-durable-transport-bullmq carries steps over BullMQ/Redis — the path for steps that run in another process or another language (e.g. a Python worker).

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

How it works

• ctx.step dispatches a task to a per-name queue: <prefix>-tasks-<name>[@<partition>].
• A worker registered for that step name consumes it, runs the handler, and adds the result to the shared <prefix>-results queue.
• The engine-side instance consumes results and checkpoints them.

Run one instance engine-side (consumes results) and one per worker process (registers its handlers, and dispatches on the transport):

const transport = new BullMQTransport({ connection: { host: 'localhost', port: 6379 } });

DurableModule.forRoot({ store, transport });

const transport = new BullMQTransport({
  connection: { host: 'localhost', port: 6379 },
  partition: 'payments', // optional isolation suffix — omit for a single-tenant deployment
});
transport.handle('payments.charge-card', async (input) => ({ chargeId: await charge(input) }));

Concurrency

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

• a fixed number — always runs exactly that many in parallel. Defaults to 1 (one task at a time).
• 'adaptive' — an AdaptiveController self-tunes the limit from a latency gradient, a RAM brake, and backpressure (errors/stalls), instead of you guessing a number.
• { mode: 'adaptive', ... } — adaptive with overrides (min, max, start, ramCeilingPct, cpuCeilingPct, tickMs).

Watch an adaptive worker breathe — spike the load and it grows into its headroom; fire the RAM brake and it backs off before memory melts:

const transport = new BullMQTransport({
  connection: { host: 'localhost', port: 6379 },
  concurrency: { mode: 'adaptive', min: 2, max: 32, ramCeilingPct: 85 },
});

Total parallelism across a fleet is concurrency × replicas. Either mode publishes its live status — current limit, in-flight count, and (adaptive) the last adjustment — on the worker's heartbeat as a WorkerStatus. See Flow control for the full set of tunables and how the Telescope Workers dashboard renders it.

Priority

Per-call priority on ctx.step follows the engine's convention: higher wins. BullMQ's own job priority is the opposite — lower numbers run first — so the transport translates one into the other before enqueuing, keeping "higher = more urgent" true end-to-end regardless of transport. Leave priority unset for the default FIFO ordering.

Namespace

namespace segments every queue/stream/key this instance touches, so a single Redis can host several isolated deployments — e.g. one per developer — without crosstalk. Leave it unset (or "default") for production: names stay byte-identical to the un-namespaced scheme. Any other value inserts a -<namespace> segment into every name. See Tenancy for the operator/tenant model this enables.

Crash recovery

A worker that crashes or stalls mid-step never produces a StepResult, so without help the step's checkpoint would stay pending forever. BullMQTransport closes that gap automatically: BullMQ's own stalled-check (run by a peer worker's Worker) marks a crashed/stalled task job failed, and a Worker.on('failed') listener bridges that into a synthetic failed StepResult — so the engine settles the checkpoint as failed and its normal durable retry re-dispatches the step. A handler business error is unaffected: runStepHandler already turns it into a successful job carrying a failed StepResult, so there's no double-publish.

No configuration needed on your part — dispatch() sets removeOnFail: { age: 24h } on every task job (overriding the default immediate removeOnFail) so a failed job's payload survives long enough for the bridge to read the runId/seq/stepId it needs, before BullMQ garbage-collects it.

Residual gap: if the only worker for a queue crashes and never restarts, no peer's stalled-check ever runs, so nothing settles the checkpoint — that needs an external liveness monitor, or the engine's opt-in remoteRedispatchMs self-heal. See Recovering a lost remote step dispatch.

Control plane & tenant protocol

Beyond Transport, BullMQTransport also implements ControlPlane — cross-instance broadcast for lifecycle events and cancellation — and carries the tenant start-run / run-request / run-reply / tenant-event protocol a store-less tenant worker uses to proxy runs through an operator. See Tenancy for the full model.

Options

The payload is the documented RemoteTask / StepResult JSON, so a non-Node worker on the same queues interoperates — that's exactly how the Python worker plugs in.