Setting Up Fallback Routing for Failed Transcodes

import asyncio
import logging
import re
from dataclasses import dataclass, field
from typing import Optional, Tuple

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("fallback_router")

@dataclass
class TranscodeConfig:
    input_path: str
    output_path: str
    primary_encoder: str = "h264_nvenc"
    fallback_encoder: str = "libx264"
    timeout: float = 300.0
    max_retries: int = 1
    stderr_window: int = 50
    fallback_delay: float = 2.0

class FallbackRouter:
    def __init__(self, config: TranscodeConfig):
        self.config = config
        self.state = {
            "attempts": 0,
            "fallback_triggered": False,
            "failure_mode": None,
            "diagnostics": []
        }

    def _build_ffmpeg_cmd(self, encoder: str) -> list[str]:
        return [
            "ffmpeg", "-y", "-i", self.config.input_path,
            "-c:v", encoder, "-preset", "medium",
            "-c:a", "aac", "-b:a", "192k",
            self.config.output_path
        ]

    async def _run_ffmpeg(self, cmd: list[str]) -> Tuple[int, str, str]:
        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.config.timeout)
            return proc.returncode or 0, stdout.decode("utf-8", errors="replace"), stderr.decode("utf-8", errors="replace")
        except asyncio.TimeoutError:
            try:
                proc.kill()
            except ProcessLookupError:
                pass
            logger.error("Transcode timed out after %.1fs", self.config.timeout)
            return -1, "", "TIMEOUT_EXCEEDED"
        except Exception as e:
            logger.error("Subprocess execution failed: %s", e)
            return -2, "", f"SUBPROCESS_ERROR: {e}"

    def _classify_failure(self, exit_code: int, stderr: str) -> Optional[str]:
        tail = "\n".join(stderr.strip().splitlines()[-self.config.stderr_window:])
        patterns = {
            "CUDA_INIT": re.compile(r"(Failed to initialize CUDA|Cannot load libcuda\.so|nvenc.*init.*error)", re.I),
            "PROFILE_MISMATCH": re.compile(r"(Error while opening encoder|Profile/level not supported)", re.I),
            "CONTAINER_DEADLOCK": re.compile(r"(Invalid data found when processing input|moov atom not found)", re.I),
            "OOM_KILLED": re.compile(r"(Out of memory|signal 9|OOMKilled)", re.I)
        }
        for mode, pattern in patterns.items():
            if pattern.search(tail):
                return mode
        return "GENERIC_FAILURE" if exit_code != 0 else None

    async def execute(self) -> bool:
        for attempt in range(self.config.max_retries + 1):
            self.state["attempts"] = attempt + 1
            encoder = self.config.fallback_encoder if self.state["fallback_triggered"] else self.config.primary_encoder
            cmd = self._build_ffmpeg_cmd(encoder)

            logger.info("Attempt %d/%d | Encoder: %s", self.state["attempts"], self.config.max_retries + 1, encoder)
            exit_code, stdout, stderr = await self._run_ffmpeg(cmd)
            failure_mode = self._classify_failure(exit_code, stderr)

            self.state["diagnostics"].append({
                "attempt": self.state["attempts"],
                "encoder": encoder,
                "exit_code": exit_code,
                "failure_mode": failure_mode,
                "stderr_tail": stderr[-200:] if stderr else None
            })

            if failure_mode is None:
                self.state["failure_mode"] = None
                logger.info("Transcode completed successfully on attempt %d.", self.state["attempts"])
                return True

            self.state["failure_mode"] = failure_mode
            logger.warning("Failure detected: %s (Exit: %d). Preparing fallback routing.", failure_mode, exit_code)

            if not self.state["fallback_triggered"]:
                self.state["fallback_triggered"] = True
                logger.info("Routing to software fallback encoder: %s", self.config.fallback_encoder)
                await asyncio.sleep(self.config.fallback_delay)
            else:
                logger.error("Fallback failed. Aborting pipeline for %s.", self.config.input_path)
                return False

        return False

    def get_diagnostics_report(self) -> dict:
        return {
            "input": self.config.input_path,
            "final_status": "SUCCESS" if self.state["attempts"] > 0 and self.state["failure_mode"] is None else "FAILED",
            "failure_classification": self.state["failure_mode"],
            "execution_trace": self.state["diagnostics"]
        }