From 360589a01959912fb9f8e9cd03a9080311ff6b2e Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Tue, 4 Aug 2026 20:15:40 -0500 Subject: [PATCH 1/2] perf: offload /api/v1/videos page read off the event loop list_videos_v1 is an async endpoint but called count_videos() and get_videos_summary() directly, so both ran on the event loop thread. get_videos_summary "Pass 2" is not cached: for every item on the page it runs parent_dir.glob(), Path.exists(), open() and json.load(). At the default limit of 50 that is roughly 200 blocking syscalls plus up to 50 JSON parses per request, all of which stall every other coroutine on the loop for the duration. Group both reads into _collect_videos_page() and dispatch them with a single asyncio.to_thread hop. One hop rather than two keeps the count and the page consistent with each other and avoids opening a second cache-refresh window between the two reads. Adds TestListVideosOffloading, which asserts where the work runs rather than only what it returns: thread identity for both calls, event-loop responsiveness while a scan is in flight, and preservation of the offset >= total short circuit. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/api/v1/router.py | 23 +++- tests/unit/test_v1_router_extended.py | 107 ++++++++++++++++++ 2 files changed, 128 insertions(+), 2 deletions(-) diff --git a/src/youtube_extension/backend/api/v1/router.py b/src/youtube_extension/backend/api/v1/router.py index 888690061..0bf288942 100644 --- a/src/youtube_extension/backend/api/v1/router.py +++ b/src/youtube_extension/backend/api/v1/router.py @@ -908,6 +908,24 @@ 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]]]: + """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. Using one hop rather than two keeps the count + and the page consistent with each other and avoids opening a second + cache-refresh window between the two reads. + """ + total = data_service.count_videos() + if offset >= total: + return total, [] + return total, data_service.get_videos_summary(limit=limit, offset=offset) + + @router.get( "/videos", response_model=dict[str, Any], @@ -921,7 +939,9 @@ async def list_videos_v1( ): """Get paginated list of processed videos""" try: - total = data_service.count_videos() + total, paginated_videos = await asyncio.to_thread( + _collect_videos_page, data_service, limit, offset + ) if offset >= total: return { "videos": [], @@ -930,7 +950,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..830566617 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,109 @@ 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" + ) From e9b738cad3d73f64a041f622dae2780780b15e4d Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Tue, 4 Aug 2026 20:47:56 -0500 Subject: [PATCH 2/2] =?UTF-8?q?perf:=20address=20review=20=E2=80=94=20own?= =?UTF-8?q?=20the=20bounds=20check=20in=20the=20helper,=20pin=20the=20hop?= =?UTF-8?q?=20count?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to 360589a0 addressing CodeRabbit's review of #1382. 1. The helper docstring overclaimed. One `asyncio.to_thread` hop is not an atomic snapshot: `DataService` backs both reads with a TTL cache that holds no lock and shares no snapshot object between them, so the entry can still expire — or be refreshed by another worker — between `count_videos()` and `get_videos_summary()`. Reworded to say the grouped hop *narrows* that window to a single thread hand-off rather than eliminating it. 2. The `offset >= total` predicate was duplicated: once inside `_collect_videos_page` and again in `list_videos_v1`. The helper now returns `(total, page, past_end)` and the endpoint branches on `past_end`, so the bounds check lives in exactly one place. Still one dispatch hop. 3. The three offloading tests did not pin the *number* of hops — they all pass for a two-hop implementation that awaits `to_thread` separately per call. Added `test_page_read_uses_exactly_one_to_thread_hop`, which wraps (rather than replaces) the real `asyncio.to_thread` so the work still runs on a worker thread, and asserts exactly one dispatch of `_collect_videos_page`. Negative control NC-4 confirms the new test is load-bearing and the gap was real: under a two-hop implementation the three original tests still pass and only the new test fails, reporting the two dispatched mocks by name. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/api/v1/router.py | 25 ++++++++---- tests/unit/test_v1_router_extended.py | 39 +++++++++++++++++++ 2 files changed, 56 insertions(+), 8 deletions(-) diff --git a/src/youtube_extension/backend/api/v1/router.py b/src/youtube_extension/backend/api/v1/router.py index 0bf288942..1db1799b2 100644 --- a/src/youtube_extension/backend/api/v1/router.py +++ b/src/youtube_extension/backend/api/v1/router.py @@ -910,20 +910,29 @@ 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]]]: +) -> 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. Using one hop rather than two keeps the count - and the page consistent with each other and avoids opening a second - cache-refresh window between the two reads. + ``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, [] - return total, data_service.get_videos_summary(limit=limit, offset=offset) + return total, [], True + return total, data_service.get_videos_summary(limit=limit, offset=offset), False @router.get( @@ -939,10 +948,10 @@ async def list_videos_v1( ): """Get paginated list of processed videos""" try: - total, paginated_videos = await asyncio.to_thread( + total, paginated_videos, past_end = await asyncio.to_thread( _collect_videos_page, data_service, limit, offset ) - if offset >= total: + if past_end: return { "videos": [], "total": total, diff --git a/tests/unit/test_v1_router_extended.py b/tests/unit/test_v1_router_extended.py index 830566617..f2edf3686 100644 --- a/tests/unit/test_v1_router_extended.py +++ b/tests/unit/test_v1_router_extended.py @@ -2505,3 +2505,42 @@ def test_offset_beyond_total_skips_the_page_read(self): 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}" + )