Dead-Letter Queue Patterns for Failed Royalty Records

Not every ingestion failure should be retried, and treating every failure as retryable is how a single malformed DSP payload turns into a wedged pipeline that blocks an entire reporting period’s payout run. Some records — a stream event with a truncated ISRC, a split-percentage field that sums to 140%, a territory code that doesn’t exist in the ISO table — are permanently unprocessable in their current form and need to be quarantined, not resubmitted. Building a dead-letter queue as a first-class part of Error Handling & Retry Mechanisms gives royalty operations teams a durable, queryable holding area for these records, complete with the failure reason and the original payload, so they can be fixed and replayed without losing a single dollar of unreconciled revenue.

Dead-Letter Storage Trade-offs

Where a rejected record lands — a message queue topic, a relational table, or object storage — changes how quickly operators can triage failures and how safely records can be redriven.

A relational DLQ table is usually the right default for royalty pipelines because triage-by-query matters more than raw throughput; queue-based DLQs earn their complexity only once ingestion volume exceeds what a single Postgres instance can comfortably absorb.

Prerequisites & Assumptions

Examples use Python 3.11+, pydantic>=2, httpx, and a Postgres-backed quarantine table reachable via psycopg or SQLAlchemy. Records entering the DLQ are assumed to have already failed validation inside the ingestion path described in Implementing Exponential Backoff for Failed API Syncs — that is, they exhausted retries or failed a validation check that backoff alone cannot fix.

Implementation

Step 1: Define the Quarantine Schema

The quarantine record needs to preserve enough context to diagnose and replay the failure without re-fetching from the DSP, which may no longer have the same data available.

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Any

class QuarantinedRecord(BaseModel):
    batch_id: str
    source_payload: dict[str, Any]      # the raw, unmodified record as received
    failure_reason: str                  # human-readable summary
    error_code: str = Field(pattern=r"^[A-Z_]+$")   # e.g. "ISRC_MALFORMED"
    territory_code: str | None = None
    reporting_period: str | None = None
    first_seen_at: datetime
    retry_count: int = 0
    status: str = "quarantined"          # quarantined | redriven | resolved | discarded

Keeping source_payload as an untouched dict — not a partially-parsed model — is what makes redrive possible after the underlying bug is fixed, since the original bytes are still available to re-run through a corrected parser.

Step 2: Route Failures to the Dead-Letter Table

Wrap the validation boundary so that any ValidationError writes a QuarantinedRecord instead of silently discarding the row.

from pydantic import ValidationError
import sqlalchemy as sa

def ingest_with_dlq(raw_record: dict, batch_id: str, engine: sa.Engine) -> None:
    from myapp.models import RoyaltyLineItem  # the strict, production schema

    try:
        RoyaltyLineItem.model_validate(raw_record)
    except ValidationError as exc:
        quarantined = QuarantinedRecord(
            batch_id=batch_id,
            source_payload=raw_record,
            failure_reason=str(exc.errors()[0]["msg"]),
            error_code="SCHEMA_VALIDATION_FAILED",
            territory_code=raw_record.get("territory_code"),
            reporting_period=raw_record.get("reporting_period"),
            first_seen_at=datetime.utcnow(),
        )
        with engine.begin() as conn:
            conn.execute(
                sa.text("""
                    INSERT INTO royalty_dlq
                    (batch_id, source_payload, failure_reason, error_code,
                     territory_code, reporting_period, first_seen_at, retry_count, status)
                    VALUES (:batch_id, :source_payload, :failure_reason, :error_code,
                            :territory_code, :reporting_period, :first_seen_at, 0, 'quarantined')
                """),
                quarantined.model_dump(mode="json"),
            )

Step 3: Alert on Dead-Letter Growth

A DLQ that grows silently is a payout delay waiting to be discovered during a monthly close. Poll the quarantine table on a schedule and alert when the growth rate crosses a threshold.

def check_dlq_growth(engine: sa.Engine, window_minutes: int = 60, alert_threshold: int = 500) -> bool:
    with engine.connect() as conn:
        count = conn.execute(
            sa.text("""
                SELECT COUNT(*) FROM royalty_dlq
                WHERE status = 'quarantined'
                  AND first_seen_at > NOW() - INTERVAL '1 minute' * :window
            """),
            {"window": window_minutes},
        ).scalar_one()
    if count > alert_threshold:
        # emit to your paging system, tagged with the dominant error_code for triage
        return True
    return False

Group the alert by error_code and territory_code so on-call engineers see immediately whether the spike is one systemic bug (e.g., a DSP changed its territory code format) or scattered, unrelated failures.

Step 4: Redrive After a Fix

Once the root cause is patched, replay quarantined records through the corrected validation path rather than manually re-keying them.

def redrive_dlq(engine: sa.Engine, error_code: str, engine_ingest_fn) -> int:
    redriven = 0
    with engine.begin() as conn:
        rows = conn.execute(
            sa.text("SELECT id, source_payload, batch_id FROM royalty_dlq "
                     "WHERE status = 'quarantined' AND error_code = :code"),
            {"code": error_code},
        ).fetchall()
        for row in rows:
            try:
                engine_ingest_fn(row.source_payload, row.batch_id)
                conn.execute(
                    sa.text("UPDATE royalty_dlq SET status='resolved', retry_count=retry_count+1 "
                             "WHERE id=:id"),
                    {"id": row.id},
                )
                redriven += 1
            except Exception:
                conn.execute(
                    sa.text("UPDATE royalty_dlq SET retry_count=retry_count+1 WHERE id=:id"),
                    {"id": row.id},
                )
    return redriven

Verification & Validation

After a redrive, reconcile the count of resolved records against the count of new rows that landed in the gold-tier royalty table for the same batch_id, and re-run split-sum assertions on the newly ingested rows to confirm right_share_pct totals still resolve to 100% per work. Record the redrive operation — operator, error_code, row count, and timestamp — as an entry in the append-only audit log, since a payout recalculation triggered by a redrive must be traceable back to the specific quarantine batch that caused it. Track dlq_resolution_rate (resolved ÷ total quarantined per period) as a standing metric; a rate that stays flat despite fixes shipping usually means the redrive job isn’t actually re-validating against the corrected code path.

Edge Cases & Gotchas

Poison records that fail redrive repeatedly will otherwise loop forever if retry_count isn’t checked before each attempt — cap redrive attempts (e.g., 3) and move records past that threshold to a terminal discarded status requiring manual intervention. Redriving without an idempotency key is the most dangerous failure mode: if the corrected ingestion path re-inserts a record that a parallel process already recovered manually, the same stream event gets counted twice and inflates a payout; always key redrive inserts on a deterministic identifier (isrc + reporting_period + dsp_transaction_id) with an upsert, never a blind insert. source_payload frequently contains payee banking or contact fields that count as sensitive data, so the quarantine table needs the same field-level encryption and RBAC boundary as the production tables, not an exemption because “it’s just quarantine.” Finally, a quarantine schema written for one DSP’s payload shape will start rejecting valid redrive attempts once that DSP changes its ERN namespace or field names — version the source_payload alongside a schema_version column so old quarantined rows don’t silently fail a redrive against a newer parser.

Related

Records typically reach the dead-letter table only after exhausting the retry budget described in Implementing Exponential Backoff for Failed API Syncs, and once resolved they flow into the same partitioned storage covered in the broader Error Handling & Retry Mechanisms reference.