Skip to content
ironflow
Ironflow is pre-1.0: APIs can change between releases. Pin a version. What is ready →

Step Primitives

Steps are the building blocks of workflows. Each step is memoized—if a workflow restarts, completed steps aren’t re-executed. This provides effectively-once execution via idempotent memoization (NATS JetStream is at-least-once; memoization makes step results stable across retries).

step.run(name, fn) — Execute and memoize

Section titled “step.run(name, fn) — Execute and memoize”

Executes a function and caches the result. If the workflow restarts, the cached result is returned instead of re-executing.

const user = await step.run("fetch-user", async () => {
  return await db.users.find(event.data.userId);
});

Key behaviors:

  • Results are persisted to the database
  • On workflow restart, cached results are returned
  • Keep step names unique within a function (see Step Naming Best Practices)
  • Return value is JSON-serializable
  • A step is durable once the engine has recorded it. A crash before that runs the step again: for a pull worker, within one checkpoint interval (default 1 second) of the step finishing; for push mode, before the handler returns or yields. Make step bodies idempotent
  • Takes an optional third argument to cap this step’s wall-clock time — step.run(name, fn, { timeout: "30s" }) in TypeScript, ironflow.Run(ctx, name, fn, ironflow.WithTimeout(30*time.Second)) in Go, ctx.step.run(name, fn, timeout="30s") in Python. It overrides the function’s stepTimeout / StepTimeout; exceeding it raises StepTimeoutError, which is retryable

step.sleep(name, duration) — Pause execution

Section titled “step.sleep(name, duration) — Pause execution”

Waits durably for a specified duration. The run reports waiting and resumes after the remaining time across server restarts. An operator hold reports paused; explicit Resume preserves the original wake time.

await step.sleep("wait-24h", "24h"); // "1h", "30m", "7d"

Duration formats:

  • "500ms" — 500 milliseconds
  • "30s" — 30 seconds
  • "5m" — 5 minutes
  • "2h" — 2 hours
  • "7d" — 7 days
  • 5000 — a bare number is milliseconds

step.sleepUntil(name, until) — Pause until a specific time

Section titled “step.sleepUntil(name, until) — Pause until a specific time”

Waits until a specific date/time. Like step.sleep, the run reports waiting and the original deadline survives server restarts and operator hold/resume.

// Sleep until a specific ISO 8601 timestamp (must be in the future)
const target = new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString(); // 24h from now
await step.sleepUntil("wait-24h", target);


// Or pass a Date object
const tomorrow = new Date(Date.now() + 24 * 60 * 60 * 1000);
await step.sleepUntil("wait-until-tomorrow", tomorrow);

Key behaviors:

  • Accepts a Date or an ISO 8601 string (TypeScript), time.Time (Go), or a timezone-aware datetime or ISO string (Python)
  • The TS and Python SDKs throw if the target is in the past; the Go SDK currently lacks this check — caller is responsible
  • Results are memoized like all other steps

step.waitForEvent(name, filter) — Wait for correlated event

Section titled “step.waitForEvent(name, filter) — Wait for correlated event”

match is a dot-separated JSON path, not a comparison expression. Paths accept letters, digits, underscores, and hyphens in each segment, with optional data. or $. prefixes. Without matchValue, the engine snapshots the value at that path in the triggering event. A missing or null trigger value fails the yield; a missing or null incoming value never matches.

Use a non-empty matchValue to correlate against a literal that is absent from the triggering event, such as the current run ID:

const approval = await step.waitForEvent("approval", {
  event: "approval.received",
  match: "data.runId",
  matchValue: run.id, // `run` comes from the handler context, beside `event` and `step`
  timeout: "1h",
});
approval = await ctx.step.wait_for_event(
    "approval", event="approval.received", match="data.runId",
    match_value=ctx.run.id, timeout="1h",
)

matchValue requires match. An omitted or empty value uses the triggering event. The Go SDK exposes the same option as EventFilter.MatchValue, and Python as match_value=; the HTTP yield field is match_value.

Pauses the workflow until a matching event arrives or the timeout expires.

A matching event can arrive before the workflow reaches this step. When the step parks, Ironflow checks retained events in the same environment, stored since the run was created and no later than the wait’s deadline. This check uses server storage time, not the event’s publisher-supplied timestamp, and examines the oldest eligible events first. The scheduler retries the check after a restart and before timing out the step.

The event that a wait returns carries the authenticated emitter that the engine recorded: emittedBy in TypeScript, EmittedBy in Go and emitted_by in Python. Examples are apikey:<id>, user:<id>, webhook:<source-id>, nats and platform. The engine sets it from the credential of the caller, and the event body cannot change it. The value is empty when the engine reports none. approve() returns the same value beside approver. approver is free text from the event data, and the caller chooses it. An operator with runs:cancel can inject or patch a step output, so emittedBy proves the emitter only for a step that an event resolved. This follows ADR 0129.

An event already consumed by an earlier completed wait in the same run cannot satisfy a later gate. Neither can an event with the same name that was stored before an earlier completed wait for that name ended, so a duplicate reply sent before the first gate closed is not reused. The event that started the run cannot satisfy its own wait. Events stored before the run was created, or removed by retention, cannot satisfy this history check. approve() uses the same behavior, with the run ID as its correlation value, so an early approval does not need to be resent.

const approval = await step.waitForEvent("wait-approval", {
  event: "order.approved",
  match: "data.orderId", // matches event.data.orderId
  timeout: "7d",
});


// The handler only resumes when a matching event arrives.
// If the timeout expires first, the run is failed server-side with
// "waitForEvent timed out" — this line is never reached on timeout.
console.log("Approved by:", approval.data.approvedBy);

Branch on a wait timeout

Section titled “Branch on a wait timeout”

By default, a wait deadline fails the run. To branch on absence, opt in to a completed, memoized null result. The engine still checks accepted matching events through the deadline before settling the timeout. The policy and result survive restarts; replay returns the same absence without waiting again. Live delivery and replay use server storage time: a reply accepted after the deadline cannot satisfy the wait, even if its publisher timestamp is earlier. A reply accepted on time remains eligible when delivery runs late.

If the platform cannot read event history, it defers timeout until two minutes after the persisted deadline. This window starts at the deadline, so a first failed read after a long restart can fail immediately. If replay still fails, the run fails with waitForEvent timeout outcome unknown: event history replay failed after grace period; it does not return null and guess that no reply arrived.

const reply = await step.waitForEvent("confirm-seat", {
  event: "seat.confirmed", timeout: "5m", onTimeout: "return-null",
});
if (reply === null) {
  await step.run("release-seat", async () => releaseSeat());
}

The default return type is IronflowEvent<T>; onTimeout: "return-null" makes it IronflowEvent<T> | null.

approve() keeps its fail-on-timeout default.

JavaScript yielding steps (sleep, sleepUntil, waitForEvent, invoke, and invokeAsync) suspend execution with an internal signal. Let that signal propagate. If a handler or parallel branch catches it and returns, or tries another durable operation, the SDK fails the run nonretryably with a diagnostic. The failure remains fatal even if an outer catch hides that diagnostic. Put ordinary error handling inside the step.run callback. This guard cannot prevent arbitrary network calls or other side effects performed directly inside a catch block before the SDK regains control. Python’s suspension signal derives from BaseException; do not catch it with except BaseException. Go uses a panic; do not recover it in the handler.

Options:

OptionTypeRequiredDescription
eventstringyesEvent name to wait for
onTimeout (TS) / on_timeout (Python)"fail" | "return-null"noFail the run by default; opt in to a completed null result. In Go use WaitForEventOptional.
matchstringnoField path to correlate events (e.g., "data.orderId")
matchValue (TS) / MatchValue (Go) / match_value (Python)stringnoNon-empty literal compared at match. Omitted or empty snapshots the triggering event. Requires match.
payload (TS, Python) / Payload (Go)unknown / anynoRequest data stored as the waiting step’s input — visible via getRunSteps and in the dashboard, so an approver can see what they are approving
timeoutTS string | number (ms) / Go time.Duration / Python duration ("7d", timedelta, or seconds)noMaximum time to wait (default "7d")

step.parallel(name, branches) — Execute branches concurrently

Section titled “step.parallel(name, branches) — Execute branches concurrently”

Run multiple operations in parallel with isolated step contexts:

