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 1/4] 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) From f2329a1c831b7c808642157702b1f244f649463d Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 11:30:18 -0500 Subject: [PATCH 2/4] test: replace fixed poll budget with a wall-clock deadline The event-loop liveness test polled a fixed 500 iterations while rmtree was parked on an unset Event. That is a ~5s budget expressed as an iteration count, so a loaded CI box that is slow to hand asyncio.to_thread a worker would fail the test for scheduling reasons rather than for the behaviour under test. Poll against time.monotonic() with a 30s deadline instead. The loop still exits as soon as the worker starts (milliseconds in practice), so the test is not slower; it simply stops being brittle under load. Also assert the tick count directly. Each completed tick is one turn of the event loop taken while the deletion was in flight, which is precisely the property being proven, and it was previously only implied by reaching the assertion at all. The polling is deliberately retained rather than replaced with a blocking wait on the Event: blocking the loop to wait for proof that the loop is not blocked would invert the test. Re-verified against pre-change semantics -- reducing the helper to a direct _cleanup() call still yields 3 failed, 2 passed, so this test is no weaker than before. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/unit/test_transcript_action_workflow.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/tests/unit/test_transcript_action_workflow.py b/tests/unit/test_transcript_action_workflow.py index f566d6f96..506597b2e 100644 --- a/tests/unit/test_transcript_action_workflow.py +++ b/tests/unit/test_transcript_action_workflow.py @@ -1261,14 +1261,23 @@ def blocking_rmtree(path, *args, **kwargs): TranscriptActionWorkflow._cleanup_download_artifacts(None, temp_root) ) - for _ in range(500): - if started.is_set(): - break + # 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 started.is_set(), "cleanup never started" + assert ticks >= 1, "event loop never yielded while cleanup was running" assert not cleanup.done() release.set() From 4ab5a7ddc8d491161c0b80e29e10740a7d304aa3 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 11:42:38 -0500 Subject: [PATCH 3/4] perf: make temp-tree cleanup total to avoid masking exceptions The cleanup helper runs from a `finally` block. Its `Path.exists()` probes performed a stat that can itself raise `OSError`, which would propagate out of the `finally` and replace the exception already in flight. Remove both probes. `unlink()` raises `FileNotFoundError` (an `OSError`, already caught) for absent paths and `rmtree(ignore_errors=True)` is a no-op, so the guards were redundant as well as unsafe. Also harden the cancellation test's poll loop to a wall-clock deadline instead of a fixed iteration budget, matching the sibling test. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../workflows/transcript_action_workflow.py | 20 +++++++- tests/unit/test_transcript_action_workflow.py | 48 +++++++++++++++++-- 2 files changed, 62 insertions(+), 6 deletions(-) diff --git a/src/youtube_extension/services/workflows/transcript_action_workflow.py b/src/youtube_extension/services/workflows/transcript_action_workflow.py index c072d88d8..e2b19b58e 100644 --- a/src/youtube_extension/services/workflows/transcript_action_workflow.py +++ b/src/youtube_extension/services/workflows/transcript_action_workflow.py @@ -880,15 +880,31 @@ async def _cleanup_download_artifacts( 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. + + ``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: - if video_path is not None and video_path.exists(): + # 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 ``OSError``, which is exactly the + # masking this must avoid. ``unlink`` already raises + # ``FileNotFoundError`` for absent paths and ``rmtree`` is a no-op. + if video_path is not None: try: video_path.unlink() except OSError: pass - if temp_root is not None and temp_root.exists(): + if temp_root is not None: shutil.rmtree(temp_root, ignore_errors=True) await asyncio.shield(asyncio.to_thread(_cleanup)) diff --git a/tests/unit/test_transcript_action_workflow.py b/tests/unit/test_transcript_action_workflow.py index 506597b2e..d1bdf80b4 100644 --- a/tests/unit/test_transcript_action_workflow.py +++ b/tests/unit/test_transcript_action_workflow.py @@ -1329,11 +1329,12 @@ def slow_rmtree(path, *args, **kwargs): TranscriptActionWorkflow._cleanup_download_artifacts(None, temp_root) ) - for _ in range(500): - if started.is_set(): - break + # 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) - assert started.is_set(), "cleanup never started" task.cancel() with pytest.raises(asyncio.CancelledError): @@ -1344,3 +1345,42 @@ def slow_rmtree(path, *args, **kwargs): # 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() From 627ebe20ef504db3b760b4b8b03fc12657fe75cb Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 12:01:12 -0500 Subject: [PATCH 4/4] fix: make download-artifact cleanup total for non-OSError failures The cleanup helper runs from a `finally`, so anything it raises replaces the exception already propagating. Its guards caught `OSError` only, but neither call is OSError-total: a NUL byte in a path makes `Path.unlink` raise `ValueError: embedded null character`, and makes `shutil.rmtree` raise the same from its internal `lstat` despite `ignore_errors=True` -- that flag suppresses `OSError` alone. Both branches now catch `Exception` and log at debug with `exc_info`. `CancelledError` is a `BaseException`, so cancellation still propagates. Not reachable in production today: `temp_root` comes from `mkdtemp` and a NUL `video_path` is already rejected by the `exists()` guard in `_download`. This corrects a false totality claim in the contract. Verified: reverting to the `OSError`-only guards fails the new test with the escaping `ValueError` (1 failed, 6 passed); restored 115 passed. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../workflows/transcript_action_workflow.py | 23 +++++++++++++----- tests/unit/test_transcript_action_workflow.py | 24 +++++++++++++++++++ 2 files changed, 41 insertions(+), 6 deletions(-) diff --git a/src/youtube_extension/services/workflows/transcript_action_workflow.py b/src/youtube_extension/services/workflows/transcript_action_workflow.py index e2b19b58e..8c3b0574d 100644 --- a/src/youtube_extension/services/workflows/transcript_action_workflow.py +++ b/src/youtube_extension/services/workflows/transcript_action_workflow.py @@ -895,17 +895,28 @@ async def _cleanup_download_artifacts( 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 ``OSError``, which is exactly the - # masking this must avoid. ``unlink`` already raises - # ``FileNotFoundError`` for absent paths and ``rmtree`` is a no-op. + # 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 OSError: - pass + 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: - shutil.rmtree(temp_root, ignore_errors=True) + 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)) diff --git a/tests/unit/test_transcript_action_workflow.py b/tests/unit/test_transcript_action_workflow.py index d1bdf80b4..7b2cb577e 100644 --- a/tests/unit/test_transcript_action_workflow.py +++ b/tests/unit/test_transcript_action_workflow.py @@ -1384,3 +1384,27 @@ async def failing_operation(): # 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 + )