Designing Dead-Letter Queue Retry for DSR Workers

A dead-letter queue (DLQ) is the difference between a Data Subject Request pipeline that can prove it never lost a sub-task and one that quietly drops an unfinished erasure into the void. Within the broader Async Polling & Queue Management design, and the wider Cross-System Data Discovery & Sync stage it sits in, the DLQ is where a worker sends a message it cannot process after a bounded number of attempts — not to abandon it, but to isolate it for repair and replay. This page is about the design and routing of that queue: the topic layout, the metadata a dead-lettered message must carry, poison-message isolation, and the replay path that re-injects a fixed task without duplicating a completed one. It deliberately does not cover retry timing — how long to wait between attempts and how to keep that within the statutory clock is the separate concern of Configuring exponential backoff within SLA budgets. For monitoring the backlog that feeds these workers, see Monitoring Kafka consumer lag for DSR SLAs; and for the distributed execution itself, Implementing Celery for async polling.

The stakes are specific: an erasure sub-task under GDPR Article 17 that vanishes because a worker crashed on a malformed record is not a lost log line — it is a subject whose data was never deleted and a controller that cannot demonstrate it fulfilled the request. The DLQ exists so that no DSR sub-task can ever silently disappear.

Prerequisites #

• Python 3.11+ for datetime.UTC and PEP 604 unions.
• Pydantic v2 (pydantic>=2.6) for the RetryEnvelope that travels with every message.
• A broker with a separate durable queue or topic for dead letters — a Celery/RabbitMQ dsr.dlq.manual queue, a Kafka dsr.dlq topic, or an SQS queue with a RedrivePolicy. The design below is broker-agnostic; only the publish call changes.
• A durable store for the envelope metadata (the same append-only evidence store the audit trail uses), so a dead-lettered erasure is provably retained, not merely parked in memory.

This page assumes the worker topology and priority queues from Implementing Celery for async polling are already in place; here we add the routing rules that decide when a task leaves the retry loop.

Step-by-step implementation #

Step 1 — Model the retry envelope #

Every message a DSR worker handles must carry enough context to be replayed and audited without reaching back into a database. That means the payload plus a header block: the request_id, the legal_basis the request is being fulfilled under, the current attempt count, the cap, and the last error. Losing any of these turns a DLQ entry into an unactionable mystery. Model the envelope with Pydantic v2 so an incomplete one is rejected at the boundary.

from datetime import datetime, UTC
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator


class RetryEnvelope(BaseModel):
    """Header block carried with every DSR sub-task through retry and dead-letter."""

    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    request_id: str
    source_system: str
    legal_basis: Literal["gdpr_art_15", "gdpr_art_17", "ccpa_1798_105", "ccpa_1798_100"]
    action: Literal["access", "deletion", "rectification"]
    attempt: int = Field(default=0, ge=0)
    max_attempts: int = Field(default=5, ge=1)
    last_error: str | None = None
    first_seen_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

    @field_validator("request_id", "source_system")
    @classmethod
    def not_blank(cls, v: str) -> str:
        """A dead-lettered task with no request_id cannot be traced back to a subject."""
        if not v:
            raise ValueError("request_id and source_system are mandatory")
        return v

    @property
    def exhausted(self) -> bool:
        """True once this task has used its whole attempt budget."""
        return self.attempt >= self.max_attempts

Step 2 — Route on failure: re-enqueue or dead-letter #

The handler owns exactly one decision — has this task exhausted its attempt budget? If not, increment attempt and re-enqueue. If so, publish the envelope to the DLQ. The rule that makes this safe for compliance is that both branches persist the message somewhere durable before the original is acknowledged: there is no window in which a task exists in neither the work queue nor the DLQ.

from collections.abc import Callable


def handle_failure(
    envelope: RetryEnvelope,
    error: Exception,
    requeue: Callable[[RetryEnvelope], None],
    to_dlq: Callable[[RetryEnvelope], None],
) -> str:
    """Route a failed sub-task to a bounded retry or the dead-letter queue.

    Returns the action taken ("requeued" or "dead_lettered") for the audit log.
    The caller must acknowledge the original message only after this returns.
    """
    stamped = envelope.model_copy(
        update={"attempt": envelope.attempt + 1, "last_error": repr(error)[:500]}
    )
    if stamped.exhausted:
        to_dlq(stamped)            # durable publish before the original is acked
        return "dead_lettered"
    requeue(stamped)               # durable publish before the original is acked
    return "requeued"

Step 3 — Isolate poison messages separately #

A poison message is one that crashes the consumer itself — a malformed body, an unparseable header — rather than merely failing an upstream call. It must not be allowed to consume its full retry budget, because each attempt takes down the worker and can wedge the whole consumer group. Track a crash count keyed by the message’s delivery id and short-circuit to the DLQ well before max_attempts.

def is_poison(crash_count: int, poison_threshold: int = 2) -> bool:
    """True once a message has crashed the consumer enough times to isolate it.

    A poison message bypasses the normal attempt budget and goes straight to the
    DLQ, so one unparseable body cannot repeatedly kill the consumer group.
    """
    return crash_count >= poison_threshold


def consume(envelope: RetryEnvelope, crash_count: int, *, process, requeue, to_dlq) -> str:
    """Process one message, isolating poison before it can exhaust a worker pool."""
    if is_poison(crash_count):
        to_dlq(envelope.model_copy(update={"last_error": "poison: repeated consumer crash"}))
        return "dead_lettered_poison"
    try:
        process(envelope)
        return "ok"
    except Exception as exc:  # noqa: BLE001 — the router classifies, not swallows
        return handle_failure(envelope, exc, requeue, to_dlq)

Step 4 — Replay from the DLQ after a fix #

A DLQ that only fills up is a leak with extra steps. The design is complete only when a fixed task can be re-injected. The replayer reads an envelope, resets the attempt budget so the task gets a fresh set of tries, and publishes it back to the work queue — but it must be idempotent, because replaying a task whose deletion actually did complete would either double-process or, worse, re-expose data. Guard replay against the same request_id + source_system dedup key the consumers use.

from collections.abc import Iterable


def replay(
    envelopes: Iterable[RetryEnvelope],
    already_completed: set[tuple[str, str]],
    requeue: Callable[[RetryEnvelope], None],
) -> list[str]:
    """Re-inject dead-lettered sub-tasks, skipping any that already completed.

    Resets the attempt counter so a fixed task gets a fresh budget, but never
    re-injects a task whose (request_id, source_system) is known-complete.
    """
    replayed: list[str] = []
    for env in envelopes:
        key = (env.request_id, env.source_system)
        if key in already_completed:
            continue               # never re-run a finished erasure or export
        requeue(env.model_copy(update={"attempt": 0, "last_error": None}))
        replayed.append(env.request_id)
    return replayed

Configuration reference #

Verification #

Prove the two invariants that keep an erasure from vanishing: a task at its cap is dead-lettered rather than dropped, and a completed task is never replayed.

def test_exhausted_task_is_dead_lettered_not_dropped() -> None:
    """At the attempt cap, routing must publish to the DLQ."""
    sink: list[RetryEnvelope] = []
    env = RetryEnvelope(
        request_id="r-1", source_system="crm", legal_basis="gdpr_art_17",
        action="deletion", attempt=4, max_attempts=5,
    )
    action = handle_failure(
        env, RuntimeError("upstream 500"),
        requeue=lambda e: (_ for _ in ()).throw(AssertionError("must not requeue")),
        to_dlq=sink.append,
    )
    assert action == "dead_lettered"
    assert sink[0].request_id == "r-1" and sink[0].legal_basis == "gdpr_art_17"


def test_replay_skips_completed_tasks() -> None:
    """A task whose (request_id, source_system) already completed is never re-injected."""
    env = RetryEnvelope(
        request_id="r-2", source_system="billing", legal_basis="ccpa_1798_105",
        action="deletion",
    )
    out = replay([env], already_completed={("r-2", "billing")}, requeue=lambda e: None)
    assert out == []

Expected: handle_failure returns dead_lettered and the envelope reaches the sink with its legal_basis and request_id intact; replay returns an empty list for a known-complete task.

Troubleshooting #

A sub-task is neither in the work queue nor the DLQ Root cause: the original message was acknowledged before the requeue or DLQ publish committed, so a crash in that window dropped it. Fix: acknowledge only after the durable publish returns (the ack_after_route invariant), and use publisher confirms so the broker guarantees the write.

The DLQ fills with the same message forever Root cause: a poison message is being retried through the normal budget, crashing the consumer each time and never reaching a terminal state. Fix: apply the Step 3 poison check keyed on delivery id so repeated consumer crashes short-circuit to the DLQ regardless of attempt count.

Replay double-deletes or re-exports data Root cause: the replayer re-injected a task whose sub-task had already completed, because the dedup key was not consulted. Fix: pass the known-complete (request_id, source_system) set to replay and skip any match, exactly as in Step 4.

Dead-lettered entries lack the legal basis for triage Root cause: the DLQ stores only the raw payload, not the envelope headers. Fix: publish the full RetryEnvelope — request id, legal basis, attempt count, last error — so a compliance operator can reconstruct why the task failed and under which obligation it must still be finished.

Retries keep firing past the deadline Root cause: max_attempts is bounded but the cumulative wait is not, so the task exhausts its budget after the SLA has already lapsed. Fix: this is a timing concern, not a routing one — bound the total elapsed retry time against the budget as described in Configuring exponential backoff within SLA budgets.

Frequently Asked Questions #

Related #

• Async Polling & Queue Management — the parent stage whose failure taxonomy this dead-letter routing implements.
• Configuring exponential backoff within SLA budgets — the retry-timing companion that decides how long between the attempts this design counts.
• Monitoring Kafka consumer lag for DSR SLAs — how to watch the backlog so DLQ growth surfaces before it threatens a deadline.
• Implementing Celery for async polling — the worker and queue topology this routing plugs into.
• Cross-System Data Discovery & Sync — the discovery stage these workers belong to.