Absolutely — here are Prefect 3.x examples for those backward recovery / compensation failure modes. I’m assuming you want Saga-style orchestration patterns in Prefect 3, where: - forward steps commit business mutations - a later permanent failure triggers rollback - compensations are modeled explicitly in the flow - compensation failures are observable and actionable, not hidden There isn’t a built-in “Saga engine” in Prefect 3, so the best pattern is to model compensation as first-class tasks/flows with: - explicit compensation records - verification of external state - time bounds / timeout handling - DLQ/manual-review escalation - rollback that continues past failed compensations where safe Relevant docs: - Prefect docs - Flows - Tasks --- # 1) Core Prefect 3 compensation pattern A practical pattern is: 1. Each forward step returns a compensation record 2. Store those records in a stack 3. If a permanent failure happens, iterate the stack in reverse 4. Each compensation: - validates target state - attempts reversal - verifies real external outcome - escalates if ambiguous or irreversible 5. Compensation failures are collected, not allowed to silently abort rollback Example skeleton:

python
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Any, Optional
from uuid import uuid4

from prefect import flow, task, get_run_logger


class CompensationStatus(str, Enum):
    PENDING = "pending"
    COMPENSATED = "compensated"
    FAILED = "failed"
    STALE = "stale"
    MANUAL_REVIEW = "manual_review"
    IRREVERSIBLE = "irreversible"


@dataclass
class CompensationRecord:
    step_name: str
    resource_id: str
    operation_id: str
    committed_at: datetime
    expected_version: Optional[str] = None
    metadata: dict[str, Any] = field(default_factory=dict)
    status: CompensationStatus = CompensationStatus.PENDING
    error: Optional[str] = None


@task
def send_to_dlq(record: CompensationRecord, reason: str) -> None:
    logger = get_run_logger()
    logger.error(f"DLQ: step={record.step_name} resource={record.resource_id} reason={reason}")


@task
def request_manual_review(record: CompensationRecord, reason: str) -> None:
    logger = get_run_logger()
    logger.error(
        f"MANUAL REVIEW REQUIRED: step={record.step_name} "
        f"resource={record.resource_id} reason={reason}"
    )


@flow
def rollback(records: list[CompensationRecord]) -> list[CompensationRecord]:
    logger = get_run_logger()
    failures: list[CompensationRecord] = []

    for record in reversed(records):
        try:
            <http://logger.info|logger.info>(f"Compensating {record.step_name} for {record.resource_id}")
            # dispatch to a compensation function by step_name
        except Exception as exc:
            record.status = CompensationStatus.FAILED
            record.error = str(exc)
            send_to_dlq.submit(record, f"compensation exception: {exc}")
            failures.append(record)
            continue

    return failures

That’s the baseline. Below are the four failure modes with concrete Prefect 3 examples. --- # 2) Stale State Compensation ## Problem The compensation acts on state that has drifted since the forward mutation committed. Example: - you reserve inventory - later you try to release that reservation - but the reservation already expired or was converted into a shipment - the compensation call returns “ok” or no-op - Prefect thinks rollback succeeded, but external state is still wrong ## What to do Before compensating, validate that: - the target resource still exists - it is still in the expected pre-compensation state - ideally the version / status / timestamp matches what the original step committed If not, do not mark compensated. Escalate to manual review. ## Prefect 3 example

python
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Optional

from prefect import flow, task, get_run_logger


@dataclass
class ReservationState:
    reservation_id: str
    status: str
    version: str
    expires_at: datetime


@dataclass
class CompensationRecord:
    step_name: str
    resource_id: str
    operation_id: str
    committed_at: datetime
    expected_version: Optional[str] = None
    status: str = "pending"
    error: Optional[str] = None


def fetch_reservation(reservation_id: str) -> ReservationState:
    # Replace with real API call
    return ReservationState(
        reservation_id=reservation_id,
        status="expired",
        version="v2",
        expires_at=datetime.now(timezone.utc) - timedelta(minutes=5),
    )


def release_reservation(reservation_id: str) -> bool:
    # Replace with real API call
    return True


@task
def request_manual_review(record: CompensationRecord, reason: str) -> None:
    logger = get_run_logger()
    logger.error(f"MANUAL REVIEW: {record.resource_id} reason={reason}")


@task
def compensate_inventory_reservation(record: CompensationRecord) -> CompensationRecord:
    logger = get_run_logger()
    current = fetch_reservation(record.resource_id)

    if record.expected_version and current.version != record.expected_version:
        record.status = "manual_review"
        record.error = (
            f"stale state: expected version {record.expected_version}, got {current.version}"
        )
        request_manual_review.submit(record, record.error)
        return record

    if current.status != "reserved":
        record.status = "manual_review"
        record.error = f"stale state: reservation is {current.status}, not reserved"
        request_manual_review.submit(record, record.error)
        return record

    release_reservation(record.resource_id)

    # verify
    current_after = fetch_reservation(record.resource_id)
    if current_after.status == "reserved":
        record.status = "manual_review"
        record.error = "compensation returned success but reservation still active"
        request_manual_review.submit(record, record.error)
        return record

    record.status = "compensated"
    <http://logger.info|logger.info>(f"Released reservation {record.resource_id}")
    return record


@flow
def stale_state_example():
    record = CompensationRecord(
        step_name="reserve_inventory",
        resource_id="res_123",
        operation_id="op_abc",
        committed_at=datetime.now(timezone.utc),
        expected_version="v1",
    )
    return compensate_inventory_reservation(record)

## Why this works This directly handles your failure mode: Signal: compensation returns success but external state is unchanged We don’t trust the compensation call alone; we: - validate preconditions before compensating - verify postconditions after compensating - escalate on drift instead of silently succeeding --- # 3) Cascading Compensation Failure ## Problem A compensation for step N fails and blocks compensations for N-1, N-2, etc. That is usually an orchestration bug: rollback is written as “fail-fast”, so one stuck compensation prevents the rest. ## What to do When safe, rollback should: - proceed in reverse order - catch and record compensation failures - send failed compensations to a DLQ/manual queue - continue compensating remaining steps You only stop if later compensations are unsafe without the failed one. ## Prefect 3 example ```python from dataclasses import dataclass from datetime import datetime, timezone from typing import Callable from prefect import flow, task, get_run_logger @dataclass class CompensationRecord: step_name: str resource_id: str operation_id: str committed_at: datetime status: str = "pending" error: str | None = None @task def send_to_dlq(record: CompensationRecord, reason: str) -> None: logger = get_run_logger()

logger.error(f"DLQ: {record.step_name} {record.resource_id} reason={reason}") @task def compensate_payment_auth(record: CompensationRecord) -> CompensationRecord: # pretend this fails raise RuntimeError("payment gateway void endpoint unavailable") @task def compensate_inventory(record: CompensationRecord) -> CompensationRecord: logger = get_run_logger() logger.info(f"Released inventory for {record.resource_id}") record.status = "compensated" return record @task def compensate_order_record(record: CompensationRecord) -> CompensationRecord: logger = get_run_logger() logger.info(f"Cancelled order record {record.resource_id}") record.status = "compensated" return record COMPENSATORS: dict[str, Callable] = { "authorize_payment": compensate_payment_auth, "reserve_inventory": compensate_inventory, "create_order": compensate_order_record, } @flow def rollback_with_continuation(records: list[CompensationRecord]) -> dict: logger = get_run_logger() completed = [] failed = [] for record in reversed(records): compensator = COMPENSATORS[record.step_name] try: result = compensator.fn(record) completed.append(result) except Exception as exc: record.status = "failed" record.error = str(exc) send_to_dlq.submit(record, str(exc)) failed.append(record) logger.error( f"Compensation failed for {record.step_name}/{record.resource_id}: {exc}. " f"Continuing rollback for remaining steps." ) continue return { "compensated": completed, "failed": failed, } @flow def cascading_compensation_failure_example(): records = [ CompensationRecord( step_name="create_order", resource_id="order_1", operation_id="op1", committed_at=datetime.now(timezone.utc), ), CompensationRecord( step_name="reserve_inventory", resource_id="inv_1", operation_id="op2", committed_at=datetime.now(timezone.utc), ), CompensationRecord( step_name="authorize_payment", resource_id="pay_1", operation_id="op3", committed_at=datetime.now(timezone.utc), ), ] return rollback_with_continuation(records) ``` ## Important note I used

.fn(...)

inside the flow here to keep rollback sequential and explicit. That is often simpler for compensation chains because order matters. ## Why this works It addresses your fix directly: - failed compensation goes to DLQ immediately - rollback continues for remaining safe steps - one stuck compensation does not hold the whole rollback hostage --- # 4) Compensation Latency Violation ## Problem Compensation takes too long, and downstream consumers act on state that is “about to be reversed”. Examples: - order confirmation email fires before payment reversal completes - shipment workflow sees inventory still reserved during a slow rollback window ## What to do Define a maximum compensation window per step. In Prefect, model that with: - task-level timeouts - explicit verification deadlines - gating steps on compensation status - escalation if compensation misses its deadline ## Prefect 3 example ```python from dataclasses import dataclass from datetime import datetime, timezone from prefect import flow, task, get_run_logger @dataclass class CompensationRecord: step_name: str resource_id: str operation_id: str committed_at: datetime status: str = "pending" error: str | None = None def slow_external_cancel(order_id: str) -> None: import time time.sleep(45) # simulate slow compensation @task(timeout_seconds=30) def compensate_order_with_sla(record: CompensationRecord) -> CompensationRecord: logger = get_run_logger()

