Designing a Dead-Letter Queue for Failed DVIR Payloads

A dead-letter queue for Driver Vehicle Inspection Reports is not a garbage can for messages your workers could not parse — it is a compliance holding area, and its design decides whether a failed inspection is recoverable evidence or a records-retention gap under 49 CFR § 396.3©. This guide builds a concrete queue on Redis Streams (with the equivalent Amazon SQS mapping noted where it differs), a failure envelope that preserves the original payload bytes alongside the error context and the attempt count, and replay tooling that re-enqueues a corrected payload without ever duplicating a compliance record. It is the implementation the policy layer in Exception Handling and Dead-Letter Routing sits on top of, within the wider DVIR Ingestion & Digital/Paper Parsing Workflows architecture.

Get the envelope wrong and the failure becomes useless: store only the exception message and you cannot re-derive what actually failed; store the parsed-but-invalid record instead of the original bytes and you have thrown away the evidence an auditor needs; inline driver PII into a queue that fans out to logs and you have created a second, unscoped copy of regulated data. The three properties this queue must guarantee are that a message survives the round-trip byte-for-byte, that replay is idempotent, and that a poison message stops after a bounded number of attempts instead of cycling forever.

This guide assumes the classifier from Exception Handling and Dead-Letter Routing has already decided a payload is permanently failed (or retry-exhausted); the queue below is where those payloads land. You need:

• Python 3.10+ — the code uses StrEnum, X | Y unions, and match/case.
• redis>=5 with Redis Streams — XADD/XREADGROUP/XACK give durable, at-least-once delivery with a consumer group and a per-message delivery count. On Amazon SQS the equivalent is a source queue with a redrive policy pointing at an SQS dead-letter queue and a maxReceiveCount; the envelope and replay logic here are broker-agnostic.
• pydantic (v2) — validate the envelope on the way in and on the way out, so a malformed envelope can never itself become a silent failure.
• A write-once object store — S3 with Object Lock, or any content-addressed blob store — to hold the original payload bytes keyed by SHA-256, exactly as the WORM Retention and SHA-256 Hashing for DVIR Records guide specifies.

The envelope is the unit of everything the queue does. It carries a reference to the original bytes (never a lossy reparse), the error context needed to diagnose the failure, and the attempt count that bounds replay. Reference the payload by content hash and store the bytes once; the envelope stays small and re-adding the same failing DVIR deduplicates instead of piling up copies.

import hashlib
from datetime import datetime, timezone
from enum import StrEnum
from pydantic import BaseModel, Field

MAX_ATTEMPTS = 5  # a payload delivered this many times is parked as poison, never re-enqueued


class Disposition(StrEnum):
    DEAD_LETTER = "dead_letter"
    REPLAYED = "replayed"
    POISON = "poison"       # exceeded MAX_ATTEMPTS — parked for manual review
    RESOLVED = "resolved"


class FailureEnvelope(BaseModel):
    payload_ref: str = Field(pattern=r"^[0-9a-f]{64}$")  # SHA-256 of the original bytes
    stage: str                       # where processing stopped: validate / parse / dispatch
    error_code: str                  # machine-stable symbol, e.g. "schema.vin_length"
    last_error: str                  # exception summary — no raw payload, no PII
    attempt: int = Field(ge=1)
    first_seen: datetime
    disposition: Disposition


def build_envelope(payload: bytes, stage: str, exc: Exception, attempt: int) -> FailureEnvelope:
    return FailureEnvelope(
        payload_ref=hashlib.sha256(payload).hexdigest(),
        stage=stage,
        error_code=f"{stage}.{type(exc).__name__}",
        last_error=str(exc)[:500],   # summary only
        attempt=attempt,
        first_seen=datetime.now(timezone.utc),
        disposition=Disposition.DEAD_LETTER,
    )

Write the payload bytes to the content-addressed store first, then publish the envelope. Ordering matters: if the process dies between the two, a re-run recomputes the same hash and re-stores the same bytes harmlessly, but an envelope pointing at bytes that were never stored is an unrecoverable dangling reference. Store bytes, then enqueue.

import redis

r = redis.Redis()
DLQ_STREAM = "dvir:dead-letter"


def dead_letter(payload: bytes, envelope: FailureEnvelope, blob_store) -> str:
    # 1. Persist the original bytes once, keyed by their hash (idempotent by construction).
    blob_store.put_if_absent(envelope.payload_ref, payload)
    # 2. Publish the envelope. XADD returns the durable stream ID.
    return r.xadd(DLQ_STREAM, {"envelope": envelope.model_dump_json()}).decode()

Read the dead-letter stream through a consumer group so every envelope is acknowledged exactly once it is handled, and unacknowledged envelopes remain claimable after a crash. Redis tracks a per-message delivery count in the pending-entries list; use it to detect a payload that keeps crashing the replay tool itself.

GROUP = "dlq-replayers"

def ensure_group() -> None:
    try:
        r.xgroup_create(DLQ_STREAM, GROUP, id="0", mkstream=True)
    except redis.ResponseError:
        pass  # group already exists

def read_batch(consumer: str, count: int = 16):
    ensure_group()
    return r.xreadgroup(GROUP, consumer, {DLQ_STREAM: ">"}, count=count, block=5000)

Replay re-enqueues the original bytes — not the envelope — into the work queue, incrementing the attempt count. Before re-enqueuing, check the attempt count against MAX_ATTEMPTS: a payload that has already been delivered that many times is poison, so park it in a dedicated partition for manual inspection instead of cycling it again. Crucially, the downstream record write is keyed on payload_ref, so replaying an inspection that had already been admitted resolves to the existing record rather than inserting a duplicate — that is what makes replay safe to run repeatedly.

WORK_STREAM = "dvir:ingest"
POISON_STREAM = "dvir:poison"


def replay(entry_id: str, envelope: FailureEnvelope, blob_store) -> Disposition:
    if envelope.attempt >= MAX_ATTEMPTS:
        # Poison: park it, acknowledge the DLQ entry, do NOT re-enqueue.
        parked = envelope.model_copy(update={"disposition": Disposition.POISON})
        r.xadd(POISON_STREAM, {"envelope": parked.model_dump_json()})
        r.xack(DLQ_STREAM, GROUP, entry_id)
        return Disposition.POISON

    payload = blob_store.get(envelope.payload_ref)   # exact original bytes
    r.xadd(WORK_STREAM, {
        "payload_ref": envelope.payload_ref,          # record write dedupes on this key
        "attempt": str(envelope.attempt + 1),
    })
    blob_store.put_if_absent(envelope.payload_ref, payload)  # keep bytes retained through replay
    r.xack(DLQ_STREAM, GROUP, entry_id)
    return Disposition.REPLAYED

Three properties are non-negotiable and each maps to a test. Use fakeredis (or a disposable Redis) plus an in-memory blob store so the suite runs without infrastructure.

import pytest
from datetime import datetime, timezone


class MemBlobStore:
    def __init__(self): self._d = {}
    def put_if_absent(self, k, v): self._d.setdefault(k, v)
    def get(self, k): return self._d[k]


def test_message_survives_round_trip():
    """The exact original bytes come back byte-for-byte after a round-trip."""
    store = MemBlobStore()
    original = b"%PDF-1.4 ... raw DVIR bytes ..."
    env = build_envelope(original, "parse", ValueError("bad xref"), attempt=1)
    dead_letter(original, env, store)
    assert store.get(env.payload_ref) == original   # nothing lossy in between


def test_replay_is_idempotent():
    """Replaying the same envelope twice yields one work item per replay but one record."""
    store = MemBlobStore()
    original = b"raw bytes"
    env = build_envelope(original, "dispatch", TimeoutError("downstream"), attempt=1)
    store.put_if_absent(env.payload_ref, original)
    # The record layer keys on payload_ref; simulate its dedupe set.
    records: set[str] = set()
    for _ in range(3):
        records.add(env.payload_ref)   # stands in for the upsert keyed on the hash
    assert len(records) == 1           # replay cannot create a duplicate compliance record


def test_poison_message_stops_after_max_attempts():
    """A payload at the attempt cap is parked, never re-enqueued."""
    store = MemBlobStore()
    original = b"segfault bait"
    env = build_envelope(original, "parse", RuntimeError("worker crash"), attempt=MAX_ATTEMPTS)
    store.put_if_absent(env.payload_ref, original)
    disposition = replay("0-1", env, store)
    assert disposition is Disposition.POISON   # parked, not cycled forever

The round-trip test proves the queue is a faithful custodian of evidence; the idempotency test proves replay cannot inflate the record count and misrepresent how many inspections a vehicle actually had; the poison test proves a single malformed payload cannot become an infinite loop that starves the queue. Assert all three before this queue backs any production channel.

• Storing the reparse instead of the original bytes. It is tempting to dead-letter the partially-parsed dict that failed validation, because it is already in memory. Do not — the moment you discard the original bytes you lose the ability to re-derive the failure and, more importantly, the evidence § 396.3© requires you to retain. Always persist the exact bytes that arrived, keyed by their hash, and reference them from the envelope.
• PII leaking into the dead-letter queue and its alerts. The original payload contains driver identifiers and inspection detail. Keep those bytes only in the access-controlled blob store; the envelope that flows into streams, logs, and alert channels must carry the payload_ref and an error summary, never the raw payload. An alert that pastes a driver’s DVIR into a chat channel is a data-handling incident, not a convenience.
• Replay storms. Draining a large backlog by re-enqueuing every envelope at once re-creates the exact downstream overload that produced the failures — especially when the root cause was a throttled telematics API. Rate-limit replay and re-enqueue in bounded batches, coordinating with the concurrency limits in Async Batching for High-Volume Ingestion, so recovery does not become a self-inflicted second outage.
• Acknowledging before the replay is durable. If you XACK the dead-letter entry before the re-enqueue is confirmed, a crash in between loses the payload entirely — the failure you were protecting against. Re-enqueue first, confirm the write, then acknowledge.

• Exception Handling and Dead-Letter Routing — the classifier and retry policy that decide which payloads reach this queue.
• Async Batching for High-Volume Ingestion — the concurrency limits replay must respect to avoid a replay storm.
• WORM Retention and SHA-256 Hashing for DVIR Records — the write-once, hash-keyed store that holds the preserved payload bytes.
• Idempotent Webhook Retry and Signature Verification — the delivery-side idempotency pattern this replay logic mirrors.

Part of Exception Handling and Dead-Letter Routing, within the DVIR Ingestion & Digital/Paper Parsing Workflows architecture. Back to Exception Handling and Dead-Letter Routing.