Building a Dead-Letter Queue for Failed Grant Payloads

Route grant payloads that exhaust their retry budget to a dead-letter queue with full context: DLQ record schema, SHA-256 lineage, poison-message detection, replay tooling, and depth metrics under 2 CFR §200.334.

This guide is part of the Error Categorization & Logging section within the broader Data Ingestion & Grant Parsing Workflows framework, and it solves one narrow problem: when a grant payload has burned through every retry and still cannot be processed, where does it go so that it is neither replayed forever nor silently dropped?

The wrong answer is a logger.error(...) followed by a discarded message. A dropped payload is a lost federal submission, and under 2 CFR §200.334 the record and its provenance are things you are obligated to retain, not free to delete. A dead-letter queue (DLQ) is the durable holding tier for exactly these payloads: it captures the original bytes, the error that stopped them, the number of attempts already spent, and enough audit context that an operator can fix the root cause and replay the payload with its lineage intact. This guide builds that DLQ, its move-in trigger, its poison-message guard, and its replay path.

When to Use This Approach

Reach for a dead-letter queue when all three of these hold:

  • A payload can fail permanently, not just transiently. Backoff and re-attempt logic — which lives in Pipeline Fallback & Retry Logic — handles the transient case: a 503 from a grant portal, a lock timeout, a brief network partition. The DLQ is the terminal sink for the payload that is still failing after that retry budget is spent, plus the payload whose error is non-retryable on the first look (a schema violation will never succeed no matter how many times you replay it).
  • The failed artifact is a retained record. Grant payloads carry financial and program data that 2 CFR §200.334 requires be kept for at least three years after the final expenditure report. A dead-lettered payload is not garbage; it is a record awaiting remediation, so the DLQ store must be as durable and auditable as your primary datastore.
  • Operators need to replay, not re-submit by hand. When the bug is fixed — a corrected field map, a widened parser — you want to re-enqueue the exact original bytes, not ask a grant manager to re-key the application. That requires the DLQ to preserve the payload verbatim and its source digest.

Deciding how many retries to allow and what backoff curve to use is out of scope here and belongs to Choosing Retry Budgets and Backoff for Grant API Failures. Classifying an error as retryable versus terminal in the first place is the job of Implementing Automated Error Logging for Grant Pipelines; this page consumes that classification and acts on it.

Dead-letter path with a poison-message branch and a lineage-preserving replay loop A pipeline consumer processes a grant payload. A decision node asks whether the retry budget is exhausted or the error is non-retryable. If not — the error is retryable — the payload loops back through a backoff retry into the consumer. If yes, the payload moves into the DLQ store, a durable record holding the original payload plus source digest, the error category and attempt count, the first-seen and last-seen timestamps, and the audit context. After an operator fixes the root cause, a replay tool reads the DLQ record and re-enqueues it into the pipeline with lineage preserved. A separate branch handles poison messages: when a record's own replays keep failing a threshold number of times, it is moved to a poison quarantine instead of being replayed again. Exhausted retries land in the DLQ — never in a dropped log line Every terminal payload is retained with full context, replayable after a fix, and guarded against poison loops. exhausted after fix retryable replay fails N× Consumer grant payload budget spent / non-retryable? DLQ store original payload + source digest error category · attempt count first-seen · last-seen audit context (actor · trace id) Replay tool re-enqueue · lineage kept Pipeline re-processed from consumer Retry with backoff within retry budget Poison quarantine held for human triage

Step-by-Step Implementation

The reference implementation targets Python 3.10+ and uses only the standard library plus a Postgres driver for the store; nothing here needs a message-broker-specific SDK, because the DLQ contract is about the record, not the transport. Swap the persistence calls for your broker’s native DLQ if you run on SQS or RabbitMQ — the schema, the trigger, and the replay logic stay identical.

Step 1: Define the dead-letter record

