From 54408df6f59b8a803903a0052e1663189807a3cb Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:34:35 -0500 Subject: [PATCH] perf: isolate blocking file I/O from the shared default executor (#1234) `asyncio.to_thread` is `run_in_executor(None, ...)`, so every offload in the repo lands on one loop-wide `ThreadPoolExecutor` (16 workers here). A stalled read on NFS/FUSE/remote-backed storage permanently leaks its worker slot: cancelling the task releases the *caller*, but `concurrent.futures` cannot interrupt a thread parked in a syscall, so the worker stays stuck. Timeouts therefore bound the coroutine, never the thread. Enough concurrent stalls starve every other offload in the process. Add `utils/blocking_io.py`: a lazily-built, explicitly-sized, process-wide pool (`BLOCKING_IO_MAX_WORKERS`, default 8) plus `run_blocking()`. Blast radius is now contained to that pool instead of the shared default one. - `functools.partial` so kwargs work (`run_in_executor` accepts none). - In-flight accounting is attached to the *concurrent* future, not the asyncio one, so a timed-out caller keeps counting its still-running worker rather than hiding the leaked slot the instrumentation exists to reveal. - Throttled saturation warning (30s) with structured `extra`. - `reset_blocking_io_executor()` shuts down with `wait=False`; blocking on a stuck worker would reintroduce the very hang this contains. Migrate the three `_read_file_bytes` sites named in AC1 only (Azure, Google, AWS vision). The remaining ~93 offloads are left for incremental migration. Tests pin isolation in both directions: saturate the I/O pool and unrelated `to_thread` work still completes; saturate the *default* pool and the vision reads still complete. The second direction fails on `main`, so it is a real behavioral regression guard rather than an import smoke test. Verified: 16/16 new tests pass; 313/313 existing provider tests pass both with and without the change. Three negative controls each fail a distinct, targeted subset - routing offloads back to the default pool fails 5, leaving providers on `to_thread` fails only the 4 routing tests, and deleting the saturation warning fails only the AC3 test. Refs #1234 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../cloud_ai/providers/aws_rekognition.py | 4 +- .../cloud_ai/providers/azure_vision.py | 4 +- .../cloud_ai/providers/google_cloud.py | 4 +- src/youtube_extension/utils/blocking_io.py | 208 ++++++++++++++++ tests/unit/test_blocking_io.py | 229 ++++++++++++++++++ ...test_vision_provider_executor_isolation.py | 229 ++++++++++++++++++ 6 files changed, 675 insertions(+), 3 deletions(-) create mode 100644 src/youtube_extension/utils/blocking_io.py create mode 100644 tests/unit/test_blocking_io.py create mode 100644 tests/unit/test_vision_provider_executor_isolation.py diff --git a/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py b/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py index 9b300bdb4..eec295afa 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py +++ b/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py @@ -16,6 +16,8 @@ from datetime import datetime from typing import Any +from youtube_extension.utils.blocking_io import run_blocking + from ..base import ( AnalysisType, BaseCloudAI, @@ -484,7 +486,7 @@ async def _prepare_image_input(self, image_url: str) -> dict[str, Any]: return {'Bytes': response.content} else: # Local file - read off the event loop - return {'Bytes': await asyncio.to_thread(_read_file_bytes, image_url)} + return {'Bytes': await run_blocking(_read_file_bytes, image_url)} def _process_video_results(self, results: dict[str, Any], video_id: str, analysis_types: list[AnalysisType], diff --git a/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py b/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py index b05e4c82f..13deecaa0 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py +++ b/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py @@ -13,6 +13,8 @@ from datetime import datetime from typing import Any, Optional +from youtube_extension.utils.blocking_io import run_blocking + from ..base import ( AnalysisType, BaseCloudAI, @@ -259,7 +261,7 @@ async def _prepare_image_input(self, image_url: str) -> Optional[bytes]: return None else: # Local file - read off the event loop - return await asyncio.to_thread(_read_file_bytes, image_url) + return await run_blocking(_read_file_bytes, image_url) async def _await_ocr_call(self, deadline: float, func: Any, *args: Any, **kwargs: Any) -> Any: """Run a blocking Azure SDK call in a worker thread, bounded by a diff --git a/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py b/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py index 98e6de8c3..deb2bed65 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py +++ b/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py @@ -14,6 +14,8 @@ from datetime import datetime from typing import Any +from youtube_extension.utils.blocking_io import run_blocking + from ..base import ( AnalysisType, BaseCloudAI, @@ -188,7 +190,7 @@ async def analyze_image(self, image_url: str, image.source.image_uri = image_url else: # Local file - read off the event loop - image.content = await asyncio.to_thread(_read_file_bytes, image_url) + image.content = await run_blocking(_read_file_bytes, image_url) # Prepare features features = self._prepare_vision_features(analysis_types) diff --git a/src/youtube_extension/utils/blocking_io.py b/src/youtube_extension/utils/blocking_io.py new file mode 100644 index 000000000..3e19c3b37 --- /dev/null +++ b/src/youtube_extension/utils/blocking_io.py @@ -0,0 +1,208 @@ +"""Isolated, bounded thread pool for blocking I/O offloads. + +Why this module exists (issue #1234) +------------------------------------ +``asyncio.to_thread(fn)`` is defined as ``loop.run_in_executor(None, fn)``: it +runs on the loop's **shared default** ``ThreadPoolExecutor``, sized +``min(32, cpu_count + 4)``. Every ``to_thread`` and every +``run_in_executor(None, ...)`` caller in the process therefore competes for the +same small pool. + +That is fine for work that always terminates. It is not fine for file I/O. A +path handed to ``open()`` may resolve to NFS/FUSE/remote-backed storage and stall +without limit, and a stalled work item **permanently leaks its worker slot**: + + task = asyncio.create_task(asyncio.to_thread(blocking_read)) + task.cancel() # the *caller* is released... + # ...the worker thread is still stuck. + +``concurrent.futures`` cannot interrupt a thread that is blocked in a syscall, so +cancellation frees the awaiting coroutine and nothing else. Enough concurrent +stalls exhaust the pool and every unrelated subsystem that uses ``to_thread`` +stops making progress. + +A timeout does not fix this. ``asyncio.wait_for`` bounds the *caller*, not the +*worker*; it converts an indefinite hang into a prompt error (worth having) but +returns no capacity to the pool. The only thing that actually contains the +failure is **isolation**: give blocking I/O its own bounded pool, so a stall can +exhaust that pool and never the one the rest of the process depends on. + +Usage:: + + from youtube_extension.utils.blocking_io import run_blocking + + data = await run_blocking(_read_file_bytes, path) + +Configuration +------------- +``BLOCKING_IO_MAX_WORKERS`` — worker count for the dedicated pool. Defaults to +``8``. Invalid or non-positive values fall back to the default rather than +raising, since a bad env var must not take the process down at import time. +""" + +from __future__ import annotations + +import asyncio +import functools +import logging +import os +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from typing import Any, Callable, TypeVar + +logger = logging.getLogger(__name__) + +T = TypeVar("T") + +_MAX_WORKERS_ENV = "BLOCKING_IO_MAX_WORKERS" +_DEFAULT_MAX_WORKERS = 8 +_THREAD_NAME_PREFIX = "blocking-io" + +# Sustained saturation would otherwise emit one warning per submission and drown +# the logs at exactly the moment they matter most. +_SATURATION_WARN_INTERVAL_SECONDS = 30.0 + +_executor: ThreadPoolExecutor | None = None +_executor_lock = threading.Lock() + +_state_lock = threading.Lock() +_in_flight = 0 +_last_saturation_warning = 0.0 + + +def _configured_max_workers() -> int: + """Read the pool size from the environment, tolerating bad input.""" + raw = os.environ.get(_MAX_WORKERS_ENV) + if raw is None or not raw.strip(): + return _DEFAULT_MAX_WORKERS + try: + value = int(raw.strip()) + except (TypeError, ValueError): + logger.warning( + "%s=%r is not an integer; falling back to %d workers", + _MAX_WORKERS_ENV, + raw, + _DEFAULT_MAX_WORKERS, + ) + return _DEFAULT_MAX_WORKERS + if value <= 0: + logger.warning( + "%s=%r must be positive; falling back to %d workers", + _MAX_WORKERS_ENV, + raw, + _DEFAULT_MAX_WORKERS, + ) + return _DEFAULT_MAX_WORKERS + return value + + +def get_blocking_io_executor() -> ThreadPoolExecutor: + """Return the process-wide executor dedicated to blocking I/O. + + Built lazily under a lock so concurrent first-callers cannot race two pools + into existence. + """ + global _executor + if _executor is None: + with _executor_lock: + if _executor is None: + _executor = ThreadPoolExecutor( + max_workers=_configured_max_workers(), + thread_name_prefix=_THREAD_NAME_PREFIX, + ) + return _executor + + +def reset_blocking_io_executor() -> None: + """Dispose of the current pool so the next call rebuilds it. + + Intended for tests and for process lifecycle hooks. ``wait=False`` because a + genuinely stuck worker is precisely what this module exists to contain — + blocking shutdown on it would reintroduce the hang. + """ + global _executor, _in_flight, _last_saturation_warning + with _executor_lock: + previous, _executor = _executor, None + with _state_lock: + _in_flight = 0 + _last_saturation_warning = 0.0 + if previous is not None: + previous.shutdown(wait=False) + + +def _enter(max_workers: int) -> None: + """Account for a submission and warn (throttled) once the pool is oversubscribed.""" + global _in_flight, _last_saturation_warning + with _state_lock: + _in_flight += 1 + in_flight = _in_flight + if in_flight <= max_workers: + return + now = time.monotonic() + if now - _last_saturation_warning < _SATURATION_WARN_INTERVAL_SECONDS: + return + _last_saturation_warning = now + logger.warning( + "blocking I/O pool saturated: %d in flight for %d workers; %d call(s) " + "queued. A stalled read holds its worker until it returns.", + in_flight, + max_workers, + in_flight - max_workers, + extra={ + "in_flight": in_flight, + "max_workers": max_workers, + "queued": in_flight - max_workers, + }, + ) + + +def _exit() -> None: + global _in_flight + with _state_lock: + _in_flight = max(0, _in_flight - 1) + + +async def run_blocking( + func: Callable[..., T], + *args: Any, + timeout: float | None = None, + **kwargs: Any, +) -> T: + """Run ``func`` on the dedicated blocking-I/O pool. + + Args: + func: Blocking callable. Must be safe to run off the event loop. + *args: Positional arguments for ``func``. + timeout: Optional caller-side deadline in seconds. Bounds *this + coroutine* only — see the module docstring: a timed-out worker keeps + running until its syscall returns. Isolation, not the timeout, is + what prevents starvation. + **kwargs: Keyword arguments for ``func``. ``run_in_executor`` accepts + none, so they are bound with ``functools.partial``. + + Raises: + asyncio.TimeoutError: If ``timeout`` elapses first. + Exception: Whatever ``func`` raised, unwrapped and unmodified. + """ + executor = get_blocking_io_executor() + call = functools.partial(func, *args, **kwargs) + _enter(executor._max_workers) # noqa: SLF001 - stdlib exposes no public getter + + # Submit to the executor directly rather than via ``run_in_executor`` so the + # completion callback can be attached to the *concurrent* future. That makes + # ``_in_flight`` track real worker occupancy: a caller released by ``timeout`` + # leaves its worker running, and this keeps counting it. Attaching to the + # asyncio future instead would decrement on cancellation and hide exactly the + # leaked slots this instrumentation exists to reveal. + try: + worker_future = executor.submit(call) + except RuntimeError: + _exit() + raise + worker_future.add_done_callback(lambda _f: _exit()) + + awaitable = asyncio.wrap_future(worker_future) + if timeout is None: + return await awaitable + return await asyncio.wait_for(awaitable, timeout=timeout) diff --git a/tests/unit/test_blocking_io.py b/tests/unit/test_blocking_io.py new file mode 100644 index 000000000..8110f028f --- /dev/null +++ b/tests/unit/test_blocking_io.py @@ -0,0 +1,229 @@ +"""Unit tests for utils/blocking_io.py — executor isolation for blocking I/O. + +Covers issue #1234: every ``asyncio.to_thread`` / ``run_in_executor(None, ...)`` +call in the process shares one bounded default ``ThreadPoolExecutor``. A blocking +read that never returns (NFS/FUSE) permanently leaks a worker slot, because +cancelling the awaiting coroutine does *not* reclaim the thread. Enough +concurrent stalls exhaust the pool and unrelated subsystems stop making progress. + +The tests below pin *isolation* in both directions. Off-loop behaviour is already +covered by the provider suites; running off the loop but still on the shared pool +is exactly the failure mode #1234 describes, so "off the loop" is not sufficient. +""" + +from __future__ import annotations + +import asyncio +import logging +import sys +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +_SRC = Path(__file__).resolve().parents[2] / "src" +sys.path.insert(0, str(_SRC)) + +from youtube_extension.utils.blocking_io import ( # noqa: E402 + get_blocking_io_executor, + reset_blocking_io_executor, + run_blocking, +) + +# Bound every wait so a regression fails the suite instead of hanging it. +_TIMEOUT = 10.0 + + +@pytest.fixture(autouse=True) +def _isolated_executor(monkeypatch): + """Give each test a small, private I/O pool so saturation is exact.""" + monkeypatch.setenv("BLOCKING_IO_MAX_WORKERS", "2") + reset_blocking_io_executor() + yield + reset_blocking_io_executor() + + +class _Saturator: + """Occupies every worker of a pool until released. + + ``entered`` is released once per worker that has actually started running, + so a test can wait for genuine saturation instead of sleeping and hoping. + """ + + def __init__(self, workers: int): + self.workers = workers + self.entered = threading.Semaphore(0) + self.release = threading.Event() + + def hog(self) -> str: + self.entered.release() + # Bounded so a bug can never wedge the suite; far longer than any assert. + self.release.wait(_TIMEOUT * 3) + return "done" + + async def wait_until_saturated(self) -> None: + """Wait for genuine saturation without starving the event loop. + + Must yield between polls: the hogs are scheduled as tasks, and a task + does not begin executing until the loop gets control. A *blocking* + acquire here would deadlock the harness before any hog reached the pool + and produce a failure that looks like a product bug but is not one. + """ + deadline = time.monotonic() + _TIMEOUT + started = 0 + while started < self.workers: + if self.entered.acquire(blocking=False): + started += 1 + continue + assert time.monotonic() < deadline, ( + f"only {started}/{self.workers} workers started; the pool is " + "smaller than expected or the harness is broken" + ) + await asyncio.sleep(0.01) + + def free(self) -> None: + self.release.set() + + +# --------------------------------------------------------------------------- +# AC2 (literal): saturating the I/O pool must not block unrelated to_thread work +# --------------------------------------------------------------------------- + + +class TestIoPoolSaturationDoesNotStarveOthers: + async def test_unrelated_to_thread_completes_while_io_pool_is_saturated(self): + pool = get_blocking_io_executor() + capacity = pool._max_workers # noqa: SLF001 - asserting the bound we set + assert capacity == 2 + + sat = _Saturator(capacity) + hogs = [asyncio.create_task(run_blocking(sat.hog)) for _ in range(capacity)] + try: + await sat.wait_until_saturated() + + # The I/O pool now has zero free workers. An unrelated subsystem + # using the default executor must be completely unaffected. + result = await asyncio.wait_for( + asyncio.to_thread(lambda: "unrelated-ok"), timeout=_TIMEOUT + ) + assert result == "unrelated-ok" + finally: + sat.free() + await asyncio.gather(*hogs) + + async def test_io_offloads_do_not_run_on_the_default_executor(self): + """Direct structural check: the worker thread is not a default-pool thread.""" + loop = asyncio.get_running_loop() + previous = getattr(loop, "_default_executor", None) + default_pool = ThreadPoolExecutor( + max_workers=1, thread_name_prefix="default-probe" + ) + loop.set_default_executor(default_pool) + try: + default_thread = await asyncio.to_thread(threading.current_thread) + io_thread = await run_blocking(threading.current_thread) + + assert io_thread is not default_thread + assert not io_thread.name.startswith("default-probe"), ( + "blocking I/O ran on the shared default executor; it must use " + "the dedicated pool so a stall cannot starve other subsystems" + ) + finally: + if previous is None: + # ``set_default_executor`` rejects ``None``; restore directly. + loop._default_executor = None # noqa: SLF001 + else: + loop.set_default_executor(previous) + default_pool.shutdown(wait=False) + + +# --------------------------------------------------------------------------- +# Helper contract +# --------------------------------------------------------------------------- + + +class TestRunBlockingContract: + async def test_runs_off_the_event_loop_thread(self): + loop_thread = threading.get_ident() + worker_thread = await run_blocking(threading.get_ident) + assert worker_thread != loop_thread + + async def test_supports_keyword_arguments(self): + """``run_in_executor`` takes no kwargs; the helper must bridge that.""" + + def _join(a, b, sep="-"): + return f"{a}{sep}{b}" + + assert await run_blocking(_join, "x", "y", sep="+") == "x+y" + + async def test_propagates_exceptions_unchanged(self): + def _boom(): + raise FileNotFoundError("missing") + + with pytest.raises(FileNotFoundError, match="missing"): + await run_blocking(_boom) + + async def test_timeout_releases_the_caller(self): + """A deadline bounds the *caller*. It cannot reclaim the worker — that is + precisely why isolation, not timeouts, is the fix — but an unbounded + caller hang is still worth preventing.""" + sat = _Saturator(1) + task = asyncio.create_task(run_blocking(sat.hog, timeout=0.25)) + try: + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(task, timeout=_TIMEOUT) + finally: + sat.free() + + def test_pool_size_is_configurable_and_bounded(self, monkeypatch): + monkeypatch.setenv("BLOCKING_IO_MAX_WORKERS", "5") + reset_blocking_io_executor() + assert get_blocking_io_executor()._max_workers == 5 # noqa: SLF001 + + def test_invalid_pool_size_falls_back_to_default(self, monkeypatch): + monkeypatch.setenv("BLOCKING_IO_MAX_WORKERS", "not-a-number") + reset_blocking_io_executor() + assert get_blocking_io_executor()._max_workers > 0 # noqa: SLF001 + + def test_executor_is_a_process_wide_singleton(self): + assert get_blocking_io_executor() is get_blocking_io_executor() + + +# --------------------------------------------------------------------------- +# AC3: saturation must be observable +# --------------------------------------------------------------------------- + + +class TestSaturationIsObservable: + async def test_warns_when_the_pool_is_saturated(self, caplog): + pool = get_blocking_io_executor() + capacity = pool._max_workers # noqa: SLF001 + sat = _Saturator(capacity) + hogs = [asyncio.create_task(run_blocking(sat.hog)) for _ in range(capacity)] + try: + await sat.wait_until_saturated() + with caplog.at_level(logging.WARNING, logger="youtube_extension.utils.blocking_io"): + extra = asyncio.create_task(run_blocking(lambda: "queued")) + await asyncio.sleep(0) # let the submit path run + sat.free() + assert await asyncio.wait_for(extra, timeout=_TIMEOUT) == "queued" + + saturation_records = [ + r for r in caplog.records if "saturat" in r.getMessage().lower() + ] + assert saturation_records, ( + "pool saturation produced no warning; exhaustion must be " + "observable before it becomes an outage" + ) + record = saturation_records[0] + assert getattr(record, "max_workers", None) == capacity + finally: + sat.free() + await asyncio.gather(*hogs) + + async def test_no_warning_when_pool_is_idle(self, caplog): + with caplog.at_level(logging.WARNING, logger="youtube_extension.utils.blocking_io"): + await run_blocking(lambda: None) + assert not [r for r in caplog.records if "saturat" in r.getMessage().lower()] diff --git a/tests/unit/test_vision_provider_executor_isolation.py b/tests/unit/test_vision_provider_executor_isolation.py new file mode 100644 index 000000000..a053d9c23 --- /dev/null +++ b/tests/unit/test_vision_provider_executor_isolation.py @@ -0,0 +1,229 @@ +"""Executor-isolation regression tests for the three vision providers (#1234). + +The providers already read local image bytes off the event loop (#1232/#1233). +That is necessary but not sufficient: the read still lands on the *shared* +default ``ThreadPoolExecutor``, so it shares fate with every other +``asyncio.to_thread`` / ``run_in_executor(None, ...)`` caller in the process +(96 call sites at time of writing, on a pool of ``min(32, cpu + 4)``). + +Each test here saturates the **default** executor and then asserts the provider's +local image read still completes. Against the pre-#1234 code these fail with +``TimeoutError`` — the read queues behind unrelated work and never runs. That is +the starvation #1234 describes, expressed as an executable assertion. + +Deliberately complementary to the existing ``*_runs_on_worker_thread`` tests: +those pin *where* the read runs, these pin *whose pool it competes for*. +""" + +from __future__ import annotations + +import asyncio +import sys +import threading +import time +import types as _types +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +_SRC = Path(__file__).resolve().parents[2] / "src" +sys.path.insert(0, str(_SRC)) + +from youtube_extension.integrations.cloud_ai.base import AnalysisType # noqa: E402 +from youtube_extension.integrations.cloud_ai.providers.aws_rekognition import ( # noqa: E402 + AWSRekognition, +) +from youtube_extension.integrations.cloud_ai.providers.azure_vision import ( # noqa: E402 + AzureVision, +) +from youtube_extension.integrations.cloud_ai.providers.google_cloud import ( # noqa: E402 + GoogleCloudAI, +) +from youtube_extension.utils.blocking_io import reset_blocking_io_executor # noqa: E402 + +_TIMEOUT = 5.0 +_DEFAULT_POOL_WORKERS = 2 + +AWS_CONFIG = { + "aws_access_key_id": "test-access-key-id", + "aws_secret_access_key": "test-secret-access-key", + "region": "us-east-1", +} +AZURE_CONFIG = { + "subscription_key": "test-key-abc", + "endpoint": "https://eastus.api.cognitive.microsoft.com/", +} +GCP_CONFIG = {"project_id": "my-gcp-project"} + + +class _DefaultExecutorSaturator: + """Fills the loop's default executor, leaving zero free workers. + + Shrinking the default pool first makes saturation exact and machine + independent — otherwise the number of hogs needed depends on ``cpu_count``. + """ + + def __init__(self, loop, workers: int = _DEFAULT_POOL_WORKERS): + self._loop = loop + self._workers = workers + self._previous = getattr(loop, "_default_executor", None) + self._pool = ThreadPoolExecutor( + max_workers=workers, thread_name_prefix="saturated-default" + ) + self._entered = threading.Semaphore(0) + self._release = threading.Event() + self._tasks: list[asyncio.Task] = [] + + def _hog(self) -> None: + self._entered.release() + self._release.wait(_TIMEOUT * 3) + + async def __aenter__(self): + self._loop.set_default_executor(self._pool) + self._tasks = [ + asyncio.create_task(asyncio.to_thread(self._hog)) + for _ in range(self._workers) + ] + # Must yield: a freshly created task does not run until the loop gets + # control, so a *blocking* acquire here would deadlock the harness + # before a single hog ever reached the pool. + deadline = time.monotonic() + _TIMEOUT + entered = 0 + while entered < self._workers: + if self._entered.acquire(blocking=False): + entered += 1 + continue + assert time.monotonic() < deadline, ( + f"default executor never saturated ({entered}/{self._workers} hogs " + "started); the harness is broken, not the code under test" + ) + await asyncio.sleep(0.01) + return self + + async def __aexit__(self, *exc_info): + self._release.set() + if self._tasks: + await asyncio.gather(*self._tasks, return_exceptions=True) + if self._previous is None: + # No default executor existed before us; ``set_default_executor`` + # rejects ``None``, so restore the pristine state directly. + self._loop._default_executor = None # noqa: SLF001 + else: + self._loop.set_default_executor(self._previous) + self._pool.shutdown(wait=False) + return False + + async def assert_default_pool_is_blocked(self): + """Positive control: prove the saturation actually bites. + + Without this, a provider test could pass simply because the pool was + never really full, and the whole file would be vacuous. + """ + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(asyncio.to_thread(lambda: "x"), timeout=0.5) + + +@pytest.fixture(autouse=True) +def _fresh_io_pool(): + reset_blocking_io_executor() + yield + reset_blocking_io_executor() + + +class TestVisionReadsSurviveDefaultExecutorSaturation: + async def test_aws_local_read_is_not_starved(self, tmp_path): + image = tmp_path / "frame.jpg" + image.write_bytes(b"AWS-PAYLOAD") + provider = AWSRekognition(AWS_CONFIG) + + async with _DefaultExecutorSaturator(asyncio.get_running_loop()) as sat: + await sat.assert_default_pool_is_blocked() + result = await asyncio.wait_for( + provider._prepare_image_input(str(image)), timeout=_TIMEOUT + ) + + assert result == {"Bytes": b"AWS-PAYLOAD"} + + async def test_azure_local_read_is_not_starved(self, tmp_path): + image = tmp_path / "frame.jpg" + image.write_bytes(b"AZURE-PAYLOAD") + provider = AzureVision(AZURE_CONFIG) + + async with _DefaultExecutorSaturator(asyncio.get_running_loop()) as sat: + await sat.assert_default_pool_is_blocked() + result = await asyncio.wait_for( + provider._prepare_image_input(str(image)), timeout=_TIMEOUT + ) + + assert result == b"AZURE-PAYLOAD" + + async def test_google_local_read_is_not_starved(self, tmp_path): + image = tmp_path / "frame.jpg" + image.write_bytes(b"GCP-PAYLOAD") + provider = GoogleCloudAI(GCP_CONFIG) + + image_instance = MagicMock() + image_instance.source = MagicMock() + feature_type = MagicMock() + feature_type.LABEL_DETECTION = "LABEL_DETECTION" + feature_cls = MagicMock() + feature_cls.Type = feature_type + mock_vision = MagicMock() + mock_vision.Image = MagicMock(return_value=image_instance) + mock_vision.Feature = feature_cls + + client = AsyncMock() + client.annotate_image = AsyncMock( + return_value=MagicMock( + label_annotations=[], + localized_object_annotations=[], + text_annotations=[], + face_annotations=[], + safe_search_annotation=None, + error=MagicMock(message=""), + ) + ) + provider._vision_client = client + + patched_modules = patch.dict( + "sys.modules", + { + "google": _types.ModuleType("google"), + "google.cloud": _types.ModuleType("google.cloud"), + "google.cloud.vision": mock_vision, + }, + ) + + async with _DefaultExecutorSaturator(asyncio.get_running_loop()) as sat: + await sat.assert_default_pool_is_blocked() + with patched_modules: + await asyncio.wait_for( + provider.analyze_image( + str(image), [AnalysisType.LABEL_DETECTION] + ), + timeout=_TIMEOUT, + ) + + assert image_instance.content == b"GCP-PAYLOAD" + + async def test_http_url_branch_still_performs_no_disk_read(self, tmp_path): + """Isolation must not alter URL handling — Azure fetches those itself.""" + provider = AzureVision(AZURE_CONFIG) + async with _DefaultExecutorSaturator(asyncio.get_running_loop()): + result = await asyncio.wait_for( + provider._prepare_image_input("https://example.com/img.jpg"), + timeout=_TIMEOUT, + ) + assert result is None + + async def test_missing_file_still_raises_file_not_found(self, tmp_path): + """Routing through a different executor must not swallow I/O errors.""" + provider = AzureVision(AZURE_CONFIG) + async with _DefaultExecutorSaturator(asyncio.get_running_loop()): + with pytest.raises(FileNotFoundError): + await asyncio.wait_for( + provider._prepare_image_input(str(tmp_path / "nope.jpg")), + timeout=_TIMEOUT, + )