Retry Logic & Dead Letter Queues

As part of the Pipeline Automation & Batch Processing layer, retry logic and a dead letter queue form the fault-tolerance backbone that decides which failures the pipeline absorbs silently and which it quarantines for a human. This component protects a throughput-and-integrity SLA: a network blip during remote ingest, a transient codec library crash, or an ephemeral storage stall must never corrupt a deliverable, orphan an intermediate file, or stall the batch — yet a genuinely unrecoverable job must stop retrying before it burns GPU quota and be sealed for analysis. Get this layer wrong and transient errors either cascade into half-written outputs or hammer a recovering upstream into a second outage; get it right and the rest of the pipeline can assume that everything past the retry boundary either succeeded or is safely in the DLQ.

Prerequisites & Environment

This component assumes the broker-backed task fabric described in Celery task routing for video jobs and the scheduler in orchestrating pipelines with Airflow. Retries and DLQ routing are cross-cutting machinery layered on top of both, so the environment below is deliberately minimal.

Pin every version in the worker image rather than the host; reproducible retries depend on byte-identical binaries, which is exactly what dockerizing media processing containers guarantees. The relevant environment variables:

Architecture & Queue Topology

Retries and the DLQ sit beside the work queues rather than inside them. A worker claims a task with acks_late=True, so the broker only drops the message once the task either succeeds or is explicitly acknowledged after being sealed into the DLQ — a crash mid-transcode therefore redelivers rather than vanishes. Transient failures are re-scheduled by the worker onto the same queue after a computed delay; permanent failures and budget exhaustion are published to a dedicated media.dlq Redis stream. A separate, low-concurrency replay consumer drains the DLQ on demand, never automatically, so a poison payload can never form a tight retry loop.

The key invariant: a task can only leave the retry loop in one of two ways — success, or a sealed DLQ record. There is no third “logged and forgotten” path, because silent drops are precisely what make media batches non-reproducible.

Step-by-Step Implementation

1. Define a per-job-class retry policy

Different media work has different economics. A ffprobe metadata pass is cheap and flaky, so it tolerates many fast attempts; a 4K transcode is expensive, so it caps low and backs off hard. Encode that as data, not as scattered decorators.

# celery==5.3.6, python>=3.10
from dataclasses import dataclass

@dataclass(frozen=True)
class RetryPolicy:
    max_retries: int
    backoff_base: int      # seconds
    backoff_max: int       # ceiling on a single delay
    jitter: bool

POLICIES: dict[str, RetryPolicy] = {
    "cpu.metadata":  RetryPolicy(max_retries=15, backoff_base=2,  backoff_max=60,  jitter=True),
    "gpu.transcode": RetryPolicy(max_retries=3,  backoff_base=10, backoff_max=600, jitter=True),
    "io.upload":     RetryPolicy(max_retries=8,  backoff_base=2,  backoff_max=120, jitter=True),
}

Verify the policy table loads and is exhaustive for your routed queues:

python -c "from policies import POLICIES; assert {'cpu.metadata','gpu.transcode','io.upload'} <= POLICIES.keys(); print('ok')"

2. Make the task idempotent with a content key

A retry must be safe to run twice. Derive a deterministic output key from the input content hash so a re-execution overwrites the same object instead of appending a duplicate, and short-circuit if the artifact already exists. The asset_sha256 contract comes straight from the upstream gate described in media validation and error routing.

# python>=3.10
from pathlib import Path

def transcode_output_key(asset_sha256: str, profile: str) -> str:
    # Deterministic: same input + profile always maps to the same object.
    return f"derived/{profile}/{asset_sha256}.mp4"

def already_done(store, key: str) -> bool:
    return store.exists(key)  # idempotency guard before any expensive work

Verify two runs of the same payload produce one object, not two:

python -m pytest tests/test_idempotency.py::test_double_run_single_artifact -q

3. Apply exponential backoff with jitter

Let Celery own the retry scheduling and compute the delay from the policy. Randomized jitter spreads retries across a window so a fleet of workers does not stampede a recovering media server in lockstep.

# celery==5.3.6
import random
from celery import shared_task
from policies import POLICIES

@shared_task(
    bind=True,
    acks_late=True,                 # redeliver on worker crash
    autoretry_for=(TransientMediaError,),
    retry_backoff=True,             # base*2**(n-1)
    retry_jitter=True,              # full jitter
    max_retries=POLICIES["gpu.transcode"].max_retries,
    retry_backoff_max=POLICIES["gpu.transcode"].backoff_max,
)
def transcode(self, payload: dict) -> dict:
    if already_done(store, transcode_output_key(payload["asset_sha256"], payload["profile"])):
        return {"status": "success", "skipped": True}
    return run_ffmpeg(payload)      # raises TransientMediaError or PermanentMediaError

Verify the delay sequence grows and stays under the ceiling:

python -c "import policies as p; b=p.POLICIES['gpu.transcode']; \
print([min(b.backoff_base*2**(n), b.backoff_max) for n in range(b.max_retries+1)])"
# -> [10, 20, 40, 80]  (80 < 600 ceiling)

4. Route exhausted tasks to the DLQ with a sealed payload

When max_retries is hit, or a PermanentMediaError (bad container, missing codec, auth failure) is raised, seal the full execution context into the DLQ. The payload is validated against a Pydantic model before it is written, so a malformed DLQ record can never itself become a debugging dead end.

# pydantic==2.6.1, redis==5.0.1
from datetime import datetime, timezone
from pydantic import BaseModel, Field

class DLQRecord(BaseModel):
    job_id: str
    queue: str
    attempt_count: int = Field(ge=1)
    error_code: str                       # e.g. ffmpeg exit code or "AUTH_401"
    error_class: str                      # "permanent" | "exhausted"
    original_payload_hash: str            # ties back to asset_sha256
    payload: dict
    stack_trace: str
    enqueued_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

def to_dlq(dlq, record: DLQRecord) -> None:
    dlq.xadd("media.dlq", {"body": record.model_dump_json()})  # schema-validated on construct

Hook it into the task’s failure handler:

@transcode.on_failure
def seal(self, exc, task_id, args, kwargs, einfo):
    record = DLQRecord(
        job_id=task_id, queue="gpu.transcode",
        attempt_count=self.request.retries + 1,
        error_code=getattr(exc, "code", "UNKNOWN"),
        error_class="exhausted" if self.request.retries >= self.max_retries else "permanent",
        original_payload_hash=args[0]["asset_sha256"],
        payload=args[0], stack_trace=str(einfo),
    )
    to_dlq(dlq, record)

Verify a forced permanent failure lands exactly one well-formed record:

redis-cli -n 9 XLEN media.dlq        # expect it to increment by 1
redis-cli -n 9 XRANGE media.dlq - + COUNT 1   # inspect the sealed payload

5. Replay DLQ entries under a dry-run guard

Recovery is deliberate, never automatic. A replay validates the hash, confirms the codec is available again, runs a dry pass, and only then re-enqueues — so restoring an upstream dependency does not unleash a flood of unverified jobs.

# python>=3.10
def replay(dlq, entry_id: str, *, dry_run: bool = True) -> str:
    raw = dlq.xrange("media.dlq", entry_id, entry_id)[0][1]["body"]
    record = DLQRecord.model_validate_json(raw)
    if record.payload["asset_sha256"] != record.original_payload_hash:
        return "rejected: payload tampered"
    if not codec_available(record.payload["codec"]):
        return "deferred: codec still missing"
    if dry_run:
        return "dry-run ok"
    transcode.apply_async(args=[record.payload], queue=record.queue)
    dlq.xdel("media.dlq", entry_id)
    return "re-enqueued"

Verify a dry-run replay touches no work queue:

python -m pytest tests/test_replay.py::test_dry_run_does_not_enqueue -q

Data Contracts

Every DLQ record is the source of truth for a failed job; downstream triage tooling depends on these fields existing and being typed exactly so it can reconstruct state without any volatile in-memory cache.

Resilience Patterns

The retry policy is bounded exponential backoff with full jitter, capped per job class so a recovering upstream is never stampeded and a 4K transcode never silently consumes more than its three attempts. Idempotency is guaranteed by deriving the output key from the input content hash and short-circuiting when the artifact already exists, which makes every retry — and every replay — safe to run more than once. DLQ routing is the terminal state for anything the loop cannot fix: permanent errors are sealed immediately, exhausted budgets are sealed on the final failure, and both carry the full payload plus stack trace. The acks_late setting closes the last gap, ensuring a worker that dies mid-job causes a redelivery rather than a silent loss. Together these turn an unreliable distributed system into one with exactly two terminal outcomes per task: a verified artifact, or a triage-ready DLQ record.

Observability & Debugging

Instrument the boundary, not the happy path. The signals that matter are retry pressure (are we masking a degrading dependency?) and DLQ depth (is something systemically broken?).

# prometheus-client==0.20.0
from prometheus_client import Counter, Gauge

RETRIES = Counter("media_task_retries_total", "Retries", ["queue", "error_code"])
DLQ_ENQUEUED = Counter("dlq_messages_enqueued_total", "DLQ inserts", ["queue", "error_class"])
DLQ_DEPTH = Gauge("dlq_depth", "Current DLQ length", ["queue"])
FAIL_BY_CODEC = Counter("transcode_failure_by_codec_total", "Failures", ["codec", "error_code"])

Differentiate expected transient spikes (a brief CDN rate limit) from systemic degradation (storage latency climbing for ten minutes) at the alert layer: page on sustained dlq_depth growth, not on a single retry burst.

Performance Tuning

Set worker_prefetch_multiplier=1 for long media jobs so a worker reserves only the task it is actively processing — otherwise a crash forfeits a whole batch of prefetched work to redelivery. Keep GPU queues at -c 1 (one transcode per VRAM budget) and let the backoff ceiling, not concurrency, absorb retry pressure. Size RETRY_BACKOFF_MAX against the realistic recovery time of the slowest upstream: 600 seconds suits a media origin that occasionally restarts, but pushing it to thousands of seconds just hides outages that should page. Give the DLQ store a TTL (14 days is a sane default) so triaged records expire rather than growing unbounded, and run the replay consumer at concurrency 1 so recovery is paced and observable rather than a thundering re-enqueue.

Frequently Asked Questions

Related

• Up to the parent overview: Pipeline Automation & Batch Processing
• The queues these retries re-schedule onto: Celery Task Routing for Video Jobs
• The scheduler whose on_failure_callback feeds the DLQ: Orchestrating Pipelines with Airflow
• Why pinned images make retries reproducible: Dockerizing Media Processing Containers
• The upstream gate that classifies failures: Media Validation & Error Routing