Prefect Flow State Transitions Explained

A Prefect run’s state is a claim about execution, not about output, and for a spatial pipeline those come apart in a specific and expensive way: a flow can finish Completed having produced a mosaic with a hole in it. Understanding which transitions are possible — and which of them your fan-out can reach without failing — is what lets you make the final state mean “the product is right” rather than “the code returned”.

When to Use This Pattern#

• A mapped fan-out can partially fail, which is every tile pipeline.
• Alerting is driven by run state, so the state has to be trustworthy.
• Retries and timeouts are in play, and their interaction with states is not obvious.
• Somebody outside the team reads the run history and takes the colour at face value.

Complete Working Example#

The states that matter, and a final-state rule that reflects the product rather than the execution.

from __future__ import annotations

from prefect import flow, task, get_run_logger
from prefect.states import Completed, Failed


@task(retries=2, retry_delay_seconds=[15, 60], timeout_seconds=900, tags=["tile"])
def build_tile(tile: Tile) -> str:
    return render(tile)


@flow(name="tile-build", timeout_seconds=7200)
def tile_build(tiles: list[Tile], coverage_floor: float = 0.995):
    # return_state=True is the whole trick: without it, one failed mapped task
    # raises and the flow ends before the reduction can judge the shortfall.
    states = build_tile.map(tiles, return_state=True)

    done = [s for s in states if s.is_completed()]
    lost = [t for t, s in zip(tiles, states) if not s.is_completed()]
    coverage = len(done) / len(tiles)

    log = get_run_logger()
    log.info("tiles: %d/%d built, coverage %.4f", len(done), len(tiles), coverage)
    for t in lost[:20]:
        log.warning("tile missing: %s", t.path)

    if coverage < coverage_floor:
        # An explicit Failed carries a message an operator can act on, which the
        # default propagated exception does not.
        return Failed(message=f"coverage {coverage:.4f} < {coverage_floor}: "
                              f"{len(lost)} tile(s) missing, first {lost[:5]}")

    mosaic = assemble([s.result() for s in done])
    return Completed(message=f"coverage {coverage:.4f}", data=mosaic)

Crashed is the state most worth distinguishing, and the one most alerting rules lump in with Failed. A crash means the process disappeared — an out-of-memory kill, a reclaimed spot instance, a node drain — so no exception handler ran, no cleanup happened and no message was recorded. That points at capacity or infrastructure, and the response is to resize, reschedule or resume. A Failed run raised something, which points at data or logic and needs a person to read the message. Routing both to the same channel with the same severity teaches everyone to ignore the channel.

Parameter & Option Reference#

Verification & Testing#

from prefect.testing.utilities import prefect_test_harness


def test_partial_failure_fails_the_flow(monkeypatch) -> None:
    monkeypatch.setattr("mymodule.render", flaky(fail_every=8))
    with prefect_test_harness():
        state = tile_build(make_tiles(412), return_state=True)
    assert state.is_failed()
    assert "coverage" in state.message and "missing" in state.message


def test_full_success_reports_the_number(monkeypatch) -> None:
    monkeypatch.setattr("mymodule.render", lambda t: t.path)
    with prefect_test_harness():
        state = tile_build(make_tiles(50), return_state=True)
    assert state.is_completed() and "1.0000" in state.message


def test_retries_do_not_hide_a_persistent_failure(monkeypatch) -> None:
    calls: list[str] = []
    monkeypatch.setattr("mymodule.render", lambda t: calls.append(t.path) or fail())
    with prefect_test_harness():
        state = tile_build(make_tiles(3), return_state=True)
    assert state.is_failed()
    assert len(calls) == 9, "each tile should be attempted three times, not more"

The third test is worth keeping because retry counts are easy to change and hard to notice. A retries=2 on a task inside a flow that is itself retried twice produces nine attempts per tile, not three, and against a rate-limited upstream that is the difference between a run that finishes and a run that gets everybody’s requests refused for an hour. Asserting the total attempt count pins the interaction that no single setting expresses.

There is a related asymmetry between task state and flow state that catches people out. A task run’s state is authoritative about that task: it retried twice, it eventually raised, here is the exception. A flow run’s state is a summary, and by default the summary is “did anything escape”. That default is right for a linear pipeline where any failure is fatal, and wrong for a fan-out where partial failure is the normal case and the interesting question is how partial. Returning an explicit state is how you replace a summary you did not choose with one you did.

The same reasoning applies to what goes in the message. Prefect will store whatever string you return, and it appears in the run list, in notifications and in the API — so it is the highest-leverage sentence in the whole pipeline. The habit worth forming is to write it for somebody who has not seen the code: a count, a threshold, and one concrete example of what is missing. That is three values the flow already has, and it turns a run history into something that can be scanned rather than investigated.

Common Pitfalls#

• Mapping without return_state=True. The first failed tile raises and the flow ends before any coverage judgement can be made.
• Treating Completed as “the product is right”. It means the callable returned; the coverage rule is what connects the two.
• Alerting on Crashed and Failed identically. One is capacity, one is data, and they need different people.
• Retrying at both levels. Attempts multiply, and the multiplication lands on somebody else’s server.
• No flow-level timeout. A run stuck in Running occupies its slot indefinitely and never reaches a state anything can alert on.
• Discarding the state message. A Failed with a bare traceback and a Failed naming twelve missing tiles cost the same to produce.

Frequently Asked Questions#

Related#

• Prefect vs Dagster for GIS workloads — the model these states belong to
• How to structure a DAG for raster processing — where the coverage check sits in the graph
• Limiting DAG fan-out with concurrency groups — why multiplied retries matter
• Alerting on dead-letter queue growth — where the failed tiles go next
• Resuming a failed mosaic from its last checkpoint — the response to a crash

← Prefect vs Dagster for GIS Workloads