Celery Task Routing for Video Jobs

As part of the Pipeline Automation & Batch Processing layer, Celery task routing is the control plane that decides where each video job runs, while the orchestrator decides when. This component protects a throughput-and-isolation SLA: a 4K HDR transcode must land only on a worker with NVENC and sufficient VRAM, a thumbnail extraction must never block behind it, and no malformed payload may poison a shared queue. Transcoding 4K masters, extracting podcast stems, and generating adaptive-bitrate ladders each impose distinct compute profiles, memory ceilings, and I/O patterns; when they share one queue, resource contention is inevitable. Routing turns a flat worker pool into a specialized, fault-tolerant fabric where hardware affinity, payload contracts, and predictable throughput are enforced as infrastructure code.

Prerequisites & Environment

Reproducible routing starts with a pinned runtime and an externalized configuration manifest. The layer assumes Python 3.10+ (for match statements and precise type hints), a broker that supports priority and per-queue routing, and a result backend that survives worker restarts. Routing rules are never hard-coded in source — they load from a version-controlled YAML manifest so the same topology validates across development, staging, and production. Configuration drift is treated as a critical failure vector, so the manifest, not the shell, owns queue names and concurrency caps.

# Python 3.10+
# celery==5.3.6           # task routing, retries, autoscale
# kombu==5.3.5            # Queue/Exchange/binding primitives
# redis==5.0.3            # broker + result backend (or RabbitMQ via amqp)
# pydantic==2.6.4         # payload + result-contract validation
# prometheus-client==0.20.0  # per-queue metrics exporter
# structlog==24.1.0       # JSON structured logging

GPU transcode workers need a host with an NVENC-capable card and the driver stack baked into the image; CPU workers for thumbnailing and waveform generation need only fast local scratch. Budget per-worker memory against the peak frame buffer of the largest job class a queue accepts — a 4K HDR encode can hold 4–6 GB of VRAM plus host RSS, while a 1080p thumbnail extraction stays under 300 MB. The following environment variables govern everything dynamic; static routing rules stay in the manifest:

This component consumes already-validated assets. Container integrity, stream topology, and codec identification are settled upstream in the Media Ingestion & Format Architecture pipeline, and the worker images themselves are built per the Dockerizing Media Processing Containers standard, so the routing layer can assume each job points at a real, probeable asset and spend its cycles deciding placement rather than re-inspecting media.

Architecture & Queue Topology

The routing layer is a deterministic dispatcher sitting between an orchestrator and a set of hardware-segregated worker pools. Producers validate a payload, resolve it to exactly one queue through the task_routes map, and publish; consumers are launched with -Q bound to specific queues so a job can only ever execute on matching hardware. A production topology separates video.transcode.gpu, video.thumbnail.cpu, audio.extract, and metadata.parse into distinct logical channels, with a dedicated dead-letter queue catching anything that exhausts its retries. This isolation prevents a runaway FFmpeg process from starving lightweight database updates or webhook dispatches.

The orchestrator emits Celery tasks only when upstream validation or storage provisioning completes; integrating routing with the Orchestrating Pipelines with Airflow DAG keeps queue topology, retry policy, and concurrency caps identical across environments. The broker functions purely as a traffic controller — it guarantees an H.264 encode never lands on a worker lacking NVENC, but it owns no business logic. Idempotency is anchored on the input SHA-256 so a redelivered message can never double-publish a rendition.

Step-by-Step Implementation

Each step ends with a verification command you can run before wiring the next one.

1. Declare queues and the task_routes map. Define every queue explicitly through Kombu so the broker provisions durable bindings, and route static task names to their hardware tier. Loading from the manifest keeps this declarative:

# Python 3.10+
from celery import Celery
from kombu import Queue

app = Celery("media", broker="redis://broker:6379/0",
             backend="redis://broker:6379/1")

app.conf.task_queues = (
    Queue("video.transcode.gpu"),
    Queue("video.thumbnail.cpu"),
    Queue("audio.extract"),
    Queue("metadata.parse"),
    Queue("video.dlq"),               # terminal sink for exhausted retries
)

app.conf.task_routes = {
    "tasks.transcode_hls":   {"queue": "video.transcode.gpu"},
    "tasks.make_thumbnails": {"queue": "video.thumbnail.cpu"},
    "tasks.extract_audio":   {"queue": "audio.extract"},
    "tasks.parse_metadata":  {"queue": "metadata.parse"},
}
app.conf.task_default_queue = "metadata.parse"     # never the GPU pool
app.conf.worker_prefetch_multiplier = 1            # long jobs: no hoarding

Verify the broker registered all five queues:

celery -A tasks inspect active_queues | grep -E 'video\.|audio\.|metadata\.'

2. Validate the job payload before dispatch. Reject poison messages at the producer boundary so workers only ever receive structurally sound instructions. A Pydantic model validates codec, resolution, and storage URI before the task is published:

# Python 3.10+
from pydantic import BaseModel, field_validator

ACCEPTED_CODECS = {"h264", "hevc", "prores", "vp9"}

class VideoJob(BaseModel):
    asset_sha256: str          # idempotency key
    source_uri: str            # s3://… or gs://…
    codec: str
    width: int
    height: int
    hdr: bool = False

    @field_validator("codec")
    @classmethod
    def known_codec(cls, v: str) -> str:
        if v not in ACCEPTED_CODECS:
            raise ValueError(f"unsupported codec: {v}")
        return v

Verify a malformed payload is rejected before it reaches the broker:

python -c "from jobs import VideoJob; VideoJob(asset_sha256='x', source_uri='s3://b/k', codec='mjpeg', width=3840, height=2160)" 2>&1 | grep -q "unsupported codec" && echo REJECTED

3. Route dynamically by codec and resolution. Static names cover most tasks, but a single transcode task whose target hardware depends on resolution needs a Router class. Celery consults routers before the static map, so this is where hardware affinity is computed:

# Python 3.10+
class VideoRouter:
    def route_for_task(self, name, args=None, kwargs=None, **opts):
        if name != "tasks.transcode_hls":
            return None                       # fall through to task_routes
        job = kwargs or {}
        # 4K / HDR demands NVENC + VRAM; SD can run on CPU libx264
        if job.get("hdr") or job.get("width", 0) >= 3840:
            return {"queue": "video.transcode.gpu", "priority": 9}
        if job.get("width", 0) >= 1920:
            return {"queue": "video.transcode.gpu", "priority": 5}
        return {"queue": "video.thumbnail.cpu", "priority": 1}

app.conf.task_routes = (VideoRouter(), app.conf.task_routes)

Verify a 4K job resolves to the GPU queue without publishing it:

python -c "from router import VideoRouter; print(VideoRouter().route_for_task('tasks.transcode_hls', kwargs={'width':3840,'hdr':True}))"

4. Bind workers to queues with hardware affinity. A GPU host runs a single-concurrency worker consuming only the transcode queue; CPU hosts run higher concurrency across the light queues. The -Q flag is the enforcement mechanism — a worker physically cannot pull from a queue it is not bound to:

# GPU host — one NVENC job at a time, never oversubscribe VRAM
celery -A tasks worker -Q video.transcode.gpu -c 1 \
       --pool=solo -n gpu@%h --max-tasks-per-child 50

# CPU host — light, parallel jobs
celery -A tasks worker -Q video.thumbnail.cpu,audio.extract,metadata.parse \
       -c 4 --pool=prefork -n cpu@%h

Verify each worker is bound to exactly the queues you intend:

celery -A tasks inspect active_queues | grep -A1 'gpu@'

5. Verify routing with a mock-publish integration test. Routing correctness must be asserted in CI, not discovered under production load. Publish a representative payload with task_always_eager disabled and assert the resolved queue:

# Python 3.10+  (pytest)
def test_4k_hdr_routes_to_gpu():
    opts = app.amqp.router.route(
        {}, "tasks.transcode_hls",
        kwargs={"width": 3840, "height": 2160, "hdr": True},
    )
    assert opts["queue"] == "video.transcode.gpu"

def test_sd_never_hits_gpu():
    opts = app.amqp.router.route({}, "tasks.transcode_hls",
                                 kwargs={"width": 640, "height": 360})
    assert opts["queue"] != "video.transcode.gpu"

Verify the routing suite is green before promoting the manifest:

pytest -q tests/test_routing.py && echo ROUTING_OK

Data Contracts

The handoff between producer and worker is governed by an explicit schema. Validating this contract at the ingress boundary is what makes retries idempotent and quarantine precise — a worker that trusts the contract never re-probes the asset.

The target_queue echoed back in the result is what lets you detect routing drift across deployments: if two environments resolve the same payload to different queues, their routers have diverged. Assets whose codec or resolution falls outside the accepted set are not transcoded here — they are rejected at validation and returned upstream rather than dispatched to a worker that would only fail.

Resilience Patterns

A video task is a side-effecting subprocess, so every invocation is treated as potentially failing and never partially. Idempotency is anchored on asset_sha256: renditions are written to a content-addressed path, and a redelivered message that finds a completed output re-acknowledges instead of re-encoding, making at-least-once broker delivery safe.

Retries use bounded exponential backoff with jitter so a storage-API blip never triggers a thundering herd. The decision to retry hinges on classifying the failure first — retrying a permanently broken asset just burns the same cycles three times:

# Python 3.10+
TRANSIENT = (ConnectionError, TimeoutError)

@app.task(bind=True, max_retries=3, acks_late=True,
          autoretry_for=TRANSIENT, retry_backoff=True,
          retry_backoff_max=600, retry_jitter=True)
def transcode_hls(self, **payload):
    try:
        return run_transcode(payload)
    except (FileNotFoundError, UnsupportedCodecError) as exc:
        # deterministic failure: do not retry, quarantine immediately
        send_to_dlq(payload, error=str(exc))
        raise self.replace(dlq_marker(payload))

Transient errors — ConnectionError, HTTP 5xx, temporary codec-lock contention — warrant retries; permanent errors — FileNotFoundError, invalid container format, unsupported pixel format — route straight to video.dlq with the payload and structured error metadata preserved. DLQ consumers trigger automated triage: re-download a source asset, notify editorial, or fall back to a software encoder. The shared retry-and-DLQ machinery — visibility timeouts, poison-message detection, replay tooling — is specified once in the Retry Logic & Dead Letter Queues patterns and reused here rather than reinvented per worker. The Celery task retry documentation covers the autoretry_for and max_retries semantics in full.

Observability & Debugging

Routing complexity demands proportional observability. Export per-queue metrics with a Prometheus client running alongside the workers, and standardize structured log fields so a stalled job can be traced to its queue and host in one query. Run with acks_late=True and worker_prefetch_multiplier=1 so a crash mid-transcode redelivers the job instead of losing it.

Instrument these metrics:

• celery_queue_depth{queue} — gauge of un-leased tasks; the backpressure signal
• celery_task_latency_seconds{task,queue} — histogram of execution wall-clock
• celery_retries_total{task} — counter; a spike on audio.extract flags upstream rate limiting
• celery_dlq_depth — gauge of quarantined payloads; should trend to zero
• gpu_worker_saturation — in-flight transcodes ÷ GPU_QUEUE_CONCURRENCY

Standardize these log fields on every task: task_id, asset_sha256, queue_name, worker_hostname, attempt, and execution_time. Correlate them with distributed-tracing spans to find latency hotspots. The error messages you will actually see map to concrete causes:

When debugging stalled jobs, inspect the broker’s acknowledgment state and confirm bindings with celery -A tasks inspect active_queues; a worker silently subscribed to the GPU queue is the most common cause of an H.264 job landing on a host without NVENC.

Performance Tuning

Throughput is governed by three knobs: per-queue concurrency, prefetch, and pool type. Set GPU concurrency from VRAM, not core count — a 1080p transcode may comfortably run at -c 4, but 4K HDR often requires -c 1 to avoid OOM kills. Keep worker_prefetch_multiplier=1 for long jobs so a worker does not reserve a backlog it cannot start, which would otherwise starve idle peers. Choose --pool=prefork for CPU-bound thumbnailing and --pool=solo for GPU workers whose codec library is not thread-safe, and bound --max-tasks-per-child so a leaking FFmpeg subprocess is recycled before it degrades the host.

For Kubernetes, bind GPU workers to labeled nodes with node selectors and taints, and use NVIDIA device passthrough so a pod can only schedule where its hardware exists. Provision identical broker virtual hosts and routing exchanges across environments with infrastructure-as-code, then validate the topology in CI before deployment. Always benchmark concurrency against actual peak RSS and VRAM per task rather than nominal limits — measured ceilings are the only ones that survive a 4K HDR master.

Frequently Asked Questions

Related

• Up to the parent overview: Pipeline Automation & Batch Processing
• The worker images these queues consume: Dockerizing Media Processing Containers
• The scheduler that emits these tasks: Orchestrating Pipelines with Airflow
• Shared retry and quarantine machinery: Retry Logic & Dead Letter Queues
• The CPU-bound audio sibling stage: FFmpeg batch processing for podcasts