From 94329ada3e3584c9cde10b3972116c491e0375c7 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 11:21:29 -0500 Subject: [PATCH] perf: offload transcript download cleanup to a worker thread The `finally` block in `_fallback_transcript_with_gemini` deleted the downloaded video and recursively removed its temp tree directly on the event loop. `Path.exists`, `Path.unlink` and `shutil.rmtree` are all blocking syscalls, and the temp tree can hold a merged mp4 plus unmerged `.fNNN` fragments, so every request that reaches the Gemini video fallback stalls the loop for the duration of the delete. Move the cleanup into `_cleanup_download_artifacts`, a static helper that runs the same logic under `asyncio.to_thread`. The call is wrapped in `asyncio.shield` because the original inline code was uncancellable: a bare `await` in a `finally` can be interrupted by a second cancellation, which would turn a loop stall into a disk leak. Filesystem semantics are preserved verbatim, including the `exists()` guards, `except OSError` and `ignore_errors=True`. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../workflows/transcript_action_workflow.py | 40 ++++- tests/unit/test_transcript_action_workflow.py | 146 ++++++++++++++++++ 2 files changed, 179 insertions(+), 7 deletions(-) diff --git a/src/youtube_extension/services/workflows/transcript_action_workflow.py b/src/youtube_extension/services/workflows/transcript_action_workflow.py index cb21fe673..c072d88d8 100644 --- a/src/youtube_extension/services/workflows/transcript_action_workflow.py +++ b/src/youtube_extension/services/workflows/transcript_action_workflow.py @@ -849,13 +849,7 @@ async def _fallback_transcript_with_gemini( if file_result.error: errors.append(file_result.error) finally: - if video_path.exists(): - try: - video_path.unlink() - except OSError: - pass - if temp_root and temp_root.exists(): - shutil.rmtree(temp_root, ignore_errors=True) + await self._cleanup_download_artifacts(video_path, temp_root) error_message = errors[0] if errors else "Gemini transcription failed" logger.warning("Gemini transcription fallback failed: %s", error_message) @@ -867,6 +861,38 @@ async def _fallback_transcript_with_gemini( "error": error_message, } + @staticmethod + async def _cleanup_download_artifacts( + video_path: Path | None, + temp_root: Path | None, + ) -> None: + """Remove downloaded video artifacts without blocking the event loop. + + The Gemini file fallback downloads a whole video into a temporary tree. + Because the format chain may fall back to separate video/audio streams, + that tree can hold the merged output plus unmerged ``.fNNN`` fragments, + so the removal is unbounded disk work. It therefore runs in a worker + thread using the same ``to_thread`` idiom as ``_download_video_file``. + + The await is shielded because this runs from a ``finally`` block. The + previous inline implementation was synchronous and so uncancellable, + which meant cleanup always completed; an unshielded await would let a + cancellation delivered during the ``finally`` skip cleanup and leak the + tree. Shielding preserves that "always cleans up" property while still + yielding the loop, and still re-raises ``CancelledError`` to the caller. + """ + + def _cleanup() -> None: + if video_path is not None and video_path.exists(): + try: + video_path.unlink() + except OSError: + pass + if temp_root is not None and temp_root.exists(): + shutil.rmtree(temp_root, ignore_errors=True) + + await asyncio.shield(asyncio.to_thread(_cleanup)) + async def _record_metric( self, name: str, diff --git a/tests/unit/test_transcript_action_workflow.py b/tests/unit/test_transcript_action_workflow.py index de021dba1..f566d6f96 100644 --- a/tests/unit/test_transcript_action_workflow.py +++ b/tests/unit/test_transcript_action_workflow.py @@ -1,6 +1,11 @@ from __future__ import annotations +import asyncio import datetime +import pathlib +import shutil +import threading +import time from dataclasses import asdict from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -1189,3 +1194,144 @@ async def test_success_via_youtube_url(self): assert result["text"] == "Gemini transcript" assert result["source"] == "gemini_video" + + +# --------------------------------------------------------------------------- +# _cleanup_download_artifacts +# --------------------------------------------------------------------------- + +class TestCleanupDownloadArtifactsOffEventLoop: + """The Gemini fallback must not delete downloaded video trees inline. + + ``_fallback_transcript_with_gemini`` downloads a full video into a + temporary tree, so the ``finally`` cleanup is unbounded disk work. These + tests pin that work to a worker thread rather than the event loop. + """ + + async def test_cleanup_runs_off_event_loop(self, monkeypatch, tmp_path): + loop_thread = threading.get_ident() + seen: dict[str, int] = {} + calls = {"unlink": 0, "rmtree": 0} + + real_unlink = pathlib.Path.unlink + real_rmtree = shutil.rmtree + + def recording_unlink(self, *args, **kwargs): + seen["unlink"] = threading.get_ident() + calls["unlink"] += 1 + return real_unlink(self, *args, **kwargs) + + def recording_rmtree(path, *args, **kwargs): + seen["rmtree"] = threading.get_ident() + calls["rmtree"] += 1 + return real_rmtree(path, *args, **kwargs) + + monkeypatch.setattr(pathlib.Path, "unlink", recording_unlink) + monkeypatch.setattr(shutil, "rmtree", recording_rmtree) + + temp_root = tmp_path / "gemini_video_abc" + temp_root.mkdir() + video_path = temp_root / "auJzb1D-fag.mp4" + video_path.write_bytes(b"video-bytes") + + await TranscriptActionWorkflow._cleanup_download_artifacts( + video_path, temp_root + ) + + # Guard against a vacuous pass: both primitives must really have run. + assert calls == {"unlink": 1, "rmtree": 1} + assert seen["unlink"] != loop_thread + assert seen["rmtree"] != loop_thread + + async def test_cleanup_does_not_block_event_loop(self, monkeypatch, tmp_path): + started = threading.Event() + release = threading.Event() + + def blocking_rmtree(path, *args, **kwargs): + started.set() + # Fail instead of hanging CI if the loop never gets to resume. + assert release.wait(timeout=10), "event loop blocked during cleanup" + + monkeypatch.setattr(shutil, "rmtree", blocking_rmtree) + + temp_root = tmp_path / "gemini_video_abc" + temp_root.mkdir() + + cleanup = asyncio.create_task( + TranscriptActionWorkflow._cleanup_download_artifacts(None, temp_root) + ) + + for _ in range(500): + if started.is_set(): + break + await asyncio.sleep(0.01) + + # Reaching here while rmtree is still parked proves the loop kept + # running concurrently with the deletion. + assert started.is_set(), "cleanup never started" + assert not cleanup.done() + + release.set() + await asyncio.wait_for(cleanup, timeout=10) + + async def test_cleanup_removes_artifacts(self, tmp_path): + temp_root = tmp_path / "gemini_video_abc" + temp_root.mkdir() + video_path = temp_root / "auJzb1D-fag.mp4" + video_path.write_bytes(b"video-bytes") + fragment = temp_root / "auJzb1D-fag.f140.m4a" + fragment.write_bytes(b"audio-bytes") + + await TranscriptActionWorkflow._cleanup_download_artifacts( + video_path, temp_root + ) + + assert not video_path.exists() + assert not fragment.exists() + assert not temp_root.exists() + + async def test_cleanup_survives_missing_paths(self, tmp_path): + # Mirrors the pre-existing guards: absent artifacts are not an error. + await TranscriptActionWorkflow._cleanup_download_artifacts( + tmp_path / "gone.mp4", tmp_path / "gone_dir" + ) + await TranscriptActionWorkflow._cleanup_download_artifacts(None, None) + + async def test_cleanup_completes_when_task_cancelled(self, monkeypatch, tmp_path): + # The replaced inline code was synchronous and therefore uncancellable, + # so cleanup always ran. The shielded await must preserve that. + started = threading.Event() + finished = threading.Event() + real_rmtree = shutil.rmtree + + def slow_rmtree(path, *args, **kwargs): + started.set() + time.sleep(0.3) + real_rmtree(path, *args, **kwargs) + finished.set() + + monkeypatch.setattr(shutil, "rmtree", slow_rmtree) + + temp_root = tmp_path / "gemini_video_abc" + temp_root.mkdir() + (temp_root / "auJzb1D-fag.mp4").write_bytes(b"video-bytes") + + task = asyncio.create_task( + TranscriptActionWorkflow._cleanup_download_artifacts(None, temp_root) + ) + + for _ in range(500): + if started.is_set(): + break + await asyncio.sleep(0.01) + assert started.is_set(), "cleanup never started" + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert finished.wait(timeout=10), "shielded cleanup did not finish" + assert not temp_root.exists() + + # Let the shielded inner task settle before the loop closes. + await asyncio.sleep(0.1)