Rollback and compensation

Use task.with_options(rollback=...) to undo side effects when a task fails, and implement the saga pattern by manually compensating completed steps in reverse order.

When a task writes to an external system — say, charging a payment or creating a database row — a failure partway through the workflow leaves those side effects in place. Retries and fallbacks handle transient failures; rollback handles the permanent ones, giving you a named function to undo what the task already did.

Attaching a rollback function

Pass a callable to rollback= on task.with_options(). When the task fails and has no fallback (and no remaining retries), Flux calls the rollback function with the same arguments that were passed to the task:

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


async def delete_record(record_id: str, data: dict) -> None:
    """Undo the create_record side effect."""
    print(f"rolling back: deleting record {record_id}")


@task.with_options(rollback=delete_record)
async def create_record(record_id: str, data: dict) -> dict:
    print(f"creating record {record_id}")
    raise IOError("database unreachable")


@workflow
async def sync_record(ctx: ExecutionContext) -> None:
    await create_record("rec-1", {"name": "Alice"})


if __name__ == "__main__":
    try:
        sync_record.run()
    except Exception:
        pass
# creating record rec-1
# rolling back: deleting record rec-1

The rollback function receives *args, **kwargs exactly as the task received them. It can be a plain def or an async def — Flux calls it with maybe_awaitable, so both forms work.

What rollback does not do automatically

Each task’s rollback function runs only for that one task. When step3 fails:

This is the core distinction from database transactions: Flux has no built-in saga coordinator. To compensate completed steps, you write the compensation logic yourself — which is the saga pattern.

The saga pattern

A saga breaks a long-running operation into a sequence of steps, each paired with a compensation. When one step fails, you undo all previously completed steps in reverse order. In Flux, the saga looks like this:

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


# ── compensation tasks ──────────────────────────────────────────────────────

@task
async def undo_create_order(order_id: str) -> None:
    print(f"  compensate: cancel order {order_id}")


@task
async def undo_reserve_inventory(order_id: str) -> None:
    print(f"  compensate: release inventory for {order_id}")


@task
async def undo_charge_payment(order_id: str) -> None:
    print(f"  compensate: refund payment for {order_id}")


# ── forward tasks ────────────────────────────────────────────────────────────

@task.with_options(rollback=undo_create_order)
async def create_order(order_id: str) -> str:
    print(f"  ok: create_order({order_id})")
    return f"order:{order_id}"


@task.with_options(rollback=undo_reserve_inventory)
async def reserve_inventory(order_id: str) -> str:
    print(f"  ok: reserve_inventory({order_id})")
    return f"reserved:{order_id}"


@task.with_options(rollback=undo_charge_payment)
async def charge_payment(order_id: str) -> str:
    print(f"  ok: charge_payment({order_id})")
    raise ValueError("payment gateway down")


# ── saga workflow ─────────────────────────────────────────────────────────────

@workflow
async def place_order(ctx: ExecutionContext[str]) -> None:
    order_id = ctx.input
    completed = []

    try:
        await create_order(order_id)
        completed.append("create_order")

        await reserve_inventory(order_id)
        completed.append("reserve_inventory")

        await charge_payment(order_id)
        completed.append("charge_payment")

    except Exception:
        # charge_payment's rollback= ran automatically for the failing step.
        # Now compensate each previously completed step in reverse order.
        for step in reversed(completed):
            if step == "reserve_inventory":
                await undo_reserve_inventory(order_id)
            elif step == "create_order":
                await undo_create_order(order_id)
        raise


if __name__ == "__main__":
    try:
        place_order.run("ORD-001")
    except Exception:
        pass

Running this prints:

  ok: create_order(ORD-001)
  ok: reserve_inventory(ORD-001)
  ok: charge_payment(ORD-001)
  compensate: refund payment for ORD-001    ← automatic via rollback=
  compensate: release inventory for ORD-001 ← manual, reversed
  compensate: cancel order ORD-001          ← manual, reversed

The verified execution order, confirmed against Flux 0.56.0:

  1. All forward steps run in declaration order.
  2. The failing step’s rollback= function runs first (automatically, before the exception propagates to the workflow).
  3. The workflow’s except block calls the remaining compensation tasks in reverse order of completion.

Keeping compensation tasks idempotent

A compensation task may be called more than once: if the workflow is interrupted mid-compensation and replayed, Flux re-runs tasks whose outcome was not yet checkpointed. Design compensation functions so that calling them twice produces the same result as calling them once — for example, by using an idempotency key on the external API call, or by checking whether the resource still exists before deleting it.

Handling a failed rollback

If the rollback function itself raises, Flux records a TASK_ROLLBACK_FAILED event and re-raises the exception. The workflow fails, and no further compensation runs.

If the compensation can safely be skipped (for example, another process already cleaned up the resource), catch and log the error inside the rollback body. If the compensation may fail transiently, decorate the compensation function with retry_max_attempts so it retries before giving up. If neither applies, emit a structured log entry or call an alerting task before re-raising so the operations team knows which step needs manual cleanup.