Fault Correlation Ticket Routing Automation

Replaying Dead-Letter Queue Events Safely

A dead-letter queue (DLQ) is where a fault pipeline puts the events it could not safely process: a syslog line that failed schema construction, a trap whose OID had no mapping, an event whose category was ambiguous during a bad ontology reload. Quarantining those events is the right call in the moment — it keeps the primary correlation stream clean and unblocked. But quarantine is not a graveyard. After the upstream parser is fixed, the missing rule is authored, or the schema is migrated, those events often represent real faults that were never ticketed. Replaying them is how a NOC recovers the signal it deferred.

The danger is that a naive replay is far more destructive than the original failure. Replaying a two-hour-old DLQ backlog straight back into ingestion can re-open incidents that operators already resolved, page on-call engineers for outages that ended, and — if a single poison message throws on every attempt — spin an infinite retry loop that starves the live pipeline. A safe replay is therefore not a for loop over the DLQ; it is a controlled, idempotent, rate-bounded re-ingestion with a hard cap on how many times any one message may be tried. Done wrong, replay adds hours to mean time to resolution (MTTR) by burying real alarms under resurrected noise; done right, it recovers deferred faults with zero duplicate tickets.

Schema alignment and taxonomy anchor

This page is the recovery path for Error Categorization Pipelines, the deterministic-first classification stage of the Ingestion & Parsing Workflows data plane that quarantines any event it cannot confidently tag. That stage is the single owner of malformed-payload and low-confidence triage; this reference owns the narrower job of getting those quarantined events back into the pipeline safely once the condition that quarantined them is resolved.

Every DLQ entry is stored with its original raw payload, the ValidationError or classification reason that quarantined it, and the schema_version in force when it failed. Replay reconstructs the canonical record defined in Event Schema Design and re-submits it through the same categorization path a live event would take — never a side channel — so a replayed interface-error event is classified by exactly the logic described in Categorizing Network Interface Errors Automatically. The idempotency that makes this safe is anchored on the raw_payload_hash already carried by every canonical event, which is what lets the pipeline recognize a replayed event as the same fault it deferred rather than a new one.

Safe dead-letter replay gating flowA quarantined event is read from the dead-letter queue. A schema-version gate checks whether the event schema matches the current version; stale events are re-quarantined for later migration. Passing events reach an idempotency check against a shared state store keyed by raw payload hash; an already-seen key is suppressed as a duplicate and never re-ingested. New keys reach a poison-cap gate that compares the attempt count against a maximum; events over the cap are parked to a poison queue for manual inspection. Events under the cap are re-ingested through the normal categorization path with exponential backoff between attempts.DLQ readraw + reasonschemacurrent?yesseenbefore?noundercap?yesRe-ingestnormal path · backoffstalere-quarantineyessuppress dupnopark: poison queue

An idempotent, capped replay worker

The replay worker below reads from the DLQ, applies four gates in order — schema-version, idempotency, poison-cap, backoff — and re-submits survivors through the same categorization entry point a live event uses. The idempotency store is a shared key-value set of already-processed payload hashes with a TTL, so a replay cannot re-open an incident that a concurrent live event already created.

import asyncio
import logging
import time
from enum import Enum
from typing import Any, Optional

from pydantic import BaseModel, ConfigDict, Field, ValidationError

logger = logging.getLogger("dlq_replay")

CURRENT_SCHEMA_VERSION = 3
MAX_ATTEMPTS = 5              # poison-message cap: never retry forever
IDEMPOTENCY_TTL_SEC = 6 * 3600


class ReplayOutcome(str, Enum):
    REINGESTED = "reingested"
    DUPLICATE = "duplicate"
    SCHEMA_STALE = "schema_stale"
    POISONED = "poisoned"


class DeadLetter(BaseModel):
    # The envelope the categorization stage wrote when it quarantined the event.
    model_config = ConfigDict(strict=True, extra="forbid")

    raw_payload_hash: str
    raw: dict[str, Any]
    reason: str
    schema_version: int
    attempts: int = Field(default=0, ge=0)
    first_seen: float = Field(gt=0)


