Avoid buffering raw frames during watchdog resets (#24663)

* Don't buffer detect ffmpeg stdout when resetting the capture thread

reset_capture_thread waited for the terminated detect process with
communicate() on the no-frames and stale sub stream paths. communicate()
reads the process's raw frames into memory until it exits, and a stalled
detect process can write hundreds of duplicate frames per second before
it does. Use wait() on every path, as the fps-exceeded path has since
#19110; the timeout and kill still bound the wait.

* Update frigate/video/ffmpeg.py

accurately explain why wait() is safe

Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>

* test(watchdog): remove non-regression child process test

The Python child exits on SIGTERM, so this test also passes with the original communicate() reset. Keep the two mock tests that distinguish the fix and remove the unused imports.

---------

Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>

frigate/test/test_camera_watchdog.py

@@ -1,8 +1,9 @@
 """Tests for per stream recording health tracking in the camera watchdog."""
 
+import subprocess as sp
 import unittest
 from datetime import UTC, datetime, timedelta
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, call, patch
 
 from frigate.config import FrigateConfig
 from frigate.const import STREAM_TYPE_MAIN, STREAM_TYPE_SUB
@@ -62,6 +63,39 @@ def build_watchdog(
     return watchdog
 
 
+class TestCameraWatchdogReset(unittest.TestCase):
+    def _build_resettable_watchdog(self) -> CameraWatchdog:
+        watchdog = build_watchdog()
+        watchdog.ffmpeg_detect_process = MagicMock()
+        watchdog.capture_thread = MagicMock()
+        watchdog.capture_thread.is_alive.return_value = False
+        watchdog.start_ffmpeg_detect = MagicMock()
+        return watchdog
+
+    def test_reset_does_not_buffer_raw_frames(self):
+        watchdog = self._build_resettable_watchdog()
+        process = watchdog.ffmpeg_detect_process
+
+        watchdog.reset_capture_thread()
+
+        process.terminate.assert_called_once()
+        process.wait.assert_called_once_with(timeout=30)
+        process.communicate.assert_not_called()
+        watchdog.start_ffmpeg_detect.assert_called_once()
+
+    def test_timed_out_reset_kills_and_reaps_without_reading_stdout(self):
+        watchdog = self._build_resettable_watchdog()
+        process = watchdog.ffmpeg_detect_process
+        process.wait.side_effect = [sp.TimeoutExpired("ffmpeg", 30), 0]
+
+        watchdog.reset_capture_thread()
+
+        process.kill.assert_called_once()
+        self.assertEqual(process.wait.call_args_list, [call(timeout=30), call()])
+        process.communicate.assert_not_called()
+        watchdog.start_ffmpeg_detect.assert_called_once()
+
+
 class TestCameraWatchdogStreamHealth(unittest.TestCase):
     def _build_watchdog(
         self, sub_enabled: bool = True, output_args: dict | None = None

frigate/video/ffmpeg.py

@@ -310,26 +310,20 @@ class CameraWatchdog(threading.Thread):
         self._check_config_updates()
         return self.config.enabled
 
-    def reset_capture_thread(
-        self, terminate: bool = True, drain_output: bool = True
-    ) -> None:
+    def reset_capture_thread(self, terminate: bool = True) -> None:
         if terminate:
+            # The capture thread keeps draining stdout, so wait() is enough here.
+            # communicate() would also buffer every frame ffmpeg writes before it
+            # exits, which can be many duplicates after a stall.
             self.ffmpeg_detect_process.terminate()
             try:
                 self.logger.info("Waiting for ffmpeg to exit gracefully...")
-
-                if drain_output:
-                    self.ffmpeg_detect_process.communicate(timeout=30)
-                else:
-                    self.ffmpeg_detect_process.wait(timeout=30)
+                self.ffmpeg_detect_process.wait(timeout=30)
             except sp.TimeoutExpired:
                 self.logger.info("FFmpeg did not exit. Force killing...")
                 self.ffmpeg_detect_process.kill()
 
-                if drain_output:
-                    self.ffmpeg_detect_process.communicate()
-                else:
-                    self.ffmpeg_detect_process.wait()
+                self.ffmpeg_detect_process.wait()
 
         # Update reconnects
         now = datetime.now().timestamp()
@@ -497,7 +491,7 @@ class CameraWatchdog(threading.Thread):
                         f"{self.config.name} exceeded fps limit. Exiting ffmpeg..."
                     )
                     if can_restart:
-                        self.reset_capture_thread(drain_output=False)
+                        self.reset_capture_thread()
                         last_restart_time = now
             elif now - self.capture_thread.current_frame.value > 20:
                 self.detect_status.send("offline", now)