don't race the capture thread when stopping ffmpeg

stop_all_ffmpeg called communicate() on the detect ffmpeg even when the capture thread was still reading stdout, which happened on every ffmpeg, record, and record_sub config restart and when a camera stalled during a stop. The two readers could buffer duplicate frames without bound or hang the stop. When the capture thread is still alive, it now terminates and waits, then kills, so the live thread drains the pipe and ffmpeg can finish its segment. Disable and shutdown still use stop_ffmpeg.

frigate/test/test_camera_watchdog.py

@@ -96,6 +96,52 @@ class TestCameraWatchdogReset(unittest.TestCase):
         watchdog.start_ffmpeg_detect.assert_called_once()
 
 
+class TestCameraWatchdogStop(unittest.TestCase):
+    def _build_stopping_watchdog(self, capture_reading: bool) -> CameraWatchdog:
+        watchdog = build_watchdog()
+        watchdog.ffmpeg_detect_process = MagicMock()
+        watchdog.capture_thread = MagicMock()
+        watchdog.capture_thread.is_alive.return_value = capture_reading
+        return watchdog
+
+    def test_live_capture_thread_terminates_without_reading_stdout(self):
+        watchdog = self._build_stopping_watchdog(capture_reading=True)
+        process = watchdog.ffmpeg_detect_process
+
+        with patch("frigate.video.ffmpeg.stop_ffmpeg") as stop_ffmpeg:
+            watchdog.stop_all_ffmpeg()
+
+        process.terminate.assert_called_once()
+        process.wait.assert_called_once_with(timeout=30)
+        process.kill.assert_not_called()
+        process.communicate.assert_not_called()
+        stop_ffmpeg.assert_not_called()
+        self.assertIsNone(watchdog.ffmpeg_detect_process)
+
+    def test_live_capture_thread_kills_ffmpeg_that_ignores_sigterm(self):
+        watchdog = self._build_stopping_watchdog(capture_reading=True)
+        process = watchdog.ffmpeg_detect_process
+        process.wait.side_effect = [sp.TimeoutExpired("ffmpeg", 30), 0]
+
+        watchdog.stop_all_ffmpeg()
+
+        process.kill.assert_called_once()
+        self.assertEqual(process.wait.call_args_list, [call(timeout=30), call()])
+        process.communicate.assert_not_called()
+
+    def test_joined_capture_thread_uses_stop_ffmpeg(self):
+        watchdog = self._build_stopping_watchdog(capture_reading=False)
+        process = watchdog.ffmpeg_detect_process
+        watchdog.capture_thread.is_alive.side_effect = [True, False]
+
+        with patch("frigate.video.ffmpeg.stop_ffmpeg") as stop_ffmpeg:
+            watchdog.stop_all_ffmpeg()
+
+        stop_ffmpeg.assert_called_once_with(process, watchdog.logger)
+        process.terminate.assert_not_called()
+        self.assertIsNone(watchdog.ffmpeg_detect_process)
+
+
 class TestCameraWatchdogStreamHealth(unittest.TestCase):
     def _build_watchdog(
         self, sub_enabled: bool = True, output_args: dict | None = None

frigate/video/ffmpeg.py

@@ -690,14 +690,26 @@ class CameraWatchdog(threading.Thread):
     def stop_all_ffmpeg(self):
         """Stop all ffmpeg processes (detection and others)."""
         logger.debug(f"Stopping all ffmpeg processes for {self.config.name}")
+        capture_reading = False
         if self.capture_thread is not None and self.capture_thread.is_alive():
             self.capture_thread.join(timeout=5)
             if self.capture_thread.is_alive():
+                capture_reading = True
                 self.logger.warning(
                     f"Capture thread for {self.config.name} did not stop gracefully."
                 )
         if self.ffmpeg_detect_process is not None:
-            stop_ffmpeg(self.ffmpeg_detect_process, self.logger)
+            # A live capture thread drains stdout, so communicate() would race it
+            # for the pipe. Without a reader, ffmpeg needs communicate() to exit.
+            if capture_reading:
+                self.ffmpeg_detect_process.terminate()
+                try:
+                    self.ffmpeg_detect_process.wait(timeout=30)
+                except sp.TimeoutExpired:
+                    self.ffmpeg_detect_process.kill()
+                    self.ffmpeg_detect_process.wait()
+            else:
+                stop_ffmpeg(self.ffmpeg_detect_process, self.logger)
             self.ffmpeg_detect_process = None
         for p in self.ffmpeg_other_processes[:]:
             if p["process"] is not None: