Designing a Dead-Letter Queue for Failed Filing Retrievals #

This guide is part of the Error Categorization & Retry Logic area within the Secretary of State Portal & API Ingestion framework. It covers the terminal boundary of the retry pipeline: what happens to a filing retrieval after every retry has been spent and the response still is not usable. The one behaviour the compliance layer forbids is a silent drop — an entity that quietly falls out of a status sweep because its retrieval failed and nobody was told. A dead-letter queue is how the system guarantees that every exhausted attempt is captured, visible, and replayable rather than lost.

Scope of This Page #

This page covers the design and implementation of the dead-letter queue (DLQ) for failed retrievals: the envelope schema that preserves the original request, its classification, and its full attempt history; the idempotency key that dedupes re-enqueued failures so one struggling entity does not flood the queue; the replay guard that bounds how many times an operator or scheduler may re-drive a message before it is quarantined as a poison message; and the operator visibility that makes the backlog inspectable. It assumes the response has already been labelled by the deterministic classifier in classifying transient vs. terminal errors from state portals and that a transient attempt has already been paced by the exponential backoff engine. It excludes the retry mechanics themselves; the DLQ is where the pipeline lands after those are exhausted.

The Constraint: A Delinquent Entity Must Never Be Silently Dropped #

A dropped retrieval is not a lost log line — it is an entity whose compliance status is now unknown, sitting in a portfolio that will be certified as complete. If a Delaware corporation’s good-standing check fails and the attempt vanishes, the sweep reports the portfolio clean while a franchise-tax delinquency under 8 Del. C. § 502 accrues penalty and 1.5% monthly interest under § 510, unseen. California’s suspension path under Cal. Corp. Code § 2205 and New York’s grace-free biennial statement under N.Y. BCL § 408 make the same point: the cost of a missed retrieval lands on the entity’s statutory clock, not the pipeline’s. The DLQ exists to convert an exhausted attempt into a durable, inspectable obligation so that a human or a scheduled replay always closes the loop before a deadline does. Every message carries enough context — request, classification, attempt history — to be reconstructed and re-driven under audit, which is also what SOX internal-control expectations require of a compliance record.

Prerequisites #

• Python 3.10+ — for X | Y unions, match on message state, and modern dataclasses.
• A durable store behind the queue — Redis, SQS, or a Postgres table with an append-only audit companion. The module below defines a small store interface and an in-memory reference implementation so it runs as-is.
• Standard library: json, logging, hashlib, time, uuid, dataclasses, enum.
• An upstream Classification (class + verdict) from the error classifier, so the envelope records why the retrieval failed, not merely that it did.
• A replay driver — the callable that re-executes a retrieval — supplied by the caller, so the DLQ owns lifecycle and idempotency while the caller owns transport.

Implementation: An Envelope, an Idempotency Key, and a Replay Guard #

The diagram traces one exhausted attempt through its full lifecycle: it is wrapped in an envelope, enqueued under an idempotency key that dedupes repeats, made visible for triage, and then either replayed to resolution or — once the replay guard is exhausted — quarantined as a poison message.

The module below is the complete DLQ: a DeadLetterEnvelope, a small durable-store interface with an in-memory reference implementation, a writer that dedupes on the idempotency key, and a consumer that enforces the replay guard and quarantines poison messages. Comments mark the compliance-critical lines.

from __future__ import annotations

import hashlib
import json
import logging
import time
import uuid
from dataclasses import dataclass, field, asdict
from enum import Enum
from typing import Callable, Protocol

# Structured JSON logging — every lifecycle transition is an auditable event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("ingestion.dead_letter")


class MessageState(str, Enum):
    QUEUED = "queued"          # awaiting triage or scheduled replay
    RESOLVED = "resolved"      # a replay retrieved a usable response — loop closed
    QUARANTINED = "quarantined"  # poison message — replay guard exhausted, needs a human


