Discrepancy Routing and Remediation › Routing Discrepancies to Dead-Letter Queues

# Routing Discrepancies to Dead-Letter Queues

This page answers a question that decides whether a reconciliation pipeline is trustworthy: when a discrepancy cannot be auto-remediated, where does it go, and how do you guarantee it is actually handled rather than quietly forgotten? It sits under the discrepancy routing and remediation reference and treats the dead-letter queue as a first-class, monitored system — because the most dangerous failure in reconciliation is a discrepancy that was detected, deferred, and then never looked at again.

# Problem Framing

Your classifier routes uncertain discrepancies to a dead-letter queue for deferred handling. The risk is that the queue becomes a black hole: records pile up, no one processes them, and a real data-integrity problem sits unresolved behind a dashboard that says reconciliation is “running.” Two properties prevent this. First, each dead-letter record must be self-describing — carrying enough context to be reprocessed without re-running the whole reconciliation. Second, the queue must have a reprocessing SLA and depth-and-age alerting, so a record that sits too long or a queue that grows without bound raises an alarm. A poison message — one that fails every reprocessing attempt — must be capped and escalated, never retried forever.

# Implementation

Each record captures the discrepancy, its source verdict, and a retry count. The processor reprocesses with bounded retries, escalates poison messages, and exposes depth and age so a stuck queue is visible.

import logging
import time
from dataclasses import dataclass, field, asdict
from typing import Callable, Dict, List, Optional

logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", level=logging.INFO
)
logger = logging.getLogger("recon.arch.dlq")


@dataclass
class DeadLetter:
    discrepancy_id: str
    category: str
    context: Dict            # everything needed to reprocess without re-reconciling
    enqueued_ms: int
    attempts: int = 0
    max_attempts: int = 5


@dataclass
class DeadLetterQueue:
    _items: List[DeadLetter] = field(default_factory=list)
    escalate: Optional[Callable[[DeadLetter], None]] = None

    def enqueue(self, dl: DeadLetter) -> None:
        self._items.append(dl)

    def depth(self) -> int:
        return len(self._items)

    def oldest_age_ms(self, now_ms: int) -> int:
        return max((now_ms - i.enqueued_ms for i in self._items), default=0)

    def reprocess(self, handler: Callable[[DeadLetter], bool], now_ms: int) -> Dict[str, int]:
        """Attempt each record; retry up to a cap, then escalate the poison messages."""
        resolved = poisoned = retried = 0
        remaining: List[DeadLetter] = []
        for dl in self._items:
            dl.attempts += 1
            try:
                if handler(dl):
                    resolved += 1
                    continue
            except Exception as exc:                       # noqa: BLE001 - never crash the drain
                logger.warning("reprocess error id=%s: %s", dl.discrepancy_id, exc)
            if dl.attempts >= dl.max_attempts:
                poisoned += 1
                logger.error("poison message id=%s after %d attempts", dl.discrepancy_id, dl.attempts)
                if self.escalate:
                    self.escalate(dl)
            else:
                retried += 1
                remaining.append(dl)
        self._items = remaining
        return {"resolved": resolved, "retried": retried, "poisoned": poisoned}

# Dead-Letter Strategy Trade-Off

The retry and escalation policy trades resolution speed against the risk of hammering a broken target.

# Key Implementation Notes

• Make each record self-describing. Store the full reprocessing context in the dead-letter record, not a reference that requires re-running the reconciliation. A dead-letter that can only be understood by replaying the whole job is one no one will process.
• Cap retries and escalate poison messages. A record that fails every attempt is a poison message; retrying it forever wastes resources and hides a systemic problem. Cap attempts, escalate to human review, and stop — the same bounded-degradation discipline as the fallback chain implementation.
• Alert on depth and age, not just depth. A queue at constant depth can still be failing if the same records are aging; alert on the oldest record’s age so a slow black hole is caught as readily as a fast-growing one.
• The queue is in-scope for data policy. Dead-letter records carry divergent field values, so apply the masking and retention the security boundaries for reconciliation reference defines; a dead-letter store is not operational exhaust.

# Verification

Assert that a transient failure retries, a resolved record leaves the queue, and a poison message escalates after the cap.

escalated: List[str] = []
dlq = DeadLetterQueue(escalate=lambda dl: escalated.append(dl.discrepancy_id))
dlq.enqueue(DeadLetter("d1", "value_diverged", {"fix": 1}, enqueued_ms=0, max_attempts=2))
dlq.enqueue(DeadLetter("d2", "missing_in_target", {"fix": 2}, enqueued_ms=0, max_attempts=2))

# d1 always fails (poison), d2 resolves on first try.
def handler(dl): return dl.discrepancy_id == "d2"
r1 = dlq.reprocess(handler, now_ms=1000)
assert r1["resolved"] == 1 and r1["retried"] == 1        # d2 done, d1 retried
r2 = dlq.reprocess(handler, now_ms=2000)
assert r2["poisoned"] == 1 and escalated == ["d1"]       # d1 capped and escalated
assert dlq.depth() == 0
logger.info("dead-letter reprocessing verified")

# Operational Considerations

Set a reprocessing SLA — a maximum age before a record must be resolved or escalated — and alert when the oldest record breaches it, so a stuck queue pages before it becomes an integrity incident. Rate-limit the reprocessing drain so a recovered target is not immediately stormed by a backlog, and run the drain on a schedule aligned to the reconciliation cadence. Expose dlq_depth, dlq_oldest_age_ms, poison_total, and resolved_total on a single panel so an operator sees both growth and staleness, and write every enqueue, resolution, and escalation to the same append-only audit trail the PCI-DSS audit trail format defines. Treat a rising poison rate as a signal that an upstream verdict or a target write path is broken, not as records to be cleared.

# Related

• Discrepancy routing and remediation — the parent reference whose uncertain-path this implements.
• Fallback chain implementation — the bounded-degradation pattern the retry cap mirrors.
• Security boundaries for reconciliation — the data policy a dead-letter store inherits.
• Exactly-once reconciliation with idempotent sinks — why a reprocessed remediation is safe to retry.