Retrying Failed Tile Tasks Without Reprocessing Everything

Make geospatial pipeline retries cheap: idempotent staged writes, a completion index, exponential backoff, poison-tile quarantine and reruns that touch only what failed.

Three things make a retry cheap: a staged write so a killed task leaves nothing behind, a completion marker written last so “done” is unambiguous, and a skip check the task performs before doing any work. With those in place, rerunning a 6,000-tile job after 40 failures costs 40 tiles of compute. Without them it costs 6,000 — and teams respond by not retrying, which is worse.

This page is the retry mechanics. The surrounding graph design is in Orchestrating Geospatial ML Pipelines.

The completion marker, not the output file, decides what to skip A retried tile task checks for a completion marker. If the marker exists and records the current model version, the tile is skipped. If the marker is absent the tile is processed. If the marker exists but records an older model version, the tile is reprocessed because the output is stale. Check the marker first — it is the only unambiguous signal read _done marker for this tile and grid present, version matches → skip, cost ≈ one HEAD request absent → process, stage, then write the marker last present, older version → reprocess: the output is stale, not missing Checking only “does the .tif exist” accepts a truncated file as done.

Why This Fails in Geospatial ML Pipelines

The naive skip check — “does the output object exist?” — is unsafe precisely when it matters. A worker killed mid-upload leaves a partial object at the final key. On the retry, the check sees it, declares the tile done, and the run publishes a mosaic with a corrupt patch. Because GeoTIFF readers happily open a truncated file and return data up to the cut, the corruption travels downstream and surfaces weeks later as an inexplicable band of nonsense.

The second failure is retrying deterministic errors. Exponential backoff is designed for transience; a tile whose source scene contains an invalid geometry will fail identically on every attempt. Three retries × 6,000 tiles × a two-minute backoff can turn a ten-minute failure into an hour of nothing, and it consumes the API quota that the healthy tiles need.

Third, staleness. When the model is redeployed, yesterday’s completion markers are still there, so a rerun skips every tile and republishes the old predictions under a new run id. The marker has to record what produced the output — model version, code version, source scene id — so “done” means “done with the thing we are asking for now”.

Core Principles

  • Stage, then promote, then mark. The marker is written last, after the object is complete.
  • Put identity in the marker. Model version, source scene, code version.
  • Skip on the marker, never on the object.
  • Backoff for transient errors only; classify and quarantine deterministic ones.
  • Let the gate decide, not the tile. Failures are counted, then judged in aggregate.
  • Make the skip cheap. One HEAD request, not a read of the raster.

Production-Ready Code

from __future__ import annotations

import json
import logging
import os
import tempfile
from dataclasses import dataclass

import boto3
import rasterio
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)
s3 = boto3.client("s3")


class PoisonTile(Exception):
    """A deterministic failure — retrying will not help."""


@dataclass(frozen=True)
class TileTarget:
    bucket: str
    prefix: str          # e.g. "predictions/utm33n_10m/2026-08-04/"
    tile_id: str
    model_version: str
    scene_id: str

    @property
    def object_key(self) -> str:
        return f"{self.prefix}{self.tile_id}.tif"

    @property
    def marker_key(self) -> str:
        return f"{self.prefix}_markers/{self.tile_id}.json"


def already_done(t: TileTarget) -> bool:
    """True only when a marker exists AND records the identity we are asking for."""
    try:
        body = s3.get_object(Bucket=t.bucket, Key=t.marker_key)["Body"].read()
    except ClientError as exc:
        if exc.response["Error"]["Code"] in ("NoSuchKey", "404"):
            return False
        raise
    marker = json.loads(body)
    fresh = (marker.get("model_version") == t.model_version
             and marker.get("scene_id") == t.scene_id)
    if not fresh:
        logger.info("tile %s marker is stale (%s vs %s) — reprocessing",
                    t.tile_id, marker.get("model_version"), t.model_version)
    return fresh


def write_tile_and_mark(t: TileTarget, array, profile) -> str:
    """Stage, promote, then mark. A kill at any point leaves a consistent state."""
    tmp_key = f"{t.object_key}.{os.getpid()}.tmp"
    with tempfile.NamedTemporaryFile(suffix=".tif") as fh:
        with rasterio.open(fh.name, "w", **profile) as dst:
            dst.write(array, 1)
            dst.update_tags(model_version=t.model_version, scene_id=t.scene_id)
        s3.upload_file(fh.name, t.bucket, tmp_key)

    s3.copy_object(Bucket=t.bucket, Key=t.object_key,
                   CopySource={"Bucket": t.bucket, "Key": tmp_key})
    s3.delete_object(Bucket=t.bucket, Key=tmp_key)

    s3.put_object(Bucket=t.bucket, Key=t.marker_key,
                  Body=json.dumps({"tile_id": t.tile_id,
                                   "model_version": t.model_version,
                                   "scene_id": t.scene_id}).encode())
    return t.object_key


def process_tile_resumable(t: TileTarget, compute) -> str:
    """Skip-or-compute with deterministic-failure classification."""
    if already_done(t):
        logger.debug("tile %s already complete", t.tile_id)
        return t.object_key
    try:
        array, profile = compute(t.tile_id)
    except (ValueError, KeyError) as exc:          # data problems, not infrastructure
        raise PoisonTile(f"tile {t.tile_id}: {exc}") from exc
    return write_tile_and_mark(t, array, profile)


def quarantine(t: TileTarget, reason: str) -> None:
    """Record a poison tile so the gate can count it and a human can inspect it."""
    s3.put_object(Bucket=t.bucket, Key=f"{t.prefix}_quarantine/{t.tile_id}.json",
                  Body=json.dumps({"tile_id": t.tile_id, "reason": reason,
                                   "model_version": t.model_version}).encode())
    logger.error("quarantined tile %s: %s", t.tile_id, reason)

Step-by-Step Walkthrough

Step 1 — Wire the skip check into the mapped task

from airflow.decorators import task
from airflow.exceptions import AirflowFailException


@task(retries=3, retry_exponential_backoff=True)
def process_tile(spec: dict, tile_id: str) -> str | None:
    from geoml.tiles import TileTarget, process_tile_resumable, PoisonTile, quarantine

    target = TileTarget(bucket="predictions", prefix=spec["staging_prefix"],
                        tile_id=tile_id, model_version=spec["model_version"],
                        scene_id=spec["scene_id"])
    try:
        return process_tile_resumable(target, compute=run_inference_for_tile)
    except PoisonTile as exc:
        quarantine(target, str(exc))
        raise AirflowFailException(str(exc))       # fail WITHOUT retrying

AirflowFailException is the important detail: it marks the task failed and skips the remaining retries, so a deterministic failure costs one attempt rather than four.

Step 2 — Measure what a rerun actually costs

import time

t0 = time.perf_counter()
done = [t for t in tiles if already_done(TileTarget(**base, tile_id=t))]
print(f"{len(done)}/{len(tiles)} already complete, "
      f"checked in {time.perf_counter() - t0:.1f}s")

The check should take milliseconds per tile. If it is slow, it is reading the raster instead of the marker.

Step 3 — Let the gate judge the failures

def validate_with_tolerance(spec: dict, produced: list[str],
                            max_missing_frac: float = 0.002) -> dict:
    expected = set(tiles_for_grid(spec["grid_id"]))
    got = {k.rsplit("/", 1)[-1].removesuffix(".tif") for k in produced if k}
    missing = expected - got
    quarantined = list_quarantine(spec["staging_prefix"])

    frac = len(missing) / max(1, len(expected))
    ok = frac <= max_missing_frac
    return {"ok": ok, "missing": sorted(missing)[:20], "n_missing": len(missing),
            "missing_frac": round(frac, 5), "quarantined": quarantined}

A tolerance turns a binary outcome into a policy: 3 missing tiles out of 6,000 publishes with a recorded gap; 600 missing stops the run.

Step 4 — Rerun only what failed

# Airflow: clear just the failed mapped instances, not the whole DAG run
airflow tasks clear landcover_daily \
    --task-regex "process_tile" --only-failed \
    --start-date 2026-08-04 --end-date 2026-08-04 --yes

Because every successful tile has a marker, even a full clear is cheap — the completed tiles return after a single HEAD request each.

Classify the error before deciding to retry Failures are split into two branches. Transient errors such as connection timeouts, rate limiting and worker eviction go to exponential backoff retry. Deterministic errors such as invalid geometry, a missing band or a malformed CRS go directly to quarantine, because retrying them cannot succeed. Not every failure deserves a retry tile task raised transient timeout · HTTP 429 · 5xx · eviction → retry with backoff (3×) succeeds ~95% of the time deterministic invalid geometry · missing band · bad CRS → quarantine, do not retry retrying burns quota for nothing The classification lives in the task’s except clauses — make it explicit, not incidental.

Verification

import json
import pytest


def test_marker_is_written_after_the_object(moto_s3):
    """A kill between upload and marker must leave the tile NOT done."""
    t = TileTarget("bkt", "p/", "t0001", "v7", "S2_20260804")
    s3.copy_object(Bucket="bkt", Key=t.object_key,
                   CopySource={"Bucket": "bkt", "Key": "seed.tif"})
    assert not already_done(t), "object present but unmarked must not count as done"


def test_stale_marker_triggers_reprocessing(moto_s3):
    t = TileTarget("bkt", "p/", "t0001", "v8", "S2_20260804")
    s3.put_object(Bucket="bkt", Key=t.marker_key,
                  Body=json.dumps({"model_version": "v7",
                                   "scene_id": "S2_20260804"}).encode())
    assert not already_done(t)


def test_rerun_skips_completed_tiles(moto_s3):
    t = TileTarget("bkt", "p/", "t0001", "v7", "S2_20260804")
    calls = []
    write_tile_and_mark(t, arr, profile)
    process_tile_resumable(t, compute=lambda tid: calls.append(tid) or (arr, profile))
    assert calls == [], "a completed tile was recomputed"


def test_poison_tile_is_not_retried():
    t = TileTarget("bkt", "p/", "bad", "v7", "S2_20260804")
    def boom(_): raise ValueError("invalid geometry")
    with pytest.raises(PoisonTile):
        process_tile_resumable(t, compute=boom)


def test_gate_tolerates_a_few_missing_tiles():
    report = validate_with_tolerance({"grid_id": "g"}, produced=keys[:-3],
                                     max_missing_frac=0.002)
    assert report["ok"] and report["n_missing"] == 3

The first test is the one that pays for itself: it encodes the ordering constraint that makes every other guarantee here work.

What a rerun costs, with and without markers Two bars compare rerun cost after forty failures in a six thousand tile job. Without markers, all six thousand tiles are reprocessed. With markers, five thousand nine hundred and sixty are skipped after a cheap head request and only forty are recomputed. Rerun after 40 failures in a 6 000-tile job 6 000 tiles no markers ~5 h of compute 40 tiles with markers ~2 min + 6 000 HEADs

FAQ

Is checking whether the output exists enough to skip a tile?

Only with atomic writes. A direct write killed halfway leaves a truncated object that the check accepts. Use a staged write plus a completion marker written last, carrying the run identity.

How many retries should a tile task get?

Three with exponential backoff covers the transient cases. Beyond that a tile is usually failing deterministically, so quarantine it — retrying burns quota the healthy tiles need.

Should a single poison tile fail the whole run?

No, but it must be visible. Let the mapped tasks finish, count the failures, and let the validation gate apply a tolerance — small gaps publish with a record, systemic failures stop the run.

Part of: Orchestrating Geospatial ML Pipelines Part of: Geospatial MLOps and Model Deployment