From 14cf5e31ff87500766608c8f4ef27a648da2879b Mon Sep 17 00:00:00 2001 From: ldm0 Date: Wed, 21 Oct 2026 15:37:07 +0800 Subject: [PATCH] fix(frontend-smoke): honor observation timeout during CDP setup --- moli-frontend-smoke/README.md | 5 +- .../moli_frontend_smoke/cdp.py | 51 ++++++----- .../moli_frontend_smoke/runner.py | 4 +- .../tests/test_cdp_timeouts.py | 85 +++++++++++++++++++ moli-frontend-smoke/tests/test_runner.py | 8 +- 5 files changed, 128 insertions(+), 25 deletions(-) create mode 100644 moli-frontend-smoke/tests/test_cdp_timeouts.py diff --git a/moli-frontend-smoke/README.md b/moli-frontend-smoke/README.md index 91fcc141a5..a34ae78278 100644 --- a/moli-frontend-smoke/README.md +++ b/moli-frontend-smoke/README.md @@ -96,7 +96,10 @@ concurrency, but a release conclusion should also repeat the reference run and c hash across scheduling modes. Each worker keeps one browser-level CDP connection and creates an isolated browser context per -case. CI may pass `--reference-infrastructure-retries 1` to re-observe a case after a Chromium +case. Discovery, the WebSocket handshake, and context/target setup use the configured observation +timeout plus protocol headroom, just like the later observation commands. Discovery and the +handshake share one connection deadline. Timeout diagnostics identify the command or event that +was pending. CI may pass `--reference-infrastructure-retries 1` to re-observe a case after a Chromium connection or protocol failure. The retry does not apply to fixture errors, normalization errors, Moli failures, or DOM differences, and the original failure remains recorded in `summary.json`. diff --git a/moli-frontend-smoke/moli_frontend_smoke/cdp.py b/moli-frontend-smoke/moli_frontend_smoke/cdp.py index e5492dcd3d..5ba3fd6e3f 100644 --- a/moli-frontend-smoke/moli_frontend_smoke/cdp.py +++ b/moli-frontend-smoke/moli_frontend_smoke/cdp.py @@ -31,17 +31,19 @@ class CdpCommandError(CdpError): _DOM_ENABLE_PARAMS = {"includeWhitespace": "all"} -def _read_json_url(url: str) -> dict[str, Any]: +def _read_json_url(url: str, *, timeout: float = 2.0) -> dict[str, Any]: opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) - with opener.open(url, timeout=2) as response: + with opener.open(url, timeout=timeout) as response: value = json.loads(response.read().decode("utf-8")) if not isinstance(value, dict): raise CdpError(f"unexpected CDP discovery response: {value!r}") return value -async def discover_websocket_url(endpoint: str) -> str: - payload = await asyncio.to_thread(_read_json_url, endpoint.rstrip("/") + "/json/version") +async def discover_websocket_url(endpoint: str, *, timeout: float = 2.0) -> str: + payload = await asyncio.to_thread( + _read_json_url, endpoint.rstrip("/") + "/json/version", timeout=timeout, + ) url = payload.get("webSocketDebuggerUrl") if not isinstance(url, str) or not url: raise CdpError(f"CDP discovery did not include webSocketDebuggerUrl: {payload}") @@ -90,11 +92,15 @@ class RawCdpClient: ) -> dict[str, Any]: message_id = await self.send(method, params, session_id=session_id) deadline = asyncio.get_running_loop().time() + timeout + timeout_message = f"timed out waiting for {method} response after {timeout:g}s" while True: remaining = deadline - asyncio.get_running_loop().time() if remaining <= 0: - raise CdpError(f"timed out waiting for {method} response") - message = await asyncio.wait_for(self.recv(), timeout=remaining) + raise CdpError(timeout_message) + try: + message = await asyncio.wait_for(self.recv(), timeout=remaining) + except TimeoutError as error: + raise CdpError(timeout_message) from error if message.get("id") == message_id: if "error" in message: raise CdpCommandError(method, message["error"]) @@ -121,24 +127,29 @@ class RawCdpClient: if matches(message): return message deadline = asyncio.get_running_loop().time() + timeout + timeout_message = f"timed out waiting for {method} event after {timeout:g}s" while True: remaining = deadline - asyncio.get_running_loop().time() if remaining <= 0: - raise CdpError(f"timed out waiting for {method}") - message = await asyncio.wait_for(self.recv(), timeout=remaining) + raise CdpError(timeout_message) + try: + message = await asyncio.wait_for(self.recv(), timeout=remaining) + except TimeoutError as error: + raise CdpError(timeout_message) from error self.events.append(message) if matches(message): return message -async def connect(endpoint: str) -> RawCdpClient: - url = await discover_websocket_url(endpoint) - websocket = await websockets.connect( - url, - open_timeout=5, - max_size=32 * 1024 * 1024, - proxy=None, - ) +async def connect(endpoint: str, *, timeout: float = 10.0) -> RawCdpClient: + async with asyncio.timeout(timeout): + url = await discover_websocket_url(endpoint, timeout=timeout) + websocket = await websockets.connect( + url, + open_timeout=timeout, + max_size=32 * 1024 * 1024, + proxy=None, + ) return RawCdpClient(websocket=websocket) @@ -425,18 +436,20 @@ async def observe_case( frames: list[DomFrameObservation] = [] final_root: dict[str, Any] | None = None diagnostics: dict[str, Any] = {} + command_timeout = max(10.0, timeout_ms / 1000 + 2.0) try: if client is None: - client = await connect(endpoint) + client = await connect(endpoint, timeout=command_timeout) else: client.events.clear() - context = await client.command("Target.createBrowserContext") + context = await client.command("Target.createBrowserContext", timeout=command_timeout) context_id = context.get("browserContextId") if not isinstance(context_id, str) or not context_id: raise CdpError(f"Target.createBrowserContext returned {context}") target = await client.command( "Target.createTarget", {"browserContextId": context_id, "url": "about:blank"}, + timeout=command_timeout, ) target_id = target.get("targetId") if not isinstance(target_id, str) or not target_id: @@ -444,12 +457,12 @@ async def observe_case( attached = await client.command( "Target.attachToTarget", {"targetId": target_id, "flatten": True}, + timeout=command_timeout, ) session_id = attached.get("sessionId") if not isinstance(session_id, str) or not session_id: raise CdpError(f"Target.attachToTarget returned {attached}") - command_timeout = max(10.0, timeout_ms / 1000 + 2.0) await client.command("Runtime.enable", session_id=session_id, timeout=command_timeout) await client.command("Page.enable", session_id=session_id, timeout=command_timeout) await client.command( diff --git a/moli-frontend-smoke/moli_frontend_smoke/runner.py b/moli-frontend-smoke/moli_frontend_smoke/runner.py index b9221d3582..00222e0a7f 100644 --- a/moli-frontend-smoke/moli_frontend_smoke/runner.py +++ b/moli-frontend-smoke/moli_frontend_smoke/runner.py @@ -296,7 +296,9 @@ async def _observe_phase_case( if client is None: connect_started = time.perf_counter() try: - client = await connect(endpoint) + client = await connect( + endpoint, timeout=max(10.0, timeout_ms / 1000 + 2.0), + ) except Exception as error: observation = EngineObservation( engine=engine, diff --git a/moli-frontend-smoke/tests/test_cdp_timeouts.py b/moli-frontend-smoke/tests/test_cdp_timeouts.py new file mode 100644 index 0000000000..cc9cb35777 --- /dev/null +++ b/moli-frontend-smoke/tests/test_cdp_timeouts.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import asyncio +from contextlib import suppress +import json + +import pytest +from websockets.asyncio.server import serve + +from moli_frontend_smoke.cdp import CdpError, RawCdpClient, connect + + +@pytest.mark.parametrize("timeout", [0.25, 5.0]) +def test_connection_deadline_covers_discovery_and_websocket_handshake(timeout: float) -> None: + async def scenario() -> None: + async def handshake(_connection, _request): + await asyncio.sleep(0.15) + + async def websocket_handler(websocket): + await websocket.wait_closed() + + async with serve( + websocket_handler, "127.0.0.1", 0, process_request=handshake, + ) as websocket_server: + port = websocket_server.sockets[0].getsockname()[1] + websocket_url = f"ws://127.0.0.1:{port}/devtools/browser/probe" + handlers: set[asyncio.Task] = set() + + async def discovery(reader, writer): + task = asyncio.current_task() + handlers.add(task) + try: + await reader.readuntil(b"\r\n\r\n") + await asyncio.sleep(0.15) + body = json.dumps({"webSocketDebuggerUrl": websocket_url}).encode() + writer.write( + b"HTTP/1.1 200 OK\r\nConnection: close\r\n" + + f"Content-Length: {len(body)}\r\n\r\n".encode() + + body + ) + await writer.drain() + finally: + writer.close() + with suppress(ConnectionError): + await writer.wait_closed() + handlers.discard(task) + + async with await asyncio.start_server(discovery, "127.0.0.1", 0) as server: + port = server.sockets[0].getsockname()[1] + endpoint = f"http://127.0.0.1:{port}" + if timeout < 0.3: + # Each phase fits individually, but their combined duration + # exceeds the one connection deadline. + with pytest.raises(TimeoutError): + await connect(endpoint, timeout=timeout) + else: + client = await connect(endpoint, timeout=timeout) + await client.websocket.close() + await asyncio.gather(*handlers) + + asyncio.run(scenario()) + + +class _SilentSocket: + async def send(self, _message): + pass + + async def recv(self): + await asyncio.Event().wait() + + +@pytest.mark.parametrize("operation", ["command", "event"]) +def test_timeout_diagnostic_identifies_the_pending_operation(operation: str) -> None: + async def scenario() -> None: + client = RawCdpClient(websocket=_SilentSocket()) # type: ignore[arg-type] + if operation == "command": + with pytest.raises(CdpError, match="Target.createTarget response after 0.01s"): + await client.command("Target.createTarget", timeout=0.01) + else: + with pytest.raises(CdpError, match="Page.loadEventFired event after 0.01s"): + await client.wait_event( + "Page.loadEventFired", session_id="page", timeout=0.01, + ) + + asyncio.run(scenario()) diff --git a/moli-frontend-smoke/tests/test_runner.py b/moli-frontend-smoke/tests/test_runner.py index 49db3f1632..90f8e9cf2a 100644 --- a/moli-frontend-smoke/tests/test_runner.py +++ b/moli-frontend-smoke/tests/test_runner.py @@ -106,7 +106,7 @@ def test_observation_worker_reuses_one_cdp_connection(monkeypatch) -> None: clients: list[_FakeClient] = [] observed_clients: list[_FakeClient] = [] - async def fake_connect(_endpoint): + async def fake_connect(_endpoint, **_kwargs): client = _FakeClient() clients.append(client) return client @@ -141,7 +141,7 @@ def test_chromium_retries_only_infrastructure_failures(monkeypatch) -> None: clients: list[_FakeClient] = [] attempts = 0 - async def fake_connect(_endpoint): + async def fake_connect(_endpoint, **_kwargs): client = _FakeClient() clients.append(client) return client @@ -188,7 +188,7 @@ def test_chromium_retries_only_infrastructure_failures(monkeypatch) -> None: def test_fixture_failure_is_never_retried(monkeypatch) -> None: attempts = 0 - async def fake_connect(_endpoint): + async def fake_connect(_endpoint, **_kwargs): return _FakeClient() async def fake_observe_case(*, engine, **_kwargs): @@ -228,7 +228,7 @@ def test_fixture_failure_is_never_retried(monkeypatch) -> None: def test_moli_failure_is_never_retried(monkeypatch) -> None: attempts = 0 - async def fake_connect(_endpoint): + async def fake_connect(_endpoint, **_kwargs): return _FakeClient() async def fake_observe_case(*, engine, **_kwargs):