Writing Idempotent Airflow Tasks for Imagery Ingestion

One-sentence answer: give every artefact a deterministic identity, check the manifest before fetching, write to a run-scoped staging key, validate, then copy to the final key and insert the manifest row in one committed transaction — after that a retry is a no-op by construction.

Context

Retries are not an edge case in a seasonal pipeline; they are the normal way work completes. Tokens expire, object stores return 503, a worker is evicted mid-download, and a backfill overlaps a scheduled run. If any of those can produce a second copy of a scene, then every scheduling decision becomes a correctness decision and the team stops re-running things — which is exactly when the pipeline becomes unmaintainable.

The property to build is narrow and precise: running the task twice leaves the same state as running it once. This guide implements it for imagery, under orchestrating seasonal pipelines with Airflow, using the manifest table from PostGIS schema design for farm data.

What a crash leaves behind at each point in the task A timeline of the ingestion task with three crash points marked. A crash during fetch or validation leaves only a staging object that expires. A crash after the final copy but before the manifest commit leaves an object with no manifest row, which weekly reconciliation reports. After the commit, any retry finds the manifest row and exits without work. manifest lookup fetch → staging key validate copy → final key manifest insert + commit crash here staging object only — expires in 7 days crash here object without a manifest row — reconciliation finds it; a retry rewrites the key after commit: every retry exits 0 The unsafe ordering — manifest first, copy second — can mark work done that never happened. Never do that.

Prerequisites

Beyond the parent topic’s stack: an object store client, a PostgreSQL connection, and the ingest_manifest table with its unique constraint on source, external identifier and checksum. Without that constraint none of this works — the insert simply adds another row.

Step-by-step

1. Identity first. Source, item identifier and content checksum. The checksum is what distinguishes a reprocessed scene from the original.

Three ways to ask whether the work is already done A table comparing three completion checks. Checking that the output object exists proves only that bytes are somewhere, and object listings lag. Checking its size proves the write finished but nothing about validation. A committed manifest row proves a complete, validated artefact was published, which is the only guarantee that survives a retry. What the task checks What it proves What it misses Output object exists a listing or a HEAD Bytes are somewhere Listings lag; a half-written object looks whole Output size looks right a size comparison The write finished Says nothing about validation or identity Manifest row committed source, id and checksum A complete artefact was published Nothing — this is the guarantee

2. Look before fetching. A manifest hit is a successful exit, not a skip to be logged as a warning.

3. Stage under the run identifier, so two concurrent runs cannot write the same staging key.

4. Validate the staged bytes before they are visible to anything downstream.

5. Promote, then commit the manifest row in one transaction.

PYTHON 
import hashlib
from dataclasses import dataclass

from sqlalchemy import create_engine, text


@dataclass(frozen=True)
class Artifact:
    source: str            # 'sentinel-2-l2a'
    external_id: str       # the STAC item id
    checksum: str          # sha256 of the fetched bytes
    final_key: str         # published/s2/{tile}/{date}/{band}.tif


def already_published(conn, art: Artifact) -> bool:
    return conn.execute(text("""
        SELECT 1 FROM ingest_manifest
        WHERE source = :s AND external_id = :e AND checksum = :c
    """), {"s": art.source, "e": art.external_id, "c": art.checksum}).first() is not None


def ingest_one(store, dsn: str, *, source: str, item_id: str, href: str,
               final_key: str, run_id: str) -> dict:
    """Fetch, validate and publish one imagery artefact. Safe to call any number of times."""
    engine = create_engine(dsn, future=True)
    staging_key = f"staging/{run_id}/{source}/{item_id}"

    # 1. Cheap pre-check on identity that does not need the bytes: item id alone.
    with engine.connect() as conn:
        seen = conn.execute(text(
            "SELECT checksum FROM ingest_manifest WHERE source = :s AND external_id = :e"
        ), {"s": source, "e": item_id}).all()

    # 2. Fetch to staging. The bytes are needed to compute the checksum.
    body = store.fetch(href)
    checksum = hashlib.sha256(body).hexdigest()
    if any(row.checksum == checksum for row in seen):
        print(f"{item_id}: already published with checksum {checksum[:12]} — no-op")
        return {"published": 0, "reason": "manifest hit"}

    store.put(staging_key, body)

    # 3. Validate before it becomes visible anywhere.
    validate_staged(store, staging_key)

    art = Artifact(source, item_id, checksum, final_key)
    with engine.begin() as conn:                    # one transaction, commits on exit
        if already_published(conn, art):            # re-check under the transaction
            return {"published": 0, "reason": "concurrent publisher won"}
        store.copy(staging_key, final_key)          # object copy, then register
        conn.execute(text("""
            INSERT INTO ingest_manifest (source, external_id, checksum, object_key)
            VALUES (:s, :e, :c, :k)
            ON CONFLICT (source, external_id, checksum) DO NOTHING
        """), {"s": art.source, "e": art.external_id, "c": art.checksum, "k": art.final_key})
    return {"published": 1, "key": final_key, "checksum": checksum}


def validate_staged(store, key: str) -> None:
    """Refuse to publish anything that is not a readable, projected raster."""
    import rasterio
    with rasterio.open(store.url_for(key)) as src:
        assert src.crs is not None, f"{key}: no CRS"
        assert src.count >= 1, f"{key}: no bands"
        assert src.width > 0 and src.height > 0, f"{key}: zero-sized raster"
        assert src.profile.get("tiled", False), f"{key}: not internally tiled"

Inline verification — the test that most pipelines lack, and the one that catches a broken refactor:

PYTHON 
first = ingest_one(store, DSN, source="sentinel-2-l2a", item_id="S2B_T30UXB_20260612",
                   href=HREF, final_key=KEY, run_id="run-1")
second = ingest_one(store, DSN, source="sentinel-2-l2a", item_id="S2B_T30UXB_20260612",
                    href=HREF, final_key=KEY, run_id="run-2")

assert first["published"] == 1, "the first ingestion published nothing — fixture problem"
assert second["published"] == 0, (
    "the second ingestion published again — the manifest is not being consulted, "
    "or the insert is outside the publishing transaction")

rows = engine.connect().execute(text(
    "SELECT count(*) FROM ingest_manifest WHERE external_id = 'S2B_T30UXB_20260612'")).scalar()
assert rows == 1, f"{rows} manifest rows for one artefact"

Settings worth being deliberate about

Setting Default here Why it matters
Identity source + item id + checksum The checksum is what distinguishes a reprocessed scene from the original. Identity on the item alone silently keeps the first version forever
Staging key staging/{run_id}/… Must be unique per attempt. Keying on the logical date instead means a retry writes over the failed attempt’s partial object
Ordering copy, then insert, then commit Leaves at worst an unregistered object, which reconciliation finds. The reverse order can mark work complete that never happened
Manifest constraint UNIQUE (source, external_id, checksum) Without it, ON CONFLICT DO NOTHING is a no-op and every retry adds a row
Staging retention 7 days Long enough to inspect a failed promotion; short enough that staging does not quietly become a second archive
Reconciliation cadence weekly Compares the published prefix against the manifest in both directions. Objects without rows mean a crashed promotion; rows without objects mean something deleted published data

The pre-check and the in-transaction re-check are both necessary and do different jobs. The first avoids a pointless download in the common case; the second decides the winner when two runs pass the first check at the same moment.

Gotchas and edge cases

  • A checksum requires the bytes, so the cheap pre-check has to be on the identifier. The two-stage check above avoids re-downloading in the common case while still detecting a reprocessed scene whose identifier is unchanged but whose content differs.
The ordering that decides what a crash costs Two panels comparing the order of the object copy and the manifest insert. Copying first and registering afterwards can leave an unregistered object, which reconciliation finds and a retry repairs. Registering first can leave a manifest row with no object, which makes every later run believe the work is done. Copy, then register — safe 1. copy staging object to the final key 2. insert the manifest row 3. commit A crash after step 1 leaves an unregistered object. Weekly reconciliation lists it and a retry rewrites it. Worst case: harmless duplicate work. Register, then copy — unsafe 1. insert the manifest row and commit 2. copy staging object to the final key A crash after step 1 leaves a row with no object. Every later run believes the work is finished. Nothing retries it; the gap is permanent. Worst case: silently missing data.

  • Writing directly to the final key is the mistake this whole pattern exists to avoid. A crash mid-write leaves a truncated object at the path everything downstream reads, and it looks perfectly valid to a listing.

  • The transaction covers the database, not the object store. A crash between the copy and the commit leaves an unregistered object. That is the safe direction, and the weekly reconciliation in the section overview finds it. Reversing the order is not safe.

  • Concurrent publishers race. Two runs can pass the pre-check simultaneously. The unique constraint decides the winner and ON CONFLICT DO NOTHING makes the loser harmless — but only if the constraint actually exists.

  • Airflow retries the whole task, not the failed step. That is fine here and it is the point: the retried task re-runs from the top, finds the manifest row if the previous attempt got that far, and exits.

  • run_id in the staging key must be genuinely unique per attempt. Using the logical date instead means a retry writes to the same staging key as the failed attempt, which reintroduces the truncated-object problem one level down.

This guide is part of Orchestrating Seasonal Pipelines with Airflow — see there for cadence, pools and dataset-triggered derivation.

Parent section Orchestrating Seasonal Pipelines with Airflow Related Backfilling a Season of Field Imagery Without Duplicates