Saga

Saga compensation primitives. Import paths:

from flexiq.workflows.saga import (
    CompensationContext,
    SagaDepthExceededError,
    SagaOrchestrator,
    current_compensation_context,
)

For the conceptual overview, see Sagas.

@queue.task(compensates=…)

Register a compensator for a task. Accepted forms:

@queue.task
def refund(forward_args, forward_kwargs, forward_result): ...

@queue.task(compensates=refund)
def charge(amount: int) -> str: ...

The compensator must accept exactly three positional arguments:

Workflow.step(…, compensates=…)

Override or set the compensator for a single workflow step:

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

Accepted: TaskWrapper, str, None, or the INHERIT_COMPENSATOR sentinel (the default — falls back to whatever @queue.task(compensates=…) declared).

current_compensation_context()

Returns the CompensationContext for the currently-executing compensator, or None outside of one. Useful for introspecting the forward execution without unpacking positional args.

from flexiq.workflows.saga import current_compensation_context

@queue.task
def refund(forward_args, forward_kwargs, forward_result):
    ctx = current_compensation_context()
    if ctx is not None:
        logger.info(
            "compensating %s/%s (forward job %s)",
            ctx.workflow_run_id,
            ctx.workflow_node_name,
            ctx.forward_job_id,
        )

CompensationContext

Frozen dataclass passed in via the contextvar. Fields:

Workflow states added by sagas

WorkflowState.is_terminal() returns True for Compensated and CompensationFailed. WorkflowRun.wait() unblocks on any of the six terminal states (Completed, CompletedWithFailures, Failed, Cancelled, Compensated, CompensationFailed).

Node statuses added by sagas

Only Completed and CacheHit nodes are eligible for compensation (is_compensable() returns True for those two). Failed, skipped, pending, and ready nodes are never compensated — their side effects never happened.

Events

Subscribe to these on the queue's event bus via queue.on_event(event_type, callback):

SagaDepthExceededError

Raised when nested saga propagation (sub-workflow whose child is also a saga whose child is also a saga, …) exceeds the hard cap of 10 levels. Almost always means a workflow graph has accidental sub-workflow recursion.

Idempotency

Every compensation job is enqueued with idempotency_key=f"compensation:{run_id}:{node_name}". The first call creates the job; a duplicate (e.g. after a tracker restart) hits the existing partial-index dedup in all three storage backends and becomes a no-op. Compensator function bodies should also be designed to be safely re-runnable — the at-most-one guarantee is at the job level, but the worker can still retry the job's body on transient failures before exhaustion.

Sub-workflow propagation

When a parent workflow's saga reaches a sub-workflow node whose child run carries its own saga, the parent propagates compensation into the child:

1. Parent compensation dispatches the sub-workflow node.
2. Orchestrator detects the node is in sub_workflow_refs and the child has compensators.
3. Child saga starts (depth = parent depth + 1).
4. When the child saga finalises (Compensated or CompensationFailed), the parent's node is marked accordingly and the parent's wave advances.

If the parent step has its own explicit compensates=…, that wins — no propagation happens. See Sagas for a worked example.