logger.info(f"Starting compensation for {record.resource_id}") slow_external_cancel(record.resource_id) record.status = "compensated" return record @task def route_latency_violation(record: CompensationRecord, reason: str) -> None: logger = get_run_logger() logger.error( f"COMPENSATION SLA VIOLATION for {record.resource_id}: {reason}. " f"Route to async recovery/manual review." ) @task def gate_customer_notification(order_id: str, compensation_status: str) -> None: logger = get_run_logger() if compensation_status in {"pending", "failed", "manual_review"}: logger.warning( f"Notification for {order_id} blocked because compensation status is {compensation_status}" ) return logger.info(f"Sending customer notification for {order_id}") @flow def compensation_latency_violation_example(): record = CompensationRecord( step_name="cancel_order", resource_id="order_123", operation_id="op_123", committed_at=datetime.now(timezone.utc), ) try: result = compensate_order_with_sla(record) status = result.status except Exception as exc: record.status = "manual_review" record.error = f"compensation exceeded max window: {exc}" route_latency_violation(record, record.error) status = record.status gate_customer_notification(record.resource_id, status) return record ``` ## Why this works This maps to your fix: - define max acceptable compensation window - if the step can’t compensate fast enough, escalate - gate downstream actions on compensation state ## Extra recommendation If downstream consumers are external systems, the stronger pattern is: - publish business events only after compensation-sensitive windows close, or - include a status like

order_state=provisional

until rollback risk is resolved Prefect can orchestrate that, but the enforcement usually lives in surrounding application architecture. Docs: - Tasks in Prefect --- # 5) Silent Compensation Success on Permanent External Failure ## Problem The compensation API returns success, but the original mutation is already irreversible. Examples: - payment already settled - shipment already dispatched - email already sent - external processor accepted the action and cannot truly undo it This is the most dangerous one because the workflow looks “green”. ## What to do Treat these as escalations, not normal compensations: - compensation call is not enough - you must verify external business state - if the operation is irreversible, mark

manual_review

/

irreversible

- do not record “compensated” until confirmed ## Prefect 3 example ```python from dataclasses import dataclass from datetime import datetime, timezone from typing import Literal from prefect import flow, task, get_run_logger @dataclass class PaymentState: payment_id: str status: Literal["authorized", "voided", "settled", "refunded"] @dataclass class CompensationRecord: step_name: str resource_id: str operation_id: str committed_at: datetime status: str = "pending" error: str | None = None def void_payment(payment_id: str) -> dict: # Fake API says "success" return {"ok": True} def fetch_payment_state(payment_id: str) -> PaymentState: # Real audit source says payment is settled and irreversible return PaymentState(payment_id=payment_id, status="settled") @task def request_manual_review(record: CompensationRecord, reason: str) -> None: logger = get_run_logger() logger.error(f"MANUAL REVIEW REQUIRED: {record.resource_id} reason={reason}") @task def compensate_payment(record: CompensationRecord) -> CompensationRecord: logger = get_run_logger() response = void_payment(record.resource_id) if not response.get("ok"): record.status = "failed"

