diff --git a/src/youtube_extension/backend/api/v1/router.py b/src/youtube_extension/backend/api/v1/router.py index 888690061..1db1799b2 100644 --- a/src/youtube_extension/backend/api/v1/router.py +++ b/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,8 +948,10 @@ 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, @@ -930,7 +959,6 @@ async def list_videos_v1( "offset": offset, "has_more": False, } - paginated_videos = data_service.get_videos_summary(limit=limit, offset=offset) return { "videos": paginated_videos, diff --git a/tests/unit/test_v1_router_extended.py b/tests/unit/test_v1_router_extended.py index 49dd3cb28..f2edf3686 100644 --- a/tests/unit/test_v1_router_extended.py +++ b/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}" + )