Python/Modules/Workflows/

Sagas

Roll back completed workflow steps in reverse order when a run fails — declare a compensator per step and flexiq handles the rest.

A saga is a workflow where steps that succeed register a compensator — a task that undoes that step's side effect. If a later step fails, flexiq automatically rolls back every already-completed step by running its compensator, in reverse-dependency order, so the run ends in a consistent state instead of a half-applied one. There's no separate "saga" API — any DAG workflow becomes a saga the moment one of its steps declares a compensator.

Basic example

python

from flexiq import Queue
from flexiq.workflows import Workflow

queue = Queue()

@queue.task()
def unreserve_inventory(forward_args, forward_kwargs, forward_result):
    inventory_api.release(forward_result["order_id"])

@queue.task(compensates=unreserve_inventory)
def reserve_inventory(order_id: str) -> dict:
    inventory_api.reserve(order_id)
    return {"reserved": True, "order_id": order_id}

@queue.task()
def refund_payment(forward_args, forward_kwargs, forward_result):
    payment_api.refund(forward_result["charge_id"])

@queue.task(compensates=refund_payment)
def charge_payment(order_id: str) -> dict:
    return {"order_id": order_id, "charge_id": payment_api.charge(order_id)}

@queue.task()
def ship_order(order_id: str):
    raise RuntimeError("Warehouse offline")
# No compensator for ship — if it never succeeded, there's nothing to undo.

wf = Workflow(name="checkout")
wf.step("reserve", reserve_inventory, args=("4212",))
wf.step("charge", charge_payment, args=("4212",), after="reserve")
wf.step("ship", ship_order, args=("4212",), after="charge")

# A worker is already running elsewhere (queue.run_worker()).
run = queue.submit_workflow(wf)
status = run.wait()
# ship fails → charge rolled back (refund_payment) → reserve rolled back (unreserve_inventory)
# status.state is WorkflowState.COMPENSATED

queue.task("reserveInventory", (orderId: string) => ({ reserved: true, orderId }));
queue.task("unreserveInventory", (result: { reserved: boolean; orderId: string }) => { /* undo */ });

queue.task("chargePayment", (orderId: string) => ({ charged: true, orderId }));
queue.task("refundPayment", (result: { charged: boolean; orderId: string }) => { /* undo */ });

queue.task("shipOrder", (orderId: string) => { throw new Error("Warehouse offline"); });
// No compensate for ship — if it never succeeded, there's nothing to undo.

const handle = queue.workflows
  .define("checkout")
  .step("reserve", "reserveInventory", { compensate: "unreserveInventory" })
  .step("charge", "chargePayment", { after: "reserve", compensate: "refundPayment" })
  .step("ship", "shipOrder", { after: "charge" })
  .submit();

queue.runWorker();

const run = await handle.wait();
// ship fails → charge rolled back (refundPayment) → reserve rolled back (unreserveInventory)
// run.state === "compensated"

java

Task<Integer> reserve = Task.of("reserveInventory", Integer.class);
Task<Integer> unreserve = Task.of("unreserveInventory", Integer.class);
Task<Integer> charge = Task.of("chargePayment", Integer.class);
Task<Integer> refund = Task.of("refundPayment", Integer.class);
Task<Integer> ship = Task.of("shipOrder", Integer.class);
// No compensator for ship — if it never succeeded, there's nothing to undo.

int orderId = 4212; // the workflow input every step receives

Workflow checkout = Workflow.named("checkout")
        .step(Step.of("reserve", reserve, orderId).compensate(unreserve).build())
        .step(Step.of("charge", charge, orderId).after("reserve").compensate(refund).build())
        .step(Step.of("ship", ship, orderId).after("charge").maxRetries(0).build());

WorkflowRun run = queue.submitWorkflow(checkout);

try (Worker worker = queue.worker()
        .handle(reserve, id -> id)
        .handle(charge, id -> id)
        .handle(ship, id -> { throw new IllegalStateException("Warehouse offline"); })
        .handle(refund, chargeResult -> chargeResult)      // undo charge
        .handle(unreserve, reserveResult -> reserveResult) // undo reserve
        .trackWorkflows()
        .start()) {
    WorkflowStatus status = run.await(Duration.ofSeconds(30));
    // ship fails → charge rolled back (refund) → reserve rolled back (unreserve)
    System.out.println(status.state); // COMPENSATED
    System.out.println(status.node("charge").orElseThrow().status); // COMPENSATED
}

Workflows are an opt-in dependency rather than part of the shell — flexiq = { version = "2.0.0", features = ["workflows"] } re-exports the flexiq-workflows crate as flexiq::workflows. A step records its compensator in StepMetadata::compensate:

rust

use std::collections::{HashMap, HashSet};

use flexiq::flexiq_core::wire::{encode_call, WireValue};
use flexiq::workflows::dagron_core::DAG;
use flexiq::workflows::lifecycle::{submit_workflow, SubmitStaticWorkflowRequest};
use flexiq::workflows::{StepMetadata, WorkflowSqliteStorage, WorkflowStorageBackend};
use flexiq::{FlexiQ, SqliteStorage, StorageBackend};

// One SQLite handle, cloned — clones share the pool, so the job tables and the
// workflow tables are the same database.
let base = SqliteStorage::new("tasks.db")?;
let queue = FlexiQ::from_storage(StorageBackend::Sqlite(base.clone()))?;
let wf = WorkflowStorageBackend::Sqlite(WorkflowSqliteStorage::new(base, None)?);

let mut dag: DAG<()> = DAG::new();
for step in ["reserve", "charge", "ship"] {
    dag.add_node(step.to_string(), ())?;
}
dag.add_edge("reserve", "charge", Some(1.0), None)?;
dag.add_edge("charge", "ship", Some(1.0), None)?;

let step = |task: &str, compensate: Option<&str>| StepMetadata {
    task_name: task.to_string(),
    compensate: compensate.map(str::to_string),
    // ship has no compensator — if it never succeeded, there's nothing to undo.
    ..Default::default()
};

// Positional arguments only; the envelope writer is the cross-SDK one.
let order = encode_call(&[WireValue::Text("4212".into())], &[]);

let run = submit_workflow(
    queue.storage(),
    &wf,
    SubmitStaticWorkflowRequest {
        name: "checkout".into(),
        version: 1,
        dag_bytes: dag.to_json(|_| None)?.into_bytes(),
        step_metadata: HashMap::from([
            ("reserve".into(), step("reserve_inventory", Some("unreserve_inventory"))),
            ("charge".into(), step("charge_payment", Some("refund_payment"))),
            ("ship".into(), step("ship_order", None)),
        ]),
        node_payloads: HashMap::from([
            ("reserve".into(), order.clone()),
            ("charge".into(), order.clone()),
            ("ship".into(), order),
        ]),
        queue_default: "default".into(),
        params_json: None,
        deferred_node_names: HashSet::new(),
        cache_hit_nodes: HashMap::new(),
        parent_run_id: None,
        parent_node_name: None,
        default_timeout_ms: 300_000,
        default_priority: 0,
        default_max_retries: 3,
        result_ttl_ms: None,
        namespace: None,
    },
)?;
println!("{}", run.run_id);

submit_workflow writes the definition, the run, a WorkflowNode per step, and one job per step wired through depends_on — so the DAG runs in order on an ordinary worker. Nothing in Rust rolls it back. flexiq-workflows is the storage half of the engine: it has the compensate field, the Compensating / Compensated / CompensationFailed states, and the three setters that write them, but no tracker that notices a step failing and decides what to undo. Submission reads only task_name, queue, max_retries, timeout_ms and priority off a StepMetadata; compensate is recorded and otherwise ignored. Driving rollback is yours: watch node status through WorkflowStorage::get_workflow_nodes and enqueue the compensators yourself.

How it works

A step's forward execution fails with its retries exhausted.
flexiq collects every step that completed and declared a compensator — steps that are pending or skipped never ran, and a step that itself failed has nothing to undo, so all three are left alone.
Those steps are grouped into reverse-topological waves — the most downstream completed steps first, so a step is never rolled back before whatever depends on it. Compensators within one wave run in parallel; the next wave only starts once the current one fully drains.
Each compensator is enqueued idempotently (see Idempotency) with the forward step's result available to it.
Once a wave finishes: if every compensator in it succeeded, the next wave starts. If any failed, rollback stops there — remaining steps are left un-rolled-back, since further undo may depend on the failed step's own undo — and the run ends CompensationFailed instead of Compensated.

Watch a saga roll back

Fail a step and the completed ones undo in reverse-topological order. Steps that never ran, and the failed step itself, are left alone.

Loading demo…

A cache-hit step is never compensated: its forward task didn't actually run in this run, so there's no local side effect to undo. Only a step whose status is NodeStatus.COMPLETED is eligible — a CACHE_HIT is excluded even if it has a compensate.

Declaring compensators

Each step names a rollback task; when the run fails, flexiq calls it with the forward step's outcome:

python

# The compensator receives exactly three positional args: the forward
# task's args, kwargs, and return value.
@queue.task()
def refund_payment(forward_args: tuple, forward_kwargs: dict, forward_result: dict) -> None:
    payment_api.refund(forward_result["charge_id"])

@queue.task(compensates=refund_payment)
def charge_payment(order_id: str) -> dict:
    return {"order_id": order_id, "charge_id": payment_api.charge(order_id)}

// The compensate task receives the forward step's return value as its
// single positional argument.
queue.task("chargePayment", async (orderId: string) => {
  const charge = await stripe.charge(orderId);
  return { chargeId: charge.id, orderId }; // compensator needs chargeId
});

queue.task("refundPayment", async (result: { chargeId: string; orderId: string }) => {
  await stripe.refund(result.chargeId);
});

java

// The compensation task receives the forward step's return value as its payload.
record Charge(String chargeId, String orderId) {}

Task<Charge> refund = Task.of("refundPayment", Charge.class);

worker.handle(refund, result -> {
    stripe.refund(result.chargeId());
    return null;
});

compensate is a task name, stored as a plain Option<String> on the step's metadata:

rust

StepMetadata {
    task_name: "charge_payment".into(),
    compensate: Some("refund_payment".into()),
    ..Default::default()
}

There is no argument contract to write against, because no Rust code enqueues that task: what a compensator is called with is decided by whatever enqueues it, which here is your own code. A compensator is an ordinary #[flexiq::task] function — give it the parameters you intend to pass.

Decorator-level default

@queue.task(compensates=…) sets a task's default compensator wherever it's used; Workflow.step(…, compensates=…) overrides it (or disables compensation) for one specific step:

python

wf.step("charge", charge_payment, compensates=refund_v2)  # override decorator default
wf.step("charge", charge_payment)                          # inherit decorator default
wf.step("charge", charge_payment, compensates=None)         # disable compensation

forward_args, forward_kwargs, and forward_result are fully populated for both static and deferred workflow nodes. Call current_compensation_context() inside a compensator body to also access the workflow run ID, node name, and forward job ID.

Coming from BullMQ? There's no built-in saga/compensation concept in BullMQ — you'd hand-roll rollback logic in your own failure handler. flexiq's compensate option and automatic reverse-order rollback are flexiq-only.

Before reaching for StepMetadata::compensate, know that the field is inert: nothing collects completed nodes, nothing builds waves, and nothing enqueues the task it names. A saga that actually rolls back is written one of two ways.

One task, durable steps. The forward work and its undo live in the same body, so the undo can see what the forward step returned. This is the shape to reach for first:

rust

#[flexiq::task]
fn checkout(order_id: String) -> flexiq::Outcome<()> {
    let mut step = flexiq::current_step();

    let hold = step.run_keyed("reserve", &order_id, || reserve(&order_id))?;
    let charge = step.run_keyed("charge", &order_id, || charge(&order_id))?;

    // A step that may need undoing commits its failure as a value.
    let shipped: Result<(), String> = step.run_keyed("ship", &order_id, || {
        Ok(ship(&order_id).map_err(|e| e.to_string()))
    })?;

    if let Err(reason) = shipped {
        // Reverse order, innermost first — the ordering waves would have given you.
        step.run_keyed("refund", &order_id, || refund(&charge))?;
        step.run_keyed("unreserve", &order_id, || unreserve(&hold))?;
        return Err(flexiq::TaskError::fatal(reason).into());
    }
    Ok(())
}

Memoization does the idempotency: a committed step is not re-run on a later attempt, so a rollback interrupted halfway resumes rather than double-refunding.

Keyed steps, not `run`

