Replaying Dead-Letter Queues for AFC Event Streams

When an automated fare collection (AFC) tap event fails schema validation, dropping it loses revenue and swallowing it hides a producer bug — so the only safe move is to capture it in a dead-letter queue (DLQ) with the exact failure reason, fix the root cause, then replay it. This page builds that loop: a DLQ record that preserves the raw payload and error, a classifier that separates transient schema failures from permanently poison payloads, and a replayer keyed on an event hash so redelivering a fixed tap can never double-count a fare. It is the recovery path for Schema Validation Pipelines inside the Fare Data Ingestion & GTFS-RT Sync pipeline.

The stakes are asymmetric. A tap that never replays is fare revenue that vanishes from month-end close; a tap that replays twice is a rider charged twice and a settlement discrepancy the clearinghouse bounces back. Idempotency is therefore not a nicety here — it is the property that makes replay safe to run repeatedly, including after a crash mid-drain.

Replay flow

The lifecycle below traces one event from validation failure to either successful re-entry or permanent quarantine. The idempotency set is the gate that makes the whole loop replayable: an event whose hash was already settled is a silent no-op, so re-running the drain from the start is harmless.

Step 1 — The dead-letter record

A DLQ entry has one job: preserve enough context to replay the event later without guessing. That means the raw payload exactly as received, the machine-readable failure reason, a stable content hash for idempotency, and a bounded retry count so a genuinely poison record cannot loop forever. The hash is computed over the business identity of the tap — event id, device, timestamp, fare — not over the whole noisy payload, so a replay after stripping an extra key still matches the original.

import hashlib
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional

logger = logging.getLogger("afc_dlq")


class FailureClass(str, Enum):
    TRANSIENT_SCHEMA = "transient_schema"   # fixable: extra key, coercible type
    POISON = "poison"                        # unfixable: no identity, corrupt


