From fa4002cbe2d962135fafb86cabcbbc17a01b08f8 Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Wed, 19 Aug 2026 08:17:49 -0500 Subject: [PATCH] fix clip download deadlock from unread ffmpeg stderr (#24032) ffmpeg's stderr was piped but never read, so recording segments that generate more than 64 KB of ffmpeg warnings blocked ffmpeg mid-write, stranding the streaming thread and its anyio threadpool token for good. Enough of those and every sync endpoint stops responding until restart. The trigger is how noisy the segments are, not how long the clip is. Send stderr to a temp file instead, and guarantee ffmpeg teardown and playlist cleanup on every exit path, including client disconnect. Also fixes two bugs the deadlock hid: the failure branch was unreachable because returncode is None mid-loop, so the playlist file leaked and ffmpeg's logs were never reported. Playlist files now get a unique name so concurrent requests for one range cannot delete each other's input. Extracts the terminate helper motion search already had into frigate/util/ffmpeg.py, now shared by both streaming call sites. --- frigate/api/media.py | 84 +++++++++---- frigate/jobs/motion_search_decode.py | 22 +--- frigate/test/test_clip_download.py | 171 +++++++++++++++++++++++++++ frigate/util/ffmpeg.py | 19 +++ 4 files changed, 253 insertions(+), 43 deletions(-) create mode 100644 frigate/test/test_clip_download.py diff --git a/frigate/api/media.py b/frigate/api/media.py index aabec09e6d..6d010efd14 100644 --- a/frigate/api/media.py +++ b/frigate/api/media.py @@ -6,11 +6,13 @@ import logging import math import os import subprocess as sp +import tempfile import time +from collections.abc import Iterator from datetime import UTC, datetime, timedelta from enum import Enum from pathlib import Path as FilePath -from typing import Any +from typing import IO, Any from urllib.parse import unquote import cv2 @@ -47,6 +49,7 @@ from frigate.const import ( from frigate.models import Event, Previews, Recordings, Regions, ReviewSegment from frigate.output.preview import get_most_recent_preview_frame from frigate.track.object_processing import TrackedObjectProcessor +from frigate.util.ffmpeg import terminate_ffmpeg_stream from frigate.util.file import ( get_event_snapshot_bytes, get_event_snapshot_path, @@ -70,6 +73,12 @@ logger = logging.getLogger(__name__) # normal hour needs ~360, one clip per recording file NGINX_VOD_MAX_CLIPS = 1080 +# tail of ffmpeg's stderr kept for the clip download failure log +CLIP_STDERR_LOG_BYTES = 8192 + +# how long a drained clip download waits for ffmpeg to exit on its own +CLIP_FFMPEG_EXIT_TIMEOUT = 10 + class VodStreamPreference(str, Enum): """Stream pin for the path-segment VOD route. @@ -465,6 +474,53 @@ async def submit_recording_snapshot_to_plus( ) +def _read_stderr_tail(stderr_file: IO[bytes]) -> str: + """Read back the last CLIP_STDERR_LOG_BYTES of a captured stderr file.""" + stderr_file.seek(0, os.SEEK_END) + stderr_file.seek(max(0, stderr_file.tell() - CLIP_STDERR_LOG_BYTES)) + return stderr_file.read().decode("utf-8", "replace") + + +def _run_clip_download(ffmpeg_cmd: list[str], file_path: str) -> Iterator[bytes]: + """Stream an ffmpeg concat remux to the client, always cleaning up after it.""" + stderr_file = None + ffmpeg = None + + try: + stderr_file = tempfile.TemporaryFile() + ffmpeg = sp.Popen(ffmpeg_cmd, stdout=sp.PIPE, stderr=stderr_file) + + while True: + data = ffmpeg.stdout.read(8192) + + if not data: + break + + yield data + + try: + # wait rather than signal, so the real exit code survives + ffmpeg.wait(timeout=CLIP_FFMPEG_EXIT_TIMEOUT) + except sp.TimeoutExpired: + pass + finally: + if ffmpeg is not None: + # read before terminating: a None here is our teardown, not a failure + exit_code = ffmpeg.poll() + terminate_ffmpeg_stream(ffmpeg) + + if exit_code: + logger.error( + "Failed to generate clip, ffmpeg logs: %s", + _read_stderr_tail(stderr_file), + ) + + if stderr_file is not None: + stderr_file.close() + + FilePath(file_path).unlink(missing_ok=True) + + @router.get( "/{camera_name}/start/{start_ts}/end/{end_ts}/clip.mp4", dependencies=[Depends(require_camera_access)], @@ -476,26 +532,6 @@ async def recording_clip( start_ts: float, end_ts: float, ): - def run_download(ffmpeg_cmd: list[str], file_path: str): - with sp.Popen( - ffmpeg_cmd, - stderr=sp.PIPE, - stdout=sp.PIPE, - text=False, - ) as ffmpeg: - while True: - data = ffmpeg.stdout.read(8192) - if data is not None and len(data) > 0: - yield data - else: - if ffmpeg.returncode and ffmpeg.returncode != 0: - logger.error( - f"Failed to generate clip, ffmpeg logs: {ffmpeg.stderr.read()}" - ) - else: - FilePath(file_path).unlink(missing_ok=True) - break - def get_clip_query(stream_type: str): return ( Recordings.select( @@ -529,7 +565,9 @@ async def recording_clip( status_code=400, ) - file_name = sanitize_filename(f"playlist_{camera_name}_{start_ts}-{end_ts}.txt") + file_name = sanitize_filename( + f"playlist_{camera_name}_{start_ts}-{end_ts}_{os.urandom(4).hex()}.txt" + ) file_path = os.path.join(CACHE_DIR, file_name) with open(file_path, "w") as file: clip: Recordings @@ -577,7 +615,7 @@ async def recording_clip( ] return StreamingResponse( - run_download(ffmpeg_cmd, file_path), + _run_clip_download(ffmpeg_cmd, file_path), media_type="video/mp4", ) diff --git a/frigate/jobs/motion_search_decode.py b/frigate/jobs/motion_search_decode.py index 4b1d518013..3b6e0cf4cc 100644 --- a/frigate/jobs/motion_search_decode.py +++ b/frigate/jobs/motion_search_decode.py @@ -17,6 +17,7 @@ import numpy as np from frigate.config import CameraConfig from frigate.ffmpeg_presets import parse_preset_hardware_acceleration_decode +from frigate.util.ffmpeg import terminate_ffmpeg_stream from frigate.util.services import auto_detect_hwaccel logger = logging.getLogger(__name__) @@ -88,25 +89,6 @@ def _read_exact(stream: IO[bytes], size: int) -> bytes | None: return bytes(buf) -def _terminate(proc: sp.Popen[bytes]) -> None: - """Stop an ffmpeg decode process promptly.""" - # Close the read end first so a blocked ffmpeg write unblocks (ffmpeg then - # sees a broken pipe), then signal it. The resulting ffmpeg write error is - # harmless and goes to the captured stderr. - if proc.stdout is not None: - try: - proc.stdout.close() - except OSError: - pass - if proc.poll() is None: - proc.terminate() - try: - proc.wait(timeout=5) - except sp.TimeoutExpired: - proc.kill() - proc.wait() - - KEYFRAME_MAX_GAP_SECONDS = 2.0 @@ -222,7 +204,7 @@ def _run_vod_decode( count += 1 yield frame finally: - _terminate(proc) + terminate_ffmpeg_stream(proc) stderr_file.close() if count == 0 and software_retry and not should_stop(): diff --git a/frigate/test/test_clip_download.py b/frigate/test/test_clip_download.py new file mode 100644 index 0000000000..4342e85b7b --- /dev/null +++ b/frigate/test/test_clip_download.py @@ -0,0 +1,171 @@ +"""Tests for the recording clip download stream.""" + +import os +import subprocess as sp +import sys +import tempfile +import threading +import unittest +from unittest.mock import patch + +from frigate.api.media import _run_clip_download + +# more than the 64 KB a pipe holds, so an undrained stderr blocks ffmpeg +STDERR_FLOOD_BYTES = 256 * 1024 +PAYLOAD = b"0123456789" * 512 + + +def fake_ffmpeg(*statements: str) -> list[str]: + """Build an argv that stands in for ffmpeg, running the given statements.""" + return [sys.executable, "-c", "\n".join(("import sys, time", *statements))] + + +class TestRunClipDownload(unittest.TestCase): + def setUp(self): + handle, self.playlist_path = tempfile.mkstemp(suffix=".txt") + os.close(handle) + + def tearDown(self): + if os.path.exists(self.playlist_path): + os.unlink(self.playlist_path) + + def collect(self, ffmpeg_cmd: list[str], timeout: float = 30.0) -> bytes: + """Drain the generator on a worker thread so a deadlock fails the test.""" + chunks: list[bytes] = [] + errors: list[BaseException] = [] + + def drain() -> None: + try: + chunks.extend(_run_clip_download(ffmpeg_cmd, self.playlist_path)) + except BaseException as err: + errors.append(err) + + thread = threading.Thread(target=drain, daemon=True) + thread.start() + thread.join(timeout) + + self.assertFalse( + thread.is_alive(), "clip download did not finish, ffmpeg is deadlocked" + ) + + if errors: + raise errors[0] + + return b"".join(chunks) + + def test_streams_full_clip_when_ffmpeg_floods_stderr(self): + """A warning flood past the pipe buffer must not stall the download.""" + data = self.collect( + fake_ffmpeg( + f"sys.stderr.write('w' * {STDERR_FLOOD_BYTES})", + "sys.stderr.flush()", + f"sys.stdout.buffer.write({PAYLOAD!r})", + ) + ) + + self.assertEqual(data, PAYLOAD) + self.assertFalse(os.path.exists(self.playlist_path)) + + def test_streams_clip_written_before_stderr_flood(self): + data = self.collect( + fake_ffmpeg( + f"sys.stdout.buffer.write({PAYLOAD!r})", + "sys.stdout.flush()", + f"sys.stderr.write('w' * {STDERR_FLOOD_BYTES})", + ) + ) + + self.assertEqual(data, PAYLOAD) + + def test_logs_ffmpeg_output_and_removes_playlist_on_failure(self): + with patch("frigate.api.media.logger") as logger: + data = self.collect( + fake_ffmpeg( + "sys.stderr.write('something went wrong')", + "sys.exit(1)", + ) + ) + + self.assertEqual(data, b"") + logger.error.assert_called_once() + self.assertIn("something went wrong", logger.error.call_args.args[1]) + self.assertFalse(os.path.exists(self.playlist_path)) + + def test_logs_only_the_tail_of_a_flooded_stderr(self): + with patch("frigate.api.media.logger") as logger: + self.collect( + fake_ffmpeg( + f"sys.stderr.write('w' * {STDERR_FLOOD_BYTES})", + "sys.exit(1)", + ) + ) + + logged = logger.error.call_args.args[1] + self.assertLess(len(logged), STDERR_FLOOD_BYTES) + + def test_does_not_log_a_successful_download(self): + with patch("frigate.api.media.logger") as logger: + self.collect(fake_ffmpeg(f"sys.stdout.buffer.write({PAYLOAD!r})")) + + logger.error.assert_not_called() + + def test_removes_playlist_when_ffmpeg_cannot_start(self): + with self.assertRaises(OSError): + self.collect(["/nonexistent-ffmpeg-binary"]) + + self.assertFalse(os.path.exists(self.playlist_path)) + + def test_closes_the_stdout_pipe_after_a_successful_download(self): + processes: list[sp.Popen] = [] + real_popen = sp.Popen + + def spy(*args, **kwargs): + process = real_popen(*args, **kwargs) + processes.append(process) + return process + + with patch("subprocess.Popen", spy): + self.collect(fake_ffmpeg(f"sys.stdout.buffer.write({PAYLOAD!r})")) + + self.assertTrue(processes[0].stdout.closed) + + def test_terminating_a_lingering_ffmpeg_is_not_logged_as_a_failure(self): + """A complete download whose ffmpeg overstays is a success, not an error.""" + lingering = fake_ffmpeg( + "import os", + f"os.write(1, {PAYLOAD!r})", + "os.close(1)", + "time.sleep(30)", + ) + + with patch("frigate.api.media.CLIP_FFMPEG_EXIT_TIMEOUT", 0.5): + with patch("frigate.api.media.logger") as logger: + data = self.collect(lingering) + + self.assertEqual(data, PAYLOAD) + logger.error.assert_not_called() + + def test_client_disconnect_kills_ffmpeg_and_removes_playlist(self): + processes: list[sp.Popen] = [] + real_popen = sp.Popen + + def spy(*args, **kwargs): + process = real_popen(*args, **kwargs) + processes.append(process) + return process + + forever = fake_ffmpeg( + "while True:", + " sys.stdout.buffer.write(b'x' * 4096)", + " sys.stdout.flush()", + ) + + with patch("subprocess.Popen", spy): + stream = _run_clip_download(forever, self.playlist_path) + self.assertTrue(next(stream)) + # Starlette never closes the generator itself, so a real disconnect + # reaches this path only once the frame is finalized + stream.close() + + self.assertIsNotNone(processes[0].poll(), "ffmpeg outlived the request") + self.assertFalse(os.path.exists(self.playlist_path)) diff --git a/frigate/util/ffmpeg.py b/frigate/util/ffmpeg.py index 87601b91d6..fed2fd2c3f 100644 --- a/frigate/util/ffmpeg.py +++ b/frigate/util/ffmpeg.py @@ -24,6 +24,25 @@ def stop_ffmpeg(ffmpeg_process: sp.Popen[Any], logger: logging.Logger): ffmpeg_process = None +def terminate_ffmpeg_stream(proc: sp.Popen[Any]) -> None: + """Stop an ffmpeg process whose stdout is being read over a pipe.""" + # Close the read end first so a blocked ffmpeg write unblocks (ffmpeg then + # sees a broken pipe), then signal it. The resulting ffmpeg write error is + # harmless and goes to the captured stderr. + if proc.stdout is not None: + try: + proc.stdout.close() + except OSError: + pass + if proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=5) + except sp.TimeoutExpired: + proc.kill() + proc.wait() + + def start_or_restart_ffmpeg( ffmpeg_cmd, logger, logpipe: LogPipe, frame_size=None, ffmpeg_process=None ) -> sp.Popen[Any]: