Day 53: State Management · Applied Decision Intelligence & Data Architecture

Written lesson

Time	Block
0:00 – 0:23	53.1 The state model
0:23 – 0:45	53.2 Inspecting states without crashing: `return_state=True`
0:45 – 0:53	Break
0:53 – 1:15	53.3 Custom states: Completed with warnings
1:15 – 1:40	53.4 A written state policy, implemented in code
1:40 – 1:48	Break
1:48 – 2:03	53.5 State hooks: alerts on failure
2:03 – 2:18	53.6 Artifacts: a run report economists can read
2:18 – 2:30	53.7 Capstone: `daily_refresh` v2 and failure drills

53.1 The State Model (23 min)

Every flow run and task run moves through states. The state type says where it is; the state name can add detail; the message explains why.

Scheduled ──► Pending ──► Running ──┬──► Completed      (success; name may be "Cached", or custom)
                                    ├──► Failed         (the code raised an error, or returned a Failed state)
                                    ├──► Crashed        (the process died: killed, out of memory, infrastructure)
                                    └──► Cancelled      (someone or something stopped it)
                          (Retrying is a Running state named "Retrying": Days 54–57)

Type	Meaning	Our pipeline example
Completed	finished as intended	dbt build succeeded
Failed	finished, but the work didn't succeed	GX critical checkpoint failed
Crashed	the run couldn't finish at all	Codespace restarted mid-run
Cancelled	stopped deliberately	you cancelled a backfill from the UI

How a flow's final state is decided

A flow's final state follows from what happens inside it:

If the flow raises an exception (including an unhandled task failure) → Failed.
If the flow returns a state object → that state.
Otherwise → Completed.

Point 2 is the key to today: you can choose the final state explicitly, based on your own rules.

Try it 53.1

In cpi_briefing.py, change the vector ID to a non-existent one (e.g. 999999999) and run the flow. Which task fails, what's its state message, and what's the flow's final state? Why didn't compute_measures run at all? Restore the ID.

Answer

fetch_cpi fails with the RuntimeError about StatCan's status (Day 28: HTTP 200 but FAILED inside). Its exception propagates into the flow, so the flow is Failed too. compute_measures never runs because its input never existed. Prefect shows no run for it, rather than a failed one, which correctly signals "not attempted".

53.2 Inspecting States Without Crashing: `return_state=True` (23 min)

By default, calling a failed task raises inside the flow, which ends the flow. Often you want to look at the outcome and decide. Pass return_state=True:

from prefect import flow, task


@task
def fragile(x: int) -> int:
    if x < 0:
        raise ValueError("negative input")
    return x * 2


@flow(log_prints=True)
def state_demo():
    ok = fragile(2, return_state=True)
    bad = fragile(-1, return_state=True)

    for s in (ok, bad):
        print(f"type={s.type.value:<10} name={s.name:<10} failed={s.is_failed()}")

    print("ok result:", ok.result())
    print("bad result:", bad.result(raise_on_failure=False))   # the exception object, not raised
    print("bad message:", bad.message)


if __name__ == "__main__":
    state_demo()

State method / attribute	Returns
`s.is_completed()`, `s.is_failed()`, `s.is_crashed()`	booleans
`s.type`	`COMPLETED`, `FAILED`, …
`s.name`	display name (`"Completed"`, `"Failed"`, or custom)
`s.message`	human-readable explanation
`s.result()`	the task's return value (raises if failed)
`s.result(raise_on_failure=False)`	the value, or the exception if failed

The flow above ends Completed: it handled the failure itself. Whether that's right depends on your policy, which we'll write explicitly in 53.4.

53.3 Custom States: Completed with Warnings (23 min)

Our tools have a middle outcome that plain Completed/Failed can't express: exit code 2, meaning "done, but look at this" (partial ingestion, quarantined release, GX warnings). Prefect lets a task or flow return a state with a custom name and message:

from prefect.states import Completed, Failed

Completed(message="all checks passed")
Completed(name="CompletedWithWarnings", message="GX warnings: L32 revisions > 0.3pp")
Failed(message="critical GX failure: L01 rate out of range")

The type remains COMPLETED or FAILED, so everything that depends on success still works correctly. The name shows up in the UI, so a warning run is visibly different from a clean one.

A step task that interprets its exit code

Add to orchestration/tasks/commands.py:

from prefect.states import Completed, Failed

WARNING = "CompletedWithWarnings"


@task(tags=["pipeline-step"])
def run_step_with_state(name: str, cmd: list[str], cwd: str | None = None,
                        warning_codes: tuple[int, ...] = (2,)):
    """Like run_step, but returns a State that encodes our exit-code convention."""
    result = run_step.fn(name, cmd, cwd)            # reuse the logic without a nested task run
    code, dur = result["exit_code"], result["duration_s"]
    if code == 0:
        return Completed(message=f"{name} ok ({dur}s)", data=result)
    if code in warning_codes:
        return Completed(name=WARNING, message=f"{name} finished with warnings ({dur}s)",
                         data=result)
    return Failed(message=f"{name} failed with exit code {code} ({dur}s)", data=result)

run_step.fn(...) calls the original function behind a task decorator, without creating another task run. It's useful for reusing a task's logic inside another task.
data=result attaches the step's dictionary to the state, so state.result() still returns it (for a Failed state, use raise_on_failure=False).

53.4 A Written State Policy, Implemented in Code (25 min)

Before coding, write the policy down. It's a decision about data risk, and should be reviewable by non-engineers, just like the quality plan (Day 41).

`orchestration/STATE_POLICY.md`

# Daily refresh — state policy

| Step | Completed | CompletedWithWarnings | Failed | Downstream if not Completed |
|------|-----------|-----------------------|--------|-----------------------------|
| ingest_api | ok | some series failed (exit 2) | nothing loaded (exit 1) | CONTINUE: warehouse keeps last good data |
| wap_statcan_tables | ok | warnings or some table quarantined (exit 2) | all changed tables quarantined (exit 1) | CONTINUE: only published data is used |
| dbt_build | ok | — | models or tests failed | STOP transform; SKIP quality; flow FAILED |
| gx_warehouse | ok | warnings (exit 2) | critical failure (exit 1) | flow FAILED: marts must not be used |

