From 4945cf1886e4fcd1a7e2e0f37086bce491ec1679 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:06:44 -0500 Subject: [PATCH 1/3] perf: scan processed-video cache off the event loop GET /api/v2/videos/list is declared async but its whole body was blocking filesystem work: a stat, a directory glob, and one open()+json.load() per cached video, with no bound on entry count. The handler never awaited, so the loop was stalled for the full scan and no other request could be served. Extract the scan into a module-level _collect_processed_videos_sync() helper and dispatch it with asyncio.to_thread(), matching the pattern used in #1194, #1196, #1228, #1233, #1240, #1245 and #1251. The scan logic is moved verbatim, so the response payload, newest-first ordering, per-entry corrupt-file skip and empty-list fallbacks are unchanged. Measured on a 2,000-entry cache, peak event-loop stall drops from ~193 ms to ~2 ms. Scan wall time is unchanged: this is a latency and fairness fix, not a throughput one. Closes #1287 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/real_api_endpoints.py | 89 +++++--- tests/unit/test_real_api_endpoints.py | 193 ++++++++++++++++++ 2 files changed, 249 insertions(+), 33 deletions(-) diff --git a/src/youtube_extension/backend/real_api_endpoints.py b/src/youtube_extension/backend/real_api_endpoints.py index 2fd932310..fe411c074 100644 --- a/src/youtube_extension/backend/real_api_endpoints.py +++ b/src/youtube_extension/backend/real_api_endpoints.py @@ -7,10 +7,12 @@ and cost monitoring instead of mock data. """ +import asyncio import json import logging import os from datetime import datetime, timezone +from pathlib import Path from typing import Any, Optional from fastapi import BackgroundTasks, FastAPI, HTTPException @@ -26,6 +28,54 @@ # Configure logging logger = logging.getLogger(__name__) + +def _collect_processed_videos_sync(cache_dir: Path) -> list[dict[str, Any]]: + """Scan the processing cache directory and parse every cached result. + + This performs blocking filesystem work (directory stat, glob, and one + ``open()``/``json.load()`` per cache entry) and is therefore intended to be + executed in a worker thread via :func:`asyncio.to_thread` rather than + directly on the event loop. + + Malformed or unreadable entries are skipped individually so that a single + corrupt file cannot fail the whole listing. + """ + processed_videos: list[dict[str, Any]] = [] + + if not cache_dir.exists(): + return processed_videos + + for cache_file in cache_dir.glob("*_processed.json"): + try: + with open(cache_file, encoding='utf-8') as f: + video_data = json.load(f) + + processed_videos.append({ + "id": video_data.get('video_id'), + "video_url": video_data.get('video_url'), + "title": video_data.get('metadata', {}).get('title', 'Unknown'), + "channel": video_data.get('metadata', {}).get('channel_title', 'Unknown'), + "duration": video_data.get('metadata', {}).get('duration', 'Unknown'), + "processed_at": video_data.get('timestamp'), + "has_transcript": video_data.get('transcript', {}).get('has_transcript', False), + "ai_analysis_success": video_data.get('ai_analysis', {}).get('success', False), + "total_cost": video_data.get('cost_breakdown', {}).get('total_cost', 0.0), + "analysis": video_data.get('ai_analysis', {}), + "createdAt": video_data.get('timestamp'), + "updatedAt": video_data.get('timestamp') + }) + except Exception as e: + logger.warning(f"Error loading cached video {cache_file}: {e}") + + # Sort by processing timestamp + processed_videos.sort( + key=lambda x: x.get('processed_at', ''), + reverse=True + ) + + return processed_videos + + # Pydantic models for API requests/responses class VideoProcessingRequest(BaseModel): video_url: str = Field(..., description="YouTube video URL or ID") @@ -188,41 +238,14 @@ async def get_processed_videos_list(): try: processor = get_real_video_processor() - # Get cached processed videos - cache_dir = processor.cache_dir - processed_videos = [] - - if cache_dir.exists(): - for cache_file in cache_dir.glob("*_processed.json"): - try: - with open(cache_file, encoding='utf-8') as f: - video_data = json.load(f) - - processed_videos.append({ - "id": video_data.get('video_id'), - "video_url": video_data.get('video_url'), - "title": video_data.get('metadata', {}).get('title', 'Unknown'), - "channel": video_data.get('metadata', {}).get('channel_title', 'Unknown'), - "duration": video_data.get('metadata', {}).get('duration', 'Unknown'), - "processed_at": video_data.get('timestamp'), - "has_transcript": video_data.get('transcript', {}).get('has_transcript', False), - "ai_analysis_success": video_data.get('ai_analysis', {}).get('success', False), - "total_cost": video_data.get('cost_breakdown', {}).get('total_cost', 0.0), - "analysis": video_data.get('ai_analysis', {}), - "createdAt": video_data.get('timestamp'), - "updatedAt": video_data.get('timestamp') - }) - except Exception as e: - logger.warning(f"Error loading cached video {cache_file}: {e}") - - # Sort by processing timestamp - processed_videos.sort( - key=lambda x: x.get('processed_at', ''), - reverse=True + # The cache scan stats a directory, globs it, and reads/parses one + # JSON file per cached video. That is unbounded blocking I/O which + # would otherwise stall the event loop for every concurrent request, + # so it runs in a worker thread. + return await asyncio.to_thread( + _collect_processed_videos_sync, processor.cache_dir ) - return processed_videos - except Exception as e: logger.error(f"Error getting processed videos list: {e}") return [] diff --git a/tests/unit/test_real_api_endpoints.py b/tests/unit/test_real_api_endpoints.py index ddf59263b..913859a8b 100644 --- a/tests/unit/test_real_api_endpoints.py +++ b/tests/unit/test_real_api_endpoints.py @@ -20,11 +20,16 @@ from __future__ import annotations +import asyncio +import contextlib import json import sys +import threading +import time from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest from fastapi import FastAPI from fastapi.testclient import TestClient @@ -43,6 +48,7 @@ VideoAnalysisResponse, VideoProcessingRequest, VideoValidationRequest, + _collect_processed_videos_sync, init_real_api_services, setup_real_api_endpoints, ) @@ -832,3 +838,190 @@ def test_empty_results_returns_empty_list(self, client, mock_youtube): response = client.post("/api/v2/search-videos?query=unusual+query") assert response.json()["total_results"] == 0 assert response.json()["results"] == [] + + +# =========================================================================== +# GET /api/v2/videos/list - blocking I/O offload (performance regression) +# =========================================================================== + + +class _ThreadRecordingCacheDir: + """Stand-in for ``processor.cache_dir`` that records the scanning thread. + + Delegates to a real :class:`~pathlib.Path` so the endpoint keeps its normal + behaviour, while capturing which thread performed each blocking filesystem + operation. + """ + + def __init__(self, real_dir: Path) -> None: + self._real = real_dir + self.scan_thread_ids: list[int] = [] + + def exists(self) -> bool: + self.scan_thread_ids.append(threading.get_ident()) + return self._real.exists() + + def glob(self, pattern: str): + self.scan_thread_ids.append(threading.get_ident()) + return list(self._real.glob(pattern)) + + +class TestVideosListOffloadsBlockingIO: + """The cache scan must not run on the event loop thread.""" + + def test_cache_scan_runs_off_the_event_loop_thread( + self, api_app, mock_processor, mock_youtube, mock_cost_monitor, tmp_cache + ): + tmp_cache.mkdir(parents=True, exist_ok=True) + _write_cache_file(tmp_cache, "auJzb1D-fag") + + recording_dir = _ThreadRecordingCacheDir(tmp_cache) + mock_processor.cache_dir = recording_dir + + # get_real_video_processor() is invoked by the handler *on the event + # loop thread*, immediately before the scan is dispatched. Recording it + # here gives us the loop's thread id without assuming the test itself + # runs on that loop. + loop_thread_ids: list[int] = [] + + def _record_loop_thread(): + loop_thread_ids.append(threading.get_ident()) + return mock_processor + + with ( + patch( + "youtube_extension.backend.real_api_endpoints.get_real_video_processor", + side_effect=_record_loop_thread, + ), + patch( + "youtube_extension.backend.real_api_endpoints.get_youtube_service", + return_value=mock_youtube, + ), + patch( + "youtube_extension.backend.real_api_endpoints.cost_monitor", + mock_cost_monitor, + ), + ): + with TestClient(api_app, raise_server_exceptions=False) as c: + response = c.get("/api/v2/videos/list") + + assert response.status_code == 200 + assert len(response.json()) == 1 + + assert loop_thread_ids, "handler never resolved the processor" + assert recording_dir.scan_thread_ids, "cache directory was never scanned" + + loop_thread_id = loop_thread_ids[0] + assert all( + tid != loop_thread_id for tid in recording_dir.scan_thread_ids + ), ( + "blocking cache scan ran on the event loop thread " + f"({loop_thread_id}); observed {recording_dir.scan_thread_ids}" + ) + + async def test_event_loop_stays_responsive_during_cache_scan( + self, api_app, mock_processor, mock_youtube, mock_cost_monitor, tmp_cache + ): + """A slow scan must not starve other tasks on the loop.""" + tmp_cache.mkdir(parents=True, exist_ok=True) + + scan_duration = 0.30 + + class _SlowCacheDir: + def exists(self) -> bool: + return True + + def glob(self, pattern: str): + time.sleep(scan_duration) + return [] + + mock_processor.cache_dir = _SlowCacheDir() + + heartbeats = 0 + + async def _heartbeat(): + nonlocal heartbeats + while True: + await asyncio.sleep(0.01) + heartbeats += 1 + + with ( + patch( + "youtube_extension.backend.real_api_endpoints.get_real_video_processor", + return_value=mock_processor, + ), + patch( + "youtube_extension.backend.real_api_endpoints.get_youtube_service", + return_value=mock_youtube, + ), + patch( + "youtube_extension.backend.real_api_endpoints.cost_monitor", + mock_cost_monitor, + ), + ): + transport = httpx.ASGITransport(app=api_app) + async with httpx.AsyncClient( + transport=transport, base_url="http://testserver" + ) as ac: + ticker = asyncio.create_task(_heartbeat()) + try: + response = await ac.get("/api/v2/videos/list") + finally: + ticker.cancel() + with contextlib.suppress(asyncio.CancelledError): + await ticker + + assert response.status_code == 200 + # A responsive loop ticks ~30x during a 0.30s scan. Assert a very + # conservative fraction of that to stay robust on loaded CI runners, + # while still failing outright when the loop is fully blocked. + assert heartbeats >= 5, ( + f"event loop was starved during the cache scan (ticks={heartbeats})" + ) + + def test_offloaded_scan_returns_same_payload(self, client, mock_processor, tmp_cache): + """Offloading must not change the response contract.""" + tmp_cache.mkdir(parents=True, exist_ok=True) + _write_cache_file(tmp_cache, "auJzb1D-fag") + + response = client.get("/api/v2/videos/list") + + assert response.status_code == 200 + payload = response.json() + assert payload == _collect_processed_videos_sync(tmp_cache) + + +class TestCollectProcessedVideosSync: + """Direct coverage of the extracted blocking helper.""" + + def test_missing_directory_returns_empty_list(self, tmp_path): + assert _collect_processed_videos_sync(tmp_path / "absent") == [] + + def test_empty_directory_returns_empty_list(self, tmp_cache): + tmp_cache.mkdir(parents=True, exist_ok=True) + assert _collect_processed_videos_sync(tmp_cache) == [] + + def test_corrupt_entry_is_skipped_without_failing_the_scan(self, tmp_cache): + tmp_cache.mkdir(parents=True, exist_ok=True) + (tmp_cache / "bad_processed.json").write_text("{invalid json", encoding="utf-8") + _write_cache_file(tmp_cache, "auJzb1D-fag") + + result = _collect_processed_videos_sync(tmp_cache) + + assert [v["id"] for v in result] == ["auJzb1D-fag"] + + def test_results_are_sorted_by_timestamp_descending(self, tmp_cache): + tmp_cache.mkdir(parents=True, exist_ok=True) + _write_cache_file(tmp_cache, "vid_a", {"timestamp": "2026-01-01T00:00:00Z"}) + _write_cache_file(tmp_cache, "vid_b", {"timestamp": "2026-06-01T00:00:00Z"}) + + result = _collect_processed_videos_sync(tmp_cache) + + assert [v["id"] for v in result] == ["vid_b", "vid_a"] + + def test_non_matching_files_are_ignored(self, tmp_cache): + tmp_cache.mkdir(parents=True, exist_ok=True) + (tmp_cache / "notes.txt").write_text("ignore me", encoding="utf-8") + (tmp_cache / "other.json").write_text("{}", encoding="utf-8") + + assert _collect_processed_videos_sync(tmp_cache) == [] From bfb60d1d735661c8659a2845ccc6bcc7336f4ace Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 21:18:20 +0000 Subject: [PATCH 2/3] style: Black-format _collect_processed_videos_sync helper Normalize string quotes to double and wrap the dict-append and sort call in _collect_processed_videos_sync to satisfy the 88-char limit, addressing the CodeRabbit review on #1288. Behaviour-preserving: diff is confined to the new helper and the reformat is Black's own AST-equivalent output (verified with --target-version py311). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_013rG7vUAn6z9tXoEuA3dAqz --- .../backend/real_api_endpoints.py | 47 +++++++++++-------- 1 file changed, 28 insertions(+), 19 deletions(-) diff --git a/src/youtube_extension/backend/real_api_endpoints.py b/src/youtube_extension/backend/real_api_endpoints.py index fe411c074..4e8136798 100644 --- a/src/youtube_extension/backend/real_api_endpoints.py +++ b/src/youtube_extension/backend/real_api_endpoints.py @@ -47,31 +47,40 @@ def _collect_processed_videos_sync(cache_dir: Path) -> list[dict[str, Any]]: for cache_file in cache_dir.glob("*_processed.json"): try: - with open(cache_file, encoding='utf-8') as f: + with open(cache_file, encoding="utf-8") as f: video_data = json.load(f) - processed_videos.append({ - "id": video_data.get('video_id'), - "video_url": video_data.get('video_url'), - "title": video_data.get('metadata', {}).get('title', 'Unknown'), - "channel": video_data.get('metadata', {}).get('channel_title', 'Unknown'), - "duration": video_data.get('metadata', {}).get('duration', 'Unknown'), - "processed_at": video_data.get('timestamp'), - "has_transcript": video_data.get('transcript', {}).get('has_transcript', False), - "ai_analysis_success": video_data.get('ai_analysis', {}).get('success', False), - "total_cost": video_data.get('cost_breakdown', {}).get('total_cost', 0.0), - "analysis": video_data.get('ai_analysis', {}), - "createdAt": video_data.get('timestamp'), - "updatedAt": video_data.get('timestamp') - }) + processed_videos.append( + { + "id": video_data.get("video_id"), + "video_url": video_data.get("video_url"), + "title": video_data.get("metadata", {}).get("title", "Unknown"), + "channel": video_data.get("metadata", {}).get( + "channel_title", "Unknown" + ), + "duration": video_data.get("metadata", {}).get( + "duration", "Unknown" + ), + "processed_at": video_data.get("timestamp"), + "has_transcript": video_data.get("transcript", {}).get( + "has_transcript", False + ), + "ai_analysis_success": video_data.get("ai_analysis", {}).get( + "success", False + ), + "total_cost": video_data.get("cost_breakdown", {}).get( + "total_cost", 0.0 + ), + "analysis": video_data.get("ai_analysis", {}), + "createdAt": video_data.get("timestamp"), + "updatedAt": video_data.get("timestamp"), + } + ) except Exception as e: logger.warning(f"Error loading cached video {cache_file}: {e}") # Sort by processing timestamp - processed_videos.sort( - key=lambda x: x.get('processed_at', ''), - reverse=True - ) + processed_videos.sort(key=lambda x: x.get("processed_at", ""), reverse=True) return processed_videos From 11bbda84171707874f5655cdccc249ef8c4463da Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:36:17 -0500 Subject: [PATCH 3/3] test: prove per-file cache read is off the event loop The thread-recording cache directory previously asserted only that exists()/glob() ran off-loop, and relied on the helper extraction to imply the per-entry open()/json.load() moved with them. glob() now yields path-like proxies whose __fspath__ records the calling thread. Because open() resolves a non-str argument through __fspath__, this captures the thread at the exact moment each blocking read starts, so the read is proven off-loop rather than inferred. Verified by reverting only the handler call site to the inline form: the new assertion fails independently with "blocking cache entry read ran on the event loop thread". Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/unit/test_real_api_endpoints.py | 52 ++++++++++++++++++++++----- 1 file changed, 43 insertions(+), 9 deletions(-) diff --git a/tests/unit/test_real_api_endpoints.py b/tests/unit/test_real_api_endpoints.py index 913859a8b..09b6b50af 100644 --- a/tests/unit/test_real_api_endpoints.py +++ b/tests/unit/test_real_api_endpoints.py @@ -845,17 +845,39 @@ def test_empty_results_returns_empty_list(self, client, mock_youtube): # =========================================================================== +class _ThreadRecordingPath: + """Path-like proxy that records the thread performing the per-file read. + + ``open()`` resolves a non-``str`` argument through ``__fspath__``, so this + captures the calling thread at the exact moment the blocking read starts — + rather than inferring it from the enclosing directory scan. + """ + + def __init__(self, real_path: Path, read_thread_ids: list[int]) -> None: + self._real = real_path + self._read_thread_ids = read_thread_ids + + def __fspath__(self) -> str: + self._read_thread_ids.append(threading.get_ident()) + return str(self._real) + + def __str__(self) -> str: + return str(self._real) + + class _ThreadRecordingCacheDir: """Stand-in for ``processor.cache_dir`` that records the scanning thread. Delegates to a real :class:`~pathlib.Path` so the endpoint keeps its normal behaviour, while capturing which thread performed each blocking filesystem - operation. + operation — both the directory-level ``exists()``/``glob()`` and the + per-entry ``open()``. """ def __init__(self, real_dir: Path) -> None: self._real = real_dir self.scan_thread_ids: list[int] = [] + self.read_thread_ids: list[int] = [] def exists(self) -> bool: self.scan_thread_ids.append(threading.get_ident()) @@ -863,7 +885,10 @@ def exists(self) -> bool: def glob(self, pattern: str): self.scan_thread_ids.append(threading.get_ident()) - return list(self._real.glob(pattern)) + return [ + _ThreadRecordingPath(p, self.read_thread_ids) + for p in self._real.glob(pattern) + ] class TestVideosListOffloadsBlockingIO: @@ -912,13 +937,20 @@ def _record_loop_thread(): assert recording_dir.scan_thread_ids, "cache directory was never scanned" loop_thread_id = loop_thread_ids[0] - assert all( - tid != loop_thread_id for tid in recording_dir.scan_thread_ids - ), ( + assert all(tid != loop_thread_id for tid in recording_dir.scan_thread_ids), ( "blocking cache scan ran on the event loop thread " f"({loop_thread_id}); observed {recording_dir.scan_thread_ids}" ) + # The directory scan and the per-entry read are separate blocking + # operations; assert the reads moved off-loop too rather than inferring + # it from the helper extraction. + assert recording_dir.read_thread_ids, "no cache entry was ever read" + assert all(tid != loop_thread_id for tid in recording_dir.read_thread_ids), ( + "blocking cache entry read ran on the event loop thread " + f"({loop_thread_id}); observed {recording_dir.read_thread_ids}" + ) + async def test_event_loop_stays_responsive_during_cache_scan( self, api_app, mock_processor, mock_youtube, mock_cost_monitor, tmp_cache ): @@ -975,11 +1007,13 @@ async def _heartbeat(): # A responsive loop ticks ~30x during a 0.30s scan. Assert a very # conservative fraction of that to stay robust on loaded CI runners, # while still failing outright when the loop is fully blocked. - assert heartbeats >= 5, ( - f"event loop was starved during the cache scan (ticks={heartbeats})" - ) + assert ( + heartbeats >= 5 + ), f"event loop was starved during the cache scan (ticks={heartbeats})" - def test_offloaded_scan_returns_same_payload(self, client, mock_processor, tmp_cache): + def test_offloaded_scan_returns_same_payload( + self, client, mock_processor, tmp_cache + ): """Offloading must not change the response contract.""" tmp_cache.mkdir(parents=True, exist_ok=True) _write_cache_file(tmp_cache, "auJzb1D-fag")