class IdempotencyStore:
    """Async wrapper over a shared key-value set of processed payload hashes."""

    def __init__(self, redis) -> None:
        self._redis = redis  # redis.asyncio client, injected for testability

    async def seen(self, key: str) -> bool:
        return bool(await self._redis.exists(f"replay:seen:{key}"))

    async def mark(self, key: str) -> None:
        # SET NX with a TTL: the first writer wins, later replays see a duplicate.
        await self._redis.set(f"replay:seen:{key}", "1",
                              nx=True, ex=IDEMPOTENCY_TTL_SEC)


async def replay_one(
    entry: DeadLetter,
    store: IdempotencyStore,
    reingest,                # async callable: submits into the live categorizer
    poison_queue: asyncio.Queue,
) -> ReplayOutcome:
    """Apply every safety gate to a single quarantined event."""

    # Gate 1 — schema version. An event captured under an older schema must not
    # be forced through today's model; hold it for a dedicated migration pass.
    if entry.schema_version != CURRENT_SCHEMA_VERSION:
        logger.info("hash=%s held: schema v%d != v%d",
                    entry.raw_payload_hash, entry.schema_version,
                    CURRENT_SCHEMA_VERSION)
        return ReplayOutcome.SCHEMA_STALE

    # Gate 2 — idempotency. If this fault was already processed (by a live event
    # or an earlier replay), re-ingesting it would open a duplicate ticket.
    if await store.seen(entry.raw_payload_hash):
        return ReplayOutcome.DUPLICATE

    # Gate 3 — poison cap. A message that has failed MAX_ATTEMPTS times is not
    # transient; parking it stops an infinite loop from starving the pipeline.
    if entry.attempts >= MAX_ATTEMPTS:
        await poison_queue.put(entry.model_dump())
        logger.warning("hash=%s parked after %d attempts",
                       entry.raw_payload_hash, entry.attempts)
        return ReplayOutcome.POISONED

    # Gate 4 — bounded retry with exponential backoff. Re-submit through the
    # SAME categorization path a live event takes, never a bypass.
    delay = min(0.5 * (2 ** entry.attempts), 30.0)  # capped backoff
    await asyncio.sleep(delay)
    try:
        await reingest(entry.raw)
    except (ValidationError, RuntimeError) as exc:
        entry.attempts += 1
        logger.warning("hash=%s replay attempt %d failed: %s",
                       entry.raw_payload_hash, entry.attempts, exc)
        raise  # caller re-enqueues with the incremented attempt count

    # Mark only AFTER a successful re-ingest so a crash mid-flight retries safely.
    await store.mark(entry.raw_payload_hash)
    return ReplayOutcome.REINGESTED

Ordering the gates matters. The schema check runs first because there is no point deduplicating an event that cannot be modelled; idempotency runs before the poison cap because a duplicate should be suppressed cheaply without consuming an attempt; the backoff runs last because it is the only gate that costs wall-clock time. Marking the idempotency key only after a successful re-ingest is deliberate — if the worker crashes between reingest and mark, the event is simply retried, never lost.

Async processing hook

Replay must never contend with live ingestion for the correlation engine’s capacity. The driver below paces itself with its own bounded concurrency and yields to the event loop between entries, so a large backlog drains in the background without starving fresh faults.

async def replay_driver(
    dlq_source: asyncio.Queue,
    store: IdempotencyStore,
    reingest,
    poison_queue: asyncio.Queue,
    max_concurrency: int = 8,   # cap replay's share of the categorizer
) -> dict[str, int]:
    """Drain the DLQ under bounded concurrency, re-enqueuing transient failures."""
    sem = asyncio.Semaphore(max_concurrency)
    tally: dict[str, int] = {o.value: 0 for o in ReplayOutcome}

    async def worker(entry: DeadLetter) -> None:
        async with sem:  # bound how much of the live categorizer replay may use
            try:
                outcome = await replay_one(entry, store, reingest, poison_queue)
                tally[outcome.value] += 1
            except Exception:
                # Transient failure: put it back with the higher attempt count
                # so exponential backoff widens on the next pass.
                await dlq_source.put(entry)

    async with asyncio.TaskGroup() as tg:
        while not dlq_source.empty():
            raw = await dlq_source.get()
            entry = raw if isinstance(raw, DeadLetter) else DeadLetter.model_validate(raw)
            tg.create_task(worker(entry))
            await asyncio.sleep(0)  # keep the loop responsive to live traffic
    return tally

The semaphore is the throttle that keeps replay a background citizen: it never lets more than max_concurrency re-ingestions run at once, so a 50,000-entry backlog cannot flood the categorizer that live faults also depend on. Because every gate and every I/O call is awaited, a single event loop drains the backlog while continuing to serve fresh telemetry.

Mitigation and hardening

Replay introduces its own failure modes, and each needs a concrete containment path:

  1. Duplicate incidents. The idempotency store is the primary defense: a payload hash already marked within the TTL window is suppressed, so a replayed event cannot re-open an incident a live event already created. Size the TTL to at least the longest realistic gap between a fault and its resolution — six hours is a sound default for transport incidents — so a slow-resolving fault is still recognized as the same one on replay.
  2. Poison-message loops. Without the attempt cap, a payload that throws on every re-ingest is retried forever, and its exponential backoff eventually dominates the worker. Capping attempts and parking the message to a poison queue for human inspection converts an infinite loop into a bounded, alertable backlog. Alert when the poison queue grows, because a rising park rate means a class of events no fix has addressed.
  3. Schema-version skew. An event quarantined under an older schema may not construct against today’s model, and forcing it through would either fail or, worse, coerce into a subtly wrong record. Gating on schema_version holds stale events for an explicit migration pass rather than mangling them, keeping the replay path honest about what it can safely process.
  4. Resurrected noise. Even valid replays can page for outages that already ended. Stamp each re-ingested event with a replayed=true marker and its original event_time, so the correlation and routing tiers can suppress notification for events whose fault window has clearly closed while still recording them for audit.

Operational hardening notes

Tuning a replay pipeline is a balance between draining fast enough to recover deferred signal and slowly enough to never disturb live ingestion. The backoff base and cap set the pace for transient failures: a 0.5-second base doubling to a 30-second ceiling means a genuinely flaky downstream is retried patiently without hammering it, while the max_concurrency semaphore caps replay’s share of the categorizer so the live p99 evaluation latency never regresses during a drain. Keep the poison cap low — five attempts is plenty, because anything failing five times with widening backoff is structural, not transient — so poison messages surface quickly instead of consuming capacity. Run replay during a quiet window when possible, but never gate it on human scheduling for security-relevant events, because a quarantined spoofed-source trap is a signal that should be reviewed promptly, not left dormant. Instrument the outcome tally as counters — reingested, duplicate, schema_stale, poisoned — and watch the ratio: a healthy replay is mostly reingested with a thin tail of duplicate, while a spike in schema_stale means a migration is overdue and a spike in poisoned means a defect class needs an engineer, not another retry. Held to these controls, DLQ replay recovers the faults the pipeline responsibly deferred with zero duplicate tickets and no measurable impact on the sub-2-second live ticket-creation budget.

Frequently Asked Questions

How does idempotent replay avoid opening duplicate tickets?

Every canonical event carries a raw payload hash, and the replay worker checks that hash against a shared idempotency store before re-ingesting. If the hash was already marked, either by a live event or an earlier replay within the time-to-live window, the event is suppressed as a duplicate and never re-enters the pipeline. The key is marked only after a successful re-ingest, so a crash mid-flight simply retries rather than losing the event or double-counting it.

What is a poison-message cap and why is it necessary?

A poison-message cap is a hard limit on how many times a single quarantined event may be retried, typically about five attempts. Without it, an event that throws on every re-ingest is retried forever, and its exponential backoff eventually dominates the worker and starves live traffic. When an event reaches the cap it is parked to a separate poison queue for human inspection instead of being retried again, converting an infinite loop into a bounded, alertable backlog.

Why gate replay on schema version?

An event quarantined under an older schema version may not construct against the current model, and forcing it through risks either a hard failure or a silent coercion into a subtly wrong record that then misroutes. The schema-version gate compares each entry against the current version and holds mismatched events for a dedicated migration pass rather than replaying them blind, so the replay path never mangles an event it cannot faithfully model.

How does replay avoid disturbing live ingestion?

Replay runs under a bounded concurrency semaphore that caps how many re-ingestions execute at once, so it can never consume more than a fixed share of the categorizer that live faults also use. Between entries the driver yields to the event loop, and exponential backoff spaces out retries of transient failures. Together these keep a large backlog draining quietly in the background while fresh telemetry continues to meet its latency budget.