Library Catalog Data Sync Pipelines

Recovering Dead-Letter Queue Messages in Catalog Ingestion

This page answers one narrow operational question: why do catalog-update messages that repeatedly fail simply disappear — the record never lands in the ILS, no exception reaches an operator, and the only trace is a dead-letter queue (DLQ) whose depth keeps climbing — and how do you build a recovery process that replays the recoverable ones and quarantines the rest? It sits under Async Batch Processing for Catalog Updates, which covers the broker-backed ingestion pipeline end to end, and within the broader Catalog Ingestion & ILS Sync Pipelines architecture. If you run any Celery- or RQ-backed catalog worker, a growing DLQ with nobody consuming it is a silent data-loss bug you will eventually discover the hard way — during a collection audit that finds records that should exist but do not.

Dead-letter recovery worker classifying failed catalog messages into replay or quarantine A dead-letter queue holding catalog-update messages that exhausted their retries feeds a recovery worker. The worker classifies each message's stored failure metadata into two dispositions. Transient failures — a database timeout, an ILS API 503, a broker connection reset — are replayed back onto the ingest queue with bounded exponential backoff. Permanent failures — malformed MARC, a schema-contract violation, an unresolvable foreign key — are routed to a quarantine store for human review. The replay path re-enters the ingest queue idempotently so a message that already committed is not double-applied. An append-only recovery log at the bottom records every classify, replay and quarantine decision. dead-letter queue retries exhausted depth climbing recovery worker read failure metadata classify each message transient vs permanent TRANSIENT → replay timeout · 503 · reset bounded backoff PERMANENT → quarantine bad MARC · schema violation human review ingest queue idempotent commit to ILS append-only recovery log every CLASSIFY · REPLAY · QUARANTINE decision recorded — each dead-letter message accounted for

Problem Framing

The symptom is silence. The catalog ingestion pipeline reports healthy — worker concurrency is nominal, the main queue drains, throughput graphs look fine — yet a subset of records that publishers or the ILS export layer emitted never appear in the catalog. There is no failed task alert because, from the broker’s perspective, nothing failed: the messages were retried the configured number of times, exhausted their attempts, and were routed to the dead-letter queue exactly as designed. The design just forgot to consume it.

The one metric that betrays the leak is DLQ depth. Left unmonitored it climbs monotonically, one poison message at a time. A quick check on the broker makes the accumulation visible:

text
$ redis-cli LLEN catalog_ingest.dlq
(integer) 4127

$ celery -A catalog inspect stats | grep -A2 dead_letter
  "dead_letter_routed": 4127,
  "dead_letter_consumed": 0        # nothing has ever drained this queue

$ redis-cli LINDEX catalog_ingest.dlq 0
{"task":"ingest_marc_record","bib_id":"991000482","args":["<base64 MARC>"],
 "failure":{"exc_type":"MarcLeaderError","attempts":5,"last_seen":"2026-07-14T02:11Z"}}

Four thousand records have been quietly dropped over months. Some are genuinely broken and belong in a review queue; many are victims of a database failover or an ILS API outage that lasted ninety seconds and has long since resolved. Both cases sit in the same undifferentiated pile, and neither is being processed.

Root Cause

There are two independent faults, and both have to be fixed.

The first is that there is no consumer for the dead-letter queue at all. Most Celery and RQ setups configure a DLQ as a safety net — task_reject_on_worker_lost, a retry ceiling, a dead-letter exchange — and then treat reaching it as the end of the message’s life. Routing to the DLQ is where the pipeline’s responsibility silently ends, so the queue becomes a write-only sink. Nothing replays it, nothing alerts on its depth, and the records inside it are, operationally, lost.

The second fault is that transient and permanent failures are not distinguished. A message lands in the DLQ for one of two fundamentally different reasons. It may carry a poison payload — malformed MARC that fails leader parsing, a record that violates the ingest schema validation contract, a foreign key that references a branch that does not exist — and no amount of retrying will ever make it succeed. Or it may be a perfectly valid message that happened to run during a transient dependency outage: a database failover, an ILS REST API returning 503, a broker connection reset. The retry ceiling treats both identically. Once the same five attempts are burned — whether against a genuinely broken record or against a momentarily unreachable database — the message is dead-lettered with no record of which kind of failure it was.

Blindly replaying the whole DLQ is therefore not a fix: it would re-run thousands of poison messages that will only fail and dead-letter again, an expensive loop that never terminates. The recovery process must read the failure metadata each message already carries, decide which class of failure it was, and act differently on each — replay the transient ones with backoff, quarantine the permanent ones for a human.

Solution

Build a dedicated recovery worker that drains the DLQ on a schedule, classifies each message by the exception recorded when it died, and routes it. Transient failures are re-enqueued onto the normal ingest queue with bounded exponential backoff and a hard replay ceiling; permanent failures are written to a durable quarantine store for human triage. Crucially, the replay is idempotent — a message that already committed on an earlier attempt must not be double-applied — and the original payload plus its full failure metadata are preserved so nothing is destroyed by the recovery step itself. This is the same quarantine discipline the parent Async Batch Processing for Catalog Updates pipeline applies to live failures, extended to the messages that already fell through.

python
from __future__ import annotations

import enum
import logging
from dataclasses import dataclass

logger = logging.getLogger("catalog.dlq.recovery")

# Exceptions that indicate a momentary dependency problem, not a bad payload.
# Replaying these after the dependency recovers is expected to succeed.
TRANSIENT_EXC: frozenset[str] = frozenset({
    "OperationalError",      # DB failover / connection reset
    "TimeoutError",
    "ConnectionResetError",
    "ILSServiceUnavailable",  # ILS REST API 5xx
    "BrokerConnectionError",
})

# Exceptions that mean the payload itself is invalid. Retrying is pointless.
PERMANENT_EXC: frozenset[str] = frozenset({
    "MarcLeaderError",
    "SchemaValidationError",
    "RecordDecodeError",
    "ForeignKeyViolation",
})

MAX_REPLAYS = 3  # bounded: a replayed message may fail at most this many more times


class Route(enum.Enum):
    REPLAY = "replay"
    QUARANTINE = "quarantine"


@dataclass(frozen=True)
class DeadLetter:
    bib_id: str
    payload: bytes
    exc_type: str
    attempts: int
    replays: int  # how many times the recovery worker has already re-enqueued it


def classify(dead: DeadLetter) -> Route:
    """Decide whether a dead-lettered message is worth replaying.

    Unknown exception types are treated as permanent and quarantined, so an
    unclassified failure fails closed into human review rather than looping.
    """
    if dead.replays >= MAX_REPLAYS:
        logger.warning(
            "replay_ceiling_reached",
            extra={"bib_id": dead.bib_id, "exc_type": dead.exc_type,
                   "replays": dead.replays},
        )
        return Route.QUARANTINE
    if dead.exc_type in TRANSIENT_EXC:
        return Route.REPLAY
    if dead.exc_type not in PERMANENT_EXC:
        logger.warning(
            "unclassified_failure_quarantined",
            extra={"bib_id": dead.bib_id, "exc_type": dead.exc_type},
        )
    return Route.QUARANTINE

The dispatch loop reads each dead letter, classifies it, and takes the routed action inside an explicit try/except so a failure to recover a message never itself loses the message:

python
from __future__ import annotations

import contextlib
from collections.abc import Iterator

from celery import Celery

app = Celery("catalog")


def recover_dead_letters(
    letters: Iterator[DeadLetter],
    *,
    quarantine: "QuarantineStore",
    backoff_base_s: float = 30.0,
) -> None:
    """Drain the DLQ once: replay transient failures, quarantine the rest."""
    for dead in letters:
        route = classify(dead)
        try:
            if route is Route.REPLAY:
                # Bounded exponential backoff keyed on prior replay count.
                countdown = backoff_base_s * (2 ** dead.replays)
                app.send_task(
                    "ingest_marc_record",
                    args=[dead.payload],
                    countdown=countdown,
                    headers={"bib_id": dead.bib_id, "replays": dead.replays + 1},
                )
                logger.info(
                    "dead_letter_replayed",
                    extra={"bib_id": dead.bib_id, "exc_type": dead.exc_type,
                           "countdown_s": countdown, "replay": dead.replays + 1},
                )
            else:
                with contextlib.closing(quarantine.begin()) as tx:
                    tx.store(dead)  # preserves payload + failure metadata verbatim
                logger.info(
                    "dead_letter_quarantined",
                    extra={"bib_id": dead.bib_id, "exc_type": dead.exc_type,
                           "attempts": dead.attempts},
                )
        except Exception:
            # Never drop a message because recovery failed; leave it on the DLQ.
            logger.exception(
                "recovery_failed_message_retained",
                extra={"bib_id": dead.bib_id, "exc_type": dead.exc_type},
            )
            raise

Two properties make this safe to run against a real backlog. First, ingest_marc_record must already be idempotent — keyed on bib_id with an upsert, so replaying a message whose earlier attempt actually committed (the failure came after the write) is a no-op rather than a duplicate record. Second, replay is bounded: the replays counter rides in the task headers, classify quarantines once it reaches MAX_REPLAYS, and the backoff widens each round, so a message that keeps failing a transient check cannot loop forever — it graduates to quarantine and a human sees it.

Compliance or Privacy Impact

A recovery worker touches full catalog payloads, and dead-lettered messages are the ones most likely to contain the odd record that carries patron-linked data — a hold note, a circulation-derived field, an acquisition record with a requester. The quarantine store therefore inherits the same handling obligations as any other persistent copy of ingest data. Two points matter in practice:

Handled this way, recovery narrows the risk surface rather than widening it: instead of thousands of records sitting indefinitely in an unmonitored broker queue, each is either committed, replayed, or held in a governed store with a retention clock.

Verification

Confirm classification is correct and that replay is both bounded and idempotent. The test asserts a transient exception is replayed, a permanent one is quarantined, an unknown one fails closed to quarantine, and a message at the replay ceiling stops replaying.

python
def _dead(exc_type: str, replays: int = 0) -> DeadLetter:
    return DeadLetter(
        bib_id="991000482", payload=b"<marc>", exc_type=exc_type,
        attempts=5, replays=replays,
    )


def test_transient_is_replayed() -> None:
    assert classify(_dead("OperationalError")) is Route.REPLAY


def test_permanent_is_quarantined() -> None:
    assert classify(_dead("MarcLeaderError")) is Route.QUARANTINE


def test_unknown_exception_fails_closed() -> None:
    # An exception type in neither set must never be replayed blindly.
    assert classify(_dead("SomeNewError")) is Route.QUARANTINE


def test_replay_ceiling_forces_quarantine() -> None:
    # Even a transient failure stops replaying once the ceiling is hit.
    assert classify(_dead("TimeoutError", replays=MAX_REPLAYS)) is Route.QUARANTINE

For the running pipeline, add a monitor on DLQ depth and on the ratio of dead_letter_consumed to dead_letter_routed: a DLQ whose depth trends upward while consumption stays flat is the exact signature this page exists to eliminate, and it should page rather than accumulate. After the first full drain, verify idempotency directly — replay a batch you know already committed and assert the catalog row count is unchanged, proving the upsert absorbed the duplicates rather than creating them.