https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> i need prefect example for 8 forward ...
# ask-marvin
o
@Marvin i need prefect example for 8 forward recovery failure modes # Forward Recovery Failure Modes Forward recovery—`retry(policy)` applied to distributed, state-mutating steps under partial failure—introduces a distinct class of failures. These are a leading cause of production incidents in orchestrated systems. * ## 1. Misclassification Problem Forward recovery is applied to a permanent failure. The orchestrator retries indefinitely against a condition that will never succeed, consuming resources and blocking correct recovery. Signal Retries exhaust their budget with no transient error patterns (e.g., consistent 4xx, deterministic exceptions). Fix Explicitly classify failures before acting: * Transient → retry * Permanent → immediate compensation / fail-fast * ## 2. Retry Storms Problem Concurrent workflows retry in synchronized waves after a shared failure, overwhelming the recovering system and re-triggering the outage. Fix * Exponential backoff with jitter * Global/system-level retry budgets (not just per-instance limits) * ## 3. Side-Effect Amplification Problem Retries on non-idempotent operations duplicate external effects (e.g., multiple charges, duplicate emails). Fix * Enforce idempotency keys on all external calls * If unsupported downstream: * Persist call intent before execution * Deduplicate on retry via orchestrator state * ## 4. Temporal Drift Problem State assumed valid at initial execution becomes stale by retry time, leading to externally incorrect outcomes despite internal consistency. Fix * Revalidate preconditions before retry * Treat time-sensitive dependencies as invalid across retry boundaries * ## 5. Semantic Drift Problem Retries yield different valid outputs over time (e.g., pricing changes), causing internal inconsistency across steps. Fix * Treat non-deterministic steps as checkpoints * Persist first successful output * Replay instead of re-executing on retry * ## 6. Unbounded Duration Problem Workflows with no time limits run indefinitely, consuming scheduler capacity, locks, and system resources. Fix * Enforce TTLs at workflow and step level * Route terminal failures to a DLQ for inspection * ## 7. Compensation Surface Expansion Problem Retries increase the number of side effects requiring compensation. Compensation complexity grows with retry depth, not just workflow length. Fix * Strict retry limits on state-mutating steps * Prefer fast failure + compensation over deep retries * ## 8. Backpressure & Throughput Collapse Problem Retries amplify load on already degraded systems, pushing them past failure thresholds. Simultaneously, accumulation of failed workflows degrades orchestrator performance (state store pressure, scheduler latency). Fix * Circuit breakers at orchestrator boundaries * Global retry throttling / budgets * TTL + DLQ to remove terminal workflows from active scheduling
m
✅ thought for 244.6 seconds
Here’s a Prefect 3.x example set for each of the 8 forward recovery failure modes you listed. I’m assuming you want practical code patterns you can adapt, not just conceptual mapping. The key Prefect features involved are: - 
retries
- 
retry_delay_seconds
- 
retry_jitter_factor
- 
retry_condition_fn
- 
timeout_seconds
- task caching / custom cache keys - transactions - concurrency limits Relevant docs: - Tasks - Flows - Task caching - Global concurrency limits - Transactions --- # 1) Misclassification Retry only transient failures; fail fast on permanent ones.
Copy code
python
from prefect import flow, task
from prefect.client.schemas import TaskRun
from prefect.states import State

def retry_only_transient(task, task_run: TaskRun, state: State) -> bool:
    try:
        state.result()
        return False
    except Exception as exc:
        if isinstance(exc, (TimeoutError, ConnectionError, OSError)):
            return True

        if isinstance(exc, (ValueError, TypeError, KeyError, PermissionError)):
            return False

        return False

@task(
    retries=3,
    retry_delay_seconds=[2, 5, 10],
    retry_condition_fn=retry_only_transient,
)
def mutate_remote_record(record_id: str):
    if record_id == "bad-input":
        raise ValueError("record_id is invalid")
    raise ConnectionError("temporary network issue")

@flow
def misclassification_example():
    mutate_remote_record("bad-input")
