Day 54: Retries and Timeouts · Applied Decision Intelligence & Data Architecture

Written lesson

Time	Block
0:00 – 0:20	54.1 Where should retries live?
0:20 – 0:45	54.2 Task retries: counts, delays, backoff, jitter
0:45 – 0:53	Break
0:53 – 1:20	54.3 Retrying only what time can fix
1:20 – 1:40	54.4 Exit codes are an API: fixing an ambiguity
1:40 – 1:48	Break
1:48 – 2:05	54.5 Timeouts and retry budgets
2:05 – 2:30	54.6 Mini-project: a per-step retry policy, drilled

54.1 Where Should Retries Live? (20 min)

You already retry at one level: dlt's HTTP helper retries individual requests (Day 29). Adding Prefect retries creates layers, and layers multiply:

HTTP request retries (dlt, Day 29)     × 5 attempts per request
   inside a task (Prefect retries)     × 3 attempts per step
      inside a flow (flow retries)     × 2 attempts per run
                                       = up to 30 attempts at one failing request

Level	Retries what	Good for	Our choice
HTTP request	one API call	brief blips, 429/5xx	keep (Day 29)
Task (step)	a whole pipeline step	an outage lasting minutes (provider maintenance)	add today, for transient failures only
Flow	the entire run	infrastructure hiccups (rare)	usually no: too coarse and expensive

Principle: each layer should cover a different timescale. HTTP retries handle seconds; task retries handle minutes; the schedule (Day 56) handles hours, since tomorrow's run is the ultimate retry.

54.2 Task Retries: Counts, Delays, Backoff, Jitter (25 min)

Create orchestration/flows/retry_demo.py:

import random

from prefect import flow, get_run_logger, task
from prefect.runtime import task_run
from prefect.tasks import exponential_backoff


@task(retries=3,
      retry_delay_seconds=exponential_backoff(backoff_factor=2),   # 2s, 4s, 8s
      retry_jitter_factor=0.5)                                      # ±50% randomness
def flaky_fetch(p_fail: float = 0.6) -> str:
    logger = get_run_logger()
    attempt = task_run.run_count                   # 1 on the first try, 2 on the first retry, …
    logger.info(f"attempt {attempt}")
    if random.random() < p_fail:
        raise ConnectionError(f"simulated outage on attempt {attempt}")
    return f"succeeded on attempt {attempt}"


@flow(log_prints=True)
def retry_demo():
    print(flaky_fetch())


if __name__ == "__main__":
    random.seed(7)                                  # reproducible demo
    retry_demo()

python -m orchestration.flows.retry_demo

In the UI, the task run shows its state history: Running → Retrying → Running → … → Completed (or Failed if all attempts failed). Each attempt's logs are kept.

Option	Meaning
`retries=3`	up to 3 additional attempts (4 in total)
`retry_delay_seconds=10`	wait 10 s between attempts
`retry_delay_seconds=[10, 60, 300]`	a different wait before each retry
`exponential_backoff(backoff_factor=b)`	waits of b, 2b, 4b, … seconds (Day 9's idea)
`retry_jitter_factor=0.5`	randomise each wait by up to ±50%
`prefect.runtime.task_run.run_count`	which attempt this is (1, 2, …)

Why jitter? If many runs fail at the same moment (a provider outage at 08:30 ET), identical backoff would make them all retry at the same instant too, hammering the provider when it recovers. Randomness spreads them out.

Changing retries per call

A task's defaults can be overridden for one call, without redefining it:

flaky_fetch.with_options(retries=0)(p_fail=0.6)                     # no retries this time
flaky_fetch.with_options(retries=5, retry_delay_seconds=1)(0.9)

We'll use with_options to give each pipeline step its own retry policy.

54.3 Retrying Only What Time Can Fix (28 min)

Day 29's rule, at the orchestration level: retry transient failures; never retry deterministic ones. A dbt model with a SQL error will fail identically on every attempt, and retrying only delays the alert.

Step failure	Transient?	Retry?
API ingestion: nothing loaded (provider down)	usually yes	✓ after a few minutes
StatCan download interrupted	yes	✓
StatCan release quarantined by GX (Day 50)	no: the data is bad	✗ never
`dbt build` failed (SQL error, failing test)	no	✗
GX critical failure	no	✗
process crashed (out of memory, Codespace hiccup)	often	✓ once

Prefect decides whether to retry with a retry_condition_fn: a function that receives the failed state and returns True to retry.

A transient-error type and a retry condition

Add to orchestration/tasks/commands.py:

class TransientStepError(RuntimeError):
    """A step failed in a way that waiting might fix (outage, interrupted download)."""


def retry_if_transient(task, task_run, state) -> bool:
    """Retry only when the task failed with TransientStepError."""
    try:
        state.result()
    except TransientStepError:
        return True
    except Exception:
        return False
    return False

state.result() re-raises the exception that failed the task, so the except clauses inspect its type. Only TransientStepError triggers a retry.

A step task that raises for transient codes

from prefect.runtime import task_run as current_task_run
from prefect.states import Completed, Failed


@task(tags=["pipeline-step"], retry_condition_fn=retry_if_transient)
def run_policy_step(name: str, cmd: list[str], cwd: str | None = None,
                    warning_codes: tuple[int, ...] = (2,),
                    transient_codes: tuple[int, ...] = (),
                    drill: dict | None = None):
    """Run a step; raise TransientStepError (→ retried) or return a policy State."""
    logger = get_run_logger()
    attempt = current_task_run.run_count

    forced = (drill or {}).get(name)
    if forced is not None:                                  # drills (Day 53), now attempt-aware
        code = forced["code"] if attempt <= forced.get("attempts", 99) else 0
        logger.warning(f"DRILL: {name} attempt {attempt} forced to exit {code}")
        result = {"step": name, "exit_code": code, "duration_s": 0.0}
    else:
        result = run_step.fn(name, cmd, cwd)

    code = result["exit_code"]
    if code in transient_codes:
        raise TransientStepError(f"{name}: transient exit {code} on attempt {attempt}")
    if code == 0:
        return Completed(message=f"{name} ok", data=result)
    if code in warning_codes:
        return Completed(name=WARNING, message=f"{name} finished with warnings", data=result)
    return Failed(message=f"{name} failed with exit code {code}", data=result)

The drill format is now {"ingest_api": {"code": 1, "attempts": 1}}, meaning "exit 1 on the first attempt only", so you can watch a retry succeed.

54.4 Exit Codes Are an API: Fixing an Ambiguity (20 min)

Look closely at ingest_statcan_tables.py (Day 50):

Exit code	Current meaning
0	everything published
2	published with warnings, or some tables quarantined
1	all changed tables quarantined

But Python also exits with 1 when a script crashes with an unhandled exception, such as an interrupted download. So 1 means both "data rejected by quality checks" (never retry) and "the script crashed" (retry). The orchestrator can't tell them apart.

Exit codes are an interface between programs. Like a function's return type, they deserve a documented design. Ambiguity here directly causes wrong operational decisions.

The fix

Give quarantine its own code. In ingest_statcan_tables.py, change the end of main():

    if quarantined and not published:
        return 3          # 3 = all changed tables quarantined (data rejected; not transient)
    if quarantined or warned:
        return 2
    return 0

Document the convention for every script in orchestration/EXIT_CODES.md:

| Script | 0 | 1 | 2 | 3 |
|---|---|---|---|---|
| ingest.py | ok | nothing loaded / crash (transient) | partial | — |
| ingest_statcan_tables.py | ok | crash, e.g. download failure (transient) | warnings or partial quarantine | all quarantined (NOT transient) |
| dbt build | ok | models or tests failed | — | — |
| quality/run_quality.sh | ok | critical failure | warnings | — |

And update orchestration/STATE_POLICY.md (Day 53) so that WAP code 3 counts as failed for the policy but is never retried.

54.5 Timeouts and Retry Budgets (18 min)

Timeouts

A step that hangs (a stalled download, a locked database) is worse than one that fails, because nothing alerts until someone notices. Give every step a timeout:

run_policy_step.with_options(timeout_seconds=1800)(...)     # fail after 30 minutes

A timed-out task enters a Failed state named TimedOut. Whether it's retried depends on the retry condition. Our condition only retries TransientStepError, so to retry timeouts too, extend it:

def retry_if_transient(task, task_run, state) -> bool:
    if state.name == "TimedOut":
        return True
    try:
        state.result()
    except TransientStepError:
        return True
    except Exception:
        return False
    return False

(Timeouts stop the task from waiting. Make sure the underlying subprocess also stops, e.g. by giving ingest_statcan_tables.py its own download timeouts, as Day 30's code already does.)

Retry budgets

Before choosing numbers, compute the worst case: how long could a run take if everything fails slowly?

Step	Timeout	Retries	Delays	Worst case
ingest_api	30 min	2	5 + 10 min	3 × 30 + 15 = 105 min
wap_statcan_tables	60 min	1	15 min	2 × 60 + 15 = 135 min
dbt_build	30 min	0	—	30 min
gx_warehouse	20 min	0	—	20 min
Total				~4.8 hours

If the refresh runs at 17:30 ET and the dashboard must be ready by 08:00 the next morning, a ~5-hour worst case is acceptable. If it had to be ready within an hour, you'd need shorter timeouts or fewer retries. Retry settings are a business decision, derived from the deadline.

54.6 Mini-Project: A Per-Step Retry Policy, Drilled (25 min)

Centralise the policy

Add to orchestration/settings.py:

STEP_POLICIES = {
    #  name                  retries  delays (s)      timeout (s)  transient codes  warning codes
    "ingest_api":         dict(retries=2, delays=[300, 600], timeout=1800,
                               transient=(1,), warning=(2,)),
    "wap_statcan_tables": dict(retries=1, delays=[900],      timeout=3600,
                               transient=(1,), warning=(2,)),
    "dbt_deps":           dict(retries=1, delays=[60],       timeout=600,
                               transient=(1,), warning=()),
    "dbt_build":          dict(retries=0, delays=[],         timeout=1800,
                               transient=(),   warning=()),
    "dbt_docs":           dict(retries=0, delays=[],         timeout=600,
                               transient=(),   warning=()),
    "gx_warehouse":       dict(retries=0, delays=[],         timeout=1200,
                               transient=(),   warning=(2,)),
}

(dbt deps downloads packages from the internet, so a failure there is usually transient.)

A helper that applies the policy

In orchestration/tasks/commands.py:

from orchestration.settings import STEP_POLICIES


def step(name: str, cmd: list[str], cwd: str | None = None, drill: dict | None = None):
    """Run a pipeline step with its policy; always return a State (never raises)."""
    p = STEP_POLICIES[name]
    configured = run_policy_step.with_options(
        name=f"step:{name}",
        retries=p["retries"],
        retry_delay_seconds=p["delays"] or 0,
        retry_jitter_factor=0.2 if p["retries"] else None,
        timeout_seconds=p["timeout"],
    )
    return configured(name, cmd, cwd,
                      warning_codes=p["warning"], transient_codes=p["transient"],
                      drill=drill, return_state=True)

Update daily_refresh.py so every step goes through step(...) (replacing run_step_with_state), and pass the flow's drill parameter down to the subflows.

Drills

For quick drills, temporarily set the delays in STEP_POLICIES to a few seconds, then:

daily_refresh(drill={"ingest_api": {"code": 1, "attempts": 1}})      # retried once → recovers → Completed
daily_refresh(drill={"ingest_api": {"code": 1, "attempts": 99}})     # exhausts retries → ingestion failed
daily_refresh(drill={"wap_statcan_tables": {"code": 3, "attempts": 1}})  # quarantine: NOT retried
daily_refresh(drill={"dbt_build": {"code": 1, "attempts": 1}})       # NOT retried; flow Failed

For each drill, confirm in the UI: the number of attempts, the delays between them, the final task state, and the flow's final state according to STATE_POLICY.md. Record the results in orchestration/DRILL_RESULTS.md and restore the real delays afterwards.

Day 54 checklist

I can explain retry layers (HTTP, task, flow, schedule) and avoid multiplying them
I configure retries with backoff and jitter, and can see attempts in the UI
A retry_condition_fn retries only transient failures; quality rejections and SQL errors are never retried
Exit codes are unambiguous and documented (quarantine = 3)
Every step has a timeout; the retry budget fits the business deadline
Retry behaviour is centralised in STEP_POLICIES and proven by drills

git add . && git commit -m "Day 54: retries, retry conditions, timeouts, exit-code API" && git push

Retries and Timeouts

What you'll learn today

Today's plan