Article

Making Spatial Tasks Idempotent for Safe Retries

Write to a name nobody else can collide with, promote it to the final name atomically, and record completion only after the promotion. A task that runs twice then produces the same bytes at the same path, with no window in which a consumer can observe a half-written raster. It builds on the Distributed Task Queues for Spatial Jobs guide, part of the broader Spatial Batch Processing & Async Workflows reference.

Prerequisites

  • Python 3.10 or later
  • pip install rasterio and whichever queue you run
  • Storage that supports either an atomic rename (a POSIX filesystem) or a conditional write (most object stores)

What “Runs Twice” Actually Looks Like

Duplicate execution is not a rare pathology; it is the normal consequence of the delivery guarantee you want. Three ordinary events cause it.

Three ordinary causes of a second run A worker killed between finishing the work and acknowledging the message leaves the message on the queue. A task slower than the visibility timeout is redelivered while the first attempt is still running. An operator re-running a failed batch re-enqueues units that already succeeded. killed before ack work finished output written then SIGKILL the message returns and is delivered again visibility timeout task takes 20 minutes timeout is 10 both run at once the worst case — two writers, same instant operator re-run batch failed at 80% someone re-enqueues all of it the common case, and the easiest to overlook

The middle column is the one that produces corruption rather than waste, because two processes write the same destination concurrently. Everything below is aimed at making that specific case safe.

Complete Working Implementation

# gistools/idempotent.py
from __future__ import annotations

import hashlib
import os
import tempfile
from contextlib import contextmanager
from pathlib import Path


def unit_key(**parts) -> str:
    """A stable identifier for a unit of work, from its inputs alone."""
    blob = "|".join(f"{k}={parts[k]!r}" for k in sorted(parts))
    return hashlib.sha256(blob.encode()).hexdigest()[:16]


@contextmanager
def atomic_output(final: Path):
    """Yield a temporary path; promote it to `final` only on clean exit."""
    final.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=final.parent, prefix=final.name + ".",
                               suffix=".tmp")
    os.close(fd)
    tmp = Path(tmp)
    try:
        yield tmp
        os.replace(tmp, final)          # atomic within one filesystem
    except BaseException:
        tmp.unlink(missing_ok=True)     # leave nothing behind on failure
        raise


def already_done(final: Path, expected_key: str) -> bool:
    """True if a previous run produced this exact output."""
    marker = final.with_suffix(final.suffix + ".key")
    return final.exists() and marker.exists() and \
           marker.read_text().strip() == expected_key

Using it inside a task:

import rasterio
from rasterio.windows import Window

from gistools.idempotent import atomic_output, already_done, unit_key


def warp_window(src: str, final: str, dst_crs: str,
                col_off: int, row_off: int, width: int, height: int) -> str:
    final = Path(final)
    key = unit_key(src=src, dst_crs=dst_crs, col=col_off, row=row_off,
                   w=width, h=height, version=2)

    if already_done(final, key):
        return str(final)               # a second delivery costs nothing

    with atomic_output(final) as tmp:
        with rasterio.open(src) as s:
            window = Window(col_off, row_off, width, height)
            profile = s.profile | {"crs": dst_crs, "width": width,
                                   "height": height,
                                   "transform": s.window_transform(window)}
            with rasterio.open(tmp, "w", **profile) as d:
                d.write(s.read(window=window))

    final.with_suffix(final.suffix + ".key").write_text(key)
    return str(final)

Step Annotations

  1. unit_key includes a version field. Bumping it invalidates every previously-written output without deleting anything, which is what you want when the transform logic changes and old outputs are no longer correct.
  2. tempfile.mkstemp in the destination directory. Creating the temporary alongside the final file guarantees the rename stays within one filesystem — os.replace across filesystems is not atomic and, on some platforms, not permitted.
  3. The marker is written after the rename, not before. A crash between the two leaves a valid output with no marker, so the next run redoes the work — wasteful and correct. The reverse order would claim completion for a file that does not exist.
  4. except BaseException rather than except Exception. A KeyboardInterrupt or a CancelledError must also clean up the temporary file, and neither derives from Exception.
  5. already_done checks both the output and the marker. Someone deleting the output but not the marker — or the reverse — should cause a rebuild, not a false skip.

