From 4d8d250e51f172e2e6fa4a4475402f4c91886242 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 17:46:37 -0500 Subject: [PATCH 1/2] perf(video): run enhanced-result persistence off the event loop `EnhancedVideoProcessor._save_enhanced_result` is `async def`, but every byte of work inside it was synchronous and executed on the event loop: 1. `save_dir.mkdir(parents=True, exist_ok=True)` - directory syscalls 2. `open(filepath,'w') / f.write(markdown)` - the full analysis doc 3. `json.dump(metadata, f, indent=2)` - serialises AND writes incrementally, so a large metadata dict became many small `write()` syscalls rather than one It is called from `process_video` (line 189), which is reached in production via `service_container.py:253` -> `video_processor_factory .get_video_processor()` -> `EnhancedVideoProcessor()`. While a video's results were being saved, every other in-flight request on that worker was stalled. This change hands the whole group - mkdir, both writes, and the `json.dumps` - to a worker thread in a *single* `asyncio.to_thread` dispatch, so the save costs one thread hop rather than one per syscall, and the serialisation cost is paid off-loop too. Writes are also made atomic. `open(path,'w')` truncates before it writes, so a crash or a concurrent reader can leave/observe a half-written analysis on disk. `_atomic_write_text` writes a sibling temp file and `os.replace`s it into place; the temp name carries the pid and thread id so two writers cannot collide, and it is unlinked if the write fails. This pre-empts the lost-serialisation class of bug that review caught on #1194: moving a write off-loop removes the event loop's implicit serialisation, so the write must become atomic. Honest framing: this does NOT make saving faster. It stops saving from stalling the event loop, and it stops partial files from being visible. Tests: 104 passed = 98 pre-existing (zero edits) + 6 new. Non-vacuity proven by reverting both dimensions simultaneously (atomic write -> plain open, to_thread -> direct call): exactly 4 targeted failures / 100 passed, restored -> 104. 4 of the 6 new tests discriminate; 2 are guards. ruff: exact parity with origin/main (identical 6 pre-existing findings). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/enhanced_video_processor.py | 67 +++++++- tests/unit/test_enhanced_video_processor.py | 153 ++++++++++++++++++ 2 files changed, 212 insertions(+), 8 deletions(-) diff --git a/src/youtube_extension/backend/enhanced_video_processor.py b/src/youtube_extension/backend/enhanced_video_processor.py index 8b4a86e52..76c782e25 100644 --- a/src/youtube_extension/backend/enhanced_video_processor.py +++ b/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 @@ -725,6 +727,50 @@ 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. + """ + save_dir.mkdir(parents=True, exist_ok=True) + EnhancedVideoProcessor._atomic_write_text(filepath, markdown) + EnhancedVideoProcessor._atomic_write_text( + metadata_file, json.dumps(metadata, indent=2, default=str) + ) + async def _save_enhanced_result(self, video_id: str, metadata: dict, markdown: str) -> str: """Save enhanced results to organized directory structure""" try: @@ -733,19 +779,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) diff --git a/tests/unit/test_enhanced_video_processor.py b/tests/unit/test_enhanced_video_processor.py index 818e7e45f..794eef8d0 100644 --- a/tests/unit/test_enhanced_video_processor.py +++ b/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,153 @@ 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]}" From fd5f49ede7a0f005fb6bee9cad7899f51be62d75 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 17:58:26 -0500 Subject: [PATCH 2/2] fix(video): write the result markdown/metadata pair atomically as a pair Review finding (@Copilot, enhanced_video_processor.py:772): the two writes in `_write_result_files` are individually atomic but were not held together. Result filenames carry only one-second precision, so two saves of the same video inside the same second resolve to the same two paths and can interleave as A-markdown, B-markdown, B-metadata, A-metadata -- leaving B's markdown paired with A's metadata. Hold both writes under a module-level `_RESULT_WRITE_LOCK`. The lock is a `threading.Lock` because the writes execute in the worker thread, and it is held only across two file writes, so it never blocks the event loop. Saves are a once-per-video-completion operation, so global serialisation of the write pair costs nothing measurable. - 106 tests pass (104 + 2 new in `TestConcurrentSavesWriteMatchedPairs`) - Non-vacuity: removing the lock yields exactly 1 targeted failure / 105 passed; the single-writer guard test still passes Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/enhanced_video_processor.py | 22 +++++-- tests/unit/test_enhanced_video_processor.py | 64 +++++++++++++++++++ 2 files changed, 82 insertions(+), 4 deletions(-) diff --git a/src/youtube_extension/backend/enhanced_video_processor.py b/src/youtube_extension/backend/enhanced_video_processor.py index 76c782e25..f7bd0da45 100644 --- a/src/youtube_extension/backend/enhanced_video_processor.py +++ b/src/youtube_extension/backend/enhanced_video_processor.py @@ -31,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) @@ -764,12 +773,17 @@ def _write_result_files( 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) - EnhancedVideoProcessor._atomic_write_text(filepath, markdown) - EnhancedVideoProcessor._atomic_write_text( - metadata_file, json.dumps(metadata, indent=2, default=str) - ) + 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""" diff --git a/tests/unit/test_enhanced_video_processor.py b/tests/unit/test_enhanced_video_processor.py index 794eef8d0..1f22583a5 100644 --- a/tests/unit/test_enhanced_video_processor.py +++ b/tests/unit/test_enhanced_video_processor.py @@ -1370,3 +1370,67 @@ async def writer(): 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}