Retrying Failed Raster Tasks in Prefect

Give a raster task a retry budget and an exponential backoff schedule so transient S3 and GDAL failures recover on their own. The minimal form is a few keyword arguments on @task:

from prefect import task
from prefect.tasks import exponential_backoff

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=10),  # 10s, 20s, 40s
    retry_jitter_factor=0.5,   # spread concurrent retries so they don't align
    timeout_seconds=120,       # abandon a hung read instead of blocking the run
)
def read_band(url: str):
    import rasterio
    with rasterio.open(url) as src:   # transient /vsicurl/ drops raise RasterioIOError
        return src.read(1)

That alone recovers most flaky network reads. The rest of this page shows how to retry only the errors worth retrying. It sits under Orchestrating Pipelines with Prefect, which introduces tasks and flows; here we go deep on resilience alone and leave scheduling to its own guide.

Why This Arises in Remote Sensing Workflows

Raster pipelines lean heavily on remote I/O. Every scene read through GDAL’s /vsicurl/ or /vsis3/ backend is a chain of HTTP range requests against object storage, and object storage is eventually consistent and occasionally throttled. A busy run pulling hundreds of Sentinel-2 assets will, statistically, hit a dropped connection, a 503 SlowDown, or a stalled socket — not because anything is wrong, but because that is how S3 behaves under load. Without retries, one such blip fails the whole flow run and discards all the work already done.

The trap is retrying indiscriminately. A RasterioIOError from a dropped connection clears on a second attempt; a ValueError because two bands have mismatched shapes will fail identically forever. Retrying the second kind burns the retry budget, delays the inevitable failure by minutes, and buries the real cause under duplicate log lines. Effective resilience means separating transient errors, which deserve a retry, from deterministic errors, which should fail fast.

For where this fits, the parent section is Cloud Execution & Orchestration; the compute layer these tasks often wrap is covered in Scaling Raster Processing with Dask.

Transient versus deterministic failures

The decision tree below is the core of the page: classify the exception, retry transient ones with backoff, and let deterministic ones fail immediately.

Environment & Setup

pip install "prefect>=2.14,<3" "rasterio>=1.3" "botocore>=1.31"

No Prefect server is required to exercise retries — they run in-process when you call the flow directly — but a running server records each attempt in the UI, which is invaluable when tuning backoff.

Complete Working Example

This task reads a raster from S3 and is hardened end to end: bounded retries, exponential backoff with jitter, a timeout, and a retry_condition_fn that retries transient network errors while letting logic errors fail on the first attempt.

from prefect import task, get_run_logger
from prefect.tasks import exponential_backoff
from prefect.states import State
from rasterio.errors import RasterioIOError
from botocore.exceptions import ClientError

# Botocore error codes that indicate a transient, retryable condition
TRANSIENT_S3_CODES = {"500", "503", "SlowDown", "RequestTimeout", "Throttling"}


def is_transient(task, task_run, state: State) -> bool:
    """
    retry_condition_fn: return True to allow a retry, False to fail fast.

    Prefect calls this with the failed run's final state. We fetch the raised
    exception and retry only errors known to be transient.
    """
    try:
        exc = state.result(raise_on_failure=False)
    except Exception as e:      # noqa: BLE001 - the state carries the exception
        exc = e

    if isinstance(exc, RasterioIOError):
        return True             # dropped /vsicurl/ connection, partial response
    if isinstance(exc, ClientError):
        code = exc.response.get("Error", {}).get("Code", "")
        return code in TRANSIENT_S3_CODES
    if isinstance(exc, TimeoutError):
        return True             # stalled socket that our timeout tripped
    return False                # ValueError, KeyError, CRS mismatch → fail fast


@task(
    name="read-s3-raster",
    retries=4,
    retry_delay_seconds=exponential_backoff(backoff_factor=10),  # 10, 20, 40, 80s
    retry_jitter_factor=0.5,       # ± up to 50% so concurrent retries desynchronize
    retry_condition_fn=is_transient,
    timeout_seconds=180,           # a hung read is treated as a failure and retried
)
def read_s3_raster(url: str) -> tuple:
    """Read band 1 from a remote COG. Transient failures are retried automatically."""
    import rasterio
    logger = get_run_logger()
    with rasterio.open(url) as src:          # RasterioIOError here is transient
        if src.count < 1:
            # Deterministic: a bandless file will never become valid → fail fast
            raise ValueError(f"{url} has no bands")
        data = src.read(1)
    logger.info("Read %s shape=%s", url, data.shape)
    return data.shape, str(data.dtype)

The two failure classes are handled by two different mechanisms. retries and the backoff schedule govern how a retry happens; retry_condition_fn governs whether one happens at all. A RasterioIOError cycles through the backoff schedule; the ValueError for a bandless file returns False from is_transient and fails immediately, surfacing the real problem without wasting attempts.

Variant Patterns

1. Per-attempt delays without exponential_backoff

When you want explicit control rather than a computed curve, pass a plain list. Its length should match retries; Prefect uses the last value for any extra attempts.

@task(retries=3, retry_delay_seconds=[5, 30, 120])
def flaky_download(url: str):
    ...   # waits 5s, then 30s, then 120s between the three retries

2. Retry only on a specific botocore throttle code

If you want to be stricter — retry throttling but not generic 500s — narrow the condition function to the exact code.

def retry_on_throttle(task, task_run, state) -> bool:
    exc = state.result(raise_on_failure=False)
    return (
        isinstance(exc, ClientError)
        and exc.response.get("Error", {}).get("Code") == "SlowDown"
    )

@task(retries=5, retry_delay_seconds=exponential_backoff(backoff_factor=5),
      retry_condition_fn=retry_on_throttle)
def get_object(key: str):
    ...

3. Timeout without retry for a watchdog task

Some tasks should be abandoned if slow but never retried — for instance a best-effort thumbnail render. Set a timeout and leave retries at zero.

@task(timeout_seconds=30)   # no retries: a slow render is skipped, not repeated
def render_preview(path: str):
    ...

A task that exceeds timeout_seconds is marked failed; downstream tasks that depend on it are skipped, while independent branches of the flow continue.

Common Errors

Deterministic errors keep retrying and waste minutes No retry_condition_fn is set, so every exception is retried. Add one that returns False for ValueError, KeyError, and CRS or shape mismatches so they fail on the first attempt.

retry_delay_seconds list shorter than retries Prefect reuses the final list element for the remaining attempts, so [5, 30] with retries=4 waits 5, 30, 30, 30. Match the list length to retries, or use exponential_backoff to avoid the mismatch entirely.

A hung read blocks the whole run indefinitely A stalled /vsicurl/ socket will never raise on its own. Set timeout_seconds so Prefect cancels the attempt; combined with retries, the next attempt gets a fresh connection.

Related

• Orchestrating Pipelines with Prefect — the parent guide covering tasks, flows, mapping, caching, and schedules
• Scheduling Sentinel-2 Downloads with Prefect — the scheduled downloader whose fetch task these retry settings harden
• Scaling Raster Processing with Dask — the distributed compute layer these resilient tasks often wrap
• Reading a COG over S3 Without Downloading It — the /vsis3/ read path that produces the transient errors this page retries