Replaying Dead-Letter Reconciliation Messages Without Double-Posting

A reconciliation message lands on the dead-letter queue for one of two reasons: a transient failure that a downstream fix will resolve, or a permanently malformed payload that will never process cleanly. Once an engineer ships the fix, every quarantined message with that failure signature needs to re-enter the pipeline — but a naive requeue risks posting the same ledger entry twice, or looping a poison message through the consumer forever. This page implements a safe replay path within Dead-Letter Queue Remediation: an envelope model that preserves the original failure context, a triage step that separates recoverable payloads from unrecoverable ones, an idempotent replay that dedups on source_hash before a message ever reaches the consumer again, and a structured audit trail that survives a compliance review.

Prerequisites

Messages have already been routed onto the dead-letter queue by Dead-Letter Queue Remediation, each carrying its original payload, failure code, and attempt count.
The downstream fix that caused the original failure has been deployed and verified against a sample of the quarantined payloads.
A dedup store is available (Redis or an equivalent key-value store with atomic SETNX) that the replay path can query independently of the primary matching database.
Python 3.10+ with pydantic v2 installed; monetary fields use Decimal, never float.
A structured logger is configured to emit trace_id, source_hash, and match_decision on every replay decision.
The consumer that will receive replayed messages exposes the same idempotent write contract used by Building an Asyncio Approval Queue with Backpressure, so a replayed message cannot bypass its own concurrency limits.

Step 1 — Model the DLQ envelope

The envelope wraps the original payload without mutating it. source_hash is computed once, at the point of original ingestion, and travels with the message through every retry so replay can dedup against it later. attempt_count and failure_code are what triage reads to decide whether a message is worth replaying at all.

python

from pydantic import BaseModel, ConfigDict, field_validator
from decimal import Decimal
from datetime import datetime, timezone
from enum import Enum
from uuid import UUID
import hashlib
import logging

log = logging.getLogger("dlq.replay")

class FailureCode(str, Enum):
    TRANSIENT_TIMEOUT = "TRANSIENT_TIMEOUT"
    SCHEMA_MISMATCH = "SCHEMA_MISMATCH"
    DOWNSTREAM_5XX = "DOWNSTREAM_5XX"
    MALFORMED_PAYLOAD = "MALFORMED_PAYLOAD"

class DlqEnvelope(BaseModel):
    model_config = ConfigDict(frozen=True)

    trace_id: UUID
    original_payload: dict
    failure_code: FailureCode
    attempt_count: int
    amount_usd: Decimal
    source_hash: str
    quarantined_at: datetime          # timezone-aware UTC

    @field_validator("quarantined_at")
    @classmethod
    def must_be_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("quarantined_at must be timezone-aware")
        return v.astimezone(timezone.utc)

    @field_validator("amount_usd")
    @classmethod
    def two_dp(cls, v: Decimal) -> Decimal:
        return v.quantize(Decimal("0.01"))

def compute_source_hash(payload: dict) -> str:
    canonical = repr(sorted(payload.items()))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

source_hash must be derived from the payload’s stable business fields — batch reference, counterparty, amount, value date — not from an ingestion timestamp or a retry-generated ID, otherwise every replay attempt produces a different hash and dedup never fires.

Step 2 — Triage and classify quarantined messages

Triage separates messages worth replaying from messages that will only fail again. A MALFORMED_PAYLOAD envelope that already exhausted max_replay_attempts is unrecoverable and moves to permanent quarantine instead of re-entering the pipeline.

python

from enum import Enum

class TriageDecision(str, Enum):
    RECOVERABLE = "RECOVERABLE"
    UNRECOVERABLE = "UNRECOVERABLE"

RECOVERABLE_CODES = {FailureCode.TRANSIENT_TIMEOUT, FailureCode.DOWNSTREAM_5XX}

def triage(envelope: DlqEnvelope, max_replay_attempts: int) -> TriageDecision:
    if envelope.failure_code not in RECOVERABLE_CODES:
        return TriageDecision.UNRECOVERABLE
    if envelope.attempt_count >= max_replay_attempts:
        return TriageDecision.UNRECOVERABLE
    return TriageDecision.RECOVERABLE

SCHEMA_MISMATCH and MALFORMED_PAYLOAD are excluded from RECOVERABLE_CODES deliberately: a schema fix changes the consumer’s contract, so those envelopes need a manual reprocessing decision rather than an automatic replay, even after the deploy.