Why this helps: Prefect will only retry when the failure class is actually recoverable. --- # 2) Retry Storms Add backoff + jitter and combine with a concurrency limit.
Copy code
python
from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task(
    retries=5,
    retry_delay_seconds=[1, 2, 4, 8, 16],
    retry_jitter_factor=0.5,
)
def call_shared_api(item_id: int):
    with concurrency("shared-api", occupy=1, timeout_seconds=5):
        raise ConnectionError("upstream unavailable")

@flow
def retry_storm_example():
    for i in range(20):
        call_shared_api.submit(i)
Why this helps: - jitter reduces synchronized retry waves - concurrency limits reduce herd pressure on the dependency To inspect/create concurrency limits in Prefect Cloud / Server, see: Global concurrency limits --- # 3) Side-Effect Amplification Protect non-idempotent actions with an idempotency key or cache key.
Copy code
python
from datetime import timedelta
from prefect import flow, task
from prefect.context import TaskRunContext

def payment_cache_key(task_run_context: TaskRunContext, inputs: dict):
    return inputs["idempotency_key"]

@task(
    retries=2,
    cache_key_fn=payment_cache_key,
    cache_expiration=timedelta(hours=24),
)
def charge_customer(customer_id: str, amount: float, idempotency_key: str):
    print(f"Charging {customer_id} for {amount}")
    return {"status": "charged", "customer_id": customer_id, "amount": amount}

@flow
def side_effect_amplification_example():
    charge_customer(
        customer_id="cust-123",
        amount=49.99,
        idempotency_key="order-987"
    )
Why this helps: if the task is retried with the same key, Prefect can reuse the prior successful result instead of reissuing the side effect. If downstream systems support their own idempotency keys, send the same key there too. That’s the safest pattern. Docs: - Task caching --- # 4) Temporal Drift Revalidate time-sensitive assumptions before retry. ```python from datetime import datetime, timedelta, timezone from prefect import flow, task from prefect.client.schemas import TaskRun from prefect.states import State
EXPIRATION_WINDOW = timedelta(minutes=5) def retry_if_not_expired(task, task_run: TaskRun, state: State) -> bool: submitted_at = task_run.expected_start_time or task_run.start_time if submitted_at is None: return True now = datetime.now(timezone.utc) if now - submitted_at > EXPIRATION_WINDOW: return False return True @task( retries=3, retry_delay_seconds=[30, 60, 120], retry_condition_fn=retry_if_not_expired, ) def reserve_inventory(sku: str): raise ConnectionError("inventory service timeout") @flow def temporal_drift_example(): reserve_inventory("sku-001")
Copy code
*Why this helps:* by the time a retry happens, the original preconditions may no longer be valid. This pattern stops retries once the action is too stale to be correct.

You can also use task/flow timeouts:

```python
@task(timeout_seconds=30, retries=2)
def do_work():
    ...
Docs: - Tasks - Flows --- # 5) Semantic Drift Checkpoint the first valid output and replay it instead of recomputing.
Copy code
python
from datetime import timedelta
from prefect import flow, task
from prefect.cache_policies import INPUTS

@task(
    cache_policy=INPUTS,
    cache_expiration=timedelta(hours=1),
)
def fetch_price(product_id: str):
    print("Calling pricing service")
    return {"product_id": product_id, "price": 100}

@task(retries=2)
def create_invoice(product_id: str):
    price_snapshot = fetch_price(product_id)
    # downstream tasks should use this frozen snapshot
    return {
        "product_id": product_id,
        "billed_price": price_snapshot["price"],
    }

@flow
def semantic_drift_example():
    return create_invoice("prod-42")
Why this helps: if pricing changes between attempts, later steps still use the original checkpointed result instead of silently drifting. This is especially useful for: - pricing - exchange rates - feature flags - schema-derived outputs Docs: - Task caching --- # 6) Unbounded Duration Always bound retries and execution time.
Copy code
python
from prefect import flow, task

@task(
    retries=4,
    retry_delay_seconds=[10, 30, 60, 120],
    timeout_seconds=20,
)
def flaky_step():
    raise TimeoutError("dependency never recovers")

@flow(timeout_seconds=300)
def unbounded_duration_example():
    flaky_step()
Why this helps: - task-level timeout bounds each attempt - retry count bounds total attempts - flow-level timeout prevents the whole workflow from lingering forever For a DLQ-style pattern in Prefect, a common approach is: - let the flow fail terminally - use an automation / webhook / notification for inspection or requeue Docs: - Flows - Automations --- # 7) Compensation Surface Expansion Limit retries on state-mutating tasks and use transactions/compensation hooks.
Copy code
python
from prefect import flow, task
from prefect.transactions import transaction

def refund_payment(payment_id: str):
    print(f"Refunding payment {payment_id}")

@task(retries=1)
def create_order():
    with transaction(key="order-123") as txn:
        payment_id = "pay-001"
        txn.set("payment_id", payment_id)

        # imagine this succeeded externally
        print(f"Captured payment {payment_id}")

        # later step fails
        raise RuntimeError("warehouse allocation failed")

@flow
def compensation_surface_example():
    try:
        create_order()
    except Exception:
        refund_payment("pay-001")
        raise
Why this helps: deep retries on mutating steps enlarge the compensation set. Prefer: - very small retry budgets on mutating tasks - fail fast - compensate explicitly Docs: - Transactions --- # 8) Backpressure & Throughput Collapse
Throttle retries globally and fail fast when the system is degraded.
Copy code
python
from prefect import flow, task
from prefect.concurrency.sync import concurrency
from prefect.concurrency.sync import AcquireConcurrencySlotTimeoutError

def retry_unless_backpressure(task, task_run, state):
    try:
        state.result()
        return False
    except Exception as exc:
        if isinstance(exc, AcquireConcurrencySlotTimeoutError):
            return False
        if isinstance(exc, (ConnectionError, TimeoutError)):
            return True
        return False

@task(
    retries=3,
    retry_delay_seconds=[5, 15, 30],
    retry_jitter_factor=0.3,
    retry_condition_fn=retry_unless_backpressure,
)
def overloaded_dependency_call():
    with concurrency("fragile-system", occupy=1, timeout_seconds=2):
        raise ConnectionError("temporary outage")

@flow
def backpressure_example():
    for _ in range(100):
        overloaded_dependency_call.submit()
Why this helps: - concurrency limit protects the fragile dependency - timeout on acquiring slots prevents infinite waiting - retry policy stops when the issue is local saturation/backpressure Docs: - Global concurrency limits --- # A compact “all 8” cheat sheet | Failure mode | Prefect pattern | |---|---| | Misclassification | 
retry_condition_fn
to separate transient vs permanent failures | | Retry storms | exponential backoff + 
retry_jitter_factor
+ concurrency limits | | Side-effect amplification | 
cache_key_fn
/ task caching + downstream idempotency keys | | Temporal drift | retry condition checks freshness / deadlines; 
timeout_seconds
| | Semantic drift | checkpoint upstream outputs with caching and replay them | | Unbounded duration | bound 
retries
, 
retry_delay_seconds
, task and flow 
timeout_seconds
| | Compensation surface expansion | low retries on mutating tasks + 
transaction()
+ explicit compensation | | Backpressure collapse | global concurrency limits + fail-fast retry classification | --- # Suggested production defaults For state-mutating distributed steps, a good starting posture in Prefect is:
Copy code
python
@task(
    retries=2,
    retry_delay_seconds=[5, 20],
    retry_jitter_factor=0.3,
    timeout_seconds=30,
    retry_condition_fn=retry_only_transient,
)
def mutate(...):
    ...
And then add: - idempotency keys for every external mutation - global concurrency limits for fragile systems - cached checkpoints for nondeterministic reads - automations for terminal failures --- If you want, I can also turn this into one of these: 1. a single runnable Prefect demo flow containing all 8 modes 2. a blog-style markdown doc you can paste into internal docs 3. a side-by-side “bad pattern vs good Prefect pattern” table
4 Views
Open in Slack
PreviousNext
Powered by