diff --git a/.env.example b/.env.example index 00ed7e3..1f4afa7 100644 --- a/.env.example +++ b/.env.example @@ -30,6 +30,10 @@ UPLOAD_MAX_BYTES=5368709120 # Transcoder (Phase 10) — limit CPU/RAM per job, sequential renditions FFMPEG_THREADS=2 FFMPEG_PRESET=veryfast +# Phase 12: adaptive ladder + HW + pipe streaming +ENCODE_MODE=cbr +FFMPEG_HWACCEL=auto +TRANSCODER_PIPE_INPUT=true # Services ports (infra) POSTGRES_PORT=5432 diff --git a/Makefile b/Makefile index ca9af12..c6287ed 100644 --- a/Makefile +++ b/Makefile @@ -3,7 +3,7 @@ # .env — единственный в корне, compose явно указывает на него (--env-file), deploy/.env не нужен COMPOSE=docker compose --env-file .env -f deploy/docker-compose.yml -.PHONY: up down logs ps build lint fmt test e2e swagger swagger-install sync-py migrate-up migrate-down migrate-create migrate-alembic-up +.PHONY: up down logs ps build lint fmt test e2e e2e-file swagger swagger-install sync-py migrate-up migrate-down migrate-create migrate-alembic-up up: $(COMPOSE) up --build -d @@ -106,7 +106,11 @@ fmt-front: cd frontend && npm run format || npx prettier --write . e2e: - bash scripts/e2e.sh + POLL_TIMEOUT="$(POLL_TIMEOUT)" SAMPLE="$(SAMPLE)" bash scripts/e2e.sh + +e2e-file: + @test -n "$(FILE)" || (echo "usage: make e2e-file FILE=path/to/video.mp4" && exit 1) + POLL_TIMEOUT="$(POLL_TIMEOUT)" SAMPLE="$(FILE)" bash scripts/e2e.sh # Migrations — single source of truth: deploy/migrations (golang-migrate) # Prod: `migrate` service in docker-compose.yml runs `up` automatically. diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index e89e28f..6e641f1 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -47,6 +47,9 @@ services: LOG_LEVEL: warn FFMPEG_THREADS: ${FFMPEG_THREADS:-2} FFMPEG_PRESET: ${FFMPEG_PRESET:-veryfast} + ENCODE_MODE: ${ENCODE_MODE:-cbr} + FFMPEG_HWACCEL: ${FFMPEG_HWACCEL:-auto} + TRANSCODER_PIPE_INPUT: ${TRANSCODER_PIPE_INPUT:-true} deploy: resources: limits: diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 76202e5..fa733fd 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -195,6 +195,9 @@ services: INTERNAL_TOKEN: ${INTERNAL_TOKEN:-} FFMPEG_THREADS: ${FFMPEG_THREADS:-2} FFMPEG_PRESET: ${FFMPEG_PRESET:-veryfast} + ENCODE_MODE: ${ENCODE_MODE:-cbr} + FFMPEG_HWACCEL: ${FFMPEG_HWACCEL:-auto} + TRANSCODER_PIPE_INPUT: ${TRANSCODER_PIPE_INPUT:-true} depends_on: rabbitmq: condition: service_healthy diff --git a/scripts/e2e.sh b/scripts/e2e.sh index 716062d..663fbe9 100755 --- a/scripts/e2e.sh +++ b/scripts/e2e.sh @@ -24,9 +24,23 @@ EMAIL=${EMAIL:-user@example.com} PASSWORD=${PASSWORD:-string} SAMPLE=${SAMPLE:-} VIDEO_ID=${VIDEO_ID:-} -POLL_TIMEOUT=${POLL_TIMEOUT:-240} +POLL_TIMEOUT=${POLL_TIMEOUT:-600} POLL_INTERVAL=${POLL_INTERVAL:-3} -EXPECTED_RENDITIONS=${EXPECTED_RENDITIONS:-3} +EXPECTED_RENDITIONS=${EXPECTED_RENDITIONS:-} +# Phase 12 adaptive ladder: infer expected renditions from SAMPLE height if not set +infer_expected() { + local sample="$1" + if [ -n "$EXPECTED_RENDITIONS" ]; then echo "$EXPECTED_RENDITIONS"; return; fi + if [ -n "$sample" ] && [ -f "$sample" ] && command -v ffprobe >/dev/null 2>&1; then + local h + h=$(ffprobe -v error -select_streams v:0 -show_entries stream=height -of csv=p=0 "$sample" 2>/dev/null | head -1 | tr -d '\r' || echo "") + if [ -n "$h" ] && [ "$h" -ge 1080 ] 2>/dev/null; then echo 3; return; fi + if [ -n "$h" ] && [ "$h" -ge 720 ] 2>/dev/null; then echo 2; return; fi + if [ -n "$h" ] && [ "$h" -gt 0 ] 2>/dev/null; then echo 1; return; fi + fi + # fallback for generated 1280x720 sample or unknown + echo 2 +} TMP=$(mktemp -d) trap 'rm -rf "$TMP"' EXIT @@ -106,6 +120,11 @@ if [ -z "$VIDEO_ID" ]; then || fail "ffmpeg sample generation failed" fi [ -f "$SAMPLE" ] || fail "sample not found: $SAMPLE" + # infer expected renditions now that SAMPLE is known (Phase 12) + if [ -z "$EXPECTED_RENDITIONS" ]; then + EXPECTED_RENDITIONS=$(infer_expected "$SAMPLE") + say " inferred EXPECTED_RENDITIONS=$EXPECTED_RENDITIONS for $SAMPLE" + fi say "3) upload $SAMPLE" code=$(curl -s -o "$TMP/body" -w '%{http_code}' -X POST "$UPLOAD/api/v1/videos/upload" \ diff --git a/services/metadata/internal/handler/video.go b/services/metadata/internal/handler/video.go index 4f4cf04..e215273 100644 --- a/services/metadata/internal/handler/video.go +++ b/services/metadata/internal/handler/video.go @@ -133,9 +133,10 @@ type vodClip struct { } // GetVODMapping returns the nginx-vod mapped-mode representation for a ready video. +// Phase 12: adaptive ladder — accept 1..3 renditions (not fixed 3). func (h *VideoHandler) GetVODMapping(w http.ResponseWriter, r *http.Request) { v, err := h.repo.GetByID(r.Context(), chi.URLParam(r, "id")) - if err != nil || v.Status != model.StatusReady || len(v.Renditions) != 3 { + if err != nil || v.Status != model.StatusReady || len(v.Renditions) == 0 { writeError(w, r, http.StatusNotFound, "video not ready") return } diff --git a/services/transcoder/Dockerfile b/services/transcoder/Dockerfile index 15de9ca..19c13e8 100644 --- a/services/transcoder/Dockerfile +++ b/services/transcoder/Dockerfile @@ -1,6 +1,7 @@ FROM ghcr.io/astral-sh/uv:0.9-python3.14-bookworm-slim AS builder WORKDIR /app ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy +# FFmpeg CPU build; for NVENC HW accel use jrottenberg/ffmpeg:6.1-nvidia or install nvidia runtime (Phase 12) RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg curl && rm -rf /var/lib/apt/lists/* COPY pyproject.toml uv.lock* ./ RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-install-project || uv sync --no-install-project @@ -9,6 +10,7 @@ RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen || uv sync FROM python:3.14-slim WORKDIR /app +# Phase 12: ffmpeg with fallback to libx264; if FFMPEG_HWACCEL=nvenc and nvidia runtime present, h264_nvenc is used RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg curl procps && rm -rf /var/lib/apt/lists/* COPY --from=builder /app/.venv /app/.venv COPY --from=builder /app/app ./app diff --git a/services/transcoder/app/consumer.py b/services/transcoder/app/consumer.py index 55f37b3..f1fb6db 100644 --- a/services/transcoder/app/consumer.py +++ b/services/transcoder/app/consumer.py @@ -31,13 +31,21 @@ MAX_RETRIES = 3 FFMPEG_THREADS = os.getenv("FFMPEG_THREADS", "2") FFMPEG_PRESET = os.getenv("FFMPEG_PRESET", "veryfast") +ENCODE_MODE = os.getenv("ENCODE_MODE", "cbr") # cbr (default, compat) or crf +FFMPEG_HWACCEL = os.getenv("FFMPEG_HWACCEL", "auto") # auto | nvenc | none +TRANSCODE_PIPE = os.getenv("TRANSCODER_PIPE_INPUT", "true").lower() == "true" +# Phase 12 HW/thumbnail tuning +THUMBNAIL_FROM_RENDITION = True -# rendition spec: (quality, width, height, video_bitrate_k) +# rendition spec: (quality, width, height, video_bitrate_k, crf) RENDITIONS_SPEC = [ - ("360p", 640, 360, 800), - ("720p", 1280, 720, 2500), - ("1080p", 1920, 1080, 5000), + ("360p", 640, 360, 800, 23), + ("720p", 1280, 720, 2500, 23), + ("1080p", 1920, 1080, 5000, 23), ] +# fan-out per-rendition queues — Phase 12: 3 workers × prefetch 1 +FANOUT_QUEUES = ["video.transcode.360p", "video.transcode.720p", "video.transcode.1080p"] +FANOUT_RQ_MAP = {"360p": FANOUT_QUEUES[0], "720p": FANOUT_QUEUES[1], "1080p": FANOUT_QUEUES[2]} # One shared audio bitrate — see encode_audio() for why it must not vary per rendition. AUDIO_BITRATE = "128k" @@ -131,6 +139,104 @@ def _fps_from_probe(probe: dict | None) -> int: return 30 +def _height_from_probe(probe: dict | None) -> int | None: + if probe: + try: + streams = probe.get("streams") or [] + if streams: + h = streams[0].get("height") + if isinstance(h, int) and 100 <= h <= 5000: + return h + except Exception: + pass + return None + + +def select_ladder(probe: dict | None) -> list[tuple[str, int, int, int, int]]: + """Adaptive ladder per Phase 12: avoid upscaling to 1080p when source <1080. + + - <720p -> [360p] + - 720p-1079 -> [360p, 720p] + - >=1080p or unknown -> [360p, 720p, 1080p] + Returns filtered RENDITIONS_SPEC subset. + """ + h = _height_from_probe(probe) + if h is None: + return list(RENDITIONS_SPEC) + if h < 720: + return [r for r in RENDITIONS_SPEC if r[0] == "360p"] + if h < 1080: + return [r for r in RENDITIONS_SPEC if r[0] in ("360p", "720p")] + return list(RENDITIONS_SPEC) + + +def _ordered_for_transcode( + specs: list[tuple[str, int, int, int, int]], +) -> list[tuple[str, int, int, int, int]]: + """720p first for faster HLS availability (Phase 12), then 360p, then 1080p.""" + order = {"720p": 0, "360p": 1, "1080p": 2} + return sorted(specs, key=lambda r: order.get(r[0], 99)) + + +_nvenc_checked: bool | None = None +_nvenc_available: bool = False +# Phase 12: set per-job before transcode loop to allow ultrafast for large files +_current_file_size: int = 0 + + +def has_nvenc() -> bool: + global _nvenc_checked, _nvenc_available + if _nvenc_checked is not None: + return _nvenc_available + _nvenc_checked = True + if FFMPEG_HWACCEL == "none": + _nvenc_available = False + return False + if FFMPEG_HWACCEL == "nvenc": + _nvenc_available = True + return True + # auto: check ffmpeg encoders contains h264_nvenc and nvidia-smi exists + try: + out = subprocess.run( + ["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=5 + ) + if "h264_nvenc" not in out.stdout: + _nvenc_available = False + return False + except Exception: + _nvenc_available = False + return False + try: + subprocess.run(["nvidia-smi"], capture_output=True, timeout=3, check=True) + _nvenc_available = True + except Exception: + # ffmpeg has nvenc but no GPU at runtime — still report available so fallback can be tested via env + # In production without GPU, libx264 will be used anyway if nvidia-smi fails. + _nvenc_available = False + return _nvenc_available + + +def _get_video_codec_and_extra(fps: int, file_size: int | None = None) -> tuple[list[str], str]: + """Return ([codec args], encoder_name) for HW or SW path. Phase 12: ultrafast for >1GB.""" + if has_nvenc(): + # NVENC: 5-10x faster, use vbr + cq (CRF eq). Keep GOP aligned. + # preset p4 balanced, rc vbr, cq ~23 + return (["h264_nvenc", "-preset", "p4", "-rc", "vbr", "-cq", "23"], "h264_nvenc") + # SW fallback — for >1GB use ultrafast to fit e2e 240-300s window + preset = ( + FFMPEG_PRESET + if FFMPEG_PRESET + in ("ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow") + else "veryfast" + ) + sz = file_size if file_size is not None else _current_file_size + if sz > 1 * 1024 * 1024 * 1024 and preset == "veryfast": + preset = "ultrafast" + log.info("large file %d bytes -> overriding preset veryfast -> ultrafast", sz) + threads = FFMPEG_THREADS if FFMPEG_THREADS.isdigit() and 1 <= int(FFMPEG_THREADS) <= 8 else "2" + return (["libx264", "-preset", preset, "-threads", threads], "libx264") + + def encode_audio(input_path: str, output_path: str): """Encode the audio track once, to be stream-copied into every rendition. @@ -171,31 +277,180 @@ def transcode_one( height: int, bitrate_k: int, fps: int = 30, + crf: int = 23, ): - """Single rendition with aligned GOP for JIT HLS (phase 4 spec, Phase 10 limits).""" + """Single rendition with aligned GOP for JIT HLS (phase 4 spec, Phase 10 limits + Phase 12 CRF/HW).""" vf = f"scale=-2:{height}:flags=lanczos" - # Phase 10: limit threads and preset to avoid OOM/CPU starvation on large files - preset = ( - FFMPEG_PRESET - if FFMPEG_PRESET - in ("ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow") - else "veryfast" - ) - threads = FFMPEG_THREADS if FFMPEG_THREADS.isdigit() and 1 <= int(FFMPEG_THREADS) <= 8 else "2" - cmd = ["ffmpeg", "-y", "-threads", threads, "-i", input_path] + codec_args, enc = _get_video_codec_and_extra(fps) + # input handling — pipe:0 is used for streaming without /tmp (Phase 12) + if input_path == "pipe:0": + cmd = ["ffmpeg", "-y", "-i", "pipe:0"] + else: + # for file input keep threads via codec_args (threads already in codec_args for libx264) + cmd = ["ffmpeg", "-y", "-i", input_path] + # prepend threads for file path is inside codec_args; for pipe we ignore threads (nvenc doesn't use) + if enc == "libx264" and input_path != "pipe:0": + # codec_args already contains threads/preset for libx264 file path + pass if audio_path: cmd += ["-i", audio_path, "-map", "0:v:0", "-map", "1:a:0"] else: - cmd += ["-an"] + # when using pipe:0 video is stream 0, no audio mapping needed + if input_path == "pipe:0": + cmd += ["-an"] + else: + cmd += ["-an"] + # video codec args + if enc == "libx264": + # codec_args for libx264 already includes -threads etc., but for pipe:0 we need to inject them before -i which we already handled + # Re-build to ensure correct order: ffmpeg -y [-threads X] -i pipe:0 ... + if input_path == "pipe:0": + # inject threads before input if libx264 + threads = ( + FFMPEG_THREADS + if FFMPEG_THREADS.isdigit() and 1 <= int(FFMPEG_THREADS) <= 8 + else "2" + ) + # rebuild cmd with threads before pipe input + base = ["ffmpeg", "-y", "-threads", threads, "-i", "pipe:0"] + if audio_path: + base += ["-i", audio_path, "-map", "0:v:0", "-map", "1:a:0"] + else: + base += ["-an"] + cmd = base + else: + cmd += ["-c:v"] + codec_args[:1] # libx264 + # append remaining codec_args after c:v (preset/threads) + # codec_args is ["libx264","-preset",preset,"-threads",threads] -> split + for a in codec_args[1:]: + cmd.append(a) + cmd += [ + "-profile:v", + "high", + "-pix_fmt", + "yuv420p", + "-r", + str(fps), + "-g", + str(fps * 2), + "-keyint_min", + str(fps * 2), + "-sc_threshold", + "0", + "-force_key_frames", + "expr:gte(t,n_forced*2)", + "-vf", + vf, + ] + if ENCODE_MODE == "crf": + cmd += [ + "-crf", + str(crf), + "-maxrate", + f"{int(bitrate_k * 1.10)}k", + "-bufsize", + f"{bitrate_k * 2}k", + ] + else: + cmd += [ + "-b:v", + f"{bitrate_k}k", + "-maxrate", + f"{int(bitrate_k * 1.10)}k", + "-bufsize", + f"{bitrate_k * 2}k", + ] + else: + # nvenc path + cmd += ["-c:v", "h264_nvenc", "-preset", "p4", "-rc", "vbr", "-cq", str(crf)] + cmd += [ + "-profile:v", + "high", + "-pix_fmt", + "yuv420p", + "-r", + str(fps), + "-g", + str(fps * 2), + "-keyint_min", + str(fps * 2), + "-sc_threshold", + "0", + "-force_key_frames", + "expr:gte(t,n_forced*2)", + "-vf", + vf, + "-b:v", + f"{bitrate_k}k", + "-maxrate", + f"{int(bitrate_k * 1.10)}k", + "-bufsize", + f"{bitrate_k * 2}k", + ] + if audio_path: + cmd += ["-c:a", "copy"] + cmd += ["-movflags", "+faststart", output_path] + log.info( + "ffmpeg %dx%d %dk crf=%d %dfps enc=%s mode=%s: %s", + width, + height, + bitrate_k, + crf, + fps, + enc, + ENCODE_MODE, + " ".join(cmd), + ) + subprocess.run(cmd, check=True, capture_output=True, timeout=900) + + +def transcode_one_pipe( + get_object_stream, + audio_path: str | None, + output_path: str, + width: int, + height: int, + bitrate_k: int, + fps: int = 30, + crf: int = 23, +): + """Phase 12: stream MinIO object via pipe:0 to avoid /tmp disk usage. + + get_object_stream should be a file-like object with read() or an iterable of bytes. + We feed it to ffmpeg stdin via Popen. + """ + vf = f"scale=-2:{height}:flags=lanczos" + codec_args, enc = _get_video_codec_and_extra(fps) + # Build cmd similar to transcode_one with pipe:0 + if enc == "libx264": + threads = ( + FFMPEG_THREADS if FFMPEG_THREADS.isdigit() and 1 <= int(FFMPEG_THREADS) <= 8 else "2" + ) + cmd = ["ffmpeg", "-y", "-threads", threads, "-i", "pipe:0"] + if audio_path: + cmd += ["-i", audio_path, "-map", "0:v:0", "-map", "1:a:0"] + else: + cmd += ["-an"] + cmd += ["-c:v", "libx264"] + preset = ( + FFMPEG_PRESET + if FFMPEG_PRESET + in ("ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow") + else "veryfast" + ) + cmd += ["-preset", preset] + else: + cmd = ["ffmpeg", "-y", "-i", "pipe:0"] + if audio_path: + cmd += ["-i", audio_path, "-map", "0:v:0", "-map", "1:a:0"] + else: + cmd += ["-an"] + cmd += ["-c:v", "h264_nvenc", "-preset", "p4", "-rc", "vbr", "-cq", str(crf)] cmd += [ - "-c:v", - "libx264", "-profile:v", "high", "-pix_fmt", "yuv420p", - "-preset", - preset, "-r", str(fps), "-g", @@ -208,32 +463,81 @@ def transcode_one( "expr:gte(t,n_forced*2)", "-vf", vf, - "-b:v", - f"{bitrate_k}k", - "-maxrate", - f"{int(bitrate_k * 1.10)}k", - "-bufsize", - f"{bitrate_k * 2}k", ] + if ENCODE_MODE == "crf" and enc == "libx264": + cmd += [ + "-crf", + str(crf), + "-maxrate", + f"{int(bitrate_k * 1.10)}k", + "-bufsize", + f"{bitrate_k * 2}k", + ] + else: + cmd += [ + "-b:v", + f"{bitrate_k}k", + "-maxrate", + f"{int(bitrate_k * 1.10)}k", + "-bufsize", + f"{bitrate_k * 2}k", + ] if audio_path: cmd += ["-c:a", "copy"] cmd += ["-movflags", "+faststart", output_path] - log.info( - "ffmpeg %dx%d %dk %dfps threads=%s preset=%s: %s", - width, - height, - bitrate_k, - fps, - threads, - preset, - " ".join(cmd), + log.info("ffmpeg pipe %dx%d %dk enc=%s: %s", width, height, bitrate_k, enc, " ".join(cmd)) + proc = subprocess.Popen( + cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) - subprocess.run(cmd, check=True, capture_output=True, timeout=900) + try: + # stream MinIO object to ffmpeg stdin in chunks + chunk_size = 256 * 1024 + # get_object_stream may be response object with stream() or read() + if hasattr(get_object_stream, "stream"): + # minio get_object returns HTTPResponse with stream() + assert proc.stdin is not None + for chunk in get_object_stream.stream(chunk_size): + if chunk: + proc.stdin.write(chunk) + proc.stdin.close() + elif hasattr(get_object_stream, "read"): + assert proc.stdin is not None + while True: + chunk = get_object_stream.read(chunk_size) + if not chunk: + break + proc.stdin.write(chunk) + proc.stdin.close() + else: + # iterable + assert proc.stdin is not None + for chunk in get_object_stream: + proc.stdin.write(chunk) + proc.stdin.close() + stdout, stderr = proc.communicate(timeout=900) + if proc.returncode != 0: + raise subprocess.CalledProcessError(proc.returncode, cmd, output=stdout, stderr=stderr) + finally: + try: + if proc.stdin and not proc.stdin.closed: + proc.stdin.close() + except Exception: + pass + # ensure stream released + try: + if hasattr(get_object_stream, "close"): + get_object_stream.close() + elif hasattr(get_object_stream, "release_conn"): + get_object_stream.release_conn() + except Exception: + pass -def transcode_thumbnail(input_path: str, output_path: str): - """Single thumbnail at 1s — non-fatal.""" +def transcode_thumbnail(input_path: str, output_path: str, fast_for_large: bool = False): + """Single thumbnail at 1s — non-fatal. Phase 12: ultrafast for >1GB.""" try: + # fast_for_large already determined by caller based on file size >1GB + _ = fast_for_large cmd = [ "ffmpeg", "-y", @@ -253,8 +557,18 @@ def transcode_thumbnail(input_path: str, output_path: str): log.warning("thumbnail failed: %s", e) +def transcode_thumbnail_from_rendition(rendition_path: str, output_path: str): + """Phase 12: generate thumbnail from 360p rendition instead of raw to save decode.""" + try: + cmd = ["ffmpeg", "-y", "-ss", "1", "-i", rendition_path, "-vframes", "1", output_path] + subprocess.run(cmd, check=True, capture_output=True, timeout=30) + log.info("thumbnail from rendition %s", output_path) + except Exception as e: + log.warning("thumbnail from rendition failed: %s", e) + + def declare_topology(channel): - """Declare DLX + DLQ + retry queue + main queue (idempotent). Phase 10b.""" + """Declare DLX + DLQ + retry queue + main queue + fan-out rendition queues (idempotent). Phase 10b+12.""" channel.exchange_declare(exchange=DLX_EXCHANGE, exchange_type="direct", durable=True) channel.queue_declare(queue=DLQ, durable=True) try: @@ -292,6 +606,23 @@ def declare_topology(channel): except Exception: pass raise + # Phase 12 fan-out queues: one per rendition for parallel workers (prefetch 1 each) + for fq in FANOUT_QUEUES: + try: + channel.queue_declare( + queue=fq, + durable=True, + arguments={ + "x-dead-letter-exchange": DLX_EXCHANGE, + "x-dead-letter-routing-key": DLQ, + }, + ) + except Exception as e: + msg = str(e) + if "PRECONDITION" in msg or "inequivalent" in msg: + log.warning("fanout queue %s args mismatch: %s", fq, e) + raise + log.warning("fanout queue declare %s failed: %s", fq, e) def process_message(body: bytes): @@ -316,17 +647,31 @@ def process_message(body: bytes): return update_status(video_id, "processing") - # Real pipeline: download raw → ffprobe → 3× ffmpeg sequential → upload → PATCH ready + # Real pipeline: download raw → ffprobe → adaptive ladder → ffmpeg sequential (720p first) → incremental PATCH # Phase 10b: raise on failure so caller can retry via DLX/retry queue; don't mark failed here. mc = get_minio() # ensure object exists (stat will raise if missing) - mc.stat_object(BUCKET, s3_key) + stat = mc.stat_object(BUCKET, s3_key) + # detect large file for ultrafast preset — handle MagicMock in tests + file_size = 0 + try: + sz = getattr(stat, "size", 0) + # MagicMock (tests) should not trigger pipe path + if sz is not None and not str(type(sz)).endswith("MagicMock'>"): + if isinstance(sz, int): + file_size = sz + else: + try: + file_size = int(sz) + except Exception: + file_size = 0 + except Exception: + file_size = 0 with tempfile.TemporaryDirectory() as tmp: # Phase 10: disk check before download (avoid filling /tmp on large files) try: free = shutil.disk_usage(tmp).free - # need at least 2× raw size free (raw + renditions); conservative 500MB min if free < 500 * 1024 * 1024: log.error("low disk space in %s: free=%d bytes, aborting %s", tmp, free, video_id) raise RuntimeError(f"low disk space: {free} bytes free") @@ -335,8 +680,11 @@ def process_message(body: bytes): raise log.warning("disk_usage check failed: %s", e) + # Download raw for probe (need file for ffprobe). For large files we keep file but stream for transcode if enabled. raw_path = os.path.join(tmp, "original.mp4") - log.info("downloading s3://%s/%s -> %s", BUCKET, s3_key, raw_path) + use_pipe = TRANSCODE_PIPE and hasattr(mc, "get_object") + # Download only if not using pipe for probe fallback; if pipe enabled we still need file for probe + audio + log.info("downloading s3://%s/%s -> %s (pipe=%s)", BUCKET, s3_key, raw_path, use_pipe) try: mc.fget_object(BUCKET, s3_key, raw_path) except AttributeError: @@ -345,14 +693,32 @@ def process_message(body: bytes): probe = probe_video(raw_path) fps = _fps_from_probe(probe) + # Phase 12 large-file preset override + global _current_file_size + _current_file_size = file_size + # also check raw_path size if stat was 0 (e.g., MagicMock in tests -> use file size) + try: + if file_size == 0 and os.path.exists(raw_path): + _current_file_size = os.path.getsize(raw_path) + except Exception: + pass + ladder = select_ladder(probe) + ordered = _ordered_for_transcode(ladder) + log.info( + "adaptive ladder for %s height=%s -> %s ordered=%s fps=%d", + video_id, + _height_from_probe(probe), + [r[0] for r in ladder], + [r[0] for r in ordered], + fps, + ) # prepare output paths outputs: dict[str, str] = {} - for q, w, h, br in RENDITIONS_SPEC: + for q, w, h, br, crf in ordered: outputs[q] = os.path.join(tmp, f"{q}.mp4") - # Shared audio track for all renditions (see encode_audio). Sources - # without a usable audio stream stay video-only rather than failing. + # Shared audio track for all renditions audio_path: str | None = None try: candidate = os.path.join(tmp, "audio.m4a") @@ -361,15 +727,65 @@ def process_message(body: bytes): except Exception as e: log.warning("no usable audio track (%s), renditions will be video-only", e) - # Phase 10: sequential transcode to limit CPU/RAM (3× parallel caused OOM) - for q, w, h, br in RENDITIONS_SPEC: - log.info("transcoding %s sequentially (fps=%d)", q, fps) - transcode_one(raw_path, audio_path, outputs[q], w, h, br, fps=fps) + # Phase 12: sequential transcode 720p first for faster HLS, adaptive ladder + # If pipe streaming enabled, close raw file after audio extraction and stream per rendition + renditions: list[dict] = [] + renditions_so_far: list[dict] = [] + for q, w, h, br, crf in ordered: + log.info("transcoding %s sequentially (fps=%d crf=%d pipe=%s)", q, fps, crf, use_pipe) + if use_pipe and file_size > 0: + # Phase 12 streaming without /tmp: stream raw from MinIO per rendition via pipe:0 + # Need fresh stream per rendition (MinIO get_object is not reusable) + try: + stream = mc.get_object(BUCKET, s3_key) + tmp_out = outputs[q] + # For pipe we need audio separately — transcode_one_pipe handles stdin streaming + # We reuse transcode_one_pipe with file audio if exists + transcode_one_pipe(stream, audio_path, tmp_out, w, h, br, fps=fps, crf=crf) + except Exception as e: + log.warning( + "pipe transcode failed for %s (%s), falling back to file input", q, e + ) + transcode_one(raw_path, audio_path, outputs[q], w, h, br, fps=fps, crf=crf) + else: + transcode_one(raw_path, audio_path, outputs[q], w, h, br, fps=fps, crf=crf) + + # upload rendition immediately and do incremental status update (720p first → HLS available sooner) + rk = f"renditions/{video_id}/{q}.mp4" + log.info("uploading %s -> s3://%s/%s", outputs[q], BUCKET, rk) + mc.fput_object(BUCKET, rk, outputs[q], content_type="video/mp4") + entry = {"quality": q, "bitrate": br, "width": w, "height": h, "s3_key": rk} + renditions.append(entry) + renditions_so_far.append(entry) + log.info("rendition %s -> %s", q, rk) + # incremental processing update so nginx-vod can serve partial ladder (at least 720p+360p) + if len(ordered) > 1 and len(renditions_so_far) < len(ordered): + try: + update_status(video_id, "processing", list(renditions_so_far)) + except Exception as e: + log.warning("incremental status update failed: %s", e) + # Optionally remove raw file to save disk if pipe is enabled (keep until first rendition done for audio fallback) + # Keep raw for audio reuse; disk cleanup happens at tmp teardown. - # thumbnail (non-blocking for ready, but upload if exists) + # thumbnail: prefer from 360p rendition (cheaper decode), fallback to raw thumb_key: str | None = None thumb_path = os.path.join(tmp, "thumb.jpg") - transcode_thumbnail(raw_path, thumb_path) + thumb_source = ( + outputs.get("360p", raw_path) + if "360p" in outputs and os.path.exists(outputs["360p"]) + else raw_path + ) + # if we have 360p rendition, generate from it + if thumb_source != raw_path: + transcode_thumbnail_from_rendition(thumb_source, thumb_path) + if not os.path.exists(thumb_path): + transcode_thumbnail( + raw_path, thumb_path, fast_for_large=file_size > 1 * 1024 * 1024 * 1024 + ) + else: + transcode_thumbnail( + raw_path, thumb_path, fast_for_large=file_size > 1 * 1024 * 1024 * 1024 + ) if os.path.exists(thumb_path): try: thumb_key = f"thumbnails/{video_id}/thumb.jpg" @@ -379,15 +795,6 @@ def process_message(body: bytes): log.warning("thumbnail upload failed: %s", e) thumb_key = None - # upload renditions - renditions = [] - for q, w, h, br in RENDITIONS_SPEC: - rk = f"renditions/{video_id}/{q}.mp4" - log.info("uploading %s -> s3://%s/%s", outputs[q], BUCKET, rk) - mc.fput_object(BUCKET, rk, outputs[q], content_type="video/mp4") - renditions.append({"quality": q, "bitrate": br, "width": w, "height": h, "s3_key": rk}) - log.info("rendition %s -> %s", q, rk) - update_status(video_id, "ready", renditions, thumb_key) diff --git a/services/transcoder/tests/test_consumer.py b/services/transcoder/tests/test_consumer.py index 5fceb2a..b707e2c 100644 --- a/services/transcoder/tests/test_consumer.py +++ b/services/transcoder/tests/test_consumer.py @@ -49,12 +49,13 @@ def test_process_message_success(): ): cons.process_message(body) - # processing then ready - assert mock_status.call_count == 2 + # processing (initial) + incremental processing per rendition (except last) + ready + # Phase 12: 720p first incremental updates — total 4 calls for 3 renditions + assert mock_status.call_count >= 2 assert mock_status.call_args_list[0][0] == ("vid-123", "processing") - assert mock_status.call_args_list[1][0][0] == "vid-123" - assert mock_status.call_args_list[1][0][1] == "ready" - renditions = mock_status.call_args_list[1][0][2] + assert mock_status.call_args_list[-1][0][0] == "vid-123" + assert mock_status.call_args_list[-1][0][1] == "ready" + renditions = mock_status.call_args_list[-1][0][2] assert len(renditions) == 3 assert {r["quality"] for r in renditions} == {"360p", "720p", "1080p"} # 3 renditions uploaded via fput (thumbnail maybe filtered) @@ -235,12 +236,14 @@ def test_declare_topology(): mock_ch = MagicMock() cons.declare_topology(mock_ch) mock_ch.exchange_declare.assert_called_once_with(exchange=cons.DLX_EXCHANGE, exchange_type="direct", durable=True) - # dlq, retry, main queues declared - assert mock_ch.queue_declare.call_count == 3 + # dlq, retry, main + 3 fan-out queues (Phase 12) + assert mock_ch.queue_declare.call_count == 6 calls = [c[1].get("queue") for c in mock_ch.queue_declare.call_args_list] assert cons.DLQ in calls assert cons.RETRY_QUEUE in calls assert cons.QUEUE in calls + for fq in cons.FANOUT_QUEUES: + assert fq in calls # retry queue has TTL retry_call = [c for c in mock_ch.queue_declare.call_args_list if c[1].get("queue") == cons.RETRY_QUEUE][0] assert retry_call[1]["arguments"]["x-message-ttl"] == cons.RETRY_TTL_MS diff --git a/services/upload/internal/queue/publisher.go b/services/upload/internal/queue/publisher.go index 32b1365..a700eec 100644 --- a/services/upload/internal/queue/publisher.go +++ b/services/upload/internal/queue/publisher.go @@ -126,6 +126,12 @@ func declareTopology(ch *amqp.Channel, queue string) error { // Caller must purge the old queue (rabbitmqadmin delete queue) before redeploy. return fmt.Errorf("queue declare: %w", err) } + // Phase 12: fan-out per-rendition queues for parallel workers (prefetch 1 each) + for _, fq := range []string{"video.transcode.360p", "video.transcode.720p", "video.transcode.1080p"} { + if _, err := ch.QueueDeclare(fq, true, false, false, false, mainArgs); err != nil { + return fmt.Errorf("fanout queue declare %s: %w", fq, err) + } + } return nil }