diff --git a/src/youtube_extension/services/workflows/transcript_action_workflow.py b/src/youtube_extension/services/workflows/transcript_action_workflow.py index cb21fe673..d3a16ed50 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,70 @@ 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 ran to completion once entered; an unshielded await + would let a cancellation delivered during the ``finally`` skip cleanup + and leak the tree. Shielding preserves that property while still + yielding the loop, and still re-raises ``CancelledError`` to the caller. + + The bound is process lifetime, not an absolute guarantee: cleanup + continues unless the process exits. A ``SIGKILL``, a hard crash, or + interpreter shutdown before the worker thread finishes still leaves the + tree behind, and nothing in this helper can prevent that. + + ``CancelledError`` is deliberately allowed to propagate rather than + being suppressed in favour of any exception already in flight. Python + chains the in-flight exception onto it as ``__context__``, so no + diagnostic information is lost, whereas swallowing it would report a + cancelled task as ``cancelled() is False`` and defeat + ``asyncio.timeout``. The caller in ``_extract_transcript`` catches + ``Exception``, so a suppressed cancellation would be downgraded to a + per-source error and the pipeline would keep issuing network calls + after the request was abandoned. + """ + + def _cleanup() -> None: + # Both branches are total. This runs from a ``finally``, so raising + # here would replace whatever exception is already propagating. + # + # Note the absence of ``exists()`` probes: ``exists()`` performs a + # stat and can itself raise, which is exactly the masking this must + # avoid. ``unlink`` already raises ``FileNotFoundError`` for absent + # paths and ``rmtree`` tolerates them. + # + # The guards catch ``Exception``, not ``OSError``, because neither + # call is OSError-total. A path holding a NUL byte makes ``unlink`` + # raise ``ValueError: embedded null character``, and makes + # ``rmtree`` raise the same from its internal ``lstat`` despite + # ``ignore_errors=True`` -- that flag only suppresses ``OSError``. + # ``CancelledError`` is a ``BaseException``, so it still propagates. + if video_path is not None: + try: + video_path.unlink() + except Exception: # noqa: BLE001 - must not mask in-flight error + logger.debug("Cleanup failed for %s", video_path, exc_info=True) + if temp_root is not None: + try: + shutil.rmtree(temp_root, ignore_errors=True) + except Exception: # noqa: BLE001 - must not mask in-flight error + logger.debug("Cleanup failed for %s", temp_root, exc_info=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..7b2cb577e 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,217 @@ 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) + ) + + # Poll against a wall-clock deadline rather than a fixed iteration + # count: a loaded CI box may take a while to hand the cleanup closure a + # worker thread, and a fixed budget would fail for scheduling reasons + # rather than for the behaviour under test. The polling itself is the + # assertion -- each completed tick is one turn of the event loop taken + # while rmtree is parked -- so this cannot be replaced by a blocking + # wait without destroying what the test proves. + deadline = time.monotonic() + 30.0 + ticks = 0 + while not started.is_set(): + assert time.monotonic() < deadline, "cleanup never started" + await asyncio.sleep(0.01) + ticks += 1 + + # Reaching here while rmtree is still parked proves the loop kept + # running concurrently with the deletion. + assert ticks >= 1, "event loop never yielded while cleanup was running" + 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) + ) + + # Wall-clock deadline rather than a fixed iteration budget: a loaded + # CI runner can stretch each sleep well past 10ms. + deadline = time.monotonic() + 30.0 + while not started.is_set(): + assert time.monotonic() < deadline, "cleanup never started" + await asyncio.sleep(0.01) + + 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) + + async def test_cleanup_does_not_mask_in_flight_exception( + self, monkeypatch, tmp_path + ): + """A cleanup failure must never replace the exception being propagated. + + The helper runs from a ``finally``. If it raised, it would discard the + real error and report a spurious filesystem fault instead. Both removal + primitives are therefore total. + """ + + def exploding_unlink(self, *args, **kwargs): + raise PermissionError("read-only filesystem") + + def exploding_stat(self, *args, **kwargs): + raise OSError("stat exploded") + + monkeypatch.setattr(pathlib.Path, "unlink", exploding_unlink) + # ``Path.exists()`` is implemented via ``stat()``. Patching stat proves + # the helper never probes a path in a way that could itself raise. + # ``shutil.rmtree`` is left real: its ``ignore_errors=True`` is the + # documented mechanism that makes the directory removal total, so + # patching it away would test a guarantee the code never claimed. + monkeypatch.setattr(pathlib.Path, "stat", exploding_stat) + + class Boom(Exception): + pass + + async def failing_operation(): + try: + raise Boom("the real error") + finally: + await TranscriptActionWorkflow._cleanup_download_artifacts( + tmp_path / "video.mp4", tmp_path / "absent_tree" + ) + + # The original exception survives; no OSError leaks out of cleanup. + with pytest.raises(Boom, match="the real error"): + await failing_operation() + + async def test_cleanup_is_total_for_non_oserror_failures(self): + """A NUL byte in either path must not escape as ``ValueError``. + + ``shutil.rmtree(..., ignore_errors=True)`` only suppresses ``OSError``: + a NUL byte makes its internal ``lstat`` raise ``ValueError``, and + ``Path.unlink`` raises the same directly. Because the helper runs from + a ``finally``, either escape would replace the in-flight exception -- + the exact defect the removal of the ``exists()`` probes fixed. No + mocking is used, so this exercises the real stdlib behaviour. + """ + nul_video = pathlib.Path("/tmp/eventrelay-nul\x00.mp4") + nul_root = pathlib.Path("/tmp/eventrelay-nul\x00-dir") + + # Premise: the bare calls really are not OSError-total. + with pytest.raises(ValueError, match="null"): + nul_video.unlink() + with pytest.raises(ValueError, match="null"): + shutil.rmtree(nul_root, ignore_errors=True) + + # Contract: the helper swallows both and returns normally. + await TranscriptActionWorkflow._cleanup_download_artifacts( + nul_video, nul_root + )