Concurrency quotas

A multi-tenant app wants one tenant unable to occupy every worker: "a company can have at most 8 chat turns executing at once". That is a quota on starting runs, not a mutex — the ninth start should be refused right away ("busy, try again in a moment", an HTTP 429), not parked behind the other eight. concurrency is exactly that.

Declaring a quotaCopy Anchor Link

On the workflow, deriving the key from the input:

import { BaseWorkflow } from '@adonis-agora/durable'
import type { WorkflowCtx } from '@adonis-agora/durable'

export default class ChatTurnWorkflow extends BaseWorkflow {
  static workflow = {
    name: 'chat.turn',
    version: '1',
    concurrency: {
      key: (input: unknown) => `tenant:${(input as TurnInput).tenantId}`,
      limit: 8,
    },
  }

  async run(ctx: WorkflowCtx, input: TurnInput) {
    /* ... */
  }
}

The same concurrency option is accepted by the low-level engine.register(name, version, fn, { concurrency }) and by engine.registerRemote(...).

Or per start — a concurrency in the start options overrides the workflow's config for that start:

await ChatTurnWorkflow.dispatch(input, {
  concurrency: { key: `tenant:${tenantId}`, limit: plan.maxActiveTurns },
})

// or, lower level:
await engine.start('chat.turn', input, runId, {
  concurrency: { key: `tenant:${tenantId}`, limit: plan.maxActiveTurns },
})

A start that would exceed the quota throws ConcurrencyLimitError (with workflow, key, limit and active) and creates nothing — map it to a 429:

import { ConcurrencyLimitError } from '@adonis-agora/durable'

try {
  await ChatTurnWorkflow.dispatch(input)
} catch (error) {
  if (error instanceof ConcurrencyLimitError) {
    // error.workflow, error.key, error.limit, error.active
    return response.tooManyRequests({ message: 'Busy, retry shortly' })
  }
  throw error
}

OptionsCopy Anchor Link

• key — the quota bucket. On the workflow it's a function of the input; return undefined to leave that start unlimited. Per start it's a plain string. The key is global, not per workflow: every workflow started with tenant:acme draws on the same quota, so one limit can cover all the kinds of work a tenant runs.
• limit — max runs of the key in flight. On the workflow it may be a (sync or async) function of the key, for per-plan limits: limit: (key) => plans.maxActive(key). Per start it's a number. limit <= 0 rejects every start.
• countStatuses — which statuses occupy a slot. Default: every non-terminal status (pending, running, suspended, blocked), exported as DEFAULT_CONCURRENCY_STATUSES. Narrow it to ['pending', 'running'] when runs parked on a human (suspended on a signal) shouldn't hold a slot.

How it's enforcedCopy Anchor Link

The run carries the tag concurrency:<key>, and start counts the runs with that tag in countStatuses before creating the new one. The count goes through the store's countRuns — one COUNT(*) on the Lucid store, also implemented by the in-memory store; a custom store without countRuns falls back to the length of a listRuns listing. The store is the only source of truth, so there is nothing to leak or reconcile when a process crashes: a slot frees the moment its run leaves the counted statuses.

• An idempotent re-start of an existing runId returns that run and is never rejected — the existing-run check happens before the quota check.
• The count spans every namespace that shares the store, like the key itself. With scopeReads on, the store's reads are fenced to the engine's namespace, so the quota only counts that namespace's runs. Put the tenant in the key when the limit is per tenant.
• A per-start concurrency replaces the workflow's quota for that start entirely (key, limit and countStatuses); it is not merged with it.
• The quota applies wherever the workflow is started: engine.start / Wf.dispatch / Wf.start, child workflows (ctx.child / ctx.startChild — a rejected child start is delivered to the parent as a failed child), event triggers (a rejected subscriber is skipped) and schedules.
• Persisted schedules take a concurrency quota of their own; an over-limit window is skipped and recorded as the schedule's lastError.

Quota, singleton, or flow control?Copy Anchor Link