perf: offload the video-detail lookup off the event loop

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

@@ -975,6 +975,67 @@ async def list_videos_v1(
         raise HTTPException(status_code=500, detail="Internal server error")
 
 
+# Shared concurrency gate for the uncached filesystem walks this router hands to
+# ``asyncio.to_thread``.
+#
+# 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
+# removes the trap outright -- each loop gets its own semaphore.
+#
+# The weak keying bounds growth; it is not a guarantee of collection, and the
+# same fast-path asymmetry is why. ``WeakKeyDictionary`` holds its *values*
+# strongly, and the waiting path above stores the loop on the semaphore, so a
+# gate that has ever been contended keeps its own weak key reachable and is
+# never evicted on its own. Uncontended gates still fall out by themselves;
+# contended ones are reclaimed by ``_discard_closed_fs_walk_gates`` below. Under
+# the production deployment -- one Uvicorn worker, one long-lived loop -- this is
+# a single entry either way, so it only matters where loops are created
+# repeatedly, as they are in tests.
+#
+# The budget is deliberately shared by every endpoint that performs one of these
+# walks, rather than one gate per endpoint. The resource being protected is the
+# single default ``ThreadPoolExecutor``, sized ``min(32, cpu_count + 4)`` and so
+# as small as five workers. Two independent gates of four would each be reasoning
+# locally about a global resource and could between them occupy every worker --
+# precisely the starvation a gate exists to prevent. One budget of four always
+# leaves at least one worker for unrelated ``to_thread`` callers.
+_FS_WALK_MAX_CONCURRENCY = 4
+# Maps a running event loop -> the ``asyncio.Semaphore`` bound to that loop.
+_fs_walk_gates: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary()
+_fs_walk_gates_lock = threading.Lock()
+
+
+def _discard_closed_fs_walk_gates() -> None:
+    """Drop registry entries whose event loop has been closed.
+
+    Callers must hold ``_fs_walk_gates_lock``. Deletions are applied only after
+    the comprehension has finished, because a ``WeakKeyDictionary`` must not
+    change size while it is being iterated.
+    """
+    for closed in [loop for loop in _fs_walk_gates if loop.is_closed()]:
+        del _fs_walk_gates[closed]
+
+
+def _get_fs_walk_gate() -> asyncio.Semaphore:
+    """Return the filesystem-walk concurrency gate bound to the running loop."""
+    loop = asyncio.get_running_loop()
+    with _fs_walk_gates_lock:
+        gate = _fs_walk_gates.get(loop)
+        if gate is None:
+            _discard_closed_fs_walk_gates()
+            gate = asyncio.Semaphore(_FS_WALK_MAX_CONCURRENCY)
+            _fs_walk_gates[loop] = gate
+        return gate
+
+
 @router.get(
     "/videos/{video_id}",
     summary="Get Video Details",
@@ -985,7 +1046,16 @@ async def get_video_detail_v1(
 ):
     """Get detailed info for specific video"""
     try:
-        video_detail = data_service.get_video_detail(video_id)
+        # ``get_video_detail`` runs an uncached recursive glob over the
+        # enhanced-analysis tree, stats every match, then opens and reads a
+        # metadata file and the full markdown body. That is blocking I/O whose
+        # cost grows with the corpus, so it is dispatched to a worker thread
+        # instead of being run on the event loop, and it shares the router's
+        # filesystem-walk budget so a burst cannot monopolise the executor.
+        async with _get_fs_walk_gate():
+            video_detail = await asyncio.to_thread(
+                data_service.get_video_detail, video_id
+            )
 
         if not video_detail:
             raise HTTPException(status_code=404, detail=f"Video not found: {video_id}")
@@ -999,37 +1069,6 @@ 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]],
@@ -1048,7 +1087,7 @@ async def get_learning_log_v1(data_service: DataService = Depends(get_data_servi
         # 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():
+        async with _get_fs_walk_gate():
             learning_log = await asyncio.to_thread(data_service.get_learning_log)
         return learning_log
     except Exception as e: