Use workflow steps

Break workflows into persisted, resumable checkpoints.

By default, workflows time out after 5 minutes. Steps let you break a workflow into a series of checkpoints that are persisted individually. If a workflow is interrupted, it resumes from the last completed step.

Basic steps

The step function takes a unique name and a handler. The return value is persisted and reused if the step has already completed:

export default new Workflow({
  name: "dataPipeline",
  handler: async ({ step }) => {
    const data = await step("fetch-data", async () => {
      return await fetchDataFromAPI()
    })

    const processed = await step("process-data", async () => {
      return processData(data)
    })

    await step("store-results", async () => {
      await saveResults(processed)
    })
  },
})

Step names should be unique. If the same name runs twice, the second call returns the first one’s cached result instead of running.

Steps can be nested. A parent step completes when all its sub-steps complete.

Retries

Steps retry automatically on failure. Control the number of attempts with maxAttempts (defaults to 5):

const data = await step("fetch-data", async ({ attempt }) => {
  console.log(`Attempt #${attempt}`)
  return await fetchDataFromAPI()
}, { maxAttempts: 10 })

If a step exhausts all retries, it throws an error. This fails the entire workflow unless you catch it:

try {
  await step("risky-operation", async () => {
    return await unstableAPICall()
  }, { maxAttempts: 3 })
} catch (err) {
  console.log("Operation failed after 3 attempts:", err)
}

Sleep

Pause the workflow for a duration (in milliseconds) or until a specific time:

await step.sleep("wait-5-min", 5 * 60 * 1000)

await step.sleepUntil("wait-until-noon", new Date("2025-01-15T12:00:00Z"))

The workflow is suspended during sleep, and no further compute is used.

Progress

Use .progress() to record a checkpoint without running any work. This is useful for observability when you want to mark milestones in a long handler:

await step.progress("Started import")
// ...
await step.progress("Halfway done")
// ...
await step.progress("Finished")

Listen

Use .listen() to suspend the workflow and set its status to "listening". The workflow then waits to be resumed by an external event. You can specify this event by passing its name in as an argument:

await step.listen("wait-for-payment")

Fail and abort

Use .fail() or .abort() to stop execution explicitly:

// Mark as failed with a message
await step.fail("User verification required")

// Abort immediately without marking as failed
step.abort()

Child workflows

You can start another workflow and wait for it to complete:

import ProcessingWorkflow from "../workflows/processing"

const result = await step.executeWorkflow(
  "process-data",
  ProcessingWorkflow,
  { data: inputData }
)

Or wait for an already-running workflow:

const childWorkflow = await ChildWorkflow.start({})
const result = await step.waitForWorkflow("wait-for-child", childWorkflow.id)

Parallel processing

map

Process an array of items in parallel and collect results:

const results = await step.map(
  "process-users",
  users,
  async (user, { i }) => await processUser(user),
  { concurrency: 5, maxAttempts: 3 }
)

forEach

Process items in parallel without collecting results:

await step.forEach(
  "notify-users",
  users,
  async (user) => await sendNotification(user),
  { concurrency: 10 }
)

batch

Process items in sequential batches:

await step.batch(
  "bulk-insert",
  records,
  async (batch) => await database.bulkInsert(batch),
  { batchSize: 100 }
)