diff --git a/src/youtube_extension/backend/api/v1/router.py b/src/youtube_extension/backend/api/v1/router.py index 1db1799b2..4ae3f97e5 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,37 @@ async def get_video_detail_v1( raise HTTPException(status_code=500, detail="Internal server error") +# Concurrency gate for the learning-log walk. +# +# 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() +_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]], @@ -1006,7 +1039,17 @@ 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. + # + # 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 f2edf3686..5a8540e8a 100644 --- a/tests/unit/test_v1_router_extended.py +++ b/tests/unit/test_v1_router_extended.py @@ -2544,3 +2544,319 @@ 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" + + 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` 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() + + 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 + + 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