fix unbounded recordings_info growth for cameras with no cache segments

A record-enabled camera whose record stream produces no cache segments
never appears in grouped_recordings, so the per-camera prune in
RecordingMaintainer.move_files() never runs for it. Its
object_recordings_info and audio_recordings_info buffers then grow
without bound until the recording process is OOM-killed (discussion
#23451).

Run a prune every move_files() cycle for cameras absent from
grouped_recordings, dropping entries older than the longest a segment
could still wait in cache before being matched
(MAX_SEGMENTS_IN_CACHE * MAX_SEGMENT_DURATION * 2). Cameras present in
grouped_recordings are left untouched and keep their existing prune.

Add a regression test asserting that an absent camera's stale entries
are dropped (recent ones kept) while a present camera's entries are
left intact.

frigate/record/maintainer.py

@@ -42,6 +42,8 @@ from frigate.util.services import get_video_properties
 
 logger = logging.getLogger(__name__)
 
+STALE_RECORDINGS_INFO_TTL = MAX_SEGMENTS_IN_CACHE * MAX_SEGMENT_DURATION * 2
+
 
 class SegmentInfo:
     def __init__(
@@ -301,6 +303,8 @@ class RecordingMaintainer(threading.Thread):
                 RecordingsDataTypeEnum.saved.value,
             )
 
+        self._expire_stale_recordings_info(grouped_recordings)
+
         recordings_to_insert: list[Optional[dict[str, Any]]] = await asyncio.gather(
             *tasks
         )
@@ -311,6 +315,21 @@ class RecordingMaintainer(threading.Thread):
             [r for r in recordings_to_insert if r is not None],
         )
 
+    def _expire_stale_recordings_info(
+        self, grouped_recordings: defaultdict[str, list[dict[str, Any]]]
+    ) -> None:
+        expire_before = datetime.datetime.now().timestamp() - STALE_RECORDINGS_INFO_TTL
+        for recordings_info in (
+            self.object_recordings_info,
+            self.audio_recordings_info,
+        ):
+            for camera in list(recordings_info.keys()):
+                if camera in grouped_recordings:
+                    continue
+                info = recordings_info[camera]
+                while info and info[0][0] < expire_before:
+                    info.pop(0)
+
     def drop_segment(self, cache_path: str) -> None:
         Path(cache_path).unlink(missing_ok=True)
         self.end_time_cache.pop(cache_path, None)

frigate/test/test_maintainer.py

@@ -115,6 +115,46 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
         self.assertIsNone(result)
         maintainer.drop_segment.assert_called_once_with(cache_path)
 
+    async def test_expire_stale_recordings_info_drops_only_absent_cameras(self):
+        config = MagicMock(spec=FrigateConfig)
+        config.cameras = {}
+        stop_event = MagicMock()
+        maintainer = RecordingMaintainer(config, stop_event)
+
+        now = datetime.datetime.now().timestamp()
+        ancient = now - 86400
+        recent = now - 1
+
+        maintainer.object_recordings_info["present_cam"] = [(ancient, [], [], [])]
+        maintainer.audio_recordings_info["present_cam"] = [(ancient, 0, [])]
+
+        maintainer.object_recordings_info["absent_cam"] = [
+            (ancient, [], [], []),
+            (recent, [], [], []),
+        ]
+        maintainer.audio_recordings_info["absent_cam"] = [
+            (ancient, 0, []),
+            (recent, 0, []),
+        ]
+
+        grouped_recordings = {"present_cam": [{"start_time": ancient}]}
+
+        maintainer._expire_stale_recordings_info(grouped_recordings)
+
+        self.assertEqual(
+            maintainer.object_recordings_info["present_cam"], [(ancient, [], [], [])]
+        )
+        self.assertEqual(
+            maintainer.audio_recordings_info["present_cam"], [(ancient, 0, [])]
+        )
+
+        self.assertEqual(
+            maintainer.object_recordings_info["absent_cam"], [(recent, [], [], [])]
+        )
+        self.assertEqual(
+            maintainer.audio_recordings_info["absent_cam"], [(recent, 0, [])]
+        )
+
 
 if __name__ == "__main__":
     unittest.main()