@dataclass(frozen=True)
class AttemptRecord:
    ts: float
    error_class: str   # from the upstream classifier — why this attempt failed
    status: int


@dataclass
class DeadLetterEnvelope:
    """Everything needed to reconstruct and re-drive one failed retrieval, under audit."""
    idempotency_key: str
    entity_id: str
    jurisdiction: str
    request: dict[str, str]              # method, url, and the minimal request identity
    error_class: str                     # terminal classification at exhaustion
    attempts: list[AttemptRecord] = field(default_factory=list)
    replay_count: int = 0
    state: MessageState = MessageState.QUEUED
    first_seen: float = field(default_factory=time.time)
    message_id: str = field(default_factory=lambda: str(uuid.uuid4()))

    @staticmethod
    def idempotency_key_for(entity_id: str, jurisdiction: str, request: dict[str, str]) -> str:
        # Stable identity: the same failed retrieval always hashes to the same key,
        # so re-enqueuing a struggling entity dedupes instead of flooding the queue.
        signature = json.dumps(
            {"e": entity_id, "j": jurisdiction, "m": request.get("method"), "u": request.get("url")},
            sort_keys=True,
        )
        return hashlib.sha256(signature.encode("utf-8")).hexdigest()


class DeadLetterStore(Protocol):
    def get(self, key: str) -> DeadLetterEnvelope | None: ...
    def put(self, env: DeadLetterEnvelope) -> None: ...
    def delete(self, key: str) -> None: ...
    def pending(self) -> list[DeadLetterEnvelope]: ...


class InMemoryStore:
    """Reference store; swap for Redis/SQS/Postgres in production without touching callers."""
    def __init__(self) -> None:
        self._data: dict[str, DeadLetterEnvelope] = {}

    def get(self, key: str) -> DeadLetterEnvelope | None:
        return self._data.get(key)

    def put(self, env: DeadLetterEnvelope) -> None:
        self._data[env.idempotency_key] = env

    def delete(self, key: str) -> None:
        self._data.pop(key, None)

    def pending(self) -> list[DeadLetterEnvelope]:
        # Operator visibility: the backlog, oldest first, for triage by age and class.
        return sorted(
            (e for e in self._data.values() if e.state is MessageState.QUEUED),
            key=lambda e: e.first_seen,
        )


class DeadLetterQueue:
    """Writer + consumer with idempotent enqueue and a bounded replay guard."""

    def __init__(self, store: DeadLetterStore, *, max_replays: int = 3) -> None:
        self._store = store
        self._max_replays = max_replays  # replay budget before a message is quarantined

    def enqueue(self, env: DeadLetterEnvelope) -> DeadLetterEnvelope:
        """Idempotent write: a failed retrieval already in the queue is merged, not duplicated."""
        existing = self._store.get(env.idempotency_key)
        if existing is not None:
            # Same entity/request already dead-lettered — append attempt history, do not re-add.
            existing.attempts.extend(env.attempts)
            self._store.put(existing)
            logger.info(json.dumps({
                "event": "dlq_duplicate_merged", "key": env.idempotency_key[:12],
                "entity_id": env.entity_id, "attempts": len(existing.attempts),
            }))
            return existing
        self._store.put(env)
        logger.warning(json.dumps({
            "event": "dlq_enqueued", "key": env.idempotency_key[:12],
            "entity_id": env.entity_id, "jurisdiction": env.jurisdiction,
            "error_class": env.error_class,
        }))
        return env

    def replay(self, key: str, driver: Callable[[DeadLetterEnvelope], bool]) -> MessageState:
        """Re-drive one message through `driver`; enforce the replay guard and poison quarantine."""
        env = self._store.get(key)
        if env is None or env.state is not MessageState.QUEUED:
            return MessageState.RESOLVED if env is None else env.state

        # REPLAY GUARD: a message that has burned its replay budget is a poison message.
        # Quarantine it for a human rather than looping forever and starving the queue.
        if env.replay_count >= self._max_replays:
            env.state = MessageState.QUARANTINED
            self._store.put(env)
            logger.error(json.dumps({
                "event": "dlq_quarantined", "key": key[:12], "entity_id": env.entity_id,
                "replay_count": env.replay_count, "reason": "replay_budget_exhausted",
            }))
            return MessageState.QUARANTINED

        env.replay_count += 1
        succeeded = driver(env)  # caller owns transport; the DLQ owns lifecycle + idempotency
        if succeeded:
            env.state = MessageState.RESOLVED
            self._store.delete(key)  # loop closed — remove so it never re-drives
            logger.info(json.dumps({
                "event": "dlq_resolved", "key": key[:12], "entity_id": env.entity_id,
                "replay_count": env.replay_count,
            }))
            return MessageState.RESOLVED

        # Failed replay: persist the incremented count so the guard converges on quarantine.
        env.attempts.append(AttemptRecord(time.time(), env.error_class, 0))
        self._store.put(env)
        logger.warning(json.dumps({
            "event": "dlq_replay_failed", "key": key[:12], "entity_id": env.entity_id,
            "replay_count": env.replay_count, "max_replays": self._max_replays,
        }))
        return MessageState.QUEUED

    def snapshot(self) -> list[dict[str, object]]:
        """Operator-facing view of the backlog for dashboards and audit export."""
        return [asdict(e) for e in self._store.pending()]

