States and State Hooks

The full Prefect state model, Failed versus Crashed, return_state, allow_failure, result handling, and lifecycle hooks.

Every chapter so far has leaned on the word “state” and trusted you to fill in the meaning. A retry happens because a task run entered AwaitingRetry. A cache hit shows up as Cached. A deployment that missed its window is Late. Automations fire on state changes. We’ve been spending this vocabulary without ever laying it out, because states are the substrate everything else in Prefect sits on — and Airflow’s much smaller state set is one of the places the two tools diverge hardest. This is the chapter that stops hand-waving.

In Airflow a task instance is success, failed, skipped, up_for_retry, upstream_failed, queued, running, or a handful of others, and the set is essentially fixed enum values the scheduler writes to a column. Prefect’s states are richer, they carry data, they carry a message, and they’re the thing your code can hold in its hand and interrogate. Treating them as mere UI labels is the fastest way to write brittle Prefect — so let’s treat them as what they are: the contract between your code, the API, and every reaction the platform can have to a run.

The full taxonomy

A Prefect state has a type (one of a fixed set) and often a more specific name layered on top. The types you’ll actually reason about:

  • Scheduled — the run exists and has a time it’s meant to start. Nothing is executing yet. Late and AwaitingRetry are both names under the Scheduled type, which is a detail that surprises people: a run waiting to retry is, mechanically, just a run scheduled for a moment in the future.
  • Late — a scheduled run whose start time has passed by more than a small grace period and no worker has picked it up. This is the state you set alerts on when a work pool has no healthy workers.
  • Pending — the run has been claimed and is preparing to execute; the gap between “scheduled” and “actually running.”
  • Running — code is executing right now.
  • Retrying — a specific Running-type name for the execution attempt that follows a retry delay. AwaitingRetry is the wait; Retrying is the re-run itself.
  • Paused — the flow run hit a pause and is waiting, holding its place in memory or in the process. Used for interactive input and manual gates.
  • Suspended — like paused, but the run exits — the process is torn down and the run will be rehydrated later when resumed. The difference from Paused matters for anything long-lived: a suspended run isn’t holding an infrastructure slot.
  • Cancelling — cancellation was requested and Prefect is tearing down the run’s infrastructure.
  • Cancelled — that teardown finished; the run was deliberately stopped.
  • Completed — terminal success. The unit did its work and returned.
  • Cached — terminal, and also a success, but the result came from a prior run’s cache rather than fresh execution (the caching chapter is where this one is earned).
  • Failed — terminal failure caused by your code — an exception propagated out of the task or flow.
  • Crashed — terminal failure caused by the environment, not the code. The worker died, the pod was OOM-killed, the process got a SIGKILL. This is the state Airflow has no honest word for, and it gets its own section below.

The single most useful way to group these is terminal versus non-terminal. A terminal state is the end of the line: Completed, Cached, Failed, Crashed, Cancelled. Once a run enters one, it doesn’t leave on its own — the only way past a terminal state is a new run or an explicit retry that creates a fresh attempt. Everything else — Scheduled, Late, Pending, Running, Retrying, AwaitingRetry, Paused, Suspended, Cancelling — is non-terminal, a stage the run is passing through. Automations, alerting, and your own return_state inspection all care intensely about which side of that line a state falls on, because “is this run done, and how did it end?” is the question most orchestration logic is really asking.

There’s a second useful axis: is_final (terminal) versus is_completed/is_failed/is_crashed/is_cancelled, which slice the terminal states by outcome. Cached reports is_completed() == True — a cache hit is a success, and code that checks is_completed() shouldn’t have to special-case it. That’s deliberate: the outcome predicates are about what the run means, not how it got there.

The state lifecycle, in prose

Trace one task run through its life. It’s created Scheduled, with a start time. A worker (or the flow’s task runner) picks it up; it moves to Pending while inputs resolve and infrastructure readies, then to Running when your function body actually begins. From Running there are three doors. The clean one: the function returns, and the run goes Completed (or Cached, if a cache key matched before the body even ran). The code-failure door: the function raises, and if retries remain the run goes AwaitingRetry (a Scheduled name) — it waits out retry_delay_seconds, re-enters as Retrying, and loops back through Running. When retries are exhausted, the door closes on Failed. The third door is the one nobody chooses: the process disappears out from under the run, and Prefect marks it Crashed.

Flow runs walk the same path with a couple of extra rooms. A flow run can enter Paused or Suspended when it hits an interactive pause, sitting non-terminal until something resumes it. It can be asked to stop, moving Cancelling → Cancelled. And a flow run’s terminal state is normally derived from its task runs: if any task run finished Failed or Crashed and nothing caught it, the flow run ends Failed; if all the ones that matter finished Completed/Cached, the flow run is Completed. That derivation is why “the flow failed” and “a task in the flow failed” are usually the same sentence — but not always, which is exactly what allow_failure and return_state let you pry apart.