The DLQ is only as useful as the context each record carries. Model it as a frozen dataclass so a dead-lettered payload is immutable once written: the original bytes, the SHA-256 digest of those bytes, the error category, how many attempts were already spent, first- and last-seen timestamps, and the audit context (who or what was processing it, and the trace id that ties it to upstream logs).

python
import logging
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

AUDIT_LOGGER = logging.getLogger("grant_pipeline.dlq")
AUDIT_LOGGER.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
AUDIT_LOGGER.addHandler(_handler)


class ErrorCategory(str, Enum):
    RETRYABLE = "retryable"          # transient: 5xx, timeout, lock contention
    SCHEMA_INVALID = "schema_invalid"  # terminal: payload violates contract
    AUTH_REJECTED = "auth_rejected"    # terminal: credentials/authorization
    UNKNOWN = "unknown"                # terminal until reclassified


@dataclass(frozen=True)
class DeadLetterRecord:
    """Immutable retained record for a payload that failed terminally."""
    payload: bytes
    source_digest: str            # SHA-256 of payload, the lineage anchor
    error_category: ErrorCategory
    error_detail: str
    attempt_count: int
    first_seen: datetime
    last_seen: datetime
    trace_id: str
    actor: str
    replay_count: int = 0
    audit: dict = field(default_factory=dict)

Making the record frozen=True is a compliance choice, not a stylistic one: a retained record under 2 CFR §200.334 should not be silently mutated after the fact. replay_count is the only field that advances, and it does so by writing a new record version, never by editing bytes in place.

Step 2: Decide the move-to-DLQ trigger

A payload moves to the DLQ under exactly two conditions: it exhausted its retry budget while carrying a retryable error, or it arrived with a non-retryable (terminal) category and should never be retried at all. Encode that as one pure predicate so the decision is testable in isolation and identical everywhere it is called.

python
def should_dead_letter(
    error_category: ErrorCategory,
    attempt_count: int,
    retry_budget: int,
) -> bool:
    """Return True when a payload is terminal and must leave the retry loop."""
    if error_category is ErrorCategory.RETRYABLE:
        # transient error: dead-letter only once the budget is fully spent
        exhausted = attempt_count >= retry_budget
        if exhausted:
            AUDIT_LOGGER.warning(
                "Retry budget exhausted | attempts=%d budget=%d", attempt_count, retry_budget
            )
        return exhausted
    # any non-retryable category is terminal on first sight
    AUDIT_LOGGER.warning("Non-retryable error | category=%s", error_category.value)
    return True

The key parameter is retry_budget, which this function reads but does not set — the budget and its backoff schedule are owned upstream by Pipeline Fallback & Retry Logic. Keeping the predicate pure means a unit test can assert every branch without a live broker.

Step 3: Preserve the source digest and audit trail

Before writing to the store, recompute the SHA-256 digest from the payload bytes and confirm it matches the digest the upstream stage recorded. This is the lineage anchor: it proves the bytes in the DLQ are byte-for-byte the ones that failed, which is what lets an auditor bind a dead-lettered record to its origin under the traceability expectation of 2 CFR §200.302.

python
class DigestMismatchError(RuntimeError):
    """Raised when payload bytes do not match the recorded source digest."""


def build_dead_letter_record(
    payload: bytes,
    expected_digest: str,
    error_category: ErrorCategory,
    error_detail: str,
    attempt_count: int,
    trace_id: str,
    actor: str,
    first_seen: datetime,
) -> DeadLetterRecord:
    """Construct a DLQ record, verifying digest lineage before it is retained."""
    computed = hashlib.sha256(payload).hexdigest()
    if computed != expected_digest:
        AUDIT_LOGGER.error(
            "Digest mismatch | expected=%s computed=%s trace=%s",
            expected_digest[:12], computed[:12], trace_id,
        )
        raise DigestMismatchError(f"payload digest drift for trace {trace_id}")

    now = datetime.now(timezone.utc)
    record = DeadLetterRecord(
        payload=payload,
        source_digest=computed,
        error_category=error_category,
        error_detail=error_detail,
        attempt_count=attempt_count,
        first_seen=first_seen,
        last_seen=now,
        trace_id=trace_id,
        actor=actor,
        audit={"dead_lettered_at": now.isoformat(), "reason": error_category.value},
    )
    AUDIT_LOGGER.info(
        "Dead-lettered | digest=%s category=%s attempts=%d trace=%s",
        computed[:12], error_category.value, attempt_count, trace_id,
    )
    return record

Raising DigestMismatchError rather than swallowing the mismatch is deliberate: if the bytes changed between failure and dead-lettering, you have a corruption bug that must not be hidden inside a quarantine record.

Step 4: Detect poison messages

A poison message is a payload that fails every time it is processed — often because it triggers a deterministic bug — and, if replayed blindly, it will loop through the DLQ forever and can starve the queue. Guard against it by capping replay_count: once a record has been replayed a threshold number of times and keeps landing back in the DLQ, route it to a poison quarantine for human triage instead of re-enqueuing it again.

python
POISON_REPLAY_THRESHOLD = 3


def classify_poison(record: DeadLetterRecord) -> bool:
    """Return True when a record has exhausted its replay allowance."""
    if record.replay_count >= POISON_REPLAY_THRESHOLD:
        AUDIT_LOGGER.error(
            "Poison message | digest=%s replays=%d category=%s trace=%s",
            record.source_digest[:12], record.replay_count,
            record.error_category.value, record.trace_id,
        )
        return True
    return False

POISON_REPLAY_THRESHOLD is the one lever to tune per corpus: set it low (2–3) for high-volume portal feeds where a looping payload is expensive, higher for feeds where transient upstream outages are common and a few extra replays are cheap.

Step 5: Build lineage-preserving replay tooling

When the root cause is fixed, replay must re-enqueue the original payload while advancing the audit trail — not resetting it. The replay produces a new record version with replay_count + 1 and an appended audit entry, so the history of every attempt survives. Poison records are refused here rather than re-enqueued.

python
class ReplayBlockedError(RuntimeError):
    """Raised when a poison record is submitted for replay."""


def replay_record(
    record: DeadLetterRecord,
    enqueue,                       # callable[[bytes, str], None] -> pipeline entry
) -> DeadLetterRecord:
    """Re-enqueue the original payload, preserving and extending its lineage."""
    if classify_poison(record):
        raise ReplayBlockedError(
            f"record {record.source_digest[:12]} is poison; route to quarantine"
        )

    # verify the bytes still match their anchor before re-injecting them
    if hashlib.sha256(record.payload).hexdigest() != record.source_digest:
        raise DigestMismatchError(f"replay digest drift for {record.trace_id}")

    enqueue(record.payload, record.trace_id)
    now = datetime.now(timezone.utc)
    replayed = DeadLetterRecord(
        **{**record.__dict__,
           "replay_count": record.replay_count + 1,
           "last_seen": now,
           "audit": {**record.audit,
                     f"replay_{record.replay_count + 1}": now.isoformat()}},
    )
    AUDIT_LOGGER.info(
        "Replayed | digest=%s replay=%d trace=%s",
        record.source_digest[:12], replayed.replay_count, record.trace_id,
    )
    return replayed

The enqueue callable is injected so the same replay logic works whether the pipeline entry point is a Postgres NOTIFY, an SQS SendMessage, or an in-process queue — the lineage bookkeeping is broker-agnostic.

Step 6: Emit dead-letter-queue depth metrics

A DLQ that fills silently is as dangerous as a dropped payload: it means terminal failures are accumulating and nobody is triaging them. Emit the queue depth (broken out by error category) on a fixed cadence so an alert fires when the backlog crosses a threshold, and record the oldest record’s age so a slow leak is visible before it becomes an SLA breach.

