Replaying Dead-Letter Media Jobs Safely

This page sits beneath Retry Logic & Dead Letter Queues within the broader Pipeline Automation & Batch Processing layer, and it addresses the moment after the incident is over: a dead-letter queue holds three thousand failed media jobs, the underlying bug is fixed, and now you have to put those jobs back into the pipeline without republishing episodes that already went out, without re-triggering the same poison message a hundred times, and without a thundering herd that knocks the fixed service straight back over.

Problem Framing

A dead-letter queue (DLQ) is where a message lands after its delivery and retry budget is exhausted. That is the easy part; the broker does it for you. The hard part is the replay, because a DLQ replay is a re-injection of side effects. A naive for msg in dlq: publish(msg) is dangerous for three independent reasons, and a safe drain has to defend against all three at once.

First, duplicates. Many dead-lettered media jobs failed after doing partial work — an encode finished and the asset was uploaded, but the ACK never reached the broker, so the job was retried and eventually dead-lettered. Blindly replaying it publishes the same episode to the RSS feed twice. Second, poison messages: a subset of jobs fail deterministically because the payload itself is malformed, so replaying them just walks them straight back into the DLQ, burning worker time on every lap. Third, load: draining three thousand jobs at line rate slams the downstream service that just recovered.

You can see all three in the operator’s log the instant a naive replay starts:

INFO  replay     drained=3000  reinjected=3000  rate=unbounded
WARN  duplicate  asset=ep214  action=rss_publish  note="episode already live; second publish created dup GUID"
WARN  poison     asset=ep209  dlq_returns=4  reason="unparseable chapter JSON; will never succeed"
ERROR downstream  service=packager  status=503  note="recovered service re-saturated by replay burst"

The fix is a replay controller that decides per message whether to quarantine or replay, makes every re-publish idempotent, drains under a rate limit, and records an audit row for each decision so the operation is reconstructable after the fact.

Solution Architecture

The controller treats the DLQ as read-input, never as a work queue to blindly forward. Each message passes a poison gate (has it already bounced too many times, or does its payload fail a schema check?). Poison messages divert to a quarantine queue for human triage; they are not replayed. Survivors go through an idempotency gate backed by a durable key store: if the job’s idempotency key is already marked complete, the message is dropped as a confirmed duplicate rather than republished. Only genuinely-unfinished jobs are re-published, and even those go out through a token-bucket rate limiter. Every branch writes an audit record.

Because the controller only ever reads the DLQ and re-publishes to the live work queue, the original dead-lettered messages remain in place until each is explicitly ACKed after its decision is durably recorded — so a crash mid-drain resumes exactly where it stopped, with no lost and no double-processed jobs.

Implementation

1. Detect poison messages before anything else

Every dead-lettered message carries a delivery history in its headers. The x-death header that RabbitMQ dead-letter exchanges stamp on a message records how many times it has been dead-lettered; a message that has bounced more than a small threshold is poison by definition and must not re-enter the pipeline. Combine that count with a schema check on the payload so malformed jobs are caught even on their first appearance.

# celery==5.4.0  kombu==5.4.0  redis==5.0.7  pika==1.3.2  (broker: RabbitMQ 3.13)
import hashlib
import json
import logging
from typing import Optional

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

MAX_DLQ_RETURNS = 3   # more bounces than this = deterministic failure = poison

def dlq_return_count(headers: dict) -> int:
    """Sum the RabbitMQ x-death counts across all dead-letter reasons."""
    return sum(entry.get("count", 0) for entry in headers.get("x-death", []))

def is_poison(payload: dict, headers: dict) -> Optional[str]:
    """Return a reason string if the message must be quarantined, else None."""
    if dlq_return_count(headers) > MAX_DLQ_RETURNS:
        return f"exceeded {MAX_DLQ_RETURNS} dead-letter returns"
    # Payload-level schema guard: a job with no asset_id or bad chapter JSON
    # will fail deterministically forever — quarantine it, do not replay.
    if not payload.get("asset_id"):
        return "missing asset_id"
    chapters = payload.get("chapters")
    if chapters is not None:
        try:
            json.loads(chapters) if isinstance(chapters, str) else chapters
        except (ValueError, TypeError):
            return "unparseable chapter JSON"
    return None

2. Guard re-publish with an idempotency key

The idempotency key must be derived from the job’s intended effect, not from the message id — two retries of the same publish carry different message ids but the same effect. For a media job, hash the stable inputs: asset id, target stage, and content version. Before re-publishing, check a durable store (here Redis SETNX, per the Redis command reference); if the key is already marked complete, the job’s side effect already happened and the message is a confirmed duplicate to drop.

import redis

kv = redis.Redis(host="redis", port=6379, db=3, decode_responses=True)

def idempotency_key(payload: dict) -> str:
    """Stable hash of the job's intended effect — identical across retries."""
    basis = f"{payload['asset_id']}:{payload['stage']}:{payload.get('version', 1)}"
    return "idem:" + hashlib.sha256(basis.encode()).hexdigest()[:32]

def already_applied(key: str) -> bool:
    """True if this effect was already committed by a prior (successful) run."""
    return kv.get(key) == "done"

def mark_applied(key: str, ttl_seconds: int = 604800) -> None:
    """Called by the WORKER after the side effect commits, not by the replayer."""
    kv.set(key, "done", ex=ttl_seconds)   # 7-day retention window

The critical discipline: the worker calls mark_applied only after the publish/upload actually commits. The replayer only ever reads the key. That split is what makes a replay safe — the record of “this episode is live” is written by the code that made it live, so the replayer can trust it.

3. Drain under a rate limit with a full audit trail

Now tie it together. Pull from the DLQ, classify, route, and — for genuine replays — re-publish through a token bucket so the recovered downstream is never re-saturated. Every decision writes an audit row before the DLQ message is ACKed, so the operation is fully reconstructable. Re-publishing itself uses ordinary Celery task dispatch, whose retry and acknowledgement semantics are covered in the Celery task retry documentation.

import time

RATE_TOKENS_PER_SEC = 50
_bucket = {"tokens": RATE_TOKENS_PER_SEC, "ts": time.monotonic()}

def take_token() -> None:
    """Simple token bucket: blocks until a re-publish slot is available."""
    while True:
        now = time.monotonic()
        _bucket["tokens"] = min(
            RATE_TOKENS_PER_SEC,
            _bucket["tokens"] + (now - _bucket["ts"]) * RATE_TOKENS_PER_SEC,
        )
        _bucket["ts"] = now
        if _bucket["tokens"] >= 1:
            _bucket["tokens"] -= 1
            return
        time.sleep(0.02)

def audit(msg_id: str, decision: str, key: str, reason: str = "") -> None:
    kv.rpush("dlq:audit", json.dumps({
        "msg_id": msg_id, "decision": decision, "idempotency_key": key,
        "reason": reason, "ts": time.time(),
    }))

def replay_one(msg_id: str, payload: dict, headers: dict, quarantine, work_queue) -> str:
    key = idempotency_key(payload)
    poison_reason = is_poison(payload, headers)
    if poison_reason:
        quarantine.publish(payload)
        audit(msg_id, "quarantine", key, poison_reason)
        return "quarantine"
    if already_applied(key):
        audit(msg_id, "drop_duplicate", key)
        return "drop_duplicate"
    take_token()                                   # rate-limited re-injection
    work_queue.publish(payload, headers={"idempotency_key": key})
    audit(msg_id, "replayed", key)
    return "replayed"

For a partial-batch replay — say, only the 400 jobs for one show that failed a specific stage — filter on the payload before calling replay_one, and the same idempotency and audit guarantees apply unchanged. The decision of whether a whole class of messages should be replayed at all, versus permanently quarantined, is the quarantine-versus-replay call the parent guide on Retry Logic & Dead Letter Queues frames in depth.

Verification

Prove the two guarantees operators actually care about: no duplicates, and a complete audit trail.

# 1. Dry-run classification: count how many messages WOULD quarantine vs replay,
#    without publishing anything. A high poison ratio means the bug isn't fixed.
python -c "
from replay import classify_dlq
counts = classify_dlq('media.dlq', dry_run=True)
print(counts)   # e.g. {'replayed': 2761, 'drop_duplicate': 205, 'quarantine': 34}
"

# 2. Duplicate check: replay, then assert no idempotency key was published twice.
python -c "
from replay import drain
report = drain('media.dlq', limit=500)
assert report['double_published'] == 0, report
print('no duplicates:', report)
"

# 3. Audit completeness: every drained message must have exactly one audit row.
redis-cli LLEN dlq:audit
rabbitmqctl list_queues name messages | grep -E 'media.dlq|media.quarantine'
# expected: audit count == messages drained; media.dlq drops to 0; quarantine holds only poison

A healthy drain shows the DLQ depth falling monotonically to zero, the quarantine queue holding only the genuinely-broken jobs, the duplicate-drop count matching the number of jobs that had already completed before dead-lettering, and one audit row per message. Export drain rate, quarantine ratio, and double_published (which must stay at 0) to Prometheus, and alert if the quarantine ratio spikes — that means you started the replay before the root cause was actually fixed.

Failure Modes & Edge Cases

Sizing the rate limiter and the retry budget that feeds the DLQ in the first place is a capacity question that overlaps with worker-pool design; when the replayed jobs are GPU-bound, route them back through the priority-aware topology in Priority Queues for Mixed GPU/CPU Media Jobs so a replay burst does not starve live traffic.

FAQ

Related

• Up to the parent guide: Retry Logic & Dead Letter Queues
• Route replayed GPU jobs without starving live traffic: Priority Queues for Mixed GPU/CPU Media Jobs
• Where dead-lettered jobs originate: Celery Task Routing for Video Jobs
• Orchestrated replays as a scheduled backfill: Orchestrating Pipelines with Airflow
• Pipeline overview: Pipeline Automation & Batch Processing