Handling retries and dead-letter queues in async report pipelines

A single malformed inspection PDF should never be able to take down an overnight batch, yet a processor with no failure handling behaves exactly like that: a network blip while downloading a report raises an exception that either aborts the whole run or gets silently swallowed, and a genuinely corrupt file gets retried forever by a well-meaning loop until the municipal submission window closes. This page adds the missing failure-handling layer to the worker pool described in Async Batch Processing Pipelines: bounded exponential-backoff retries with jitter, a sha256-derived idempotency key so a retried or reprocessed report can never double-write a tower’s compliance record, and a dead-letter queue that captures the failure reason and an audit hash for anything that never recovers. By the end you will have a runnable asyncio module that turns transient failures into automatic recoveries and permanent failures into an inspectable queue instead of a stalled batch.

Prerequisites & Context #

Before extending an existing pipeline with the pattern below, have the following in place:

• Python 3.11+ for asyncio.sleep-based backoff and dataclasses with default factories.
• The base concurrency layer already running. This page assumes the semaphore-bound worker pool from Async Batch Processing for Multi-Site Structural Reports already exists; retries and the dead-letter queue wrap around each call to that extraction code, they do not replace it.
• A stable source_uri per report. Whether it is an S3 key, an SFTP path, or a municipal portal URL, the idempotency key is only as trustworthy as this identifier — if the same file can arrive under two different URIs, dedupe silently fails.
• Somewhere durable to hold dead letters. The example below keeps them in memory for clarity; production pipelines should persist dead letters the same way they persist the audit trail described in Automated Structural Report Parsing & Document Ingestion, since a dead-lettered report is itself a compliance-relevant event.
• A clear line between transient and permanent failure. A timed-out download and a structurally corrupt PDF are not the same problem, and treating them the same either wastes the whole retry budget on a file that will never parse, or gives up on a file that would have succeeded on the second attempt.

Step-by-Step Implementation #

Each step closes one specific way an unguarded pipeline stalls or double-processes a tower.

Step 1 — Derive an idempotency key before anything else. Compute idempotency_key = sha256(site_id + source_uri) the moment a task enters the processor, before the first extraction attempt. Because the key is derived only from identifiers and never from file contents, it stays stable across every retry of the same task, and a dead letter replayed hours later still maps to the exact same key — the property that makes a “process once” guarantee possible.

Step 2 — Wrap extraction in a try/except that separates failure categories. A worker calls the existing extraction routine inside a try block and catches two distinct shapes of failure: transient exceptions such as TimeoutError or ConnectionError that are worth retrying, and a custom PoisonMessageError for anything structurally unrecoverable — an unreadable PDF header, a report for a site_id that does not exist in the compliance database, or a page count of zero. Anything not explicitly recognized as transient should be treated as poison rather than silently retried; that default keeps unknown bugs from burning the entire retry budget before anyone notices.

Step 3 — Apply exponential backoff with jitter and a hard attempt cap. On a transient failure, sleep for base_delay * 2 ** (attempt - 1) plus a small random jitter before trying again, and stop after max_attempts. The jitter matters as much as the exponential curve: without it, every report from the same batch that failed on the same downstream outage retries in perfect lockstep and re-creates the exact spike that caused the outage.

Step 4 — Route exhausted or poisoned items to a dead-letter queue. When a report raises PoisonMessageError, or the retry loop reaches max_attempts without success, append it to a dead-letter queue entry carrying the failure reason and an audit_hash computed from the idempotency key and the reason string. That hash gives a compliance auditor a tamper-evident answer to “why didn’t TWR-8842’s Q2 report land” without needing the original file back.

Step 5 — Reprocess the dead-letter queue safely through the same idempotency check. When an upstream issue is fixed or a corrected file is re-uploaded under the same source_uri, replay dead letters through the identical process() path used for new tasks. Because the idempotency key is recomputed the same way and checked against a dedupe store before any extraction runs, a report that already succeeded through some other path is skipped rather than written twice, and a report that still fails lands back in the dead-letter queue with an updated reason instead of vanishing.

Figure: a task computes its idempotency key once, then loops through bounded exponential-backoff retries; PoisonMessageError and an exhausted retry budget both exit to the dead-letter queue, which replays safely back through the same idempotency check.

Complete Runnable Example #

The module below implements all five steps: an idempotency key derived with hashlib.sha256, a bounded retry loop with exponential backoff and jitter, a PoisonMessageError that skips straight to the dead-letter queue, and a reprocess_dead_letters method that replays failures through the same dedupe check.

import asyncio
import hashlib
import logging
import random
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("retry_dlq_processor")

class PoisonMessageError(Exception):
    """Raised when a report is structurally invalid and must never be retried."""

@dataclass
class ReportTask:
    site_id: str          # e.g. TWR-8842
    source_uri: str       # stable storage key, e.g. s3://reports/twr8842_q2.pdf
    idempotency_key: str = ""
    attempt: int = 0

@dataclass
class DeadLetter:
    task: ReportTask
    reason: str
    audit_hash: str

def make_idempotency_key(site_id: str, source_uri: str) -> str:
    return hashlib.sha256(f"{site_id}:{source_uri}".encode()).hexdigest()

class RetryingProcessor:
    def __init__(self, max_attempts: int = 4, base_delay: float = 0.4):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.processed: Dict[str, Any] = {}       # idempotency_key -> result, dedupe store
        self.dead_letters: List[DeadLetter] = []

    async def _extract(self, task: ReportTask) -> Dict[str, Any]:
        if "corrupt" in task.source_uri:
            raise PoisonMessageError(f"unparseable structure in {task.source_uri}")
        if random.random() < 0.55 and task.attempt < 3:
            raise TimeoutError(f"transient timeout fetching {task.source_uri}")
        return {"site_id": task.site_id, "pages": 12}

    def _to_dead_letter(self, task: ReportTask, reason: str) -> None:
        audit_hash = hashlib.sha256(f"{task.idempotency_key}:{reason}".encode()).hexdigest()
        self.dead_letters.append(DeadLetter(task, reason, audit_hash))
        logger.error("routed_to_dlq site=%s reason=%s hash=%s", task.site_id, reason, audit_hash[:12])

    async def process(self, task: ReportTask) -> Optional[Dict[str, Any]]:
        task.idempotency_key = make_idempotency_key(task.site_id, task.source_uri)
        if task.idempotency_key in self.processed:
            logger.info("skip_duplicate site=%s key=%s", task.site_id, task.idempotency_key[:12])
            return self.processed[task.idempotency_key]

        while task.attempt < self.max_attempts:
            task.attempt += 1
            try:
                result = await self._extract(task)
                self.processed[task.idempotency_key] = result
                logger.info("report_succeeded site=%s attempt=%d key=%s",
                            task.site_id, task.attempt, task.idempotency_key[:12])
                return result
            except PoisonMessageError as exc:
                self._to_dead_letter(task, str(exc))
                return None
            except Exception as exc:
                delay = self.base_delay * (2 ** (task.attempt - 1)) + random.uniform(0, 0.25)
                logger.warning("report_retry site=%s attempt=%d delay=%.2f error=%s",
                               task.site_id, task.attempt, delay, exc)
                await asyncio.sleep(delay)

        self._to_dead_letter(task, f"exhausted {self.max_attempts} attempts")
        return None

    async def reprocess_dead_letters(self) -> None:
        pending, self.dead_letters = self.dead_letters, []
        for entry in pending:
            entry.task.attempt = 0
            await self.process(entry.task)

if __name__ == "__main__":
    async def main() -> None:
        proc = RetryingProcessor(max_attempts=4)
        batch = [
            ReportTask("TWR-8842", "s3://reports/twr8842_q2.pdf"),
            ReportTask("TWR-9017", "s3://reports/twr9017_corrupt.pdf"),
        ]
        await asyncio.gather(*(proc.process(t) for t in batch))
        print(f"dead-lettered: {len(proc.dead_letters)}")
        await proc.reprocess_dead_letters()
        print(f"dead-lettered after replay: {len(proc.dead_letters)}")

    asyncio.run(main())

Verification & Expected Output #

Running the module produces one structured line per attempt, a routed_to_dlq line for the poisoned file, and a summary before and after replay:

2026-03-18 09:41:02 | WARNING | report_retry site=TWR-8842 attempt=1 delay=0.51 error=transient timeout fetching s3://reports/twr8842_q2.pdf
2026-03-18 09:41:02 | ERROR | routed_to_dlq site=TWR-9017 reason=unparseable structure in s3://reports/twr9017_corrupt.pdf hash=7c1e9a2f0b6d
2026-03-18 09:41:03 | INFO | report_succeeded site=TWR-8842 attempt=2 key=41c8de77a0b9
dead-lettered: 1
dead-lettered after replay: 1

TWR-8842 recovers on its second attempt because the injected failure is transient; TWR-9017 never retries at all because _extract raises PoisonMessageError on the first call, and the replay at the end confirms the dead letter is still genuinely poisoned rather than silently dropped. To assert this behavior in a test, check outcomes rather than exact timing: assert proc.processed contains the successful key, assert len(proc.dead_letters) == 1, and assert proc.dead_letters[0].task.site_id == "TWR-9017". If a failure surfaces as a report succeeding twice in the log, that means the idempotency check ran after extraction instead of before it — move the self.processed lookup back to the top of process().

Gotchas & Edge Cases #

• Overly broad except clauses reclassify poison as transient. If the retry loop’s except Exception block sits above the except PoisonMessageError block instead of below it, every poisoned file burns its entire retry budget in backoff delay before finally reaching the dead-letter queue. Order the specific exception first and keep the transient catch-all narrow — ideally naming TimeoutError, ConnectionError, and a handful of known-retryable types rather than bare Exception.
• A re-uploaded, corrected file needs a deliberate versioning decision. Because the idempotency key is derived from source_uri, a genuinely corrected report saved under a new date-stamped filename gets a new key and processes as a fresh task — usually what you want. But if the correction is saved back to the same source_uri, the old dead-letter entry’s key still matches, and replaying it will pick up the fixed file automatically; overwriting a source_uri in place is the trigger for that path, so treat it as a policy choice, not an accident.
• Fixed jitter beats no jitter, but a growing dead-letter queue with no reprocessing cadence beats neither. Exponential backoff with jitter prevents synchronized retry storms, but a dead-letter queue is not self-healing — without a scheduled call to reprocess_dead_letters (hourly, or triggered on upstream recovery), sites accumulate in the queue and a compliance deadline can pass with the report sitting there, technically preserved but effectively still missing.

FAQ #

Related #

• Up to the parent architecture: Async Batch Processing Pipelines
• Sibling task: Async Batch Processing for Multi-Site Structural Reports
• The parent architecture: Automated Structural Report Parsing & Document Ingestion