Reprocessing Dead-Letter Geotasks Safely

Reprocessing a dead-letter geotask safely means replaying each stored failure at most a capped number of times, spacing attempts with backoff, and de-duplicating by geometry hash so a replay never double-writes a feature that partially succeeded before. The reconciliation job claims a batch of pending rows, reprojects and re-runs each one idempotently, bumps the attempt counter, and moves permanently-failed rows to a quarantined state instead of looping on them forever. This page gives you a complete, idempotent replay loop over a PostGIS DLQ.

It is the draining half of Dead-Letter Queues for Failed Geotasks and consumes exactly the rows written by storing failed geometries in a PostGIS dead-letter queue.

When to Use This Pattern#

Run a reconciliation replay job when:

• Failures accumulate in a durable DLQ and you need them cleared without manual re-submission.
• The original failure was transient (a WFS timeout, a rate limit) and the feature will likely succeed now.
• You must guarantee that replaying a half-completed geotask does not insert a feature twice into the target table.
• You want permanently-broken payloads (corrupt geometry, unknown CRS) removed from the active queue and parked for human review.

Core Principles#

Four rules keep a replay job from doing harm:

1. Cap attempts. Every row carries an attempts counter. Past a ceiling (say 5), the row is quarantined, never retried again. This prevents an unfixable geometry from consuming compute forever.
2. Back off between attempts. Do not replay a row the instant it fails again. Gate re-selection on last_seen plus a growing delay, so a still-degraded downstream service is not hammered — the same discipline as exponential backoff for API rate limits.
3. Deduplicate by geometry hash. The write into the target table must be idempotent on the stored geom_hash. If a prior partial run already wrote the feature, the replay upsert is a no-op rather than a duplicate. This is the DLQ-local application of idempotency keys in spatial ETL.
4. Claim rows atomically. Use SELECT ... FOR UPDATE SKIP LOCKED so multiple reconciliation workers can drain the same DLQ without processing the same row twice.

Complete Working Example#

The loop claims a bounded batch of eligible rows, replays each through the real processing function, and transitions status based on the outcome. Reprojection uses the row’s stored EPSG so the geometry is handled in the correct CRS.

from __future__ import annotations

import logging
from typing import Any

import psycopg
from psycopg.rows import dict_row
from pyproj import CRS, Transformer
from shapely import wkb
from shapely.ops import transform as shp_transform
from shapely.geometry.base import BaseGeometry

logger = logging.getLogger(__name__)

MAX_ATTEMPTS = 5
BACKOFF_BASE_SECONDS = 30  # multiplied by attempts for the re-eligibility gate
TARGET_CRS = CRS.from_epsg(3857)


def claim_batch(conn: psycopg.Connection, limit: int = 50) -> list[dict[str, Any]]:
    """Atomically claim pending rows that have cooled past their backoff."""
    sql = """
        SELECT id, task_id, ST_AsBinary(geom) AS wkb, epsg, geom_hash, attempts
        FROM failed_geotasks
        WHERE status = 'pending'
          AND attempts < %(max_attempts)s
          AND last_seen < now() - (%(backoff)s * attempts) * interval '1 second'
        ORDER BY first_seen
        FOR UPDATE SKIP LOCKED
        LIMIT %(limit)s;
    """
    with conn.cursor(row_factory=dict_row) as cur:
        cur.execute(
            sql,
            {"max_attempts": MAX_ATTEMPTS, "backoff": BACKOFF_BASE_SECONDS, "limit": limit},
        )
        return cur.fetchall()


def reproject(geom: BaseGeometry, source_epsg: int) -> BaseGeometry:
    source = CRS.from_epsg(source_epsg)
    if source == TARGET_CRS:
        return geom
    transformer = Transformer.from_crs(source, TARGET_CRS, always_xy=True)
    return shp_transform(transformer.transform, geom)


def upsert_feature(conn: psycopg.Connection, geom: BaseGeometry, geom_hash: str) -> None:
    """Idempotent write: geom_hash uniqueness makes a replay a no-op."""
    sql = """
        INSERT INTO processed_features (geom, geom_hash)
        VALUES (ST_SetSRID(ST_GeomFromWKB(%(wkb)s), 3857), %(hash)s)
        ON CONFLICT (geom_hash) DO NOTHING;
    """
    with conn.cursor() as cur:
        cur.execute(sql, {"wkb": geom.wkb, "hash": geom_hash})


def replay_dlq(conn: psycopg.Connection) -> None:
    rows = claim_batch(conn)
    for row in rows:
        geom = wkb.loads(bytes(row["wkb"]))
        try:
            projected = reproject(geom, row["epsg"])
            upsert_feature(conn, projected, row["geom_hash"])
        except Exception as exc:  # replay failed again
            new_attempts = row["attempts"] + 1
            status = "quarantined" if new_attempts >= MAX_ATTEMPTS else "pending"
            conn.execute(
                "UPDATE failed_geotasks "
                "SET attempts = %s, last_seen = now(), status = %s, error = %s "
                "WHERE id = %s",
                (new_attempts, status, str(exc), row["id"]),
            )
            logger.warning("replay failed id=%s -> %s: %s", row["id"], status, exc)
        else:
            conn.execute(
                "UPDATE failed_geotasks SET status = 'replayed', last_seen = now() "
                "WHERE id = %s",
                (row["id"],),
            )
            logger.info("replayed id=%s task=%s", row["id"], row["task_id"])
    conn.commit()  # single commit releases the FOR UPDATE locks for the batch

The eligibility filter does the backoff arithmetic in SQL: a row on its third attempt is not re-selected until 90 seconds after its last try (BACKOFF_BASE_SECONDS * attempts). ON CONFLICT (geom_hash) DO NOTHING is the linchpin — if an earlier crash wrote the feature before the DLQ row was marked replayed, the retry sees the conflict and skips the insert, so no duplicate geometry lands in processed_features.

Parameter & Option Reference#

Verification & Testing#

Assert the two behaviors that make replay safe — de-duplication and quarantine on exhaustion:

def test_replay_is_idempotent(conn) -> None:
    seed_dlq_row(conn, task_id="t1", geom_hash="abc", epsg=4326)
    replay_dlq(conn)
    replay_dlq(conn)  # second pass must not double-write
    n = conn.execute(
        "SELECT count(*) FROM processed_features WHERE geom_hash = 'abc'"
    ).fetchone()[0]
    assert n == 1


def test_exhausted_row_is_quarantined(conn) -> None:
    seed_dlq_row(conn, task_id="t2", geom_hash="bad", epsg=9999)  # invalid EPSG
    for _ in range(MAX_ATTEMPTS + 1):
        force_backoff_elapsed(conn, "t2")
        replay_dlq(conn)
    status = conn.execute(
        "SELECT status FROM failed_geotasks WHERE task_id = 't2'"
    ).fetchone()[0]
    assert status == "quarantined"

Operationally, watch the queue drain with a status rollup, and confirm nothing is stuck cycling:

SELECT status, count(*) AS n, max(attempts) AS worst
FROM failed_geotasks
GROUP BY status
ORDER BY status;

A healthy job shows pending shrinking, replayed growing, and quarantined holding only genuinely unfixable rows. A pending count that never falls while attempts climbs means your backoff gate or downstream dependency needs attention.

Common Pitfalls#

• Non-idempotent target writes. Without ON CONFLICT (geom_hash), a replay after a partial success inserts the feature twice and corrupts spatial joins downstream.
• No attempt ceiling. A corrupt geometry or unknown EPSG replayed forever burns compute and never drains; cap attempts and quarantine.
• Replaying without backoff. Immediately retrying a row whose downstream is still degraded re-fails it instantly and inflates the attempt counter; gate on last_seen.
• Concurrent workers without SKIP LOCKED. Two reconciliation workers claiming the same rows double-process them; lock and skip so each row is drained once.
• Reprojecting with the wrong source CRS. Ignoring the row’s stored epsg and assuming a global source projection silently mis-places features; always transform from the recorded EPSG.

Related: This job drains rows created by storing failed geometries in a PostGIS dead-letter queue, reuses the spacing logic of exponential backoff for API rate limits, depends on idempotency keys in spatial ETL for duplicate-free writes, and complements circuit breakers for external WMS services that stop failures reaching the DLQ in the first place — all within Resilience & Failure Handling for GIS Pipelines.

← Back to Dead-Letter Queues for Failed Geotasks