Workflows & steps

Workflows

A workflow is registered on the engine by name and version; its body is the deterministic async function (ctx, input) => … the engine executes and replays.

import engine from '@adonis-agora/durable/services/main'
import type { WorkflowCtx } from '@adonis-agora/durable'

engine.register('checkout', '1', async (ctx: WorkflowCtx, order: Order) => {
  /* ... deterministic orchestration ... */
})

register(name, version, fn, opts?) takes an options object for tags, an executionTimeout ('30s' or ms), a singleton config, input validation, and event triggers. Start runs with engine.start('checkout', input, runId) — pass the registered name and a stable runId. (Starting the same runId twice is a no-op, which is what makes scheduling and webhooks idempotent.)

1@Workflow({ name: 'checkout', version: '1' })2export default class CheckoutWorkflow {3  constructor(4    private inventory: InventorySteps,5    private payments: PaymentSteps,6    private shipping: ShippingSteps,7    private email: EmailSteps,8  ) {}9 10  async run(ctx: WorkflowCtx, order: Order) {11    const hold = await ctx.step(this.inventory.reserve, order)12    const charge = await ctx.step(this.payments.charge, { order, hold })13    await ctx.waitForSignal(`packed:${order.id}`)14    const label = await ctx.step(this.shipping.ship, order)15    await ctx.step(this.email.confirm, { order, label })16    return { chargeId: charge.id, tracking: label.tracking }17  }18}

Steps

ctx.step(refOrName, input, opts?) is the one durable step primitive. It is always dispatched: the engine routes it to a handler registered under the step's name, suspends the run (zero compute) until the result lands, then resumes with it — checkpointed and replay-safe. The handler runs on whatever worker serves that name, in this process or another; the workflow code is the same either way.

There are three ways to author the handler a ctx.step routes to.

1// the workflow dispatches it by name and awaits the result:2await ctx.step(this.inventory.reserve, order)

1. By string name

The zero-decorator baseline — dispatch by the literal handler name. Nothing to import at the call site, and it works across runtimes (e.g. a Python worker serving the same name):

const charge = await ctx.step<{ chargeId: string }>(
  'payments:charge-card',
  { orderId: order.id, amountCents: order.total },
)

The handler is registered under that same name — see serving a handler.

2. @Step decorator (recommended)

Mark a class method with @Step. Steps are discovered from app/steps (config stepsPath, plus the @adonis-agora/durable/hooks/steps assembler barrel) and served by name automatically. Pass the method reference to ctx.step for full input/output types:

import { Step } from '@adonis-agora/durable'
import { z } from 'zod'

export default class PaymentSteps {
  @Step({
    name: 'payments:charge-card',
    input: z.object({ orderId: z.number().int(), amountCents: z.number().int() }),
    output: z.object({ chargeId: z.string() }),
    retries: 3,
  })
  async chargeCard(input: { orderId: number; amountCents: number }) {
    return { chargeId: await stripe.charge(input) }
  }
}

// in the workflow (with the step class injected as `this.payments`):
const charge = await ctx.step(this.payments.chargeCard, { orderId: order.id, amountCents: order.total })

Bare @Step() derives the routing name as `${ClassName}.${method}` (refactor-safe); @Step('custom:name') overrides it; the object form adds opt-in input/output zod schemas and a def-level retry/liveness policy.

3. defineStep

defineStep(name, fn, config?) builds a typed step handler without a class — the direct swap for a function-form workflow. It returns a ref you pass to ctx.step, and it is discoverable/servable by name:

import { defineStep } from '@adonis-agora/durable'
import { z } from 'zod'

export const chargeCard = defineStep(
  'payments:charge-card',
  async (input: { orderId: number; amountCents: number }) => ({ chargeId: await stripe.charge(input) }),
  {
    input: z.object({ orderId: z.number().int(), amountCents: z.number().int() }),
    output: z.object({ chargeId: z.string() }),
    retries: 3,
  },
)

// in the workflow:
const charge = await ctx.step(chargeCard, { orderId: order.id, amountCents: order.total })

Step schemas are optional — a bare @Step() or defineStep(name, fn) carries none, and the engine skips validation when they're absent. When present, input is parsed before the handler runs and output before its result is returned (at the serve boundary).

Dispatch options

ctx.step accepts a third StepDispatchOpts argument to route the call and override the step's declared policy field-by-field: { queue, priority, fairnessKey, transport, requires, retries, backoff, backoffMs, backoffMaxMs, jitter, timeoutMs, pickupTimeoutMs }.

await ctx.step(chargeCard, input, { queue: 'payments', priority: 10, retries: 5 })

See Retries & timeouts for the durable-retry, timeoutMs and pickupTimeoutMs liveness semantics, Flow control for queues, and Transports for pinning transport.

requires — refuse to dispatch into a queue nobody serves

requires names the capabilities a live worker must advertise to run this call:

await ctx.step('billing.charge', input, { requires: ['saga'] })

If no live worker advertises every name — or none is protocol-compatible — the run parks as blocked with a precise reason (which capability is missing, how many live workers were considered) instead of dispatching into a queue that will never be drained. Nothing is thrown into the workflow body; the run waits for a capable worker to appear and is visible as blocked in the dashboard.

Omit it and the step runs anywhere, which is the default. A per-call requires replaces the one declared on the @Step — it does not merge with it. See capability-aware routing.

Serving a step handler

Steps under app/steps are discovered and served by name automatically (the convention mirrors app/workflows). The low-level escape hatch is transport.handle(name, fn) — register a handler on a worker-side transport under the same name a ctx.step dispatches to:

// worker side — the transport serving this name
transport.handle('payments:charge-card', async (input: { orderId: number; amountCents: number }) => {
  return { chargeId: await stripe.charge(input) }
})

Routing is by name: a step's queue/wire token is derived from its (sanitized) name, optionally isolated onto its own worker pool with a partition. (The old per-step group option is deprecated — accepted but ignored.)

Local (in-process) steps

ctx.localStep(name, fn, opts?) runs a unit of work in-process: the body executes once right here in the engine, then its result is checkpointed and replayed — never dispatched to a worker. Reach for it for cheap in-process work, or as an escape hatch when a dispatched step is overkill.

const quote = await ctx.localStep('quote', () => pricing.fetch(order), { retries: 3 })

The body receives a step logger as its argument, so you can emit annotations from inside it. It supports the same StepOptions as a dispatched step (retries, backoff, timeout, and compensate for in-process sagas) — see Retries & backoff.

In-flight visibility

When a local step's body begins, the engine emits a step.started lifecycle event and (by default) writes a running checkpoint, so a long step shows up in the dashboard the moment it starts — not only once it finishes. The running checkpoint is a placeholder overwritten by the step's completed/failed result; it never short-circuits replay (only a completed checkpoint does), so a crash mid-body simply re-runs the step.

Toggle it in config/durable.ts:

export default defineConfig({
  trackStepStart: false, // default true
})

The step.started event fires either way; the flag only gates the extra checkpoint write. Set it to false on hot paths with many short local steps to halve their checkpoint writes, at the cost of in-flight visibility that survives a page reload.

Singleton workflows: one run per key

Some workflows must not run concurrently for the same subject. Two syncs for one shop racing each other produce interleaved writes; two "recompute this tenant's billing" runs at once double-charge. A singleton workflow serializes runs by a key derived from the input:

export default class SyncShopWorkflow extends BaseWorkflow {
  static workflow = {
    name: 'sync-shop',
    version: '1',
    singleton: {
      key: (input) => (input as { shopId: string }).shopId,
      limit: 1,
      maxQueueDepth: 20,
    },
  }

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

Start a run whose key is already at its limit and nothing fails — the run is created, parked as suspended, and admitted the moment a slot frees. Admission is FIFO by creation time and race-free across engine instances (the ordering comes from the store, not from in-memory state), and a settling run wakes the next in line immediately rather than making it wait out a poll tick.

Back-pressure

Without maxQueueDepth, a producer that outruns the workflow simply builds an unbounded queue of suspended runs — the failure surfaces late, as a store full of runs that will not finish for hours. Setting maxQueueDepth turns that into an immediate, visible rejection:

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

try {
  await SyncShopWorkflow.dispatch({ shopId })
} catch (error) {
  if (error instanceof SingletonQueueFullError) {
    // error.workflow, error.key, error.maxQueueDepth
    return response.tooManyRequests({ retryAfter: 60 })
  }
  throw error
}

The cap is limit + maxQueueDepth — in the example above, one running plus twenty queued. The check happens before the run is created, so a rejected start leaves nothing behind to clean up.

Fan-out (parallel steps)

Run steps concurrently with Promise.all — checkpoints stay deterministic because each step's position is taken in the synchronous prefix before any await:

const [a, b] = await Promise.all([
  ctx.localStep('a', () => doA()),
  ctx.localStep('b', () => doB()),
])

Fatal errors

Any thrown error is retried up to the step's limit. To stop retrying a business failure that a retry can't fix, throw FatalError — it fails the run immediately:

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

await ctx.localStep('charge', async () => {
  const res = await stripe.charge(order)
  if (res.declined) throw new FatalError('card declined', 'declined')
  return res
})

The optional second argument is a machine-readable code that ends up on the run's structured error.

Catching errors — rethrow control-flow signals

If you wrap a step in your own try/catch inside the workflow body — to run a cleanup or compensating path on failure — you must rethrow control-flow signals untouched. Suspend-the-run operations (ctx.sleep, ctx.waitForSignal, ctx.continueAsNew) unwind the current turn by throwing: they are not real failures, and running a failure path on one records extra commands into history that a later replay never produces — which surfaces as a NonDeterminismError when the run resumes.

Use isWorkflowControlFlowSignal(error) to let those through before treating an error as a genuine failure:

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

try {
  await ctx.step('chargeCard', input)
} catch (error) {
  if (isWorkflowControlFlowSignal(error)) throw error // suspend / continue-as-new: rethrow as-is
  await ctx.step('refund', input) // a REAL failure — safe to compensate here
  throw error
}

The predicate is true for the engine's control-flow signals (currently a suspend and a continue-as-new) regardless of which module instance threw them — it checks a stamped marker (CONTROL_FLOW_SIGNAL), not instanceof, so it stays correct across a duplicated module copy. It is deliberately false for a cancelled run (a terminal outcome you may legitimately want to observe) and for a FatalError or a step rejection (real failures your catch is meant to handle) — so this guard never swallows a genuine error. Most of the time you don't need it: prefer the built-in saga compensation over hand-rolled try/catch, and reach for isWorkflowControlFlowSignal only when you truly must catch inside the workflow body.

Steps vs. sub-process events

These look similar in the dashboard but are fundamentally different — the distinction is durability:

• A step (ctx.step or ctx.localStep) is a durable checkpoint. The engine records its result, so on a crash, retry, or replay it is not re-executed — the saved result is replayed. It's a first-class node in the run graph and the unit of recovery.

• A sub-process event is a log annotation emitted inside a step via the step logger's log.sub(name, status) — e.g. one entry per item in a fan-out the step performs internally. It is not durable on its own: it's metadata attached to the parent step's checkpoint. If the parent step retries, all of its sub-process events are produced again. Use it for visibility, not recovery.

// One durable step that internally processes N items and records each outcome for visibility.
await ctx.localStep('process-batch', async (log) => {
  for (const item of batch) {
    try {
      await handle(item)
      log.sub(item.id, 'ok') // a sub-process event — shown under the step, not a checkpoint
    } catch (e) {
      log.sub(item.id, 'failed', String(e))
    }
  }
})

The step logger also offers log.debug/info/warn/error(message, data?) for leveled logs, log.subProcess(name, body) for nested, phased sub-work, and log.subEvent({ ... }) for a fully-specified sub-process entry (its own id, group, phase, status, message and data) when log.sub's three arguments are not enough. Rule of thumb: if you want each unit to retry or replay independently, make it a step (or a child workflow). If you only want to see what happened inside one durable unit, emit sub-process events.

Reporting liveness from inside a long step

A step that runs for minutes has a problem the logger cannot solve: from the engine's side, "working hard" and "the worker died" look identical. log.heartbeat(progress?) is the signal that separates them:

@Step()
async reindexCatalog(input: { shopId: string }, log: StepLogger) {
  const pages = await catalog.pageCount(input.shopId)

  for (let page = 1; page <= pages; page += 1) {
    await catalog.reindexPage(input.shopId, page)
    log.heartbeat({ page, pages })
  }

  return { pages }
}

Each beat does three things:

• It rearms the step's timeoutMs. That window means "maximum silence", not "maximum duration" — so a step that beats every few seconds can run for an hour under a timeoutMs: 30_000 and never time out, while a worker that dies is caught within the window. See Retries & timeouts.
• It persists the progress payload on the step's checkpoint, so durable:runs and the dashboard can show where a long step is, not just that it is running.
• It costs almost nothing. Beats are throttled — at most one every 5 seconds leaves the worker, and the engine persists at most one every 10 seconds per step, with an exemption so the first beat carrying a progress payload always lands. Call it as often as is convenient; the throttles do the rationing.

Beat from a dispatched step, where there is a worker whose silence is meaningful. Inside ctx.localStep the body runs in the engine itself, so there is nothing to report liveness to and log.heartbeat is a no-op.

Deterministic sources — ctx.now & ctx.sideEffect

Because the body replays, never read a value that changes on each run. The context exposes deterministic, checkpointed sources — each records its value the first time and replays it afterwards:

const issuedAt = await ctx.now()                              // epoch ms — captured once, replayed verbatim
const nonce = await ctx.sideEffect(() => crypto.randomUUID()) // any generated value, captured once
const sampled = (await ctx.sideEffect(() => Math.random())) < 0.1

ctx.now() is the deterministic wall-clock. ctx.sideEffect(fn) is the general deterministic capture: it runs fn once, checkpoints the result, and returns that same value on every replay without re-running fn — wrap any non-determinism you control (a UUID/ULID, Math.random(), a config/env read) in it. For real work with side effects — a DB write, an API call — use a step instead, since sideEffect's fn must be effectively pure (it runs only once and is not re-executed on replay).

See Versioning & determinism for why and how the runtime guard works.

Tags

Label runs with tags to find them later. Static tags on register apply to every run; per-run tags are added at start. Both are merged onto the run and are searchable in the dashboard.

engine.register('pipeline', '1', async (ctx) => { /* ... */ }, { tags: ['etl', 'critical'] })

// per-run tags merge with the static ones → run.tags = ['etl', 'critical', 'nightly']
await engine.start('pipeline', input, runId, { tags: ['nightly'] })

The dashboard shows a run's tags and a tag filter. Programmatically, query by tag with RunQuery.tag (engine.listRuns({ tag: 'etl' })) or the dashboard API's ?tag= param.

Delayed starts — startAt

Start a run now but have it execute later by passing startAt (epoch ms or a Date) to engine.start:

// Created durably right away; the body first executes at 09:00 tomorrow.
await engine.start('send-reminder', { orderId }, runId, { startAt: tomorrowAt9am })

The run is created immediately but parked suspended with its wake timer set to startAt; the ordinary durable-timer poller starts it when due. That makes the delay crash-safe with no ctx.sleep polluting the workflow body or its history — the body's first checkpoint is its first real step, and a fix-and-replay of the run doesn't re-wait the delay. A startAt in the past (or absent) starts immediately. One event-lifecycle nuance: a delayed run skips the run.started pending → running hop and goes suspended → … when it fires.

Run origin — package attribution

When several packages start workflows on one engine — your app plus a couple of @adonis-agora/* libraries — "whose code produced this run?" is worth answering without archaeology. Stamp it with origin, at any of three levels (most specific wins):

// On the class — every run of this workflow:
static workflow = { name: 'sync-catalog', version: '1', origin: '@acme/catalog-pipeline' }

// On a manual registration:
engine.register('sync-catalog', '1', fn, { origin: '@acme/catalog-pipeline' })

// On one start — overriding the registration, for a shared workflow started on another package's behalf:
await engine.start('sync-catalog', input, runId, { origin: '@acme/billing' })

The origin lands on run.origin, is filterable (engine.listRuns({ origin: '@acme/catalog-pipeline' }), the dashboard API's ?origin= param), and powers the console's origin facet — runs with no origin render as "unknown" there. It's attribution, not routing: use namespace to pin where a run executes.