diff --git a/src/youtube_extension/backend/real_api_endpoints.py b/src/youtube_extension/backend/real_api_endpoints.py index 2fd932310..4e8136798 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,63 @@ # 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 +247,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..09b6b50af 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,224 @@ 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 _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 — 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()) + return self._real.exists() + + def glob(self, pattern: str): + self.scan_thread_ids.append(threading.get_ident()) + return [ + _ThreadRecordingPath(p, self.read_thread_ids) + for p in 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}" + ) + + # 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 + ): + """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) == []