diff --git a/src/youtube_extension/backend/services/real_video_processor.py b/src/youtube_extension/backend/services/real_video_processor.py index d2d5ba909..d5f1aa698 100644 --- a/src/youtube_extension/backend/services/real_video_processor.py +++ b/src/youtube_extension/backend/services/real_video_processor.py @@ -12,6 +12,8 @@ import json import logging import os +import tempfile +from contextlib import suppress from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional @@ -35,6 +37,9 @@ # Configure logging logger = logging.getLogger(__name__) +# Cached results older than this are treated as misses. +_CACHE_TTL_SECONDS = 86400 # 24 hours + class RealVideoProcessor: """ Complete real video processing service @@ -70,6 +75,60 @@ def _get_cache_path(self, video_id: str) -> Path: """Get cache file path for video""" return self.cache_dir / f"{video_id}_processed.json" + @staticmethod + def _read_cache_file(cache_path: Path) -> Optional[tuple[dict[str, Any], float]]: + """Read and parse a cache entry. + + Blocking: performs ``open``/``fstat``/``json.load``. Always run this off + the event loop. Keeping the whole sequence in a single call also avoids a + stat/read race across separate thread hops. + + The age is taken from ``fstat`` on the already-open descriptor rather than + from a separate ``stat`` on the path, so the timestamp always describes the + exact bytes being parsed even if the path is republished concurrently. + + Returns ``(payload, cache_age_seconds)`` for a fresh entry, else ``None``. + """ + try: + handle = open(cache_path, encoding='utf-8') + except FileNotFoundError: + return None + + with handle as f: + cache_age = datetime.now().timestamp() - os.fstat(f.fileno()).st_mtime + # Deliberately the positive form, mirroring the original + # ``cache_age < 86400`` guard: a non-finite timestamp compares False + # here and falls through to a miss, exactly as it did before. + if cache_age < _CACHE_TTL_SECONDS: + return json.load(f), cache_age + + return None + + @staticmethod + def _write_cache_file(cache_path: Path, payload: dict[str, Any]) -> None: + """Serialize ``payload`` to ``cache_path``. + + Blocking: performs ``open``/``json.dump``/``replace``. Always run off the + event loop. + + Publishes atomically. Serializing straight into ``cache_path`` would + truncate it up front, so a concurrent reader could observe an empty or + half-written entry. Writing to a sibling temp file and ``os.replace``-ing + it into position means readers only ever see a complete entry; the temp + file shares the cache directory so the rename stays on one filesystem. + """ + fd, tmp_name = tempfile.mkstemp( + dir=cache_path.parent, prefix=f'.{cache_path.name}.', suffix='.tmp' + ) + try: + with os.fdopen(fd, 'w', encoding='utf-8') as f: + json.dump(payload, f, indent=2, ensure_ascii=False, default=str) + os.replace(tmp_name, cache_path) + except BaseException: + with suppress(OSError): + os.unlink(tmp_name) + raise + 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: @@ -77,18 +136,16 @@ async def _load_from_cache(self, video_id: str) -> Optional[dict[str, Any]]: try: cache_path = self._get_cache_path(video_id) - if cache_path.exists(): - # Check if cache is recent (less than 24 hours old) - cache_age = datetime.now().timestamp() - cache_path.stat().st_mtime - if cache_age < 86400: # 24 hours - with open(cache_path, encoding='utf-8') as f: - cached_result = json.load(f) + entry = await asyncio.to_thread(self._read_cache_file, cache_path) + if entry is None: + return None - cached_result['cached'] = True - cached_result['cache_age_hours'] = round(cache_age / 3600, 2) + cached_result, cache_age = entry + cached_result['cached'] = True + cached_result['cache_age_hours'] = round(cache_age / 3600, 2) - logger.info(f"📁 Using cached result for {video_id} (age: {cached_result['cache_age_hours']}h)") - return cached_result + logger.info(f"📁 Using cached result for {video_id} (age: {cached_result['cache_age_hours']}h)") + return cached_result except Exception as e: logger.warning(f"Error loading cache: {e}") @@ -106,8 +163,7 @@ async def _save_to_cache(self, video_id: str, result: dict[str, Any]): # Remove cache-specific fields before saving clean_result = {k: v for k, v in result.items() if k not in ['cached', 'cache_age_hours']} - with open(cache_path, 'w', encoding='utf-8') as f: - json.dump(clean_result, f, indent=2, ensure_ascii=False, default=str) + await asyncio.to_thread(self._write_cache_file, cache_path, clean_result) logger.debug(f"💾 Saved result to cache: {cache_path}") diff --git a/tests/unit/test_real_processors.py b/tests/unit/test_real_processors.py index d371a02f9..b01b60dbb 100644 --- a/tests/unit/test_real_processors.py +++ b/tests/unit/test_real_processors.py @@ -1132,6 +1132,197 @@ async def test_load_from_cache_handles_corrupt_file(self, tmp_path): assert result is None +class _ThreadRecordingJSON: + """Proxy around the real ``json`` module that records executing thread ids. + + Used to prove that parse/serialize work is handed to a worker thread rather + than running inline on the event loop thread. + """ + + def __init__(self, real_json): + self._real = real_json + self.load_threads = [] + self.dump_threads = [] + + def load(self, *args, **kwargs): + import threading + self.load_threads.append(threading.get_ident()) + return self._real.load(*args, **kwargs) + + def dump(self, *args, **kwargs): + import threading + self.dump_threads.append(threading.get_ident()) + return self._real.dump(*args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + +class TestCacheDiskIOOffEventLoop: + """Cache disk I/O must not block the event loop thread. + + These assert on *which thread* executes the blocking work (identity, not + wall-clock timing) so they are deterministic under CI load. + """ + + async def test_load_from_cache_parses_off_event_loop(self, tmp_path): + import threading + + proc = _make_video_processor(tmp_path) + cache_path = proc._get_cache_path("auJzb1D-fag") + cache_path.write_text(json.dumps({"video_id": "auJzb1D-fag", "success": True})) + + # Reach the module globals via a method that exists both before and + # after this change; sibling test modules rebind youtube_extension.* + # in sys.modules, so patching a re-imported module object is unreliable. + module_globals = type(proc)._load_from_cache.__globals__ + recorder = _ThreadRecordingJSON(module_globals["json"]) + loop_thread = threading.get_ident() + + with patch.dict(module_globals, {"json": recorder}): + result = await proc._load_from_cache("auJzb1D-fag") + + assert result is not None + assert result["video_id"] == "auJzb1D-fag" + assert recorder.load_threads, "json.load was never invoked" + assert loop_thread not in recorder.load_threads, ( + "cache parse ran on the event loop thread" + ) + + async def test_save_to_cache_serializes_off_event_loop(self, tmp_path): + import threading + + proc = _make_video_processor(tmp_path) + module_globals = type(proc)._save_to_cache.__globals__ + recorder = _ThreadRecordingJSON(module_globals["json"]) + loop_thread = threading.get_ident() + + with patch.dict(module_globals, {"json": recorder}): + await proc._save_to_cache("auJzb1D-fag", {"video_id": "auJzb1D-fag"}) + + assert proc._get_cache_path("auJzb1D-fag").exists() + assert recorder.dump_threads, "json.dump was never invoked" + assert loop_thread not in recorder.dump_threads, ( + "cache serialize ran on the event loop thread" + ) + + async def test_load_from_cache_treats_exact_ttl_as_stale(self, tmp_path): + """Boundary: an entry aged exactly the TTL is a miss, matching prior behaviour.""" + import os + import time + + proc = _make_video_processor(tmp_path) + cache_path = proc._get_cache_path("auJzb1D-fag") + cache_path.write_text(json.dumps({"video_id": "auJzb1D-fag"})) + + boundary = time.time() - 86400 + os.utime(cache_path, (boundary, boundary)) + + assert await proc._load_from_cache("auJzb1D-fag") is None + + async def test_cache_roundtrip_preserves_payload(self, tmp_path): + proc = _make_video_processor(tmp_path) + payload = { + "video_id": "auJzb1D-fag", + "success": True, + "ai_analysis": {"summary": "nested", "topics": ["a", "b"]}, + } + + await proc._save_to_cache("auJzb1D-fag", payload) + loaded = await proc._load_from_cache("auJzb1D-fag") + + assert loaded is not None + assert loaded["ai_analysis"] == payload["ai_analysis"] + assert loaded["cached"] is True + assert loaded["cache_age_hours"] >= 0 + + async def test_load_from_cache_rejects_non_finite_age(self, tmp_path): + """A non-finite age must be a miss, matching the original ``< TTL`` guard. + + ``NaN`` compares False against both ``<`` and ``>=``, so expressing the + staleness check in the negated form would silently serve an entry the + previous implementation discarded. + """ + proc = _make_video_processor(tmp_path) + cache_path = proc._get_cache_path("auJzb1D-fag") + cache_path.write_text(json.dumps({"video_id": "auJzb1D-fag"})) + + module_globals = type(proc)._load_from_cache.__globals__ + real_datetime = module_globals["datetime"] + + class _NaNNow: + @staticmethod + def timestamp(): + return float("nan") + + class _NaNClock: + @staticmethod + def now(*args, **kwargs): + return _NaNNow + + def __getattr__(self, name): + return getattr(real_datetime, name) + + with patch.dict(module_globals, {"datetime": _NaNClock()}): + assert await proc._load_from_cache("auJzb1D-fag") is None + + async def test_save_to_cache_publishes_atomically(self, tmp_path): + """The destination must never be observable in a truncated state. + + Serializing straight into the destination truncates it before the new + bytes land, so a concurrent reader can observe an empty file. This + asserts the write is staged elsewhere and renamed into place. + """ + proc = _make_video_processor(tmp_path) + cache_path = proc._get_cache_path("auJzb1D-fag") + await proc._save_to_cache("auJzb1D-fag", {"video_id": "auJzb1D-fag", "generation": 1}) + + module_globals = type(proc)._save_to_cache.__globals__ + real_json = module_globals["json"] + observed = [] + + class _ObservingJSON: + def dump(self, *args, **kwargs): + # Mid-write: whatever is visible at the destination path must + # still be the previous complete entry. + observed.append(cache_path.read_text()) + return real_json.dump(*args, **kwargs) + + def __getattr__(self, name): + return getattr(real_json, name) + + with patch.dict(module_globals, {"json": _ObservingJSON()}): + await proc._save_to_cache( + "auJzb1D-fag", {"video_id": "auJzb1D-fag", "generation": 2} + ) + + assert observed, "json.dump was never invoked" + assert observed[0].strip(), ( + "destination was truncated before the replacement entry was complete" + ) + assert json.loads(observed[0])["generation"] == 1 + assert json.loads(cache_path.read_text())["generation"] == 2 + + async def test_save_to_cache_leaves_no_temp_file_on_failure(self, tmp_path): + """A failed serialize must not leave a partial temp file in the cache dir.""" + proc = _make_video_processor(tmp_path) + module_globals = type(proc)._save_to_cache.__globals__ + real_json = module_globals["json"] + + class _FailingJSON: + def dump(self, *args, **kwargs): + raise ValueError("serialization boom") + + def __getattr__(self, name): + return getattr(real_json, name) + + with patch.dict(module_globals, {"json": _FailingJSON()}): + await proc._save_to_cache("auJzb1D-fag", {"video_id": "auJzb1D-fag"}) + + assert not proc._get_cache_path("auJzb1D-fag").exists() + assert list(proc.cache_dir.iterdir()) == [], "a temp file was left behind" + + class TestProcessVideo: async def test_returns_cached_result_when_available(self, tmp_path): proc = _make_video_processor(tmp_path)