fix(worker): FFmpeg 超时保护 - 防止渲染 hang 住导致 worker 永久阻塞

根因：新引擎 UnifiedRenderService 所有 FFmpeg 调用通过 run_ffmpeg 执行，但 subprocess.run 未设置 timeout，FFmpeg hang 住时 worker 线程永久阻塞。旧引擎 compose_video 有单独的 timeout=3600，但新引擎路径没有。修复： 1. run_ffmpeg 新增默认超时 1800s（30分钟），支持自定义传参 2. 捕获 TimeoutExpired 并打 error 日志后重新抛出 3. probe_video_info 新增 timeout=15s 超时保护 4. 7个单元测试覆盖超时逻辑影响范围：所有通过 run_ffmpeg 调用的 FFmpeg 命令（unified_render_service 所有渲染/混音/合并操作）
feat(api): 新增渲染结果内部下载接口 - 支持按视频ID/任务ID获取预签名URL
2026-07-13 11:05:13 +08:00 · 2026-07-13 11:05:13 +08:00
5 changed files with 424 additions and 0 deletions
@@ -13,6 +13,7 @@ from app.api.routes.generated_videos import router as generated_videos_router
 from app.api.routes.generation_tasks import router as generation_tasks_router
 from app.api.routes.health import router as health_check_router
 from app.api.routes.ingest_jobs import router as ingest_jobs_router
+from app.api.routes.internal_render import router as internal_render_router
 from app.api.routes.jobs import router as jobs_router
 from app.api.routes.projects import router as projects_router
 from app.api.routes.recipes import router as recipes_router
@@ -156,3 +157,7 @@ api_router.include_router(
    feature_flags_router,
    tags=["Internal"],
 )
+api_router.include_router(
+    internal_render_router,
+    tags=["Internal"],
+)
@@ -0,0 +1,120 @@
+"""渲染结果内部下载接口。
+
+通过内部 API Key 鉴权，为灰度对比工具等内部系统提供渲染结果下载能力。
+
+API:
+  GET  /api/v1/internal/render/videos/{video_id}/download-url  - 获取单个视频下载URL
+  GET  /api/v1/internal/render/tasks/{task_id}/videos          - 获取任务下所有视频及下载URL
+
+鉴权：X-API-Key header，走内部 API Key 验证
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.api.routes.auth import _verify_internal_api_key
+from app.core.storage import OSSStorageService, get_storage_service
+from app.dependencies import get_generated_video_repository
+from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/internal/render", tags=["Internal"])
+
+
+class InternalRenderVideoItem(BaseModel):
+    """内部渲染视频项。"""
+
+    video_id: str
+    generation_task_id: str
+    project_id: str
+    name: str
+    file_url: str
+    file_size: int | None = None
+    duration: float | None = None
+    width: int | None = None
+    height: int | None = None
+    fps: float | None = None
+    status: str
+    download_url: str
+
+
+class InternalRenderTaskVideosResponse(BaseModel):
+    """任务下所有渲染视频响应。"""
+
+    task_id: str
+    count: int
+    videos: list[InternalRenderVideoItem]
+
+
+class InternalRenderDownloadUrlResponse(BaseModel):
+    """单个视频下载URL响应。"""
+
+    video_id: str
+    download_url: str
+
+
+def _video_to_item(video: Any, download_url: str) -> InternalRenderVideoItem:
+    """将 GeneratedVideo 领域对象转为响应项。"""
+    return InternalRenderVideoItem(
+        video_id=video.id,
+        generation_task_id=video.generation_task_id,
+        project_id=video.project_id,
+        name=video.name,
+        file_url=video.file_url,
+        file_size=getattr(video, "file_size", None),
+        duration=getattr(video, "duration", None),
+        width=getattr(video, "width", None),
+        height=getattr(video, "height", None),
+        fps=getattr(video, "fps", None),
+        status=video.status,
+        download_url=download_url,
+    )
+
+
+@router.get("/videos/{video_id}/download-url", response_model=InternalRenderDownloadUrlResponse)
+def get_render_video_download_url(
+    video_id: str,
+    _: bool = Depends(_verify_internal_api_key),
+    generated_video_repository: Any = Depends(get_generated_video_repository),
+    storage_service: OSSStorageService = Depends(get_storage_service),
+) -> InternalRenderDownloadUrlResponse:
+    """获取单个渲染视频的下载URL（预签名）。"""
+    video = generated_video_repository.get(video_id)
+    if video is None:
+        raise HTTPException(status_code=404, detail=f"GeneratedVideo {video_id} not found")
+
+    download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400)
+    logger.info("内部渲染下载URL生成: video_id=%s", video_id)
+    return InternalRenderDownloadUrlResponse(video_id=video_id, download_url=download_url)
+
+
+@router.get("/tasks/{task_id}/videos", response_model=InternalRenderTaskVideosResponse)
+def get_render_task_videos(
+    task_id: str,
+    status: str | None = Query(None, description="按状态筛选，如 completed/failed"),
+    _: bool = Depends(_verify_internal_api_key),
+    generated_video_repository: Any = Depends(get_generated_video_repository),
+    storage_service: OSSStorageService = Depends(get_storage_service),
+) -> InternalRenderTaskVideosResponse:
+    """获取生成任务下所有渲染视频及下载URL。"""
+    videos = generated_video_repository.list_by_generation_task(task_id)
+
+    # 状态筛选
+    if status:
+        videos = [v for v in videos if v.status == status]
+
+    items = []
+    for video in videos:
+        download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400)
+        items.append(_video_to_item(video, download_url))
+
+    logger.info("内部渲染任务视频查询: task_id=%s count=%d", task_id, len(items))
+    return InternalRenderTaskVideosResponse(
+        task_id=task_id,
+        count=len(items),
+        videos=items,
+    )
@@ -42,6 +42,10 @@ XFADE_TRANSITION_MAP: dict[str, str] = {

 DEFAULT_TRANSITION_DURATION = 0.5

+# FFmpeg 执行默认超时（秒），防止 FFmpeg hang 住导致 worker 永久阻塞
+# 默认 30 分钟，足够处理大部分短视频渲染；超长视频可单独传参覆盖
+DEFAULT_FFMPEG_TIMEOUT = 1800
+

 # ── FFmpeg 执行 ───────────────────────────────────────────────────────────────

@@ -50,12 +54,14 @@ def run_ffmpeg(
    command: list[str],
    *,
    capture_output: bool = True,
+    timeout: int | None = DEFAULT_FFMPEG_TIMEOUT,
 ) -> tuple[str, str]:
    """执行 FFmpeg 命令。

    Args:
        command: 完整的 ffmpeg 命令列表（含 "ffmpeg" 本身）
        capture_output: 是否捕获 stdout/stderr