record.error = "payment void API returned failure" return record verified = fetch_payment_state(record.resource_id) if verified.status == "voided": record.status = "compensated" logger.info(f"Payment {record.resource_id} successfully voided") return record if verified.status == "settled": record.status = "irreversible" record.error = "payment already settled; cannot be voided automatically" request_manual_review.submit(record, record.error) return record record.status = "manual_review" record.error = f"ambiguous payment state after compensation attempt: {verified.status}" request_manual_review.submit(record, record.error) return record @flow def silent_success_permanent_failure_example(): record = CompensationRecord( step_name="authorize_payment", resource_id="pay_123", operation_id="op_123", committed_at=datetime.now(timezone.utc), ) return compensate_payment(record) ``` ## Why this works It directly handles your fix: - irreversible external actions are not treated like normal transactions - orchestrator does not mark them “cleanly compensated” - human intervention is required before resolution is considered complete --- # 6) A fuller Saga example in Prefect 3 If you want all four ideas together, here’s a compact orchestration pattern. ```python from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any from prefect import flow, task, get_run_logger @dataclass class CompensationRecord: step_name: str resource_id: str operation_id: str committed_at: datetime expected_version: str | None = None metadata: dict[str, Any] = field(default_factory=dict) status: str = "pending" error: str | None = None @task def create_order() -> CompensationRecord: return CompensationRecord( step_name="create_order", resource_id="order_123", operation_id="op_order", committed_at=datetime.now(timezone.utc), ) @task def reserve_inventory() -> CompensationRecord: return CompensationRecord( step_name="reserve_inventory", resource_id="inv_123", operation_id="op_inv", committed_at=datetime.now(timezone.utc), expected_version="v1", ) @task def authorize_payment() -> CompensationRecord: return CompensationRecord( step_name="authorize_payment", resource_id="pay_123", operation_id="op_pay", committed_at=datetime.now(timezone.utc), ) @task def ship_order() -> None: raise RuntimeError("shipping service permanent failure") @task def send_to_dlq(record: CompensationRecord, reason: str): logger = get_run_logger() logger.error(f"DLQ: {record.step_name}/{record.resource_id} reason={reason}") @task def request_manual_review(record: CompensationRecord, reason: str): logger = get_run_logger() logger.error(f"MANUAL REVIEW: {record.step_name}/{record.resource_id} reason={reason}") def compensate_record(record: CompensationRecord) -> CompensationRecord: # Replace with real dispatch logic using the examples above if record.step_name == "authorize_payment": record.status = "irreversible" record.error = "payment may already be settled" elif record.step_name == "reserve_inventory": record.status = "manual_review" record.error = "inventory state drift detected" else: record.status = "compensated" return record @flow def order_saga(): logger = get_run_logger() compensation_stack: list[CompensationRecord] = [] try: compensation_stack.append(create_order()) compensation_stack.append(reserve_inventory()) compensation_stack.append(authorize_payment()) ship_order() return {"status": "completed"} except Exception as exc:

logger.error(f"Forward path failed permanently: {exc}") rollback_failures = [] for record in reversed(compensation_stack): try: result = compensate_record(record) if result.status in {"manual_review", "irreversible"}: request_manual_review.submit(result, result.error or "review required") elif result.status == "failed": send_to_dlq.submit(result, result.error or "failed") rollback_failures.append(result) except Exception as rollback_exc: record.status = "failed" record.error = str(rollback_exc) send_to_dlq.submit(record, record.error) rollback_failures.append(record) continue return { "status": "rolled_back_with_escalations", "rollback_failures": [r.resource_id for r in rollback_failures], }

---

# 7) Prefect 3 features that help here

## Task timeouts
Use `@task(timeout_seconds=...)` for compensation windows.

## Retries
Use retries for *transient* compensation failures, but not as a substitute for backward recovery logic.

Example:
```python
@task(retries=3, retry_delay_seconds=10)
def call_external_compensation():
    ...

## Subflows A good production pattern is to make rollback its own flow: - main flow handles forward path - rollback subflow processes compensation stack - manual-review flow handles escalations Docs: - Flows - Tasks ## Logging and auditability Use

get_run_logger()

heavily and log: - original mutation id - compensation id - expected state - observed state - verification source - escalation reason That matters a lot for reconciliation. --- # 8) Recommended design rules for these four failure modes ## For stale state compensation - record expected external version/status at commit time - validate before compensation - verify after compensation - escalate on drift ## For cascading compensation failure - never write rollback as fail-fast by default - continue compensating remaining safe steps - DLQ failed compensations immediately - preserve reverse-order execution ## For compensation latency violation - define a compensation SLA per step - enforce with task timeouts - block downstream actions until compensation resolves or is explicitly waived - route timed-out compensations to async/manual recovery ## For silent success on permanent external failure - never trust 200/OK alone - verify against system-of-record or audit API - distinguish reversible vs irreversible operations - irreversible => manual review, not “compensated” --- # 9) One important architecture note Prefect can orchestrate the rollback correctly, but it cannot by itself guarantee business consistency if the external systems: - don’t expose versioned state - don’t support verification - don’t support idempotency keys - have irreversible side effects For robust Sagas, the external APIs should ideally support: - idempotency keys - read-after-write verification - explicit statuses like

authorized/settled/voided

- version or etag checks - compensating actions that are semantically distinct from “success” --- If you want, I can turn this into either: 1. a single complete Prefect 3 sample project with mocked services, or 2. a realistic order/payment/inventory saga with

prefect deploy

-ready structure.