https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> i need prefect 3 examples for the bac...
# ask-marvin
o
@Marvin i need prefect 3 examples for the backward healing failires Backward Recovery Failure Modes Backward recovery — executing compensating transactions in response to a permanent failure — is the mechanism that restores a Saga to a consistent business-defined state. It fails quietly: the workflow appears to handle the error while leaving the system in a worse state than if it had never run. The failure modes below are distinct from forward recovery failures; they emerge from the compensation logic itself, not from retries. 1. Missing Compensation A step executes successfully and mutates external state, but no compensating transaction is defined. If a later step fails permanently, that mutation is stranded — the system has no path back to consistency. Signal: post-failure audits revealing orphaned records, reserved-but-never-released resources, or charges with no corresponding order. Fix: every state-mutating step must have a paired compensation defined before the step executes, not as an afterthought. No compensation, no mutation. 2. Non-Atomic Compensation The compensating transaction itself fails partway through, leaving the system in a partially compensated state — worse than either the committed state or the clean pre-transaction state. Signal: compensation acknowledged as started but external state still partially mutated; inconsistency between internal workflow state and downstream system records. Fix: compensations must be idempotent and retriable. A compensation that cannot be safely retried is not a compensation — it is a new failure surface. Apply the same idempotency discipline to compensations as to forward steps. 3. Wrong Compensation Order Compensations executed out of sequence produce referential integrity violations or business logic errors. Releasing an inventory reservation before voiding the payment that depended on it may trigger downstream re-allocation before the payment void completes. Signal: compensation completes without error but downstream systems are in a logically inconsistent state; errors appearing in systems that weren't part of the failed step. Fix: compensation order must be explicitly defined and enforced by the orchestrator, not assumed. Default to strict reverse-sequential unless the domain explicitly permits non-linear compensation. 4. Compensation of Already-Compensated State (Double Compensation) A compensation executes twice — due to an orchestrator restart, a duplicate trigger, or a race between concurrent failure handlers — producing a state that is doubly-reversed: a payment voided twice, inventory released twice. Signal: resource balances going negative; duplicate refund records; idempotency violations detected in downstream systems. Fix: compensations must be idempotent by key. The orchestrator must track compensation state durably so a restart replays the compensation status, not the compensation action.
m
✅ thought for 166.5 seconds
Here are Prefect 3 examples for each backward recovery failure mode, framed as Saga compensation patterns. Important note: in Prefect 3, the cleanest building blocks for this are: - 
@flow
/ 
@task
- normal Python 
try/except/finally
- durable state in an external system or persisted result store - explicit compensation functions - reverse-order compensation stacks I’d recommend not depending purely on “best effort” in-memory rollback logic for business-critical sagas. Persist compensation state by key. --- # 1. Missing Compensation Problem: a step mutates external state, but you forgot to define how to undo it. ## Bad example
Copy code
python
from prefect import flow, task

@task
def charge_customer(order_id: str) -> str:
    # external mutation
    return f"charge-{order_id}"

@task
def reserve_inventory(order_id: str) -> str:
    raise RuntimeError("inventory unavailable")

@flow
def order_flow(order_id: str):
    charge_id = charge_customer(order_id)
    reserve_inventory(order_id)
If 
reserve_inventory
fails, the charge remains. ## Better Prefect 3 pattern Define the compensation before or at the same time as the mutation is introduced into the flow.
Copy code
python
from prefect import flow, task, get_run_logger

@task
def charge_customer(order_id: str) -> str:
    return f"charge-{order_id}"

@task
def refund_customer(charge_id: str):
    print(f"Refunding {charge_id}")

@task
def reserve_inventory(order_id: str) -> str:
    raise RuntimeError("inventory unavailable")