Failed versus Crashed: the distinction Airflow won’t draw

Here is the difference worth internalizing. Failed means your code raised. A SQL statement errored, an API returned 500, an assertion tripped, a ValueError you threw yourself. The failure is inside the workflow’s logic, and it is reproducible — run the same code against the same inputs and you’ll likely fail the same way. Crashed means the code never got to finish because the ground gave way underneath it. The Kubernetes node was preempted. The container hit its memory limit and the kernel killed it. The worker lost its network connection to the API and the heartbeat stopped. The failure is outside the workflow’s logic, and it is often transient — the same code on a healthy machine would have sailed through.

Airflow collapses both of these into failed. A task that raised a genuine bug and a task whose pod got OOM-killed land in the same red box, and the DAG author has to go spelunking in logs to tell them apart — assuming the logs even survived the crash, which after an OOM kill they frequently don’t. That collapse has real costs:

  • Alerting. A Crashed run usually means “look at your infrastructure,” not “look at your code.” Paging the on-call data engineer for an OOM kill when the fix is a bigger work pool is noise; paging the platform team for a genuine Failed assertion is misrouted. When the states are distinct you can route them to different places — and Prefect’s automations let you do exactly that, firing one action on Crashed and another on Failed.
  • Retries. Retrying a Crashed run is almost always correct — transient infra hiccups are the textbook case for “just try again.” Retrying a deterministic Failed run three times just fails three times slower and buries the real error under duplicate alerts. Knowing which you’re dealing with tells you whether a retry is hope or superstition.
  • SLOs and post-mortems. “We had four Failed runs and eleven Crashed runs this week” is two completely different conversations — one about a flaky model, one about undersized infrastructure. A single failed count hides that.

You don’t have to do anything to get this distinction; Prefect assigns Crashed automatically when it detects abnormal termination (a lost heartbeat, a non-zero exit from infrastructure rather than a propagated exception). Your job is to stop treating “failure” as one bucket and start reacting to the two kinds differently — starting with the hook that only fires on one of them.

Inspecting state from code

By default a task or flow call behaves like a normal function: it returns its result, or it raises. That’s the right default and it’s what most code should want. But sometimes you need the state object itself — to branch on it, to pull a failed result out without blowing up, to decide something the orchestrator can’t decide for you. That’s what return_state=True is for.

from prefect import flow, task

@task
def fetch_inventory(store_id: int) -> dict:
    if store_id == 7:
        raise RuntimeError("store 7 feed is down")
    return {"store_id": store_id, "titles": 1200}

@flow(log_prints=True)
def nightly_rollup():
    state = fetch_inventory(7, return_state=True)   # does NOT raise

    if state.is_failed():
        print(f"skipping store 7: {state.message}")
        inventory = {"store_id": 7, "titles": 0}     # fall back
    else:
        inventory = state.result()

    print(f"rolling up {inventory['titles']} titles")

With return_state=True, the call returns a State no matter how the task ended, and the exception stays boxed inside it instead of propagating. Now you can ask the state questions: state.is_completed(), state.is_failed(), state.is_crashed(), state.is_cancelled(), state.is_final(). The .message attribute carries the human-readable reason. And when you’re ready for the actual return value, state.result() unpacks it — but with a knob:

data = state.result(raise_on_failure=False)

state.result() on a failed state re-raises the original exception by default — sensible, because usually you called .result() because you wanted the value and its absence is an error. But raise_on_failure=False says “hand me whatever’s there and let me decide,” returning the exception object rather than throwing it. That’s the controlled-access path: you get to look at the failure as data instead of having control flow ripped out from under you. Use it when a failed upstream is a branch in your logic, not an abort.

This is a genuinely different posture from Airflow. There, a task instance’s outcome isn’t a first-class object your Python can hold; you reach for TriggerRule.ALL_DONE, or an airflow.exceptions.AirflowSkipException, or you poke the metadata DB, to express “continue even though something upstream didn’t succeed.” In Prefect the state is a value, so “look at how the upstream ended and decide” is just an if.

allow_failure in dependencies

There’s a cousin to return_state for the specific case where a downstream task should run even though an upstream one failed, and should receive the failed state as input rather than being cancelled by the propagation. That’s allow_failure:

from prefect import flow, task
from prefect.states import State
from prefect import allow_failure

@task
def load_primary() -> str:
    raise RuntimeError("primary warehouse unreachable")

@task
def load_backup() -> str:
    return "loaded into backup"

@task
def notify(primary_result, backup_result: str) -> None:
    # primary_result is the FAILED state, not a raised exception
    if isinstance(primary_result, State) and primary_result.is_failed():
        print(f"primary failed ({primary_result.message}); backup: {backup_result}")
    else:
        print("both loads succeeded")

