diff --git a/tests/app/routers/test_event_loop_blocking.py b/tests/app/routers/test_event_loop_blocking.py index a50fde4a5e0..3acc424a32a 100644 --- a/tests/app/routers/test_event_loop_blocking.py +++ b/tests/app/routers/test_event_loop_blocking.py @@ -7,13 +7,24 @@ dead rather than as a slow search. Declaring these routes `def` hands them to Starlette's threadpool instead, leaving the -loop free. These tests pin that property down: the slowness is simulated with a -synchronous sleep in the service layer, so they assert the *dispatch mechanism* rather -than any particular query's speed, and stay fast and deterministic in CI. +loop free. These tests pin that property down deterministically: the service layer blocks +on an Event the TEST controls (released only after the probe completes), so the assertion +is causal rather than a wall-clock latency threshold. An earlier version asserted +"probe elapsed < BLOCKING_SECONDS / 2", which a descheduled CI runner could fail with a +free loop (observed: a 1.09s probe on a loaded macOS runner) - machine noise and a blocked +loop produced the same timing signature. Here, machine noise cannot fail the test: the +handler cannot finish before the probe unless loop dispatch is broken, because only the +test releases it. + +The failure mode is equally deterministic: an `async def` regression blocks the loop +inside the handler's wait, the probe cannot be answered until that wait times out +(FAILURE_TIMEOUT_SECONDS, paid only when the guard actually catches a regression), and by +then the handler has finished - the finished-flag assertion fires. """ import asyncio -import time +import threading +from dataclasses import dataclass, field from unittest.mock import MagicMock import pytest @@ -26,11 +37,13 @@ from invokeai.app.services.session_queue.session_queue_common import SessionQueueItemSummary from invokeai.app.services.shared.pagination import OffsetPaginatedResults -# Long enough that a blocked event loop is unmistakable, short enough to keep the suite fast. -BLOCKING_SECONDS = 1.0 +# Upper bound on the blocked-handler wait. The healthy path never waits on it (the test +# releases the handler as soon as the probe returns); it only bounds how long a genuine +# `async def` regression takes to surface as a failure. +FAILURE_TIMEOUT_SECONDS = 30.0 # A trivial route with no auth dependency and no database access. If the loop is free, this -# answers in single-digit milliseconds no matter what else the server is doing. +# answers no matter what else the server is doing. PROBE_ROUTE = "/api/v1/app/version" @@ -39,9 +52,33 @@ def anyio_backend() -> str: return "asyncio" +@dataclass +class BlockingServices: + """The patched invoker plus the test's handles on its blocking behavior. + + - `started` is set the moment a slow handler begins executing (wherever that is: the + threadpool if the route is a healthy `def`, the event loop if it regressed). + - `release` is set by the test to let the handler return. + - `finished` is set when the handler's wait ends - via `release`, or via the + FAILURE_TIMEOUT_SECONDS backstop when a blocked loop prevented the release. + """ + + invoker: MagicMock + started: threading.Event = field(default_factory=threading.Event) + release: threading.Event = field(default_factory=threading.Event) + finished: threading.Event = field(default_factory=threading.Event) + + def block(self) -> None: + self.started.set() + try: + self.release.wait(timeout=FAILURE_TIMEOUT_SECONDS) + finally: + self.finished.set() + + @pytest.fixture -def blocking_invoker(monkeypatch: pytest.MonkeyPatch) -> MagicMock: - """Point every router at services whose gallery/image reads block for BLOCKING_SECONDS. +def blocking_services(monkeypatch: pytest.MonkeyPatch) -> BlockingServices: + """Point every router at services whose gallery/image reads block until released. Patching the attribute on the class itself covers all routers at once - they share the single `ApiDependencies` object rather than importing their own copy. @@ -51,24 +88,26 @@ def blocking_invoker(monkeypatch: pytest.MonkeyPatch) -> MagicMock: # multiuser mode and answer every request with 401 before the route is ever reached. invoker.services.configuration.multiuser = False + services = BlockingServices(invoker=invoker) + def slow_list_item_names(**_: object) -> GalleryItemNamesResult: - time.sleep(BLOCKING_SECONDS) + services.block() return GalleryItemNamesResult(items=[], starred_count=0, total_count=0) def slow_get_image_names(**_: object) -> ImageNamesResult: - time.sleep(BLOCKING_SECONDS) + services.block() return ImageNamesResult(image_names=[], starred_count=0, total_count=0) def slow_list_items(**_: object) -> OffsetPaginatedResults[GalleryItem]: - time.sleep(BLOCKING_SECONDS) + services.block() return OffsetPaginatedResults[GalleryItem](limit=10, offset=0, total=0, items=[]) def slow_get_item_names(**_: object) -> GalleryItemNames: - time.sleep(BLOCKING_SECONDS) + services.block() return GalleryItemNames(item_names=[], starred_count=0, total_count=0) def slow_queue_item_summaries(**_: object) -> list[SessionQueueItemSummary]: - time.sleep(BLOCKING_SECONDS) + services.block() return [] invoker.services.gallery.list_item_names.side_effect = slow_list_item_names @@ -78,31 +117,7 @@ def slow_queue_item_summaries(**_: object) -> list[SessionQueueItemSummary]: invoker.services.session_queue.get_queue_item_summaries_by_ids.side_effect = slow_queue_item_summaries monkeypatch.setattr(ApiDependencies, "invoker", invoker, raising=False) - return invoker - - -async def _probe_latency_while_busy( - client: AsyncClient, slow_route: str, params: dict, json_body: dict | None = None -) -> tuple[float, asyncio.Task]: - """Start `slow_route`, then time a probe request issued while it is still running. - - The clock starts before yielding to the slow request, so a blocked loop shows up as - probe latency even though the probe itself never got a chance to be dispatched. - """ - started = time.perf_counter() - if json_body is None: - slow_request = asyncio.create_task(client.get(slow_route, params=params)) - else: - slow_request = asyncio.create_task(client.post(slow_route, params=params, json=json_body)) - # Hand control to the slow request so it reaches its route handler before we probe. - for _ in range(10): - await asyncio.sleep(0) - - response = await client.get(PROBE_ROUTE) - elapsed = time.perf_counter() - started - - assert response.status_code == 200 - return elapsed, slow_request + return services @pytest.mark.anyio @@ -118,20 +133,39 @@ async def _probe_latency_while_busy( ], ) async def test_slow_gallery_read_leaves_the_event_loop_free( - blocking_invoker: MagicMock, slow_route: str, params: dict, json_body: dict | None + blocking_services: BlockingServices, slow_route: str, params: dict, json_body: dict | None ) -> None: transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: - elapsed, slow_request = await _probe_latency_while_busy(client, slow_route, params, json_body) - - assert elapsed < BLOCKING_SECONDS / 2, ( - f"{PROBE_ROUTE} took {elapsed:.2f}s while {slow_route} was running. The slow route is " - f"executing its blocking database work on the event loop, so the server answers " - f"nothing else until it finishes. Declare the route `def` instead of `async def`." - ) - assert not slow_request.done(), ( - "The slow request finished before the probe was even dispatched, so nothing was " - "measured concurrently - the event loop was blocked for its full duration." - ) - - await slow_request + if json_body is None: + slow_request = asyncio.create_task(client.get(slow_route, params=params)) + else: + slow_request = asyncio.create_task(client.post(slow_route, params=params, json=json_body)) + + try: + # Wait until the slow handler is actually executing before probing, so the probe + # is provably concurrent with it. `to_thread` keeps this wait off the event loop: + # if the handler regressed to running ON the loop, this await can only resume + # after the handler's timeout backstop - which the finished-flag assertion below + # then reports as the failure, rather than the test deadlocking. + handler_started = await asyncio.to_thread(blocking_services.started.wait, FAILURE_TIMEOUT_SECONDS) + assert handler_started, f"{slow_route} never reached its (patched) service call." + + response = await client.get(PROBE_ROUTE) + + assert response.status_code == 200 + assert not blocking_services.finished.is_set(), ( + f"{PROBE_ROUTE} could not be answered until {slow_route}'s handler had finished, and the " + f"handler is only released by this test AFTER the probe - so the route is executing its " + f"blocking database work on the event loop, and the server answers nothing else while it " + f"runs. Declare the route `def` instead of `async def`." + ) + assert not slow_request.done(), ( + "The slow request completed before the probe, so nothing was measured concurrently - " + "the event loop was blocked for its full duration." + ) + finally: + # Whatever happened above, let the handler (and the pending request) finish so + # neither a threadpool worker nor the task outlives the test. + blocking_services.release.set() + await slow_request