diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7073bae773..87f0878e2d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -478,6 +478,11 @@ jobs: uv sync --project moli-cdp-smoke --locked npm ci --prefix moli-cdp-smoke uv run --project moli-cdp-smoke --no-sync playwright install --with-deps chromium + - name: Repeat target semantics for shutdown investigation + run: | + uv run --project moli-cdp-smoke --no-sync moli-cdp-smoke \ + --browser chromium --group target-semantics --repeat 20 --jobs 4 \ + --output-dir target/smoke/cdp-chromium-ci/shutdown-repeat - name: Run full CDP smoke on Chromium run: | # Piped core handlers ignore ulimit, and can block Page.crash teardown. diff --git a/moli-cdp-smoke/README.md b/moli-cdp-smoke/README.md index ad7daee675..894dc9cfd4 100644 --- a/moli-cdp-smoke/README.md +++ b/moli-cdp-smoke/README.md @@ -29,6 +29,13 @@ observes the parent's return code independently of stdout/stderr EOF: a descendant may retain an inherited output pipe after the browser has exited, which can leave `asyncio.subprocess.Process.wait()` pending. Failures identify the session case and lifecycle stage, including acknowledgment and process exit. +Failed cases retain their close response, completed cases, browser log tail, and +connection state in the worker's `observed` result before any socket or process +cleanup. On Linux they also capture the owned PID's status, thread wait channels +and syscalls, and descendant process states. This distinguishes a live or stuck +browser from an already reaped browser whose descendants still own output pipes; +diagnostic collection does not relax the natural-exit deadline or count forced +cleanup as a successful exit. Lifecycle artifacts retain total FDs for both engines. Chromium also records FDs for regular files inside its owned profile: delayed database initialization diff --git a/moli-cdp-smoke/moli_cdp_smoke/groups/browser_close.py b/moli-cdp-smoke/moli_cdp_smoke/groups/browser_close.py index 8f1bb7e080..749c51571f 100644 --- a/moli-cdp-smoke/moli_cdp_smoke/groups/browser_close.py +++ b/moli-cdp-smoke/moli_cdp_smoke/groups/browser_close.py @@ -5,7 +5,7 @@ import json from typing import Any from ..assertions import SmokeError, assert_equal, wait_until -from ..process import wait_for_process_exit +from ..process import process_exit_diagnostics, wait_for_process_exit from ..raw_cdp import RawCdpClient, connect_raw_cdp, connect_raw_cdp_websocket from ..serve import browser_identity, start_browser, stop_browser, wait_for_browser_endpoint @@ -13,6 +13,12 @@ from ..serve import browser_identity, start_browser, stop_browser, wait_for_brow _OMITTED = object() +class BrowserCloseError(SmokeError): + def __init__(self, message: str, observed: dict[str, Any]) -> None: + super().__init__(message) + self.observed = observed + + async def _command( client: RawCdpClient, method: str, params: dict[str, Any] | None = None, *, session_id: str | None = None, @@ -134,6 +140,8 @@ async def browser_close_instance_lifecycle(endpoint: str, fixture: str) -> dict[ serving = await start_browser() clients: list[RawCdpClient] = [] stage = "startup" + child_endpoint = None + response = None try: child_endpoint = await wait_for_browser_endpoint(serving) await browser_identity(child_endpoint, identity["engine"]) @@ -168,7 +176,32 @@ async def browser_close_instance_lifecycle(endpoint: str, fixture: str) -> dict[ assert_equal(code, 0, "natural process exit") observed.append({"case": name, "response": response, "exitCode": code, "rejected": rejected}) except Exception as error: - raise SmokeError(f"{name} ({stage}): {type(error).__name__}: {error}") from error + # Keep the pre-cleanup state: closing the clients or sending SIGTERM + # first would conceal whether the browser itself was still alive. + diagnostics = process_exit_diagnostics(serving.process) + observed_failure = { + "browser": identity, + "completedCases": observed, + "failedCase": { + "case": name, "scope": scope, "stage": stage, + "endpoint": child_endpoint, "closeResponse": response, + "process": diagnostics, + "connections": [ + {"state": client.websocket.state.name, + "closeCode": client.websocket.close_code, + "closeReason": client.websocket.close_reason} + for client in clients + ], + "browserLogTail": serving.logs[-80:], + }, + } + process_state = diagnostics.get("process", {}) + raise BrowserCloseError( + f"{name} ({stage}): {type(error).__name__}: {error}; " + f"pid={serving.process.pid} returncode={diagnostics['returncode']} " + f"state={process_state.get('State')} wchan={process_state.get('wchan')}", + observed_failure, + ) from error finally: await asyncio.gather( *(client.websocket.close() for client in clients), return_exceptions=True, diff --git a/moli-cdp-smoke/moli_cdp_smoke/groups/target_semantics.py b/moli-cdp-smoke/moli_cdp_smoke/groups/target_semantics.py index 13b5566117..d040a81720 100644 --- a/moli-cdp-smoke/moli_cdp_smoke/groups/target_semantics.py +++ b/moli-cdp-smoke/moli_cdp_smoke/groups/target_semantics.py @@ -10,7 +10,7 @@ from urllib.parse import quote from ..assertions import SmokeError, assert_equal, record_contract, wait_until from ..progress import await_with_progress from ..raw_cdp import RawCdpClient, RawCdpError, connect_raw_cdp -from .browser_close import browser_close_instance_lifecycle +from .browser_close import BrowserCloseError, browser_close_instance_lifecycle RawSemanticScenario = Callable[[str, str], Awaitable[dict[str, Any]]] @@ -37,17 +37,18 @@ async def run_target_semantics_group( item.scenario(endpoint, fixture), ) except Exception as error: - results.append( - { - "name": item.name, - "ok": False, - "contract": item.contract, - "source": item.source, - "commands": item.commands, - "errorType": type(error).__name__, - "error": str(error), - } - ) + failed = { + "name": item.name, + "ok": False, + "contract": item.contract, + "source": item.source, + "commands": item.commands, + "errorType": type(error).__name__, + "error": str(error), + } + if isinstance(error, BrowserCloseError): + failed["observed"] = error.observed + results.append(failed) else: record_contract( results, diff --git a/moli-cdp-smoke/moli_cdp_smoke/process.py b/moli-cdp-smoke/moli_cdp_smoke/process.py index 1fb0c3e770..cdb4d2c099 100644 --- a/moli-cdp-smoke/moli_cdp_smoke/process.py +++ b/moli-cdp-smoke/moli_cdp_smoke/process.py @@ -3,9 +3,12 @@ from __future__ import annotations import asyncio import os import signal +import sys from collections.abc import Mapping, Sequence from dataclasses import dataclass from os import PathLike +from pathlib import Path +from typing import Any _PROCESS_CLEANUP_TIMEOUT_SECONDS = 2.0 @@ -52,6 +55,71 @@ async def wait_for_process_exit( return process.returncode +def process_exit_diagnostics(process: asyncio.subprocess.Process) -> dict[str, Any]: + """Snapshot a failed natural exit before cleanup changes the evidence.""" + result: dict[str, Any] = {"pid": process.pid, "returncode": process.returncode} + if not sys.platform.startswith("linux"): + result["procfsUnavailable"] = sys.platform + return result + + def read(path: Path) -> str: + try: + return path.read_text(encoding="utf-8", errors="replace").strip() + except OSError as error: + return f"{type(error).__name__}: {error}" + + def describe(pid: int) -> dict[str, Any]: + directory = Path(f"/proc/{pid}") + status = read(directory / "status") + fields = ("Name", "State", "PPid", "Threads", "VmRSS") + description: dict[str, Any] = {"pid": pid, "wchan": read(directory / "wchan")} + if directory.exists(): + description.update( + (key, value.strip()) + for line in status.splitlines() + if ":" in line + for key, value in [line.split(":", 1)] + if key in fields + ) + else: + description["procfsError"] = status + return description + + result["process"] = describe(process.pid) + tasks = Path(f"/proc/{process.pid}/task") + try: + thread_dirs = sorted(tasks.iterdir(), key=lambda path: int(path.name)) + except OSError as error: + result["threadsError"] = f"{type(error).__name__}: {error}" + thread_dirs = [] + result["threads"] = [ + {"tid": int(path.name), "name": read(path / "comm"), + "wchan": read(path / "wchan"), "syscall": read(path / "syscall")} + for path in thread_dirs[:64] + ] + result["threadsTruncated"] = len(thread_dirs) > 64 + descendants = [] + pending = [process.pid] + seen = {process.pid} + while pending and len(descendants) < 64: + pid = pending.pop(0) + children = read(Path(f"/proc/{pid}/task/{pid}/children")) + for value in children.split(): + if not value.isdecimal(): + continue + child = int(value) + if child in seen: + continue + seen.add(child) + pending.append(child) + descendants.append(describe(child)) + if len(descendants) == 64: + break + result["descendants"] = descendants + result["descendantsTruncated"] = len(descendants) == 64 + return result + + def _consume_task_result(task: asyncio.Task[object]) -> None: try: task.exception() diff --git a/moli-cdp-smoke/tests/test_browser_close.py b/moli-cdp-smoke/tests/test_browser_close.py new file mode 100644 index 0000000000..623d8fc4c2 --- /dev/null +++ b/moli-cdp-smoke/tests/test_browser_close.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import asyncio +import sys +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from moli_cdp_smoke.groups import browser_close, target_semantics +from moli_cdp_smoke.process import terminate_process_tree, wait_for_process_exit + + +class BrowserCloseDiagnosticsTests(unittest.IsolatedAsyncioTestCase): + async def test_acknowledgment_without_exit_still_fails_and_preserves_pre_cleanup_state(self) -> None: + process = await asyncio.create_subprocess_exec( + sys.executable, "-c", "import time; print('ready', flush=True); time.sleep(30)", + stdout=asyncio.subprocess.PIPE, + start_new_session=True, + ) + await asyncio.wait_for(process.stdout.readline(), 5) + serving = SimpleNamespace(process=process, logs=["stderr: shutdown checkpoint"]) + socket = SimpleNamespace( + state=SimpleNamespace(name="OPEN"), close_code=None, close_reason=None, + close=AsyncMock(), + ) + root = SimpleNamespace(websocket=socket) + acknowledgment = {"id": 1000, "result": {}} + identity = {"engine": "chromium"} + + async def bounded_exit(child, *, timeout_seconds): + self.assertEqual(timeout_seconds, 10, "natural exit keeps the original deadline") + return await wait_for_process_exit(child, timeout_seconds=0.02) + + async def stop(child): + socket.close.assert_awaited_once() + await terminate_process_tree(child.process) + + try: + with ( + patch.object(browser_close, "browser_identity", AsyncMock(return_value=identity)), + patch.object(browser_close, "start_browser", AsyncMock(return_value=serving)), + patch.object(browser_close, "wait_for_browser_endpoint", AsyncMock(return_value="http://child")), + patch.object(browser_close, "connect_raw_cdp", AsyncMock(return_value=root)), + patch.object(browser_close, "_close_request", AsyncMock(return_value=acknowledgment)), + patch.object(browser_close, "wait_for_process_exit", bounded_exit), + patch.object(browser_close, "stop_browser", stop), + ): + with self.assertRaisesRegex(browser_close.BrowserCloseError, "root .*natural process exit.*pid=") as raised: + await browser_close.browser_close_instance_lifecycle("http://survivor", "http://fixture") + + observed = raised.exception.observed + failed = observed["failedCase"] + self.assertEqual(failed["closeResponse"], acknowledgment) + self.assertEqual(failed["stage"], "natural process exit") + self.assertEqual(failed["process"]["pid"], process.pid) + self.assertIsNone(failed["process"]["returncode"], "snapshot precedes forced cleanup") + self.assertEqual(failed["connections"][0]["state"], "OPEN") + self.assertEqual(failed["browserLogTail"], serving.logs) + self.assertEqual(observed["completedCases"], []) + self.assertIsNotNone(process.returncode, "failed scenario still cleans up its child") + finally: + await terminate_process_tree(process) + + async def test_worker_result_retains_structured_failure_evidence(self) -> None: + observed = {"completedCases": [{"case": "root", "exitCode": 0}], + "failedCase": {"case": "nested-browser-session", "stage": "natural process exit"}} + error = browser_close.BrowserCloseError("natural process exit timed out", observed) + scenario = AsyncMock(side_effect=error) + contract = target_semantics.RawSemanticContract("close", "contract", "Chromium", ["Browser.close"], scenario) + results = [] + + with patch.object(target_semantics, "_raw_semantic_contracts", return_value=(contract,)): + await target_semantics.run_target_semantics_group("http://child", "http://fixture", results) + + self.assertFalse(results[0]["ok"]) + self.assertEqual(results[0]["observed"], observed) + self.assertEqual(results[0]["error"], str(error)) + + +if __name__ == "__main__": + unittest.main() diff --git a/moli-cdp-smoke/tests/test_external_process.py b/moli-cdp-smoke/tests/test_external_process.py index e11d1ba815..2896a68333 100644 --- a/moli-cdp-smoke/tests/test_external_process.py +++ b/moli-cdp-smoke/tests/test_external_process.py @@ -9,7 +9,11 @@ import unittest from moli_cdp_smoke.assertions import SmokeError from moli_cdp_smoke.groups.external_process import run_external_json_process -from moli_cdp_smoke.process import terminate_process_tree, wait_for_process_exit +from moli_cdp_smoke.process import ( + process_exit_diagnostics, + terminate_process_tree, + wait_for_process_exit, +) class ExternalProcessTests(unittest.IsolatedAsyncioTestCase): @@ -84,6 +88,52 @@ class ExternalProcessTests(unittest.IsolatedAsyncioTestCase): class NaturalProcessExitTests(unittest.IsolatedAsyncioTestCase): + @unittest.skipUnless(sys.platform.startswith("linux"), "diagnostic regression uses procfs") + async def test_failed_exit_captures_owned_process_and_descendant_before_cleanup(self) -> None: + descendant = "import time; print('ready', flush=True); time.sleep(30)" + parent = ( + "import subprocess, sys; " + f"child = subprocess.Popen([sys.executable, '-c', {descendant!r}]); " + "child.wait()" + ) + process = await asyncio.create_subprocess_exec( + sys.executable, "-c", parent, + stdout=asyncio.subprocess.PIPE, + start_new_session=True, + ) + try: + self.assertEqual(await asyncio.wait_for(process.stdout.readline(), 5), b"ready\n") + with self.assertRaises(TimeoutError): + await wait_for_process_exit(process, timeout_seconds=0.02) + + diagnostics = process_exit_diagnostics(process) + + self.assertEqual(diagnostics["pid"], process.pid) + self.assertIsNone(diagnostics["returncode"]) + self.assertIn("State", diagnostics["process"]) + self.assertTrue(diagnostics["threads"]) + self.assertEqual(diagnostics["threads"][0]["tid"], process.pid) + self.assertEqual(len(diagnostics["descendants"]), 1) + self.assertEqual(int(diagnostics["descendants"][0]["PPid"]), process.pid) + self.assertIsNone(process.returncode, "diagnostics must not stop the process") + json.dumps(diagnostics) + finally: + await terminate_process_tree(process) + + @unittest.skipUnless(sys.platform.startswith("linux"), "diagnostic regression uses procfs") + async def test_diagnostics_preserve_a_reaped_process_status(self) -> None: + process = await asyncio.create_subprocess_exec( + sys.executable, "-c", "raise SystemExit(7)", + ) + await wait_for_process_exit(process, timeout_seconds=5) + + diagnostics = process_exit_diagnostics(process) + + self.assertEqual(diagnostics["returncode"], 7) + self.assertIn("FileNotFoundError", diagnostics["process"]["procfsError"]) + self.assertEqual(diagnostics["threads"], []) + self.assertEqual(diagnostics["descendants"], []) + @unittest.skipUnless(os.name == "posix", "inherited-pipe regression is POSIX-specific") async def test_parent_exit_does_not_wait_for_descendant_output_eof(self) -> None: descendant = "import time; time.sleep(30)"