Flow control

Steps fan out work to workers, and sometimes you need to throttle or prioritize that fan-out: a third-party API allows only N requests per minute, a downstream service falls over above some concurrency, an urgent job shouldn't wait behind a backlog of routine ones, or one noisy tenant shouldn't starve the rest. nestjs-durable gives you four independent levers for this:

• Queues — register a named cap (concurrency and/or rateLimit) that ctx.step calls opt into.
• Per-call admission hints — priority and fairnessKey on the ctx.step(..., opts) call, which decide who gets the next slot when a queue is contended.
• Worker/transport concurrency — how many dispatched tasks this process runs at once, fixed or self-tuning (adaptive).
• Admission scope — in-process (per engine instance) by default, or fleet-wide across every replica via RedisAdmissionBackend.

All four are durable in the sense that matters here: a call that can't be admitted doesn't busy-wait or hold the run in memory — it re-suspends, and the timer poller retries admission later, so the limit survives a crash.

1@Workflow({ name: 'send-receipt', version: '1' })2export class SendReceiptWorkflow {3  constructor(private readonly notify: NotificationsWorker) {}4 5  async run(ctx: WorkflowCtx, job: EmailJob) {6    const sent = await ctx.step(this.notify.sendEmail, job, {7      queue: 'emails',8      priority: job.urgent ? 10 : 0,9      fairnessKey: job.tenantId,10    });11    return { messageId: sent.messageId };12  }13}

The walkthrough above dispatches one step through a queue with both priority and fairnessKey set. When the queue has a free slot, the call is admitted immediately. When it's contended, the admission controller picks the rightful next waiter in this order: higher priority first, then (for a queue configured with fairness: 'key') the least-recently-served fairnessKey, then arrival order as the final tiebreak — the same comparison both the built-in in-process controller and RedisAdmissionBackend implement. Nothing is held in memory while a call waits: the run just re-suspends with a retry time and comes back later.

And here it is as a live system — dots are runs dispatching steps through a queue into a fixed pool of worker slots. Hit burst and watch backpressure work:

Registering a queue

Register a queue on the engine with engine.registerQueue, or declare it on the NestJS module's queues option (which registers it at startup). A QueueConfig has:

• name — the queue's name, referenced from ctx.step(handler, input, { queue: name }).
• concurrency — the maximum number of steps in flight at once for this queue, on this engine instance. Omit for unlimited.
• rateLimit — a fixed-window rate limit of { limit, periodMs }: at most limit admissions per periodMs. Checked before concurrency, and unaffected by priority/fairness. Omit for unlimited.
• retryMs — how long (ms) a blocked call waits before re-checking for a free slot. Defaults to 1000.
• fairness — 'key' round-robins admission across distinct fairnessKeys so one key can't monopolize the budget; 'fifo' (default) is plain first-come.
• order — the arrival tiebreak among otherwise-equal waiters (same priority, same fairness rank): 'fifo' (default, earliest first) or 'lifo' (most recent first, a stack).

DurableModule.forRoot({
  store,
  transport,
  queues: [
    // Rate-limit a third-party API to 100 calls/minute:
    { name: 'shipping-api', rateLimit: { limit: 100, periodMs: 60_000 } },
    // Cap a heavy downstream to 5 concurrent steps, re-checking every 500ms when full:
    { name: 'pdf-render', concurrency: 5, retryMs: 500 },
    // Both at once, PLUS fair admission across tenants under contention:
    {
      name: 'emails',
      concurrency: 10,
      rateLimit: { limit: 1000, periodMs: 3_600_000 },
      fairness: 'key',
    },
  ],
});

The equivalent imperative form, when you hold the engine directly:

engine.registerQueue({ name: 'emails', concurrency: 10, rateLimit: { limit: 1000, periodMs: 3_600_000 } });

Registering the same name again replaces the queue's config.

Concurrency caps

concurrency bounds how many admitted-but-not-yet-settled steps a queue allows at once, on this engine instance. A slot is held from the moment a call is admitted until the step's result lands (or the run is cancelled) — so accounting tracks real in-flight work, not just dispatch — and released for the next blocked waiter to claim on its next poll.

{ name: 'pdf-render', concurrency: 5, retryMs: 500 }

Rate limits

rateLimit is a fixed window: at most limit admissions per periodMs. It's checked first, ahead of concurrency and any priority/fairness ordering — a queue that's rate-limited blocks every call equally regardless of priority, until the window rolls over.

{ name: 'shipping-api', rateLimit: { limit: 100, periodMs: 60_000 } }

Watch the window budget drain and refill — a burst rides the windows through, limit at a time:

Using a queue from a workflow

Subject a step dispatch to a queue by naming it in the call options — and, optionally, hint how it should be admitted when the queue is contended:

@Injectable()
export class NotificationsWorker {
  @Step({
    name: 'notify.email',
    input: z.object({ to: z.string().email(), template: z.string() }),
    output: z.object({ messageId: z.string() }),
  })
  async sendEmail(input: { to: string; template: string }): Promise<{ messageId: string }> {
    // ...send the message...
    return { messageId: await this.mailer.send(input) };
  }
}

// in the workflow — admitted through the `emails` queue, urgent jobs jump the line,
// and each tenant gets a fair share of the slot when the queue is full:
const sent = await ctx.step(this.notify.sendEmail, job, {
  queue: 'emails',
  priority: job.urgent ? 10 : 0,
  fairnessKey: job.tenantId,
});

Priority

priority on the ctx.step call opts (StepDispatchOpts.priority) is admission priority within a queue: higher is admitted first when a slot is contended (default 0); it has no effect without a queue. The same value also rides along as broker priority: for a transport that supports job priority (BullMQ), it's carried onto the dispatched RemoteTask and translated into a BullMQ job priority, so an urgent task can jump ahead of already-enqueued lower-priority ones at the worker's own queue too — not just at the flow-control admission gate. Transports without broker priority support silently ignore it.

await ctx.step(this.notify.sendEmail, job, { queue: 'emails', priority: job.urgent ? 10 : 0 });

Fairness

fairnessKey on the call opts is the fairness bucket — e.g. a tenant id — used when the queue is registered with fairness: 'key'. The queue round-robins across distinct keys (the least-recently-served key wins the next contended slot), so one busy key can't monopolize the budget. It defaults to the run id when omitted, so unrelated runs are still fair to each other even if you never set it explicitly. Priority always wins over fairness; fairness only breaks ties within the same priority tier, and arrival order (order: 'fifo' | 'lifo') breaks whatever's left.

// queue config: opt into fairness
{ name: 'emails', concurrency: 10, fairness: 'key' }

// call site: the fairness bucket
await ctx.step(this.notify.sendEmail, job, { queue: 'emails', fairnessKey: job.tenantId });

How admission works (and why it's durable)

When a queued call is reached, the engine asks the queue's admission backend to admit one unit of work. If the queue is at its concurrency limit or has exhausted its rate-limit window, admission is blocked and the backend returns the epoch-ms time at which admission may next succeed. The engine does not dispatch: it re-suspends the run with that retry time as its wakeAt. The durable-timer poller — the same one that wakes suspended runs whose timers come due — retries admission when the time arrives, and if it's still blocked, the run simply re-suspends again.

Because the retry time is persisted on the run and the poller drives it, the limit is durable: a process that crashes while runs are parked waiting for a slot loses nothing — the parked runs come back when their timers are due. No run is ever held in memory waiting for a slot.

Worker / transport concurrency

Flow-control queues throttle admission into a dispatch; worker concurrency throttles how many already-dispatched tasks this process executes at once. It's a transport-level option — on BullMQTransport, concurrency (a ConcurrencyOption) is the BullMQ Worker's own concurrency for the tasks it consumes:

const transport = new BullMQTransport({
  connection: { host: 'localhost', port: 6379 },
  concurrency: 8, // this pod runs up to 8 tasks in parallel; defaults to 1
});

Total parallelism across a fleet is concurrency × replicas. Pass 'adaptive' instead of a fixed number to let a self-tuning controller pick the limit at runtime instead of you guessing it:

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

The adaptive controller (AdaptiveController in @dudousxd/durable-worker) runs an AIMD gradient loop on a tickMs interval (default 2000): it grows the limit when there's headroom, shrinks it when latency shows queuing or errors spike (backpressure), and hard-brakes on RAM (ramCeilingPct, default 85%) or, if configured, CPU (cpuCeilingPct). Every adjustment is recorded as a WorkerAdjust — { at, from, to, reason } — so a dashboard can show why the limit moved, not just that it did. Tunables:

• min / max — the floor and ceiling the controller won't cross. Defaults 1 / 32.
• start — the initial limit, clamped into [min, max]. Defaults to min.
• ramCeilingPct — RSS percent of the memory ceiling that triggers the hard brake. Default 85.
• cpuCeilingPct — CPU percent ceiling; off (undefined) by default.
• tickMs — control-loop period in ms. Default 2000.

Either way — fixed or adaptive — the live concurrency, in-flight count, and (in adaptive mode) the last adjustment ride the worker's heartbeat as a WorkerStatus payload, which the Telescope Workers dashboard renders per worker group so you can see saturation and self-tuning behavior without instrumenting anything yourself.

Local vs. global admission

By default, flow-control accounting is per engine instance (in-process) — the DBOS workerConcurrency tier. For the common single-orchestrator deployment this is exactly what you want, correct without any cross-process coordination. If you run several orchestrator replicas, each enforces the limit independently, so a concurrency: 5 queue admits up to 5 per replica, not 5 fleet-wide.

For a true cross-instance cap, pass a RedisAdmissionBackend (from @dudousxd/nestjs-durable-admission-redis) as the admission option — it moves concurrency, rate-limit, and priority/fairness ordering into Redis so every replica shares the same accounting:

import { RedisAdmissionBackend } from '@dudousxd/nestjs-durable-admission-redis';

DurableModule.forRoot({
  store,
  transport,
  queues: [{ name: 'emails', concurrency: 10, rateLimit: { limit: 1000, periodMs: 3_600_000 } }],
  admission: new RedisAdmissionBackend({
    connection: { host: 'localhost', port: 6379 },
  }),
});

Concurrency slots are tracked per owning instance and reclaimed only once that instance's liveness heartbeat lapses (instanceTtlMs, default 30s) — so a live pod keeps its slot for the full step duration no matter how long it runs, while a crashed pod's slots free within the TTL window. Rate limiting is a Redis fixed-window counter, and priority/fairness ordering runs through the same atomic Lua acquire the in-process controller mirrors in logic, so behavior is identical — just fleet-wide instead of per-replica. This targets a single (non-cluster) Redis instance; size limits with that in mind.

Which knob when