Workflows & steps

Workflows

A workflow is a provider decorated with @Workflow; its run(ctx, input) method is the deterministic body the engine executes and replays.

@Workflow({ name: 'checkout', version: '1' })
export class CheckoutWorkflow {
  constructor(private readonly inventory: InventoryService) {}
  async run(ctx: WorkflowCtx, order: Order) {
    /* ... */
  }
}

Register it as a normal provider — DurableModule discovers it. Start runs with WorkflowService.start(CheckoutWorkflow, input) — pass the workflow class for a typed input, or a name string for a cross-runtime workflow.

Class-first starts — MyWorkflow.start / MyWorkflow.execute

Extend the optional DurableWorkflow base class and the workflow starts from the class itself — no engine or WorkflowService injection at the call site:

@Workflow({ name: 'checkout', version: '1' })
export class CheckoutWorkflow extends DurableWorkflow {
  constructor(private readonly inventory: InventoryService) {
    super();
  }
  async run(ctx: WorkflowCtx, order: Order) {
    /* ... */
  }
}

// anywhere — a controller, a consumer, another workflow's body:
const { runId } = await CheckoutWorkflow.start(order); // fire-and-forget, typed input
const result = await CheckoutWorkflow.execute(order); //  run-and-await, typed output

Each method means the same thing everywhere; only the mechanism adapts to where you call it:

Inside-ness is detected through the ambient workflow context the engine installs around every body execution, so calling InnerWorkflow.execute(input) inside another workflow's run is literally ctx.child(InnerWorkflow, input) — the child shows up linked under its parent in the dashboard, and both calls are deterministic on replay. opts.id pins the run id (outside) or child id (inside) for idempotent starts; execute's opts.timeoutMs bounds the outside wait.

The types are inferred from the subclass's own run signature — start checks the input, execute resolves Awaited<ReturnType<run>> — so there is no manual type parameter anywhere. In a store-less tenant worker the class is bound to a start facade that cannot wait: start works, execute outside a workflow throws a clear error.

Which to reach for

The statics and ctx.child/ctx.startChild do the same thing inside a workflow — but they read differently, and that difference is the guidance:

• Outside a workflow — prefer the statics. A controller or consumer that only needs to kick off a run shouldn't inject WorkflowService for it; CheckoutWorkflow.start(order) is the API the class earns, and there's no ambiguity out here — outside, start is engine.start.
• Inside a workflow body — prefer ctx.child / ctx.startChild. The ctx is the visible determinism boundary: everything that touches the outside world goes through it, so a reviewer scanning for ctx. sees every history operation at a glance. Inner.execute() does the identical thing, but the call site no longer looks like a durable operation — and ctx also covers what the statics can't: a child by name (ctx.child('processing', input) — a cross-runtime workflow with no TS class) and fan-out with ctx.all.
• The statics inside a body earn their place in deep helpers — a domain service several calls below run that would otherwise thread ctx through three signatures just to spawn a child. The ambient context resolves it cleanly there.

Steps

A step is the one durable primitive: ctx.step(...) runs a unit of work and checkpoints its result, so on a crash, retry, or replay it is not re-executed — the saved result is returned. There is no local/remote placement choice: every step is dispatched and engine-scheduled — it runs on whatever worker serves its name, the run suspends with zero compute until the result lands, then resumes with it.

You declare the work as a @Step-decorated provider method and call it from the workflow with ctx.step:

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

The handler's single argument is the step input; its return value is the step output. Two call forms:

// by method reference — typed by the handler's own signature, refactor-safe, autocompleted:
const held = await ctx.step(this.inventory.reserve, order);

// by name — for a cross-runtime handler (e.g. a Python @step) with no JS reference to import:
const out = await ctx.step<ProcResult>('processing:proc', input);

A real workflow strings several steps together — the body is just orchestration, and each ctx.step is a durable checkpoint. Walk a checkout run line by line:

1@Workflow({ name: 'checkout', version: '1' })2export class CheckoutWorkflow {3  constructor(4    private readonly inventory: InventoryService,5    private readonly payments: PaymentsService,6    private readonly shipping: ShippingService,7    private readonly email: EmailService,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');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}

Declaring a step with @Step

@Step marks a provider method as a step handler. Three forms:

• @Step() — bare: the routing name is derived from the method as `${ClassName}.${method}` (e.g. InventoryService.reserve) — refactor-safe, no magic string. Carries no retry/timeout and no runtime validation; the method's compile-time types are the only check.
• @Step('custom:name') — an explicit name (stable across refactors, or a cross-runtime contract with a non-JS worker).
• @Step({ name?, input?, output?, retries?, backoff?, backoffMs?, backoffMaxMs?, jitter?, timeoutMs? }) — optional name override, opt-in runtime zod schemas (input validates before the method runs, output before its result is handed back), and a def-level retry/backoff/liveness policy. backoffMaxMs caps an exponential backoff so it doesn't grow without bound — see Retries & backoff.

@Injectable()
export class PaymentsService {
  @Step({ retries: 3, backoff: 'exp' })
  async chargeCard(input: { orderId: string; amountCents: number }) {
    return { chargeId: await this.stripe.charge(input) };
  }
}

The policy on @Step is what ctx.step reads off the method reference; a per-call ctx.step(ref, input, { retries }) overrides it field-by-field.

In-flight visibility

When a step's handler 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-step simply re-runs it.

Toggle this with the engine's trackStepStart option (default true). The step.started event fires either way — the live event stream always sees the start; the flag only gates the extra checkpoint write. Set it to false on hot paths with many short steps to halve their checkpoint writes, at the cost of reload-survivable in-flight visibility.

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.step(this.svc.doA, x),
  ctx.step(this.svc.doB, y),
]);

Long-running loops — ctx.continueAsNew

A workflow that loops forever — polling for new work, or a long-lived accumulator parked on ctx.waitForSignal — keeps writing a checkpoint on every turn it takes. Left unbounded, that history grows forever and every replay gets slower re-reading it. ctx.continueAsNew(input?) is the reset: it ends the current run and hands off to a fresh execution of the same workflow with a clean history.

continueAsNew(input?: unknown): Promise<never>;

It's terminal — the call always throws internally, so code written after it never runs (return its result, same as return/throw). The current run completes normally (no output), and the next execution starts under a new, traceable id: <runId> → <runId>~1 → <runId>~2 … Whatever the next iteration needs to keep going must be carried forward explicitly through input — the fresh run starts with empty history, not a continuation of the old one.

@Workflow({ name: 'poll-inbox', version: '1' })
export class PollInboxWorkflow {
  constructor(private readonly inbox: InboxService) {}

  async run(ctx: WorkflowCtx, input: { mailbox: string; cursor?: string }) {
    const { messages, nextCursor } = await ctx.step(this.inbox.fetchSince, {
      mailbox: input.mailbox,
      cursor: input.cursor,
    });
    for (const message of messages) {
      await ctx.startChild(ProcessMessageWorkflow, message);
    }
    await ctx.sleep('30s');
    // End this run and start fresh with the new cursor — history resets instead of growing forever.
    return ctx.continueAsNew({ mailbox: input.mailbox, cursor: nextCursor });
  }
}

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 '@dudousxd/nestjs-durable-core';

@Step()
async charge(order: Order) {
  const res = await this.stripe.charge(order);
  if (res.declined) throw new FatalError('card declined', 'declined');
  return res;
}

Steps vs. sub-process events

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

• A step (ctx.step) 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. Use a step for any unit that should survive a crash or not be redone.

• 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.

The step logger arrives as the handler's second argument:

// One durable step that internally processes N items and records each outcome for visibility.
@Step()
async processBatch(batch: Item[], log: StepLogger) {
  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));
    }
  }
}

Rule of thumb: if you want each unit to retry or replay independently, make it a step (or a child workflow when it has its own internal steps you want to see). If you only want to see what happened inside one durable unit, emit sub-process events.

Emitting from deep inside a step

The second argument only reaches the handler itself. When the code that actually knows the progress lives a few layers down — a shared batch inserter, an HTTP client, a parser — threading the logger through every signature on the way is the wrong trade. currentStep() reads the logger of the step running on the current async path instead (an AsyncLocalStorage, so concurrent step invocations never see each other's):

import { currentStep, subProcess } from '@dudousxd/nestjs-durable-core';

// A generic utility. It takes no logger, and its callers were not edited.
async function insertRows(rows: Row[]) {
  for (const chunk of chunks(rows, 100)) {
    await db.insert(chunk);
    currentStep()?.subEvent({ id: 'insert', name: 'insert', phase: `${done} rows` });
  }
}

Module-level shortcuts cover the rest of the logger surface: sub(...), subEvent(...) and subProcess(name, body, opts?).

Log lines come in two spellings, and both are supported on purpose:

import { info, log, warn } from '@dudousxd/nestjs-durable-core';

info('parsed header');                 // level known at the call site — the idiomatic TS form,
warn('row 42 skipped', { reason });    // one function per StepLogger method

log(mapLevel(record.severity), record.message); // level computed — the Python-symmetric form

debug / info / warn / error are the per-level shortcuts, mirroring the same four methods on a StepLogger instance so ambient code reads exactly like code that was handed one. log(level, …) takes the level as a value — the literal twin of the Python SDK's log(level, message, data) — which is what you want when the level comes from data rather than from the source line. Neither subsumes the other, so pick by where the level comes from.

Outside a step everything is a no-op — currentStep() is undefined, log/sub/subEvent do nothing, and subProcess still runs its body but emits nothing. That is deliberate: it is what lets a utility be instrumented once and stay usable on a non-durable path and in unit tests, with no if at the call site.

Step interceptors

For cross-cutting concerns around step execution — timing, structured logging, tracing, error enrichment — register an onion-style interceptor instead of repeating the same wrapper in every step handler. At the engine level that's engine.use(interceptor), where interceptor is (invocation, next) => Promise<unknown>: call next() to run the step body (or the next interceptor in the chain) and return — or transform — its result; throw to fail the step. First-registered runs outermost.

The NestJS package wraps this as a decorator so an interceptor is just another provider:

import { type DurableStepInterceptor, StepInterceptor } from '@dudousxd/nestjs-durable';
import type { StepInvocation } from '@dudousxd/nestjs-durable-core';
import { Injectable, Logger } from '@nestjs/common';

@Injectable()
@StepInterceptor()
export class StepLoggingInterceptor implements DurableStepInterceptor {
  private readonly logger = new Logger(StepLoggingInterceptor.name);

  async intercept(invocation: StepInvocation, next: () => Promise<unknown>): Promise<unknown> {
    const startedAt = Date.now();
    try {
      const result = await next();
      this.logger.log(`${invocation.workflow} ${invocation.stepName} ok in ${Date.now() - startedAt}ms`);
      return result;
    } catch (err) {
      this.logger.warn(`${invocation.workflow} ${invocation.stepName} failed: ${String(err)}`);
      throw err;
    }
  }
}

DurableModule discovers any @Injectable() marked with @StepInterceptor() on boot and registers it with the engine — no manual engine.use call needed. invocation carries runId, workflow, stepName, seq, and the 1-based attempt number.

Tags

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

@Workflow({ name: 'pipeline', version: '1', tags: ['etl', 'critical'] })
export class PipelineWorkflow {
  /* ... */
}

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

The dashboard shows a run's tags in the list and detail, and a tag filter box (clicking a tag filters the list). Programmatically, query by tag with RunQuery.tag or the dashboard API's ?tag= param.

Search attributes

Tags are a flat label; search attributes are typed, queryable key/value metadata on a run — an amount, a tier, a customerId — for the range/equality filters a tag alone can't express. Set them from inside the workflow with ctx.upsertSearchAttributes(attrs):

upsertSearchAttributes(attrs: SearchAttributes): Promise<void>;

type SearchAttributes = Record<string, string | number | boolean>;

async run(ctx: WorkflowCtx, order: Order) {
  await ctx.upsertSearchAttributes({ amount: order.total, tier: order.customer.tier });
  /* ... */
}

It's a shallow merge into the run's searchAttributes — keys you don't pass are kept — and it's durable and exactly-once: recorded at this position on the first run and skipped on replay, so calling it inside a loop doesn't multiply writes. Use it instead of injecting the StateStore to mutate the run you're executing (store.updateRun(ctx.runId, …) becomes ctx.upsertSearchAttributes(…)).

Querying by attribute — RunQuery

The dashboard and store.listRuns filter on these through RunQuery:

interface RunQuery {
  workflow?: string;
  status?: RunStatus;
  /** Each plural field matches ANY of its values, ANDed with the other predicates. An empty set
   *  matches nothing; `origins` takes `null` as a member (the runs nothing could attribute). */
  statuses?: RunStatus[];
  workflows?: string[];
  namespaces?: string[];
  origins?: (string | null)[];
  /** Exact match against a run's tags; `tags` matches a run carrying ANY of them. */
  tag?: string;
  tags?: string[];
  /** Typed/range predicates over searchAttributes, ANDed together. `in` carries a `values` SET,
   *  matched as OR — the one thing two `eq` predicates on a key cannot express, since they are
   *  ANDed and no run has one attribute with two values. */
  attributes?: (
    | { key: string; op: 'eq' | 'ne' | 'gt' | 'gte' | 'lt' | 'lte'; value: string | number | boolean }
    | { key: string; op: 'in'; values: (string | number | boolean)[] }
  )[];
  namespace?: string;
  limit?: number;
  offset?: number;
}

attributes is where search attributes pay off — range and inequality predicates a tag can't express, ANDed together (pair it with workflow/status/tag to bound the scan on large stores):

const highValuePending = await store.listRuns({
  workflow: 'checkout',
  status: 'suspended',
  attributes: [
    { key: 'amount', op: 'gte', value: 20000 },
    { key: 'tier', op: 'eq', value: 'pro' },
  ],
});

statuses is the other predicate worth knowing: use it instead of one listRuns call per status when you need several — e.g. counting running + suspended runs in a single scan.