The queue is deliberately narrow: it owns the envelope, the idempotency key, and the replay guard, and nothing else. Transport is injected as a driver callable so the DLQ never re-implements the retrieval it is protecting, and the store is a Protocol so Redis, SQS, or an append-only Postgres table drops in without changing a caller. A resolved message is deleted so it can never re-drive; a poison message is quarantined, not deleted, so an operator can still see and investigate it.

Configuration Reference #

Every knob is chosen for a compliance reason, not a performance one.

Failure Modes and Fallback Routing #

Each DLQ failure maps back to the parent area’s error categorization & retry logic taxonomy and has a defined remediation.

1. A single entity floods the queue with repeated failures (operational overload). Without a stable idempotency key, every retry cycle would enqueue a fresh message for the same entity, burying real work. The sha256(entity · jurisdiction · request) key makes enqueue merge repeats into one envelope, appending attempt history so the record grows in depth, not in count.
2. A poison message loops forever (system). A retrieval that fails deterministically — a decommissioned endpoint, a permanently malformed request — would otherwise consume the replay budget indefinitely. The replay guard quarantines it after max_replays, moving it to a manual-review state and freeing the queue. Quarantine, not deletion, preserves the record for engineering triage.
3. A statutory block is dead-lettered as if it were transient (statutory). If the upstream classifier mislabels a franchise-tax hold as transient, its exhausted attempt arrives here carrying the wrong error_class. Because the envelope preserves the full classification and attempt history, an operator can see the misclassification during triage and route the entity to the penalty-avoidance path where priority scoring algorithms rank it against its grace window, rather than replaying a request that can never succeed.
4. A replay succeeds transiently but the response is unusable (data-validation). A driver that returns True on a bare 200 would mark a wrapped error as resolved. The success path requires the driver to re-classify the replayed response — the same content-first check from classifying transient vs. terminal errors from state portals — so a 200 HTML page keeps the message queued rather than falsely closing the loop.
5. The process restarts with messages in flight (durability). The in-memory reference store loses state on restart; production must back the DeadLetterStore protocol with Redis, SQS, or a Postgres table so an exhausted retrieval survives a deploy. The idempotency key guarantees that re-enqueuing after recovery is safe and non-duplicating.

Frequently Asked Questions #

Related #

• Error Categorization & Retry Logic
• Classifying Transient vs. Terminal Errors From State Portals
• Async Polling & Rate Limiting
• Priority Scoring Algorithms
• Secretary of State Portal & API Ingestion