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 worksCopy Anchor Link

• 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) }));

ConcurrencyCopy Anchor Link

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). The defaults are min: 1, max: 32, start: min, ramCeilingPct: 85, no CPU ceiling, and tickMs: 2000. To pin the limit instead, pass a plain number.

A number that is not finite or is out of range falls back to its default: a fixed value, min, max, start or tickMs below 1 (or NaN, Infinity) uses the default, and so does a ramCeilingPct of 0 or less. A bad cpuCeilingPct turns the CPU ceiling off. A fractional count is rounded down.

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.

Result concurrencyCopy Anchor Link

concurrency governs the task queues a worker consumes. The engine side consumes the shared <prefix>-results queue, and resultConcurrency sets how many results this instance handles at once. It defaults to 1.

Each result settles its step's checkpoint and then resumes the run inline. With one result at a time, a pod drives one run turn at a time off results, so the results of a wide fan queue up behind it. Raise resultConcurrency on the instances that consume results:

const transport = new BullMQTransport({
  connection: { host: 'localhost', port: 6379 },
  resultConcurrency: 16, // results of different runs handled in parallel
});

Results of the same run still resume it in order on an instance. Settling a result (claiming its checkpoint, checking a step fan's join barrier) runs in parallel; only the turn each result triggers waits for the previous turn of that run on this instance. A run never gets two concurrent turns from one instance, and different runs proceed in parallel. The engine applies the same ordering to any transport that delivers results concurrently, including the in-process ones.

A value that is not a finite number of at least 1 (NaN, Infinity, 0) falls back to the default of 1; a fractional value is rounded down.

Result deliveryCopy Anchor Link

A result job whose handler throws (a store error at peak load, say) is retried up to 5 times with exponential backoff starting at 500 ms. A result that still fails is kept in the results queue for 24 hours instead of being removed, and the transport logs that it gave up on it, with the step and run ids.

PriorityCopy Anchor Link

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.

NamespaceCopy Anchor Link

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 recoveryCopy Anchor Link

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.

The bridge lives in the consumer, not the queue. A worker in another language consumes the same queues, and the bridge only exists where somebody implemented it: the TypeScript transport here, and the Python durable-worker (from the release that ships its own failed listener — earlier versions had none, so a Python fleet's steps orphaned where a Node fleet's did not). If you run a worker on an SDK without it, treat the checkpoint as unsettled and rely on the engine's remoteRedispatchMs self-heal below.

Residual gaps, both of which the engine's opt-in remoteRedispatchMs self-heal covers (see Retries):

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

See also Recovering a lost remote step dispatch.

Control plane & tenant protocolCopy Anchor Link

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.

OptionsCopy Anchor Link

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.