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

@@ -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",
     )
 

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():

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))

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]: