Flow control

Dispatched steps fan out work to workers, and sometimes you need to throttle that fan-out: a third-party API allows only N requests per minute, or a downstream service falls over above some concurrency. Flow-control queues cap how much work ctx.step admits at once — a concurrency limit, a rate limit (fixed or sliding window), or both — and, crucially, they do it durably: a call that can't be admitted doesn't busy-wait or hold the run in memory, it re-suspends and is retried by the timer poller, so the limit survives a crash.

There are four independent levers:

• Queues — 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 with admissions.redis.

Dots below are runs dispatching steps through a queue into a fixed pool of worker slots — hit burst and watch backpressure work:

Registering a queueCopy Anchor Link

Declare queues in config/durable.ts with queues, or register them on the engine with engine.registerQueue. Both take the same QueueConfig:

• name — the queue's name, referenced from ctx.step(step, input, { queue: name }).
• concurrency — the maximum number of steps in flight at once for this queue. Omit for unlimited.
• rateLimit — { limit, periodMs, algorithm? }: at most limit admissions per periodMs. algorithm is 'fixed' (default) or 'sliding' — see below. Omit for unlimited.
• retryMs — how long (ms) a concurrency-blocked call waits before re-checking for a free slot. Defaults to 1000. (A rate-limited call is told to come back when the window allows, not after retryMs.)
• fairness — 'fifo' (default) or 'key' (round-robin by the call's fairnessKey).
• order — 'fifo' (default) or 'lifo' arrival ordering among same-priority waiters.

Registering the same name again replaces the queue's config (on the in-process backend, with a fresh controller). A call naming a queue that was never registered is not gated at all.

// Rate-limit a third-party API to 100 calls/minute:
engine.registerQueue({ name: 'shipping-api', rateLimit: { limit: 100, periodMs: 60_000 } })

// Cap a heavy downstream to 5 concurrent steps, re-checking every 500ms when full:
engine.registerQueue({ name: 'pdf-render', concurrency: 5, retryMs: 500 })

// Both at once: at most 10 in flight AND no more than 1000/hour:
engine.registerQueue({ name: 'emails', concurrency: 10, rateLimit: { limit: 1000, periodMs: 3_600_000 } })

// A third-party cap enforced as a rolling window — never more than 100 in ANY 60s span:
engine.registerQueue({ name: 'openai', rateLimit: { limit: 100, periodMs: 60_000, algorithm: 'sliding' } })

The declarative form registers the same queues at boot, on every process that builds the engine:

export default defineConfig({
  queues: [
    { name: 'shipping-api', rateLimit: { limit: 100, periodMs: 60_000 } },
    { name: 'pdf-render', concurrency: 5, retryMs: 500 },
  ],
})

Two queues entries with the same name throw at boot. A later engine.registerQueue of a name already in queues replaces it, like any re-registration.

Concurrency capsCopy Anchor Link

concurrency bounds how many admitted-but-not-yet-settled steps a queue allows at once, on this engine instance (fleet-wide with the Redis backend). Accounting tracks real in-flight work, not just dispatch — see how admission works for when a slot is taken and released.

Rate limitsCopy Anchor Link

rateLimit is a hard window: at most limit admissions per periodMs. It is checked first, ahead of concurrency and of any priority/fairness ordering — a rate-limited queue blocks every call equally, whatever its priority, until the window allows another admission.

Fixed vs sliding windowCopy Anchor Link

algorithm: 'fixed' (the default, and the historical behavior) resets a counter each period — simple, but a burst straddling a window boundary can admit up to 2× limit in a periodMs span (100 at the end of one window, 100 more the instant the next opens). algorithm: 'sliding' counts admissions over a rolling window instead: an admission is denied while limit admissions already happened in the trailing periodMs, and a blocked call is told to retry at the precise instant the oldest one ages out. That steady-state smoothness is exactly what a third-party API cap cares about — the provider is counting a rolling window too. Both algorithms work on the in-process controller and on the Redis backend (where the rolling count runs inside the same atomic Lua script).

Using a queue from a workflowCopy Anchor Link

Subject a dispatched step to a queue by naming it in the call options:

export const sendEmail = defineStep(
  'notify:email',
  async (input: { to: string; template: string }) => ({ messageId: await mailer.send(input) }),
  {
    input: z.object({ to: z.string().email(), template: z.string() }),
    output: z.object({ messageId: z.string() }),
  },
)

// in the workflow — admitted through the `emails` queue, with a priority and fairness key:
const sent = await ctx.step(
  sendEmail,
  { to: order.email, template: 'receipt' },
  { queue: 'emails', priority: 10, fairnessKey: order.tenantId },
)

When the queue has a free slot, the call is admitted immediately. When it is contended, the admission controller picks the rightful next waiter in this order: higher priority first, then (for a queue registered with fairness: 'key') the least-recently-served fairnessKey, then arrival order (order: 'fifo' | 'lifo') as the final tiebreak — the same comparison the in-process controller and the Redis backend implement.

PriorityCopy Anchor Link

priority on the call opts is admission priority within a queue: higher is admitted first when a slot is contended (absent = unprioritised). The same value also rides along as broker priority: it is carried onto the dispatched RemoteTask, and the bullmq transport translates it into a BullMQ job priority (BullMQ runs the lowest number first, so the engine maps "higher wins" onto it), so an urgent task can also jump ahead of already-enqueued lower-priority ones at the worker's own queue. The queue transport maps it onto its adapter's job priority, and the db transport stores it on the task row and claims higher priority first. Transports without broker priority ignore it.

FairnessCopy Anchor Link

fairnessKey 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. Priority always wins over fairness; fairness only breaks ties within a priority tier, and arrival order breaks whatever is left.

// queue config: opt into fairness
engine.registerQueue({ name: 'emails', concurrency: 10, fairness: 'key' })

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

1@Workflow({ name: 'send-receipt', version: '1' })2export default class SendReceiptWorkflow {3  constructor(private notify: NotifySteps) {}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}

How admission works (and why it's durable)Copy Anchor Link

When a queued call is reached, the engine asks the queue's controller 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 controller returns the epoch-ms time at which admission may next succeed (retryAt). The engine does not dispatch: it re-suspends the run with that retry time as its wakeAt. The durable-timer poller, which already 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.

A slot is held from the moment a call is admitted until the step's result lands, at which point the slot is released and the next blocked run can be admitted. The queue name is stored on the step's checkpoint, so whichever instance consumes the result releases the slot — the dispatching pod does not have to be alive. A late result for a run that was cancelled (or has already failed) still releases its slot. Two consequences:

• A durable retry re-admits. A step whose attempt failed gives its slot back when that failed result lands, and goes through admission again when its backoff is up, so a step waiting out its backoff holds nothing. A timeoutMs step is the exception: it is admitted once and keeps its slot across its timeout re-dispatches, until the await on the dispatching instance ends.
• A lost job keeps its slot. A step whose result never arrives holds its slot until the step is re-driven and settles (remoteRedispatchMs, or engine.redispatchPending(runId) — see Lost remote dispatch), or until the holder goes away: a restart of the process, for the in-process backend, or a lapsed instance heartbeat, for Redis.

A blocked run normally retries at its retryAt. With the Redis backend, a released slot is also announced over Redis pub/sub, so each instance re-tries its blocked runs on that queue right away instead of waiting for the tick.

Capping one fan: ctx.all concurrencyCopy Anchor Link

A queue caps a step across every run that names it. To cap a single fan instead, pass concurrency to ctx.all:

// At most 500 processLine steps in flight for this run; the rest wait their wave.
const results = await ctx.all(processLine, lines, { concurrency: 500 })

The fan goes out in waves. Over a step, a wave is concurrency steps and the next is sent once every step of the previous one has settled. Over a child workflow, two waves of floor(concurrency / 2) children are in flight and the next goes out as soon as one settles. Either way each settled wave costs the parent one turn, not one turn per result, and the cap holds across instances because the waves come from the parent's checkpoints, not from in-memory slots.

A step fan does not accept a queue: ctx.all(step, inputs, { queue }) throws. When a step must also respect a fleet-wide rate limit, enforce it in the handler, or fan out over a child workflow whose ctx.step names the queue.

Scope: per engine instanceCopy Anchor Link

By default, flow-control accounting is per engine instance. For the common single-orchestrator deployment this is exactly what you want, and it's correct without any cross-process coordination. If you run several worker replicas, each enforces the limit independently (so a concurrency: 5 queue admits up to 5 per replica). The per-replica count is also approximate: a result consumed by another replica releases the slot on that replica's counter, not on the one that admitted it. For a truly global (and exact) cap, use the Redis backend below.

Worker / transport concurrencyCopy Anchor Link

A queue's concurrency caps what the engine admits. How many already-dispatched tasks a worker executes at once is a separate, transport-level knob — transports.bullmq({ concurrency }), transports.queue({ concurrency }) or transports.db({ concurrency }):

bullmq: transports.bullmq({
  connection: { host: '127.0.0.1', port: 6379 },
  concurrency: 8, // this pod runs up to 8 tasks per task queue in parallel; defaults to 1
}),

Total parallelism across a fleet is concurrency × replicas. Pass 'adaptive' (or { mode: 'adaptive', min, max, start, tickMs, ramCeilingPct, cpuCeilingPct }) instead of a number to let a control loop move the limit at runtime (defaults: min 1, max 32, start = min, tickMs 2000, ramCeilingPct 85, cpuCeilingPct off): it grows when latency shows headroom, shrinks when latency shows queuing or errors/stalls spike, and hard-brakes on RAM (and on CPU when cpuCeilingPct is set). Either way the live limit, in-flight count, RSS, CPU, throughput, p95 and the last adjustment with its reason ride the worker heartbeat, and the dashboard's worker cards show them. A Python worker has its own equivalent — see Python: concurrency and memory.

Distributed admission — RedisCopy Anchor Link

The @adonis-agora/durable/admission-redis subpath replaces the per-instance admission accounting with a Redis-backed one, so a concurrency or rateLimit cap holds across every engine instance. It's the tier you reach for when several worker pods must share one global limit. It ships in the main @adonis-agora/durable package; install @adonisjs/redis (or ioredis) as the extra peer.

npm i @adonisjs/redis

Select it by name in config/durable.ts with the admissions factory — a lazy thunk, so @adonisjs/redis is imported only when this backend is actually the one in use:

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

export default defineConfig({
  // … plus your transport/store (see Getting Started)
  admission: admissions.redis({ connection: 'main' }),
})

connection names an @adonisjs/redis connection from config/redis.ts, exactly like controlPlanes.redis(...) and stores.lucid(...) — the host and credentials live in one place. The queues themselves are still declared with queues or engine.registerQueue(...); the backend only changes where the accounting happens.

Need a backend of your own — a database counter, a different broker? Implement AdmissionBackend (register, handles, tryAdmit, release, and optionally onFreed to wake blocked runs early instead of making them wait out a retry tick), prove it against runAdmissionBackendContract from the conformance kit, and pass the instance straight through:

export default defineConfig({
  admission: new MyAdmissionBackend(Date.now),
})

Either form works: admission accepts a ready backend or a lazy thunk that builds one at boot.

A held slot is reclaimed only once its owner's heartbeat lapses: a live pod keeps its slot for the full step duration no matter how long it runs, and a crashed pod's slots free within instanceTtlMs. A waiter re-registers on every retry, so a still-trying call never expires; one that gave up (its run was cancelled) is pruned after waiterTtlMs so it can't sit as a phantom best-waiter and deadlock the rest.

The Lua script touches several keys in one call, so the backend targets a single (non-cluster) Redis.

Which knob whenCopy Anchor Link