python
from collections import Counter
from typing import Sequence


def emit_dlq_depth_metrics(records: Sequence[DeadLetterRecord]) -> dict[str, object]:
    """Compute and log DLQ depth and age metrics for alerting."""
    by_category: Counter[str] = Counter(r.error_category.value for r in records)
    now = datetime.now(timezone.utc)
    oldest_age_s = max(
        ((now - r.first_seen).total_seconds() for r in records), default=0.0
    )
    metrics: dict[str, object] = {
        "dlq_depth_total": len(records),
        "dlq_depth_by_category": dict(by_category),
        "dlq_oldest_age_seconds": round(oldest_age_s, 1),
    }
    log = AUDIT_LOGGER.warning if len(records) else AUDIT_LOGGER.info
    log("DLQ metrics | %s", metrics)
    return metrics

Scrape the returned dict into your metrics backend and alert on dlq_depth_total and dlq_oldest_age_seconds together — a nonzero depth is expected, but a growing depth or an aging oldest record is the signal that remediation has stalled.

Verification

Confirm the DLQ behaves correctly with four checks:

  1. The trigger fires only on terminal conditions. Assert should_dead_letter(ErrorCategory.RETRYABLE, attempt_count=2, retry_budget=5) is False, that it becomes True at attempt_count=5, and that any non-retryable category returns True at attempt_count=0.
  2. Digest lineage is enforced. Build a record with a correct digest and confirm it succeeds; then tamper one byte of the payload and assert build_dead_letter_record raises DigestMismatchError rather than retaining mismatched bytes.
  3. Poison messages are refused, not looped. Set replay_count to POISON_REPLAY_THRESHOLD and assert replay_record raises ReplayBlockedError and never calls enqueue.
  4. Replay preserves and extends lineage. Replay a healthy record and assert the returned record has replay_count incremented by one, the same source_digest, and a new replay_N key in its audit dict — proof that history is appended, not overwritten.

A compliant dead-letter emits one WARNING naming the category and attempt count; a replay emits one INFO carrying the digest and replay number; a poison detection emits an ERROR. Ship those to a write-once tier so the trail satisfies the three-year retention window under 2 CFR §200.334.

python
digest = hashlib.sha256(b'{"grant_id": 42}').hexdigest()
rec = build_dead_letter_record(
    payload=b'{"grant_id": 42}', expected_digest=digest,
    error_category=ErrorCategory.SCHEMA_INVALID, error_detail="missing total",
    attempt_count=1, trace_id="t-001", actor="parser-worker-3",
    first_seen=datetime.now(timezone.utc),
)
assert rec.source_digest == digest and len(rec.source_digest) == 64
replayed = replay_record(rec, enqueue=lambda payload, trace: None)
assert replayed.replay_count == 1 and replayed.source_digest == digest

Common Errors & Fixes

Error Cause Fix
Payloads vanish after max retries Terminal failures logged and discarded, never stored Call should_dead_letter at budget exhaustion and persist a DeadLetterRecord; a retained record is required by 2 CFR §200.334.
DigestMismatchError on write Payload bytes were mutated (e.g. re-encoded) before dead-lettering Store the exact original bytes; never normalize or re-serialize a failed payload before it reaches the DLQ.
DLQ never drains, same record loops Poison message replayed blindly on every fix attempt Cap replay_count with classify_poison and route to quarantine at POISON_REPLAY_THRESHOLD.
Replay loses the attempt history New record overwrites audit instead of appending Merge into a copy ({**record.audit, ...}) and increment replay_count; keep the record frozen.
Retryable errors dead-lettered too early Trigger ignores the budget and treats every failure as terminal Only dead-letter RETRYABLE errors once attempt_count >= retry_budget; let backoff run first.
Backlog discovered days late No depth metric or alert on the DLQ Emit emit_dlq_depth_metrics on a schedule and alert on total depth and oldest-record age together.