Configuring Dead-Letter Queues for Failed Ingest Tasks

A retry policy protects you from transient faults; it does nothing for a task that fails the same way every time. When a corrupt manifest or an unreadable preservation master enters the Async Task Queuing for Batches stage of the Automated Ingestion & Batch Scanning Workflows pipeline, exhausting max_retries simply replays the same deterministic failure — hammering the storage array on every attempt and, once retries are spent, either dropping the message or wedging the worker. This page shows how to give those poison messages a deliberate resting place: a dead-letter exchange and a quarantine queue that capture the failed payload and its failure reason as a durable preservation event, then hand a human a review-and-replay workflow instead of letting an irreplaceable master vanish or retry-storm the archive.

# Root-Cause Analysis: Why Terminal Failures Need Their Own Path

A dead-letter path only earns its complexity because a specific class of failure cannot be resolved by waiting and trying again. These are the concrete causes that produce poison messages in a digitization queue:

1. Deterministic decode failures. A truncated or byte-corrupt master TIFF raises the same struct.error or libtiff read failure on every attempt. No amount of backoff repairs the bytes on disk, so each retry is wasted I/O against the storage array.
2. Schema-invalid manifests. A manifest missing color_profile or carrying a non-numeric resolution_dpi fails validation identically forever. This is the class the Batch Validation Schemas reject at the boundary — but a task that slips past the boundary check still needs somewhere terminal to land.
3. Unresolvable format identification. A file whose PUID cannot be resolved, or whose signature conflicts with its extension, blocks promotion and will never self-heal without a human decision.
4. Serialization poison. A message whose body cannot be deserialized crashes the worker before the task body runs, so autoretry_for never even engages — the broker keeps redelivering a message that kills every consumer that touches it.
5. Exhausted transient retries. A genuinely transient fault (a storage timeout) that outlives its retry budget is indistinguishable, at exhaustion, from a permanent one. Once max_retries is spent it must be treated as terminal and escalated, exactly as Error Handling & Retry Logic prescribes for partial batch failures.

The unifying property is that redelivery is pointless and actively harmful: it consumes broker capacity, saturates storage bandwidth, and — with early acknowledgment — risks silently discarding the payload. The state machine below shows the escape hatch: a message that exhausts its retries is dead-lettered to a quarantine queue where it rests, is inspected, and is either replayed after a fix or discarded with a recorded reason. This is the natural continuation of the reliability discipline established in Implementing Celery for Asynchronous Ingestion Tasks, which guarantees a task is never lost — this page decides what happens when it can never succeed.

A terminal failure is dead-lettered to quarantine, recorded as a preservation event, then either replayed after a human fix or discarded with a documented reason — never retried into a storm.

# Configuring the RabbitMQ Dead-Letter Exchange

On an AMQP broker, dead-lettering is a native queue feature. A message is dead-lettered when it is rejected with requeue=false, when its per-message TTL expires, or when it exceeds a queue length limit. You declare the main ingest queue with a dead-letter exchange (x-dead-letter-exchange) and a routing key (x-dead-letter-routing-key) that point at a quarantine queue. The quarantine queue itself is durable and has no consumer wired to auto-retry — it is an inbox for human review.

# Declare the dead-letter (quarantine) exchange + queue first.
rabbitmqadmin declare exchange name=ingest.dlx type=direct durable=true
rabbitmqadmin declare queue name=ingest.quarantine durable=true \
  arguments='{"x-queue-type":"quarantine"}'
rabbitmqadmin declare binding source=ingest.dlx destination=ingest.quarantine \
  routing_key=quarantined

# Declare the MAIN queue so terminal rejects flow to the DLX.
# x-message-ttl bounds how long a stuck message lingers before it is dead-lettered.
rabbitmqadmin declare queue name=ingest.main durable=true arguments='{
  "x-dead-letter-exchange": "ingest.dlx",
  "x-dead-letter-routing-key": "quarantined",
  "x-message-ttl": 86400000
}'

The equivalent declaration in Python via pika keeps the same arguments under version control so every environment provisions an identical topology:

import logging
import pika

logger = logging.getLogger("archival.ingest.dlq")


def declare_ingest_topology(connection: pika.BlockingConnection) -> None:
    """Provision the main ingest queue and its dead-letter quarantine path."""
    channel = pika.BlockingConnection.channel(connection)
    channel.exchange_declare(exchange="ingest.dlx", exchange_type="direct", durable=True)
    channel.queue_declare(queue="ingest.quarantine", durable=True)
    channel.queue_bind(
        queue="ingest.quarantine", exchange="ingest.dlx", routing_key="quarantined"
    )
    channel.queue_declare(
        queue="ingest.main",
        durable=True,
        arguments={
            "x-dead-letter-exchange": "ingest.dlx",
            "x-dead-letter-routing-key": "quarantined",
            "x-message-ttl": 86_400_000,  # 24h in ms — cap on a stuck message's lifetime
        },
    )
    logger.info(
        "ingest_topology_declared",
        extra={"main_queue": "ingest.main", "quarantine_queue": "ingest.quarantine"},
    )

When a consumer calls basic_nack(delivery_tag, requeue=False) on a poison message, RabbitMQ moves it — headers, body, and all — to ingest.quarantine and stamps an x-death header recording the reason, original queue, and count. That header is the raw material for the preservation event below.

# Routing Terminal Failures to Quarantine in Python

The consumer must distinguish a transient exception (retry) from a terminal one (quarantine), and on quarantine it must emit a structured record capturing why before the message leaves its hands. The record is the same evidence an auditor needs, so it doubles as a PREMIS event: quarantining an irreplaceable master is a preservation action, and it emits a PREMIS event so the object’s provenance shows exactly when and why it was set aside.

import datetime as dt
import hashlib
import json
import logging
import uuid
from dataclasses import asdict, dataclass

import pika

logger = logging.getLogger("archival.ingest.dlq")


class TerminalIngestError(Exception):
    """Raised for deterministic failures that must not be retried."""


@dataclass(frozen=True)
class QuarantineRecord:
    """A durable, PREMIS-aligned record of why a task was dead-lettered."""

    event_id: str
    event_type: str
    event_datetime: str
    object_identifier: str
    failure_reason: str
    exception_class: str
    retry_count: int
    payload_sha256: str
    outcome: str = "quarantined"


def build_quarantine_record(
    payload: bytes, object_id: str, exc: Exception, retry_count: int
) -> QuarantineRecord:
    """Capture the failed payload and reason as a preservation event."""
    return QuarantineRecord(
        event_id=str(uuid.uuid4()),
        event_type="quarantine",
        event_datetime=dt.datetime.now(dt.timezone.utc).isoformat(),
        object_identifier=object_id,
        failure_reason=str(exc),
        exception_class=type(exc).__name__,
        retry_count=retry_count,
        payload_sha256=hashlib.sha256(payload).hexdigest(),
    )


def on_ingest_message(
    channel: "pika.channel.Channel",
    method: "pika.spec.Basic.Deliver",
    properties: "pika.spec.BasicProperties",
    body: bytes,
) -> None:
    """Process one ingest message; dead-letter terminal failures to quarantine."""
    headers = properties.headers or {}
    object_id = headers.get("object_id", "unknown")
    retry_count = int(headers.get("x-retry-count", 0))
    try:
        process_master(json.loads(body))
        channel.basic_ack(delivery_tag=method.delivery_tag)
        logger.info("ingest_ok", extra={"object_id": object_id})
    except TerminalIngestError as exc:
        record = build_quarantine_record(body, object_id, exc, retry_count)
        persist_preservation_event(record)  # durable store, not the result backend
        # requeue=False routes the message through x-dead-letter-exchange to quarantine.
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        logger.warning(
            "ingest_quarantined",
            extra={**asdict(record)},
        )
    except (ConnectionError, TimeoutError) as exc:
        # Transient: let the retry/backoff layer handle redelivery.
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
        logger.info(
            "ingest_retry",
            extra={"object_id": object_id, "reason": str(exc), "retry_count": retry_count},
        )

The persist_preservation_event call must write to durable storage — an append-only ledger or database — and never to the Celery/Redis result backend, which expires. That distinction is what keeps the quarantine record admissible as chain-of-custody evidence.

# The Redis and Celery Equivalent

Redis has no native dead-letter exchange, so Celery’s application layer supplies the same guarantee. Catch the terminal exception, publish the payload and reason to a dedicated quarantine queue with send_task, and let the original task succeed (ack) so the broker does not redeliver the poison. A custom Task.on_failure hook centralizes this so no task can forget it.

import logging

from celery import Celery, Task

logger = logging.getLogger("archival.ingest.dlq")
app = Celery("archival_ingest")


class QuarantiningTask(Task):
    """Base task that dead-letters terminal failures to a quarantine queue."""

    autoretry_for = (ConnectionError, TimeoutError)
    max_retries = 3

    def on_failure(
        self, exc: Exception, task_id: str, args: tuple, kwargs: dict, einfo: object
    ) -> None:
        """Fires only after retries are exhausted or on a non-retryable error."""
        if isinstance(exc, TerminalIngestError) or self.request.retries >= self.max_retries:
            app.send_task(
                "archival.tasks.hold_in_quarantine",
                kwargs={
                    "object_id": kwargs.get("object_id", "unknown"),
                    "reason": str(exc),
                    "exception_class": type(exc).__name__,
                    "original_task_id": task_id,
                },
                queue="quarantine",
            )
            logger.warning(
                "celery_task_quarantined",
                extra={"task_id": task_id, "reason": str(exc)},
            )

Because quarantine has no worker consuming it for reprocessing, a message parked there waits for a human. A reviewer inspects the reason, repairs the manifest or re-fetches the master, and replays it by re-submitting to the main queue with a fresh object_id correlation — closing the loop drawn in the state machine.

# Validation and Verification

Confirm the dead-letter path behaves before you trust it with irreplaceable material:

• Force a poison message. Publish a deliberately corrupt manifest and confirm the message lands in ingest.quarantine after exactly one processing attempt — not max_retries attempts. Inspect the x-death header (rabbitmqadmin get queue=ingest.quarantine) and verify it names the original queue and reason.
• Assert the preservation event exists. Query the durable event store for the quarantine event and confirm its payload_sha256 matches sha256 of the message body you published. A missing or mismatched digest means the record cannot prove what was quarantined.
• Verify no retry storm. Watch storage-array I/O during the test; a correctly configured terminal failure produces one read attempt, not a saturating loop. Compare rabbitmqctl list_queues name messages on the main queue before and after — the poison message should leave the main queue, not cycle in it.
• Test replay idempotency. Replay a repaired message and confirm it completes exactly once and links back to the original quarantine event, so provenance shows the full fail-then-succeed history.

# Edge Cases and Gotchas

Three archival-specific traps deserve emphasis. First, a multi-page bound-volume TIFF may fail only on page 400 — the failure is terminal for that object but the record must name the byte offset, or a reviewer cannot reproduce it. Second, proprietary scanner extensions (a vendor’s private TIFF tag) can trip a strict validator; decide once whether such tags are terminal or tolerated, and encode that decision in the classifier rather than leaving it to each worker. Third, never let a quarantine queue double as an archival store — it holds a copy of a payload pending a decision, and the durable preservation event, not the queued message, is the system of record. For the AMQP semantics, consult the RabbitMQ Dead Letter Exchanges reference, and the Python json module documentation for hardening the deserialization boundary.

# Frequently Asked Questions

# When should a task go to the dead-letter queue instead of retrying?

Retry only when the fault is plausibly transient — a storage timeout, a dropped connection, a momentarily unreachable service — and bound those retries with backoff. Route to quarantine the moment the failure is deterministic (a corrupt master, a schema-invalid manifest, an unresolvable format) or the moment a transient retry budget is exhausted. The test is simple: if trying again with identical inputs would fail identically, it belongs in quarantine, because further retries only burn broker and storage capacity without any chance of success.

# What exactly should the quarantine record capture?

Enough to reconstruct and adjudicate the failure without the original worker: a unique event identifier, the object identifier, an ISO-8601 UTC timestamp, the exception class and message, the retry count at failure, and a SHA-256 digest of the failed payload. The digest lets a reviewer prove the quarantined bytes are the exact bytes that failed. Because setting a master aside is itself a preservation action, this record is emitted as a PREMIS event so the object’s provenance chain shows the quarantine and, later, any replay or discard.

# Why not just retry poison messages a few more times?

Because a deterministic failure has a zero-percent success rate no matter how many times you retry, and each attempt costs a full read of the master against the storage array plus broker round-trips. At batch scale this becomes a retry storm that starves healthy work and can trip storage rate limits. Worse, with early acknowledgment an exhausted retry can silently discard the payload. A dead-letter path converts that harmful loop into a single, recorded, recoverable event.

# How do I safely replay a quarantined ingest task?

Fix the underlying cause first — repair the manifest, re-fetch a clean master, or update the format ruleset — then re-submit the payload to the main ingest queue as a fresh task correlated to the original quarantine event. Do not simply requeue the untouched message: nothing changed, so it will dead-letter again. Confirm the replay completes exactly once and that its success event links back to the quarantine event, giving auditors the complete fail-then-recover history.

# Related

• Async Task Queuing for Batches — the parent stage: queue contract and broker guarantees this dead-letter path extends.
• Error Handling & Retry Logic — the backoff and idempotent-retry policies that decide when a failure becomes terminal.
• Emitting PREMIS Events for Preservation Actions — how a quarantine record is recorded as a durable, auditable preservation event.