Building a Video Transcoding Job Queue with Python RQ and Redis

Our multi-region fetch cron pulls trending video metadata for 8 regions on a staggered schedule. Until a few months ago, the same PHP process that wrote those rows into SQLite also shelled out to ffmpeg to build preview artifacts: one 6-second WebM loop and three WebP thumbnails per new video. That was fine at 5 videos per region. It fell apart at 40. LiteSpeed kills a request at 180 seconds, and a single 720p loop takes 4 to 11 seconds depending on source bitrate. The cron would die somewhere inside region three, leave half-written .webm files on disk, and the next run would redo all of it because nothing had been marked complete.

The fix was not a cleverer ffmpeg invocation. It was moving the transcode work out of the request lifecycle entirely. What follows is the queue that replaced it: Redis as the broker, Python RQ workers on a small VPS, PHP 8.4 as the producer, and finished artifacts shipped back through the same FTP automation the rest of the site deploys with. The preview loops on TrendVidStream all come out of this pipeline.

Why RQ and not Celery

Honest answer: Celery works too. I picked RQ for three reasons that hold at this size.

• One dependency. RQ needs Redis and nothing else. Celery's broker/result-backend split invites a second piece of infrastructure, and I did not want to run RabbitMQ for a workload that peaks around 400 jobs a day.
• Failed jobs are a real dead letter queue. RQ keeps a FailedJobRegistry with the full traceback and the original arguments. Requeueing is one command, not a custom replay script.
• The source is small enough to read. When a job vanished, I read the worker loop and found my own bug in an afternoon.

The tradeoff is real: RQ workers fork per job, no threads and no eventlet. That rules out Windows and makes it a bad fit for 5ms jobs. Against a 7-second ffmpeg run, the fork cost is noise.

The shape of the pipeline

Four moving parts:

1. PHP cron (producer). After the region loop writes new video rows, it pushes a compact JSON intent onto a Redis list. It does not know RQ exists.
2. Bridge worker. One Python process doing BLPOP on that list, routing by priority, and enqueueing real RQ jobs.
3. Transcode workers. Three RQ workers, each running ffmpeg in a subprocess with a hard timeout.
4. Publisher. A dependent job that uploads artifacts over FTP and only then marks the row visible in SQLite.

Why the bridge, instead of having PHP enqueue directly into RQ? RQ serializes jobs with pickle by default, and its internal job hash layout is not a public contract; it shifted between the 1.x and 2.x lines. You can write a PHP-side serializer that RQ will eat, and people have, but you are then pinned to an implementation detail forever. A 50-line bridge keeps the RQ-specific format in Python where it belongs and gives one obvious place to put dedup and rate control.

The PHP producer

The producer's only real job is idempotency. A video id must never be queued twice, because the expensive part is not the enqueue, it is the 7 seconds of VP9 encoding on the other end.

<?php
declare(strict_types=1);

final class TranscodeIntentQueue
{
    private const LIST_KEY = 'vw:transcode:intents';
    private const SEEN_KEY = 'vw:transcode:seen';
    private const SEEN_TTL = 604800; // 7 days

    public function __construct(private readonly Redis $redis) {}

    public static function connect(string $host = '127.0.0.1', int $port = 6379): self
    {
        $r = new Redis();
        $r->connect($host, $port, 2.0);
        $r->setOption(Redis::OPT_READ_TIMEOUT, 2.0);
        return new self($r);
    }

    /** @param list<array{id:string,region:string,duration:int}> $videos */
    public function pushBatch(array $videos, int $priority = 5): int
    {
        $pushed = 0;
        foreach ($videos as $v) {
            // sAdd returns 0 when the id was already a member -> already queued.
            if ($this->redis->sAdd(self::SEEN_KEY, $v['id']) === 0) {
                continue;
            }
            $this->redis->rPush(self::LIST_KEY, json_encode([
                'video_id'  => $v['id'],
                'region'    => $v['region'],
                'duration'  => $v['duration'],
                'priority'  => $priority,
                'queued_at' => time(),
            ], JSON_THROW_ON_ERROR));
            $pushed++;
        }
        $this->redis->expire(self::SEEN_KEY, self::SEEN_TTL);
        return $pushed;
    }
}

// Called at the end of each region pass in the fetch cron.
$queue = TranscodeIntentQueue::connect();
$n = $queue->pushBatch($repo->newVideosSince($lastRun), $region === 'US' ? 1 : 5);
fwrite(STDERR, sprintf('[transcode] queued %d intents for %s%s', $n, $region, PHP_EOL));

That trailing EXPIRE slides the whole set forward on every push, so in practice the set never expires while the cron is healthy. That is deliberate. I would rather carry a few thousand stale ids in Redis than re-encode a back catalogue because the key aged out mid-week. If the cron stops for seven days I have a bigger problem than duplicate transcodes.

One thing worth calling out: the dedup happens before the push, not inside the worker. Producer-side dedup is cheap and exact. Worker-side dedup means you have already paid for the network hop and the fork.

The bridge

BLPOP with a short timeout, decode, route, enqueue. Nothing clever.

# bridge.py -- turns PHP intents into RQ jobs.
import json
import logging
import os
import signal
import sys

from redis import Redis
from rq import Queue, Retry

LIST_KEY = 'vw:transcode:intents'
_queues = {}
log = logging.getLogger('bridge')
running = True


def _stop(signum, frame):
    global running
    running = False


def queue_for(priority, conn):
    name = 'transcode_high' if priority <= 2 else 'transcode_default'
    if name not in _queues:
        _queues[name] = Queue(name, connection=conn, default_timeout=900)
    return _queues[name]


def main():
    signal.signal(signal.SIGTERM, _stop)
    signal.signal(signal.SIGINT, _stop)
    conn = Redis.from_url(os.environ.get('REDIS_URL', 'redis://127.0.0.1:6379/0'))

    while running:
        item = conn.blpop(LIST_KEY, timeout=5)
        if item is None:
            continue
        try:
            intent = json.loads(item[1])
        except ValueError:
            log.error('undecodable intent dropped: %r', item[1][:200])
            continue

        q = queue_for(intent.get('priority', 5), conn)
        q.enqueue(
            'tasks.transcode_preview',
            video_id=intent['video_id'],
            region=intent['region'],
            source_duration=intent['duration'],
            job_id='preview:' + intent['video_id'],
            retry=Retry(max=3, interval=[30, 120, 600]),
            job_timeout=900,
            result_ttl=3600,
            failure_ttl=259200,
        )
        log.info('enqueued %s on %s', intent['video_id'], q.name)
    return 0


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
    sys.exit(main())

A correction to a belief I held for a while: passing job_id does not give you unique-job semantics. RQ will happily push the same id onto the queue list twice and run it twice; the second enqueue just overwrites the job hash. The id is for lookups and log correlation. The actual dedup is that Redis set in the PHP producer, plus an existence check inside the task itself.

BLPOP costs one blocking connection. If you run the bridge under systemd, set Restart=always and keep the timeout non-infinite, so a Redis failover surfaces as a reconnect rather than a process wedged forever on a dead socket.

The transcode task

The important property here is that partial output is never visible. Everything lands in a staging directory and gets moved into place with a single rename.

# tasks.py
import os
import shutil
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path

WORK_ROOT = Path(os.environ.get('TRANSCODE_WORK', '/var/lib/vw/work'))
OUT_ROOT = Path(os.environ.get('TRANSCODE_OUT', '/var/lib/vw/out'))
FFMPEG = os.environ.get('FFMPEG_BIN', '/usr/bin/ffmpeg')


class TranscodeError(RuntimeError):
    pass


@dataclass(frozen=True)
class PreviewSpec:
    loop_seconds: int = 6
    width: int = 480
    crf: int = 34
    thumb_count: int = 3


def _run(cmd, timeout):
    proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                          timeout=timeout, check=False, start_new_session=True)
    if proc.returncode != 0:
        tail = proc.stderr.decode('utf-8', 'replace')[-800:]
        raise TranscodeError('ffmpeg exit ' + str(proc.returncode) + ': ' + tail)