Every step on a compensating path must use run_keyed. An unkeyed step is matched by position in the recorded sequence, and the undo branch asks for steps the earlier attempt never recorded there — the run fails with a step divergence before any compensator body executes. A keyed step is matched by key wherever it sits, and one that was never recorded is simply new ground.

The failing step must not abort

Note that ship returns its failure rather than raising it. A step body that returns Err commits nothing and leaves its own position outstanding, and the session refuses to start another step while one is — so the very next run_keyed fails with started while step 'ship' is still uncommitted, surfacing as a retryable error and a retry loop that never reaches the rollback.

So: forward steps you will never compensate can use ? freely. The one whose failure triggers the rollback has to hand the failure back as data, the way Result<(), String> does above.

Separate jobs, driven by you. If the rollback has to be its own job, the node row has columns for it — set_workflow_node_compensation_job moves a node to Compensating and records the job id, then set_workflow_node_compensated or set_workflow_node_compensation_failed closes it out. The reverse-topological ordering is yours to compute; topological_order gives you the forward order to walk backwards.

Steps without compensators

A step that doesn't declare a compensator is skipped during rollback, and so is a step that never ran because it was downstream of the step that failed — only a completed step with a compensator is rolled back:

python

wf = Workflow(name="partial-saga")
wf.step("a", task_a, args=(1,), compensates=undo_a)  # rolled back if the run fails after a completes
wf.step("b", task_b, args=(2,), after="a")            # no compensator — left as-is
wf.step("c", task_c, args=(3,), after="b", compensates=undo_c)

# If b fails: only a is rolled back (b never completed; c never ran).
# If c fails: c has no completed result, so only a is rolled back.

queue.workflows
  .define("partial-saga")
  .step("a", "taskA", { compensate: "undoA" }) // rolled back if run fails after a completes
  .step("b", "taskB", { after: "a" })           // no compensate — left as-is
  .step("c", "taskC", { after: "b", compensate: "undoC" })
  .submit();

// If b fails: only a is rolled back (b never completed; c never ran).
// If c fails: c has no result yet, so only a is rolled back.

java

Workflow partial = Workflow.named("partial-saga")
        .step(Step.of("a", taskA, 1).compensate(undoA).build()) // rolled back if run fails after a completes
        .step("b", taskB, 2, "a")                               // no compensator — left as-is
        .step(Step.of("c", taskC, 3).after("b").compensate(undoC).build());

// If b fails: only a is rolled back (b never completed; c never ran).
// If c fails: c has no completed result, so only a is rolled back.

compensate defaults to None, so a step opts out by saying nothing:

rust

let steps = HashMap::from([
    ("a".to_string(), StepMetadata {
        task_name: "task_a".into(),
        compensate: Some("undo_a".into()),
        ..Default::default()
    }),
    ("b".to_string(), StepMetadata {
        task_name: "task_b".into(),
        ..Default::default() // no compensator
    }),
]);

Which steps are eligible is still a decision your own rollback code makes. WorkflowNodeStatus::is_compensable() is the rule to key off: true for Completed and CacheHit, false for everything else — so a step that never ran, and a step that failed, are both excluded.

Final run states

State	Meaning
`WorkflowState.COMPLETED"completed"WorkflowState.COMPLETEDWorkflowState::Completed`	All steps succeeded — no compensation ran
`WorkflowState.FAILED"failed"WorkflowState.FAILEDWorkflowState::Failed`	A step failed and no completed step had a compensator
`WorkflowState.COMPENSATED"compensated"WorkflowState.COMPENSATEDWorkflowState::Compensated`	A step failed; every applicable compensator succeeded
`WorkflowState.COMPENSATION_FAILED"compensation_failed"WorkflowState.COMPENSATION_FAILEDWorkflowState::CompensationFailed`	A step failed; at least one compensator also failed

Manual intervention

CompensationFailed needs a human: inspect each step's status (

run.status().nodeshandle.nodes()status.nodeswf.get_workflow_nodes(run_id)

) to find which rollback failed and whether data is left partially consistent.

These four are a subset of WorkflowState's six terminal states — is_terminal() is also True for CANCELLED (compensation never runs on a cancelled run) and COMPLETED_WITH_FAILURES, covered next.

Read that table as a vocabulary, not a state machine that runs. Every one of those four is a value you pass to update_workflow_run_state yourself; nothing transitions a run on its own.

WorkflowState has ten variants in all — the four above plus Pending, Running, Paused, Compensating, Cancelled and CompletedWithFailures. What it also ships is the legal-move table, which is worth using even though the writer does not:

rust

use flexiq::workflows::WorkflowState;

assert!(WorkflowState::Running.can_transition_to(WorkflowState::Compensating));
assert!(WorkflowState::Compensating.can_transition_to(WorkflowState::Compensated));
// Terminal is terminal — nothing leaves Compensated.
assert!(!WorkflowState::Compensated.can_transition_to(WorkflowState::Running));

update_workflow_run_state does not call it. Gate on it before the write or a retried handler will happily move a Compensated run back to Running. Note which edges exist: Compensating is reachable from Running, Completed, CompletedWithFailures and Failed, and Cancelled has no outgoing edge at all — a cancelled run cannot be compensated, which matches the rule everywhere else.

Compensating on continue-mode workflows

By default, on_failure="continue" workflows never trigger compensation — they run every step regardless of individual failures and always reach COMPLETED_WITH_FAILURES if any step failed. Opt in with compensate_on_continue=True:

python

wf = Workflow(
    name="resilient_checkout",
    on_failure="continue",
    compensate_on_continue=True,
)
wf.step("charge", charge_payment, args=("4212",))
wf.step("ship", ship_order, args=("4212",), after="charge")

run = queue.submit_workflow(wf)

With compensate_on_continue=True, once the run settles, a run with at least one failed step transitions from COMPLETED_WITH_FAILURES into Compensating — compensators for every succeeded step then run in reverse-topological order, same as a fail_fast saga, and the run ends Compensated or CompensationFailed. Without it, a continue-mode run with failures just stays COMPLETED_WITH_FAILURES and no compensation runs.

Node: no continue-mode compensation

The Node builder has no run-level failure policy, so there is no compensateOnContinue equivalent. on_failure here is only a per-step condition ("on_success" | "on_failure" | "always"), not a continue-vs-abort mode for the whole run. Saga compensation is declared per step with compensate and runs only when the run itself fails.

Java: no continue-mode compensation

The Workflow builder has no run-level failure policy and no compensateOnContinue. onFailure() is only a per-step condition shorthand. Compensation is declared per step with Step.Builder.compensate(...) and rolls back only when the run fails.

No run-level failure policy to opt into

SubmitStaticWorkflowRequest carries no on_failure mode and no compensate_on_continue. Fail-fast versus continue is not a setting here — it is what your own code does when it sees a failed node.

CompletedWithFailures is still a real state you can write, and can_transition_to allows CompletedWithFailures → Compensating, so the continue-then-compensate shape is expressible. It is just a sequence of writes, not a flag.

Inside a single durable-step task the distinction is which shape the step's value takes:

rust

// Fail fast — the abort ends the attempt here.
let charge = step.run_keyed("charge", &order_id, || charge(&order_id))?;

// Continue — the failure is committed as the step's value, so the body runs on.
let shipped: Result<(), String> = step.run_keyed("ship", &order_id, || {
    Ok(ship(&order_id).map_err(|e| e.to_string()))
})?;

Only the second one keeps going. .is_ok() on the first would not: an aborted step commits nothing and leaves its position outstanding, so the next step in that attempt is refused. Continue mode here means a step that succeeds at recording a failure, which is also what makes the outcome survive a retry.

Idempotency

Every compensation job is enqueued with a deterministic idempotency key — compensation:{run_id}:{node_name} — so a compensation that's re-triggered (for example after a worker or tracker restart mid-rollback) hits the existing dedup and becomes a no-op instead of running the rollback twice.

That guarantee is at the job level, not the function body: a worker can still retry a compensator on a transient failure before its own retries are exhausted. Write compensator bodies to be safely re-runnable, the same way you would any other task.

Combining with conditions

Conditions and compensation compose. An on_failure step runs first as the run settles toward failure; once the run is failed, compensation then rolls back the completed compensable steps:

python

wf = Workflow(name="safe-checkout")
wf.step("reserve", reserve_inventory, args=("4212",), compensates=unreserve_inventory)
wf.step("charge", charge_payment, args=("4212",), after="reserve", compensates=refund_payment)
wf.step("ship", ship_order, args=("4212",), after="charge")
wf.step("notify_failure", send_failure_email, args=("4212",), after="ship", condition="on_failure")

queue.workflows
  .define("safe-checkout")
  .step("reserve", "reserveInventory", { compensate: "unreserveInventory" })
  .step("charge", "chargePayment", { after: "reserve", compensate: "refundPayment" })
  .step("ship", "shipOrder", { after: "charge" })
  .step("notifyFailure", "sendFailureEmail", {
    after: "ship",
    condition: "on_failure",
  })
  .submit();

java

int orderId = 4212; // the workflow input every step receives

Workflow safeCheckout = Workflow.named("safe-checkout")
        .step(Step.of("reserve", reserve, orderId).compensate(unreserve).build())
        .step(Step.of("charge", charge, orderId).after("reserve").compensate(refund).build())
        .step("ship", ship, orderId, "charge")
        .step(Step.of("notifyFailure", sendFailureEmail, orderId)
                .onFailure()
                .after("ship")
                .build());

condition is a step field, and like compensate it is an opaque string that submission records but never evaluates:

rust

StepMetadata {
    task_name: "send_failure_email".into(),
    condition: Some("on_failure".into()),
    ..Default::default()
}

submit_workflow pre-enqueues every non-deferred node up front and wires it with depends_on, and a dependency counts as satisfied only once it reaches Complete — so an on_failure step submitted this way waits forever behind the step it was meant to react to. Branch on outcome the other way round: list the node in deferred_node_names so it gets a WorkflowNode row and no job, then enqueue it yourself once the run's node statuses tell you to.

Events

Saga lifecycle emits dedicated events on the queue's event bus, subscribed via queue.on_event(event_type, callback):

Event	Fires when
`WORKFLOW_COMPLETED_WITH_FAILURES`	A `continue`-mode run finished with at least one failure
`WORKFLOW_COMPENSATING`	The run enters the compensation phase
`WORKFLOW_COMPENSATED`	Every compensator succeeded
`WORKFLOW_COMPENSATION_FAILED`	At least one compensator failed
`NODE_COMPENSATING`	A step's compensator was enqueued
`NODE_COMPENSATED`	A step's compensator finished successfully
`NODE_COMPENSATION_FAILED`	A step's compensator failed after retries

Node: no saga events

The Node SDK emits only the four job-lifecycle events — job.completed, job.retrying, job.dead, job.cancelled — via queue.on(...). There are no WORKFLOW_COMPENSATING / NODE_COMPENSATED-style saga events. Track compensation progress by polling the run handle: handle.status() and handle.nodes().

Java: no saga events

Java's EventName enum has only the four job outcomes — SUCCESS, RETRY, DEAD, CANCELLED. No workflow/saga transition events are emitted. Poll WorkflowRun / status.nodes for compensation progress.

There is no event bus — not for saga transitions and not for job outcomes. WorkerBuilder exposes no hook at all; flexiq_core::Worker::on_outcome is the nearest seam and the builder does not reach it.

For a saga written as durable steps this matters less than it looks, because the rollback is a branch in code you wrote — put the notification on the branch:

rust

if let Err(reason) = shipped {
    notify_rollback(&order_id, &reason);   // your own transport
    step.run_keyed("refund", &order_id, || refund(&charge))?;
}

Send it outside the step closures, not inside one. A step body is memoized: on a replay it does not run, so a notification in there fires once for the first attempt and never again — which is right for a charge and wrong for an alert.

For a saga spread over workflow nodes, the compensation columns are the record to poll: get_workflow_nodes(run_id) and look at status, compensation_started_at, compensation_completed_at and compensation_error. WorkflowNodeStatus::{Compensating, Compensated, CompensationFailed} are the three to watch, and CompensationFailed is the one that wants a human.

Sagas

Basic example

How it works

Watch a saga roll back

Declaring compensators

Decorator-level default

Steps without compensators

Final run states

Compensating on continue-mode workflows

Idempotency

Combining with conditions

Events

See also

Basic example

How it works

Watch a saga roll back

Declaring compensators

Decorator-level default

Steps without compensators

Final run states

Compensating on continue-mode workflows

Idempotency

Combining with conditions

Events

See also