One Named Gotcha: Shapefiles Have Five Files and One Name

An atomic rename works on a single file. A shapefile is at least three and often five, and renaming them one at a time leaves a window in which a reader sees a .shp with no .dbf — which GDAL will open and report as having no attributes rather than failing.

Two ways out, and the first is strongly preferable:

# 1. Write a single-file format instead. GeoPackage and FlatGeobuf are one file each.
final = final.with_suffix(".gpkg")

# 2. If a shapefile is genuinely required, rename the directory, not the files.
with atomic_output_dir(final.parent / final.stem) as tmpdir:
    write_shapefile(tmpdir / "layer.shp", gdf)
# os.replace on a directory is atomic on POSIX, provided the target does not exist

The directory approach works but is fiddly: the target must not already exist, so a re-run has to remove the previous directory first, which briefly leaves nothing at the destination. If anything downstream reads that path continuously, the single-file format is the only clean answer.

The ordering of the three steps is the whole guarantee, and each possible crash point leaves a state that is either correct or merely wasteful — never wrong.

Every crash point leaves a safe state Crashing before the write leaves nothing. During the write leaves an orphan temporary file that a sweep removes. After the rename leaves valid output with no marker, so the unit is redone. After the marker, the unit is complete and a redelivery is a no-op. start writing tmp renamed marker written nothing exists redo it an orphan .tmp sweep and redo valid, unmarked redo — wasteful only complete redelivery is free No crash point leaves a partial file at the destination name, which is the property consumers rely on.

Reversing steps three and four — recording completion before promoting the file — moves the third column from wasteful to wrong, because a later run would skip a unit whose output does not exist.

Verification

# Run the same unit twice; the second should be a no-op and the output unchanged.
python -c "from gistools.tasks import warp_window as w;            a=w('in.tif','out/part.tif','EPSG:3857',0,0,256,256);            import hashlib,pathlib;            h1=hashlib.sha256(pathlib.Path(a).read_bytes()).hexdigest();            w('in.tif','out/part.tif','EPSG:3857',0,0,256,256);            h2=hashlib.sha256(pathlib.Path(a).read_bytes()).hexdigest();            print('identical' if h1==h2 else 'DIFFERENT')"

# Kill mid-write and confirm nothing partial survives.
timeout 0.3 python -c "from gistools.tasks import warp_window as w;                        w('big.tif','out/part.tif','EPSG:3857',0,0,8192,8192)"
ls out/ | grep -c '\.tmp$'      # expect 0 — the context manager cleaned up
test -f out/part.tif && echo "UNEXPECTED partial output" || echo "clean"

The second block is the important one. A pipeline that passes the first check and fails the second is not idempotent — it is merely deterministic, and it will still corrupt an output when a worker is reclaimed at the wrong moment.

Idempotency for the Other Output Types

The temporary-then-promote pattern covers files. Two other things a spatial task commonly writes need their own treatment.

Making each kind of side effect safe to repeat A file is made idempotent by writing to a temporary name and promoting atomically. A database write is made idempotent by an upsert keyed on the unit identifier. An appended record is made idempotent by carrying the unit key so duplicates can be collapsed downstream. a raster or vector file temporary key, then an atomic promote covered above a row in PostGIS upsert on the unit key, never a bare insert one row per unit an appended log or metric carry the unit key so duplicates collapse dedupe downstream

The middle row is the one that silently produces duplicates. A bare insert in a retried task leaves two rows describing the same window, and nothing errors — the count is simply wrong from then on.

Testing Idempotency Deliberately

Idempotency is a property that holds until someone changes the write path, and the only way to keep it is a test that fails when it breaks. Two tests cover it.

The first runs the same unit twice and asserts the output is byte-identical and that no temporary files remain. It catches the ordinary regression: a write that stopped going through the temporary path, or a marker written before the promotion.

The second is more valuable and slightly harder: kill the process partway through the write and assert that the destination is either absent or complete, never partial. A subprocess with a short timeout does it, and the assertion is a simple existence-and-openability check.

def test_partial_write_leaves_nothing(tmp_path):
    out = tmp_path / "part.tif"
    subprocess.run([sys.executable, "-c", WRITE_BIG_SNIPPET, str(out)],
                   timeout=0.4, check=False)
    assert not out.exists() or rasterio.open(out).width > 0
    assert not list(tmp_path.glob("*.tmp"))

The not out.exists() or ... shape is deliberate: both outcomes are correct, and asserting only one of them makes the test flaky on a fast machine.

Versioning the Output

The version field in the unit key is worth more thought than it usually gets. It is what lets you invalidate previously-written outputs when the logic that produced them changes, without deleting anything.

Bumping it means the next run treats every output as absent and rebuilds. Not bumping it when the logic changed means a resumed run mixes outputs from two versions of the code, which is the kind of inconsistency that surfaces months later in an analysis nobody can reproduce.

The rule that works is to bump the version whenever the bytes a unit would produce could differ: a changed resampling method, a different compression setting, a fixed bug in the transform. Not for a refactor that provably produces identical output, and not for a change to logging or error handling. Recording the version in the output’s metadata as well as in the key makes the question answerable afterwards, which is worth the one extra tag.

Idempotency Is a Property of the Whole Unit

It is easy to make the write idempotent and leave the rest of the unit with side effects that are not. Three are common in spatial pipelines and each needs its own treatment.

A row inserted into a catalogue table is the most frequent. A bare insert in a retried task leaves two rows describing one window, and nothing errors — the count is simply wrong from then on. An upsert keyed on the unit identifier fixes it, and the key must be the same one the output uses so the two cannot disagree.

A metric incremented at the end of a unit double-counts under retry. Counting from the records afterwards rather than incrementing in the task removes the problem entirely, and it is more accurate anyway because a task killed after incrementing but before finishing would otherwise count.

A notification — an email, a webhook, a message onto another queue — is the one with no clean solution inside the task. Sending it from a separate step that runs once after the batch completes, rather than from each unit, is the arrangement that works. It also produces better notifications: one message saying what happened, rather than ten thousand saying that a tile finished.

Cleaning Up After Interrupted Runs

Temporary files and keys accumulate whenever a process dies at the wrong moment, and a pipeline that never sweeps them will eventually fill a volume or a bucket.

On a filesystem, a sweep at start-up is sufficient: remove anything matching the temporary pattern that is older than the longest a unit could reasonably take. Using an age threshold rather than removing everything matters, because a concurrent run may be legitimately holding a temporary file right now.

On an object store, a lifecycle rule is better than a sweep, because it runs whether or not your pipeline does. A rule expiring objects whose key matches the temporary suffix after a day costs nothing and removes the whole category. The equivalent rule for aborting incomplete multipart uploads is worth adding at the same time, since those are invisible in a listing and still billed.

Both are the kind of thing that gets written after the first incident. Writing them at the start costs ten minutes and removes an incident.

Verifying Idempotency in Production

Testing proves the property holds for the cases you thought of. A cheap production check proves it is still holding.

The check is a comparison between the number of units the batch believes it completed and the number of distinct outputs that exist. They should be equal. A count of outputs lower than the completed count means something recorded completion without producing a file; higher means something produced a file that was not recorded — usually a leftover from an earlier run under a different key scheme.

Running that comparison at the end of every batch and failing the run when they disagree turns a silent divergence into an immediate one. It costs one listing and one query, and it is the only check that would catch a subtly broken key function, which is the failure mode that survives every unit test because the key is consistent — just not unique.