const [userResult, orderResult, inventoryResult] = await step.parallel(
  "fetch-all-data",
  [
    async (s) => s.run("fetch-user", async () => fetchUser(userId)),
    async (s) => s.run("fetch-order", async () => fetchOrder(orderId)),
    async (s) => s.run("check-inventory", async () => checkInventory(productId)),
  ],
  {
    concurrency: 2,        // Max 2 branches running at once
    onError: "failFast",   // Stop on first error (default) or "allSettled"
  }
);

Key behaviors:

  • Each branch receives a scoped step client (TypeScript) or *BranchContext (Go). It is load-bearing: a branch is not itself a recorded step. Work done directly in the callback body persists nothing and re-runs in full on every retry; work routed through the enclosing function’s step client is memoized outside the branch’s scope. Both compile, so the SDK warns on either shape: a step claimed on the enclosing client while the fan-out is running (its index comes from a counter shared by every branch, so a resume can cross the wires), or every branch skipping the scope. See the opt-out in the options table below. TypeScript warns on the second shape only. The Python SDK raises neither warning.
  • Results are returned in order regardless of completion order
  • Supports concurrency limits to control parallelism
  • Two error modes: failFast (stop immediately) or allSettled (complete all)

Options:

OptionTypeDefaultDescription
concurrencynumberunlimitedMaximum concurrent branches
onError"failFast" | "allSettled""failFast"Error handling strategy
expectScopedClient (TypeScript)booleantrueSet false to silence the unscoped-branch warning
SkipScopedClientCheck (Go)boolfalseSet true to silence the unscoped-branch warning

The two SDKs spell the opt-out with opposite names and opposite polarity, and both default to warning. Go struct fields zero-value to false, so an ExpectScopedClient bool would have defaulted every existing caller into the opt-out.

step.map(name, items, fn) — Fan-out/fan-in pattern

Section titled “step.map(name, items, fn) — Fan-out/fan-in pattern”

Process an array of items in parallel:

const userIds = ["user_1", "user_2", "user_3"];
const users = await step.map(
  "fetch-all-users",
  userIds,
  async (userId, s, index) => {
    return await s.run(`fetch-user-${index}`, async () => {
      return await fetchUserDetails(userId);
    });
  },
  {
    concurrency: 5,        // Max 5 concurrent operations
    onError: "allSettled", // Complete all items even if some fail
  }
);

Use cases:

  • Processing multiple orders simultaneously
  • Fetching data from multiple sources
  • Sending notifications to multiple users
  • Batch processing with controlled parallelism

Options:

OptionTypeDefaultDescription
concurrencynumberunlimitedMaximum concurrent operations
onError"failFast" | "allSettled""failFast"Error handling strategy
expectScopedClient (TypeScript)booleantrueSet false to silence the unscoped-branch warning
SkipScopedClientCheck (Go)boolfalseSet true to silence the unscoped-branch warning

The two SDKs spell the opt-out with opposite names and opposite polarity, and both default to warning. Go struct fields zero-value to false, so an ExpectScopedClient bool would have defaulted every existing caller into the opt-out.

Step Naming Best Practices

Section titled “Step Naming Best Practices”

Step names must be unique within a function and should be descriptive:

// Good: descriptive, unique names
await step.run("validate-order-items", ...);
await step.run("calculate-shipping-cost", ...);
await step.run("process-payment-stripe", ...);


// Bad: generic or duplicate names
await step.run("step1", ...);
await step.run("process", ...);
await step.run("process", ...);  // Allowed: the SDK adds a counter to the step ID, but the step list is hard to read
# Good: descriptive, unique names
await ctx.step.run("validate-order-items", validate_items)
await ctx.step.run("calculate-shipping-cost", calculate_shipping)
await ctx.step.run("process-payment-stripe", charge)
// Good: descriptive, unique names
ironflow.Run(ctx, "validate-order-items", validateItems)
ironflow.Run(ctx, "calculate-shipping-cost", calculateShipping)
ironflow.Run(ctx, "process-payment-stripe", charge)

No SDK rejects a repeated name: each one adds a per-name counter to the step ID. Keep names unique anyway, so the step list stays readable.

Guidelines:

  • Use kebab-case for consistency
  • Include the action and target (e.g., fetch-user, send-email)
  • Make names unique even in loops (e.g., process-item-${index})

Deploying changes to workflows with existing runs

