Dead-Letter Queue Forensics for Vision Pipelines

This walkthrough sits under Error Handling in Computer Vision Pipelines and solves one precise task: turning a growing dead-letter queue (DLQ) of captures and detections that failed processing into a ranked list of root causes and a safe, repeatable replay. The parent layer owns the guards that route a bad frame into quarantine; this page owns what happens next, once thousands of those quarantined messages have piled up and someone has to answer why and which ones can we safely reprocess? Draining a DLQ by blindly re-submitting every message is the single most common way to turn one incident into two — a replay storm that re-fails the same poison messages, fires duplicate restock triggers, and buries the genuine signal. The disciplined alternative is forensic: enrich each message, bucket it by a stable failure signature, quantify the buckets to find the dominant cause, fix that cause, and only then replay through an idempotent harness gated against poison. Each step below is independently verifiable.

Prerequisites & Context Jump to heading#

Before you can run forensics, the DLQ has to carry enough context that a message is self-describing. Confirm the following are in place. The enrichment step assumes each message already speaks the platform’s typed vocabulary; if yours only holds a bare image_uri and a stack trace, fix the producer first — the same Error Handling in Computer Vision Pipelines layer that quarantines a frame is where those fields are stamped.

• Runtime: Python 3.11+ on the triage host; no GPU is needed to bucket and rank, only to replay through inference.
• A typed DLQ record: each message must carry the source image reference (image_uri), the stage that failed (ingestion, inference, postprocess), the reject_reason, the registry_revision (the model and catalog revision active at the moment of failure), the capture_timestamp, and the platform keys capture_id, store_id, fixture_id, and planogram_id. Without registry_revision you cannot tell a message that failed under a since-fixed model from one that will re-fail on replay.
• A durable broker with a DLQ topic. The queue semantics — ordering, redelivery, and the second-order “poison” topic this page writes to — come from Message Broker Patterns for Capture Events; read that first if your DLQ is an ad-hoc database table rather than a real broker sink.
• An idempotency ledger: a set or key-value store that records which messages have been successfully replayed, so a re-run of the harness is safe. This is the same guard pattern a webhook consumer needs, described in Real-Time Compliance Alerting & Webhooks.

A note on terms: a failure signature is the small tuple of fields that makes two failures “the same” for triage; a poison message is one that will re-fail every time it is replayed and must be held, not looped.

Step 1 — Enrich and Bucket Messages by Failure Signature Jump to heading#

Raw DLQ messages are noise until they are grouped. The unit of grouping is a failure signature: a stable, low-cardinality tuple that answers “same root cause?” without over-splitting. Signature on the axes that actually predict whether a fix and replay will work — the failing stage, the reject_reason, and the registry_revision — and deliberately exclude high-cardinality fields like capture_id or capture_timestamp, which would shatter every message into its own bucket.

from __future__ import annotations

from collections import defaultdict
from dataclasses import dataclass
from typing import Iterable, NewType

CaptureId = NewType("CaptureId", str)


@dataclass(frozen=True, slots=True)
class DeadLetter:
    """One message drained from the vision pipeline's dead-letter queue."""
    capture_id: CaptureId
    store_id: str
    fixture_id: str
    planogram_id: str
    image_uri: str
    stage: str                 # ingestion | inference | postprocess
    reject_reason: str         # schema_invalid | low_confidence | ...
    registry_revision: str     # model + catalog revision at failure time
    capture_timestamp: str     # ISO-8601, UTC
    attempts: int = 0
    trace_id: str = ""


@dataclass(frozen=True, slots=True)
class Signature:
    stage: str
    reject_reason: str
    registry_revision: str


def failure_signature(dl: DeadLetter) -> Signature:
    """Collapse a message to the axes that make two failures 'the same'."""
    return Signature(
        stage=dl.stage.strip().lower(),
        reject_reason=dl.reject_reason.strip().lower(),
        registry_revision=dl.registry_revision.strip(),
    )


def bucket_by_signature(
    messages: Iterable[DeadLetter],
) -> dict[Signature, list[DeadLetter]]:
    """Group every DLQ message under its failure signature."""
    buckets: dict[Signature, list[DeadLetter]] = defaultdict(list)
    for dl in messages:
        buckets[failure_signature(dl)].append(dl)
    return dict(buckets)

Normalizing case and whitespace in the signature matters: a producer that emits Low_Confidence in one release and low_confidence in the next would otherwise create two phantom buckets for one cause. Keep the signature deliberately coarse — if you later need to split, you can always sub-group a bucket, but you cannot un-shatter one that was signatured on a timestamp.

Step 2 — Quantify Each Bucket and Find the Dominant Cause Jump to heading#

With messages bucketed, rank them. Forensics is triage, and triage is Pareto: in a real fleet, one or two signatures usually account for the majority of the queue, and fixing that cause drains most of the backlog. Compute each bucket’s absolute count and its share of the total, sort descending, and keep an example capture_id per bucket so a human can pull one image to confirm the diagnosis.

@dataclass(frozen=True, slots=True)
class BucketStat:
    signature: Signature
    count: int
    share: float
    example_capture_id: CaptureId


def rank_buckets(buckets: dict[Signature, list[DeadLetter]]) -> list[BucketStat]:
    """Rank signatures by volume; the head of the list is the cause to fix first."""
    total = sum(len(msgs) for msgs in buckets.values())
    if total == 0:
        return []
    stats = [
        BucketStat(
            signature=sig,
            count=len(msgs),
            share=round(len(msgs) / total, 4),
            example_capture_id=msgs[0].capture_id,
        )
        for sig, msgs in buckets.items()
    ]
    return sorted(stats, key=lambda s: (s.count, s.signature.reject_reason), reverse=True)


def dominant_cause(stats: list[BucketStat], alert_share: float = 0.40) -> BucketStat | None:
    """Return the top bucket, and let the caller alert when one cause dominates."""
    if not stats:
        return None
    top = stats[0]
    if top.share >= alert_share:
        # One signature owns most of the queue — fix it before any replay.
        return top
    return top

The tie-break on reject_reason keeps the ranking deterministic when two buckets have equal counts, which matters when this runs on a schedule and you diff yesterday’s ranking against today’s. A dominant bucket whose signature is a stale registry_revision tells you the fix is a redeploy; a dominant schema_invalid at stage="ingestion" tells you a capture-app release changed a field — a completely different remediation, surfaced by the same ranking.

Step 3 — Build a Deterministic, Idempotent Replay Harness Jump to heading#

Only once the dominant cause is fixed do you replay — and replay must be safe to run twice. The guarantee comes from an idempotency key derived from the fields that define “the same work”: the capture_id, the stage, and the registry_revision it will be reprocessed under. Before re-submitting a message, the harness checks a ledger; if the key is already recorded, it skips. This is what makes a second run of the drain a no-op instead of a duplicate-verdict generator.

import hashlib
import logging

logger = logging.getLogger("cv.dlq.replay")


def idempotency_key(dl: DeadLetter) -> str:
    """Stable key: same capture, same stage, same revision == same work."""
    raw = f"{dl.capture_id}|{dl.stage.lower()}|{dl.registry_revision}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]


class ReplayLedger:
    """Records idempotency keys that have already been replayed successfully."""

    def __init__(self, seen: set[str] | None = None) -> None:
        self._seen: set[str] = seen or set()

    def already_done(self, key: str) -> bool:
        return key in self._seen

    def mark_done(self, key: str) -> None:
        self._seen.add(key)

Because the key folds in registry_revision, a capture that failed under an old model and is being replayed under a new one yields a different key — so a legitimate reprocess after a fix is not blocked, while an accidental double-drain of the identical work is. Persist the ledger in the same store as your compliance records so it survives a harness restart; an in-memory set that resets on deploy will happily replay everything again.

Step 4 — Gate Replay So Poison Messages Do Not Loop Jump to heading#

The final piece is the poison gate, and it is what separates forensics from a replay storm. A message becomes poison when it has been replayed to its max_attempts or when its reject_reason has not been cleared for replay — meaning the root cause behind that signature is still open. Poison messages are routed to a separate quarantine topic, held, and paged; they never re-enter the DLQ, so they cannot loop. Everything else replays through the failing stage exactly once and is marked done.

from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class ReplayPolicy:
    max_attempts: int = 3
    cleared_reasons: frozenset[str] = frozenset()  # signatures fixed and safe to replay


class PoisonQuarantine:
    """Terminal sink for messages that must not be replayed again."""

    def __init__(self, broker) -> None:
        self._broker = broker

    def hold(self, dl: DeadLetter, detail: str) -> None:
        self._broker.publish("cv.deadletter.poison", {
            "capture_id": dl.capture_id,
            "store_id": dl.store_id,
            "fixture_id": dl.fixture_id,
            "planogram_id": dl.planogram_id,
            "stage": dl.stage,
            "reject_reason": dl.reject_reason,
            "registry_revision": dl.registry_revision,
            "attempts": dl.attempts,
            "detail": detail,
        })


def replay_message(
    dl: DeadLetter,
    *,
    stage_fn,
    ledger: ReplayLedger,
    quarantine: PoisonQuarantine,
    policy: ReplayPolicy,
) -> str:
    """Replay one message idempotently, behind the poison gate."""
    key = idempotency_key(dl)
    if ledger.already_done(key):
        logger.info("skip %s: already replayed under %s", dl.capture_id, dl.registry_revision)
        return "skipped_idempotent"

    if dl.attempts >= policy.max_attempts:
        quarantine.hold(dl, f"attempts {dl.attempts} >= max {policy.max_attempts}")
        return "poison_attempts"

    if policy.cleared_reasons and dl.reject_reason.lower() not in policy.cleared_reasons:
        # Root cause for this signature is still open; replaying would just re-fail.
        quarantine.hold(dl, f"reason '{dl.reject_reason}' not cleared for replay")
        return "poison_uncleared"

    try:
        stage_fn(dl.image_uri)
    except Exception as exc:  # a fresh failure — do not mark done, do not loop
        logger.warning("replay failed for %s: %r", dl.capture_id, exc)
        return "replay_failed"

    ledger.mark_done(key)
    logger.info("replayed %s through %s", dl.capture_id, dl.stage)
    return "replayed"

The cleared_reasons allow-list is the operational safety catch: an operator explicitly names the signatures whose root cause is fixed, and only those replay. Draining the whole DLQ therefore becomes “fix the dominant cause, add its reason to cleared_reasons, run the harness” — a controlled, auditable loop rather than a firehose.

Verification & Testing Jump to heading#

Confirm each rule deterministically rather than trusting a queue-depth graph:

1. Buckets are stable across identical input. Run bucket_by_signature twice on the same messages and assert the bucket keys and counts are identical; then feed the same reason in mixed case (Low_Confidence vs low_confidence) and assert they land in one bucket, not two.
2. Ranking is deterministic. Assert rank_buckets returns buckets in descending count order and that two equal-count buckets always sort the same way via the reject_reason tie-break, so a scheduled run is diffable day over day.
3. Replay is idempotent. Call replay_message twice for the same DeadLetter and assert the second call returns "skipped_idempotent" and that stage_fn was invoked exactly once — a call counter on a fake stage_fn proves no duplicate side effect.
4. Attempt cap poisons, does not loop. Pass a message with attempts equal to max_attempts and assert it returns "poison_attempts", stage_fn was never called, and the quarantine received exactly one record.
5. Uncleared reasons are held. With a non-empty cleared_reasons that excludes the message’s reason, assert "poison_uncleared" and that nothing re-enters the DLQ topic.

A healthy drain shows the replayed count climbing while poison_* stays flat and the dominant bucket’s share falls on the next forensic pass — the sign the fix, not the replay, drained the queue.

Troubleshooting Jump to heading#

Related Jump to heading#

• Error Handling in Computer Vision Pipelines — the parent layer that stamps the typed, self-describing records this DLQ triage depends on
• Message Broker Patterns for Capture Events — the broker and dead-letter-topic semantics, including the poison topic this harness writes to
• Real-Time Compliance Alerting & Webhooks — the same idempotent-replay and poison-guard pattern applied to a webhook consumer’s dead-letter path