Step 3 — Dedup on source_hash before replay

The dedup store is the single control that prevents a replayed message from posting to the ledger twice. A SETNX-style atomic write claims the hash before the message is handed to the consumer; if the claim fails, another worker (or an earlier replay run) already processed that hash and this attempt is skipped.

python

class DedupStore:
    """Minimal interface — back this with Redis SETNX + TTL in production."""
    def claim(self, source_hash: str, ttl_seconds: int) -> bool:
        raise NotImplementedError

class RedisDedupStore(DedupStore):
    def __init__(self, redis_client) -> None:
        self._redis = redis_client

    def claim(self, source_hash: str, ttl_seconds: int) -> bool:
        # SET key value NX EX ttl — atomic claim-and-expire in one round trip.
        return bool(
            self._redis.set(f"dlq:replayed:{source_hash}", "1", nx=True, ex=ttl_seconds)
        )

def replay_one(envelope: DlqEnvelope, dedup: DedupStore, dedup_ttl: int) -> str:
    if not dedup.claim(envelope.source_hash, dedup_ttl):
        log.info(
            "dlq.replay_skipped",
            extra={
                "trace_id": str(envelope.trace_id),
                "source_hash": envelope.source_hash,
                "match_decision": "REPLAY_DUPLICATE",
            },
        )
        return "REPLAY_DUPLICATE"
    return "CLAIMED"

The claim must happen before the message reaches the consumer, not after a successful post — claiming post-hoc leaves a window where two workers both pass triage on the same hash and both post.

Step 4 — Replay in bounded batches with backoff

Replaying the entire backlog in one pass can overwhelm the consumer the moment the fix ships, producing a fresh wave of failures. Bounded batches with exponential backoff between them keep replay traffic inside the consumer’s normal capacity.

python

import time
from dataclasses import dataclass

@dataclass
class ReplayResult:
    replayed: int
    skipped_duplicate: int
    quarantined: int

def replay_batch(
    envelopes: list[DlqEnvelope],
    dedup: DedupStore,
    consumer_post,               # callable: DlqEnvelope -> bool
    *,
    replay_batch_size: int,
    dedup_ttl: int,
    max_replay_attempts: int,
    backoff_base_ms: int,
) -> ReplayResult:
    replayed = skipped = quarantined = 0
    for offset in range(0, len(envelopes), replay_batch_size):
        chunk = envelopes[offset:offset + replay_batch_size]
        for envelope in chunk:
            decision = triage(envelope, max_replay_attempts)
            if decision is TriageDecision.UNRECOVERABLE:
                quarantined += 1
                log.warning(
                    "dlq.quarantined",
                    extra={
                        "trace_id": str(envelope.trace_id),
                        "source_hash": envelope.source_hash,
                        "match_decision": "UNRECOVERABLE_PAYLOAD",
                    },
                )
                continue
            outcome = replay_one(envelope, dedup, dedup_ttl)
            if outcome == "REPLAY_DUPLICATE":
                skipped += 1
                continue
            ok = consumer_post(envelope)
            log.info(
                "dlq.replayed",
                extra={
                    "trace_id": str(envelope.trace_id),
                    "source_hash": envelope.source_hash,
                    "match_decision": "REPLAYED" if ok else "REPLAY_FAILED",
                },
            )
            if ok:
                replayed += 1
        backoff_s = (backoff_base_ms / 1000) * (2 ** (offset // replay_batch_size))
        time.sleep(min(backoff_s, 30))
    return ReplayResult(replayed, skipped, quarantined)

The 2 ** batch_index backoff mirrors the retry discipline used by the transaction-matching pipeline itself, so a burst of replay traffic degrades gracefully rather than tripping the consumer’s own circuit breaker.

Step 5 — Route unrecoverable payloads to permanent quarantine

A message that fails triage twice, or whose attempt_count exceeds quarantine_after, is written to a separate permanent-quarantine store with its full failure history intact. Nothing in permanent quarantine re-enters the automatic replay path — it requires a human decision and, in most shops, a change ticket.

python

class QuarantineStore:
    def persist(self, envelope: DlqEnvelope, reason: str) -> None:
        raise NotImplementedError

def finalize_quarantine(
    envelope: DlqEnvelope, store: QuarantineStore, quarantine_after: int
) -> bool:
    if envelope.attempt_count < quarantine_after:
        return False
    store.persist(envelope, reason="UNRECOVERABLE_PAYLOAD")
    log.warning(
        "dlq.permanent_quarantine",
        extra={
            "trace_id": str(envelope.trace_id),
            "source_hash": envelope.source_hash,
            "match_decision": "UNRECOVERABLE_PAYLOAD",
            "attempt_count": envelope.attempt_count,
        },
    )
    return True

Configuration boundary table

Parameter	Default	Valid range	Notes
`replay_batch_size`	`100`	`10`–`1000`	Messages replayed per pass before the backoff pause
`dedup_ttl`	`86400`	`3600`–`604800`	Seconds a claimed `source_hash` blocks a second replay
`max_replay_attempts`	`3`	`1`–`10`	Attempts before triage marks a message `UNRECOVERABLE`
`quarantine_after`	`5`	`1`–`20`	Total attempt count that forces permanent quarantine
`backoff_base_ms`	`500`	`100`–`5000`	Base delay multiplied exponentially between batches

Verification and testing

A fixture with a known source_hash collision proves the dedup path actually blocks a second post, and a second fixture with an exhausted attempt_count proves triage routes to quarantine instead of looping forever.

python

from datetime import timedelta

class FakeDedupStore(DedupStore):
    def __init__(self) -> None:
        self._claimed: set[str] = set()

    def claim(self, source_hash: str, ttl_seconds: int) -> bool:
        if source_hash in self._claimed:
            return False
        self._claimed.add(source_hash)
        return True

def test_dedup_blocks_second_replay():
    envelope = DlqEnvelope(
        trace_id=UUID(int=42),
        original_payload={"batch_ref": "BATCH-2026-07-10", "amount": "1250.00"},
        failure_code=FailureCode.TRANSIENT_TIMEOUT,
        attempt_count=1,
        amount_usd=Decimal("1250.00"),
        source_hash=compute_source_hash({"batch_ref": "BATCH-2026-07-10", "amount": "1250.00"}),
        quarantined_at=datetime.now(timezone.utc) - timedelta(hours=2),
    )
    dedup = FakeDedupStore()
    assert replay_one(envelope, dedup, dedup_ttl=3600) == "CLAIMED"
    assert replay_one(envelope, dedup, dedup_ttl=3600) == "REPLAY_DUPLICATE"

def test_triage_routes_exhausted_attempts_to_quarantine():
    exhausted = DlqEnvelope(
        trace_id=UUID(int=7),
        original_payload={"batch_ref": "BATCH-2026-07-01"},
        failure_code=FailureCode.TRANSIENT_TIMEOUT,
        attempt_count=5,
        amount_usd=Decimal("40.00"),
        source_hash="deadbeef",
        quarantined_at=datetime.now(timezone.utc),
    )
    assert triage(exhausted, max_replay_attempts=3) is TriageDecision.UNRECOVERABLE

Before running a production replay, dry-run replay_batch against a read-only copy of the DLQ, diff the resulting match_decision counts against the expected recoverable/unrecoverable split from triage, and only then point consumer_post at the live consumer.

Troubleshooting

REPLAY_DUPLICATE — a message is skipped that the operator expected to replay. Root cause: an earlier replay run already claimed the source_hash and dedup_ttl has not expired. Fix: check the dedup store for the key directly; if the earlier replay genuinely failed downstream, clear that single key rather than shortening dedup_ttl globally.
UNRECOVERABLE_PAYLOAD — triage keeps quarantining messages that should now succeed. Root cause: failure_code was set to a code outside RECOVERABLE_CODES (commonly SCHEMA_MISMATCH) even though the fix addressed the underlying issue. Fix: re-classify the original failure code at the source, don’t force the replay path to treat every code as recoverable.
DLQ_OVERFLOW — the replay job falls further behind than it processes. Root cause: replay_batch_size and backoff_base_ms are tuned for steady-state traffic, not for draining a multi-day backlog. Fix: raise replay_batch_size temporarily and run the drain during a low-traffic window, watching consumer latency as the ceiling.
POISON_MESSAGE — the same envelope re-enters the DLQ immediately after every replay. Root cause: attempt_count is being reset on requeue instead of incremented, so max_replay_attempts never trips. Fix: increment attempt_count on the envelope itself before each replay and persist it, so triage can see the true history.

Part of Dead-Letter Queue Remediation within Exception Routing & Human-in-the-Loop Workflows.