InventoryAllocation

Building a Dead-Letter Queue Replay Tool

When a channel-manager sync fails past its retries, the payload lands in a dead-letter queue — but a DLQ that no one can inspect, filter, or safely replay is just a slow-motion data-loss incident, because the rate and availability updates sitting in it are exactly the ones that never reached booking_com or expedia. This page builds a command-line tool that lists what is stuck, filters it down to the records you actually want to retry, and replays them back through the pipeline without double-applying, by preserving each message’s original idempotency key. It is the operational counterpart to the classification rules in error categorization and retry logic that decided these records belonged in the DLQ in the first place.

The reader is the on-call engineer at 8am after an OTA had a two-hour outage: a few thousand availability pushes dead-lettered, and they need to replay them safely before the day’s bookings come in — without re-sending anything that actually succeeded, and without letting one genuinely poisonous message block the batch.

Prerequisites & environment

The tool reads from the DLQ, re-validates each payload, and re-submits it through the same code path a live sync uses. Pin these:

The one non-negotiable prerequisite is that the original producer stamped a stable idempotency key onto each message before it was dead-lettered. If the key is missing, a replay cannot be made safe after the fact — the fix is upstream, in the producer, not in this tool.

The replay tool reads dead-lettered messages from the dead-letter queue and processes each through four stages. First it re-validates the payload against the canonical Pydantic model. A schema-invalid message is routed to a poison quarantine and removed from the main queue. A valid message carries its original idempotency key forward. In dry-run mode the tool only reports what it would replay and changes nothing. In apply mode it re-submits the payload to the sync endpoint with the preserved idempotency key, so a record that in fact already applied is deduplicated server-side and treated as a no-op success, and only on a confirmed success is the message deleted from the dead-letter queue. Poison messages are isolated so one bad record never blocks the batch. Dead-letter queue replay flow with dry-run, idempotency preservation, and poison quarantine Replay: re-validate, preserve the key, dry-run before apply One poison message never blocks the batch; a preserved idempotency key makes replay a safe no-op if it already applied. sync:dlq Redis Stream XRANGE / XDEL re-validate Pydantic model + read idem key poison quarantine schema-invalid → isolate invalid mode? dry-run: report only apply: re-submit valid dry-run report counts by channel / error — nothing changes sync endpoint (apply) POST + Idempotency-Key: original already applied → dedup no-op success on confirmed success → XDEL remove from DLQ only after 2xx
Figure 1: Each dead-lettered message is re-validated, poison records quarantined, and valid ones replayed with their original idempotency key — deleted from the DLQ only after a confirmed success.

Step-by-step implementation

Four steps: model the envelope so the idempotency key travels with the payload, build an inspect view that summarizes the DLQ before you touch it, replay with the preserved key under a dry-run flag, and quarantine poison messages so one bad record never stalls the batch.

Step 1 — Model the dead-letter envelope

A dead-lettered message is more than its payload: it carries the original idempotency key, the failure reason that sent it here, and an attempt count. Model the envelope so the key is a first-class field that replay cannot lose.

python
from pydantic import BaseModel, Field

class DeadLetter(BaseModel):
    stream_id: str                      # Redis Stream entry id, for XDEL after success
    idempotency_key: str = Field(min_length=8)   # MUST survive into the replay
    channel: str
    endpoint: str
    property_id: str
    payload: dict                       # the original sync body, replayed verbatim
    error_class: str                    # e.g. "ota_5xx", "timeout", "schema_invalid"
    attempts: int = Field(ge=1)

    def replay_headers(self) -> dict[str, str]:
        # the original key — NOT a fresh one — is what makes replay safe
        return {"Idempotency-Key": self.idempotency_key}

Reconstructing the original idempotency key into the replay headers is the single choice that makes the whole tool safe: mint a fresh key here and the OTA treats the replay as a brand-new update, double-applying a rate change that may have actually succeeded before the failure that dead-lettered it.

Step 2 — Inspect before you touch anything

Never replay blind. Load the DLQ into Polars and summarize it by channel, error class, and property, so the operator can see whether they are looking at one OTA’s outage or a systemic schema bug before deciding what to replay.

python
import polars as pl
import redis

def load_dlq(r: redis.Redis, stream: str = "sync:dlq") -> list[DeadLetter]:
    entries = r.xrange(stream)          # [(id, {field: value}), ...]
    out = []
    for entry_id, fields in entries:
        out.append(DeadLetter(stream_id=entry_id.decode(), **_decode(fields)))
    return out

def summarize(letters: list[DeadLetter]) -> pl.DataFrame:
    return (
        pl.DataFrame([l.model_dump(exclude={"payload"}) for l in letters])
        .group_by(["channel", "error_class"])
        .agg(pl.len().alias("count"),
             pl.col("property_id").n_unique().alias("properties"))
        .sort("count", descending=True)
    )

