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.

Registering a queue

Register a queue on the engine with engine.registerQueue. A QueueConfig has:

• 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.
• fairness — 'fifo' (default) or 'key' (round-robin by the call's fairnessKey).
• order — 'fifo' (default) or 'lifo' arrival ordering among same-priority waiters.

// 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' } })

Fixed vs sliding window

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 workflow

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 },
)

A higher priority is admitted first; fairnessKey round-robins between keys when the queue's fairness is 'key', so one noisy tenant can't starve the rest.

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)

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 (or the run is cancelled), at which point the slot is released and the next blocked run can be admitted.

Scope: per engine instance

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). For a truly global cap, use the Redis backend below.

Distributed admission — Redis

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 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.