From fd5a6b2c5918a5a23ebfb5746c88788babbe4238 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 09:12:54 -0500 Subject: [PATCH 1/2] perf: offload video result cache disk I/O to worker threads RealVideoProcessor._load_from_cache and _save_to_cache ran blocking filesystem and JSON work inline inside async def, stalling the event loop for every concurrently-served request on the hot path of process_video(). Move both to asyncio.to_thread via two static helpers. _read_cache_file collapses the previous exists/stat/open/json.load sequence into a single off-loop hop, which also closes the TOCTOU window where a cache entry could be evicted between the existence check and the read. Observable behaviour is unchanged, including the 24h TTL boundary and the broad except that degrades a corrupt cache entry to a miss. Add TestCacheDiskIOOffEventLoop, which asserts on thread identity rather than wall-clock timing so it stays deterministic under CI load. The two off-loop tests fail against the pre-change implementation. Closes #1227 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/real_video_processor.py | 53 +++++++-- tests/unit/test_real_processors.py | 105 ++++++++++++++++++ 2 files changed, 146 insertions(+), 12 deletions(-) diff --git a/src/youtube_extension/backend/services/real_video_processor.py b/src/youtube_extension/backend/services/real_video_processor.py index d2d5ba909..f315d9226 100644 --- a/src/youtube_extension/backend/services/real_video_processor.py +++ b/src/youtube_extension/backend/services/real_video_processor.py @@ -35,6 +35,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 +73,35 @@ 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 ``exists``/``stat``/``open``/``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. + + Returns ``(payload, cache_age_seconds)`` for a fresh entry, else ``None``. + """ + if not cache_path.exists(): + return None + + cache_age = datetime.now().timestamp() - cache_path.stat().st_mtime + if cache_age >= _CACHE_TTL_SECONDS: + return None + + with open(cache_path, encoding='utf-8') as f: + return json.load(f), cache_age + + @staticmethod + def _write_cache_file(cache_path: Path, payload: dict[str, Any]) -> None: + """Serialize ``payload`` to ``cache_path``. + + Blocking: performs ``open``/``json.dump``. Always run off the event loop. + """ + with open(cache_path, 'w', encoding='utf-8') as f: + json.dump(payload, f, indent=2, ensure_ascii=False, default=str) + 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 +109,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 +136,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..92190cec3 100644 --- a/tests/unit/test_real_processors.py +++ b/tests/unit/test_real_processors.py @@ -1132,6 +1132,111 @@ 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 + + class TestProcessVideo: async def test_returns_cached_result_when_available(self, tmp_path): proc = _make_video_processor(tmp_path) From f4b1ad59106df561abd7dce576374adc69ba104c Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 09:26:58 -0500 Subject: [PATCH 2/2] fix: publish cache file atomically and keep TTL NaN parity Review follow-ups on the cache off-loop change. Atomic publish: _write_cache_file opened the destination with 'w', which truncates it before json.dump completes, so a concurrent _load_from_cache could observe an empty or half-written entry. Serialize into a sibling temp file and os.replace it into position instead, so readers only ever see a complete entry. The temp file shares the cache directory to keep the rename on one filesystem, and is unlinked if serialization fails. TTL parity: the staleness guard had been rewritten as `cache_age >= TTL -> miss`. NaN compares False against both < and >=, so a non-finite mtime that the original `cache_age < TTL` guard treated as a miss would instead have been served as a hit. Restored the positive form so behaviour matches the original for every value, non-finite included. Because os.replace swaps the inode, staleness is now read via os.fstat on the already-open descriptor rather than a separate stat on the path, so the timestamp and the parsed bytes always describe the same inode. Adds three regression tests; all three fail against the previous implementation rather than erroring. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/real_video_processor.py | 51 ++++++++--- tests/unit/test_real_processors.py | 86 +++++++++++++++++++ 2 files changed, 125 insertions(+), 12 deletions(-) diff --git a/src/youtube_extension/backend/services/real_video_processor.py b/src/youtube_extension/backend/services/real_video_processor.py index f315d9226..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 @@ -77,30 +79,55 @@ def _get_cache_path(self, video_id: str) -> Path: def _read_cache_file(cache_path: Path) -> Optional[tuple[dict[str, Any], float]]: """Read and parse a cache entry. - Blocking: performs ``exists``/``stat``/``open``/``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. + 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``. """ - if not cache_path.exists(): + try: + handle = open(cache_path, encoding='utf-8') + except FileNotFoundError: return None - cache_age = datetime.now().timestamp() - cache_path.stat().st_mtime - if cache_age >= _CACHE_TTL_SECONDS: - 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 - with open(cache_path, encoding='utf-8') as f: - 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``. Always run off the event loop. + 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. """ - with open(cache_path, 'w', encoding='utf-8') as f: - json.dump(payload, f, indent=2, ensure_ascii=False, default=str) + 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""" diff --git a/tests/unit/test_real_processors.py b/tests/unit/test_real_processors.py index 92190cec3..b01b60dbb 100644 --- a/tests/unit/test_real_processors.py +++ b/tests/unit/test_real_processors.py @@ -1236,6 +1236,92 @@ async def test_cache_roundtrip_preserves_payload(self, tmp_path): 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):