From 82dbcce66df2043ef16567ed037f6e90ee643238 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Tue, 4 Aug 2026 21:05:25 -0500 Subject: [PATCH 1/3] perf: offload the learning-log filesystem walk off the event loop `get_learning_log_v1` awaited nothing: it called `DataService.get_learning_log()` directly, so the whole walk ran inline on the event loop. That call is worse than the sibling endpoints already fixed. It issues its own fresh `rglob("*_enhanced.md")` rather than going through `_get_all_files_cached()`, so it is completely uncached, and it takes no `limit`/`offset`, so its cost grows linearly with the total video count forever. Per entry it also runs a `glob`, an `exists`, an `open` + `json.load` and a `stat`. The walk is now dispatched with a single `asyncio.to_thread` hop. The blocking implementation in `data_service.py` is deliberately untouched; caching and pagination for it are follow-up work. Adds `TestLearningLogOffloading` (4 tests) covering thread identity, loop responsiveness while the walk is in flight, the exact to_thread hop count, and the unchanged 500 error contract. Closes #1386 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/api/v1/router.py | 5 +- tests/unit/test_v1_router_extended.py | 134 ++++++++++++++++++ 2 files changed, 138 insertions(+), 1 deletion(-) diff --git a/src/youtube_extension/backend/api/v1/router.py b/src/youtube_extension/backend/api/v1/router.py index 1db1799b2..ebc2c91df 100644 --- a/src/youtube_extension/backend/api/v1/router.py +++ b/src/youtube_extension/backend/api/v1/router.py @@ -1006,7 +1006,10 @@ async def get_video_detail_v1( async def get_learning_log_v1(data_service: DataService = Depends(get_data_service)): """Get learning log from enhanced analysis files""" try: - learning_log = data_service.get_learning_log() + # ``get_learning_log`` walks the enhanced-analysis tree and opens a + # metadata file per entry. That is unbounded blocking I/O, so it is + # dispatched to a worker thread rather than run on the event loop. + learning_log = await asyncio.to_thread(data_service.get_learning_log) return learning_log except Exception as e: logger.error(f"Error getting learning log: {e}", exc_info=True) diff --git a/tests/unit/test_v1_router_extended.py b/tests/unit/test_v1_router_extended.py index f2edf3686..da15276d8 100644 --- a/tests/unit/test_v1_router_extended.py +++ b/tests/unit/test_v1_router_extended.py @@ -2544,3 +2544,137 @@ async def _run(): "expected exactly one asyncio.to_thread hop dispatching " f"_collect_videos_page, got {dispatched}" ) + + +class TestLearningLogOffloading: + """`get_learning_log_v1` performs an unbounded, uncached filesystem walk. + + `DataService.get_learning_log` issues its own `rglob` on every call — it does + not go through `_get_all_files_cached` — and then opens a metadata file per + entry. These tests assert *where* that work runs. A status-code assertion + passes just as happily when the walk is executed inline on the event loop, + which is exactly why the pre-existing tests for this endpoint stayed green + while production stalled. + """ + + @staticmethod + def _service(on_call=None): + svc = MagicMock() + + def _log(): + if on_call is not None: + on_call() + return [{"video_id": "vid-1", "title": "Video 1"}] + + svc.get_learning_log.side_effect = _log + return svc + + def test_walk_runs_on_a_worker_thread(self): + seen: dict[str, int] = {} + + svc = self._service( + on_call=lambda: seen.__setitem__("walk", threading.get_ident()) + ) + + async def _run(): + seen["loop"] = threading.get_ident() + return await router_module.get_learning_log_v1(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 + # assertion below would pass trivially if the call never happened. + assert "walk" in seen, "get_learning_log was never invoked" + assert result == [{"video_id": "vid-1", "title": "Video 1"}] + + assert seen["walk"] != seen["loop"], ( + "get_learning_log ran on the event loop thread; it must be offloaded" + ) + + def test_event_loop_stays_responsive_while_walk_is_in_flight(self): + """Ticks must complete *while* the walk is still running. + + Counting ticks alone is not enough: a blocking call with a timeout + eventually returns, after which the loop is free and the ticks run + anyway. So each tick is timestamped and compared against the moment the + walk actually finished. If the walk runs inline it pins the loop, and + every tick necessarily lands *after* it -- giving zero qualifying ticks. + """ + import time + + release = threading.Event() + finished_at: dict[str, float] = {} + + def _block(): + release.wait(timeout=5.0) + finished_at["walk"] = time.monotonic() + + svc = self._service(on_call=_block) + + async def _run(): + task = asyncio.create_task( + router_module.get_learning_log_v1(data_service=svc) + ) + tick_times: list[float] = [] + for _ in range(20): + await asyncio.sleep(0.005) + tick_times.append(time.monotonic()) + if len(tick_times) >= 3: + break + release.set() + return tick_times, await task + + tick_times, result = asyncio.run(_run()) + + # Anti-vacuity: the endpoint still returned its real payload, and the + # blocking work really ran to completion. + assert result == [{"video_id": "vid-1", "title": "Video 1"}] + assert "walk" in finished_at, "get_learning_log never completed" + + walk_end = finished_at["walk"] + concurrent = [t for t in tick_times if t < walk_end] + assert len(concurrent) >= 3, ( + "event loop was blocked during the walk: only " + f"{len(concurrent)} of {len(tick_times)} tick(s) completed before " + "the walk finished" + ) + + def test_walk_uses_exactly_one_to_thread_hop(self): + 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.get_learning_log_v1(data_service=svc) + + result = asyncio.run(_run()) + + # Anti-vacuity: the endpoint really ran and returned its payload. + assert result == [{"video_id": "vid-1", "title": "Video 1"}] + + assert len(dispatched) == 1, ( + "expected exactly one asyncio.to_thread hop for the learning-log " + f"walk, got {len(dispatched)}: {dispatched}" + ) + + def test_error_contract_is_unchanged(self): + """A failure inside the worker thread must still surface as a 500.""" + from fastapi import HTTPException as FastAPIHTTPException + + svc = MagicMock() + svc.get_learning_log.side_effect = RuntimeError("scan exploded") + + async def _run(): + return await router_module.get_learning_log_v1(data_service=svc) + + with pytest.raises(FastAPIHTTPException) as exc: + asyncio.run(_run()) + + assert exc.value.status_code == 500 + assert exc.value.detail == "Internal server error" From 0975d79357215e247c9380e28fc554f53df569fa Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Tue, 4 Aug 2026 21:22:38 -0500 Subject: [PATCH 2/3] perf: bound concurrent learning-log walks with a per-loop semaphore Offloading the learning-log walk to a worker thread stopped it pinning the event loop, but it left a second problem in place: the walk is uncached, so every concurrent request starts its own. The endpoint has no endpoint-specific rate limit, and the default per-client limit permits bursts, so a burst could occupy every worker in the shared default executor and starve unrelated asyncio.to_thread callers. Gate the dispatch on a semaphore capped at 4 in-flight walks. Requests over the cap wait on the event loop holding no worker thread, so the endpoint degrades by queueing instead of by monopolising the executor. asyncio.Semaphore pins itself to the first event loop that awaits it and raises RuntimeError if reused from another, so a single module-level instance would break any process that runs more than one loop. The gate is therefore created per running loop and held in a WeakKeyDictionary keyed by that loop, guarded by a threading.Lock because weakref callbacks can run arbitrary Python and the dictionary is not thread-safe. Adds three tests: the cap holds with more callers than the limit, the gate is rebuilt per event loop, and it is shared within one loop. Negative controls confirm each detects a distinct defect - removing the gate yields 7 concurrent walks against a cap of 4, a fresh gate per call fails both the cap and sharing tests, and a single module-level gate fails the per-loop rebuild test. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/api/v1/router.py | 35 +++++- tests/unit/test_v1_router_extended.py | 112 ++++++++++++++++++ 2 files changed, 146 insertions(+), 1 deletion(-) diff --git a/src/youtube_extension/backend/api/v1/router.py b/src/youtube_extension/backend/api/v1/router.py index ebc2c91df..9dbbac8ce 100644 --- a/src/youtube_extension/backend/api/v1/router.py +++ b/src/youtube_extension/backend/api/v1/router.py @@ -10,8 +10,10 @@ import asyncio import logging import os +import threading import time import uuid as _uuid +import weakref from dataclasses import asdict from datetime import datetime, timezone from typing import Any, Optional @@ -997,6 +999,30 @@ async def get_video_detail_v1( raise HTTPException(status_code=500, detail="Internal server error") +# Concurrency gate for the learning-log walk. +# +# ``asyncio.Semaphore`` binds itself to the first event loop that awaits it and +# refuses to be reused from another one, so a single module-level instance would +# break any process that runs more than one loop over its lifetime (every +# ``asyncio.run`` in the test suite, for one). The gate is therefore created +# per running loop and held weakly, so it disappears with the loop it belongs to. +_LEARNING_LOG_MAX_CONCURRENCY = 4 +# Maps a running event loop -> the ``asyncio.Semaphore`` bound to that loop. +_learning_log_gates: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary() +_learning_log_gates_lock = threading.Lock() + + +def _get_learning_log_gate() -> asyncio.Semaphore: + """Return the learning-log concurrency gate bound to the running loop.""" + loop = asyncio.get_running_loop() + with _learning_log_gates_lock: + gate = _learning_log_gates.get(loop) + if gate is None: + gate = asyncio.Semaphore(_LEARNING_LOG_MAX_CONCURRENCY) + _learning_log_gates[loop] = gate + return gate + + @router.get( "/learning-log", response_model=list[dict[str, Any]], @@ -1009,7 +1035,14 @@ async def get_learning_log_v1(data_service: DataService = Depends(get_data_servi # ``get_learning_log`` walks the enhanced-analysis tree and opens a # metadata file per entry. That is unbounded blocking I/O, so it is # dispatched to a worker thread rather than run on the event loop. - learning_log = await asyncio.to_thread(data_service.get_learning_log) + # + # The walk is also uncached, so every concurrent request starts its own. + # Without a bound those walks would be free to occupy every worker in + # the shared default executor and starve unrelated ``to_thread`` + # callers. The gate caps how many walks may be in flight; requests over + # the cap wait here on the event loop, holding no worker thread. + async with _get_learning_log_gate(): + learning_log = await asyncio.to_thread(data_service.get_learning_log) return learning_log except Exception as e: logger.error(f"Error getting learning log: {e}", exc_info=True) diff --git a/tests/unit/test_v1_router_extended.py b/tests/unit/test_v1_router_extended.py index da15276d8..8fadae1bf 100644 --- a/tests/unit/test_v1_router_extended.py +++ b/tests/unit/test_v1_router_extended.py @@ -2678,3 +2678,115 @@ async def _run(): assert exc.value.status_code == 500 assert exc.value.detail == "Internal server error" + + def test_concurrent_walks_are_capped_by_the_gate(self): + """The walk is uncached, so concurrency must be bounded. + + Offloading alone moves the stall off the event loop but lets any burst + of requests occupy every worker in the shared default executor, which + starves unrelated `to_thread` callers. This asserts the cap is real: + more callers than the limit must never produce more simultaneous walks + than the limit. + """ + import time + + limit = router_module._LEARNING_LOG_MAX_CONCURRENCY + callers = limit + 3 + + state = {"in_flight": 0, "peak": 0} + counter_lock = threading.Lock() + release = threading.Event() + + def _occupy(): + with counter_lock: + state["in_flight"] += 1 + state["peak"] = max(state["peak"], state["in_flight"]) + release.wait(timeout=5.0) + with counter_lock: + state["in_flight"] -= 1 + + svc = self._service(on_call=_occupy) + + async def _run(): + tasks = [ + asyncio.create_task(router_module.get_learning_log_v1(data_service=svc)) + for _ in range(callers) + ] + + # Wait for the first wave to reach the worker threads. + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + with counter_lock: + if state["in_flight"] >= limit: + break + await asyncio.sleep(0.01) + + # Give any unbounded overflow a chance to appear before releasing; + # without the gate all `callers` walks would be in flight by now. + await asyncio.sleep(0.25) + with counter_lock: + observed_peak = state["peak"] + + release.set() + return observed_peak, await asyncio.gather(*tasks) + + peak, results = asyncio.run(_run()) + + # Anti-vacuity: every caller really ran and got the payload back. + assert len(results) == callers + assert all(r == [{"video_id": "vid-1", "title": "Video 1"}] for r in results) + assert svc.get_learning_log.call_count == callers + + # The gate held: never more than `limit` walks at once... + assert peak <= limit, ( + f"{peak} concurrent walks observed with a cap of {limit}; the " + "concurrency gate is not bounding the shared executor" + ) + # ...and it did not over-restrict into effective serialisation. + assert peak == limit, ( + f"expected the cap of {limit} to be reached with {callers} " + f"concurrent callers, only saw {peak}" + ) + + def test_gate_is_rebuilt_for_each_event_loop(self): + """A loop-bound `Semaphore` must not leak across event loops. + + `asyncio.Semaphore` pins itself to the first loop that awaits it and + raises `RuntimeError` if reused elsewhere, so a single module-level + instance would break the second `asyncio.run` in any process. + """ + svc = self._service() + + async def _run(): + gate = router_module._get_learning_log_gate() + result = await router_module.get_learning_log_v1(data_service=svc) + return gate, result + + first_gate, first_result = asyncio.run(_run()) + second_gate, second_result = asyncio.run(_run()) + + # Anti-vacuity: both calls actually completed through the gate. + assert first_result == [{"video_id": "vid-1", "title": "Video 1"}] + assert second_result == [{"video_id": "vid-1", "title": "Video 1"}] + + assert first_gate is not second_gate, ( + "the same Semaphore was reused across two event loops; it would " + "raise RuntimeError once the first loop is closed" + ) + + def test_gate_is_shared_within_one_event_loop(self): + """Within a single loop every request must contend for the same gate.""" + svc = self._service() + + async def _run(): + a = router_module._get_learning_log_gate() + await router_module.get_learning_log_v1(data_service=svc) + b = router_module._get_learning_log_gate() + return a, b + + first, second = asyncio.run(_run()) + + assert first is second, ( + "a fresh Semaphore per call would impose no bound at all" + ) + assert svc.get_learning_log.call_count == 1 From 4c2c6674a42b37a73f615b94a864e8ddb1084385 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Tue, 4 Aug 2026 21:36:18 -0500 Subject: [PATCH 3/3] docs: correct semaphore binding semantics, add contention regression test CodeRabbit's second-round review flagged that the comment above `_LEARNING_LOG_MAX_CONCURRENCY` mis-stated when an `asyncio.Semaphore` binds to an event loop. The claim was that it binds on first acquisition. That is wrong, and verifying it changed how the fix should be described. `Semaphore.acquire` in CPython 3.12: if not self.locked(): self._value -= 1 return True # returns before _get_loop() fut = self._get_loop().create_future() # only the waiting path binds The uncontended path never touches the loop. A module-level singleton is therefore not an obvious bug that any test would catch -- it is a latent landmine. It works across any number of loops until the first acquisition that genuinely has to wait; that one pins it, and every later use from a different loop raises `RuntimeError: ... is bound to a different event loop`. The failure cannot surface in low-concurrency tests. It waits for exactly the burst this gate exists to absorb. That makes the per-loop `WeakKeyDictionary` design more justified, not less. Verified from CPython source via `inspect.getsource`, then reproduced: uncontended acquires on two different loops leave `_loop` as `None` with no error; five waiters against a cap of four pin it; a fresh loop then raises. Changes: - Rewrite the comment block to describe contention-triggered binding and the latent-landmine framing, replacing the incorrect first-use claim. - Correct the `test_gate_is_rebuilt_for_each_event_loop` docstring. - Add `test_gate_survives_a_contended_loop_then_a_fresh_loop`, which turns the proxy identity assertion into a proof that the production failure mode is prevented: it saturates the gate with `limit + 2` callers so at least one is a real waiter (the only path that pins the semaphore), then drives the endpoint on a brand-new loop and fails with a descriptive message if `RuntimeError` escapes. Negative control NC-7 (replace the per-loop accessor with a module-level singleton) fails the new test with the exact production `RuntimeError`, confirming it discriminates rather than merely passing. Focused class 8 passed; whole file 133 passed (was 132). `ruff check` findings identical to baseline (26, differing only by a +7 line shift); `ruff format --diff` neutral at 189 lines, same as baseline. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/api/v1/router.py | 17 +++-- tests/unit/test_v1_router_extended.py | 76 ++++++++++++++++++- 2 files changed, 85 insertions(+), 8 deletions(-) diff --git a/src/youtube_extension/backend/api/v1/router.py b/src/youtube_extension/backend/api/v1/router.py index 9dbbac8ce..4ae3f97e5 100644 --- a/src/youtube_extension/backend/api/v1/router.py +++ b/src/youtube_extension/backend/api/v1/router.py @@ -1001,11 +1001,18 @@ async def get_video_detail_v1( # Concurrency gate for the learning-log walk. # -# ``asyncio.Semaphore`` binds itself to the first event loop that awaits it and -# refuses to be reused from another one, so a single module-level instance would -# break any process that runs more than one loop over its lifetime (every -# ``asyncio.run`` in the test suite, for one). The gate is therefore created -# per running loop and held weakly, so it disappears with the loop it belongs to. +# A single module-level ``asyncio.Semaphore`` would be a latent landmine rather +# than an obvious bug. ``Semaphore.acquire`` only reaches ``_get_loop()`` when it +# has to wait -- the uncontended path decrements the counter and returns before +# any loop is touched. So the semaphore stays unbound, and works fine across any +# number of event loops, right up until the first time it is genuinely contended. +# That acquisition pins it, and every later use from a different loop raises +# ``RuntimeError: ... is bound to a different event loop``. +# +# The failure therefore cannot show up in low-concurrency tests; it waits for the +# exact burst this gate exists to absorb. Building the gate per running loop and +# holding it weakly removes the trap outright -- each loop gets its own semaphore, +# which is collected along with the loop it belongs to. _LEARNING_LOG_MAX_CONCURRENCY = 4 # Maps a running event loop -> the ``asyncio.Semaphore`` bound to that loop. _learning_log_gates: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary() diff --git a/tests/unit/test_v1_router_extended.py b/tests/unit/test_v1_router_extended.py index 8fadae1bf..5a8540e8a 100644 --- a/tests/unit/test_v1_router_extended.py +++ b/tests/unit/test_v1_router_extended.py @@ -2751,9 +2751,12 @@ async def _run(): def test_gate_is_rebuilt_for_each_event_loop(self): """A loop-bound `Semaphore` must not leak across event loops. - `asyncio.Semaphore` pins itself to the first loop that awaits it and - raises `RuntimeError` if reused elsewhere, so a single module-level - instance would break the second `asyncio.run` in any process. + `asyncio.Semaphore` does not bind on first use. `acquire` only calls + `_get_loop()` on the path where it must wait, so an uncontended + semaphore stays unbound and crosses loops happily. The first genuinely + contended acquisition pins it, and every later use from another loop + raises `RuntimeError`. A module-level instance would therefore pass + quiet tests and fail only under the burst this gate exists to absorb. """ svc = self._service() @@ -2790,3 +2793,70 @@ async def _run(): "a fresh Semaphore per call would impose no bound at all" ) assert svc.get_learning_log.call_count == 1 + + def test_gate_survives_a_contended_loop_then_a_fresh_loop(self): + """The real failure mode: contention binds a semaphore to its loop. + + `test_gate_is_rebuilt_for_each_event_loop` asserts gate *identity*, + which is a proxy. This asserts the consequence. It saturates the gate + hard enough to force at least one waiter — the only path that reaches + `_LoopBoundMixin._get_loop()` and pins the semaphore — and then drives + the endpoint again on a brand-new loop. A module-level singleton raises + `RuntimeError: ... is bound to a different event loop` here. + """ + import time + + limit = router_module._LEARNING_LOG_MAX_CONCURRENCY + release = threading.Event() + in_flight = 0 + peaked = threading.Event() + counter_lock = threading.Lock() + + def _occupy(): + nonlocal in_flight + with counter_lock: + in_flight += 1 + if in_flight >= limit: + peaked.set() + release.wait(timeout=5.0) + with counter_lock: + in_flight -= 1 + + async def _saturate(): + svc = self._service(on_call=_occupy) + # limit + 2 callers guarantees at least one must *wait* on the gate. + tasks = [ + asyncio.create_task(router_module.get_learning_log_v1(data_service=svc)) + for _ in range(limit + 2) + ] + deadline = time.monotonic() + 5.0 + while not peaked.is_set() and time.monotonic() < deadline: + await asyncio.sleep(0.01) + # Let the surplus callers actually queue on the semaphore. + await asyncio.sleep(0.25) + release.set() + return await asyncio.gather(*tasks) + + saturated = asyncio.run(_saturate()) + + # Anti-vacuity: the contended loop really did serve every caller. + assert peaked.is_set(), "the gate was never saturated; no waiter existed" + assert len(saturated) == limit + 2 + assert all(r == [{"video_id": "vid-1", "title": "Video 1"}] for r in saturated) + + # The actual assertion: a fresh loop must still work. + fresh_svc = self._service() + + async def _after(): + return await router_module.get_learning_log_v1(data_service=fresh_svc) + + try: + result = asyncio.run(_after()) + except RuntimeError as exc: # pragma: no cover - the regression path + raise AssertionError( + "the gate leaked across event loops after being bound by " + f"contention: {exc}" + ) from exc + + assert result == [{"video_id": "vid-1", "title": "Video 1"}] + assert fresh_svc.get_learning_log.call_count == 1