def transcode_preview(video_id, region, source_duration, spec=PreviewSpec()):
    source = WORK_ROOT / region / (video_id + '.mp4')
    if not source.exists():
        raise TranscodeError('source missing: ' + str(source))

    final_dir = OUT_ROOT / video_id
    if (final_dir / 'loop.webm').exists():
        return {'video_id': video_id, 'skipped': True}

    # Sample from 25% in -- intros are usually black frames or sponsor cards.
    start = max(0, int(source_duration * 0.25))
    staging = Path(tempfile.mkdtemp(prefix='tc-', dir=str(OUT_ROOT)))
    try:
        _run([FFMPEG, '-nostdin', '-loglevel', 'error', '-y',
              '-ss', str(start), '-t', str(spec.loop_seconds), '-i', str(source),
              '-an', '-c:v', 'libvpx-vp9', '-crf', str(spec.crf), '-b:v', '0',
              '-cpu-used', '4', '-row-mt', '1', '-threads', '2',
              '-vf', 'scale=' + str(spec.width) + ':-2,fps=24',
              str(staging / 'loop.webm')], timeout=600)

        step = max(1, source_duration // (spec.thumb_count + 1))
        for i in range(1, spec.thumb_count + 1):
            _run([FFMPEG, '-nostdin', '-loglevel', 'error', '-y',
                  '-ss', str(step * i), '-i', str(source), '-frames:v', '1',
                  '-vf', 'scale=' + str(spec.width) + ':-2', '-quality', '78',
                  str(staging / ('thumb-' + str(i) + '.webp'))], timeout=90)
    except BaseException:
        shutil.rmtree(staging, ignore_errors=True)
        raise

    try:
        os.replace(staging, final_dir)   # atomic; fails if final_dir is non-empty
    except OSError:
        shutil.rmtree(staging, ignore_errors=True)
        return {'video_id': video_id, 'skipped': True}

    return {'video_id': video_id, 'region': region,
            'bytes': sum(p.stat().st_size for p in final_dir.iterdir())}

Three details that were not obvious to me at the start:

• -nostdin is not optional. Without it, ffmpeg reads from the worker's stdin, and a forked RQ worker shares that descriptor. One malformed input file and ffmpeg consumes whatever else is queued on stdin, then the worker behaves strangely for reasons that make no sense in the traceback.
• -ss before -i seeks by keyframe and is roughly instant. After -i it decodes from the start. On a 40-minute source that is the difference between 200ms and 30 seconds.
• start_new_session=True puts ffmpeg in its own process group, so a subprocess.TimeoutExpired kill does not take the worker down with it, and a runaway encode does not survive as an orphan holding a file handle on the staging directory.

BaseException rather than Exception in that cleanup is intentional. RQ enforces job timeouts by raising a JobTimeoutException inside the worker via a signal, and depending on version and lineage that can bypass a plain except Exception. Catching BaseException and re-raising means a timed-out job still cleans up its staging directory.

Timeouts, retries, and what actually deserves a retry

The defaults will hurt you. RQ's default job timeout is 180 seconds, which is exactly the LiteSpeed limit I was trying to escape. A 4K source with a slow CRF pass can take several minutes, so the queue carries default_timeout=900 and each job sets job_timeout explicitly.

Retries need to be split by failure class, because most of them are pointless:

• Source file missing. Retrying helps: the downloader may still be writing. Backoff of 30s, 120s, 600s covers it.
• ffmpeg exit 1 on a corrupt container. Retrying is pure waste. Three attempts, three identical failures, 20 minutes of worker time burned.
• OOM kill. The worker dies with the job. RQ's death penalty never fires because nothing is alive to raise it, so the job sits in the StartedJobRegistry until its TTL expires and gets moved to failed. That path is slow and silent, and it is the main reason I cap worker memory and recycle workers.

The pragmatic version is a narrow retry: raise a distinct SourceNotReadyError for the retryable case and let everything else go straight to the failed registry, where I can look at it. A dead letter queue you actually read is worth more than a retry policy that hides the same bug three times.

Worker sizing and not melting the box

The instinct on an 8-core VPS is eight workers. That is wrong for ffmpeg, because ffmpeg is already parallel. Eight workers each spawning a default-threaded encode gives you 60-odd threads fighting over 8 cores, and per-job latency roughly triples while total throughput barely moves.

What works here: three workers, -threads 2 per encode, one core left for Redis and the bridge. Measured on our workload, p50 went from 19s per job (8 workers) to 6.8s (3 workers), with p95 at 19s. A 40-video US batch drains in about two and a half minutes.

# worker.py
import logging
import os
import resource

from redis import Redis
from rq import Queue, Worker

log = logging.getLogger('worker')


def cap_memory(soft_mb=1024):
    limit = soft_mb * 1024 * 1024
    resource.setrlimit(resource.RLIMIT_AS, (limit, limit))


def record_failure(job, exc_type, exc_value, traceback):
    job.connection.hincrby('vw:transcode:failcount',
                           job.kwargs.get('video_id', 'unknown'), 1)
    log.error('job %s failed: %s: %s', job.id, exc_type.__name__, exc_value)
    return True  # let RQ continue to its default handling


def main():
    cap_memory()
    conn = Redis.from_url(os.environ.get('REDIS_URL', 'redis://127.0.0.1:6379/0'))
    # Queue order == priority order: high is fully drained before default.
    queues = [Queue('transcode_high', connection=conn),
              Queue('transcode_default', connection=conn)]
    worker = Worker(queues, connection=conn,
                    name=os.environ.get('WORKER_NAME'),
                    exception_handlers=[record_failure])
    worker.work(with_scheduler=True, max_jobs=200)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    main()

max_jobs=200 plus Restart=always in the systemd unit is a deliberate recycle: the worker exits cleanly after 200 jobs and systemd starts a fresh one. It papers over slow leaks in long-lived processes, and it gives you a natural point where a new deploy picks up new code without a manual restart. with_scheduler=True is what makes the retry intervals actually fire; without it, retried jobs are re-enqueued immediately regardless of your backoff list.

Publishing over FTP without racing the site deploy

Our hosting is shared LiteSpeed with FTP-only access, so the artifacts have to be pushed rather than mounted. The rule that matters: a video row must not become searchable before its preview exists, or the grid renders a broken loop element. So the publish step is a dependent job, and the SQLite write, including the FTS5 index update, happens last.

# publish.py
import ftplib
import os
import sqlite3
from pathlib import Path


def publish_preview(video_id, region, **_):
    src = Path(os.environ['TRANSCODE_OUT']) / video_id
    files = sorted(p for p in src.iterdir() if p.is_file())
    if not files:
        raise RuntimeError('nothing to publish for ' + video_id)

    ftp = ftplib.FTP_TLS(os.environ['FTP_HOST'])
    ftp.login(os.environ['FTP_USER'], os.environ['FTP_PASS'])
    ftp.prot_p()
    remote_dir = '/public_html/media/previews/' + video_id
    try:
        ftp.mkd(remote_dir)
    except ftplib.error_perm:
        pass  # already there

    for path in files:
        tmp = remote_dir + '/.' + path.name + '.part'
        with path.open('rb') as fh:
            ftp.storbinary('STOR ' + tmp, fh, blocksize=65536)
        ftp.rename(tmp, remote_dir + '/' + path.name)  # atomic-ish on the server
    ftp.quit()

    db = sqlite3.connect(os.environ['SQLITE_PATH'], timeout=30)
    try:
        db.execute('PRAGMA busy_timeout = 30000')
        db.execute('UPDATE videos SET preview_ready = 1 WHERE video_id = ?', (video_id,))
        db.commit()
    finally:
        db.close()
    return {'video_id': video_id, 'files': len(files)}


# In the bridge, chain it so publish only runs on a successful transcode:
#   from rq.job import Dependency
#   t = q.enqueue('tasks.transcode_preview', video_id=vid, region=r, source_duration=d)
#   q.enqueue('publish.publish_preview', video_id=vid, region=r,
#             depends_on=Dependency(jobs=[t], allow_failure=False))

Upload to a dotfile then rename: FTP has no atomic upload, and without this the site can serve a truncated WebM to whoever requests it mid-transfer. The leading dot also keeps partials out of any directory listing the site itself walks.

The SQLite side needs busy_timeout. The publisher and the fetch cron write to the same file, and the default behaviour is an immediate database is locked rather than a wait. Since preview_ready is what the search query filters on, and the FTS5 content table is populated from the same row, an FTS5 rebuild triggered mid-publish would otherwise fail with no useful message.

What broke in production

• Staging directories accumulated. Early version cleaned up in except Exception. OOM kills and timeouts skipped it. Fixed with BaseException plus a nightly sweep of tc-* directories older than a day.
• A duplicate enqueue during a cron rerun ran two identical encodes. Both finished; the second lost the os.replace race and returned skipped. Harmless, but it is why the existence check at the top of the task exists.
• Redis maxmemory-policy was allkeys-lru. Under pressure, Redis evicted job hashes, and workers popped ids pointing at nothing. For a broker the correct setting is noeviction, so a full Redis fails loudly on write instead of silently deleting your queue.
• rq info --interval 2 is the whole monitoring stack. Queue depth and worker state in one line. Alerting is a cron that counts the FailedJobRegistry and emails when it grows.

Wrapping up

The architectural win was not RQ specifically. It was the boundary: PHP writes an intent, Python does the CPU work, and no user-facing process waits on ffmpeg. Everything after that is the same handful of rules any queue needs. Dedup at the producer. Make jobs idempotent, so a duplicate is boring rather than corrupting. Stage output and publish with a rename. Retry only what a retry can actually fix, and read your dead letter queue. Size workers to the parallelism of the work, not the core count.

If you are starting from a cron that is quietly timing out, move the slowest single step out first and leave the rest alone. That change took an afternoon and removed the failure mode entirely; the priority queues, the dependency chaining, and the FTP staging all came later, once I could see what was actually failing.