## Final flow state
- FAILED if: dbt_build failed, gx_warehouse critical, or ingestion failed completely
  (dashboards keep yesterday's consistent data, but a human must act).
- COMPLETED WITH WARNINGS if any step ended with warnings.
- COMPLETED otherwise.

Why continue after an ingestion failure? Thanks to incremental loading (Day 27) and WAP (Day 50), the warehouse is consistent even when today's ingestion fails. Rebuilding dbt keeps marts coherent and lets GX report staleness. But the run must still end Failed, so someone investigates.

The flow, following the policy

Replace orchestration/flows/daily_refresh.py:

"""Daily macro refresh — v2: explicit state policy (Day 53)."""
from prefect import flow, get_run_logger
from prefect.states import Completed, Failed

from orchestration.settings import DBT, DBT_DIR, VENV_PY
from orchestration.tasks.commands import WARNING, run_step_with_state


def outcome(state) -> str:
    """Collapse a state into 'ok' | 'warning' | 'failed'."""
    if state.is_failed() or state.is_crashed():
        return "failed"
    return "warning" if state.name == WARNING else "ok"


@flow(name="ingest")
def ingest_flow() -> dict:
    return {
        "ingest_api": run_step_with_state(
            "ingest_api", [VENV_PY, "ingest.py", "--only", "boc,fred,statcan"],
            return_state=True),
        "wap_statcan_tables": run_step_with_state(
            "wap_statcan_tables", [VENV_PY, "ingest_statcan_tables.py"], return_state=True),
    }


@flow(name="transform")
def transform_flow() -> dict:
    states = {"dbt_deps": run_step_with_state("dbt_deps", [DBT, "deps", "--quiet"],
                                              cwd=str(DBT_DIR), return_state=True)}
    if outcome(states["dbt_deps"]) == "failed":
        return states
    states["dbt_build"] = run_step_with_state("dbt_build", [DBT, "build", "--target", "prod"],
                                              cwd=str(DBT_DIR), warning_codes=(),
                                              return_state=True)
    if outcome(states["dbt_build"]) != "failed":
        states["dbt_docs"] = run_step_with_state("dbt_docs",
                                                 [DBT, "docs", "generate", "--target", "prod"],
                                                 cwd=str(DBT_DIR), return_state=True)
    return states


@flow(name="quality")
def quality_flow() -> dict:
    return {"gx_warehouse": run_step_with_state(
        "gx_warehouse", ["bash", "quality/run_quality.sh"], return_state=True)}


@flow(name="daily-refresh")
def daily_refresh():
    logger = get_run_logger()
    states = {}

    states.update(ingest_flow())                        # CONTINUE regardless (policy)
    states.update(transform_flow())

    if outcome(states.get("dbt_build", states["dbt_deps"])) == "failed":
        logger.error("dbt failed — skipping warehouse quality checks (policy)")
    else:
        states.update(quality_flow())

    summary = {name: outcome(s) for name, s in states.items()}
    for name, o in summary.items():
        logger.info(f"{name:<22} {o}")

    fatal = [n for n, o in summary.items() if o == "failed"
             and n in ("dbt_deps", "dbt_build", "gx_warehouse")]
    ingestion_dead = summary.get("ingest_api") == "failed" and \
                     summary.get("wap_statcan_tables") == "failed"
    warnings = [n for n, o in summary.items() if o != "ok"]

    if fatal or ingestion_dead:
        return Failed(message=f"Policy: FAILED — {fatal or ['all ingestion']}. See STATE_POLICY.md",
                      data=summary)
    if warnings:
        return Completed(name=WARNING, message=f"Warnings in: {warnings}", data=summary)
    return Completed(message="All steps completed cleanly", data=summary)


if __name__ == "__main__":
    daily_refresh()

Compare with pipeline.sh: every || [ $? -eq 2 ] and set +e is replaced by a named, documented decision in Python. The flow's final state and message tell anyone opening the UI exactly what happened and why.

Why do subflows return dictionaries of states instead of raising? Because the parent flow owns the policy. Subflows report; the parent decides. That separation keeps each stage reusable, so quality_flow() can run alone without dragging along the parent's rules.

53.5 State Hooks: Alerts on Failure (15 min)

A state hook is a function Prefect calls when a run enters a given state. It's the natural place for notifications, and it replaces the alert calls scattered through pipeline.sh.

from macro_ingest.resilience import notify          # Day 29's single alert channel


def alert_on_failure(flow, flow_run, state):
    notify(f"⛔ {flow.name} run '{flow_run.name}' {state.name}: {state.message}")


def alert_on_warnings(flow, flow_run, state):
    if state.name == "CompletedWithWarnings":
        notify(f"⚠ {flow.name} run '{flow_run.name}': {state.message}")


@flow(name="daily-refresh",
      on_failure=[alert_on_failure],
      on_crashed=[alert_on_failure],
      on_completion=[alert_on_warnings])
def daily_refresh():
    ...

Hook	Called when the run…
`on_completion`	reaches any Completed state (including custom names)
`on_failure`	reaches Failed
`on_crashed`	crashes (process killed, infrastructure failure)
`on_cancellation`	is cancelled

Tasks accept hooks too (@task(on_failure=[...]), with the signature (task, task_run, state)), but for our pipeline one flow-level alert that summarises everything beats one alert per step. That's the same alert-fatigue lesson as Day 29.

Hooks must never crash the flow themselves. notify() (Day 29) already catches its own errors, so a broken webhook can't turn a warning into a failure.

53.6 Artifacts: A Run Report Economists Can Read (15 min)

Artifacts are rich outputs attached to a run and shown in the UI: tables, Markdown reports and links. They turn a run from "green or red" into evidence.

Create orchestration/tasks/reporting.py:

"""Artifacts: a step table and a Markdown briefing for each refresh."""
import duckdb
from prefect import task
from prefect.artifacts import create_markdown_artifact, create_table_artifact

from orchestration.settings import ENV


@task
def publish_step_table(summary: dict, durations: dict):
    rows = [{"step": s, "outcome": o, "duration_s": durations.get(s)}
            for s, o in summary.items()]
    create_table_artifact(key="daily-refresh-steps", table=rows,
                          description="Outcome and duration of each pipeline step")


@task
def publish_briefing():
    """Latest KPIs (Day 39) and data-quality status (Day 50) as a Markdown report."""
    with duckdb.connect(ENV["MACRO_DUCKDB_PATH"], read_only=True) as con:
        kpis = con.sql("""select pillar, kpi, round(value, 2) as value, as_of, signal
                          from marts.macro_kpis order by pillar, kpi""").fetchall()
        quality = con.sql("""select checkpoint, pct_passing, failures
                             from marts.quality_latest order by checkpoint""").fetchall()

    lines = ["# Canadian macro briefing", "", "| Pillar | KPI | Value | As of | Signal |",
             "|---|---|---|---|---|"]
    lines += [f"| {p} | {k} | {v} | {a} | {s} |" for p, k, v, a, s in kpis]
    lines += ["", "## Data quality (latest GX runs)", "", "| Checkpoint | % passing | Failures |",
              "|---|---|---|"]
    lines += [f"| {c} | {pct} | {f} |" for c, pct, f in quality]
    create_markdown_artifact(key="macro-briefing", markdown="\n".join(lines),
                             description="KPIs and data quality after this refresh")

Call both from the end of daily_refresh (before returning the final state), when the transform succeeded:

    def step_data(state):
        data = state.result(raise_on_failure=False)
        return data if isinstance(data, dict) else {}

    durations = {n: step_data(s).get("duration_s") for n, s in states.items()}
    publish_step_table(summary, durations)
    if summary.get("dbt_build") != "failed":
        publish_briefing()

In the UI, open Artifacts (or the run's Artifacts tab): the step table, and a formatted briefing with every KPI, its date and its signal, plus the quality pass rates. Artifacts with the same key form a history, so you can page back through yesterday's briefing.

The briefing opens the warehouse read-only and runs after all writers have finished, so it never competes for the DuckDB lock.

53.7 Capstone: `daily_refresh` v2 and Failure Drills (13 min)

Assemble the pieces: state policy (53.4), hooks (53.5) and artifacts (53.6). Then prove the policy works without breaking real systems, by adding a drill parameter that overrides exit codes.

A drill parameter

In run_step_with_state, accept an optional force_exit: int | None = None. If set, skip the real command and pretend it exited with that code:

    if force_exit is not None:
        get_run_logger().warning(f"DRILL: {name} forced to exit {force_exit}")
        result = {"step": name, "exit_code": force_exit, "duration_s": 0.0}
    else:
        result = run_step.fn(name, cmd, cwd)

and add a flow parameter drill: dict[str, int] | None = None, passing force_exit=(drill or {}).get(step_name) to each step.

Run the drills

from orchestration.flows.daily_refresh import daily_refresh

daily_refresh(drill={"ingest_api": 2})                  # expect CompletedWithWarnings, warning alert
daily_refresh(drill={"wap_statcan_tables": 1})          # expect CompletedWithWarnings (one ingestion path failed)
daily_refresh(drill={"ingest_api": 1, "wap_statcan_tables": 1})   # expect Failed (all ingestion)
daily_refresh(drill={"dbt_build": 1})                   # expect Failed; quality skipped
daily_refresh(drill={"gx_warehouse": 1})                # expect Failed (critical quality)
daily_refresh(drill={"gx_warehouse": 2})                # expect CompletedWithWarnings

Record the results in orchestration/DRILL_RESULTS.md:

Drill	Expected final state	Observed	Alert sent?	Downstream behaviour correct?
ingest_api = 2	CompletedWithWarnings
…

A policy that hasn't been drilled is a hope, the same lesson as Day 50's fault injection, now at the orchestration level.

Recap: Days 51–53

Concept	You built
flows & tasks	tracked, logged, observable steps with typed parameters
dynamic dependencies	data dependencies and `wait_for` for side effects
concurrency	`submit`/`map` with a thread-pool runner, and the DuckDB single-writer constraint
subflows	ingest / transform / quality stages, runnable alone
wrapping tools	streamed subprocess tasks for dlt, WAP, dbt and GX
states	`return_state=True`, custom CompletedWithWarnings, explicit final states
policy	`STATE_POLICY.md`, implemented and drilled
observability	hooks for alerts; artifacts for step tables and an economist-readable briefing

Looking ahead: Days 54–57

Retries with backoff per task (a flaky FRED call retries alone, not the whole refresh).
Caching, so re-running a flow skips steps whose inputs haven't changed.
Scheduling and deployments: run daily-refresh automatically, with parameters, from a worker.
Logging in depth, and concurrency limits that enforce "one DuckDB writer at a time".

Day 53 checklist

I can describe the state lifecycle and how a flow's final state is determined
I inspect task outcomes with return_state=True and handle failures deliberately
Exit codes map to Completed, CompletedWithWarnings and Failed states
A written state policy is implemented in the parent flow; subflows report, the parent decides
Hooks send one summarising alert per run; artifacts publish a step table and a macro briefing
Drills prove every row of the state policy behaves as documented

git add orchestration && git commit -m "Day 53: state policy, custom states, hooks, artifacts, drills" && git push

State Management

What you'll learn today

Today's plan