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