API reference
Every public member of WorkflowCtx, BaseWorkflow and the authoring decorators, the WorkflowEngine's selected surface, the RunGateway, the run-detail and fan types, the DurableConfig options of config/durable.ts, the driver factories, the observability and command building blocks, the thrown error classes, the run/checkpoint status enums and the package's subpath exports — one line each, distilled from source.
This page is a flat lookup table of the public API of @adonis-agora/durable: the WorkflowCtx, BaseWorkflow and the decorators, the engine, the RunGateway, defineConfig, the driver factories, the observability and command building blocks, the errors and the statuses. It is deliberately terse. For the reasoning behind an entry, follow its link to the concept, authoring or reliability page.
Everything below is exported from the package root (@adonis-agora/durable) unless a row names a subpath.
WorkflowCtx
The context handed to a workflow's run(ctx, input). Every interaction with the outside world goes through it so the engine can checkpoint, and the body itself stays deterministic (see Durability & replay).
Properties
| Member | Signature | Notes |
|---|---|---|
runId | readonly runId: string | This run's id. |
Steps
| Member | Signature | Notes |
|---|---|---|
step | step<TInput, TOutput>(handler: StepRef<TInput, TOutput>, input: TInput, opts?: StepDispatchOpts & { compensate?: StepRef<StepUndo<TInput, TOutput>, unknown> | string }): Promise<TOutput> | Dispatch a durable step by typed reference (a @Step method or a defineStep handle). This is the one step primitive, and it is always engine-scheduled: the run suspends until the result lands. compensate registers a dispatched saga undo. See Workflows & steps. |
step (by name) | step<TOutput = unknown>(name: string, input: unknown, opts?: StepDispatchOpts & { compensate?: StepRef<StepUndo<unknown, TOutput>, unknown> | string }): Promise<TOutput> | The same dispatch addressed by string name, for a cross-runtime handler (e.g. a Python step) with no JS reference. |
localStep | localStep<TOutput>(name: string, fn: (log: StepLogger) => Promise<TOutput>, options?: StepOptions): Promise<TOutput> | A checkpointed step whose body runs in the engine process instead of being dispatched. It can carry an in-memory compensate for sagas. See Local steps. |
transaction | transaction<TOutput>(name: string, fn: (tx: unknown) => Promise<TOutput>): Promise<TOutput> | Exactly-once DB step: runs fn and writes the checkpoint in one store transaction, so the business write and the "done" marker commit together. Needs a store that supports transactions and throws otherwise. See External tasks & transactions. |
Time
| Member | Signature | Notes |
|---|---|---|
sleep | sleep(duration: string | number): Promise<void> | Durable sleep for a relative duration ('30s', '2h', '7 days', or ms). Uses no compute while waiting and resumes across restarts. See Sleep & signals. |
sleepUntil | sleepUntil(when: Date | number): Promise<void> | Durable sleep until an absolute time. The wake time is fixed on the first run, so it is replay-stable. |
now | now(): Promise<number> | Deterministic epoch-ms clock: recorded on the first run, the same value on replay. Use it instead of Date.now() in a workflow body. See Versioning & determinism. |
Signals & events
| Member | Signature | Notes |
|---|---|---|
waitForSignal | waitForSignal<TPayload>(token: string, opts?: { timeoutMs?: number }): Promise<TPayload> | Suspend until engine.signal(token, payload) arrives. timeoutMs bounds the wait and throws SignalTimeoutError when it passes. See Sleep & signals. |
waitForEvent | waitForEvent<TPayload>(name: string, opts?: { match?: Record<string, unknown>; timeoutMs?: number }): Promise<TPayload> | Suspend for a named event published with engine.publishEvent. match is a payload subset that must deep-equal, so a publish resumes only the runs it concerns. See Named events. |
setEvent | setEvent<TValue>(key: string, value: TValue): Promise<void> | Publish a queryable value that engine.getEvent(runId, key) reads while the run is in flight. See Queries & updates. |
onUpdate | onUpdate<TArg>(name: string, opts?: { timeoutMs?: number }): Promise<TArg> | A run-scoped update point: suspends until engine.update(runId, name, arg) delivers arg, after any registered validator accepts it. |
Children
| Member | Signature | Notes |
|---|---|---|
child | child<C extends WorkflowClass>(workflow: C, input: WorkflowInputOf<C>, options?: string | ChildCallOptions): Promise<WorkflowOutputOf<C>> | Start (or reattach to) a child run and suspend until it is terminal, then resume with its output. Throws a FatalError if the child failed. See Child workflows. |
child (by name) | child<TOutput>(workflow: string, input: unknown, options?: string | ChildCallOptions): Promise<TOutput> | The same, for a cross-runtime child addressed by name. |
startChild | startChild<C extends WorkflowClass>(workflow: C, input: WorkflowInputOf<C>, options?: string | ChildCallOptions): Promise<string> | Fire-and-forget: dispatch a child once and return its run id at once; the parent keeps running. Also takes a workflow name (startChild(workflow: string, input: unknown, options?)). |
all | all<C extends WorkflowClass>(workflow: C, inputs: WorkflowInputOf<C>[], opts?: AllOptions): Promise<WorkflowOutputOf<C>[]> | Run N children of the same workflow in parallel and await all of them, outputs in input order. waitAll (the default) throws an aggregate GatherError; failFast throws on the first failure and cancels the remaining siblings (best-effort). Also takes a workflow name (all<TOutput = unknown>(workflow: string, inputs: unknown[], opts?: AllOptions)). |
all (over a step) | all<TInput, TOutput>(step: StepRef<TInput, TOutput>, inputs: TInput[], opts?: AllStepsOptions): Promise<TOutput[]> | One durable step per input, joined once: the parent replays once per settled wave, not once per result. Failures surface as a GatherError. See Fanning out over a step. |
allSettled | allSettled<C extends WorkflowClass>(workflow: C, inputs: WorkflowInputOf<C>[], opts?: AllSettledOptions): Promise<Settled<WorkflowOutputOf<C>>[]> | The same fan as all in waitAll mode, resolving with one Settled per input instead of throwing. Also takes a @Step reference (opts?: AllStepsSettledOptions) or a workflow name. See Child workflows. |
External
| Member | Signature | Notes |
|---|---|---|
task | task<TResult>(name: string, dispatch: () => Promise<void>, options?: StepOptions): Promise<TResult> | An external task with async completion: runs dispatch once, then suspends until engine.completeTask or engine.failTask reports back. name must be unique per run. See External tasks. |
webhook | webhook<TPayload>(): DurableWebhook<TPayload> | Mint a durable callback handle (token plus an optional public url). await handle.wait({ timeoutMs? }) suspends until the callback lands as engine.signal(token, body). See Durable webhooks. |
callEntity | callEntity<TResult = unknown>(name: string, key: string, op: string, arg?: unknown): Promise<TResult> | Call a durable entity op (serialized per key) and await its result. See Durable entities. |
signalEntity | signalEntity(name: string, key: string, op: string, arg?: unknown): Promise<void> | Send an entity op fire-and-forget. |
State & metadata
| Member | Signature | Notes |
|---|---|---|
patched | patched(id: string): Promise<boolean> | Guard an in-place change without a new version: true for a fresh run (new branch), false for a run whose history predates the guard. See Versioning. |
continueAsNew | continueAsNew(input?: unknown): Promise<never> | End this run and start a fresh execution of the same workflow with clean history (id <runId>~N). Always throws. See Continue-as-new. |
sideEffect | sideEffect<TValue>(fn: () => TValue | Promise<TValue>): Promise<TValue> | Deterministic capture: runs fn once, checkpoints the result, and replay returns the same value without re-running fn (e.g. uuidv7(), Math.random()). |
upsertSearchAttributes | upsertSearchAttributes(attrs: Partial<A>): Promise<void> | Shallow-merge queryable metadata onto this run's searchAttributes (A is WorkflowCtx<A>'s attribute shape, default SearchAttributes). Exactly-once: skipped on replay. With a declared searchAttributes schema the merged result is validated first (SchemaValidationError on failure). See Typed search attributes and See Search attributes. |
breakpoint | breakpoint(label?: string): Promise<void> | Pause the run here until a human resumes it with engine.continue(runId) or the dashboard. See Human-in-the-loop pauses. |
Supporting option types
| Type | Shape | Notes |
|---|---|---|
StepDispatchOpts | { queue?, priority?, fairnessKey?, transport?, retries?, backoff?, backoffMs?, backoffMaxMs?, jitter?, timeoutMs?, pickupTimeoutMs?, requires? } | Per-call override of the @Step-declared policy, merged as { ...stepConfigOf(ref), ...opts }. queue routes through a flow-control queue, where priority and fairnessKey apply. transport pins a named transport in the pool. requires lists capabilities a worker must advertise. See Dispatch options. |
StepOptions | { retries?, backoff?, backoffMs?, backoffMaxMs?, jitter?, timeoutMs?, pickupTimeoutMs?, compensate?, requires? } | Options for ctx.localStep and ctx.task. compensate: () => Promise<void> registers an in-process saga undo, retried up to compensationRetries (default 1). See Retries & backoff. |
BackoffStrategy | 'fixed' | 'exp' | Constant or doubling delay between retries. |
StepUndo<TInput, TOutput> | { input: TInput; output: TOutput } | The single argument a dispatched saga undo receives: the compensated call's own input and result. |
UndoOf<H> | H extends (input: infer I, ...rest: never[]) => infer R ? StepUndo<I, Awaited<R>> : StepUndo<unknown, unknown> | Derives an undo's argument from the original step's signature, so a compensate ref is compile-checked: async cancelBooking(undo: UndoOf<FlightService['book']>) { ... }. |
AllOptions | { mode?: 'waitAll' | 'failFast'; concurrency?: number } | For ctx.all over children. concurrency sends waves of floor(concurrency / 2), two at a time. It is part of the fan's identity, so don't change it under an in-flight run. |
AllStepsOptions | Omit<StepDispatchOpts, 'queue' | 'timeoutMs' | 'pickupTimeoutMs'> & { mode?: 'waitAll' | 'failFast'; concurrency?: number } | For ctx.all over a @Step. The retry/backoff options apply to every step of the fan; concurrency sends strict waves of that size. |
AllSettledOptions / AllStepsSettledOptions | Omit<AllOptions, 'mode'> / Omit<AllStepsOptions, 'mode'> | The ctx.allSettled counterparts. |
Settled<T> | { status: 'fulfilled'; value: T } | { status: 'rejected'; id: string; error: string; code?: string; details?: unknown } | One item of an allSettled fan, in input order. id is the child run id or the step id. |
ChildCallOptions | { childId?: string; priority?: number } | Options for ctx.child and ctx.startChild; a bare string is shorthand for { childId }. priority only affects a remote child. |
DurableWebhook<TPayload> | { readonly token: string; readonly url?: string; wait(opts?: { timeoutMs?: number }): Promise<TPayload> } | What ctx.webhook() returns. url is built by the webhookUrl config option. |
DynamicWorkflowCtx — the non-overloaded view
WorkflowCtx minus its five overloaded methods, re-declared with one string-addressed signature each: step(name, input, opts?), child(workflow, input, options?), startChild(workflow, input, options?), all(workflow, inputs, opts?) and allSettled(workflow, inputs, opts?). Every WorkflowCtx is assignable to it without a cast. Depend on it (or a Pick of it) in code that drives the ctx by names, such as a graph interpreter, and in test fakes, where satisfying every overload of the full WorkflowCtx is impractical:
type InterpreterCtx = Pick<DynamicWorkflowCtx, 'runId' | 'localStep' | 'child' | 'all' | 'sleep'>
new Interpreter(ctx) // a WorkflowCtx IS a DynamicWorkflowCtx — no `as unknown as` castStepLogger
The second argument a step handler (and a ctx.localStep body) receives. Events it records are checkpointed with the step and rendered under it in the dashboard. See Steps vs. sub-process events.
| Member | Signature | Notes |
|---|---|---|
debug info warn error | (message: string, data?: unknown): void | One method per level. |
heartbeat | heartbeat(progress?: unknown): void | Report "alive mid-flight": rearms the step's timeoutMs window and persists the latest beat on the checkpoint. Throttled internally (about 5s). See Reporting liveness. |
sub | sub(name: string, status: 'ok' | 'failed' | 'skipped', message?: string, data?: unknown): void | Record a sub-step outcome. |
subEvent | subEvent(e: { id; name; group?; phase?; status?; message?; data? }): void | Record a sub-process event (a phase transition or a terminal status). |
subProcess | subProcess<T>(name: string, body: (sp: SubProcessHandle) => Promise<T> | T, opts?: { group?: string; id?: string }): Promise<T> | Time body and record ok or failed. sp.phase(label), sp.skip(reason) and sp.fail(reason) mark other outcomes. |
currentStep()
currentStep(): StepLogger | undefined, exported from @adonis-agora/durable, returns the logger of the step body running on the current async path: the same instance the body received, so what it records lands in the same event list. Every step body has one, local (ctx.step, ctx.localStep) and remote (a handler on a worker). It is undefined outside a step, so a helper deep in your code can instrument itself without being handed the logger and still runs in a unit test:
import { currentStep } from '@adonis-agora/durable'
export async function insertBatch(rows: Row[]) {
await db.table('lines').multiInsert(rows)
currentStep()?.info('inserted batch', { rows: rows.length })
}Authoring
BaseWorkflow
The base class for a workflow under app/workflows/, auto-registered at boot. See app/workflows.
| Member | Signature | Notes |
|---|---|---|
workflow | static workflow?: WorkflowOptions | The workflow's identity and options (table below). A class without it is not registrable. |
schedule | static schedule?: WorkflowScheduleConfig | WorkflowScheduleConfig[] | Colocated recurring schedule(s), merged with config.schedules (config wins on a key collision). See Scheduling. |
run | abstract run(ctx: WorkflowCtx, input: unknown): unknown | The workflow body. Annotate input with the type you want WorkflowInputOf to infer. |
deadLetter | deadLetter?(ctx: WorkflowCtx, dead: DeadLetter): unknown | Optional inline dead-letter handler, registered as <name>.dlq (same version and origin) and started as dlq:<deadRunId> when a run of this workflow goes dead. DeadLetter<TInput> is { deadRunId: string; workflow: string; input: TInput; error?: StepError }. Declaring it with static workflow.deadLetterWorkflow throws. See Dead-letter queue. |
start | static start<C extends WorkflowClass>(this: C, input: WorkflowInputOf<C>, opts?: WorkflowDispatchOptions): Promise<WorkflowOutputOf<C>> | Run and wait for the result. Outside a workflow: engine.start plus waitForRun(runId, { terminal: true }). Inside one: ctx.child. |
dispatch | static dispatch<C extends WorkflowClass>(this: C, input: WorkflowInputOf<C>, opts?: WorkflowDispatchOptions): Promise<{ runId: string }> | Fire-and-forget. Outside a workflow: engine.start. Inside one: ctx.startChild. |
WorkflowDispatchOptions is { runId?: string } & StartOptions. Outside a workflow runId sets the new run's id (default randomUUID()); inside one it is the deterministic child id, and omitting it lets the ctx derive a replay-stable one. See Child workflows.
Related helpers:
| Export | Signature | Notes |
|---|---|---|
setWorkflowEngineResolver | setWorkflowEngineResolver(resolver: WorkflowEngineResolver | undefined): void | Override how the statics find the engine outside a workflow (default: the AdonisJS container). A testing seam; pass undefined to restore. See Testing. |
getCurrentWorkflowCtx | getCurrentWorkflowCtx(): WorkflowCtx | undefined | The ambient WorkflowCtx on the executing body's async path, or undefined outside one. |
workflowName | workflowName(ref: WorkflowRef): string | Resolve a class or a name to the registered name. Throws for a class without static workflow. |
workflowMeta | workflowMeta(target: unknown): WorkflowMeta | undefined | Read a class's static workflow with version normalized to '1'. |
WorkflowOptions (static workflow)
| Option | Type | Notes |
|---|---|---|
name | string | Required. The registered, cross-runtime workflow name. |
version | string | Default '1'. Old runs keep resuming on the version they started on. See Versioning. |
tags | string[] | Labels merged onto every run, together with per-start tags. |
executionTimeout | string | number | Wall-clock budget for the whole run ('5m' or ms). An overrunning run is cancelled by sweepTimeouts. |
onEvent | string[] | Event names whose engine.publishEvent starts a run of this workflow. See Named events. |
singleton | SingletonConfig ({ key: (input: unknown) => string; limit?: number; maxQueueDepth?: number }) | Durable FIFO mutex per key. See Singleton workflows. |
concurrency | ConcurrencyConfig ({ key: (input: unknown) => string | undefined; limit: number | ((key: string) => number | Promise<number>); countStatuses?: RunStatus[] }) | Start-time quota per key, shared across workflows. An over-limit start throws ConcurrencyLimitError. See Concurrency quotas. |
origin | string | Package attribution stamped on every run (WorkflowRun.origin). Overrides the origin derived from the nearest package.json name. See Run origin. |
onError | { code?: string | { readonly code: string }; workflow: WorkflowRef }[] | Route a failed run to a handler workflow by error code (string or an error class with a static code; omit for the catch-all). See Routing a run's failure. |
deadLetterWorkflow | WorkflowRef | The workflow a dead run of this one is handed to (overrides the config default). Mutually exclusive with a deadLetter() method. See Dead-letter queue. |
inputSchema | StandardSchemaV1 | Any Standard Schema (zod, valibot, arktype, …). Validated at start before a run is created. The run stores the input as sent; run receives the schema's output, re-parsed before every turn. A failure throws SchemaValidationError. See Validating the input. |
searchAttributes | SearchAttributesSchema | A Standard Schema with a flat output: every ctx.upsertSearchAttributes is validated against the merged attributes. Must validate synchronously. See Typed search attributes. |
@Step(nameOrOptions?) and defineStep
Step(nameOrOptions?: string | StepDecoratorOptions): MethodDecorator marks a class method as a durable step handler, routed by name. Steps under app/steps/ are discovered and served automatically (config stepsPath). See Workflows & steps.
| Form | Notes |
|---|---|
@Step() | Name derived as `${ClassName}.${method}`. |
@Step('custom:name') | Explicit name, a stable cross-runtime contract. |
@Step({ name?, input?, output?, retries?, backoff?, backoffMs?, backoffMaxMs?, jitter?, timeoutMs?, pickupTimeoutMs?, requires? }) | input / output are opt-in Standard Schemas (zod, valibot, arktype, …) validated at the serve boundary; a failure is a non-retryable SchemaValidationError. The rest is the def-level dispatch policy ctx.step reads (a per-call opts overrides it field by field). See Retries & backoff. |
StepDecoratorOptions field | Type | Notes |
|---|---|---|
name | string | Explicit routing name. |
input / output | StandardSchemaV1 | Runtime schemas for the step's input and output — any Standard Schema library. The handler gets the input schema's output value; the caller gets the output schema's output value. |
retries | number | Max attempts before the step (and run) fails. |
backoff | BackoffStrategy | 'fixed' or 'exp'. |
backoffMs / backoffMaxMs | number | Base delay and upper bound in ms. |
jitter | boolean | Random 50–100% jitter on the delay. |
timeoutMs | number | Liveness window: no result or heartbeat within it fails the dispatch with RemoteStepTimeout (retryable). |
pickupTimeoutMs | number | Liveness window before the worker's first heartbeat (queue wait plus pickup). Defaults to timeoutMs. See pickupTimeoutMs. |
requires | string[] | Capabilities a live worker must advertise; with none live, the run parks blocked. See Capability-aware routing. |
| Export | Signature | Notes |
|---|---|---|
defineStep | defineStep<TInput, TOutput>(name: string, fn: (input: TInput, log: StepLogger) => TOutput | Promise<TOutput>, config?: Omit<StepDecoratorOptions, 'name'>): StepRef<TInput, TOutput> | A step handler without a class or decorator. With an input / output schema and an unannotated handler, the types are inferred: StepRef<InferInput<input>, InferOutput<output>>, handler argument InferOutput<input>. See defineStep. |
StepRef<TInput, TOutput> | ((input: TInput) => Promise<TOutput> | TOutput) & { [DURABLE_STEP_NAME]?: string; [DURABLE_STEP_CONFIG]?: StepConfig } | A typed handle onto a stamped step name and policy. |
StandardSchemaV1 / validateStandard | validateStandard(schema, value, label): Promise<InferOutput<schema>> | The vendored Standard Schema V1 type, and the helper every schema slot uses (sync or async validate; throws SchemaValidationError). An optional fourth argument sets the error's code ('invalid_input', 'invalid_output', default 'schema_validation'). |
InferSearchAttributes / readSearchAttributes / attributesOf | attributesOf(WorkflowClass, run): A | Infer a searchAttributes schema's shape; read a run's attributes typed through a schema or by workflow class (lenient: history that fails the schema reads as {}). |
stepNameOf / stepConfigOf | stepNameOf(ref: unknown): string | undefined / stepConfigOf(ref: unknown): StepConfig | undefined | Read the stamped name or policy off a reference. |
Class decorators
| Decorator | Signature | Notes |
|---|---|---|
@Scheduled | Scheduled(config: WorkflowScheduleConfig | WorkflowScheduleConfig[]): ClassDecorator | Decorator form of static schedule; stacks in source order. See The @Scheduled decorator. |
@OnEvent | OnEvent(config: EmitterTriggerConfig): ClassDecorator | Start the workflow on an exact AdonisJS emitter event ({ event: string }). Stamps static on. See Event-triggered workflows. |
@OnDiagnostic | OnDiagnostic(config: DiagnosticsTriggerConfig): ClassDecorator | Start the workflow on an @adonis-agora/diagnostics channel ({ lib: string; event?: string | RegExp }); omit event for every event of the lib. |
onError types
| Type | Shape | Notes |
|---|---|---|
OnErrorRoute | { code?: string; workflow: string } | A route with names resolved. The engine's register / registerRemote take onError?: OnErrorRoute[]. |
OnErrorInput<TInput> | { runId: string; workflow: string; input: TInput; error: StepError } | What a handler workflow is started with. See onError. |
WorkflowEngine (selected public surface)
The engine singleton is the default export of @adonis-agora/durable/services/main (bound on the standalone and control-plane roles only; a tenant pod uses runGateway from the same module, see Roles & config). It is also resolvable from the container as WorkflowEngine. This is a curated subset, not the full class.
import engine from '@adonis-agora/durable/services/main'Registering
| Member | Signature | Notes |
|---|---|---|
register | register(name: string, version: string, fn: WorkflowFn, opts?: { tags?, singleton?, concurrency?, executionTimeout?, validateInput?, inputSchema?, searchAttributesSchema?, onEvent?, eventBatch?, requires?, origin?, onError?, deadLetterWorkflow? }): void | Register a function-form workflow (WorkflowFn is (ctx: WorkflowCtx, input: unknown) => Promise<unknown>). app/workflows discovery calls this for you. See Workflows & steps. |
registerRemote | registerRemote(name: string, version: string, opts: { group: string; executor: WorkflowExecutor; tags?; singleton?; concurrency?; executionTimeout?; validateInput?; inputSchema?; requires?; onError?; deadLetterWorkflow? }): void | Register a workflow whose body runs in another SDK; the engine still owns its state. inputSchema (any Standard Schema) runs before validateInput, which sees the parsed value. See Cross-ecosystem interop. |
remote | remote(name: string, opts: { group: string; version?: string; tags?; singleton?; concurrency?; executionTimeout?; validateInput?; inputSchema?; requires?; onError?; deadLetterWorkflow? }): void | registerRemote with the executor built over the engine's primary transport; version defaults to '1'. Throws without a transport. |
handle | handle(name: string, fn: StepHandler): void | Serve a step body by name on the engine's transport. It makes the engine a StepServer, so registerStepsFromDir(engine, dir) serves an extra directory from a preload. Throws when the transport can't serve handlers. See Registering an extra directory by hand. |
registerQueue | registerQueue(config: QueueConfig): void | Register a flow-control queue ({ name, concurrency?, rateLimit?, retryMs?, fairness?, order? }). See Flow control. |
registerEntity | registerEntity<S>(name: string, config: EntityConfig<S>): void | Register a durable entity ({ initialState: () => S; handlers: Record<string, EntityHandler<S>> }). See Durable entities. |
registerUpdateValidator | registerUpdateValidator<TArg>(workflow: string, name: string, validate: UpdateValidator<TArg>): void | Register (or replace) the validator gating update for a (workflow, name) pair. See Queries & updates. |
Starting and delivering
| Member | Signature | Notes |
|---|---|---|
start | start<C extends WorkflowClass>(workflow: C, input: WorkflowInputOf<C>, runId: string, opts?: StartOptions): Promise<RunResult> | Create and enqueue a run. Idempotent by runId. Also takes a workflow name (start<TInput>(workflow: string, input: TInput, runId: string, opts?)). |
waitForRun | waitForRun(runId: string, opts?: { timeoutMs?: number; terminal?: boolean }): Promise<RunResult> | Resolve once the run settles (terminal or suspended; a run suspended while it awaits a remote workflow turn's decision is not settled yet). terminal: true waits past suspensions for a terminal status. |
signal | signal(token: string, payload: unknown): Promise<RunResult | null> | Deliver a signal; buffered if no run is waiting yet (null). On a transport with a run queue the result is the run's status once the wake is handed over, not after the resumed turn; waitForRun gives the outcome. See Signals. |
signalWithStart | signalWithStart(workflow: WorkflowRef, input: unknown, runId: string, signal: { token: string; payload?: unknown }, opts?: StartOptions): Promise<{ runId: string }> | Ensure the run exists, then signal it, race-free. |
publishEvent | publishEvent(name: string, payload: unknown, opts?: { id?: string; buffer?: boolean }): Promise<number> | Resume every run parked on ctx.waitForEvent(name, …) and start every onEvent subscriber; returns the count touched. A publish that reaches nobody is buffered once unless buffer: false. See Named events. |
update | update(runId: string, name: string, arg: unknown): Promise<UpdateResult> | Deliver a validated update to ctx.onUpdate(name). A rejection never touches the run. |
getEvent | getEvent<TValue = unknown>(runId: string, key: string): Promise<TValue | undefined> | Read the latest value a run published with ctx.setEvent. |
completeTask | completeTask(runId: string, name: string, result: unknown): Promise<RunResult | null> | Report a ctx.task(name, …) result back to its run. See External tasks. |
failTask | failTask(runId: string, name: string, error: string): Promise<RunResult | null> | Report a ctx.task failure; the run resumes and throws a FatalError at the task. |
signalEntity | signalEntity(name: string, key: string, op: string, arg?: unknown): Promise<void> | Send an entity op from outside a run. |
getEntityState | getEntityState<S = unknown>(name: string, key: string): Promise<S | undefined> | Read an entity's current durable state. |
Controlling runs
| Member | Signature | Notes |
|---|---|---|
cancel | cancel(runId: string, opts?: { compensate?: boolean }): Promise<RunResult | null> | Cancel a run. { compensate: true } resumes it first so its saga undos run in reverse. See Compensating cancellation. |
cancelWhere | cancelWhere(filter: Omit<RunQuery, 'page' | 'size'>, opts?: { compensate?: boolean }): Promise<RunResult[]> | Bulk-cancel every run matching a RunQuery filter, each through cancel. |
resume | resume(runId: string): Promise<RunResult> | Re-execute a run's body. Explicit, so it also re-drives a failed run (prefer requeue). Completed, cancelled and dead runs are never re-executed. |
requeue | requeue(runId: string): Promise<RunResult | null> | The dispatch-model retry: set the run back to pending, clear its lease and dispatch it. Returns without running the body inline. (A second parameter is internal.) |
redispatchPending | redispatchPending(runId: string): Promise<(RunResult & { redispatched: number }) | null> | Re-dispatch every remote step of a run stuck pending after a lost job. See Failure modes. |
retryWithInput | retryWithInput(runId: string, input: unknown, newRunId?: string): Promise<{ runId: string } | null> | Fix-and-replay: a new run (default id <runId>~retry~<uuid>) with corrected input and clean history. See Dead-letter queue. |
continue | continue(runId: string): Promise<RunResult | null> | Resume a run paused at ctx.breakpoint. |
deleteRun | deleteRun(runId: string): Promise<number> | Hard-delete a run and its child subtree; returns the count removed. |
purgeRuns | purgeRuns(scope: RunScope, opts?: { batchSize?: number; cancelLive?: boolean; children?: boolean }): Promise<number> | Hard-delete every run matching scope, in batches. See Purging on demand. |
purgeNamespace | purgeNamespace(namespace: string, opts?: { batchSize?: number; cancelLive?: boolean }): Promise<number> | purgeRuns({ namespace }). |
Reading
| Member | Signature | Notes |
|---|---|---|
getRun | getRun(runId: string): Promise<WorkflowRun | null> | A run's current persisted state. |
listRuns | listRuns(query: RunQuery): Promise<WorkflowRun[]> | Filter by workflow, status, tags, namespace and search attributes. See Querying by attribute. |
countRuns | countRuns(query: Omit<RunQuery, 'page' | 'size'>): Promise<number> | Exact count through the store when it supports one. |
listCheckpoints | listCheckpoints(runId: string): Promise<StepCheckpoint[]> | A run's checkpoint timeline. |
getRunDetail | getRunDetail(runId: string, options?: RunDetailOptions): Promise<RunDetail | null> | The run, its timeline and children; with summarizeFansOver every wider fan is summarized into fans. When summarizeFansOver is set and the store has checkpointStats + queryCheckpoints (the Lucid store does), the detail is built from those aggregates and carries an etag; options.ifNoneMatch naming it returns { notModified: true } without reading the timeline (the dashboard's 304). See Run detail and fan types. |
getFanMembers | getFanMembers(runId: string, group: string, query: FanPageQuery): Promise<FanPage | null> | One page of a summarized fan, versioned like getRunDetail (query.ifNoneMatch). |
runFacets | runFacets(query: RunFacetQuery): Promise<RunFacetRow[]> | How the matching runs split across (status, origin): one GROUP BY through the store's runFacets, else counted over listRuns. |
runValueFacets | runValueFacets(axis: RunValueAxis, query: RunFacetQuery, opts?: RunValueFacetOptions): Promise<RunValueFacetRow[]> | The distinct values of one filter axis (workflow, tag, …) over the matching runs, with counts: what a console's pickers list. Delegates to the store's runValueFacets, else counts a bounded scan (RUN_VALUE_FACET_SCAN, 5000 runs). |
getRunChildren | getRunChildren(parentRunId: string): Promise<string[]> | Ids of the runs this run spawned, stable after they finish. |
workerHealth | workerHealth(extra?: string[]): Promise<GroupHealth[]> | Per-group queue backlog and live worker heartbeats, over knownGroups(extra) plus every group seen in a heartbeat. Empty unless the transport can introspect health (the bullmq, queue and db transports). |
knownGroups | knownGroups(extra?: string[]): string[] | The worker groups this engine dispatches to: every registered remote workflow's group plus extra. Pass local-step groups in extra so a group with backlog and no workers still reports. |
workflowDirectory | workflowDirectory(): Promise<WorkflowDirectory> | Every workflow the live fleet can run in this engine's partition (workflows, each declared or observed), whether the fleet could be asked at all (supported), workers live on other partitions (otherPartitions) and a one-line detail. See What can I call?. |
announcedWorkflows | announcedWorkflows(): Promise<AnnouncedWorkflow[]> | workflowDirectory()'s workflows. |
unservedOnErrorRoutes | unservedOnErrorRoutes(): Promise<{ workflow: string; route: OnErrorRoute }[]> | onError routes whose handler nothing can run. See Catching a misnamed handler. |
Schedules
| Member | Signature | Notes |
|---|---|---|
listSchedules | listSchedules(now?: number): Promise<ScheduleInfo[]> | Config and colocated schedules with their live state. See Runtime control. |
setSchedulePaused | setSchedulePaused(key: string, paused: boolean): boolean | Pause or resume a schedule, fleet-wide over the control plane. |
triggerSchedule | triggerSchedule(key: string, now?: number): Promise<RunResult | null> | Start the current window now (idempotent by its run id). |
schedules | readonly schedules: ScheduleClient — create / upsert / get / list / pause / resume / trigger / delete / tick | Persisted, runtime-managed schedules stored in durable_schedules, fired when persistedSchedules is on. create throws ScheduleAlreadyExistsError; the others throw ScheduleNotFoundError on an unknown id. See Persisted schedules. |
runSchedules(engine, schedules, nowMs, opts?) is a standalone function, not an engine method: it starts each ScheduledWorkflow's current time-bucket run (idempotent through engine.start). The durable:work loop calls it for you. resolveSchedules(schedules) resolves the lazy imports and classes to registered names first (async); runSchedules refuses a schedule whose lazy import was not resolved. See Lower-level — runSchedules.
Listening and middleware
| Member | Signature | Notes |
|---|---|---|
subscribe | subscribe(listener: EngineListener): () => void | Every EngineEvent the engine emits. See Events & interceptors. |
use | use(interceptor: StepInterceptor): () => void | Onion middleware around real local-step execution; never fires on replay. |
onDead | onDead(listener: (run: WorkflowRun) => void): () => void | Runs moved to dead past maxRecoveryAttempts. See Dead-letter queue. |
onEnqueued | onEnqueued(listener: (runId: string) => void): () => void | Runs enqueued on another instance (control-plane broadcast). |
onCancel | onCancel(listener: (runId: string) => void): () => void | Cancellations on any instance, for cooperative abort. |
onStalled | onStalled(listener: (info: StalledRunInfo) => void): () => void | Stranded in-flight runs, paged by sweepStalled after stalledAfter. See The stalled-run pager. |
onEvict | onEvict(listener: (run: WorkflowRun, checkpoints: StepCheckpoint[]) => void | Promise<void>): () => void | Archive a run before retention deletes it; a throwing hook skips that deletion. See Retention. |
Worker loop
These are the phases node ace durable:work (or the embedded worker) drives on each tick. Call them yourself only when composing your own loop. See CLI.
| Member | Signature | Notes |
|---|---|---|
runPending | runPending(nowMs?: number): Promise<RunResult[]> | Pick up and execute every pending run. |
runOne | runOne(runId: string): Promise<RunResult | null> | Execute one run, e.g. from onEnqueued. |
consumeRuns | consumeRuns(): (() => Promise<void>) | undefined | Consume this engine's namespace's run queue and return the consumer's stop. The engine calls it itself when built (a second call returns the same stop); undefined without a run queue or with a custom runDispatcher. drain() stops it. |
runEnqueued | runEnqueued(run: RunDispatch | string): Promise<void> | Take one run delivered by the transport's run queue and act on its state: a pending run is run, a suspended one is leased and resumed, a finished one is skipped. The engine calls it from its own run-queue consumer. |
recoverIncomplete | recoverIncomplete(nowMs?: number): Promise<RunResult[]> | Resume runs left running by a crash or deploy. |
resumeDueTimers | resumeDueTimers(nowMs?: number): Promise<RunResult[]> | Resume suspended runs whose timer is due, at most timerBatchSize per pass. |
sweepTimeouts | sweepTimeouts(now?: number): Promise<void> | Cancel runs past their executionTimeout, cascading to children. |
sweepRetention | sweepRetention(now?: number): Promise<number> | Apply the retention policy (throttled to once a minute). |
sweepStalled | sweepStalled(now?: number): Promise<StalledRunInfo[]> | Detect stranded runs and notify onStalled listeners. |
startConsumers | startConsumers(): void | Start broker consumption deferred by consumers. Idempotent. |
beginDrain / isDraining | beginDrain(spentMs?: number): void / isDraining(): boolean | Stop picking up new work without waiting. spentMs reports shutdown wait already spent (e.g. on an in-flight tick); it is deducted from the next drain budget, and from one already waiting. Non-finite or negative values are ignored. |
drain | drain(timeoutMs?: number): Promise<void> | Graceful shutdown: stop new work and wait for in-flight executions (default 10000 ms), minus any wait reported through beginDrain(spentMs). |
StartOptions
| Option | Type | Notes |
|---|---|---|
version | string | Pin the run to this registered version instead of the newest. An unregistered version throws before any run row exists; never falls back. Convention routing cannot honour a pin. |
tags | string[] | Merged with the workflow's static tags. |
searchAttributes | SearchAttributes | Typed, queryable run data (Record<string, string | number | boolean>). |
priority | number | Dispatch priority for a remote run; higher wins. |
namespace | string | Stamp this run into another worker pool. See Tenancy. |
startAt | number | Date | Create now, park suspended, start when due. See Delayed starts. |
origin | string | Override the registration's package attribution for this run. |
concurrency | ConcurrencyQuota ({ key: string; limit: number; countStatuses?: RunStatus[] }) | A per-start quota that overrides the registered one. See Concurrency quotas. |
RunGateway
The read, control and start surface every role exposes, as runGateway from @adonis-agora/durable/services/main. On standalone and control-plane it is a StoreRunGateway over the engine; on a tenant pod it is a ProxyRunGateway that round-trips each call over the transport to the control plane's RunRequestResponder. The dashboard and the CLI read through it, so code written against it runs unchanged on every role. See Reading and steering runs.
import { runGateway } from '@adonis-agora/durable/services/main'
const detail = await runGateway.getRunDetail?.(runId, { summarizeFansOver: 50 })| Member | Signature | Notes |
|---|---|---|
topology | topology(): DurableTopology | { role: 'control-plane' | 'standalone' | 'tenant'; tenant? }. |
getRun / listRuns | getRun(runId): Promise<WorkflowRun | null> / listRuns(query: RunQuery): Promise<WorkflowRun[]> | As on the engine. |
getCheckpoints / getRunChildren | (runId: string) => Promise<StepCheckpoint[]> / Promise<string[]> | A run's timeline (the engine's listCheckpoints) and its child ids. |
getRunDetail? / getFanMembers? | as on the engine | Summarized detail and fan pages, ETag-versioned (ifNoneMatch → notModified). Optional: absent on a gateway that predates them. |
getSearchAttributes | getSearchAttributes(runId: string): Promise<SearchAttributes | undefined> | A run's search attributes. |
workerHealth | workerHealth(): Promise<GroupHealth[]> | Per-group backlog and live workers. |
start | start(workflow: WorkflowRef, input: unknown, opts?: StartRunOptions): Promise<RunResult> | StartRunOptions is StartOptions & { runId? }; a missing runId is minted. Idempotent by run id. |
signal | signal(runId: string, signal: string, payload?: unknown): Promise<RunResult | null> | Deliver a signal by token. runId lets a proxy route to the owning control plane; an unclaimed payload is buffered. |
cancel | cancel(runId: string, opts?: { compensate?: boolean }): Promise<RunResult | null> | As on the engine. |
redispatchPending | redispatchPending(runId: string): Promise<(RunResult & { redispatched: number }) | null> | As on the engine. |
retry? / continue? / retryWithInput? | retry(runId) (the engine's requeue), continue(runId), retryWithInput(runId, input) | Optional control verbs. Against a control plane that predates them (it answers unsupported-request), the dashboard degrades: retry falls back to redispatchPending, continue and retryWithInput answer null (the handlers 404). |
runFacets? / runValueFacets? | as on the engine | Status/origin split and per-axis value counts for the runs list. Optional; against an older control plane the dashboard counts a bounded listRuns scan instead. |
waitingFor? | waitingFor(runIds: string[]): Promise<Record<string, RunWaiting>> | What each suspended run is parked on (signal, webhook, child, breakpoint), for the ones parked on something. Against an older control plane the list rows carry no waiting stamp. |
subscribe | subscribe(runId: string, onEvent: (event: EngineEvent) => void): () => void | Live lifecycle events of one run (the dashboard's live tail). |
| Export | Signature | Notes |
|---|---|---|
StoreRunGateway | new StoreRunGateway(engine: RunGatewayEngine, options?: { role?: 'standalone' | 'control-plane' }) | The store-backed gateway the provider binds on the store roles. |
ProxyRunGateway | new ProxyRunGateway(transport: ProxyTransport, options: { partition: string; token?: string; requestTimeoutMs?: number }) | The tenant pod's gateway. requestTimeoutMs is the config's requestTimeoutMs. |
RunRequestResponder | new RunRequestResponder(transport: ResponderTransport, gateway: RunGateway, options?: { verifyTenant?; subscribeEngineEvents? }) | Answers proxied run requests on the control plane; a request it cannot serve is answered UNSUPPORTED_RUN_REQUEST. |
signTenantToken | signTenantToken(tenant: string, secret: string, opts?: { ttlMs?: number; now?: number }): string | Mint a pod's HMAC tenant claim (config.tenant.token); ttlMs makes it expire. |
hmacTenantVerifier | hmacTenantVerifier(secret: string | string[], capabilities?: string[]): TenantVerifier | The matching verifyTenant. A list of secrets allows rotation. See Layered tenant authentication. |
Run detail and fan types
What getRunDetail and getFanMembers return, on the engine and the RunGateway, and the helpers they are built from.
| Type / export | Shape | Notes |
|---|---|---|
RunDetail | { run, timeline, children, fans?, childCount?, etag?, notModified? } | When summarized, the members of a wide fan are left out of timeline and described in fans, and children lists at most SUMMARY_CHILD_LIMIT (200) ids, childCount the total. notModified: true comes back with an empty body when ifNoneMatch matched. |
RunDetailOptions | { summarizeFansOver?: number; ifNoneMatch?: string } | Summarize every fan with more members than summarizeFansOver. ifNoneMatch is ignored when no etag can be computed (no summarizeFansOver, or a store without checkpoint aggregates). |
FanSummary | { group, label, firstSeq, lastSeq, size, total?, notStarted?, failFast, counts, startedAt?, finishedAt?, failures } | One summarized fan; failures holds at most FAN_SUMMARY_FAILURES (20) failed members. |
FanPageQuery | { offset?, limit?, failedOnly?, ifNoneMatch? } | offset defaults to 0, limit to 100, clamped to 1–500. |
FanPage | { items: StepCheckpoint[]; total: number; etag?; notModified? } | One page of a fan's members in seq order. |
summarizeTimeline / summarizeFan / pageFan | summarizeTimeline(timeline, over) → { timeline, fans }; summarizeFan(group, members) → FanSummary; pageFan(timeline, group, query) → FanPage | The in-memory versions, over a loaded checkpoint list. |
summarizeRunFromStore / pageFanFromStore | (store, runId, over | group, …) | The same, read from a store's checkpointStats / queryCheckpoints aggregates. |
runDetailEtag / fanPageEtag / etagMatches | → string / → string / (ifNoneMatch, etag) → boolean | The ETags the detail and fan reads carry, and the weak If-None-Match comparison (* included). |
childIdsOf / parentRunIdOf | childIdsOf(checkpoints): string[] / parentRunIdOf(runId): string | undefined | The child ids a timeline references; the parent of an engine-minted child id (also stamped as parentRunId on diagnostics events). |
DurableConfig
defineConfig(config) in config/durable.ts returns a DurableConfig, a union discriminated on role: StandaloneConfig (the default when role is omitted), ControlPlaneConfig or TenantConfig. All roles share the fields below. See Roles & config.
import { defineConfig, stores, transports } from '@adonis-agora/durable'
export default defineConfig({
transport: 'queue',
transports: { queue: transports.queue({ connection: 'main' }) },
store: 'lucid',
stores: { lucid: stores.lucid({ connection: 'pg' }) },
})| Option | Type | Default | Notes |
|---|---|---|---|
role | 'standalone' | 'control-plane' | 'tenant' | 'standalone' | Explicit topology, never inferred. See Cluster. |
transport | string | in-process | Key of transports to dispatch over. See Transports. |
transports | Record<string, TransportFactory> | — | Named transports, built with the transports factory. |
store | string | in-memory | Key of stores for runs, checkpoints and timers. See State stores. |
stores | Record<string, StoreFactory> | — | Named stores, built with the stores factory. |
autoSchema | boolean | true | Provision the store's schema at boot with ensureSchema(). Set false to migrate with createDurableTables yourself. See Who provisions the schema. |
controlPlane | ControlPlane | ControlPlaneFactory | single-instance | Cross-instance lifecycle and cancellation broadcast. See Control plane. |
admission | AdmissionBackend | AdmissionFactory | in-process caps | Backend for flow-control queues; admissions.redis(...) makes caps fleet-wide. See Distributed admission. |
queues | QueueConfig[] | none | Flow-control queues registered at boot, the declarative engine.registerQueue. A duplicate name throws at boot; a later engine.registerQueue of the same name replaces it. See Flow control. |
deadLetterWorkflow | WorkflowRef (name, class or ref) | none | Fleet-default handler for dead runs, started as dlq:<deadRunId> with a DeadLetter input in the dead run's namespace. A workflow's deadLetter() method or own deadLetterWorkflow wins. See Dead-letter queue. |
webhookUrl | (token: string) => string | none | Builds DurableWebhook.url for ctx.webhook(). See Durable webhooks. |
traceparent | () => string | undefined | wired by @adonis-agora/diagnostics-otel when installed | W3C traceparent stamped on dispatched tasks. See OpenTelemetry. |
trackStepStart | boolean | true | Persist a running checkpoint when a local step starts; false halves checkpoint writes on hot paths. |
leaseMs | number | 30000 | Recovery lease duration. See Delivery under multiple instances. |
instanceId | string | random | Unique per process; run leases are held by it. |
namespace | string | 'default' on standalone; unset (operator) on control-plane | Worker-pool partition; poll and recovery paths act only on this namespace. A control-plane with no namespace is the operator and drives every pool (before 0.51 an unset namespace meant operator on every role). See Tenancy. |
maxRecoveryAttempts | number | unlimited | Crash-recovery pickups before a run moves to dead. See Dead-letter queue. |
compensationRetries | number | 1 | Attempts per saga compensation on run failure. See Retrying a transient undo. |
compensationTimeoutMs | number | 300000 | Upper bound on awaiting one dispatched compensation without its own timeoutMs. |
retention | { completed?, failed?, cancelled?, dead?: string | number; policies?: Array<{ statuses: TerminalRunStatus[]; maxAge?: string | number; maxCount?: number; scope: RunScope }> } | keep everything | Hard-delete terminal runs past an age, per status or per scope. A policy needs maxAge and/or maxCount (keep the N newest by start time within the scope). See Run retention. |
stalledAfter | string | number | '15m' | Idle time before onStalled listeners are paged. |
runDispatcher | RunDispatcher | in-process | Where a freshly started run executes. Wins over drive. |
drive | boolean | true | false: this process never executes the runs it starts, requeues or recovers (no-op dispatcher); they stay pending for durable:work or a control plane. Directly requested resumes (a signal or webhook handled here) still execute here. Throws at boot with worker: { embedded: true }. durable:work drives regardless. See Where a run actually executes. |
drainTimeoutMs | number | 10000 | Shutdown budget: on app shutdown every store-backed process (standalone, control-plane) waits this long for in-flight executions before closing the transport and control plane. A worker.drainTimeoutMs, when set, takes precedence; the embedded loop drains within that budget and is not drained twice. |
consumers | 'auto' | 'always' | 'never' | 'auto' | When this process's transport starts broker consumers; 'auto' defers in console/repl. See consumers. |
remoteRedispatchMs | number | off | Opt-in self-heal: re-dispatch a no-timeoutMs remote step still pending past this window. The step must be idempotent. See Failure modes. |
remoteRedispatchMax | number | 10 | Re-dispatches per lost step before failing it (remote_step_lost). |
remoteAdvanceSilenceMs | number | off | Liveness window of a remote workflow turn: with no decision and no run-scoped heartbeat for this long, the turn is re-dispatched (fresh taskId, same history) instead of waiting forever; heartbeats rearm it. A RemoteWorkflowExecutor's timeoutMs is the per-workflow version (the shorter wins). See Remote workflow turns. |
timerBatchSize | number | 500 | Most due timers one timer pass loads, most overdue first. |
eventBufferTtl | string | number | keep until consumed | Age after which a buffered event is expired: never delivered, and deleted by the worker tick's retention phase (StateStore.pruneBufferedEvents). |
scopeReads | boolean | false | Fence the store's run reads to namespace (a tenant-boundary view) through the store's withScope. Writes are unaffected. See Lucid store. |
reconcileMs | number | 300000 | Fallback re-drive for a run suspended on an event (child, signal, remote result) whose wake-up was lost; also the wake a recovered run with a remote step in flight is parked on (a remote workflow's run whose turn was lost is re-driven after about a second instead). 0 disables it; a non-finite or negative value falls back to the default. |
blockedPollMs | number | 5000 | How often a run parked blocked (no capable, compatible live worker) re-checks the fleet. A non-finite or sub-1 value falls back to the default. |
schedules | ScheduledWorkflow[] | none | Recurring workflows (interval or cron), merged with colocated static schedule. Name each workflow with a lazy import, () => import('#workflows/…') (resolved when the worker starts, after boot), a class, or a registered name; ScheduledWorkflow<LazyWorkflowImport<typeof X>> types input. See Scheduling. |
persistedSchedules | boolean | false | Fire the runtime-managed engine.schedules on each worker tick. |
worker | EmbeddedWorkerConfig ({ embedded?: boolean; intervalMs?: number; drainTimeoutMs?: number }) | no embedded loop | Run the worker loop inside the web process. See Running the loop in the web process. |
workflowsPath | string | string[] | false | 'app/workflows' | Directory (or directories) scanned for BaseWorkflow classes; false or [] disables discovery. A workflow name + version defined in two directories fails boot. See app/workflows. |
stepsPath | string | string[] | false | 'app/steps' | Directory (or directories) scanned for @Step classes and defineStep handlers; false or [] disables. A step name defined in two directories fails boot. |
Role-specific fields:
| Role | Field | Type | Notes |
|---|---|---|---|
control-plane | store, transport | string | Both required. |
control-plane, standalone | verifyTenant | TenantVerifier | Verifies tenant tokens on incoming requests (a standalone engine answers tenant pods too when its transport carries the store-less protocol). See Layered tenant authentication. |
tenant | store, stores | never | A store-less pod may not name a store; any value is a compile error. |
tenant | transport, partition | string | Both required. partition suffixes the pod's <name>@<tenant> routing tokens. |
tenant | tenant | { token?: string } | The pod's signed claim. |
tenant | capabilities | string[] | Extra advertised features beyond registered handler names. |
tenant | requestTimeoutMs | number | ProxyRunGateway round-trip timeout. See Thin workers. |
Driver factories
Each factory returns a lazy thunk, so a driver's peer dependency is imported only when that driver is selected.
| Factory | Signature | Notes |
|---|---|---|
transports.memory | memory(_config?: MemoryTransportConfig): TransportFactory | Test-only in-process transport (the default when no transport is named). See In-process for tests. |
transports.eventEmitter | eventEmitter(config?: EventEmitterTransportConfig): TransportFactory | Production in-process transport and control plane over a Node EventEmitter. See Single-process production. |
transports.queue | queue(config?: QueueTransportConfig): TransportFactory | Cross-process over @adonisjs/queue. See Queue transport. |
transports.bullmq | bullmq(config: BullMQTransportConfig): TransportFactory | Cross-process over BullMQ (bullmq / ioredis imported lazily). Its options include pingIntervalMs (default 30000) and the opt-in stepTimeoutMs, a wall-clock ceiling per step handler past which the step fails retryable (wedged steps). See BullMQ transport. |
transports.db | db(config?: DbTransportConfig): TransportFactory | Cross-process over the app's database through @adonisjs/lucid, no broker. See DB transport. |
stores.lucid | lucid(config?: LucidStoreConfig): StoreFactory | Persist in SQL through @adonisjs/lucid ({ connection?: string; collation?: string }). See Lucid store. |
controlPlanes.redis | redis(config?: RedisControlPlaneConfig): ControlPlaneFactory | Lifecycle and cancellation broadcast over @adonisjs/redis pub/sub. See Multi-replica: controlPlanes.redis. |
admissions.redis | redis(config?: RedisAdmissionConfig): AdmissionFactory | Fleet-wide flow-control caps in one atomic Lua script on Redis. See Distributed admission. |
The concrete classes (InMemoryTransport, EventEmitterTransport, QueueTransport, BullMQTransport, DbTransport, InMemoryStateStore, LucidStateStore, RedisControlPlane) are exported too, for passing an instance directly or writing tests. Schema helpers for a hand-written migration: createDurableTables(db: Database, connectionName?: string, options?: CreateDurableTablesOptions): Promise<void>, dropDurableTables(db: Database, connectionName?: string): Promise<void> (CreateDurableTablesOptions is { logger?, collation?, force? }; DURABLE_SCHEMA_REVISION, DURABLE_SCHEMA_META_TABLE and DURABLE_JSON_COLUMNS describe what it manages), onCorruptJsonColumn(reporter) to route the Lucid store's report of a JSON column that does not parse, createDurableTransportTables(db: Database): Promise<void> and dropDurableTransportTables(db: Database): Promise<void>. See Lucid store and DB transport.
Observability building blocks
Root exports for wiring the engine's lifecycle events into your own tooling. The /otel and /telescope subpaths hold the heavier integrations.
| Export | Signature | Notes |
|---|---|---|
collectMetrics | collectMetrics(engine: { subscribe }): MetricsCollector | Dependency-free counters; .prometheus() returns the text exposition, .snapshot() the raw MetricsSnapshot, .stop() unsubscribes. Counts only this instance's events. See Metrics. |
attachDurableDiagnostics | attachDurableDiagnostics(engine: DurableEventSource, options?: { includeRemote?: boolean }): () => void | Bridge every EngineEvent onto the @adonis-agora/diagnostics bus as durable / <event type>. The provider attaches it for you. includeRemote (default false) also forwards events heard from other instances over the control plane. See Diagnostics. |
isRemoteEvent | isRemoteEvent(event: EngineEvent): boolean | true for an event another instance produced and the control plane re-delivered here. Filter on it in a listener that records or counts, so a fleet counts each event once. |
withScheduleOrigin / asHeartbeat | <T>(fn: () => T) => T | Label fn as scheduler-driven work, or as a liveness probe an observability tool may drop. A no-op unless a driver (Telescope's) is installed. |
currentStep | currentStep(): StepLogger | undefined | The running step's logger. See currentStep(). |
Command building blocks
The pieces durable:work, durable:runs and durable:retry are built from, exported so an app can compose its own command or scheduler without re-implementing the loop. See Building your own.
| Export | Signature | Notes |
|---|---|---|
runTick | runTick(engine: WorkflowEngine, options?: { now?: number; schedules?: ScheduledWorkflow[] }): Promise<TickResult> | One worker tick, in phases: runPending, recoverIncomplete, resumeDueTimers, sweepTimeouts, schedules (config and, with persistedSchedules, persisted), sweepRetention, sweepStalled. TickResult is { pending, recovered, timers, scheduled, evicted, errors: { phase, error }[] }; a failing phase does not stop the others. |
runWorkerLoop | runWorkerLoop(engine: WorkflowEngine, options: WorkerLoopOptions): Promise<number> | runTick every intervalMs until stopSignal resolves, then drains. Starts deferred consumers first. WorkerLoopOptions is { intervalMs, stopSignal, logger: WorkerLogger, sleep?, schedules?, … }. |
listRuns | listRuns(source: RunLister, opts: { status?, statuses?, workflow?, limit? }): Promise<WorkflowRun[]> | The durable:runs query. |
attachLiveness / filterStale | attachLiveness(source, runs, now?): Promise<RunLiveness[]> / filterStale(liveRuns, thresholdMs = DEFAULT_STALE_MS) | Add each run's age and oldest pending remote step; keep the ones idle past the threshold (DEFAULT_STALE_MS is 15 minutes). |
renderRunsTable / staleHint / parseDurationMs | renderRunsTable(liveRuns): string, staleHint(runId): string, parseDurationMs(input): number | undefined | The durable:runs table, its remediation hint, and its --stale parser. |
retryRun | retryRun(engine, runId): Promise<RunResult | null> | engine.requeue(runId), what durable:retry calls. |
Discovery helpers
What the provider runs over workflowsPath / stepsPath, for a directory it does not scan (a preload, a package). See Registering an extra directory by hand.
| Export | Signature | Notes |
|---|---|---|
registerWorkflowsFromDir | registerWorkflowsFromDir(engine, dir: string | string[], factory?): Promise<WorkflowMeta[]> | Import every module under dir and register its BaseWorkflow classes. The returned metadata carries each workflow's resolved origin. |
registerWorkflowsFromBarrel | registerWorkflowsFromBarrel(engine, barrel, factory?, options?: { appRoot?: string | URL }): Promise<WorkflowMeta[]> | The same over a generated barrel (hooks/workflows). appRoot is the app the barrel was generated for: its modules derive their origin from it. Without it they're attributed to the calling package. |
registerStepsFromDir / registerStepsFromBarrel | (server: StepServer, dir | barrel) => Promise<DurableStepMeta[]> | Serve every @Step class and defineStep handler found. The engine is a StepServer. |
registerWorkflowClass / registerSteps | registerWorkflowClass(engine, cls, factory?) / registerSteps(server, exports) | Register one class, or the steps in a list of module exports. |
parseDuration | parseDuration(duration: string | number): number | The duration parser every '30s' / '7 days' option goes through. |
Errors
| Error | Thrown when |
|---|---|
FatalError | new FatalError(message, code?, details?). Throw it yourself in a step or workflow for an unrecoverable business failure (e.g. a declined card). It is never retried, in a local or a dispatched step, and fails the run at once. code and details travel with the failure to the run's error, an allSettled item, a ctx.child catch and an onError handler. See Fatal errors. |
SignalTimeoutError | The timeoutMs of ctx.waitForSignal, ctx.waitForEvent, ctx.onUpdate or a webhook's wait passes before delivery (token, timeoutMs). Catch it to take a default branch. |
NonDeterminismError | On resume, the step at a position has a different name or kind than its checkpoint: the code changed under an in-flight run without a new version (code: 'non_determinism', runId, seq). Never routed by onError. See The non-determinism guard. |
NestedWorkflowCallError | A ctx primitive (or a BaseWorkflow.start() / .dispatch() through the ambient ctx) was called from inside a step body (ctx.localStep, transaction, sideEffect, now, task's dispatch). A FatalError with code nested_workflow_call (runId, primitive, step). See Workflow primitives inside a step body. |
SchemaValidationError | A value failed a declared Standard Schema: a step's input / output, a workflow's inputSchema (thrown by start, nothing created), or its searchAttributes schema (thrown by ctx.upsertSearchAttributes). A FatalError, never retried, whose code says which schema rejected the value: invalid_input (a step's input, a workflow's inputSchema), invalid_output (a step's output) or schema_validation (a searchAttributes write, a direct validateStandard call). Its static code is schema_validation, and an onError route for it also catches invalid_input and invalid_output that no route of their own matches; issues / details.issues hold { message, path: string[] }[]. |
ConcurrencyLimitError | start whose quota key already has limit runs in flight (workflow, key, limit, active). Nothing was created. See Concurrency quotas. |
SingletonQueueFullError | start on a singleton workflow whose in-flight plus gated backlog equals limit + maxQueueDepth (workflow, key, maxQueueDepth). See Singleton workflows. |
GatherError | ctx.all: one or more children or steps of a fan failed. failures holds GatherItemFailure items ({ index, id, error, code?, details? }). See Child workflows. |
RemoteStepTimeout | A dispatched step produced no result or heartbeat within timeoutMs (stepId, timeoutMs). Retryable per the step's retries. See Retries & backoff. |
RemoteStepError | Raised inside the workflow when a remote worker reports a step failure (stepError?: StepError). |
RemoteWorkflowTimeout | An inline remote workflow advance produced no decision or run-scoped heartbeat within remoteAdvanceSilenceMs (taskId, timeoutMs). Not a failure: the engine releases the lease and crash recovery re-drives the run. With a call of the lost turn still in flight, recovery parks the run on a wake about a second out (not the reconcileMs window) and the timer pass dispatches a fresh turn; the call is not dispatched again. |
NamespaceMismatch | engine.resume on a run that belongs to a different namespace. See Tenancy. |
DuplicateDiscoveryError | Boot, when two workflowsPath / stepsPath directories (or two roots of a generated barrel) define the same workflow name + version, or the same step name, with different classes or handlers (kind, id, files). The message names both files. See Several directories. |
ScheduleAlreadyExistsError / ScheduleNotFoundError | engine.schedules.create on a taken id / any engine.schedules.* call on an unknown id (scheduleId). |
ContinueAsNew | Thrown by ctx.continueAsNew(input) to unwind the run and hand off to a fresh execution. A control-flow signal. |
WorkflowSuspended | Thrown by the engine to stop a turn and persist the suspension (wakeAt, waitSeqs). A control-flow signal. |
The thin-worker turn runtime (@adonis-agora/durable/worker, also re-exported from the root) has its own set: WorkflowStepFailedError, WorkflowGatherFailedError, WorkflowNondeterminismError and WorkflowTurnCancelled. See Thin workers.
Control-flow signals
WorkflowSuspended and ContinueAsNew are thrown internally to unwind the current turn (a durable sleep, a still-pending step, ctx.continueAsNew). They are not failures. Because they use the same exception mechanism as a real step failure, a workflow's own try/catch around ctx.step, ctx.sleep and so on can catch one, and it must rethrow it untouched before running any cleanup. Otherwise the cleanup is recorded as extra history during what replay later sees as a plain suspend, and the run fails with NonDeterminismError on resume. See Catching errors and Sagas.
| Export | Signature | Notes |
|---|---|---|
CONTROL_FLOW_SIGNAL | const CONTROL_FLOW_SIGNAL: unique symbol | Symbol.for('@agora/durable:control-flow'), the global-registry marker stamped on every control-flow signal class. Rarely used directly. |
isWorkflowControlFlowSignal | isWorkflowControlFlowSignal(error: unknown): boolean | true for WorkflowSuspended and ContinueAsNew, whichever module copy threw them. Not true for a cancellation or a FatalError / step rejection, which are real outcomes a catch may handle. |
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
throw error
}Statuses
Run status (RunStatus)
| Status | Meaning |
|---|---|
pending | Created and enqueued by start, not yet picked up. |
running | A worker is executing (or replaying) the body. |
suspended | Parked with zero compute on a sleep, a signal, event or update, an awaited step or a child. |
blocked | Parked because no live worker can run the next dispatch (a required capability is unadvertised, or every live worker is protocol-incompatible). Not terminal: the blocked-recovery poll re-drives it once a capable worker appears. See Handshake. |
completed | Finished successfully. Terminal. |
failed | Finished with an unrecovered error. Terminal. |
cancelled | Cancelled, with or without compensation, or past its executionTimeout. Terminal. |
dead | Dead-lettered: crash recovery gave up after maxRecoveryAttempts. Terminal. |
Only completed, failed, cancelled and dead are terminal (TerminalRunStatus, listed in TERMINAL_RUN_STATUSES). Only those are eligible for retention pruning, and they never transition on their own again.
Checkpoint status (StepCheckpoint['status'])
| Status | Meaning |
|---|---|
pending | A remote step was dispatched and awaits its worker's result; the run is durably suspended. |
running | A local step's body is executing in-process right now (see trackStepStart). |
completed | Settled successfully. Short-circuits on replay: the step is not re-executed. |
failed | Settled with an error. Re-awaited (remote) or re-run (local) on replay unless retries are exhausted. |
Subpath exports
| Import path | What it provides |
|---|---|
@adonis-agora/durable | Everything on this page: ctx types, BaseWorkflow, decorators, WorkflowEngine, defineConfig, factories, errors, statuses. |
@adonis-agora/durable/services/main | The WorkflowEngine singleton (default export, undefined on a tenant pod) and runGateway. See Roles & config. |
@adonis-agora/durable/durable_provider | The AdonisJS provider registered by node ace configure. See Getting started. |
@adonis-agora/durable/dashboard_provider | The dashboard provider. See Dashboard. |
@adonis-agora/durable/dashboard | Dashboard building blocks: its own defineConfig / resolveConfig / defaultAuthorize, the framework-agnostic handlers (listRuns, getRun, getRunEvent, cancelRun, retryRun, retryWithInputRun, continueRun, redispatchPendingRun, bulkAction, runValues, workers, topology, health, metricsText, deliverWebhook), dashboardEngineForRole / gatewayDashboardEngine, the auth helpers (resolveDashboardAuth, performLogin, performSession, readSession, SESSION_COOKIE_NAME) and the access-denied / login page renderers. See Dashboard and Dashboard auth. |
@adonis-agora/durable/worker | The store-less worker entry: WorkerRuntime, step registration helpers (registerSteps, registerStepsFromDir, registerStepsFromBarrel), runWorkflowTurn and its errors, the worker registries (RedisWorkerRegistry, NoopWorkerRegistry) and the descriptor / routing-token helpers. See Thin workers. |
@adonis-agora/durable/otel | attachDurableOtel, attachDurableMetrics, otelTraceparent. See OpenTelemetry. |
@adonis-agora/durable/telescope | durableTelescopeExtension, durableDashboard, durableSchedules, the data providers behind its widgets (durableStateProvider, durableWorkerHealthProvider, durableWorkerStatusProvider, durableTimeseriesProvider, …) and dedupeRunEvents. See Telescope. |
@adonis-agora/durable/testing | createTestEngine, MutableClock, assertRunStatus, assertOutput, assertStepsRan, assertStepAttempts, recordedSteps, failOnce, failTimes, captureHistory, parseRunHistory, assertReplayable, and the transport conformance kit assertTransportConformance. See Testing. |
@adonis-agora/durable/testing/conformance | State-store, admission-backend and run-queue conformance kits: runStateStoreContract, runAdmissionBackendContract, runRunDispatchContract, and StateStoreUnavailableError (throw it from a store factory to skip the contract when the backend is down). See Conformance kits. |
@adonis-agora/durable/hooks/workflows / @adonis-agora/durable/hooks/steps | Assembler hooks that generate the app/workflows and app/steps barrels (default exports; workflowsHook(options) / stepsHook(options) take { source?, importAlias?, output? }, output .adonisjs/durable/workflows.ts / steps.ts). See app/workflows. |
@adonis-agora/durable/admission-redis | RedisAdmissionBackend, for passing an instance directly. See Flow control. |
@adonis-agora/durable/control-plane-redis | RedisControlPlane, for passing an instance directly. See Control plane. |
@adonis-agora/durable/commands | The ace commands: durable:work, durable:worker, durable:runs, durable:retry, durable:export and make:workflow. See CLI. |
@adonis-agora/durable/configure | The node ace configure hook (also re-exported from the root). |