Grouping by error_class up front is what turns a scary undifferentiated backlog into a decision: a pile of ota_5xx from one channel is a safe bulk replay, whereas a pile of schema_invalid means the producer is broken and replaying would just re-fail every record — a distinction drawn straight from categorizing 4xx vs 5xx sync errors.

Step 3 — Replay with the preserved key, behind a dry-run flag

Default to dry-run: report exactly what would be replayed and change nothing. Only with --apply does the tool re-submit — and it deletes a message from the DLQ only after a confirmed success, so a replay that fails again stays put for the next attempt.

python
import httpx
import structlog

log = structlog.get_logger()

def replay(r: redis.Redis, client: httpx.Client, letters: list[DeadLetter],
           apply: bool) -> dict[str, int]:
    stats = {"replayed": 0, "skipped_dry_run": 0, "failed": 0}
    for dl in letters:
        if not apply:
            log.info("would_replay", idempotency_key=dl.idempotency_key,
                     property_id=dl.property_id, channel=dl.channel)
            stats["skipped_dry_run"] += 1
            continue
        resp = client.post(f"/sync/{dl.endpoint}", json=dl.payload,
                           headers=dl.replay_headers())
        # 200/201 = applied now; 409/200-dedup = already applied → still success
        if resp.status_code in (200, 201, 409):
            r.xdel("sync:dlq", dl.stream_id)     # remove ONLY after confirmed success
            stats["replayed"] += 1
        else:
            log.warning("replay_failed", idempotency_key=dl.idempotency_key,
                        status=resp.status_code)
            stats["failed"] += 1
    return stats

Treating a 409 Conflict (or a dedup-hit 200) as success, not failure, is what closes the double-apply gap end to end: because the original key was preserved, a record that actually applied before dead-lettering is recognized server-side and the replay becomes a safe no-op that still clears the message from the queue.

Step 4 — Quarantine poison messages so one bad record can’t block the batch

A message that fails re-validation, or fails replay repeatedly, is poison — replaying it forever just stalls everything behind it. Route it to a separate quarantine stream and remove it from the main DLQ, so the good records keep flowing.

python
POISON_STREAM = "sync:dlq:poison"

def quarantine_poison(r: redis.Redis, raw_id: str, raw_fields: dict,
                      reason: str) -> None:
    # copy the raw entry verbatim into the poison stream for human review
    r.xadd(POISON_STREAM, {**raw_fields, "quarantine_reason": reason})
    r.xdel("sync:dlq", raw_id)          # unblock the batch; do not lose the data
    log.error("dlq_poison_quarantined", stream_id=raw_id, reason=reason)

def safe_load(r: redis.Redis, stream: str = "sync:dlq") -> list[DeadLetter]:
    letters = []
    for entry_id, fields in r.xrange(stream):
        try:
            letters.append(DeadLetter(stream_id=entry_id.decode(), **_decode(fields)))
        except Exception as exc:        # unparseable envelope → poison, not a crash
            quarantine_poison(r, entry_id.decode(), _decode(fields), str(exc))
    return letters

Copying poison to a dedicated stream rather than deleting it outright preserves the evidence for a human while still unblocking the batch — the payload is never lost, but a single malformed record can no longer wedge the head of the queue and starve every valid message behind it.

Gotchas & production notes

Verification snippet

The behaviour worth proving is the safety property: replaying a record whose update already applied must not double-apply, and must still clear it from the DLQ. Stub the sync endpoint to return 409 (server-side dedup) and assert the message is removed cleanly.

python
import fakeredis

def test_replay_is_idempotent_and_clears_dlq() -> None:
    r = fakeredis.FakeStrictRedis()
    dl = DeadLetter(
        stream_id="1-0", idempotency_key="idem-booking-8842-2026-07-20",
        channel="booking_com", endpoint="availability", property_id="LON-STJ-01",
        payload={"property_id": "LON-STJ-01", "date": "2026-07-20", "avail": 3},
        error_class="ota_5xx", attempts=3)
    r.xadd("sync:dlq", {"x": "1"}, id="1-0")   # placeholder entry to be XDEL'd

    class DedupClient:                          # OTA already applied this key
        def post(self, url, json, headers):
            assert headers["Idempotency-Key"] == "idem-booking-8842-2026-07-20"
            class Resp: status_code = 409       # dedup hit, not a real conflict
            return Resp()

    stats = replay(r, DedupClient(), [dl], apply=True)
    assert stats["replayed"] == 1              # dedup 409 counts as success
    assert r.xlen("sync:dlq") == 0             # message cleared, not re-queued

test_replay_is_idempotent_and_clears_dlq()

A green run confirms the tool’s core guarantee: a record whose update had in fact already reached the OTA is replayed with its original key, recognized as a duplicate server-side, counted as a success, and removed from the DLQ — so an operator can bulk-replay a recovered backlog without any fear of double-applying a rate or availability change.

← Back to Error Categorization & Retry Logic