perf: offload /api/v1/videos page read off the event loop

src/youtube_extension/backend/api/v1/router.py

@@ -908,6 +908,33 @@ async def clear_all_cache_v1(cache_service: CacheService = Depends(get_cache_ser
 
 
 # Data Endpoints
+def _collect_videos_page(
+    data_service: DataService, limit: int, offset: int
+) -> tuple[int, list[dict[str, Any]], bool]:
+    """Gather the total video count and one page of summaries.
+
+    Both ``count_videos`` and ``get_videos_summary`` perform blocking
+    filesystem work, so they are grouped into this single synchronous helper
+    and dispatched to a worker thread by the caller with one
+    ``asyncio.to_thread`` hop instead of two.
+
+    One hop is *not* an atomic snapshot. ``DataService`` backs both reads with
+    a TTL cache that holds no lock and exposes no snapshot object shared
+    between them, so the entry can still expire -- or be refreshed by another
+    worker -- in between. Grouping the calls narrows that window to a single
+    thread hand-off rather than eliminating it; callers must still treat the
+    count and the page as independently observed values.
+
+    Returns ``(total, page, past_end)``. ``page`` is empty when ``offset`` is
+    past the end, and ``past_end`` reports that condition so the caller does
+    not re-derive the bounds check.
+    """
+    total = data_service.count_videos()
+    if offset >= total:
+        return total, [], True
+    return total, data_service.get_videos_summary(limit=limit, offset=offset), False
+
+
 @router.get(
     "/videos",
     response_model=dict[str, Any],
@@ -921,16 +948,17 @@ async def list_videos_v1(
 ):
     """Get paginated list of processed videos"""
     try:
-        total = data_service.count_videos()
-        if offset >= total:
+        total, paginated_videos, past_end = await asyncio.to_thread(
+            _collect_videos_page, data_service, limit, offset
+        )
+        if past_end:
             return {
                 "videos": [],
                 "total": total,
                 "limit": limit,
                 "offset": offset,
                 "has_more": False,
             }
-        paginated_videos = data_service.get_videos_summary(limit=limit, offset=offset)
 
         return {
             "videos": paginated_videos,

tests/unit/test_v1_router_extended.py

@@ -12,6 +12,7 @@
 
 import asyncio
 import sys
+import threading
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -2398,3 +2399,148 @@ def test_eviction_targets_least_recently_touched(self):
         d["d"] = 4   # overflow -> evict b
         assert "b" not in d
         assert {"a", "c", "d"} <= set(d.keys())
+
+
+# ===========================================================================
+# Regression: /api/v1/videos must not block the event loop (#1379)
+# ===========================================================================
+
+
+class TestListVideosOffloading:
+    """`list_videos_v1` performs unbounded, uncached filesystem work.
+
+    These tests assert *where* that work runs, not merely that the endpoint
+    returns a payload — a status-code assertion passes just as happily when
+    the scan is executed inline on the event loop.
+    """
+
+    @staticmethod
+    def _service(on_count=None, on_summary=None):
+        svc = MagicMock()
+
+        def _count():
+            if on_count is not None:
+                on_count()
+            return 1
+
+        def _summary(limit=None, offset=0):
+            if on_summary is not None:
+                on_summary()
+            return [{"video_id": "vid-1", "title": "Video 1"}]
+
+        svc.count_videos.side_effect = _count
+        svc.get_videos_summary.side_effect = _summary
+        return svc
+
+    def test_filesystem_scan_runs_on_a_worker_thread(self):
+        seen: dict[str, int] = {}
+
+        svc = self._service(
+            on_count=lambda: seen.__setitem__("count", threading.get_ident()),
+            on_summary=lambda: seen.__setitem__("summary", threading.get_ident()),
+        )
+
+        async def _run():
+            seen["loop"] = threading.get_ident()
+            return await router_module.list_videos_v1(
+                limit=10, offset=0, data_service=svc
+            )
+
+        result = asyncio.run(_run())
+
+        # Anti-vacuity: the blocking work really executed and the endpoint
+        # really produced its normal payload. Without these, the thread-identity
+        # assertions below would pass trivially if the calls never happened.
+        assert "count" in seen, "count_videos was never invoked"
+        assert "summary" in seen, "get_videos_summary was never invoked"
+        assert result["total"] == 1
+        assert result["videos"] == [{"video_id": "vid-1", "title": "Video 1"}]
+
+        assert seen["count"] != seen["loop"], (
+            "count_videos ran on the event loop thread; it must be offloaded"
+        )
+        assert seen["summary"] != seen["loop"], (
+            "get_videos_summary ran on the event loop thread; it must be offloaded"
+        )
+
+    def test_event_loop_stays_responsive_while_scan_is_in_flight(self):
+        release = threading.Event()
+        svc = self._service(on_count=lambda: release.wait(timeout=2.0))
+
+        async def _run():
+            ticks = 0
+            task = asyncio.create_task(
+                router_module.list_videos_v1(limit=10, offset=0, data_service=svc)
+            )
+            # While the scan is parked in a worker thread the loop must remain
+            # free to schedule unrelated coroutines.
+            for _ in range(20):
+                if task.done():
+                    break
+                ticks += 1
+                await asyncio.sleep(0.005)
+            release.set()
+            return ticks, await task
+
+        ticks, result = asyncio.run(_run())
+
+        assert result["total"] == 1  # anti-vacuity
+        assert ticks >= 3, (
+            f"event loop only advanced {ticks} time(s) while the scan was "
+            "running; the blocking work is starving the loop"
+        )
+
+    def test_offset_beyond_total_skips_the_page_read(self):
+        """The `offset >= total` short-circuit must survive the refactor."""
+        summary_calls = []
+        svc = self._service(on_summary=lambda: summary_calls.append(1))
+
+        result = asyncio.run(
+            router_module.list_videos_v1(limit=10, offset=100, data_service=svc)
+        )
+
+        assert result["videos"] == []
+        assert result["total"] == 1
+        assert result["has_more"] is False
+        assert summary_calls == [], (
+            "get_videos_summary should not be called when offset >= total"
+        )
+
+    def test_page_read_uses_exactly_one_to_thread_hop(self):
+        """Pin the *number* of hops, not merely that offloading happens.
+
+        The three tests above all pass for a two-hop implementation that awaits
+        ``asyncio.to_thread`` separately for ``count_videos`` and
+        ``get_videos_summary``. That variant costs an extra thread hand-off per
+        request and widens the window in which the underlying TTL cache can be
+        refreshed between the two reads, so the single grouped hop is the
+        behaviour worth protecting.
+
+        The real ``asyncio.to_thread`` is wrapped rather than replaced so the
+        work still runs on a worker thread and the endpoint keeps its normal
+        semantics.
+        """
+        svc = self._service()
+        real_to_thread = router_module.asyncio.to_thread
+        dispatched: list[str] = []
+
+        async def counting_to_thread(func, /, *args, **kwargs):
+            dispatched.append(getattr(func, "__name__", repr(func)))
+            return await real_to_thread(func, *args, **kwargs)
+
+        async def _run():
+            with patch.object(router_module.asyncio, "to_thread", counting_to_thread):
+                return await router_module.list_videos_v1(
+                    limit=50, offset=0, data_service=svc
+                )
+
+        result = asyncio.run(_run())
+
+        # Anti-vacuity: the endpoint really ran and returned its page.
+        assert result["total"] == 1
+        assert result["videos"] == [{"video_id": "vid-1", "title": "Video 1"}]
+
+        assert dispatched == ["_collect_videos_page"], (
+            "expected exactly one asyncio.to_thread hop dispatching "
+            f"_collect_videos_page, got {dispatched}"
+        )