@flow(log_prints=True)
def resilient_load():
    primary = load_primary.submit()
    backup = load_backup.submit()
    notify(allow_failure(primary), backup)

Without allow_failure, a downstream task that depends on a failed upstream is itself marked as failed-by-propagation and never runs its body — Prefect’s equivalent of Airflow’s upstream_failed. Wrapping the dependency in allow_failure overrides that: notify runs regardless, and instead of receiving load_primary’s value it receives its failed state, which it can inspect exactly like a return_state result. This is the clean way to express “try the primary, but don’t let its failure abort the notification/cleanup path” — the pattern you’d otherwise contort trigger rules to get in Airflow.

Returning a state to force an outcome

return_state=True lets the caller read a state; the mirror image is letting the callee write one. A task or flow can return a State object straight from its body, and Prefect honors it as the run’s outcome instead of deriving one from whether an exception propagated. It is how you say “this run is a failure, but not because Python raised” — a data-quality gate, or a business rule that vetoes the result.

from prefect import task
from prefect.states import State, Completed, Failed

@task
def validate_feed(rows: list[dict]) -> State:
    missing = [r for r in rows if r.get("price") is None]
    if len(missing) > len(rows) * 0.1:
        # >10% of rows missing price: fail deliberately, with a message
        return Failed(message=f"{len(missing)}/{len(rows)} rows missing a price")
    return Completed(message=f"validated {len(rows)} rows ({len(missing)} warnings)")

Returning Failed(message=...) marks the task Failed without raising — no traceback, just a terminal failure carrying the string you chose, shown in the UI and readable via state.message. Contrast it with raise ValueError("..."), which also yields Failed but wraps the exception as the state’s data, so state.result(raise_on_failure=False) hands back the exception object rather than a message. Reach for a returned State when the failure is a decision your code made — a rule or gate — and for raise when it’s an error your code hit.

State-change hooks

Inspecting state is pull. Hooks are push: functions Prefect calls for you when a run enters a particular state, wired directly onto the @flow or @task decorator. There are five, matched to the transitions worth reacting to:

  • on_running — fires when the run enters Running. Good for “mark this job started” side effects.
  • on_completion — fires on Completed (terminal success).
  • on_failure — fires on Failed (your code raised, retries exhausted).
  • on_crashed — fires on Crashed (infrastructure died). This is the hook that exists because Prefect splits Failed from Crashed; Airflow has no equivalent because it has no such split.
  • on_cancellation — fires on Cancelled.

Every hook has the same signature: it’s called with three positional arguments — the flow (or task) object, the run object, and the state that triggered it.

def hook(flow, flow_run, state):
    ...

For a task hook the first two are the task and the task run; for a flow hook, the flow and the flow run. The state argument is the same rich State you’d get from return_state, so inside a hook you have state.message, state.result(raise_on_failure=False), and the predicates all available.

Here’s a flow that puts three of them to work — a metric on success, a cleanup on failure, and a distinct infrastructure alert on crash:

import time
from prefect import flow, task

def emit_duration_metric(flow, flow_run, state):
    elapsed = (flow_run.end_time - flow_run.start_time).total_seconds()
    print(f"[metric] {flow.name} completed in {elapsed:.1f}s")

def clean_up_staging(flow, flow_run, state):
    # runs only when the flow's code raised — safe to assume staging is dirty
    print(f"[cleanup] dropping staging tables after failure: {state.message}")
    # drop_staging_tables(...)

def page_platform_team(flow, flow_run, state):
    # runs only when infra died — route to a DIFFERENT channel than on_failure
    print(f"[PAGE] {flow.name} crashed (infra): {state.message}")
    # pagerduty.trigger(service="platform", ...)

@flow(
    log_prints=True,
    on_completion=[emit_duration_metric],
    on_failure=[clean_up_staging],
    on_crashed=[page_platform_team],
)
def nightly_etl():
    time.sleep(1)
    raise RuntimeError("dbt run exited non-zero")   # → Failed → clean_up_staging

Run that and clean_up_staging fires, because the body raised — a genuine Failed. If instead the worker had been killed mid-run, page_platform_team would fire and clean_up_staging would not. That’s the payoff of the taxonomy made concrete: two different reactions to two different kinds of failure, declared right on the decorator. Note that each hook argument is a list — you can attach several functions to the same transition, and they run in order.

Task-level hooks are the right place for reactions scoped to a single unit of work — but they are not the same set, and this catches people:

Hook@flow@task
on_completionyesyes
on_failureyesyes
on_runningyesyes
on_crashedyesno
on_cancellationyesno