@flow
def order_flow(order_id: str):
    logger = get_run_logger()
    compensations = []

    try:
        charge_id = charge_customer(order_id)
        compensations.append(lambda: refund_customer(charge_id))

        reserve_inventory(order_id)

    except Exception:
        logger.error("Flow failed; starting compensation")
        for compensate in reversed(compensations):
            compensate()
        raise
## Key rule No compensation, no mutation. If a task changes external state, register its undo path immediately. --- # 2. Non-Atomic Compensation Problem: compensation itself partially fails. Example: “cancel shipment” succeeds, but “refund payment” fails. ## Bad example
Copy code
python
from prefect import flow, task

@task
def cancel_shipment(shipment_id: str):
    print(f"Cancelled shipment {shipment_id}")

@task
def refund_payment(payment_id: str):
    raise RuntimeError("refund API timeout")

@flow
def compensate_order(shipment_id: str, payment_id: str):
    cancel_shipment(shipment_id)
    refund_payment(payment_id)
Now the order is partially compensated. ## Better pattern: idempotent, retriable compensations with durable status
Copy code
python
from prefect import flow, task, get_run_logger

# Replace with a real DB/table/object store in production
COMPENSATION_STATUS = {}

@task(retries=3, retry_delay_seconds=5)
def refund_payment(payment_id: str):
    if COMPENSATION_STATUS.get(f"refund:{payment_id}") == "done":
        return

    # external refund call
    print(f"Refunding payment {payment_id}")

    COMPENSATION_STATUS[f"refund:{payment_id}"] = "done"

@task(retries=3, retry_delay_seconds=5)
def cancel_shipment(shipment_id: str):
    if COMPENSATION_STATUS.get(f"cancel:{shipment_id}") == "done":
        return

    # external cancel call
    print(f"Cancelling shipment {shipment_id}")

    COMPENSATION_STATUS[f"cancel:{shipment_id}"] = "done"

@flow
def compensate_order(shipment_id: str, payment_id: str):
    logger = get_run_logger()

    try:
        cancel_shipment(shipment_id)
        refund_payment(payment_id)
    except Exception as exc:
        logger.error(f"Compensation failed: {exc}")
        raise
## What matters Each compensation must be: - idempotent - safe to retry - durably tracked If a worker crashes mid-compensation, a rerun should continue safely. --- # 3. Wrong Compensation Order Problem: compensation runs in the wrong sequence. Forward path: 1. reserve inventory
2. charge payment 3. create shipment Backward path should usually be: 1. cancel shipment 2. refund payment 3. release inventory ## Correct Prefect 3 pattern: reverse-sequential compensation
Copy code
python
from prefect import flow, task, get_run_logger

@task
def reserve_inventory(order_id: str) -> str:
    return f"inv-{order_id}"

@task
def release_inventory(reservation_id: str):
    print(f"Released inventory {reservation_id}")

@task
def charge_payment(order_id: str) -> str:
    return f"pay-{order_id}"

@task
def refund_payment(payment_id: str):
    print(f"Refunded payment {payment_id}")

@task
def create_shipment(order_id: str) -> str:
    raise RuntimeError("shipment service failure")

@task
def cancel_shipment(shipment_id: str):
    print(f"Cancelled shipment {shipment_id}")

@flow
def order_flow(order_id: str):
    logger = get_run_logger()
    compensations = []

    try:
        reservation_id = reserve_inventory(order_id)
        compensations.append(lambda: release_inventory(reservation_id))

        payment_id = charge_payment(order_id)
        compensations.append(lambda: refund_payment(payment_id))

        shipment_id = create_shipment(order_id)
        compensations.append(lambda: cancel_shipment(shipment_id))

    except Exception:
        logger.error("Failure detected, compensating in reverse order")
        for compensate in reversed(compensations):
            compensate()
        raise
## Why this works The compensation stack is built in forward order and executed in reverse order, which matches Saga rollback semantics. --- # 4. Double Compensation Problem: compensation runs twice after restart/retry/race. Examples: - refund issued twice - inventory released twice ## Better pattern: idempotency key + durable compensation record
Copy code
python
from prefect import flow, task

