Move genai review frame extraction off the embeddings loop (#24959)

* move genai review frame extraction off the embeddings loop

- Run frame gathering for finished review items on a thread so per-frame ffmpeg calls don't stall the embeddings maintainer
- Share one path between live and regenerated review descriptions
- Stop padding review start and end times in place, which widened the stored review item
- Lock timestamp expiry in EventsPerSecond since update() and eps() now run on different threads

* serialize recording frame extraction for review descriptions

frigate/data_processing/post/review_descriptions.py

@@ -68,6 +68,9 @@ class ReviewDescriptionProcessor(PostProcessorApi):
         self.review_desc_dps = EventsPerSecond()
         self.review_desc_dps.start()
 
+        # one review's ffmpeg extraction at a time, as on the maintainer loop
+        self._frame_extraction_lock = threading.Lock()
+
     def calculate_frame_count(
         self,
         camera: str,
@@ -179,53 +182,72 @@ class ReviewDescriptionProcessor(PostProcessorApi):
             ):
                 return
 
-            image_source = camera_config.review.genai.image_source
+            threading.Thread(
-            frame_mode = camera_config.review.genai.frame_mode
+                target=self._describe_finished_review,
+                name=f"review_description_{id}",
+                daemon=True,
+                args=(
+                    camera_config,
+                    final_data,
+                    camera_config.review.genai.image_source,
+                ),
+            ).start()
 
-            if image_source == ImageSourceEnum.recordings:
+    def _describe_finished_review(
-                buffer_extension = get_recording_buffer_extension(
+        self,
-                    final_data["end_time"] - final_data["start_time"]
+        camera_config: CameraConfig,
-                )
+        final_data: dict[str, Any],
-                final_data["start_time"] -= buffer_extension
+        image_source: ImageSourceEnum,
-                final_data["end_time"] += buffer_extension
+        preview_fallback: bool = True,
+        named_objects: list[tuple[str, str]] | None = None,
+    ) -> None:
+        """Gather frames for a finished review item and start its description.
 
+        Runs off the maintainer loop since recording frames shell out to ffmpeg
+        once per frame.
+        """
+        camera = final_data["camera"]
+        id = final_data["id"]
+        genai_config = camera_config.review.genai
+        frames: list[tuple[bytes, float]] = []
+
+        if image_source == ImageSourceEnum.recordings:
+            buffer_extension = get_recording_buffer_extension(
+                final_data["end_time"] - final_data["start_time"]
+            )
+            with self._frame_extraction_lock:
                 frames = self.get_recording_frames(
                     camera,
-                    final_data["start_time"],
+                    final_data["start_time"] - buffer_extension,
-                    final_data["end_time"],
+                    final_data["end_time"] + buffer_extension,
                     height=480,  # Use 480p for good balance between quality and token usage
-                    frame_mode=frame_mode,
+                    frame_mode=genai_config.frame_mode,
                 )
 
-                if not frames:
+            if frames:
-                    # Fallback to preview frames if no recordings available
+                if genai_config.debug_save_thumbnails:
-                    logger.warning(
-                        f"No recording frames found for {camera}, falling back to preview frames"
-                    )
-                    frames = self.get_preview_frames_as_bytes(
-                        camera,
-                        final_data["start_time"],
-                        final_data["end_time"],
-                        final_data["thumb_path"],
-                        id,
-                        camera_config.review.genai.debug_save_thumbnails,
-                        frame_mode,
-                    )
-                elif camera_config.review.genai.debug_save_thumbnails:
                     self.save_debug_recording_frames(id, frames)
+            elif not preview_fallback:
+                logger.error("No recording frames are available for review item %s", id)
+                return
             else:
-                # Use preview frames
+                logger.warning(
-                frames = self.get_preview_frames_as_bytes(
+                    "No recording frames found for %s, falling back to preview frames",
                     camera,
-                    final_data["start_time"],
-                    final_data["end_time"],
-                    final_data["thumb_path"],
-                    id,
-                    camera_config.review.genai.debug_save_thumbnails,
-                    frame_mode,
                 )
 
-            self.start_analysis(camera_config, final_data, frames)
+        if not frames:
+            frames = self.get_preview_frames_as_bytes(
+                camera,
+                final_data["start_time"],
+                final_data["end_time"],
+                final_data["thumb_path"],
+                id,
+                genai_config.debug_save_thumbnails,
+                genai_config.frame_mode,
+            )
+
+        self.start_analysis(camera_config, final_data, frames, named_objects)
 
     def handle_request(self, topic: str, request_data: dict[str, Any]) -> str | None:
         if topic == EmbeddingsRequestEnum.regenerate_review_description.value:
@@ -419,31 +441,12 @@ class ReviewDescriptionProcessor(PostProcessorApi):
             logger.error("Review item %s has not ended yet", review_id)
             return
 
-        buffer_extension = get_recording_buffer_extension(
+        self._describe_finished_review(
-            final_data["end_time"] - final_data["start_time"]
-        )
-        frames = self.get_recording_frames(
-            str(review.camera),
-            final_data["start_time"] - buffer_extension,
-            final_data["end_time"] + buffer_extension,
-            height=480,
-            frame_mode=camera_config.review.genai.frame_mode,
-        )
-
-        if not frames:
-            logger.error(
-                "No recording frames are available for review item %s", review_id
-            )
-            return
-
-        if camera_config.review.genai.debug_save_thumbnails:
-            self.save_debug_recording_frames(review_id, frames)
-
-        self.start_analysis(
             camera_config,
             final_data,
-            frames,
+            ImageSourceEnum.recordings,
-            get_named_objects(
+            preview_fallback=False,
+            named_objects=get_named_objects(
                 final_data["data"].get("detections") or [],
                 self.config.all_attributes,
             ),

frigate/test/test_genai_processor_sync.py

@@ -5,6 +5,7 @@ per-camera gating those processors apply once they exist.
 """
 
 import sys
+import threading
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -19,6 +20,8 @@ for mod in _MOCK_MODULES:
     if mod not in sys.modules:
         sys.modules[mod] = MagicMock()
 
+from frigate.config.camera.review import ImageSourceEnum  # noqa: E402
+
 # imported from the maintainer to avoid tripping the circular import between
 # the maintainer and the processor modules
 from frigate.embeddings.maintainer import (  # noqa: E402
@@ -211,3 +214,58 @@ class TestObjectDescriptionCameraGating(unittest.TestCase):
 
         mock_create_thumbnail.assert_called_once()
         self.assertEqual(len(processor.tracked_events["1234.5-abcdef"]), 1)
+
+
+class TestReviewDescriptionOffLoop(unittest.TestCase):
+    """Frame extraction for a finished review must not block process_data."""
+
+    def test_recording_frames_extracted_off_the_caller_thread(self):
+        camera = MagicMock()
+        camera.review.genai.enabled = True
+        camera.review.genai.alerts = True
+        camera.review.genai.image_source = ImageSourceEnum.recordings
+        camera.review.genai.debug_save_thumbnails = False
+        config = MagicMock()
+        config.cameras = {"front": camera}
+        processor = ReviewDescriptionProcessor(
+            config, MagicMock(), MagicMock(), MagicMock()
+        )
+
+        release = threading.Event()
+        extracted = threading.Event()
+        analyzed = threading.Event()
+
+        def slow_frames(*args, **kwargs):
+            release.wait(5)
+            extracted.set()
+            return [(b"jpg", 1.0)]
+
+        with (
+            patch.object(processor, "get_recording_frames", side_effect=slow_frames),
+            patch.object(
+                processor, "start_analysis", side_effect=lambda *a, **k: analyzed.set()
+            ) as start_analysis,
+        ):
+            processor.process_data(
+                {
+                    "type": "end",
+                    "after": {
+                        "id": "1234.5-abcdef",
+                        "camera": "front",
+                        "severity": "alert",
+                        "start_time": 100.0,
+                        "end_time": 110.0,
+                    },
+                },
+                PostProcessDataEnum.review,
+            )
+
+            self.assertFalse(extracted.is_set())
+            release.set()
+            self.assertTrue(extracted.wait(5))
+
+            self.assertTrue(analyzed.wait(5))
+            start_analysis.assert_called_once()
+            final_data = start_analysis.call_args.args[1]
+            self.assertEqual(final_data["start_time"], 100.0)
+            self.assertEqual(final_data["end_time"], 110.0)

frigate/util/builtin.py

@@ -9,6 +9,7 @@ import queue
 import re
 import shlex
 import struct
+import threading
 import time
 import urllib.parse
 from collections import deque
@@ -42,6 +43,7 @@ class EventsPerSecond:
         self._max_events = max_events
         self._last_n_seconds = last_n_seconds
         self._timestamps: deque[float] = deque(maxlen=max_events)
+        self._lock = threading.Lock()
 
     def start(self) -> None:
         self._start = time.monotonic()
@@ -71,8 +73,11 @@ class EventsPerSecond:
     # remove aged out timestamps
     def expire_timestamps(self, now: float) -> None:
         threshold = now - self._last_n_seconds
-        while self._timestamps and self._timestamps[0] < threshold:
+
-            self._timestamps.popleft()
+        # update() and eps() can run on different threads
+        with self._lock:
+            while self._timestamps and self._timestamps[0] < threshold:
+                self._timestamps.popleft()
 
 
 class InferenceSpeed: