Declarative flow control

debounce, singleton, concurrency, throttle, rateLimit, batch, priority, timeouts and onFailure — declared on the workflow, enforced before the run exists.

Everything on this page is declared on workflow({...}) and enforced at the admission boundary — the moment ctx.workflows.start(...) is called, before a durable run exists.

That timing is the whole point. Once a run is enqueued, the only tools left are cancel and sleep, and neither of them un-spends the durable entity. So "run this at most once per row per fifteen minutes" cannot be a primitive you call inside the body; it has to be a property of the declaration.

Not the same as Flow control. That page covers durableQueue / processQueue / rateLimit — primitives you call inside a running workflow to bound the work it fans out. This page is about whether the run starts at all. They compose: a workflow can declare concurrency here and still use a durable queue in its body.

The shape

import { Schema } from 'effect'
import { workflow } from '@voltro/workflow/define'

export const tourNarration = workflow({
  name: 'tourNarration',
  payload: Schema.Struct({
    rowId: Schema.String,
    tenantId: Schema.String,
    editedAt: Schema.Number,
  }),
  success: Schema.Void,
  idempotencyKey: ({ rowId, editedAt }) => `tour:${rowId}:${editedAt}`,

  debounce: { key: (p) => `tour:${p.rowId}`, period: '15 minutes' },
  concurrency: { limit: 5, key: (p) => p.tenantId },
  timeouts: { start: '1 hour', finish: '10 minutes' },
  onFailure: 'narrationFailed',
})

Every key callback receives the workflow's own decoded payload type. A misspelled field is a compile error, not a key that quietly becomes the string "undefined" and collapses every row in your deployment into one bucket.

`idempotencyKey` is the execution's identity — read this first

This is the single most expensive misunderstanding in the workflow API, and getting it wrong produces a design that looks right and silently stops working.

idempotencyKey is not a dedupe window. It is the execution's identity, permanently:

const a = yield* wf.execute({ id: 'same' })
const b = yield* wf.execute({ id: 'same' })   // does NOT run — replays a's result

After the run completes the key is spent. A later, genuinely new invocation under that key is a silent no-op that returns the old output. Nothing errors and nothing logs, because from the engine's point of view you asked for a run it already has.

So a key must be unique per unit of work you want to happen:


`tour:${rowId}`	wrong if the tour can ever be re-narrated
`tour:${rowId}:${editedAt}`	right — every edit is a new unit of work

And a flow-control key is a different thing

Conflating the two is what makes "I need to re-arm a key" feel like a missing feature. It is not missing; it is two fields:

idempotencyKey — the execution's identity. Varies per unit of work.
debounce.key / singleton.key / concurrency.key — the resource runs compete for. Stable.

"One job, fifteen minutes after the last edit, latest state wins" is then the example at the top of this page: twenty edits mint twenty identities, and exactly one is ever admitted. There is no restart: true in this API because separating the two keys is the mechanism it would have been.

debounce — collapse a burst

debounce: { key: (p) => `tour:${p.rowId}`, period: '15 minutes', timeout: '1 hour' }

Starts sharing a key collapse into one pending row. The timer resets on every arrival, and the latest payload wins — which is what "narrate what settled" means.

timeout is a hard cap measured from the first start in the burst. It is optional and uncapped when unset, which is a real trade: an unbroken stream of starts arriving faster than period defers the run forever. We do not invent a default cap and we do not warn — instead the starvation is a number you can see:

voltro workflows flow

A waiting= climbing past a few multiples of your period is the signal. Set timeout when you see it, or from the start if the burst is user-driven.

singleton — one run per resource

singleton: { key: (p) => p.tenantId, mode: 'skip' }    // newcomer stands down
singleton: { key: (p) => p.tenantId, mode: 'cancel' }  // newcomer evicts the incumbent

There is no default mode: 'skip' discards the incoming request and 'cancel' discards the running one, and choosing for you would silently throw away work either way.

Under 'skip', start() returns the incumbent's handle — a real, pollable run. Under 'cancel', the incumbent is cancelled when the replacement actually starts, not when it is queued. That matters when you also declare debounce: evicting at queue time would leave the whole quiet period with the old run dead and the new one not yet begun.

concurrency — bound what is in flight

concurrency: { limit: 5, key: (p) => p.tenantId }

At most limit runs in flight per key, across every replica — the count rides the shared admissions ledger, so three replicas with limit: 5 are five runs, not fifteen. Excess starts queue and are admitted as slots free.

`pool` — share one budget across workflows

// embeddings.workflow.tsx
concurrency: { limit: 10, pool: 'openai' }
// summarize.workflow.tsx
concurrency: { limit: 10, pool: 'openai' }

Without pool, the limit bounds this workflow's runs. With it, every workflow declaring the same pool name competes for one budget — the shape a rate-limited provider forces: five workflows that each call OpenAI must share ten slots, not hold ten each.

key still partitions within the pool: give each member key: (p) => p.tenantId and the shared budget applies per tenant.

Every member of a pool must declare the same limit — the boot fails otherwise. Two limits for one budget is a contradiction, and silently picking either would enforce a number somebody did not write.

throttle vs rateLimit — late, or gone

They are mutually exclusive, and declaring both is a boot error.

throttle:  { limit: 100, period: '1 minute', key: (p) => p.tenantId }  // QUEUES the excess
rateLimit: { limit: 100, period: '1 minute', key: (p) => p.tenantId }  // DROPS the excess

Reach for throttle when every start must eventually run. Reach for rateLimit when the excess is genuinely surplus and running it late is worse than not running it.

A dropped start is never silent: start() resolves to a handle with status: 'dropped' and a retryAfterMs, and the drop is a row in the admissions ledger with its key and reason.

throttle has no burst knob. The window is sliding, so its maximum instantaneous burst is already exactly limit; a separate knob could only duplicate it.

batch — many starts, one run

export const refreshIssues = workflow({
  name: 'refreshIssues',
  payload: Schema.Struct({ items: Schema.Array(IssueRef) }),
  success: Schema.Void,
  idempotencyKey: ({ items }) => `refresh:${items.length}:${items[0]?.issueKey ?? ''}`,
  batch: { item: IssueRef, key: (i) => i.tenantId, maxSize: 100, timeout: '30 seconds' },
})

await ctx.workflows.start('refreshIssues', { tenantId, issueKey: 'ABC-1' })

Callers start it with a single item; the workflow's own payload is the batch shape. That mismatch is checked at declaration time — a decode failure on the batching replica is a failure nobody is watching.

batch.item is therefore also what an arriving start is validated against. A caller's payload is judged by the item schema; the run the drainer eventually starts is judged by the workflow's own payload. Two schemas, because there are genuinely two shapes — and a WorkflowPayloadError naming items on a start() call would be the framework asking the caller for the batch it is supposed to be building.

The timeout is a deadline, not a quiet period: it does not reset per item, or a steady trickle would never flush. batch and debounce cannot both be declared for exactly that reason.

priority

priority: (p) => (p.urgent ? 100 : 0)

Higher runs first out of the pending queue. Ties break by arrival, so an all-default deployment is FIFO rather than dialect-dependent.

Scope — read before porting BullMQ priority lanes. priority orders the admission queue only: the pending rows a deferring control (debounce, batch, throttle, concurrency), a pause, or a delayed { at } start has parked. A start that is admitted immediately never competes with anything — it goes straight to the engine, whatever any other start's priority says. There are no cross-workflow priority lanes, no preemption of running work, and no ordering between two starts that both found a free slot. If urgent starts must overtake normal ones, put both classes behind the same concurrency limit (or a shared pool), so every start passes through the queue that priority orders.

timeouts

timeouts: { start: '1 hour', finish: '10 minutes' }

start bounds how long a start may sit in the admission queue, measured from the first arrival in its group. A debounced run that never gets a quiet moment is a job that silently did not happen.
finish bounds the run itself once admitted, and also tightens the crash backstop on its concurrency slot.

Both expire into the same path as an exhausted retry: the run is recorded failed and onFailure fires. Declaring timeouts.start on a workflow that cannot defer is a boot error — it would never fire.

onFailure — the signal that replaces the sweep

onFailure: 'narrationFailed'

A workflow name, not a function. A closure cannot be journaled: the failure may be noticed by a different replica, minutes later, after the process that held it is gone.

It fires for every way a run fails to deliver — not only an exhausted retry:

the body failed and retries are spent
timeouts.finish cancelled an overrunning run
timeouts.start expired a start that never got a slot
the workflow was renamed away while starts were queued

The last two produce no run row at all, which is exactly why polling listRuns({ status: 'failed' }) could never see them.

The named workflow's payload is WorkflowFailureReport:

export const narrationFailed = workflow({
  name: 'narrationFailed',
  payload: Schema.Struct({
    workflow: Schema.String,
    payload: Schema.Unknown,
    errorTag: Schema.NullOr(Schema.String),
    errorMessage: Schema.NullOr(Schema.String),
    runId: Schema.NullOr(Schema.String),
    executionId: Schema.NullOr(Schema.String),
    reason: Schema.String,
    failedAt: Schema.Number,
  }),
  success: Schema.Void,
  idempotencyKey: ({ runId, failedAt }) => `failed:${runId ?? 'none'}:${failedAt}`,
})

There is no onFailure for an onFailure — a handler that fails is logged and not re-notified, because the alternative is one run per failure per level with no floor.

encryptSteps

encryptSteps: true

step({ input }) is journaled and shown in the dashboard, which is a feature and the reason people pass rich input. For a step carrying personal data it is also a second copy outside the .encrypted() boundary the governance plugin establishes for tables.

encryptSteps closes it, reusing the same cipher — one key, one rotation story:

governancePlugin({ fieldEncryption: { secret: 'VOLTRO_FIELD_ENCRYPTION_KEY' } })

Declaring it without that plugin configured is a boot refusal, not a warning. A plaintext fallback would leave the declaration reading as protection while every step input sat readable.

cancelOn — stop live work when a correlated event arrives

workflow({
  name: 'tourNarration',
  payload: { rowId: Schema.String, issueKey: Schema.String },
  idempotencyKey: (p) => `tour:${p.rowId}`,
  cancelOn: [{
    event: 'jira.issue.deleted',
    schema: JiraIssueDeleted,
    match: (event, payload) => event.issueKey === payload.issueKey,
  }],
})

Both sides are typed: event from the entry's own schema, payload from the workflow's.

Why it is a declaration and not a race inside the body. You can express "stop when the issue is deleted" with awaitEvent and an interrupt. That works while the body is running. It does not work while the run is sleeping for six hours, suspended on a signal, or still sitting in the admission queue — which is the case you wanted cancellation for. The event has to reach a run whose fiber is not executing anything, and only something outside the body can do that.

So it is swept: a coordinated tick reads events published since a durable watermark, resolves each declaring workflow's live runs, and cancels the ones that correlate.

It also discards queued starts. Cancelling only the running one leaves a debounced or concurrency-queued duplicate to start seconds later, against the row that was just deleted — the exact outcome the declaration was meant to prevent, arriving late enough that nobody connects the two.

Field	Meaning
`event`	The name, exactly as `ctx.events.publish` writes it
`schema`	Decoded before `match` runs. An event whose shape does not decode is reported and never matched — cancelling on an event you could not read is cancelling blind
`match`	Required. Write `match: () => true` if you really mean "every live run"
`within`	Only cancel runs started within this window before the event
`reason`	Recorded on the run's `run-cancelled` event; defaults to `cancelOn:<event>`

match has no default for the same reason singleton.mode has none: the omitted case is "cancel every live run of this workflow", which is a legitimate thing to want and a catastrophic thing to acquire by forgetting a line.

A run that started after the event is never cancelled. Without that rule, a sweep catching up after a deployment gap reads an hour of history and kills runs that started in the meantime — and the symptom (fresh work cancelled for no visible reason) looks nothing like its cause (a restart).

Bulk cancel and bulk replay

A bad deploy leaves four thousand runs that must all stop, or four thousand that must all be re-driven once the downstream is fixed.

voltro workflows cancel-many --workflow tourNarration --reason "bad deploy"
voltro workflows cancel-many --workflow tourNarration --reason "bad deploy" --commit
voltro workflows replay-many --status failed --mode redrive --limit 200 --commit

Three things are deliberately stricter than the obvious design:

--limit is required and there is no "all". The cap is the blast radius. truncated in the result says whether more matched, so "did I get all of them" stays answerable without an unbounded verb ever existing.
It is a dry run unless you pass --commit. The default for a verb that can stop a thousand runs is the one that stops none. The dashboard panel enforces the same order: the apply button does not exist until a preview has returned a number.
--reason is required for a cancel. It lands on every affected run's run-cancelled event, so "why did four thousand runs stop on the 8th" has an answer in the table an operator is already reading.

The result is per-run, not a count: succeeded, failed (with the reason for each) and skipped (with what made each ineligible) are three different outcomes. A bulk op that reports "4000 cancelled" while forty failed is how people learn not to trust bulk ops.

Eligibility is fixed by the verb: a cancel acts on running and suspended runs; replay --mode redrive on failed only (redrive resumes from the step that died, which only exists for a failure); replay --mode retry on failed and cancelled. --mode has no default because the two cost very different amounts.

Delayed one-off starts — `start(..., { at })`

Run this once, later — without a schedule and without a sleep at the top of the body:

await ctx.workflows.start('orders.remind', { orderId }, {
  at: new Date(Date.now() + 24 * 60 * 60_000),   // tomorrow, this time
})

The start is parked as a durable row in the same pending queue the controls above use (mode: 'delayed'), and the coordinated drainer fires it when at arrives — it survives restarts and fires on whichever replica drains, never from an in-process timer. The handle comes back status: 'queued' with deferral: { mode: 'delayed', dueAt }.

Semantics worth knowing:

at is an absolute instant, deliberately — not a delay duration. A delay is measured "from when?" (enqueue? admission? retry?) and every queueing system answers differently; an instant has no such ambiguity and composes with the schedule/backfill surfaces, which are also instant-based. A relative delay is one line: at: new Date(Date.now() + ms).
At at, the start becomes an ordinary ARRIVAL. Declared controls judge it as of that moment — a debounce collapses it into whatever window is open then, a rate cap can drop it (recorded, as always). { at } delays the arrival; it never outranks a control.
An at in the past starts immediately — "no earlier than" is already satisfied.
{ at, wait: true } is refused: there is no result to block on for a start that exists only as a future row.
Each { at } start is its own row. Two delayed starts never collapse into one — unlike debounce, nothing about { at } says the second supersedes the first. Want collapsing? That is debounce, and they compose.

Prefer this over sleep as the first step of the body when the wait precedes the work: a parked row costs one row, a sleeping run costs a durable execution the whole time.

What a deferred start returns

start() no longer always returns a running handle:

const handle = await ctx.workflows.start('tourNarration', payload)

if (handle.status === 'queued')  { /* handle.deferral.dueAt tells you when */ }
if (handle.status === 'dropped') { /* over a rateLimit cap; it will NOT run */ }
if (handle.status === 'skipped') { /* handle.executionId is the incumbent */ }

executionId is null for queued and dropped, because there is no execution and there may never be one. Inventing an id there would produce a handle that polls status: 'unknown' forever.

Blocking callers

ctx.workflows.run(...) and start({ wait: true }) block for the run's result, and a start that was collapsed into a future run has none. A workflow declaring debounce / batch / throttle / concurrency therefore refuses those callers with an error naming both halves. Controls with a synchronous answer — singleton, rateLimit — keep working on every path (rateLimit throws WorkflowRateLimitedError, singleton: 'skip' throws WorkflowSingletonHeldError carrying the incumbent's id).

Pausing a workflow

voltro workflows pause tourNarration --reason "deploying a fix"
voltro workflows unpause tourNarration

A pause makes starts collect, never discard — so you come back to a backlog rather than a hole in the data. unpause drains it.

The pause is a row, so it applies fleet-wide; each replica picks it up on its next drain tick (~1 s).

Seeing what happened

voltro workflows flow
voltro workflows flow --workflow tourNarration --format json

or GET /_voltro/inspect/workflows/flow-control.

This is not optional colour. A debounce that collapses nineteen starts into one is indisputably correct behaviour and indistinguishable from nineteen starts vanishing — unless something writes down that it happened. So every decision is a row in _voltro_workflow_admissions, with its key, its reason, how many starts folded into it, and how long it waited.

Three questions it answers:

Question	Where
"Twenty edits, one run — did that work, or did I lose nineteen?"	`collapsed`
"Nothing has run for an hour. Stuck, or quiet?"	`waiting=` on the queued row
"Why did my run not start?"	the ledger's `outcome` + `reason`

Cost

A workflow that declares no control takes exactly the code path it took before this feature existed — no query, no branch beyond one map lookup.

A workflow that declares one pays only for that one: an undeclared control costs zero round trips. The rate/throttle window reads at most limit rows, which is why limit is a throughput knob and not somewhere to put 10⁶.

The drainer runs on one replica per tick through the same claim arbiter the cron scheduler uses. N replicas draining at once would each see a free slot and each take it.

Measured admission throughput

Measured, not estimated — node packages/cli/scripts/admission-throughput.mjs in the framework repo drives the real gate, facade and drainer against the in-memory reference store (slope across N=500/1000/2000 starts, median of 3 sequential repeats; Apple-silicon dev machine, 2026-08):

no controls (passthrough): ~1 µs/start (~800,000 starts/s) — the do-nothing path really does nothing.
through a concurrency gate: ~110 µs/start (~9,000 starts/s) — dominated by the reference store's unindexed admission-state scan, which grows with the ledger; a real dialect serves that read from an index, but also adds its round-trips. Read this as the machinery's worst-case CPU floor, not a database benchmark.
durable park → drain → start: ~4 µs/row (~240,000 rows/s) of pure machinery per queued row.

The deployed ceiling is min(these numbers, what your database serves for the admission reads/writes) — on any SQL dialect the database is the bound long before the gate is. That is also why there is no key-group batching in the admission path: at ~9k gated starts/s worst-case CPU, batching would add a flush boundary to a path whose bound is elsewhere.

Per step, the recorder adds exactly 2 fire-and-forget store writes (insert at step start, update at settle), off the step's critical path — plus the cluster engine's own journal write, which is the durability you asked for. Hot high-step workflows can turn the introspection copy off: workflows: { recording: 'coarse' } — see Debugging.

Declarative flow control

#The shape

#idempotencyKey is the execution's identity — read this first

#And a flow-control key is a different thing

#debounce — collapse a burst

#singleton — one run per resource

#concurrency — bound what is in flight

#pool — share one budget across workflows

#throttle vs rateLimit — late, or gone

#batch — many starts, one run

#priority

#timeouts

#onFailure — the signal that replaces the sweep

#encryptSteps

#cancelOn — stop live work when a correlated event arrives

#Bulk cancel and bulk replay

#Delayed one-off starts — start(..., { at })

#What a deferred start returns

#Blocking callers

#Pausing a workflow

#Seeing what happened

#Cost

#Measured admission throughput