Idempotency Keys in Spatial ETL

In short: an idempotency key is a deterministic name for a unit of work, derived from the work’s inputs, that lets a task ask “have I already done this?” before doing it again. In spatial ETL the unit is rarely a row — it is a tile, a bundle, a mosaic window or an extent — and the key must be derived from the geometry and the coordinate reference system as well as the bytes. Get the unit right and retries become free; get it wrong and every retry doubles a feature table.

Retries are not an edge case in geospatial orchestration, they are the normal operating mode. A tile job that touches a WMS endpoint, a gdalwarp call that runs for eleven minutes on a spot worker, a COPY into PostGIS that competes with a vacuum — each of these fails often enough that any pipeline without retries stalls weekly. The moment you turn retries on, though, you have committed to a much harder property: the second run must not corrupt what the first run half-finished. That property is idempotency, and the key is how you buy it.

Prerequisites & Architecture Baseline#

• An orchestrator with per-task retry configuration — Prefect 2.14+ (@task(retries=…)) or Dagster 1.5+ (RetryPolicy)
• A transactional store for the ledger: PostgreSQL 14+ with PostGIS 3.3+ is the usual choice, because it is already in the stack
• A unique index on the key column — the ledger's correctness comes from the database constraint, not from application code
• pyproj 3.6+ and rasterio 1.3+ (or geopandas 0.14+), pinned, since a PROJ upgrade can change how a CRS serialises and therefore change a key derived from it
• Object storage with conditional writes (If-None-Match) if outputs land in S3/GCS rather than a database
• A decision, written down, about what a "unit of work" is for each flow: bundle, tile, extent or feature

Core Principles#

1. The key names the work, not the attempt. Anything that varies between attempts — a run id, a timestamp, a worker hostname, a temporary path — must stay out of the digest. If your key changes when the orchestrator retries, it is a run identifier wearing an idempotency key’s clothes, and it guarantees duplicates rather than preventing them.

2. The key covers everything that changes the output. For a reprojection task that means the source bytes, the source CRS, the target CRS, the resampling method and the nodata value. Two runs that differ in resampling produce different pixels; if the key ignores resampling, the second run is skipped and the pipeline quietly serves the first run’s output under the second run’s contract.

That question is worth applying literally, field by field, when you first define a key. It resolves the arguments that otherwise recur every few months. Does the worker’s GDAL version belong in the key? It changes the bytes on a resampling-algorithm fix, so strictly yes — but including it rebuilds the world on every image upgrade, so most teams exclude it and handle GDAL changes as an explicit, versioned rollout instead. Does the target table name belong in it? Only if one flow writes to several, in which case it is part of the effect and must be hashed. The decisions differ per pipeline; what does not differ is that they should be made once, written down, and encoded in the algorithm version rather than rediscovered.

3. Claim before you work. The ledger insert must precede the expensive operation, not follow it. Two workers that start a 40-minute mosaic at the same second will both complete it if the ledger is only written at the end. INSERT … ON CONFLICT DO NOTHING RETURNING gives you an atomic claim: exactly one worker gets a row back.

4. Separate claimed from completed. A single boolean cannot distinguish “in progress” from “done”, and that distinction is what lets a reaper find work abandoned by a worker that was terminated mid-run. Store claimed_at and completed_at, and treat a row claimed long ago and never completed as reclaimable.

5. Choose the coarsest unit that is still safely re-runnable. Feature-level keys give perfect granularity and cost a lookup per feature, which is unaffordable at a million geometries. Tile-level keys cost one lookup per tile and re-do at most one tile’s work. The right unit is usually the same one your DAG design for spatial ETL already uses as its task boundary.

6. Version the algorithm. Bake a version integer into the key’s prefix. When you change what goes into the digest — adding the nodata value, say — bumping the version re-keys everything cleanly instead of leaving a ledger where old and new keys are computed by different rules and neither can be trusted.

Production Implementation#

The implementation below is deliberately generic in its hashing and specific in what it hashes. SpatialWorkKey takes the parameters that decide the output — including the CRS and the nodata value that a tabular pipeline would never think about — and produces a stable key plus the ledger operations that make it useful.

from __future__ import annotations

import base64
import hashlib
import json
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta, timezone
from typing import Any, Mapping, Optional

import psycopg
from pyproj import CRS

ALGO_VERSION = 2
STALE_CLAIM = timedelta(hours=2)


@dataclass(frozen=True)
class TileWork:
    """Everything that decides the pixels this task will write."""

    layer: str
    tile_z: int
    tile_x: int
    tile_y: int
    source_etag: str          # the object-store ETag of the source scene
    src_crs: str              # e.g. "EPSG:32633"
    dst_crs: str              # e.g. "EPSG:3857"
    resampling: str = "bilinear"
    nodata: Optional[float] = None

    def canonical(self) -> Mapping[str, Any]:
        # Normalise the CRS through pyproj so "EPSG:3857", "epsg:3857" and an
        # equivalent WKT string all collapse to one authority code. Without this,
        # a source that starts declaring WKT re-keys every tile it touches.
        payload = asdict(self)
        payload["src_crs"] = CRS.from_user_input(self.src_crs).to_authority()
        payload["dst_crs"] = CRS.from_user_input(self.dst_crs).to_authority()
        return payload


def work_key(work: TileWork, namespace: str = "tiles") -> str:
    """Deterministic key for one tile of one layer at one source version."""
    blob = json.dumps(work.canonical(), sort_keys=True, separators=(",", ":"))
    digest = hashlib.sha256(f"{namespace}:v{ALGO_VERSION}:{blob}".encode("utf-8"))
    return base64.urlsafe_b64encode(digest.digest()[:12]).decode("ascii").rstrip("=")


def claim(conn: psycopg.Connection, key: str, work: TileWork) -> bool:
    """Atomically claim the work. True means this worker owns it.

    A claim older than STALE_CLAIM with no completion is treated as abandoned and
    is taken over — the worker that held it was killed, not merely slow.
    """
    cutoff = datetime.now(timezone.utc) - STALE_CLAIM
    with conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO tile_ledger (work_key, layer, z, x, y, claimed_at)
            VALUES (%s, %s, %s, %s, %s, now())
            ON CONFLICT (work_key) DO UPDATE
              SET claimed_at = now()
              WHERE tile_ledger.completed_at IS NULL
                AND tile_ledger.claimed_at < %s
            RETURNING work_key
            """,
            (key, work.layer, work.tile_z, work.tile_x, work.tile_y, cutoff),
        )
        return cur.fetchone() is not None


def complete(conn: psycopg.Connection, key: str, feature_count: int) -> None:
    with conn.cursor() as cur:
        cur.execute(
            "UPDATE tile_ledger SET completed_at = now(), feature_count = %s "
            "WHERE work_key = %s",
            (feature_count, key),
        )

The ledger table it depends on is small, and every column in it earns its place:

CREATE TABLE tile_ledger (
    work_key      text PRIMARY KEY,          -- the idempotency key
    layer         text        NOT NULL,
    z             smallint    NOT NULL,
    x             integer     NOT NULL,
    y             integer     NOT NULL,
    claimed_at    timestamptz NOT NULL,
    completed_at  timestamptz,               -- NULL while in flight
    feature_count integer
);

-- Finds abandoned work in one index scan rather than a sequential sweep.
CREATE INDEX tile_ledger_inflight
    ON tile_ledger (claimed_at)
    WHERE completed_at IS NULL;

Step-by-Step Walkthrough#

1. Describe the work as data, not as arguments. TileWork is a frozen dataclass, so the same values always serialise the same way. Passing the parameters loosely — as kwargs assembled at three call sites — is how a key ends up depending on argument order.
2. Normalise the coordinate reference system before hashing it. CRS.from_user_input(...).to_authority() collapses EPSG:3857, a PROJ string and the equivalent WKT into one tuple. Skipping this step means the key changes the day the upstream catalogue starts emitting WKT2, and every tile is silently rebuilt.
3. Serialise with sorted keys and no whitespace. json.dumps(..., sort_keys=True, separators=(",", ":")) removes both dictionary-order and formatting variance from the digest. This is the cheapest determinism you will ever buy.
4. Prefix with a namespace and an algorithm version. The namespace keeps staging and production ledgers apart; the version lets you change the hashing rule later without reinterpreting historical rows.
5. Claim the row before doing the work. The INSERT … ON CONFLICT DO UPDATE … WHERE completed_at IS NULL AND claimed_at < cutoff is a single atomic statement that both claims new work and reclaims abandoned work. Only the worker that gets a row back proceeds.
6. Do the expensive operation, then record completion. Between the claim and the completion the row advertises “someone is on this”, which is what stops a duplicate worker and what makes the in-flight index useful to an operator.
7. Reap abandoned claims on a schedule. A five-minute flow that lists rows older than the stale-claim window and re-queues them turns a killed spot worker into a delay rather than a hole in the mosaic.

Edge Cases & Failure Recovery#

A source that re-publishes identical bytes under a new ETag. Some object stores change the ETag on a multipart re-upload even when the content is unchanged. The key then moves, and the tile is rebuilt for no reason. If your source does this, hash the source content digest (an x-amz-checksum-sha256 if available) rather than the ETag, and accept the extra HEAD request.

A partially written output. The ledger says completed, but the object store holds a truncated GeoTIFF because the worker died between the flush and the close. Recovery is to write to a temporary key and rename — object-store renames are atomic in GCS and are a copy-then-delete in S3 — so completion is only recorded after the final object exists. Verify with gdalinfo /vsis3/bucket/key in the reaper, which fails loudly on a truncated file.

Antimeridian-crossing extents. An extent-keyed job whose bounding box crosses ±180° serialises as a box spanning the whole globe if you normalise longitudes naively. The key is then shared by two genuinely different jobs. Hash the tile indices, not the bounding box, whenever the work is tiled; where you must hash an extent, split it at the antimeridian first and key each half.

Clock skew across workers. claimed_at written by a worker whose clock is two minutes fast can make a claim look newer than it is, delaying reclamation. Write timestamps with the database’s now(), as above, never with the worker’s clock.

A retried task whose inputs changed underneath it. If the source is republished between attempt one and attempt two, the second attempt computes a different key — correctly, because it is different work. The stale in-flight row from attempt one is then reaped rather than completed. This is the intended behaviour, and it is worth asserting in a test, because the alternative implementations (keying on the task run id) hide it.

A ledger that outlives the data it describes. Tiles get deleted, layers get retired, and a ledger row that survives its output claims work is done when the artefact is gone. The cheap guard is to make deletion go through the same path: whatever removes the object also deletes or tombstones the ledger row, in one transaction where the store allows it. Where it does not — an object store and a database cannot share a transaction — delete the ledger row first. A missing row causes a rebuild, which is wasteful; a stale row causes a permanent gap, which is a defect.

Two flows keyed on the same work with different effects. A tiling flow and a statistics flow may both be keyed on (layer, z, x, y, source_etag) while writing entirely different outputs. Sharing a namespace makes each one skip the other’s work. The namespace exists precisely for this: tiles and zonal-stats are different effects on the same inputs and must not share an identity.

Configuration Reference#

Two of these settings interact in a way worth stating explicitly. The stale-claim window and the orchestrator’s own task timeout must be consistent: if the orchestrator kills a task at 90 minutes but the ledger only reclaims at 120, there is a half-hour hole in which the work is dead and nothing will pick it up. Set the stale window from the timeout, not from a guess — the same reasoning that drives timeout budgets for geotasks, where every layer’s limit is derived from the layer beneath it rather than chosen independently.

The ledger’s growth is the other operational question. One row per tile per source version is a lot of rows: a 12-zoom pyramid re-published weekly reaches tens of millions within a year. Partition the table by month on claimed_at and drop partitions older than your re-drive horizon, or keep only the most recent completed row per (layer, z, x, y) and let the historical record live in the warehouse. What you must not do is add a DELETE that runs during the pipeline, because a deleted completed row is indistinguishable from work never done.

Reaper cadence is the one setting with no good default. A pipeline whose tiles take seconds wants a one-minute reaper; a continental mosaic whose tasks run for half an hour wants a 15-minute one, because the reaper’s own overhead is otherwise larger than the work it recovers.

Frequently Asked Questions#

Related#

• Generating idempotent keys for shapefile uploads — the bundle-level recipe in full
• Idempotent PostGIS upserts for feature loads — row-level safety without a ledger
• Exponential backoff for API rate limits — the timing half of safe retries
• State management in geospatial flows — where the ledger sits in the wider state model
• Caching strategies for spatial tasks — the same digests, used for reuse rather than safety

← Resilience & Failure Handling for GIS Pipelines