Coastal & Marine Spatial Analysis Pipelines

Retrying Failed Prefect Tasks Idempotently

@task(retries=3) is one decorator and it converts a class of transient failures into invisible successes. It also converts a class of partial writes into duplicated data, and a class of real faults into a delayed, quieter version of the same fault. Which of the three you get depends entirely on what the task does when it runs a second time. This page sorts that out, within the orchestration workflow.

Root cause: a retry is a second execution, not a resumption

Prefect re-invokes the task function from the top. Anything the first attempt did to the world is still done, and the second attempt does it again. For a marine pipeline the three write patterns behave completely differently:

Overwrite a keyed object. PUT s3://bucket/tiles/0042.tif. The second attempt overwrites the first. Idempotent, safe, and the natural shape for tile output.

Append to a store. ds.to_zarr(store, append_dim="time"). The second attempt appends again. The store now holds the same timestep twice, both valid, and no error anywhere.

Mutate a counter or a ledger. db.execute("UPDATE runs SET tiles_done = tiles_done + 1"). The second attempt increments again, and the count no longer matches reality.

from prefect import task

# Unsafe: a retry appends the same day twice.
@task(retries=3)
def append_day(ds, store):
    ds.to_zarr(store, append_dim="time")

# Safe: a retry writes the same bytes to the same key.
@task(retries=3)
def write_day(ds, store, day: str):
    ds.to_zarr(f"{store}/days/{day}.zarr", mode="w")

The second form is not merely safer; it is also the one that lets the flow resume after a partial run, because the presence of the key is the record of the work.

Three Writes, One Retry Policy, Three Outcomes Three rows showing the state after one attempt and after a retry. In the keyed-object row, the first attempt writes tile 0042 and the retry overwrites it, so the final state holds exactly one tile and is identical whether one or two attempts ran. In the append row, the first attempt appends a timestep and the retry appends it again, so the store holds the same day twice with both entries valid and no error raised. In the counter row, the first attempt increments the completed count and the retry increments it again, so the ledger reports one more tile than exists. A note observes that only the first row is safe to decorate with retries, and that the retry configuration is identical in all three. the same @task(retries=3), three write paths keyed object PUT tiles/0042.tif retry → same key overwritten 1 tile ✓ append append_dim="time" retry → appends again the day appears twice ✗ counter tiles_done += 1 retry → += 1 again ledger over-counts ✗ The decorator is identical in all three. Retry safety lives in the write path, and nowhere else. Which means adding retries to an existing flow is a review of every task's writes, not a configuration change. The two failing rows are silent: nothing raises, and the state is plausible.

Step-by-step

1. Classify the failure before deciding to retry

Retrying an error that will recur identically wastes the retry budget and delays the alert. Prefect’s retry_condition_fn lets the decision be made from the exception:

import logging

import botocore.exceptions
from prefect import task

logger = logging.getLogger(__name__)

TRANSIENT = (
    botocore.exceptions.ConnectTimeoutError,
    botocore.exceptions.ReadTimeoutError,
    ConnectionResetError,
    TimeoutError,
)


def is_retryable(task, task_run, state) -> bool:
    """Retry transport failures; never retry a data or schema failure."""
    try:
        state.result(raise_on_failure=True)
    except TRANSIENT as exc:
        logger.info("retryable: %s", type(exc).__name__)
        return True
    except botocore.exceptions.ClientError as exc:
        code = exc.response.get("Error", {}).get("Code", "")
        retryable = code in {"SlowDown", "RequestTimeout", "InternalError", "503"}
        logger.info("ClientError %s: %s", code, "retryable" if retryable else "permanent")
        return retryable
    except Exception as exc:                       # noqa: BLE001 — deliberate catch-all
        logger.warning("not retrying %s: %s", type(exc).__name__, exc)
        return False
    return False


@task(retries=4, retry_delay_seconds=[2, 8, 30, 120], retry_condition_fn=is_retryable)
def fetch_tide_series(station: str, day: str):
    ...

The explicit delay list rather than a single number encodes the shape of the failure: a rate limit wants exponential backoff, and four attempts two seconds apart against a throttled endpoint is four failures rather than one.

2. Cache on the inputs so a retry can skip completed work 

from prefect import task
from prefect.cache_policies import INPUTS, TASK_SOURCE


@task(retries=3, cache_policy=INPUTS + TASK_SOURCE, cache_expiration=None)
def grid_tile(tile_id: int, manifest_uri: str, cell_size_m: float, image_digest: str):
    """Cached on its inputs and its own source, so a re-run skips finished tiles."""
    ...

Including TASK_SOURCE in the key is what stops a cached result surviving a change to the function that produced it — the same reasoning as including the image digest in a run identifier. A cache keyed only on arguments will happily return last week’s tile after the algorithm changed.

3. Never retry a validation gate 

@task(retries=0)                     # deliberately zero
def assert_drift_within_budget(residuals, budget_m: float) -> None:
    p95 = float(np.percentile(np.abs(residuals), 95))
    if p95 > budget_m:
        raise ValueError(f"control-point drift p95 {p95:.3f} m exceeds {budget_m} m")

A gate is a measurement, and measurements do not become true on the fourth attempt. Retrying one either wastes time on a deterministic failure or — worse — succeeds because something upstream changed between attempts, which converts a finding into a flake and trains everyone to re-run rather than investigate.

