Routing a run's failure — onError

Some failures are not accidents. A line that fails validation, a record a downstream system rejects, an order for a product that no longer exists: retrying won't change the answer, and failing the run isn't the right end state either, because something should happen next (record the line, tell someone, move on).

onError routes such a failure to a handler workflow. When a run ends failed with an error whose code matches, the engine starts the handler and the failed run waits for it, then settles with the handler's outcome: its return value becomes the run's output, so the run completes.

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

export class InvalidLine extends FatalError {
  static readonly code = 'invalid_line'
  constructor(message: string) {
    super(message, InvalidLine.code)
  }
}

export class RejectedLineWorkflow extends BaseWorkflow {
  static workflow = { name: 'line-rejected' }

  async run(ctx: WorkflowCtx, failure: OnErrorInput<Line>): Promise<LineResult> {
    await ctx.step(recordRejectedLine, { line: failure.input, error: failure.error.message })
    return { ok: 0, failed: 1 }
  }
}

export class LineWorkflow extends BaseWorkflow {
  static workflow = {
    name: 'line',
    onError: [{ code: InvalidLine, workflow: RejectedLineWorkflow }],
  }

  async run(ctx: WorkflowCtx, line: Line): Promise<LineResult> {
    await ctx.step(validateLine, line) // throws InvalidLine for a bad line
    await ctx.step(ingestLine, line)
    return { ok: 1, failed: 0 }
  }
}

A file that fans out over its lines with ctx.all(LineWorkflow, lines) now completes with one LineResult per line, instead of failing on the first invalid one.

How the engine runs itCopy Anchor Link

The engine does the routing, at the point where it settles a run as failed. It never looks inside the body, so the same thing happens whether the body ran in-process or on a remote worker.

1. The run's body fails with an error carrying code: 'invalid_line'. Saga compensations of the body run, as for any failure.
2. Instead of marking the run failed, the engine starts the handler as its own run, with the id <runId>.onError.<attempt> (.0 for the first handoff) and an OnErrorInput ({ runId, workflow, input, error }) in the failed run's namespace, tags the run onError:<code> (onError:* for the catch-all route), emits run.routed, and suspends the failed run on the handler the way a parent waits for a ctx.child.
3. From then on the run only follows its handler: the body is never run again, not by a recovery wake nor by a manual resume. When the handler finishes, the run settles with its outcome: completed with the handler's output, or failed with the handler's error.

The handler is an ordinary durable run: it shows up in the dashboard linked to the run it handles, its steps are checkpointed and retried, and the wait survives a crash like any child wait. The onError:<code> tag makes the handled runs a filter away in the console.

The failure's detailsCopy Anchor Link

error carries the failure's message, code and details. details is whatever JSON-serializable value the failure was raised with — new FatalError(message, code, details) — so a handler can record exactly what went wrong (e.g. a schema's validation issues):

// FatalError(message, code, details)
export class InvalidLine extends FatalError {
  static readonly code = 'invalid_line'
  constructor(message: string, issues?: unknown) {
    super(message, InvalidLine.code, issues)
  }
}

// in the step
const parsed = LineSchema.safeParse(line)
if (!parsed.success) throw new InvalidLine(`line ${line.n} is invalid`, parsed.error.issues)

// in the handler: failure.error.details is the issues array

The same details reach a parent's ctx.allSettled item. In Python: raise FatalError("line 3 is invalid", "invalid_line", details=issues).

Retrying a runCopy Anchor Link

Retrying a failed run (from the dashboard, node ace durable:retry <runId>, or engine.requeue(runId)) runs its body again. A handoff the run made before is retired: it stays in the timeline, renamed onError:retried:<handler id>, and is never followed again, and the run's onError:<code> tag is removed. If the body fails again, the engine hands the new failure to a new handler run (<runId>.onError.1, then .2, …), tagged again.

Catching a misnamed handlerCopy Anchor Link

After boot, the provider warns about any route whose handler workflow is neither registered in the app nor served by a live worker. A route unserved at boot is checked again 30 seconds later before it is reported, so a handler whose workers start after the app is not flagged. The same check is engine.unservedOnErrorRoutes().

Seeing itCopy Anchor Link

The dashboard shows the failure's code and its details (as JSON) on the failed step and run. Handoffs are counted as routed (by code) in collectMetrics and attachDurableMetrics, marked as a durable.run.routed event on the run's OpenTelemetry span, and reach Telescope through the diagnostics bridge as agora:durable:run.routed / run.unrouted.

MatchingCopy Anchor Link

Routing matches on the error's code, not its class: a dispatched step's error reaches the engine as a plain { message, code }, and a remote body's failure was never a JavaScript error at all. code is the code string, or an error class with a static code; a route without code is the catch-all, taking any failure no other route matches. An exact code match always wins over the catch-all. Two routes for the same code, or two catch-alls, are ambiguous and rejected when the workflow is registered. To send several codes to one handler, list one route per code with the same workflow.

The code comes from FatalError's second argument, from a code on whatever a step handler threw, or from the code a remote worker reported for the failed step. A failure from a step reaches the route once the step's retries are spent; a FatalError skips them (in a dispatched handler too, because it carries retryable: false). A SchemaValidationError thrown in the body is a FatalError with the code invalid_input (a step's input schema, or the workflow's stored input no longer parsing) or invalid_output (a step's output schema), so it can be routed like any other code. A route for schema_validation — or code: SchemaValidationError, whose static code it is — catches both when no route names them, as well as a rejected searchAttributes write:

onError: [
  { code: 'invalid_output', workflow: ReportBadProviderResponse }, // just this one
  { code: SchemaValidationError, workflow: QuarantineRun },        // every other schema failure
]

Never routed: cancellation (a run past its executionTimeout is cancelled too), dead-lettering, and the failures that mean the deployment is broken (non-determinism, a ctx call nested in a step). Those keep their usual meaning. A step that times out fails like any other step, so its failure is routable.

Remote workflowsCopy Anchor Link

For a workflow the engine serves by convention (a Python body, or any worker it reaches by its live queue name), the routes come from what its workers announce in their handshake descriptor. A TS worker announces them through WorkerRuntime.registerWorkflow(name, body, meta) — workflowAnnouncement(MyWorkflow) builds that meta from a class's static workflow — and the Python SDK through @worker.workflow("line", on_error=[{"code": "invalid_line", "workflow": "line-rejected"}]).

A Python worker can also declare the handler inline: .on_error(code) on the workflow function registers the handler as its own workflow, <workflow>.onError.<function>, and announces the route to it. code is a code string or an exception class with a code attribute:

@worker.workflow("line")
def line(ctx, input): ...

@line.on_error(InvalidLine)
def reject_line(ctx, failure):  # {"runId", "workflow", "input", "error"}
    ctx.step("record_rejected_line", {"line": failure["input"], "error": failure["error"]})
    return {"ok": 0, "failed": 1}

A Python workflow class does the same with @on_error(code) methods. See Python: workflows as classes.

The handler can be a workflow of any runtime; the engine only starts a run by name. If the live workers announce different routes for the workflow (a rollout in progress), the failure is not routed and the engine emits run.unrouted with the reason. The engine reuses what it read for a few seconds, so a burst of failures costs one read.

Routes are read from the worker advertisement, which every cross-process transport carries; the queue transport keeps it in a key store beside its queues (the app database by default), so with no database there it has none, and a convention workflow's failures are not routed: declare its routes where it is registered instead.

On the engine directlyCopy Anchor Link

engine.register(name, version, fn, { onError: [{ code?, workflow }] }), and engine.registerRemote takes the same option for a remote workflow registered by hand rather than served by convention.

onError, ctx.allSettled and the dead-letter queueCopy Anchor Link

Reach for onError when the failure policy belongs to the workflow, and for ctx.allSettled when it belongs to one caller.