Dead-Letter Queue Patterns for Failed Meter Reads

A meter read that exhausts its retries is not zero consumption — it is a revenue gap and an audit hole, and dropping it into a log line is how a municipality silently under-bills a whole route. The disciplined answer is a dead-letter queue (DLQ): a durable, inspectable holding area for reads that could not be committed, structured so a human can triage them and an automated driver can replay them safely once the root cause is fixed. This page sits under Error Handling & Retry Workflows, part of the broader Meter Data Ingestion & Validation Pipelines subsystem. Where the parent guide covers when a read gets dead-lettered, this page is about the queue itself — the record shape that makes triage possible, a replay driver that re-enters the exact same idempotent commit path, a cap that stops a poison message from looping forever, and the observability that turns a growing DLQ into a page rather than a surprise.

# Minimal Prerequisites

• Python 3.11+ — for StrEnum, X | None unions, and timezone-aware datetime throughout (a DLQ record’s timestamps decide retention, so naive datetimes are disqualifying).
• Pydantic v2 (pydantic>=2.6) — the DLQ record is a validated model so a malformed entry cannot itself become a second class of poison.
• decimal.Decimal for the metered payload — the stored read carries billing quantities; those remain Decimal, never float, so a replayed commit reproduces the original amount exactly.
• A stable idempotency key per read — derived upstream from meter, interval timestamp, and per-meter sequence. Replay safety rests entirely on this key already existing; this page consumes it, it does not invent it.
• A durable store with an idempotent commit path — the same keyed upsert the primary pipeline uses, so replay is structurally a no-op for reads that already committed.

Data assumptions: a read arrives at the DLQ already classified — either its transient-retry budget was spent, or a downstream returned a permanent error. The commit path the driver replays into is the identical exactly-once upsert the live pipeline uses; the DLQ never gets a private, looser write.

# Annotated Implementation

The DLQ has two operations that matter: put, which records a failure without ever fanning one read out into duplicate entries, and replay, which drains the queue back through the primary commit path. The record is keyed by idempotency key, so a read that fails, gets parked, and fails again on replay updates one entry — bumping attempts and last_seen_utc while preserving first_seen_utc for retention math. Replay’s cardinal rule: it calls the same commit the live pipeline calls, so replaying a read that already committed is a harmless no-op, and a read that keeps failing is capped and parked for a human instead of looping forever.

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Callable
from datetime import datetime, timezone
from enum import StrEnum

from pydantic import BaseModel, ConfigDict, Field

UTC = timezone.utc
POISON_CAP = 5          # replay failures tolerated before a record is parked


class DeadLetterReason(StrEnum):
    RETRIES_EXHAUSTED = "retries_exhausted"   # transient budget spent upstream
    PERMANENT_4XX = "permanent_4xx"           # contract breach — will never self-heal
    POISON = "poison"                         # keeps failing on replay


class DeadLetterRecord(BaseModel):
    """One failed meter read, keyed for safe idempotent replay and triage."""
    model_config = ConfigDict(frozen=True)

    idempotency_key: str                      # meter | ts | seq — the replay identity
    payload: dict                             # the original read, stored unmodified
    reason: DeadLetterReason
    attempts: int = Field(ge=1)               # total delivery + replay attempts
    first_seen_utc: datetime                  # when it first failed (never overwritten)
    last_seen_utc: datetime                   # most recent failure — drives retention

    def bump(self, reason: DeadLetterReason) -> "DeadLetterRecord":
        # A re-failed read updates one entry: raise the counter, keep first_seen.
        return self.model_copy(update={
            "attempts": self.attempts + 1,
            "reason": reason,
            "last_seen_utc": datetime.now(UTC),
        })


async def dead_letter_put(store: dict[str, DeadLetterRecord], key: str,
                          payload: dict, reason: DeadLetterReason) -> DeadLetterRecord:
    """Upsert on the idempotency key so one read never fans out into duplicates."""
    now = datetime.now(UTC)
    existing = store.get(key)
    if existing is None:
        record = DeadLetterRecord(
            idempotency_key=key, payload=payload, reason=reason,
            attempts=1, first_seen_utc=now, last_seen_utc=now,
        )
    else:
        record = existing.bump(reason)        # merge into the existing entry
    store[key] = record
    return record


async def replay_dead_letters(
    store: dict[str, DeadLetterRecord],
    commit: Callable[[dict, str], Awaitable[None]],   # the SAME idempotent commit path
    parked: dict[str, DeadLetterRecord],
) -> dict[str, int]:
    """Drain the DLQ through the primary commit path; park permanent and poison reads."""
    stats = {"committed": 0, "requeued": 0, "parked": 0}
    for key, record in list(store.items()):
        if record.reason is DeadLetterReason.PERMANENT_4XX:
            parked[key] = store.pop(key)      # a 4xx never self-heals — human triage
            stats["parked"] += 1
            continue
        try:
            await commit(record.payload, key)  # re-enters the exactly-once upsert
        except Exception:                      # replay still failing — reclassify
            bumped = record.bump(DeadLetterReason.POISON)
            if bumped.attempts >= POISON_CAP:
                parked[key] = bumped           # cap reached — stop looping, alert
                store.pop(key)
                stats["parked"] += 1
            else:
                store[key] = bumped            # leave it queued for the next drain
                stats["requeued"] += 1
            continue
        store.pop(key)                         # committed cleanly — leave the DLQ
        stats["committed"] += 1
    return stats

The whole safety argument reduces to one line: await commit(record.payload, key) calls the identical keyed upsert the live pipeline uses. Because that upsert is idempotent on the key, a replay of a read that in fact committed on its final live attempt collides with the existing row and does nothing — replaying the entire queue after an outage can never double-bill. The record’s split between first_seen_utc (frozen) and last_seen_utc (bumped) is what lets a retention sweep age entries by age-since-last-failure while auditors can still see how long a read has been stuck. Permanent 4xx reads are parked without replay because they will fail identically forever — the same permanent-versus-transient distinction the parent error handling and retry workflows draw at the ingestion boundary, applied here at the queue.

Figure: A failed read enters the keyed DLQ, is triaged, and either replays through the same idempotent commit path or is parked for human review once it hits the poison cap.

# Edge Cases and Billing Gotchas

The DLQ exists to make failure recoverable; these are the cases where a careless queue turns recovery back into corruption.

• Replaying an already-committed read. The dangerous illusion: a read committed on its final live attempt, but the acknowledgement was lost, so it also landed in the DLQ. Replay must go through the same keyed upsert, whose unique constraint makes the second write a no-op. If the replay path takes any shortcut around that constraint, draining the queue double-bills exactly the reads that were actually fine.
• The poison loop. A payload that fails deterministically on every replay — a truly malformed read, or one that trips a downstream bug — will otherwise cycle through the queue forever, burning capacity and drowning the signal. POISON_CAP bounds replay attempts; on the cap the record is parked, removed from the active queue, and surfaced to a human. A poison read stalling the queue is the classic DLQ failure, and the cap is the specific control that prevents it.
• DLQ growth and retention. An unbounded DLQ is its own outage: it hides the reads that matter under thousands that do not, and eventually exhausts storage. Age entries by last_seen_utc, alert on queue depth and oldest-entry age, and set a retention window long enough to satisfy the append-only audit expectations that audit logging and data governance require — parked records move to cold, write-once storage rather than being deleted.
• Partial replay. Draining is not atomic — a replay pass can commit some reads and requeue others when a downstream recovers only partially. The driver must be safe to run repeatedly: each pass commits what it can, requeues the rest with a bumped counter, and leaves the queue in a consistent state, so a second pass simply picks up where the first stopped without re-committing anything.

# Verification Snippet

Two properties are worth proving directly: replaying an already-committed read is a no-op, and a poison read is parked after the cap instead of looping forever.

def test_replay_is_idempotent_and_caps_poison() -> None:
    commits: dict[str, int] = {}          # key -> number of committing writes seen

    async def commit_ok(payload: dict, key: str) -> None:
        commits[key] = commits.get(key, 0) + 1   # the real path upserts; count it

    async def commit_fails(payload: dict, key: str) -> None:
        raise TimeoutError("downstream still down")

    async def scenario() -> None:
        store: dict[str, DeadLetterRecord] = {}
        parked: dict[str, DeadLetterRecord] = {}

        # An exhausted read lands in the DLQ, then replays cleanly and leaves.
        good = "MTR-00000001|2026-07-17T00:00:00+00:00|1"
        await dead_letter_put(store, good, {"kwh": "12.50"},
                              DeadLetterReason.RETRIES_EXHAUSTED)
        stats = await replay_dead_letters(store, commit_ok, parked)
        assert stats["committed"] == 1
        assert store == {}                       # drained
        assert commits[good] == 1

        # Re-draining an empty queue is a safe no-op (idempotent replay).
        again = await replay_dead_letters(store, commit_ok, parked)
        assert again == {"committed": 0, "requeued": 0, "parked": 0}
        assert commits[good] == 1                # never committed twice

        # A poison read that fails every replay is parked after the cap.
        bad = "MTR-00000009|2026-07-17T00:00:00+00:00|1"
        await dead_letter_put(store, bad, {"kwh": "corrupt"},
                              DeadLetterReason.RETRIES_EXHAUSTED)
        for _ in range(POISON_CAP + 2):
            await replay_dead_letters(store, commit_fails, parked)
        assert bad in parked                     # stopped looping, escalated
        assert bad not in store                  # not requeued forever

    asyncio.run(scenario())

The second replay of an empty queue committing nothing, and commits[good] staying at 1, together prove the queue is safe to drain repeatedly. Wire these assertions to the same fixtures your primary pipeline uses so a change to the commit path that breaks replay idempotence fails the suite before it reaches a billing run.

# FAQ

# Related Topics

• Error Handling & Retry Workflows — the parent guide covering classification, backoff, and circuit breakers that feed this queue.
• Implementing Exponential Backoff with Jitter in Python — the retry loop whose exhaustion routes a read here.
• Async Batch Processing for High-Volume Reads — the throughput layer that drains and replays the DLQ concurrently.
• Schema Validation & Data Quality Checks — classifies permanent payload defects so they are parked, not endlessly replayed.
• Meter Data Ingestion & Validation Pipelines — the end-to-end subsystem this recovery pattern protects.

Up: Error Handling & Retry Workflows · Meter Data Ingestion & Validation Pipelines