The Task That Always Succeeds on the Fourth Try A histogram of attempts-to-success across all tasks in a nightly flow over thirty days. The overwhelming majority succeed on the first attempt. A modest population succeeds on the second, consistent with occasional transport hiccups. One named task, the tide-series fetch, shows a distribution concentrated on the third and fourth attempts, succeeding every night but never on the first try. A note records that this task never fails the flow and therefore never pages anyone, while consuming its entire retry budget nightly and adding two minutes of backoff to every run — and that a single alert on the mean attempt count would have surfaced it on the first night. attempts to success tasks fetch_tide_series every night 1234 It never fails the flow, so it never pages. It burns its whole retry budget nightly and adds two minutes of backoff.

Verification and acceptance test 

import pytest


def test_running_a_task_twice_leaves_one_record(store, day):
    write_day.fn(dataset, store, day)
    write_day.fn(dataset, store, day)
    entries = [k for k in storage.list(f"{store}/days/") if day in k]
    assert len(entries) == 1, f"{len(entries)} entries for {day}; the write is not idempotent"


def test_a_schema_error_is_not_retried(monkeypatch):
    calls = {"n": 0}

    @task(retries=3, retry_condition_fn=is_retryable)
    def bad():
        calls["n"] += 1
        raise KeyError("expected column 'sog_kn'")

    with pytest.raises(Exception):
        bad.fn()
    assert calls["n"] == 1, f"a permanent error was retried {calls['n']} times"


def test_gates_have_no_retries(flow):
    for t in flow.tasks:
        if t.name.startswith("assert_"):
            assert t.retries == 0, f"{t.name} has retries; a gate must not be retried"

The third test is a policy check, and it is worth having because the natural instinct when adding retries to a flow is to add them everywhere.

Making an existing flow retry-safe without rewriting it

Most marine flows acquire retries after they are written, which means the question is not how to design for idempotence but how to retrofit it. Three patterns cover nearly every task in a pipeline of this kind, in increasing order of effort.

Wrap the write in a stage-and-promote. Any task whose output is a single object can be made idempotent without touching its logic: write to a run-scoped temporary key, then copy to the final key. The copy is atomic per object, so a retry either re-does the whole thing or does nothing visible.

from contextlib import contextmanager


@contextmanager
def staged_write(final_key: str, run_id: str):
    """Yield a temporary key; promote it atomically on clean exit."""
    tmp = f"_staging/{run_id}/{final_key.replace('/', '_')}"
    try:
        yield tmp
    except Exception:
        storage.delete(tmp, missing_ok=True)
        raise
    storage.copy(tmp, final_key)
    storage.delete(tmp, missing_ok=True)

Convert an append into a keyed write. A task appending a timestep becomes a task writing days/2026-03-14.zarr, with a separate, cheap consolidation step that assembles the days into one logical store. The consolidation is itself idempotent because it rebuilds from what exists.

Guard a mutation with a conditional. Where a counter or a ledger genuinely must be mutated, make the write conditional on the row not already recording this run: INSERT ... ON CONFLICT (run_id, tile_id) DO NOTHING. The second attempt is a no-op at the database rather than in application logic, which is the only place the check cannot race.

The order matters because the effort does. The first pattern is a decorator and covers most tile and product writes. The second is a schema change and covers the append case. The third needs a unique constraint and is only required where a genuine mutable aggregate exists — which, in a well-shaped marine pipeline, is usually only the run manifest.

Three Retrofits, in Increasing Order of Effort Three patterns for making an existing non-idempotent task safe to retry, arranged by the effort each requires. The first, stage and promote, wraps the write in a context manager that writes to a run-scoped temporary key and copies it to the final key on clean exit; it requires no change to the task's own logic and covers most tile and product writes. The second, keyed write, converts an append into one object per unit with a separate idempotent consolidation step; it requires a schema change and covers the append case. The third, conditional insert, guards a genuine mutation with an on-conflict-do-nothing clause; it requires a unique constraint and is needed only where a mutable aggregate genuinely exists, which in a well-shaped pipeline is usually just the run manifest. stage and promote write to _staging/run/key copy to the final key atomic per object effort: a decorator covers most product writes keyed write append → days/2026-03-14 separate consolidation step rebuilds from what exists effort: a schema change covers the append case conditional insert ON CONFLICT DO NOTHING the no-op happens in the DB where it cannot race effort: a unique constraint only for real mutable aggregates Work left to right and most tasks are done at the first box. The third is usually needed only for the run manifest. Retrofitting is a review of writes, one task at a time — not a configuration change applied to the flow. Add the retries only to tasks that have been through it.

Edge cases and gotchas

  • Prefect retries the task, not the flow. A task that succeeded is not re-run when a later task fails and the flow is retried from a failed state; the cache decides. If a task’s output depends on wall-clock time rather than on its inputs, that cached result is now stale in a way the cache key cannot see.
  • A retry after a partial multi-part upload leaves the parts. S3 charges for incomplete multipart uploads until they are aborted. An AbortIncompleteMultipartUpload lifecycle rule is the reliable cleanup, as covered in S3 lifecycle rules.
  • Backoff delays consume the flow’s SLA. Four retries at 2, 8, 30 and 120 seconds add two and a half minutes per failing task. On a flow with hundreds of tasks and a transient regional problem, the accumulated backoff can exceed the run’s entire time budget while every individual task eventually succeeds.

Up: Orchestrating Marine Pipelines with Dask and Prefect