Geospatial Portal

Isolating Poison Messages in Ingestion Queues

This guide keeps a single unprocessable record from occupying an ingestion fleet indefinitely, and makes the records that failed triageable rather than merely lost. It belongs to Scaling Celery Pipelines for Bulk Ingestion, within the Metadata Catalog Automation & Ingestion Workflows framework.

Prerequisites

  • A Celery deployment with a broker that supports message expiry and a dead-letter path.
  • Idempotent tasks, so a redelivered message is safe to process again.
  • Structured task logging with a run identifier, per aggregating Celery logs with Loki.
  • Somewhere durable to keep failed payloads, which is not the queue.

Not every failure should be retried

A retry policy that treats all exceptions alike produces the pathology this guide exists to prevent: a record that can never succeed, retried forever, consuming a worker slot on every attempt. Sorting failures by whether a retry could plausibly help is the whole of the fix.

Three failure classes, three retry policies Transient, permanent and ambiguous failures with the retry behaviour appropriate to each and the reason. FAILURE EXAMPLES POLICY transient the input is fine upstream timeout, connection reset, rate limit, database failover retry with backoff and jitter permanent the input is the problem unparseable XML, invalid geometry, missing identifier, wrong encoding no retry — dead-letter at once ambiguous an unexpected exception anything the code did not anticipate the class cannot be inferred safely a few bounded retries, then dead-letter

Step-by-step implementation

1. Classify failures explicitly in the task 

# tasks.py — the exception hierarchy IS the retry policy.
from celery import Task
from celery.exceptions import Reject

class PermanentIngestError(Exception):
    """The input cannot succeed, however many times it is tried."""

class TransientIngestError(Exception):
    """The input is fine; something else was temporarily unavailable."""

@app.task(
    bind=True,
    acks_late=True,                  # a killed worker returns the message
    reject_on_worker_lost=True,
    autoretry_for=(TransientIngestError,),
    retry_backoff=5, retry_backoff_max=600, retry_jitter=True,
    max_retries=6,
)
def ingest_record(self, payload: dict):
    try:
        record = parse(payload)      # raises PermanentIngestError on bad input
    except PermanentIngestError as exc:
        dead_letter(payload, self.request, exc)
        raise Reject(exc, requeue=False)     # never comes back to this queue

    try:
        upsert(record)
    except UpstreamUnavailable as exc:
        raise TransientIngestError(str(exc)) from exc

2. Make the dead-letter record triageable

A dead-letter queue holding bare payloads is a bin. What makes it useful is the context that lets somebody decide what to do without reproducing the failure.

# deadletter.py — everything a triage decision needs, in one row.
def dead_letter(payload: dict, request, exc: Exception) -> None:
    db.execute(
        """INSERT INTO ingest_dead_letter
             (task_id, task_name, run_id, source_id, source_native_id,
              attempts, error_type, error_message, payload, first_failed_at)
           VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s, now())
           ON CONFLICT (task_id) DO UPDATE
             SET attempts = ingest_dead_letter.attempts + 1,
                 error_message = EXCLUDED.error_message""",
        (request.id, request.task, payload.get("run_id"),
         payload.get("source_id"), payload.get("identifier"),
         request.retries + 1, type(exc).__name__, str(exc)[:2000],
         json.dumps(payload)),
    )

3. Bound what a poison message can consume 

# Two independent bounds, because either alone is insufficient.
app.conf.update(
    # A task that hangs must not hold a worker slot indefinitely.
    task_time_limit=300,          # hard kill
    task_soft_time_limit=240,     # a chance to clean up and dead-letter

    # A message older than the run it belongs to is no longer useful.
    task_default_queue="ingest",
    broker_transport_options={"visibility_timeout": 600},
)

@app.task(bind=True, soft_time_limit=240)
def ingest_record(self, payload):
    try:
        ...
    except SoftTimeLimitExceeded as exc:
        dead_letter(payload, self.request, exc)   # record it before the hard kill
        raise Reject(exc, requeue=False)

4. Watch for the pattern, not only the count

Reading the shape of the dead-letter queue Trickle, single-source spike and all-source spike, each with the likely cause and the response. a steady trickle spread across many sources stable over weeks this is data quality route it to the quality report, not to an alert one source spikes concentrated, sudden same error type throughout that source changed something a schema, an encoding, a field — contact them with examples everything spikes all sources at once starting at a deployment the pipeline changed roll back first, investigate after — this one is yours

5. Replay deliberately, and never in a loop 

# replay.py — replay is an operator action with an explicit scope and a cap.
def replay(source_id: str | None = None, error_type: str | None = None,
           limit: int = 500, dry_run: bool = True) -> int:
    rows = db.query(
        """SELECT task_id, payload FROM ingest_dead_letter
            WHERE replayed_at IS NULL
              AND (%s IS NULL OR source_id = %s)
              AND (%s IS NULL OR error_type = %s)
            ORDER BY first_failed_at
            LIMIT %s""",
        (source_id, source_id, error_type, error_type, limit))

    if dry_run:
        print(f"would replay {len(rows)} messages"); return len(rows)

    for task_id, payload in rows:
        # Mark BEFORE dispatching: a replay that fails again must not be
        # picked up by the next replay run, or the loop never terminates.
        db.execute("UPDATE ingest_dead_letter SET replayed_at = now() WHERE task_id = %s",
                   (task_id,))
        ingest_record.apply_async(args=[payload], queue="ingest-replay")
    return len(rows)

Dispatching replays to a separate queue matters: a replay of ten thousand records should not compete with tonight’s harvest for the same workers, and keeping it separate means the replay can be paused without pausing ingestion.

Where the queue’s own limits belong

Task-level policy handles individual messages; two broker-level settings bound what a pathological run can do to the system as a whole, and they are easy to leave at defaults that assume a much smaller workload.

Three bounds that protect the shared broker Queue length, message time-to-live and per-queue consumer limits, each with what it protects. queue length bounds broker memory without it, one runaway producer takes down every queue on the broker overflow to dead-letter message TTL discards stale work a task from a superseded run is worse than useless once a newer run exists set it from the run cadence consumers per queue bounds fleet capture a backlog on one queue must not consume workers the others need reserve a share per queue

The middle one is specific to ingestion and often overlooked: when a harvest is re-run because the first attempt was wrong, the tasks from the first run are still queued and will happily process stale payloads after the corrected ones. A time-to-live shorter than the interval between runs makes that impossible.

Verification 

# 1. A permanently bad record is dead-lettered on the first attempt
python3 -c "from tasks import ingest_record; ingest_record.delay({'identifier':None})"
psql -Atc "SELECT attempts, error_type FROM ingest_dead_letter ORDER BY first_failed_at DESC LIMIT 1;"
#   expect: attempts=1, a permanent error type

# 2. A transient failure is retried with backoff, not dead-lettered immediately
#    (simulate an upstream outage, then restore it)
psql -Atc "SELECT count(*) FROM ingest_dead_letter WHERE error_type='TransientIngestError';"
#   expect: 0 while the upstream is briefly down

# 3. A hung task is killed and recorded rather than holding a worker
celery -A app inspect active | grep -c ingest_record
#   expect: no task older than the soft time limit

# 4. A replay does not loop
python3 -c "import replay; replay.replay(source_id='county-highways', dry_run=False)"
python3 -c "import replay; print(replay.replay(source_id='county-highways'))"
#   expect: 0 on the second call — already-replayed rows are excluded

# 5. Dead-letter volume is visible per source and per error type
psql -Atc "SELECT source_id, error_type, count(*) FROM ingest_dead_letter
           WHERE first_failed_at > now() - interval '1 day'
           GROUP BY 1,2 ORDER BY 3 DESC LIMIT 10;"

Troubleshooting matrix

Symptom Likely cause Fix
A single record consumes the fleet Blanket autoretry_for=(Exception,) with unlimited retries Classify exceptions; dead-letter permanent failures immediately
The dead-letter table grows without triage No owner and no report Report per source and per error type; route to the quality process
A replay re-dead-letters the same rows forever Rows not marked before dispatch Mark first, then dispatch; exclude replayed rows from the next run
Workers stop consuming during a replay Replay dispatched to the ingestion queue Use a separate queue so replay can be throttled or paused
Messages vanish with no dead-letter entry Reject without recording, or an unhandled kill Record before rejecting; handle the soft time limit explicitly
Retries hammer a recovering upstream Backoff without jitter Enable jitter; cap the maximum backoff
A deployment produces thousands of dead letters The pipeline changed, not the data Roll back first; the all-source spike is the signal

FAQ

Should the dead-letter store be a queue or a table?

A table. A queue is designed for things that will be consumed shortly, and dead letters are consumed by a human decision that may be days away — often never, for records that are simply bad. A table can be queried, grouped, reported and joined against provenance, which is what triage requires.

How long should dead letters be retained?

Long enough to spot a pattern and act on it: a quarter is a reasonable default, with the payload retained for the same period so a replay is possible. Beyond that, keep the counts and error types for trend reporting and discard the payloads, which are the bulk of the storage.

What if the whole batch is bad?

Stop the run rather than dead-lettering every record individually. A rejection ratio above a threshold in the first minutes of a run is a strong signal that the mapping or the source changed, and continuing produces tens of thousands of dead letters that all have one cause — which is harder to see than a stopped run with a clear reason.

Should failures be reported to the source?

Yes, in aggregate and with examples, on the quality-report cadence rather than as they happen. “Forty records failed with an unparseable date, here are three” is actionable; a stream of individual failure notifications is not, and it is the fastest way for a partner’s technical contact to filter your messages.

Does acks_late risk processing a record twice?

Yes, and that is the correct trade when tasks are idempotent — which they must be for a bulk pipeline regardless. Processing a record twice writes the same row twice with the same result; losing a record silently is a gap nobody detects. If a task genuinely cannot be made idempotent, that is the problem to fix first.

How does this interact with the run-level counters?

Dead letters should be counted against the run’s rejected total, so the run’s terminal record reflects them, and they should be reported per source afterwards. A run that reports complete success while a thousand records sit in the dead-letter table has produced a comforting number and a wrong one.

Up one level: Scaling Celery Pipelines for Bulk Ingestion.