From e18af3c888331a3699233430a9866312b55f3d83 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:15:07 -0500 Subject: [PATCH 1/4] perf: offload cache-directory scan off the event loop (#1231) get_processing_status is an async def, but counted cache entries inline: cached_files = len(list(self.cache_dir.glob("*_processed.json"))) if self.cache_dir.exists() else 0 That is two blocking syscall sequences on the event loop thread - a stat(), then an eager opendir/readdir walk of the whole cache directory. It is awaited by a live HTTP endpoint (real_api_endpoints.py:324), so every status request stalled all concurrently-served requests for the duration of the walk, growing with the number of cached videos. Add a _count_cached_files static helper and await it via asyncio.to_thread, matching the convention established for _read_cache_file/_write_cache_file in this file. The existence check and the walk stay in one hop so the directory cannot disappear between them, and the count is accumulated lazily instead of materializing the full listing. Tests assert the offload from two independent angles - thread identity, and loop responsiveness while the scan is in flight - and neither names asyncio.to_thread, so they stay honest if the mechanism changes. Both carry explicit anti-vacuity guards. Verified: RED 2 failed/3 passed -> GREEN 5/5, 326 passed across all five related test files. Three negative controls all discriminate, including one that keeps the helper but calls it inline, proving the tests assert offloading rather than rewarding the extraction. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/real_video_processor.py | 24 ++++++- tests/unit/test_real_processors.py | 69 +++++++++++++++++++ 2 files changed, 91 insertions(+), 2 deletions(-) diff --git a/src/youtube_extension/backend/services/real_video_processor.py b/src/youtube_extension/backend/services/real_video_processor.py index d5f1aa698..a17e59281 100644 --- a/src/youtube_extension/backend/services/real_video_processor.py +++ b/src/youtube_extension/backend/services/real_video_processor.py @@ -129,6 +129,24 @@ def _write_cache_file(cache_path: Path, payload: dict[str, Any]) -> None: os.unlink(tmp_name) raise + @staticmethod + def _count_cached_files(cache_dir: Path) -> int: + """Count published cache entries under ``cache_dir``. + + Blocking: performs a ``stat`` plus a full directory walk. Always run + this off the event loop. It is reached from the status endpoint, so an + inline walk stalls every concurrently-served request for as long as the + scan takes — which grows with the number of cached videos. + + The existence check and the walk stay in a single call so the directory + cannot disappear between them, mirroring ``_read_cache_file``. The count + is accumulated lazily rather than materializing the whole listing, since + only the total is ever used. + """ + if not cache_dir.exists(): + return 0 + return sum(1 for _ in cache_dir.glob("*_processed.json")) + async def _load_from_cache(self, video_id: str) -> Optional[dict[str, Any]]: """Load processed result from cache if available""" if not self.enable_caching: @@ -475,8 +493,10 @@ async def get_processing_status(self) -> dict[str, Any]: try: cost_dashboard = await cost_monitor.get_cost_dashboard() - # Count cached files - cached_files = len(list(self.cache_dir.glob("*_processed.json"))) if self.cache_dir.exists() else 0 + # Count cached files (off the event loop; the walk is unbounded) + cached_files = await asyncio.to_thread( + self._count_cached_files, self.cache_dir + ) return { 'service_status': 'operational', diff --git a/tests/unit/test_real_processors.py b/tests/unit/test_real_processors.py index b01b60dbb..1cb2224f8 100644 --- a/tests/unit/test_real_processors.py +++ b/tests/unit/test_real_processors.py @@ -12,8 +12,10 @@ from __future__ import annotations +import asyncio import json import sys +import threading from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -1591,6 +1593,73 @@ async def test_error_in_status_returns_error_dict(self, tmp_path): assert status["service_status"] == "error" assert "error" in status + async def test_cache_scan_runs_off_the_event_loop(self, tmp_path): + """The cache-directory scan must not run on the event loop thread. + + ``get_processing_status`` is served by a live HTTP endpoint, so scanning + the cache directory inline stalls every concurrently-served request for + the duration of the walk. + """ + proc = _make_video_processor(tmp_path) + (proc.cache_dir / "abc_processed.json").write_text("{}") + + loop_thread = threading.get_ident() + scan_threads: list[int] = [] + real_glob = Path.glob + + def recording_glob(self, pattern, *args, **kwargs): + scan_threads.append(threading.get_ident()) + return real_glob(self, pattern, *args, **kwargs) + + with patch("youtube_extension.backend.services.real_video_processor.cost_monitor") as cm: + cm.get_cost_dashboard = AsyncMock(return_value={}) + with patch.object(Path, "glob", recording_glob): + status = await proc.get_processing_status() + + # Guards against a vacuous pass: an unscanned directory would satisfy + # the membership assertion trivially. + assert scan_threads, "cache directory was never scanned" + assert loop_thread not in scan_threads + assert status["cache"]["cached_videos"] == 1 + + async def test_cache_scan_does_not_stall_the_event_loop(self, tmp_path): + """The loop keeps scheduling coroutines while the scan is in flight. + + The scan blocks until a coroutine running *on the loop* releases it. If + the scan were inline that coroutine could never be scheduled, so the + gather would exceed its timeout instead of completing. + """ + proc = _make_video_processor(tmp_path) + (proc.cache_dir / "abc_processed.json").write_text("{}") + + scan_started = threading.Event() + may_finish = threading.Event() + real_glob = Path.glob + + def gated_glob(self, pattern, *args, **kwargs): + scan_started.set() + may_finish.wait(timeout=10) + return real_glob(self, pattern, *args, **kwargs) + + async def release_once_scan_starts(): + while not scan_started.is_set(): + await asyncio.sleep(0.01) + may_finish.set() + + with patch("youtube_extension.backend.services.real_video_processor.cost_monitor") as cm: + cm.get_cost_dashboard = AsyncMock(return_value={}) + with patch.object(Path, "glob", gated_glob): + status, _ = await asyncio.wait_for( + asyncio.gather( + proc.get_processing_status(), release_once_scan_starts() + ), + timeout=5, + ) + + assert scan_started.is_set(), "cache directory was never scanned" + assert may_finish.is_set() + assert status["cache"]["cached_videos"] == 1 + class TestClose: async def test_close_calls_youtube_service_close(self, tmp_path): From 38e4801f85e0abdb4793861320de5f70626a198f Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 00:54:28 +0000 Subject: [PATCH 2/4] docs: correct atomicity claim in _count_cached_files docstring The docstring overclaimed that keeping exists() and glob() in one thread hop prevents the cache directory from disappearing between them. That is not atomic. Clarify that the race is benign: glob() on a missing directory yields nothing, so the count degrades to 0 rather than raising. Addresses Copilot review thread on PR #1237. Docstring-only; no behavior change. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01W82PjcoVBdg9mhd4sdLEZ4 --- .../backend/services/real_video_processor.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/youtube_extension/backend/services/real_video_processor.py b/src/youtube_extension/backend/services/real_video_processor.py index a17e59281..77dccc153 100644 --- a/src/youtube_extension/backend/services/real_video_processor.py +++ b/src/youtube_extension/backend/services/real_video_processor.py @@ -138,10 +138,13 @@ def _count_cached_files(cache_dir: Path) -> int: inline walk stalls every concurrently-served request for as long as the scan takes — which grows with the number of cached videos. - The existence check and the walk stay in a single call so the directory - cannot disappear between them, mirroring ``_read_cache_file``. The count - is accumulated lazily rather than materializing the whole listing, since - only the total is ever used. + The existence check and the walk run in a single thread hop rather than + two, mirroring ``_read_cache_file``. This is not atomic: the directory + can still be removed between the ``exists()`` check and the ``glob`` + walk. That race is benign here — ``glob`` on a missing directory yields + nothing, so the count simply degrades to ``0`` instead of raising. The + count is accumulated lazily rather than materializing the whole listing, + since only the total is ever used. """ if not cache_dir.exists(): return 0 From da5cda8262c925f6df556a8aa07194b49c0c2b18 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 01:01:42 +0000 Subject: [PATCH 3/4] refactor: drop redundant exists() and log cache-scan failures Addresses two CodeRabbit findings on _count_cached_files: - Remove the cache_dir.exists() guard. Path.glob already yields no matches for a missing directory, so the check only added a redundant stat() and did not make the walk atomic. - Wrap the walk so a genuine filesystem failure (e.g. a permission error on a directory that does exist) is logged via logger.exception and re-raised, rather than surfacing as service_status: error with no log entry. A missing directory still counts as 0; only real OSErrors propagate. Behavior is unchanged for the existing-directory and missing-directory cases; verified across populated/missing/OSError paths. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01W82PjcoVBdg9mhd4sdLEZ4 --- .../backend/services/real_video_processor.py | 28 ++++++++++--------- 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/src/youtube_extension/backend/services/real_video_processor.py b/src/youtube_extension/backend/services/real_video_processor.py index 77dccc153..5c264b679 100644 --- a/src/youtube_extension/backend/services/real_video_processor.py +++ b/src/youtube_extension/backend/services/real_video_processor.py @@ -133,22 +133,24 @@ def _write_cache_file(cache_path: Path, payload: dict[str, Any]) -> None: def _count_cached_files(cache_dir: Path) -> int: """Count published cache entries under ``cache_dir``. - Blocking: performs a ``stat`` plus a full directory walk. Always run - this off the event loop. It is reached from the status endpoint, so an - inline walk stalls every concurrently-served request for as long as the - scan takes — which grows with the number of cached videos. - - The existence check and the walk run in a single thread hop rather than - two, mirroring ``_read_cache_file``. This is not atomic: the directory - can still be removed between the ``exists()`` check and the ``glob`` - walk. That race is benign here — ``glob`` on a missing directory yields - nothing, so the count simply degrades to ``0`` instead of raising. The + Blocking: performs a full directory walk. Always run this off the event + loop. It is reached from the status endpoint, so an inline walk stalls + every concurrently-served request for as long as the scan takes — which + grows with the number of cached videos. + + ``glob`` already yields nothing for a missing directory, so no separate + existence check is needed (it would only add a redundant ``stat`` and + would not make the walk atomic). A genuine filesystem failure — e.g. a + permission error on a directory that does exist — is logged and + re-raised rather than being silently reported as an empty cache. The count is accumulated lazily rather than materializing the whole listing, since only the total is ever used. """ - if not cache_dir.exists(): - return 0 - return sum(1 for _ in cache_dir.glob("*_processed.json")) + try: + return sum(1 for _ in cache_dir.glob("*_processed.json")) + except OSError: + logger.exception("Failed to count cached files in %s", cache_dir) + raise async def _load_from_cache(self, video_id: str) -> Optional[dict[str, Any]]: """Load processed result from cache if available""" From 059aee451ed0519bfe73e0cc872eb83024e4682a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 01:04:44 +0000 Subject: [PATCH 4/4] test: narrow cache-scan Path.glob patch to the cache directory (#1237) Addresses the remaining CodeRabbit finding on #1237. The source fix (dropping the redundant exists() guard and logging/re-raising OSError) already landed in da5cda8; this scopes the two regression tests' global Path.glob patch to the cache scan itself (self == cache_dir and pattern == "*_processed.json") so unrelated Path.glob calls can no longer trip the thread recorder or the responsiveness gate. Keeps the tests asserting production behavior, not the patch. Negative control (offload removed) still fails the thread-identity test. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_012Hfr2TcNvG7rBRtmrpBEsu --- tests/unit/test_real_processors.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_real_processors.py b/tests/unit/test_real_processors.py index 1cb2224f8..2bb4a5eaa 100644 --- a/tests/unit/test_real_processors.py +++ b/tests/unit/test_real_processors.py @@ -1605,10 +1605,15 @@ async def test_cache_scan_runs_off_the_event_loop(self, tmp_path): loop_thread = threading.get_ident() scan_threads: list[int] = [] + cache_dir = proc.cache_dir real_glob = Path.glob def recording_glob(self, pattern, *args, **kwargs): - scan_threads.append(threading.get_ident()) + # Only record the cache scan itself; a global patch would otherwise + # intercept unrelated Path.glob calls and validate the patch rather + # than production behavior. + if self == cache_dir and pattern == "*_processed.json": + scan_threads.append(threading.get_ident()) return real_glob(self, pattern, *args, **kwargs) with patch("youtube_extension.backend.services.real_video_processor.cost_monitor") as cm: @@ -1634,11 +1639,15 @@ async def test_cache_scan_does_not_stall_the_event_loop(self, tmp_path): scan_started = threading.Event() may_finish = threading.Event() + cache_dir = proc.cache_dir real_glob = Path.glob def gated_glob(self, pattern, *args, **kwargs): - scan_started.set() - may_finish.wait(timeout=10) + # Gate only the cache scan; a global patch would otherwise block on + # unrelated Path.glob calls and make the test assert the patch. + if self == cache_dir and pattern == "*_processed.json": + scan_started.set() + may_finish.wait(timeout=10) return real_glob(self, pattern, *args, **kwargs) async def release_once_scan_starts():