perf: offload video result cache disk I/O to worker threads

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,25 +75,77 @@ 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:
             return None
 
         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}")
 

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)