don't race the capture thread when stopping ffmpeg (#24884)

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