Revenue Management Pricing & forecasting pipelines
5 min read Ingestion

Quarantining invalid records with a dead-letter queue

When a record fails validation, you have two bad options and one good one. Dropping it loses data; blocking the pipeline on it stalls every good record behind it. The good option is a dead-letter queue: divert the failure, keep the stream moving, and hold the bad record for inspection and replay. This guide builds that quarantine sink, implementing a core mechanism of Data Quality & Schema Contracts in the Data Ingestion & OTA API Integration Workflows pillar.

Prerequisites

  • Python 3.11+
  • Standard library json, logging, dataclasses, datetime
  • A validation function that returns a typed record or a structured error, as built for the schema contract
  • A durable store for quarantined items (a table or queue; a JSONL file stands in here)
  • A replay path that can re-run validation after a contract fix

Step 1 — Define the dead-letter envelope

A quarantined item is worthless without context. Wrap the raw payload with the reason it failed, the contract version in force, and a timestamp, so a later reviewer can diagnose and replay it.

python 
from __future__ import annotations

import json
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone

logger = logging.getLogger("ingestion.dlq")


@dataclass(frozen=True)
class DeadLetter:
    raw_payload: dict
    reason: str
    contract_version: str
    quarantined_at: str        # ISO-8601 UTC

    @staticmethod
    def create(raw_payload: dict, reason: str, contract_version: str,
               now: datetime) -> "DeadLetter":
        return DeadLetter(
            raw_payload=raw_payload,
            reason=reason,
            contract_version=contract_version,
            quarantined_at=now.astimezone(timezone.utc).isoformat(),
        )

Step 2 — Route failures to the queue without stopping the batch

Process a batch by validating each record and diverting failures, so one malformed row never blocks the rest. The good records go forward; the bad ones land in the dead-letter sink with their reason attached.

python 
def process_batch(records: list[dict], validate, contract_version: str,
                  now: datetime) -> tuple[list, list[DeadLetter]]:
    accepted, dead = [], []
    for raw in records:
        try:
            accepted.append(validate(raw))
        except ValueError as exc:
            letter = DeadLetter.create(raw, str(exc), contract_version, now)
            dead.append(letter)
            logger.warning("dead-lettered %s: %s", raw.get("property_id"), exc)
    if dead:
        logger.info("batch committed %d, quarantined %d", len(accepted), len(dead))
    return accepted, dead


def persist_dead_letters(dead: list[DeadLetter], path: str) -> None:
    with open(path, "a", encoding="utf-8") as fh:
        for letter in dead:
            fh.write(json.dumps(asdict(letter)) + "\n")

Step 3 — Replay after a contract fix

The point of quarantine, versus deletion, is recovery. When a contract bug is fixed or a provider reverts a change, re-run validation over the queue and promote whatever now passes, leaving the rest for the next fix.

python 
def replay(path: str, validate) -> tuple[list, int]:
    recovered, still_bad = [], 0
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            letter = json.loads(line)
            try:
                recovered.append(validate(letter["raw_payload"]))
            except ValueError:
                still_bad += 1
    logger.info("replay recovered %d, still invalid %d", len(recovered), still_bad)
    return recovered, still_bad

Replay is what turns a data-quality incident into a recoverable event: after you ship the contract fix, the backlog drains itself instead of being lost.

Verification and testing

python 
from datetime import datetime


def _validate(rec: dict) -> dict:
    if rec.get("base_rate_cents", 0) <= 0:
        raise ValueError("non-positive rate")
    return rec


def test_batch_splits_and_replays(tmp_path) -> None:
    now = datetime(2026, 8, 1, tzinfo=timezone.utc)
    batch = [{"property_id": "h1", "base_rate_cents": 18000},
             {"property_id": "h2", "base_rate_cents": 0}]      # invalid
    accepted, dead = process_batch(batch, _validate, "v3", now)
    assert len(accepted) == 1 and len(dead) == 1

    path = tmp_path / "dlq.jsonl"
    persist_dead_letters(dead, str(path))
    # after a "fix", the same record still fails; replay reports it
    recovered, still_bad = replay(str(path), _validate)
    assert recovered == [] and still_bad == 1

Common pitfalls and edge cases

  • Silent drops. Catching the error and moving on without persisting loses the record; always write it to the durable sink.
  • No reason captured. A quarantined payload without its failure reason is un-diagnosable; store the exact validation error.
  • Missing contract version. Without the version in force, you cannot tell whether a fix should recover the item; stamp it.
  • Unbounded growth. A queue that only fills signals an unaddressed upstream break; alert on depth and oldest-item age.
  • Replay without idempotency. Re-promoting recovered records must use the same idempotent upsert as normal ingestion, or replay duplicates data.