perf(video): run enhanced-result persistence off the event loop

src/youtube_extension/backend/enhanced_video_processor.py

@@ -12,9 +12,11 @@
 """
 
 import asyncio
+import contextlib
 import json
 import logging
 import os
+import threading
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Optional
@@ -29,6 +31,15 @@
 
 logger = logging.getLogger(__name__)
 
+# Serialises the markdown+metadata pair written by
+# ``EnhancedVideoProcessor._write_result_files``. Result filenames carry only
+# one-second precision, so two saves of the same video inside the same second
+# resolve to the same two paths; without this lock their individually-atomic
+# writes can interleave and leave one save's markdown paired with the other's
+# metadata. Held only across two file writes, and only inside a worker thread,
+# so it never blocks the event loop.
+_RESULT_WRITE_LOCK = threading.Lock()
+
 # Optional Gemini Vision integration for frame analysis
 try:
     # Use package import (works with PYTHONPATH=src and when the real MCP server on 8010 is exercised)
@@ -725,6 +736,55 @@ def _coerce_analysis_to_structured_dict(self, text: str) -> dict[str, Any]:
             'format': 'text_coerced'
         }
 
+    @staticmethod
+    def _atomic_write_text(path: Any, contents: str) -> None:
+        """
+        Write ``contents`` to ``path`` so a reader never sees a partial file.
+
+        ``open(path, 'w')`` truncates before it writes, so a concurrent reader
+        -- or a crash mid-write -- can leave a half-written analysis on disk.
+        Write to a sibling temporary file and rename it into place instead;
+        ``os.replace`` is atomic, so the target is either the complete previous
+        file or the complete new one. The temporary name carries the pid and
+        thread id so two writers cannot collide on it.
+        """
+        tmp_path = f"{path}.{os.getpid()}.{threading.get_ident()}.tmp"
+        try:
+            with open(tmp_path, 'w', encoding='utf-8') as f:
+                f.write(contents)
+            os.replace(tmp_path, path)
+        except BaseException:
+            with contextlib.suppress(OSError):
+                os.unlink(tmp_path)
+            raise
+
+    @staticmethod
+    def _write_result_files(
+        save_dir: Any,
+        filepath: Any,
+        metadata_file: Any,
+        markdown: str,
+        metadata: dict,
+    ) -> None:
+        """
+        All blocking filesystem work for :meth:`_save_enhanced_result`.
+
+        Kept in one function so the whole save costs a single thread hop
+        rather than one per syscall. ``json.dumps`` is done here rather than
+        by the caller so the serialisation cost -- which scales with the size
+        of the metadata dict -- is paid off the event loop too.
+
+        The two writes are individually atomic *and* are held together under
+        ``_RESULT_WRITE_LOCK`` so concurrent saves cannot pair one save's
+        markdown with another's metadata.
+        """
+        save_dir.mkdir(parents=True, exist_ok=True)
+        payload = json.dumps(metadata, indent=2, default=str)
+        # Both files must land as a matched pair -- see _RESULT_WRITE_LOCK.
+        with _RESULT_WRITE_LOCK:
+            EnhancedVideoProcessor._atomic_write_text(filepath, markdown)
+            EnhancedVideoProcessor._atomic_write_text(metadata_file, payload)
+
     async def _save_enhanced_result(self, video_id: str, metadata: dict, markdown: str) -> str:
         """Save enhanced results to organized directory structure"""
         try:
@@ -733,19 +793,24 @@ async def _save_enhanced_result(self, video_id: str, metadata: dict, markdown: s
             timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
 
             save_dir = Path('youtube_processed_videos') / 'enhanced_analysis' / category
-            save_dir.mkdir(parents=True, exist_ok=True)
 
             # Save markdown with timestamp
             filename = f"{video_id}_{timestamp}_enhanced.md"
             filepath = save_dir / filename
-
-            with open(filepath, 'w', encoding='utf-8') as f:
-                f.write(markdown)
-
-            # Save metadata
             metadata_file = save_dir / f"{video_id}_{timestamp}_metadata.json"
-            with open(metadata_file, 'w', encoding='utf-8') as f:
-                json.dump(metadata, f, indent=2, default=str)
+
+            # mkdir, both writes and the JSON serialisation are synchronous and
+            # scale with the size of the analysis. Run on this event loop they
+            # stall every other request for the duration of the save, so hand
+            # the whole group to a worker thread in one dispatch.
+            await asyncio.to_thread(
+                self._write_result_files,
+                save_dir,
+                filepath,
+                metadata_file,
+                markdown,
+                metadata,
+            )
 
             logger.info(f"✅ Enhanced results saved to: {filepath}")
             return str(filepath)

tests/unit/test_enhanced_video_processor.py

@@ -7,8 +7,11 @@
 
 from __future__ import annotations
 
+import asyncio
+import contextlib
 import json
 import sys
+import time
 from contextlib import asynccontextmanager
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, mock_open, patch
@@ -1217,3 +1220,217 @@ def test_returns_instance(self):
             with patch.dict(os.environ, {"GEMINI_API_KEY": "factory-key"}, clear=False):
                 proc = get_enhanced_video_processor()
         assert isinstance(proc, EnhancedVideoProcessor)
+
+
+# ===========================================================================
+# Persistence must not stall the event loop, and must not publish partial files
+# ===========================================================================
+
+class TestSaveDoesNotBlockEventLoop:
+    """
+    `_save_enhanced_result` performs mkdir + two file writes + a `json.dumps`.
+    Run directly on the event loop those stall every other in-flight request
+    for the whole duration of the save. These tests pin the two properties the
+    change is claimed to provide: the loop keeps running, and a reader never
+    observes a half-written file.
+    """
+
+    @staticmethod
+    async def _count_heartbeats_during(coro):
+        """Return how many times the loop got to run while ``coro`` was awaited."""
+        ticks = 0
+        stop = False
+
+        async def heartbeat():
+            nonlocal ticks
+            while not stop:
+                ticks += 1
+                await asyncio.sleep(0.005)
+
+        beat = asyncio.create_task(heartbeat())
+        await asyncio.sleep(0)  # let the heartbeat reach its first await first
+        try:
+            result = await coro
+        finally:
+            stop = True
+            beat.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await beat
+        return result, ticks
+
+    async def test_save_does_not_stall_the_event_loop(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        proc = _make_processor()
+
+        real_write = _mod.EnhancedVideoProcessor._atomic_write_text
+
+        def slow_write(path, contents):
+            time.sleep(0.10)
+            real_write(path, contents)
+
+        with patch.object(
+            _mod.EnhancedVideoProcessor, "_atomic_write_text", staticmethod(slow_write)
+        ):
+            result, ticks = await self._count_heartbeats_during(
+                proc._save_enhanced_result(_VIDEO_ID, {"category": "General"}, "# md")
+            )
+
+        assert result != ""
+        # Two 0.10s writes on the loop would yield zero heartbeats.
+        assert ticks > 1, f"event loop was stalled during the save (ticks={ticks})"
+
+    async def test_json_serialisation_also_runs_off_the_loop(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        proc = _make_processor()
+
+        real_dumps = json.dumps
+
+        def slow_dumps(*args, **kwargs):
+            time.sleep(0.10)
+            return real_dumps(*args, **kwargs)
+
+        with patch.object(_mod.json, "dumps", slow_dumps):
+            result, ticks = await self._count_heartbeats_during(
+                proc._save_enhanced_result(_VIDEO_ID, {"category": "General"}, "# md")
+            )
+
+        assert result != ""
+        assert ticks > 1, f"json serialisation stalled the loop (ticks={ticks})"
+
+    async def test_save_writes_readable_markdown_and_metadata(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        proc = _make_processor()
+        metadata = {"category": "Programming", "title": "T"}
+
+        path = await proc._save_enhanced_result(_VIDEO_ID, metadata, "# Heading")
+
+        assert path != ""
+        assert Path(path).read_text(encoding="utf-8") == "# Heading"
+        meta_path = Path(path.replace("_enhanced.md", "_metadata.json"))
+        assert json.loads(meta_path.read_text(encoding="utf-8")) == metadata
+
+    async def test_no_temp_files_remain_after_a_successful_save(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        proc = _make_processor()
+
+        path = await proc._save_enhanced_result(_VIDEO_ID, {"category": "General"}, "# md")
+
+        assert path != ""
+        leftovers = list(Path(path).parent.glob("*.tmp"))
+        assert leftovers == [], f"temporary files left behind: {leftovers}"
+
+    async def test_temp_file_is_cleaned_up_when_the_write_fails(self, tmp_path):
+        target = tmp_path / "out.md"
+
+        with patch.object(_mod.os, "replace", side_effect=OSError("boom")):
+            with pytest.raises(OSError):
+                _mod.EnhancedVideoProcessor._atomic_write_text(target, "payload")
+
+        assert list(tmp_path.glob("*.tmp")) == [], "temp file survived a failed write"
+        assert not target.exists()
+
+    async def test_reader_never_observes_a_partially_written_file(self, tmp_path):
+        """
+        Rewrite one path repeatedly while a reader watches it. `open(path,'w')`
+        truncates first, so a reader can catch a short/empty file; an atomic
+        rename means every observation is a complete generation.
+        """
+        target = tmp_path / "analysis.md"
+        gen_a = "A" * 400_000
+        gen_b = "B" * 400_000
+        _mod.EnhancedVideoProcessor._atomic_write_text(target, gen_a)
+
+        observations = []
+        stop = False
+
+        async def reader():
+            while not stop:
+                with contextlib.suppress(FileNotFoundError):
+                    observations.append(target.read_text(encoding="utf-8"))
+                await asyncio.sleep(0)
+
+        async def writer():
+            for i in range(12):
+                await asyncio.to_thread(
+                    _mod.EnhancedVideoProcessor._atomic_write_text,
+                    target,
+                    gen_b if i % 2 else gen_a,
+                )
+
+        watcher = asyncio.create_task(reader())
+        await asyncio.sleep(0)
+        try:
+            await writer()
+        finally:
+            stop = True
+            watcher.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await watcher
+
+        assert observations, "reader never sampled the file"
+        bad = [len(o) for o in observations if o not in (gen_a, gen_b)]
+        assert not bad, f"reader saw {len(bad)} partial file(s), sizes={bad[:5]}"
+
+
+class TestConcurrentSavesWriteMatchedPairs:
+    """
+    Result filenames carry only one-second precision, so two saves of the same
+    video inside the same second resolve to the same two paths. Each write is
+    individually atomic, but without holding the pair together the two saves
+    can interleave and leave one save's markdown next to the other's metadata.
+    """
+
+    async def test_concurrent_saves_never_mix_markdown_with_foreign_metadata(self, tmp_path):
+        save_dir = tmp_path / "enhanced"
+        md_path = save_dir / "vid_20240101_000000_enhanced.md"
+        meta_path = save_dir / "vid_20240101_000000_metadata.json"
+
+        real_write = EnhancedVideoProcessor._atomic_write_text
+
+        def _interleaving_write(path, contents):
+            # Deterministically drive the worst-case interleaving: writer B
+            # waits long enough for A to land its markdown, and A stalls
+            # between its own two writes.
+            is_markdown = str(path).endswith(".md")
+            if is_markdown and contents.startswith("# B"):
+                time.sleep(0.05)
+            real_write(path, contents)
+            if is_markdown and contents.startswith("# A"):
+                time.sleep(0.15)
+
+        with patch.object(
+            EnhancedVideoProcessor, "_atomic_write_text", staticmethod(_interleaving_write)
+        ):
+            await asyncio.gather(
+                asyncio.to_thread(
+                    EnhancedVideoProcessor._write_result_files,
+                    save_dir, md_path, meta_path, "# A analysis", {"who": "A"},
+                ),
+                asyncio.to_thread(
+                    EnhancedVideoProcessor._write_result_files,
+                    save_dir, md_path, meta_path, "# B analysis", {"who": "B"},
+                ),
+            )
+
+        markdown = md_path.read_text(encoding="utf-8")
+        metadata = json.loads(meta_path.read_text(encoding="utf-8"))
+        winner = "A" if markdown.startswith("# A") else "B"
+
+        assert metadata["who"] == winner, (
+            f"markdown belongs to save {winner} but metadata belongs to "
+            f"save {metadata['who']} -- the pair interleaved"
+        )
+
+    async def test_pair_write_still_produces_both_files(self, tmp_path):
+        """Guard: serialising the pair must not change the single-writer result."""
+        save_dir = tmp_path / "enhanced"
+        md_path = save_dir / "v_enhanced.md"
+        meta_path = save_dir / "v_metadata.json"
+
+        await asyncio.to_thread(
+            EnhancedVideoProcessor._write_result_files,
+            save_dir, md_path, meta_path, "# only", {"who": "solo", "n": 1},
+        )
+
+        assert md_path.read_text(encoding="utf-8") == "# only"
+        assert json.loads(meta_path.read_text(encoding="utf-8")) == {"who": "solo", "n": 1}