COMPENSATION_STATUS = {}

@task
def refund_payment(payment_id: str):
    key = f"refund:{payment_id}"

    if COMPENSATION_STATUS.get(key) == "done":
        print(f"Refund already completed for {payment_id}")
        return

    # call downstream API with idempotency key = key
    print(f"Refunding {payment_id}")

    COMPENSATION_STATUS[key] = "done"

@task
def release_inventory(reservation_id: str):
    key = f"release:{reservation_id}"

    if COMPENSATION_STATUS.get(key) == "done":
        print(f"Inventory already released for {reservation_id}")
        return

    print(f"Releasing inventory {reservation_id}")

    COMPENSATION_STATUS[key] = "done"

@flow(retries=2)
def compensate(payment_id: str, reservation_id: str):
    refund_payment(payment_id)
    release_inventory(reservation_id)
If the flow retries, compensation tasks check durable state before acting. ## Key rule The orchestrator should replay status, not blindly replay side effects. --- # A fuller Prefect 3 Saga example This combines all four ideas. ```python from prefect import flow, task, get_run_logger SAGA_STATE = {} @task def reserve_inventory(order_id: str) -> str: reservation_id = f"inv-{order_id}" SAGA_STATE[f"forward{order id}inventory"] = reservation_id return reservation_id @task def release_inventory(reservation_id: str): key = f"compensatedinventory{reservation_id}" if SAGA_STATE.get(key): return print(f"Released inventory {reservation_id}") SAGA_STATE[key] = True @task def charge_payment(order_id: str) -> str: payment_id = f"pay-{order_id}" SAGA_STATE[f"forward{order id}payment"] = payment_id return payment_id @task def refund_payment(payment_id: str): key = f"compensatedpayment{payment_id}" if SAGA_STATE.get(key): return print(f"Refunded payment {payment_id}") SAGA_STATE[key] = True @task def create_shipment(order_id: str) -> str: raise RuntimeError("shipping unavailable") @task def cancel_shipment(shipment_id: str): key = f"compensatedshipment{shipment_id}" if SAGA_STATE.get(key): return
print(f"Cancelled shipment {shipment_id}") SAGA_STATE[key] = True @flow def order_saga(order_id: str): logger = get_run_logger() compensations = [] try: reservation_id = reserve_inventory(order_id) compensations.append(("inventory", reservation_id, lambda: release_inventory(reservation_id))) payment_id = charge_payment(order_id) compensations.append(("payment", payment_id, lambda: refund_payment(payment_id))) shipment_id = create_shipment(order_id) compensations.append(("shipment", shipment_id, lambda: cancel_shipment(shipment_id))) logger.info("Order completed successfully") except Exception as exc: logger.error(f"Order failed: {exc}") logger.info("Running compensations in reverse order") for name, resource_id, compensate in reversed(compensations): try: compensate() except Exception as comp_exc: logger.error(f"Compensation failed for {name}:{resource_id} - {comp_exc}") raise raise ``` --- # Practical Prefect 3 guidance ## Recommended design Use: - one flow as the saga orchestrator - one task per forward step - one task per compensation step - a reverse-order compensation list - durable idempotency keys in your DB / payment provider / inventory service ## Avoid - compensations that are only in memory - compensations without retry safety - unordered parallel rollback unless your domain explicitly allows it - assuming a flow retry means it is safe to rerun side effects ## Production note In real systems, replace the in-memory 
dict
examples with: - Postgres - Redis - S3/GCS-backed records - downstream API idempotency keys - your business database --- # Useful Prefect docs - Prefect docs - Flows - Tasks If you want, I can also turn this into: 1. a payment/inventory/order example, 2. a travel booking saga example, or 3. a Prefect 3 + async tasks + persisted compensation log example.
Open in Slack
PreviousNext
Powered by