From 757e8475c5e0df7a64366107f17ab48516b3398a Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:52:49 -0500 Subject: [PATCH 1/2] perf(cloud-ai): analyse each batch concurrently in process_batch_videos The batch loop sliced video_urls into batch_size chunks but then awaited ai.analyze_video() for each video in turn, so batch_size controlled nothing except how often the inter-batch pause fired - wall-clock cost stayed the full sum of every per-video analysis. Analyse each batch with asyncio.gather(..., return_exceptions=True) so batch_size becomes the real bound on concurrent calls to the shared upstream AI providers. isinstance(result, Exception) preserves the previous except-Exception-and-continue semantics exactly, so one failing video is logged and skipped without aborting its batch. Also hoists the loop-local 'import asyncio' to module scope. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/cloud_ai_routes.py | 32 +++-- tests/unit/test_cloud_ai_routes_batch.py | 126 ++++++++++++++++++ 2 files changed, 149 insertions(+), 9 deletions(-) create mode 100644 tests/unit/test_cloud_ai_routes_batch.py diff --git a/src/youtube_extension/backend/cloud_ai_routes.py b/src/youtube_extension/backend/cloud_ai_routes.py index 49804aba1..81d9ee986 100644 --- a/src/youtube_extension/backend/cloud_ai_routes.py +++ b/src/youtube_extension/backend/cloud_ai_routes.py @@ -4,6 +4,7 @@ FastAPI routes for cloud AI video analysis services. """ +import asyncio import logging from typing import Any, Optional @@ -377,23 +378,36 @@ async def process_batch_videos(video_urls: list[str], analysis_types: list[Analy for i in range(0, len(video_urls), batch_size): batch = video_urls[i:i+batch_size] - for video_url in batch: - try: - result = await ai.analyze_video( + # Analyse the batch concurrently. batch_size is the intended + # concurrency bound against the shared upstream AI providers, + # so the batch is exactly the right unit to run in parallel. + # Awaiting each video in turn made batch_size control nothing + # but the cadence of the pause below, while the wall-clock cost + # stayed the full sum of every per-video analysis. + batch_results = await asyncio.gather( + *( + ai.analyze_video( video_url=video_url, analysis_types=analysis_types, preferred_provider=preferred_provider, - use_fallback=True + use_fallback=True, ) - results.append(format_analysis_result(result)) - - except Exception as e: - logger.error(f"Failed to analyze video {video_url}: {e}") + for video_url in batch + ), + return_exceptions=True, + ) + + # isinstance(..., Exception) mirrors the previous + # `except Exception ... continue`: one failing video is logged + # and skipped without aborting the rest of the batch. + for video_url, result in zip(batch, batch_results, strict=True): + if isinstance(result, Exception): + logger.error(f"Failed to analyze video {video_url}: {result}") continue + results.append(format_analysis_result(result)) # Brief pause between batches if i + batch_size < len(video_urls): - import asyncio await asyncio.sleep(1) # Store results (in production, save to database or cache) diff --git a/tests/unit/test_cloud_ai_routes_batch.py b/tests/unit/test_cloud_ai_routes_batch.py new file mode 100644 index 000000000..3e77470ff --- /dev/null +++ b/tests/unit/test_cloud_ai_routes_batch.py @@ -0,0 +1,126 @@ +"""Unit tests for cloud_ai_routes.process_batch_videos batch fan-out.""" + +from __future__ import annotations + +import asyncio +import contextlib +from unittest.mock import AsyncMock, MagicMock, patch + +from youtube_extension.backend import cloud_ai_routes as routes + + +class _FakeIntegrator: + """Async context manager standing in for CloudAIIntegrator.""" + + def __init__(self, analyze): + self.analyze_video = analyze + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc_info): + return False + + +@contextlib.contextmanager +def _fake_cloud_ai(analyze): + """Run process_batch_videos against a fake integrator. + + ``routes.asyncio`` is replaced so the inter-batch ``asyncio.sleep(1)`` + pause does not add real wall-clock time. Only the module-level name is + patched, so ``asyncio`` inside these tests is untouched and ``gather`` + keeps its real behaviour. + """ + fake_asyncio = MagicMock() + fake_asyncio.gather = asyncio.gather + fake_asyncio.sleep = AsyncMock() + with ( + patch.object(routes, "get_cloud_ai_config", return_value={}), + patch.object(routes, "CloudAIIntegrator", lambda _cfg: _FakeIntegrator(analyze)), + patch.object(routes, "format_analysis_result", side_effect=lambda r: r), + patch.object(routes, "asyncio", fake_asyncio), + ): + yield fake_asyncio + + +def _tracking_analyze(state, failures=()): + """analyze_video stand-in that records peak concurrent in-flight calls.""" + + async def _analyze(video_url, **_kwargs): + state["inflight"] += 1 + state["peak"] = max(state["peak"], state["inflight"]) + try: + await asyncio.sleep(0.01) + if video_url in failures: + raise RuntimeError(f"provider rejected {video_url}") + return f"result:{video_url}" + finally: + state["inflight"] -= 1 + + return _analyze + + +class TestProcessBatchVideosFanOut: + """A batch must be analysed concurrently, bounded by batch_size. + + The previous implementation awaited each video in turn, so batch_size + controlled nothing but the cadence of the inter-batch pause while + wall-clock cost stayed the full sum of every per-video analysis. + """ + + async def test_batch_is_analysed_concurrently(self): + state = {"inflight": 0, "peak": 0} + urls = [f"https://v/{i}" for i in range(4)] + + with _fake_cloud_ai(_tracking_analyze(state)): + await routes.process_batch_videos(urls, [], None, 4, "task-1") + + assert state["peak"] == 4, ( + f"batch peaked at {state['peak']} concurrent analyze_video call(s) " + "for a batch of 4 - the batch is being analysed sequentially" + ) + + async def test_batch_size_bounds_concurrency(self): + """batch_size must remain a real bound on the shared upstream API.""" + state = {"inflight": 0, "peak": 0} + urls = [f"https://v/{i}" for i in range(6)] + + with _fake_cloud_ai(_tracking_analyze(state)): + await routes.process_batch_videos(urls, [], None, 2, "task-2") + + assert state["peak"] == 2, ( + f"batch_size=2 but peak concurrency was {state['peak']}; the batch " + "boundary must still bound in-flight provider calls" + ) + + async def test_pause_still_applied_between_batches_only(self): + state = {"inflight": 0, "peak": 0} + urls = [f"https://v/{i}" for i in range(6)] + + with _fake_cloud_ai(_tracking_analyze(state)) as fake_asyncio: + await routes.process_batch_videos(urls, [], None, 2, "task-3") + + # 6 urls / batch_size 2 = 3 batches -> 2 inter-batch pauses. + assert fake_asyncio.sleep.await_count == 2 + + async def test_failed_video_does_not_abort_the_batch(self): + state = {"inflight": 0, "peak": 0} + urls = ["https://v/0", "https://v/1", "https://v/2"] + collected = [] + + analyze = _tracking_analyze(state, failures={"https://v/0"}) + with _fake_cloud_ai(analyze): + with patch.object( + routes, "format_analysis_result", side_effect=lambda r: collected.append(r) or r + ): + await routes.process_batch_videos(urls, [], None, 3, "task-4") + + assert collected == ["result:https://v/1", "result:https://v/2"], ( + f"expected the two healthy videos to be collected, got {collected}" + ) + + async def test_empty_url_list_is_a_noop(self): + state = {"inflight": 0, "peak": 0} + with _fake_cloud_ai(_tracking_analyze(state)): + await routes.process_batch_videos([], [], None, 4, "task-5") + assert state["peak"] == 0 From 8725b51dfc3debcf87e1f2afb9cbca5d52af9b41 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 17:03:12 -0500 Subject: [PATCH 2/2] fix(cloud-ai): propagate cancellation out of batch fan-out `asyncio.gather(..., return_exceptions=True)` captures a child's CancelledError as a *value*. Because it derives from BaseException and not Exception, the `isinstance(result, Exception)` filter let it fall through to `format_analysis_result`, which raised AttributeError that the outer `except Exception` then swallowed -- silently skipping every remaining batch. Re-raise BaseException-but-not-Exception results before the per-video failure handling, restoring the propagation semantics of the original sequential `await`/`except Exception` loop. Reported by @Copilot in review of #1188. Adds 2 regression tests that fail with "DID NOT RAISE CancelledError" without this change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/cloud_ai_routes.py | 15 +++++ tests/unit/test_cloud_ai_routes_batch.py | 67 +++++++++++++++++++ 2 files changed, 82 insertions(+) diff --git a/src/youtube_extension/backend/cloud_ai_routes.py b/src/youtube_extension/backend/cloud_ai_routes.py index 81d9ee986..3824dc985 100644 --- a/src/youtube_extension/backend/cloud_ai_routes.py +++ b/src/youtube_extension/backend/cloud_ai_routes.py @@ -397,6 +397,21 @@ async def process_batch_videos(video_urls: list[str], analysis_types: list[Analy return_exceptions=True, ) + # `return_exceptions=True` also captures BaseException-only + # failures (most notably CancelledError) as *values*, so an + # `isinstance(..., Exception)` test alone would let a cancelled + # child fall through to format_analysis_result. Re-raise those + # to preserve the previous `await`/`except Exception` semantics, + # where cancellation propagated out of the loop. + cancellations = [ + result + for result in batch_results + if isinstance(result, BaseException) + and not isinstance(result, Exception) + ] + if cancellations: + raise cancellations[0] + # isinstance(..., Exception) mirrors the previous # `except Exception ... continue`: one failing video is logged # and skipped without aborting the rest of the batch. diff --git a/tests/unit/test_cloud_ai_routes_batch.py b/tests/unit/test_cloud_ai_routes_batch.py index 3e77470ff..9ddd19e19 100644 --- a/tests/unit/test_cloud_ai_routes_batch.py +++ b/tests/unit/test_cloud_ai_routes_batch.py @@ -6,6 +6,8 @@ import contextlib from unittest.mock import AsyncMock, MagicMock, patch +import pytest + from youtube_extension.backend import cloud_ai_routes as routes @@ -124,3 +126,68 @@ async def test_empty_url_list_is_a_noop(self): with _fake_cloud_ai(_tracking_analyze(state)): await routes.process_batch_videos([], [], None, 4, "task-5") assert state["peak"] == 0 + + +class TestProcessBatchVideosCancellation: + """Cancellation must propagate, not be misread as a successful result. + + ``asyncio.gather(..., return_exceptions=True)`` captures a child's + ``CancelledError`` as a *value*. Because it derives from ``BaseException`` + and not ``Exception``, an ``isinstance(result, Exception)`` filter alone + would hand it to ``format_analysis_result`` as if it were an analysis. + The previous sequential ``await``/``except Exception`` loop let + cancellation escape, so it must still escape here. + """ + + async def test_cancelled_analysis_is_re_raised_not_treated_as_a_result(self): + async def analyze(video_url, **_kwargs): + if video_url == "u2": + raise asyncio.CancelledError() + await asyncio.sleep(0) + return f"result:{video_url}" + + formatted: list[object] = [] + + with _fake_cloud_ai(analyze) as fake_asyncio: + with patch.object( + routes, + "format_analysis_result", + side_effect=lambda r: formatted.append(r) or r, + ): + with pytest.raises(asyncio.CancelledError): + await routes.process_batch_videos( + task_id="t-cancel", + video_urls=["u1", "u2", "u3", "u4"], + analysis_types=[], + preferred_provider=None, + batch_size=2, + ) + + assert not any(isinstance(r, BaseException) for r in formatted), ( + f"a CancelledError leaked into format_analysis_result: {formatted}" + ) + assert fake_asyncio.sleep.await_count == 0, ( + "cancellation did not abort the run: the inter-batch pause ran and " + "the following batch was still dispatched" + ) + + async def test_cancellation_takes_precedence_over_ordinary_failures(self): + """A cancelled peer still propagates when siblings also failed.""" + + async def analyze(video_url, **_kwargs): + if video_url == "boom": + raise RuntimeError("provider exploded") + if video_url == "cancelled": + raise asyncio.CancelledError() + await asyncio.sleep(0) + return f"result:{video_url}" + + with _fake_cloud_ai(analyze): + with pytest.raises(asyncio.CancelledError): + await routes.process_batch_videos( + task_id="t-mixed", + video_urls=["ok", "boom", "cancelled"], + analysis_types=[], + preferred_provider=None, + batch_size=3, + )