From ffed853f6ea5bccd17ef454518ef721657fc718a Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 13 Jul 2026 10:17:14 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker):=20FFmpeg=20=E8=B6=85=E6=97=B6?= =?UTF-8?q?=E4=BF=9D=E6=8A=A4=20-=20=E9=98=B2=E6=AD=A2=E6=B8=B2=E6=9F=93?= =?UTF-8?q?=20hang=20=E4=BD=8F=E5=AF=BC=E8=87=B4=20worker=20=E6=B0=B8?= =?UTF-8?q?=E4=B9=85=E9=98=BB=E5=A1=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因：新引擎 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 所有渲染/混音/合并操作） --- apps/worker/video_processing/ffmpeg_utils.py | 16 ++++ tests/unit/test_ffmpeg_timeout_protection.py | 93 ++++++++++++++++++++ 2 files changed, 109 insertions(+) create mode 100644 tests/unit/test_ffmpeg_timeout_protection.py diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index 6501dca76..dab6b4b72 100755 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -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 diff --git a/tests/unit/test_ffmpeg_timeout_protection.py b/tests/unit/test_ffmpeg_timeout_protection.py new file mode 100644 index 000000000..04ecbac22 --- /dev/null +++ b/tests/unit/test_ffmpeg_timeout_protection.py @@ -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