Conditions & Error Handling

Every .step() runs by default only when all of its predecessors completed successfully. Pass condition to change that:

const handle = queue.workflows
  .define("resilient-pipeline")
  .step("riskyStep", "riskyTask")
  .step("celebrate", "celebrateTask", {
    after: "riskyStep",
    condition: "on_success", // default — only runs if riskyStep completed
  })
  .step("recover", "recoverTask", {
    after: "riskyStep",
    condition: "on_failure", // only runs if riskyStep dead-lettered
  })
  .submit();

queue.runWorker();

const run = await handle.wait();
console.log(run.state); // "failed" — even if recover ran (see below)

Condition values

A step whose condition is not met transitions to skipped. Skipped steps propagate: all of their descendants are skipped too, unless those descendants have a separate predecessor that did complete.

How it works

The WorkflowTracker evaluates conditions when a predecessor node settles. It reads the run plan from storage, checks each dependent node's condition against the settled outcome, and either creates the deferred job or marks the node skipped. Because the tracker reconstructs run state from storage on every event, submit and execute may be different processes.

When riskyStep fails: celebrate → skipped, recover → enqueued and runs. When riskyStep succeeds: recover → skipped, celebrate → enqueued and runs.

Conditions are string predicates only

condition accepts only the three string literals above — you cannot pass a JavaScript function. Conditions are evaluated by the WorkflowTracker at runtime, which runs inside the worker process and may differ from the process that called .submit(). Because the workflow definition is serialised to and reconstructed from storage, a JS closure cannot cross that boundary. If you need dynamic branching, encode the decision in the task's return value and gate downstream work on that (for example, use a fan-out with a zero-item array to short-circuit a branch).

A failed run stays failed

When riskyStep fails and recover runs, the workflow run still ends in state "failed". The on_failure handler executes but does not "recover" the run — it is an error-handling side effect, not a circuit-breaker. To treat a failure as a non-fatal branch, model it differently (for example, wrap the risky logic in a task that catches internally and returns a status sentinel).

always steps

Use "always" for teardown or notification steps that should run regardless of the upstream outcome:

queue.workflows
  .define("with-cleanup")
  .step("provision", "provisionTask")
  .step("work", "workTask", { after: "provision" })
  .step("cleanup", "cleanupTask", {
    after: "work",
    condition: "always",
  })
  .submit();

cleanup runs whether work succeeded or failed.

Step options

Coming from BullMQ? FlowProducer runs every child unconditionally — there's no per-child on_success / on_failure branching. Conditional steps are flexiq-only; in BullMQ you'd inspect the parent job's outcome yourself and decide whether to add() the next job at all.