From 3d7312e04f2ac5470b23f3fea0977d6808a58872 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Tue, 8 Sep 2026 05:32:32 +0800 Subject: [PATCH] test(cdp): guard target teardown resource lifetimes Add a managed-process target-lifecycle smoke group covering 1792 close cycles across Target.closeTarget, Page.close, detach-then-close, and explicit-context disposal. Synchronize on exact target and load events, preserve a live peer, and navigate over HTTP after each phase. Enforce fixed Linux FD, poll-driver, and thread budgets after warmup, persist batch samples on failure, and report unsupported external/non-Linux sampling explicitly. Include group-selection and resource-contract tests. --- moli-cdp-smoke/README.md | 14 ++ .../moli_cdp_smoke/groups/target_lifecycle.py | 170 ++++++++++++++++++ moli-cdp-smoke/moli_cdp_smoke/runner.py | 20 ++- moli-cdp-smoke/tests/test_group_selection.py | 7 + moli-cdp-smoke/tests/test_target_lifecycle.py | 38 ++++ 5 files changed, 247 insertions(+), 2 deletions(-) create mode 100644 moli-cdp-smoke/moli_cdp_smoke/groups/target_lifecycle.py create mode 100644 moli-cdp-smoke/tests/test_target_lifecycle.py diff --git a/moli-cdp-smoke/README.md b/moli-cdp-smoke/README.md index 23b12798e..28705f948 100644 --- a/moli-cdp-smoke/README.md +++ b/moli-cdp-smoke/README.md @@ -164,6 +164,20 @@ and restricted-float conversion contracts. It was calibrated on 2026-09-07 against Debian `/usr/bin/chromium` 145.0.7632.116 and runs independently of IndexedDB startup coverage. +The default `target-lifecycle` process group locks down Moli's resource lifetime, +not a Chromium-specific FD count. In one server it closes 800 default-context +targets with `Target.closeTarget`, 800 with `Page.close`, 128 after detaching, +and 64 by disposing explicit contexts. Foreground/background creation alternates. +It waits for each exact `Target.targetDestroyed`, records Linux `/proc//fd` +and thread counts every batch, and checks a fixed post-warmup resource budget. +Each phase must still load a real HTTP document and preserve a live peer Page. +This catches closed renderer wakers retaining Tokio I/O drivers without changing +the test to use a fresh context for each default-context Page. Batch progress, +resource samples, and the usual server logs are retained on failure. With an +external endpoint or on non-Linux systems, protocol churn/navigation still run; +the artifact explicitly reports that FD sampling was unavailable. CI uses the +managed Linux server, so the resource assertions are mandatory there. + Covered well: - The default raw `debugger-breakpoints`, `runtime-exception`, and diff --git a/moli-cdp-smoke/moli_cdp_smoke/groups/target_lifecycle.py b/moli-cdp-smoke/moli_cdp_smoke/groups/target_lifecycle.py new file mode 100644 index 000000000..bfd039a6c --- /dev/null +++ b/moli-cdp-smoke/moli_cdp_smoke/groups/target_lifecycle.py @@ -0,0 +1,170 @@ +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path +from typing import Any + +from ..assertions import SmokeError, assert_equal, record +from ..raw_cdp import RawCdpClient, connect_raw_cdp +from ..serve import MoliServe + + +def process_resources(pid: int) -> dict[str, int] | None: + if not sys.platform.startswith("linux"): + return None + # Inspect only the managed server, not this client or other smoke workers. + # A missing/inaccessible managed process is a failure, not a skipped check. + descriptors: dict[int, str] = {} + for path in Path(f"/proc/{pid}/fd").iterdir(): + try: + descriptors[int(path.name)] = str(path.readlink()) + except FileNotFoundError: + # A transient socket can close between directory listing/readlink. + continue + return { + "fds": len(descriptors), + "eventpoll": sum(value == "anon_inode:[eventpoll]" for value in descriptors.values()), + "eventfd": sum(value == "anon_inode:[eventfd]" for value in descriptors.values()), + "threads": len(list(Path(f"/proc/{pid}/task").iterdir())), + "maxFd": max(descriptors, default=-1), + } + + +def assert_resources_bounded(baseline: dict[str, int], current: dict[str, int]) -> None: + # Permit a fixed amount of in-flight teardown/transport bookkeeping, never + # a per-iteration allowance. The old +2 FD/page leak fails at the first batch. + for key, allowance in (("fds", 8), ("eventpoll", 2), ("eventfd", 2), ("threads", 2)): + if current[key] > baseline[key] + allowance: + raise SmokeError(f"target teardown leaked {key}: baseline={baseline}, current={current}") + + +class LifecycleProbe: + def __init__(self, client: RawCdpClient) -> None: + self.client = client + self.destroyed: set[str] = set() + self.loaded: set[str] = set() + + def observe(self, message: dict[str, Any]) -> None: + if message.get("method") == "Target.targetDestroyed": + self.destroyed.add(message["params"]["targetId"]) + elif message.get("method") == "Page.loadEventFired": + self.loaded.add(message["sessionId"]) + + async def call(self, method: str, params: dict[str, Any] | None = None, + session: str | None = None) -> dict[str, Any]: + response, seen = await self.client.recv_until_id( + await self.client.send(method, params, session_id=session), timeout=10, + ) + for message in seen: + self.observe(message) + return response["result"] + + async def wait_for(self, identities: set[str], identity: str, label: str) -> None: + async def receive() -> None: + while identity not in identities: + self.observe(await self.client.recv()) + identities.remove(identity) + try: + await asyncio.wait_for(receive(), timeout=10) + except TimeoutError as error: + raise SmokeError(f"missing {label} for exact identity {identity}") from error + + async def attach(self, target: str) -> str: + return (await self.call("Target.attachToTarget", { + "targetId": target, "flatten": True, + }))["sessionId"] + + async def close(self, target: str) -> None: + result = await self.call("Target.closeTarget", {"targetId": target}) + assert_equal(result.get("success"), True, "Target.closeTarget accepted") + await self.wait_for(self.destroyed, target, "Target.targetDestroyed") + + async def cycle(self, mode: str, index: int) -> None: + params: dict[str, Any] = {"url": "about:blank", "background": bool(index % 2)} + if mode == "context": + params.update(await self.call("Target.createBrowserContext")) + target = (await self.call("Target.createTarget", params))["targetId"] + if mode == "context": + await self.call("Target.disposeBrowserContext", { + "browserContextId": params["browserContextId"], + }) + await self.wait_for(self.destroyed, target, "disposed context target") + elif mode == "page": + await self.call("Page.close", session=await self.attach(target)) + await self.wait_for(self.destroyed, target, "Page.close target destruction") + else: + if mode == "detach": + await self.call("Target.detachFromTarget", {"sessionId": await self.attach(target)}) + await self.close(target) + + async def navigate(self, url: str) -> None: + target = (await self.call("Target.createTarget", {"url": "about:blank"}))["targetId"] + session = await self.attach(target) + await self.call("Page.enable", session=session) + response = await self.call("Page.navigate", {"url": url}, session) + if response.get("errorText"): + raise SmokeError(f"navigation after target churn failed: {response}") + await self.wait_for(self.loaded, session, "post-churn Page.loadEventFired") + value = await self.call("Runtime.evaluate", { + "expression": "document.querySelector('main')?.textContent", "returnByValue": True, + }, session) + assert_equal(value.get("result", {}).get("value"), "plain ok", "real HTTP document after churn") + await self.close(target) + + +async def run_target_lifecycle_group( + endpoint: str, fixture: str, results: list[dict[str, Any]], serve: MoliServe | None, +) -> None: + pid = serve.process.pid if serve is not None else None + client = await connect_raw_cdp(endpoint) + probe = LifecycleProbe(client) + try: + await probe.call("Target.setDiscoverTargets", {"discover": True}) + for target in (await probe.call("Target.getTargets"))["targetInfos"]: + if target["type"] == "page": + await probe.close(target["targetId"]) + peer = (await probe.call("Target.createTarget", {"url": "about:blank"}))["targetId"] + peer_session = await probe.attach(peer) + await probe.call("Runtime.evaluate", {"expression": "globalThis.lifecycleSentinel = 42"}, peer_session) + + modes = (("target", 800), ("page", 800), ("detach", 128), ("context", 64)) + # Warm all closure routes and the shared network machinery once, then + # hold the same default BrowserContext and a live peer for every batch. + for mode, _ in modes: + for index in range(4): + await probe.cycle(mode, index) + await probe.navigate(f"{fixture}/plain?lifecycle-warmup") + baseline = process_resources(pid) if pid is not None else None + record(results, "target_lifecycle_baseline", { + "pid": pid, "resources": baseline, + "fdSampling": ("external-endpoint-without-owned-pid" if pid is None else + "linux-proc" if baseline is not None else "unavailable-on-this-platform"), + }) + for mode, count in modes: + for index in range(count): + await probe.cycle(mode, index) + if (index + 1) % 100 == 0 or index + 1 == count: + current = process_resources(pid) if pid is not None else None + # Persist the observation before asserting, so a failed + # run retains the resource slope and exact failing batch. + record(results, "target_lifecycle_sample", { + "mode": mode, "closed": index + 1, "resources": current, + }) + print(f"[moli-cdp-smoke] target-lifecycle {mode} {index + 1}/{count} {current}", + file=sys.stderr, flush=True) + if baseline is not None and current is not None: + assert_resources_bounded(baseline, current) + + await probe.navigate(f"{fixture}/plain?after-{mode}-{count}") + sentinel = await probe.call("Runtime.evaluate", { + "expression": "globalThis.lifecycleSentinel", "returnByValue": True, + }, peer_session) + assert_equal(sentinel.get("result", {}).get("value"), 42, "peer renderer survives teardown") + targets = (await probe.call("Target.getTargets"))["targetInfos"] + assert_equal([item["targetId"] for item in targets if item["type"] == "page"], + [peer], "no closed Page remains registered") + record(results, f"target_lifecycle_{mode}", {"closed": count, "navigation": "ok", "peer": "alive"}) + await probe.close(peer) + finally: + await client.websocket.close() diff --git a/moli-cdp-smoke/moli_cdp_smoke/runner.py b/moli-cdp-smoke/moli_cdp_smoke/runner.py index e079d36e8..660fbe4e4 100644 --- a/moli-cdp-smoke/moli_cdp_smoke/runner.py +++ b/moli-cdp-smoke/moli_cdp_smoke/runner.py @@ -62,6 +62,7 @@ from .groups.stagehand import run_stagehand_group from .groups.svg_rect import run_svg_rect_group from .groups.target_semantics import run_target_semantics_group +from .groups.target_lifecycle import run_target_lifecycle_group from .groups.tracing import run_raw_tracing_group, run_tracing_group from .groups.url_policy import run_url_policy_group from .groups.webgl_viewport import run_webgl_viewport_group @@ -87,6 +88,7 @@ ExternalGroupRunner = Callable[[str, str, list[dict[str, Any]]], Awaitable[None]] PageGroupRunner = Callable[[SmokeState], Awaitable[None]] BrowserGroupRunner = Callable[[Any, str, list[dict[str, Any]]], Awaitable[None]] +ProcessGroupRunner = Callable[[str, str, list[dict[str, Any]], MoliServe | None], Awaitable[None]] @dataclass(frozen=True) @@ -94,7 +96,7 @@ class SmokeGroup: name: str description: str phase: str - runner: RawGroupRunner | ExternalGroupRunner | PageGroupRunner | BrowserGroupRunner + runner: RawGroupRunner | ExternalGroupRunner | PageGroupRunner | BrowserGroupRunner | ProcessGroupRunner async def _await_group(group: SmokeGroup, awaitable: Awaitable[None]) -> None: @@ -417,8 +419,13 @@ async def _await_group(group: SmokeGroup, awaitable: Awaitable[None]) -> None: ) +PROCESS_GROUPS: tuple[SmokeGroup, ...] = ( + SmokeGroup("target-lifecycle", "Managed-process target churn, FD bounds, and post-close navigation.", + "process", run_target_lifecycle_group), +) + DEFAULT_GROUPS: tuple[SmokeGroup, ...] = ( - RAW_GROUPS + PAGE_GROUPS + BROWSER_GROUPS + MANAGED_EXTERNAL_GROUPS + RAW_GROUPS + PAGE_GROUPS + BROWSER_GROUPS + MANAGED_EXTERNAL_GROUPS + PROCESS_GROUPS ) ALL_GROUPS: tuple[SmokeGroup, ...] = DEFAULT_GROUPS + OPTIONAL_EXTERNAL_GROUPS DEFAULT_GROUP_NAMES: tuple[str, ...] = tuple(group.name for group in DEFAULT_GROUPS) @@ -430,6 +437,10 @@ async def _await_group(group: SmokeGroup, awaitable: Awaitable[None]) -> None: class SmokeSelection: groups: tuple[SmokeGroup, ...] + @property + def process_groups(self) -> tuple[SmokeGroup, ...]: + return tuple(group for group in self.groups if group.phase == "process") + @property def raw_groups(self) -> tuple[SmokeGroup, ...]: return tuple(group for group in self.groups if group.phase == "raw") @@ -671,6 +682,11 @@ async def async_main(argv: list[str] | None = None) -> int: raise RuntimeError("CDP endpoint was not initialized") if serve is None: await wait_for_cdp_server(endpoint, serve) + for current_group in selection.process_groups: + await _await_group( + current_group, + current_group.runner(endpoint, fixture.url, results, serve), # type: ignore[misc] + ) for current_group in selection.raw_groups: await _await_group( current_group, diff --git a/moli-cdp-smoke/tests/test_group_selection.py b/moli-cdp-smoke/tests/test_group_selection.py index 6246e9fda..fd40533aa 100644 --- a/moli-cdp-smoke/tests/test_group_selection.py +++ b/moli-cdp-smoke/tests/test_group_selection.py @@ -34,6 +34,7 @@ def test_default_runs_every_repository_managed_group(self) -> None: self.assertIn("media-error", DEFAULT_GROUP_NAMES) self.assertIn("webgl-viewport", DEFAULT_GROUP_NAMES) self.assertIn("svg-rect", DEFAULT_GROUP_NAMES) + self.assertIn("target-lifecycle", DEFAULT_GROUP_NAMES) self.assertIn("multi-page", DEFAULT_GROUP_NAMES) self.assertIn("puppeteer", DEFAULT_GROUP_NAMES) self.assertEqual( @@ -50,6 +51,12 @@ def test_only_external_environment_groups_remain_opt_in(self) -> None: ) self.assertTrue(set(optional_names).isdisjoint(DEFAULT_GROUP_NAMES)) + def test_target_lifecycle_uses_managed_process_without_playwright_context(self) -> None: + selection = resolve_group_selection(["target-lifecycle"]) + self.assertEqual(tuple(group.name for group in selection.process_groups), ("target-lifecycle",)) + self.assertFalse(selection.needs_playwright) + self.assertFalse(selection.raw_groups) + def test_registry_and_listing_mark_exact_default_set(self) -> None: self.assertEqual(len(ALL_GROUPS), len(GROUPS_BY_NAME)) listed_defaults = tuple( diff --git a/moli-cdp-smoke/tests/test_target_lifecycle.py b/moli-cdp-smoke/tests/test_target_lifecycle.py new file mode 100644 index 000000000..46155f99f --- /dev/null +++ b/moli-cdp-smoke/tests/test_target_lifecycle.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from moli_cdp_smoke.assertions import SmokeError +from moli_cdp_smoke.groups.target_lifecycle import ( + LifecycleProbe, assert_resources_bounded, process_resources, +) + + +class TargetLifecycleTests(unittest.TestCase): + def test_resource_budget_is_fixed_not_proportional_to_iterations(self) -> None: + baseline = {"fds": 24, "eventpoll": 7, "eventfd": 4, "threads": 20} + assert_resources_bounded(baseline, baseline) + assert_resources_bounded(baseline, dict(baseline, fds=28, eventpoll=8, eventfd=5)) + for key, value in (("fds", 224), ("eventpoll", 10), ("eventfd", 7), ("threads", 23)): + with self.subTest(key=key), self.assertRaises(SmokeError): + assert_resources_bounded(baseline, dict(baseline, **{key: value})) + + def test_event_identity_is_target_and_session_specific(self) -> None: + probe = LifecycleProbe(None) # type: ignore[arg-type] + probe.observe({"method": "Target.targetDestroyed", "params": {"targetId": "old"}}) + probe.observe({"method": "Page.loadEventFired", "sessionId": "session-A"}) + probe.observe({"method": "Target.targetCreated", "params": {"targetInfo": {"targetId": "new"}}}) + self.assertEqual(probe.destroyed, {"old"}) + self.assertEqual(probe.loaded, {"session-A"}) + self.assertNotIn("session-B", probe.loaded) + + def test_non_linux_fd_sampling_is_explicitly_unavailable(self) -> None: + with patch("moli_cdp_smoke.groups.target_lifecycle.sys.platform", "darwin"): + self.assertIsNone(process_resources(1234)) + + def test_missing_managed_process_is_not_silently_skipped(self) -> None: + with patch("moli_cdp_smoke.groups.target_lifecycle.sys.platform", "linux"), \ + patch("moli_cdp_smoke.groups.target_lifecycle.Path.iterdir", side_effect=FileNotFoundError): + with self.assertRaises(FileNotFoundError): + process_resources(1234)