Section titled “Deploying changes to workflows with existing runs”

A resumed handler encounters persisted outputs from completed steps. New code must still understand those outputs and preserve the meaning of unfinished work.

VersionWhat it identifies today
Configuration/history versionsRegistered function settings and snapshots in ironflow:fn:{id}. The configuration counter and history entity version are distinct fields.
Executable deployment identityThe actual handler build or artifact. Ironflow does not currently pin runs to an immutable executable identity.
Event schema versionThe shape of an event payload. Event upcasters adapt events, not completed step results.
Run’s recorded versionruns.function_version snapshots the function’s EntityVersion at run creation. It is not a handler code digest.

Registration compares configuration and metadata, and skips unchanged registrations. The Go, Node and Python SDKs stamp a code hash into the function metadata (__ironflow_code_hash), so editing a handler normally registers a new version. Go hashes the running binary. Node and Python hash the handler’s own source, so a change in a helper, a dependency or a captured value can leave the hash, and the version, unchanged. A registration without that metadata, such as a plain RegisterFunction call, retains the version when only the code changes. Pull worker capabilities read the live function record’s entity version; matching that version to a queued run does not prove that the worker has the original executable. Push dispatch also resolves the live function configuration, including its endpoint. These are current source-backed limitations, not a guarantee of executable pinning.

Safe and unsafe changes

Section titled “Safe and unsafe changes”

A compatible change preserves step identity and ordering, accepts existing output schemas, and preserves the semantics of branches, waits, and side effects for existing runs. Validate that compatibility using representative persisted outputs and runs waiting at durable steps before deploying.

Treat these changes as incompatible until you have checked existing runs:

  • Renaming, reordering, inserting, removing, or changing the type of a step can change which persisted result is reused or cause work to execute again.
  • Changing an output schema can leave new code reading an old cached shape. Editing the completed step’s callback does not recompute its stored output.
  • Changing branching or wait names, filters, or timeouts can alter the path taken after resume or conflict with a wait already registered by the old code.
  • Changing side-effect semantics can mix an old completed action with new unfinished actions. Renaming a payment step to force execution can charge again.

For incompatible changes, stop new admissions to the old function and drain outstanding runs, including sleeping, waiting, retryable, and queued runs, before replacing its handler. If both implementations must coexist, use a separate function ID for new work and retain the old handler and endpoint or workers for the old ID. Route new triggers deliberately so both functions do not consume the same work accidentally. Keep side effects idempotent across retries.

Function registration history can restore configuration. It does not preserve or restore handler code, migrate completed step results, or make an incompatible deployment safe. Keep deployable artifacts under your own deployment management.

Immutable executable IDs and routing both push and pull runs to their originating deployment are accepted future work, not implemented behavior. The agreed direction is to keep runs pending with a visible reason when their executable is unavailable, with operator alerts and explicit cancellation instead of silently using newer code. Operators would initially retain old workers or version-specific endpoints. Artifact hosting and lifecycle management remain outside that first implementation. Track it in #2166; nothing on this page changes until it ships.

Other step primitives

Section titled “Other step primitives”

Covered on neighbouring pages:

  • step.compensate — register an undo for a step; runs in reverse order on failure. See Sagas.
  • step.invoke / step.invokeAsync — call another function from inside a step (sync awaits the result; async fire-and-forget). The target function must have no event triggers. In Go: ironflow.Invoke[T](ctx, id, input) and ironflow.InvokeAsync(ctx, id, input). In Python: await ctx.step.invoke(id, input) and await ctx.step.invoke_async(id, input). The default wait is 30 seconds.
  • step.publish — publish to a pub/sub topic from inside a step. The step is memoized, so a retry after the engine recorded it does not publish again; for a retry before that point, the step sends a default idempotency key derived from its step ID, so the server drops the repeat. The server keeps the key for IRONFLOW_RUN_RETENTION_DAYS days (forever when unset or 0) and returns the original event ID (ADR 0132). Pass your own key to replace the default. It does not trigger functions — use the events API for that. In Go: ironflow.Publish(ctx, topic, data). In Python: await ctx.step.publish(topic, data). Under serve(), set server_url= or IRONFLOW_SERVER_URL; without it the run fails with “server URL not configured for publish step”.

What’s Next?

Section titled “What’s Next?”