def event_identity_hash(payload: dict[str, Any]) -> str:
    """Stable SHA-256 over the tap's business identity, not the raw envelope.

    Keying on identity means a replay of the same tap after an upstream fix
    (e.g. a stripped telemetry key) hashes identically and dedupes cleanly.
    """
    identity = {
        "event_id": payload.get("event_id"),
        "device_id": payload.get("device_id"),
        "tap_timestamp": payload.get("tap_timestamp"),
        "fare_amount_cents": payload.get("fare_amount_cents"),
    }
    canonical = json.dumps(identity, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


@dataclass
class DeadLetterRecord:
    raw_payload: dict[str, Any]
    failure_reason: str
    failure_class: FailureClass
    event_hash: str = ""
    retry_count: int = 0
    first_seen_utc: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def __post_init__(self) -> None:
        if not self.event_hash:
            self.event_hash = event_identity_hash(self.raw_payload)

The event_hash is the linchpin. Because it is derived from the tap’s identity fields rather than the full JSON, the record survives the exact fix most replays need — removing an undocumented vendor key or coercing a stringy timestamp — while still colliding with any earlier copy of the same tap.

Step 2 — Capturing failures into the DLQ

Capture happens at the validation boundary. A ValidationError is classified: if the payload carries a usable identity it is TRANSIENT_SCHEMA and eligible for replay after a fix; if it has no event_id at all, it is POISON and goes straight to quarantine because nothing keys it. The main stream only ever sees events that validated.

from pydantic import BaseModel, ValidationError


def classify_failure(payload: dict[str, Any], exc: ValidationError) -> FailureClass:
    # No stable identity means we can neither dedupe nor trace it — poison.
    if not payload.get("event_id") or not payload.get("device_id"):
        return FailureClass.POISON
    return FailureClass.TRANSIENT_SCHEMA


def capture_to_dlq(
    raw_events: list[dict[str, Any]],
    model: type[BaseModel],
) -> tuple[list[BaseModel], list[DeadLetterRecord]]:
    """Validate a batch; route valid events onward and failures into the DLQ."""
    settled: list[BaseModel] = []
    dead_letters: list[DeadLetterRecord] = []

    for payload in raw_events:
        try:
            settled.append(model.model_validate(payload))
        except ValidationError as exc:
            reason = "; ".join(
                f"{'.'.join(str(p) for p in e['loc'])}: {e['msg']}"
                for e in exc.errors()
            )
            record = DeadLetterRecord(
                raw_payload=payload,
                failure_reason=reason,
                failure_class=classify_failure(payload, exc),
            )
            logger.warning("DLQ capture [%s] hash=%s reason=%s",
                           record.failure_class.value, record.event_hash[:12], reason)
            dead_letters.append(record)

    return settled, dead_letters

Every captured record carries the human-readable reason string, so when an analyst opens the DLQ they see exactly which field violated which rule — fare_amount_cents: Input should be a valid integer — rather than a bare “rejected” flag. That reason is what tells you whether the fix belongs in the producer firmware, the model, or an upstream adapter.

Step 3 — The idempotent replayer

Replay re-validates each transient record against the (now fixed) model. The idempotency set holds the hashes of every event already settled; a record whose hash is present is skipped as a no-op, so re-running the drain — after a crash, or just cautiously twice — never double-settles. A record that still fails validation has its retry count bumped and is moved to poison once it exhausts its budget.

class DeadLetterReplayer:
    def __init__(self, model: type[BaseModel], max_retries: int = 3) -> None:
        self.model = model
        self.max_retries = max_retries
        self._settled_hashes: set[str] = set()   # idempotency guard
        self.replayed: list[BaseModel] = []
        self.poison: list[DeadLetterRecord] = []

    def prime_settled(self, hashes: set[str]) -> None:
        """Seed the idempotency set from the ledger's already-settled hashes."""
        self._settled_hashes |= hashes

    def replay(self, records: list[DeadLetterRecord]) -> list[DeadLetterRecord]:
        """Attempt replay; return records still pending (not settled, not poison)."""
        pending: list[DeadLetterRecord] = []
        for record in records:
            if record.failure_class is FailureClass.POISON:
                self.poison.append(record)
                continue

            # Idempotency: a hash already settled is a no-op, never a re-charge.
            if record.event_hash in self._settled_hashes:
                logger.info("Replay skip (already settled) hash=%s", record.event_hash[:12])
                continue

            try:
                event = self.model.model_validate(record.raw_payload)
            except ValidationError as exc:
                record.retry_count += 1
                if record.retry_count >= self.max_retries:
                    record.failure_class = FailureClass.POISON
                    record.failure_reason = f"exhausted retries: {exc.error_count()} errors"
                    self.poison.append(record)
                    logger.error("Replay -> poison hash=%s", record.event_hash[:12])
                else:
                    pending.append(record)
                continue

            self.replayed.append(event)
            self._settled_hashes.add(record.event_hash)   # record before ack
            logger.info("Replay settled hash=%s", record.event_hash[:12])

        return pending

Priming the set from the ledger’s already-settled hashes (prime_settled) is what makes replay safe against events that were partially processed before the fix landed: if the tap already reached settlement through some other path, its hash is present and the replay skips it. Recording the hash before acknowledging the main-stream write closes the crash window — a redelivery after a crash finds the hash and no-ops.

Validation & Test Cases

Three behaviors must hold: a fixable entry re-enters the stream, a duplicate replay of a settled entry is a no-op, and a poison entry stays quarantined no matter how often you drain.

from decimal import Decimal
from pydantic import ConfigDict, Field, field_validator


class ReplayTapEvent(BaseModel):
    model_config = ConfigDict(extra="forbid")
    event_id: str
    device_id: str
    fare_amount_cents: int = Field(ge=0)
    tap_timestamp: str


# Capture: one event fails on an undocumented key, one has no identity (poison).
bad_batch = [
    {"event_id": "evt-1", "device_id": "V1", "fare_amount_cents": 275,
     "tap_timestamp": "2026-07-03T08:00:00Z", "battery_pct": 90},   # fixable
    {"fare_amount_cents": 275, "tap_timestamp": "2026-07-03T08:00:00Z"},  # no id -> poison
]
settled, dlq = capture_to_dlq(bad_batch, ReplayTapEvent)
assert len(settled) == 0 and len(dlq) == 2
assert {r.failure_class for r in dlq} == {FailureClass.TRANSIENT_SCHEMA, FailureClass.POISON}

# Fix the transient record upstream: strip the rogue telemetry key.
fixable = next(r for r in dlq if r.failure_class is FailureClass.TRANSIENT_SCHEMA)
fixable.raw_payload.pop("battery_pct")

replayer = DeadLetterReplayer(ReplayTapEvent)

# 1. Fixable entry re-enters the main stream.
pending = replayer.replay(dlq)
assert len(replayer.replayed) == 1
assert len(replayer.poison) == 1           # the identity-less record stayed poison
assert pending == []

# 2. Duplicate replay is a no-op — no second settlement, hash already recorded.
replayer.replay(dlq)
assert len(replayer.replayed) == 1         # unchanged, not 2

# 3. Poison entry never replays, even primed with a clean model.
assert all(r.failure_class is FailureClass.POISON for r in replayer.poison)

The second assertion is the one that protects the rider: replaying the same DLQ list twice leaves exactly one settled event because the first pass recorded its event_hash, so the second pass sees a known hash and skips. The poison record with no event_id is never eligible — there is no key to dedupe it on and no way to trace it, so it stays quarantined for a human to inspect.

Edge Cases & Gotchas

• Prime from the ledger, not just from this run. If the process restarts, an in-memory _settled_hashes is empty; seed it from the ledger’s settled hashes on startup or a crash mid-drain can re-settle. In production back the set with a durable store (a Redis set or a unique index on the ledger).
• Hash the identity, not the timestamp of receipt. Keying on ingestion time defeats deduplication because a replay arrives at a new wall-clock instant. Key only on the tap’s own fields.
• Cap retries and alert on poison growth. A steadily growing poison quarantine is a producer that never got fixed; alert on its rate, not just its size.
• Preserve the original failure reason across retries. Overwriting it on each attempt loses the diagnostic trail; append or keep the first reason so forensics can see the original violation.
• Replay in bounded batches. Draining a million-entry DLQ in one transaction risks the same OOM the streaming validator was built to avoid; page through it.

Integration Note

This replay loop is the drain valve for the dead-letter path that Schema Validation Pipelines fills — the operational checklist there warns that a DLQ you cannot drain is a data-loss incident in slow motion, and this is the drain. It pairs directly with Migrating AFC Event Models to Pydantic v2: a stricter v2 contract will initially reject more payloads into the DLQ, so the replay path is exactly how you re-drive that backlog once the producer or model fix lands. The event_hash here is the same idempotency discipline the reconciliation engine uses to dedupe redelivered taps, so a replayed event settles on one identity end to end.

FAQ

Related

• Schema Validation Pipelines — the parent component whose rejected records fill this DLQ
• Migrating AFC Event Models to Pydantic v2 — a stricter contract that sends more payloads down this replay path
• Implementing Pydantic Models for AFC Event Streams — the validation gate that builds the audit records replayed here
• GTFS-RT Realtime Sync — a sibling stage with its own dead-letter table and replay job
• AFC System Security Boundaries — masking and retention rules for the payloads a DLQ preserves

Part of Schema Validation Pipelines, within Fare Data Ingestion & GTFS-RT Sync.