Pass on_crashed= to a @task and you don’t get a silently-ignored kwarg, you get TypeError: task() got an unexpected keyword argument 'on_crashed' — which is at least honest. The asymmetry is defensible once you think about who observes what: a crash is the runtime dying underneath you, and it’s the flow run — the thing still standing — that is in a position to notice and report. If you need per-task crash handling, put an on_crashed on the flow and inspect which task run is in a non-terminal state.

def alert_on_extract_failure(task, task_run, state):
    print(f"[alert] extract task failed: {state.message}")

@task(retries=2, on_failure=[alert_on_extract_failure])
def extract_orders() -> list[dict]:
    ...

One timing subtlety worth internalizing: on_failure fires when the run reaches its terminal Failed state — after retries are exhausted, not on each failed attempt. A task with retries=2 that fails all three times fires on_failure once, at the end. If you want to react to every attempt, that’s a different mechanism; the state hooks are about the run’s final verdict, which is almost always what you want for alerting and cleanup.

Write hooks to be idempotent. A hook can run more than once — a UI retry re-enters Failed and re-fires on_failure — so a cleanup hook should be safe to run twice: DROP TABLE IF EXISTS, not DROP TABLE. And on_crashed carries a caution people miss: it is best-effort by construction. A hook is Python in the run’s own process, so if the worker was OOM-killed there may be nothing left to invoke it. Prefect fires it when it can, but anything that must happen after a crash — paging, releasing a lease — belongs in a server-side automation, which brings us to that distinction.

Hooks versus automations

Hooks and automations look like they solve the same problem, and choosing wrong leads to grief, so draw the line clearly:

  • Hooks run in-process, inside your run’s own execution. They’re Python functions in your codebase, they execute on the same worker as the run, and they share its environment — imports, secrets, network access, the lot. That makes them ideal for reactions that need the run’s context or its Python objects: cleaning up staging tables the flow created, computing a duration from flow_run, calling a library you already have imported. Their flip side: if the process crashed, an in-process hook is on shaky ground — which is precisely why on_crashed is best-effort and why truly infrastructure-independent alerting shouldn’t rely solely on it.
  • Automations run server-side, on the Prefect API, reacting to the event stream. They don’t live in your code; you configure them in the UI or via the API, and they fire on emitted state-change events regardless of whether the originating process is still alive. That independence is their whole value: an automation watching for Crashed still fires when the worker is gone, because the server saw the state change. Automations are how you do fleet-wide, cross-flow reactions — “notify Slack whenever any production flow enters Failed,” “if this deployment is Late for 10 minutes, page” — without editing a single flow’s code.

The rule of thumb: use a hook when the reaction needs to be part of the run (it touches the run’s resources or objects, or is specific to that one flow); use an automation when the reaction should survive the run and apply across many (platform alerting, SLA monitoring, anything that must fire even if the process is dead). A robust setup uses both — an on_failure hook to clean up locally, and a server-side automation on Failed/Crashed to alert independently of the worker’s health.

How states drive the UI and automations

Everything visible in the Prefect UI is a projection of state. The colored dots on a flow run, the flat-line chart of run outcomes over time, the “Late” badge on a work pool with no workers — all of it is the server reading state transitions off the event stream and rendering them. When you filter the runs page to “show me failures,” you’re querying state type. This is why states aren’t cosmetic: the same signal that colors a dot is the signal an automation subscribes to. A run entering Crashed emits an event; the UI paints it red and any automation watching for Crashed fires, from one source of truth.

That shared substrate is the reason to get comfortable naming states precisely. “The pipeline broke” isn’t actionable. “Three task runs entered Crashed while the work pool showed zero healthy workers” tells you it’s a Late-plus-Crashed infrastructure story, points you at the work pool, and — if you’d set it up — would already have fired an automation before you looked. The next chapter picks up one of the levers that produces those states directly.

Final thoughts

Coming from Airflow, the temptation is to translate: Completed is success, Failed is failed, Cancelled is whatever the closest Airflow thing is, and move on. Resist it. The states Airflow doesn’t have are exactly the ones that make Prefect worth reasoning about — Crashed as a first-class category separate from Failed, Cached as a success that skipped the work, Suspended as a pause that gives back its infrastructure, Late as a schedulable-but-unclaimed alarm. Each of those is a distinction you were probably making anyway in Airflow, badly, with log-grepping and convention. Prefect promotes them to values your code and your platform can both see.

Treat state as data, not decoration. Hold it with return_state, branch on it with the predicates, unpack it carefully with raise_on_failure=False, keep a downstream alive across an upstream failure with allow_failure, and react to transitions with hooks in-process and automations server-side. Do that and the vague Airflow trinity of success/failed/skipped gives way to a model that actually describes what happened — which is the whole point of an orchestrator watching your work in the first place.

Next: Concurrency and Rate Limiting

Comments