+        timeout: 超时时间（秒），默认 1800s（30分钟）；None 表示不设超时（不推荐）

    Returns:
        (stdout, stderr) 元组
@@ -63,6 +69,7 @@ def run_ffmpeg(
    Raises:
        subprocess.CalledProcessError: 命令执行失败时抛出，
            异常信息包含完整 stderr 以便排查。
+        subprocess.TimeoutExpired: 超时未完成时抛出，FFmpeg 进程会被 kill。
    """
    try:
        result = subprocess.run(  # nosec B603
@@ -71,8 +78,16 @@ def run_ffmpeg(
            stdout=subprocess.PIPE if capture_output else None,
            stderr=subprocess.PIPE if capture_output else None,
            text=True,
+            timeout=timeout,
        )
        return (result.stdout or "", result.stderr or "")
+    except subprocess.TimeoutExpired as e:
+        logger.error(
+            "FFmpeg 命令超时 (%ds): command=%s",
+            timeout or -1,
+            " ".join(str(c) for c in command[:20]),
+        )
+        raise
    except subprocess.CalledProcessError as e:
        # 把完整 stderr 打到日志，方便排查 exit code 183 等问题
        stderr_text = (e.stderr or "").strip()
@@ -174,6 +189,7 @@ def probe_video_info(video_path: str) -> dict[str, Any]:
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
+            timeout=15,
        )

        import json
@@ -0,0 +1,93 @@
+"""FFmpeg 超时保护测试。
+
+验证 run_ffmpeg / probe_video_info 的超时保护机制，
+防止 FFmpeg hang 住导致 worker 永久阻塞。
+"""
+
+from __future__ import annotations
+
+import subprocess
+from unittest.mock import MagicMock, patch
+
+import pytest
+from video_processing.ffmpeg_utils import (
+    DEFAULT_FFMPEG_TIMEOUT,
+    probe_video_info,
+    run_ffmpeg,
+)
+
+# ── run_ffmpeg 超时保护 ──────────────────────────────────────────────────────
+
+
+class TestRunFFmpegTimeout:
+    """run_ffmpeg 超时保护测试。"""
+
+    def test_default_timeout_is_set(self):
+        """默认超时应为 1800 秒（30分钟）。"""
+        assert DEFAULT_FFMPEG_TIMEOUT == 1800
+
+    def test_timeout_expired_is_raised(self):
+        """超时未完成时 TimeoutExpired 异常被传播。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg", "test"], timeout=1)
+            with pytest.raises(subprocess.TimeoutExpired):
+                run_ffmpeg(["ffmpeg", "test"])
+
+    def test_custom_timeout(self):
+        """支持自定义超时时间。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=5)
+            with pytest.raises(subprocess.TimeoutExpired):
+                run_ffmpeg(["ffmpeg", "test"], timeout=5)
+
+    def test_none_timeout_disables_protection(self):
+        """timeout=None 可以禁用超时保护（不推荐）。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_result = MagicMock()
+            mock_result.stdout = ""
+            mock_result.stderr = ""
+            mock_run.return_value = mock_result
+            run_ffmpeg(["ffmpeg", "test"], timeout=None)
+            # 验证 timeout=None 被传递
+            call_kwargs = mock_run.call_args.kwargs
+            assert call_kwargs["timeout"] is None
+
+    def test_called_process_error_still_raised(self):
+        """超时异常不影响原有 CalledProcessError 的抛出。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.CalledProcessError(returncode=1, cmd=["ffmpeg"], stderr="error msg")
+            with pytest.raises(subprocess.CalledProcessError):
+                run_ffmpeg(["ffmpeg", "test"])
+
+
+# ── probe_video_info 超时保护 ────────────────────────────────────────────────
+
+
+class TestProbeVideoInfoTimeout:
+    """probe_video_info 超时保护测试。"""
+
+    def test_probe_uses_timeout(self):
+        """probe_video_info 调用 ffprobe 时应设置 timeout=15。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=15)
+            # 超时异常被捕获，返回默认值
+            result = probe_video_info("/tmp/test.mp4")
+            assert result["width"] == 1280  # DEFAULT_OUTPUT_WIDTH
+            assert result["height"] == 720  # DEFAULT_OUTPUT_HEIGHT
+
+    def test_probe_success(self):
+        """正常情况应解析 ffprobe JSON 输出。"""
+        fake_output = """
+        {
+            "streams": [{"width": 1920, "height": 1080, "r_frame_rate": "30/1", "duration": "10.5"}],
+            "format": {"duration": "10.5"}
+        }
+        """
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_result = MagicMock()
+            mock_result.stdout = fake_output
+            mock_run.return_value = mock_result
+            result = probe_video_info("/tmp/test.mp4")
+            assert result["width"] == 1920
+            assert result["height"] == 1080
+            assert abs(result["duration"] - 10.5) < 0.01
@@ -0,0 +1,190 @@
+"""渲染结果内部下载接口单元测试。
+
+测试 internal_render 路由的核心逻辑，mock 掉 repository 和 storage 依赖。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from app.api.routes.internal_render import (
+    InternalRenderDownloadUrlResponse,
+    InternalRenderTaskVideosResponse,
+    _video_to_item,
+    get_render_task_videos,
+    get_render_video_download_url,
+)
+
+# ── Helpers ────────────────────────────────────────────────────────────────
+
+
+class MockVideo:
+    """模拟 GeneratedVideo 领域对象。"""
+
+    def __init__(self, **kwargs):
+        self.id = kwargs.get("id", "video-1")
+        self.generation_task_id = kwargs.get("generation_task_id", "task-1")
+        self.project_id = kwargs.get("project_id", "proj-1")
+        self.name = kwargs.get("name", "test_video.mp4")
+        self.file_url = kwargs.get("file_url", "videos/test/output.mp4")
+        self.file_size = kwargs.get("file_size", 1024000)
+        self.duration = kwargs.get("duration", 30.5)
+        self.width = kwargs.get("width", 1080)
+        self.height = kwargs.get("height", 1920)
+        self.fps = kwargs.get("fps", 30.0)
+        self.status = kwargs.get("status", "completed")
+
+
+# ── _video_to_item 测试 ────────────────────────────────────────────────────
+
+
+class TestVideoToItem:
+    """测试视频对象转响应项。"""
+
+    def test_basic_conversion(self):
+        video = MockVideo(id="v1", generation_task_id="t1", status="completed")
+        item = _video_to_item(video, "https://oss.example.com/download?v1")
+        assert item.video_id == "v1"
+        assert item.generation_task_id == "t1"
+        assert item.status == "completed"
+        assert item.download_url == "https://oss.example.com/download?v1"
+
+    def test_missing_optional_fields(self):
+        """缺可选字段时返回 None。"""
+        video = MockVideo()
+        # 去掉可选字段
+        del video.file_size
+        del video.duration
+        item = _video_to_item(video, "https://example.com/dl")
+        assert item.file_size is None
+        assert item.duration is None
+        assert item.width == 1080  # 还在
+
+
+# ── 路由函数测试 ────────────────────────────────────────────────────────────
+
+
+class TestGetRenderVideoDownloadUrl:
+    """测试单个视频下载URL接口。"""
+
+    def test_video_exists(self):
+        video = MockVideo(id="v-abc", file_url="videos/abc/out.mp4")
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = video
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://oss.test/signed?v=abc"
+
+        result = get_render_video_download_url(
+            video_id="v-abc",
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        assert isinstance(result, InternalRenderDownloadUrlResponse)
+        assert result.video_id == "v-abc"
+        assert result.download_url == "https://oss.test/signed?v=abc"
+        mock_repo.get.assert_called_once_with("v-abc")
+        mock_storage.get_download_url.assert_called_once()
+
+    def test_video_not_found_raises_404(self):
+        from fastapi import HTTPException
+
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+        mock_storage = MagicMock()
+
+        with pytest.raises(HTTPException) as exc_info:
+            get_render_video_download_url(
+                video_id="nonexistent",
+                _=True,
+                generated_video_repository=mock_repo,
+                storage_service=mock_storage,
+            )
+        assert exc_info.value.status_code == 404
+
+    def test_download_url_long_expiry(self):
+        """过期时间应为 24 小时（86400s）。"""
+        video = MockVideo(id="v1")
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = video
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://oss.test/signed"
+
+        get_render_video_download_url(
+            video_id="v1",
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        # 验证 expires_seconds=86400
+        call_kwargs = mock_storage.get_download_url.call_args
+        assert call_kwargs.kwargs.get("expires_seconds") == 86400 or call_kwargs[1].get("expires_seconds") == 86400
+
+
+class TestGetRenderTaskVideos:
+    """测试任务视频列表接口。"""
+
+    def test_list_multiple_videos(self):
+        videos = [
+            MockVideo(id="v1", status="completed"),
+            MockVideo(id="v2", status="completed"),
+            MockVideo(id="v3", status="failed"),
+        ]
+        mock_repo = MagicMock()
+        mock_repo.list_by_generation_task.return_value = videos
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://oss.test/signed"
+
+        result = get_render_task_videos(
+            task_id="task-1",
+            status=None,
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        assert isinstance(result, InternalRenderTaskVideosResponse)
+        assert result.task_id == "task-1"
+        assert result.count == 3
+        assert len(result.videos) == 3
+
+    def test_filter_by_status(self):
+        videos = [
+            MockVideo(id="v1", status="completed"),
+            MockVideo(id="v2", status="completed"),
+            MockVideo(id="v3", status="failed"),
+        ]
+        mock_repo = MagicMock()
+        mock_repo.list_by_generation_task.return_value = videos
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://oss.test/signed"
+
+        result = get_render_task_videos(
+            task_id="task-1",
+            status="completed",
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        assert result.count == 2
+        assert all(v.status == "completed" for v in result.videos)
+
+    def test_empty_task(self):
+        mock_repo = MagicMock()
+        mock_repo.list_by_generation_task.return_value = []
+        mock_storage = MagicMock()
+
+        result = get_render_task_videos(
+            task_id="empty-task",
+            status=None,
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        assert result.count == 0
+        assert result.videos == []
Author	SHA1	Message	Date
CI Bot	aca20d9bf0	fix(worker): FFmpeg 超时保护 - 防止渲染 hang 住导致 worker 永久阻塞 CI/CD Pipeline / Frontend Lint (pull_request) Successful in 3m28s Details CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 6m18s Details CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped Details CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped Details CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped Details CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped Details CI/CD Pipeline / Deploy Production (pull_request) Has been skipped Details CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped Details CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m0s Details 根因：新引擎 UnifiedRenderService 所有 FFmpeg 调用通过 run_ffmpeg 执行，但 subprocess.run 未设置 timeout，FFmpeg hang 住时 worker 线程永久阻塞。旧引擎 compose_video 有单独的 timeout=3600，但新引擎路径没有。修复： 1. run_ffmpeg 新增默认超时 1800s（30分钟），支持自定义传参 2. 捕获 TimeoutExpired 并打 error 日志后重新抛出 3. probe_video_info 新增 timeout=15s 超时保护 4. 7个单元测试覆盖超时逻辑影响范围：所有通过 run_ffmpeg 调用的 FFmpeg 命令（unified_render_service 所有渲染/混音/合并操作）	2026-07-13 11:05:13 +08:00
CI Bot	e78c551ab1	feat(api): 新增渲染结果内部下载接口 - 支持按视频ID/任务ID获取预签名URL - 新增 /internal/render/videos/{video_id}/download-url 接口：单个视频下载URL - 新增 /internal/render/tasks/{task_id}/videos 接口：任务下所有视频+下载URL - 内部 API Key 鉴权（X-API-Key），不对外开放 - 下载URL有效期24小时，方便对比工具批量下载 - 支持按 status 筛选（completed/failed 等） - 8个单元测试覆盖核心逻辑	2026-07-13 11:05:13 +08:00