diff --git a/moli-benchmark/README.md b/moli-benchmark/README.md index b9c76d2ed5..dafd622f3c 100644 --- a/moli-benchmark/README.md +++ b/moli-benchmark/README.md @@ -420,6 +420,22 @@ Run the command with `--help` to see fixture, public-site, sampling, and output options. The pull-request workflows under `.github/workflows/` are the source of truth for CI execution and permissions. +## Sampling and failure evidence + +Resource samples retain process identities, read errors, and the raw observations +used for CPU and memory totals. Missing or incomplete reads remain unavailable; +they are not reported as zero. Thread CPU deltas and process clock ticks are +recorded separately, including thread retirement and counter precision. + +Target shutdown seals the sampler before terminating the engine and records log +reader EOF, cleanup errors, and timeout results. Startup and cancellation failures +retain their original exceptions together with the evidence collected so far. +Sequential CDP receive failures attach already observed messages through +`cdp_receive_evidence`, without consuming extra messages or replacing the error. + +These records improve measurement validity and failure diagnosis. They do not +change browser rendering policy or establish a performance improvement. + ## Development Run the core CLI tests from `moli-benchmark/`: diff --git a/moli-benchmark/moli_benchmark/raw_cdp.py b/moli-benchmark/moli_benchmark/raw_cdp.py index c1f5e863a5..7d7010f83f 100644 --- a/moli-benchmark/moli_benchmark/raw_cdp.py +++ b/moli-benchmark/moli_benchmark/raw_cdp.py @@ -6,8 +6,9 @@ import contextlib import json import time import urllib.request +from collections.abc import Callable from dataclasses import dataclass -from typing import Any, Callable +from typing import Any import websockets from websockets.asyncio.client import ClientConnection @@ -95,19 +96,39 @@ class RawCdpClient: raise RawCdpError(f"unexpected CDP payload: {payload!r}") return payload - async def recv_until_id(self, message_id: int, *, timeout: float = 10.0) -> tuple[dict[str, Any], list[dict[str, Any]]]: + async def recv_until_id( + self, + message_id: int, + *, + timeout: float = 10.0, + method: str | None = None, + ) -> tuple[dict[str, Any], list[dict[str, Any]]]: seen: list[dict[str, Any]] = [] deadline = asyncio.get_running_loop().time() + timeout - while True: - remaining = deadline - asyncio.get_running_loop().time() - if remaining <= 0: - raise RawCdpTimeoutError(f"timed out waiting for CDP response id={message_id}; seen={seen[-20:]}") - message = await asyncio.wait_for(self.recv(), timeout=remaining) - seen.append(message) - if message.get("id") == message_id: - if "error" in message: - raise RawCdpError(f"CDP command id={message_id} failed: {message['error']}") - return message, seen + try: + while True: + remaining = deadline - asyncio.get_running_loop().time() + if remaining <= 0: + raise RawCdpTimeoutError( + f"timed out waiting for CDP response id={message_id}; seen={seen[-20:]}" + ) + message = await asyncio.wait_for(self.recv(), timeout=remaining) + seen.append(message) + if message.get("id") == message_id: + if "error" in message: + raise RawCdpError( + f"CDP command id={message_id} failed: {message['error']}" + ) + return message, seen + except BaseException as error: + # Keep the same exception (including cancellation) and receiver. + # A failed response/receive must not discard already observed events. + error.cdp_receive_evidence = { + "message_id": message_id, + "method": method, + "messages": list(seen), + } + raise async def connect_raw_cdp(endpoint: str) -> RawCdpClient: diff --git a/moli-benchmark/moli_benchmark/sampling.py b/moli-benchmark/moli_benchmark/sampling.py index 2c02b26f1e..65708b3e6e 100644 --- a/moli-benchmark/moli_benchmark/sampling.py +++ b/moli-benchmark/moli_benchmark/sampling.py @@ -1,171 +1,468 @@ from __future__ import annotations +import copy +import hashlib import os import subprocess import threading import time +import traceback from dataclasses import dataclass, field +from itertools import pairwise from pathlib import Path from typing import Any +INSTRUMENTATION_SCHEMA = 2 -def _process_tree(root_pid: int) -> list[int]: - ppid_by_pid: dict[int, int] = {} - for stat_path in Path("/proc").glob("[0-9]*/stat"): - try: - raw = stat_path.read_text(encoding="utf-8", errors="replace") - except OSError: - continue - close = raw.rfind(")") - if close < 0: - continue - parts = raw[close + 2 :].split() - if len(parts) < 2: - continue - try: - pid = int(stat_path.parent.name) - ppid = int(parts[1]) - except ValueError: - continue - ppid_by_pid[pid] = ppid - result = [root_pid] - seen = {root_pid} - changed = True - while changed: - changed = False - for pid, ppid in ppid_by_pid.items(): - if pid not in seen and ppid in seen: - result.append(pid) - seen.add(pid) - changed = True +def error_evidence(error: BaseException) -> dict[str, Any]: + result = {"type": type(error).__name__, "message": str(error)} + if isinstance(error, OSError): + result.update(errno=error.errno, filename=error.filename) return result -def _read_pss_bytes(pid: int) -> int | None: +def _text_read(path: Path) -> dict[str, Any]: + record = {"path": str(path), "started_monotonic_ns": time.monotonic_ns()} try: - text = Path(f"/proc/{pid}/smaps_rollup").read_text(encoding="utf-8", errors="replace") - except OSError: - return None - for line in text.splitlines(): - if line.startswith("Pss:"): - parts = line.split() - if len(parts) >= 2: - return int(parts[1]) * 1024 - return None + record.update( + status="ok", raw=path.read_text(encoding="utf-8", errors="replace") + ) + except OSError as error: + record.update(status="os_error", error=error_evidence(error), raw=None) + record["finished_monotonic_ns"] = time.monotonic_ns() + return record + + +def _stat_read(pid: int, tid: int | None = None) -> dict[str, Any]: + path = Path(f"/proc/{pid}/stat" if tid is None else f"/proc/{pid}/task/{tid}/stat") + record = _text_read(path) + record["identity"] = None + if record["status"] == "ok": + try: + raw = record["raw"] + close = raw.rfind(")") + fields = raw[close + 2 :].split() + observed_id = int(raw[: raw.index("(")].strip()) + if close < 0 or observed_id != (pid if tid is None else tid): + raise ValueError("proc stat ID/format mismatch") + user, system, start = (int(fields[index]) for index in (11, 12, 19)) + if min(user, system, start) < 0: + raise ValueError("negative proc stat counter") + record.update( + pid=pid, + tid=tid, + start_ticks=start, + identity=f"{pid}:{start}" if tid is None else f"{pid}:{tid}:{start}", + state=fields[0], + ppid=int(fields[1]), + user_ticks=user, + system_ticks=system, + total_ticks=user + system, + ) + except (ValueError, IndexError) as error: + record.update(status="parse_error", error=error_evidence(error)) + return record + + +def _same_identity(before: dict, after: dict) -> bool: + return before.get("identity") is not None and before["identity"] == after.get( + "identity" + ) + + +def _process_tree_evidence(root_pid: int) -> tuple[list[int], dict[int, dict], dict]: + stats, failures = {}, {} + enumeration_error = None + try: + paths = list(Path("/proc").glob("[0-9]*/stat")) + except OSError as error: + paths, enumeration_error = [], error_evidence(error) + for path in paths: + pid = int(path.parent.name) + record = _stat_read(pid) + if record["status"] == "ok": + stats[pid] = record + else: + key = f"{record['status']}:{record.get('error', {}).get('errno')}" + failures[key] = failures.get(key, 0) + 1 + pids, seen = [root_pid], {root_pid} + for parent in pids: + for pid, record in stats.items(): + if record["ppid"] == parent and pid not in seen: + pids.append(pid) + seen.add(pid) + # Root is kept even when unreadable: never turn missing coverage into 0/0. + if root_pid not in stats: + stats[root_pid] = _stat_read(root_pid) + return ( + pids, + {pid: stats[pid] for pid in pids}, + { + "scope": "best_effort_non_atomic_proc_tree", + "enumerated_stat_paths": len(paths), + "scan_failures_by_status_errno": failures, + "enumeration_error": enumeration_error, + "all_descendants_observed": "unknown", + }, + ) + + +def _process_tree(root_pid: int) -> list[int]: + return _process_tree_evidence(root_pid)[0] + + +def _field_read(source: dict, field_name: str, multiplier: int = 1) -> dict: + record = {key: value for key, value in source.items() if key != "raw"} + record.update(field=field_name, value=None) + if source["status"] != "ok": + return record + for line in source["raw"].splitlines(): + fields = line.split() + if fields and fields[0] == field_name: + record["raw_line"] = line + try: + value = int(fields[1]) + if value < 0: + raise ValueError("negative procfs field") + record["value"] = value * multiplier + except (ValueError, IndexError) as error: + record.update(status="parse_error", error=error_evidence(error)) + return record + record["status"] = "missing_field" + return record + + +def _fd_read(pid: int) -> dict: + path = Path(f"/proc/{pid}/fd") + record = { + "path": str(path), + "value": None, + "started_monotonic_ns": time.monotonic_ns(), + } + try: + record.update(status="ok", value=sum(1 for _ in path.iterdir())) + except OSError as error: + record.update(status="os_error", error=error_evidence(error)) + record["finished_monotonic_ns"] = time.monotonic_ns() + return record + + +def _read_pss_bytes(pid: int) -> int | None: + return _field_read(_text_read(Path(f"/proc/{pid}/smaps_rollup")), "Pss:", 1024)[ + "value" + ] def _read_rss_bytes(pid: int) -> int | None: - try: - text = Path(f"/proc/{pid}/status").read_text(encoding="utf-8", errors="replace") - except OSError: + return _field_read(_text_read(Path(f"/proc/{pid}/status")), "VmRSS:", 1024)["value"] + + +def _read_thread_count(pid: int) -> int | None: + return _field_read(_text_read(Path(f"/proc/{pid}/status")), "Threads:")["value"] + + +def _read_fd_count(pid: int) -> int | None: + return _fd_read(pid)["value"] + + +def _read_cpu_identity(pid: int) -> tuple[tuple[int, int], int] | None: + value = _stat_read(pid) + if value["status"] != "ok": return None - for line in text.splitlines(): - if line.startswith("VmRSS:"): - parts = line.split() - if len(parts) >= 2: - return int(parts[1]) * 1024 - return None + return (pid, value["start_ticks"]), value["total_ticks"] def _read_cpu_percent(pids: list[int]) -> float | None: - if not pids: - return None + # Compatibility for other callers: this is ps lifetime %CPU, not phase CPU. try: completed = subprocess.run( - ["ps", "-o", "%cpu=", "-p", ",".join(str(pid) for pid in pids)], + ["ps", "-o", "%cpu=", "-p", ",".join(map(str, pids))], capture_output=True, text=True, timeout=1, check=False, ) - except (OSError, subprocess.TimeoutExpired): - return None - total = 0.0 - found = False - for line in completed.stdout.splitlines(): - try: - total += float(line.strip()) - found = True - except ValueError: - continue - return total if found else None - - -def _read_thread_count(pid: int) -> int | None: - try: - text = Path(f"/proc/{pid}/status").read_text(encoding="utf-8", errors="replace") - except OSError: - return None - for line in text.splitlines(): - if line.startswith("Threads:"): - parts = line.split() - if len(parts) >= 2: - return int(parts[1]) - return None - - -def _read_fd_count(pid: int) -> int | None: - try: - return sum(1 for _ in Path(f"/proc/{pid}/fd").iterdir()) - except OSError: + return ( + sum(float(line.strip()) for line in completed.stdout.splitlines()) + if completed.returncode == 0 and completed.stdout.strip() + else None + ) + except (OSError, ValueError, subprocess.TimeoutExpired): return None -def _read_cpu_identity(pid: int) -> tuple[tuple[int, int], int] | None: - try: - raw = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8", errors="replace") - except OSError: - return None - close = raw.rfind(")") - if close < 0: - return None - fields = raw[close + 2 :].split() - if len(fields) <= 19: - return None - try: - user_ticks = int(fields[11]) - system_ticks = int(fields[12]) - start_ticks = int(fields[19]) - except ValueError: - return None - return (pid, start_ticks), user_ticks + system_ticks - - -def snapshot_resources(root_pid: int, *, include_lifetime_cpu: bool = True) -> dict[str, Any]: - pids = _process_tree(root_pid) - pss_values = [_read_pss_bytes(pid) for pid in pids] - rss_values = [_read_rss_bytes(pid) for pid in pids] - thread_values = [_read_thread_count(pid) for pid in pids] - fd_values = [_read_fd_count(pid) for pid in pids] - pss_values = [value for value in pss_values if value is not None] - rss_values = [value for value in rss_values if value is not None] - thread_values = [value for value in thread_values if value is not None] - fd_values = [value for value in fd_values if value is not None] - return { +def snapshot_resources( + root_pid: int, *, include_lifetime_cpu: bool = True +) -> dict[str, Any]: + started = time.monotonic_ns() + pids, before, tree = _process_tree_evidence(root_pid) + processes = [] + for pid in pids: + status = _text_read(Path(f"/proc/{pid}/status")) + row = { + "pid": pid, + "identity_before": before[pid], + "pss": _field_read( + _text_read(Path(f"/proc/{pid}/smaps_rollup")), "Pss:", 1024 + ), + "rss": _field_read(status, "VmRSS:", 1024), + "threads": _field_read(status, "Threads:"), + "fds": _fd_read(pid), + } + row["identity_after"] = _stat_read(pid) + row["identity_valid"] = _same_identity(before[pid], row["identity_after"]) + processes.append(row) + result = { + "instrumentation_schema": INSTRUMENTATION_SCHEMA, "pids": pids, "process_count": len(pids), - "pss_bytes": sum(pss_values) if len(pss_values) == len(pids) else None, - "pss_process_count": len(pss_values), - "rss_bytes": sum(rss_values) if len(rss_values) == len(pids) else None, - "rss_process_count": len(rss_values), - "thread_count": sum(thread_values) - if len(thread_values) == len(pids) - else None, - "thread_process_count": len(thread_values), - "fd_count": sum(fd_values) if len(fd_values) == len(pids) else None, - "fd_process_count": len(fd_values), + "tree": tree, + "processes": processes, + "started_monotonic_ns": started, + "finished_monotonic_ns": time.monotonic_ns(), "cpu_percent": _read_cpu_percent(pids) if include_lifetime_cpu else None, + "cpu_percent_method": "ps_lifetime" if include_lifetime_cpu else None, + } + for name, total, count in ( + ("pss", "pss_bytes", "pss_process_count"), + ("rss", "rss_bytes", "rss_process_count"), + ("threads", "thread_count", "thread_process_count"), + ("fds", "fd_count", "fd_process_count"), + ): + values = [ + row[name]["value"] + for row in processes + if row["identity_valid"] and row[name]["status"] == "ok" + ] + result[count] = len(values) + result[total] = sum(values) if pids and len(values) == len(pids) else None + result["selected_process_reads_complete"] = ( + bool(processes) + and all( + row["identity_valid"] + and all( + row[name]["status"] == "ok" for name in ("pss", "rss", "threads", "fds") + ) + for row in processes + ) + and tree.get("enumeration_error") is None + ) + result["finished_monotonic_ns"] = time.monotonic_ns() + return result + + +def clock_ticks_evidence() -> dict: + result = {"source": "os.sysconf(SC_CLK_TCK)", "ticks_per_second": None} + try: + rate = os.sysconf("SC_CLK_TCK") + if not isinstance(rate, int) or rate <= 0: + raise ValueError("invalid SC_CLK_TCK") + result.update(ticks_per_second=rate, counter_tick_ms=1000 / rate) + except (OSError, ValueError) as error: + result["error"] = error_evidence(error) + return result + + +def cpu_snapshot(pid: int) -> dict: + """Root process ticks and its observed live threads; no descendant CPU.""" + result = { + "started_monotonic_ns": time.monotonic_ns(), + "root_before": _stat_read(pid), + "threads": {}, + "observations": [], + "enumeration_error": None, + } + try: + tasks = list(Path(f"/proc/{pid}/task").iterdir()) + except OSError as error: + tasks = [] + result["enumeration_error"] = error_evidence(error) + result["enumerated_tids"] = sorted(int(task.name) for task in tasks) + for tid in result["enumerated_tids"]: + before = _stat_read(pid, tid) + runtime = _text_read(Path(f"/proc/{pid}/task/{tid}/schedstat")) + runtime["value"] = None + if runtime["status"] == "ok": + try: + runtime["value"] = int(runtime["raw"].split()[0]) + if runtime["value"] < 0: + raise ValueError("negative schedstat runtime") + except (ValueError, IndexError) as error: + runtime.update( + status="parse_error", error=error_evidence(error), value=None + ) + after = _stat_read(pid, tid) + valid = _same_identity(before, after) and runtime["status"] == "ok" + result["observations"].append( + { + "tid": tid, + "before": before, + "runtime": runtime, + "after": after, + "valid": valid, + } + ) + if valid: + result["threads"][before["identity"]] = runtime["value"] + result["root_after"] = _stat_read(pid) + result["finished_monotonic_ns"] = time.monotonic_ns() + return result + + +def cpu_delta( + before: dict, after: dict, clock_ticks: dict, expected_root: str | None +) -> dict: + old, new = before["threads"], after["threads"] + common, created, retired = ( + old.keys() & new.keys(), + new.keys() - old.keys(), + old.keys() - new.keys(), + ) + regressed = sorted(key for key in common if new[key] < old[key]) + identities = [ + point[key].get("identity") + for point in (before, after) + for key in ("root_before", "root_after") + ] + root_valid = expected_root is not None and all( + key == expected_root for key in identities + ) + boundary_complete = root_valid and all( + point["enumeration_error"] is None + and bool(point["observations"]) + and all(row["valid"] for row in point["observations"]) + for point in (before, after) + ) + known_ns = ( + None + if regressed or not root_valid + else sum(new[key] - old[key] for key in common) + + sum(new[key] for key in created) + ) + first, last = before["root_after"], after["root_after"] + process_valid = root_valid and all( + last.get(key, -1) >= first.get(key, 0) for key in ("user_ticks", "system_ticks") + ) + ticks = last["total_ticks"] - first["total_ticks"] if process_valid else None + rate = clock_ticks["ticks_per_second"] + process_cpu = { + "method": "proc_root_stat_utime_plus_stime", + "scope": "root_process_all_threads_no_descendants", + "before": first, + "after": last, + "clock_ticks": clock_ticks, + "identity_valid": root_valid, + "counter_valid": process_valid, + "delta_ticks": ticks, + "cpu_ms": ticks / rate * 1000 if ticks is not None and rate else None, + "zero_ticks_mean": "no_exported_tick_change_not_zero_cpu", + "remote_process_clock": {"status": "unsupported_not_runtime_validated"}, + } + return { + # Historical field stays an explicitly named thread estimate. Never + # silently substitute process ticks when thread evidence is incomplete. + "cpu_ms": known_ns / 1e6 + if known_ns is not None and boundary_complete + else None, + "thread_cpu_observed": { + "method": "root_threads_schedstat_boundary_estimate", + "before": before, + "after": after, + "known_cpu_ms": known_ns / 1e6 if known_ns is not None else None, + "boundary_read_complete": boundary_complete, + "counter_regressions": regressed, + "lifetime_coverage": "not_proven_by_boundary_sampling", + }, + "created_threads": sorted(created), + "retired_threads": sorted(retired), + "process_cpu": process_cpu, + } + + +def affinity_snapshot( + pid: int, requested_cpu: int | None, expected_root: str | None +) -> dict: + result = { + "started_monotonic_ns": time.monotonic_ns(), + "root_before": _stat_read(pid), + "requested_cpu": requested_cpu, + "threads": [], + "enumeration_error": None, + } + try: + tids = sorted(int(path.name) for path in Path(f"/proc/{pid}/task").iterdir()) + except OSError as error: + tids = [] + result["enumeration_error"] = error_evidence(error) + for tid in tids: + before = _stat_read(pid, tid) + row = {"tid": tid, "before": before, "cpus": None, "error": None} + try: + row["cpus"] = sorted(os.sched_getaffinity(tid)) + except OSError as error: + row["error"] = error_evidence(error) + row["after"] = _stat_read(pid, tid) + row["identity_valid"] = _same_identity(before, row["after"]) + result["threads"].append(row) + result["root_after"] = _stat_read(pid) + result["boundary_complete"] = ( + bool(tids) + and result["enumeration_error"] is None + and expected_root is not None + and all( + result[key].get("identity") == expected_root + for key in ("root_before", "root_after") + ) + and all( + row["identity_valid"] and row["error"] is None for row in result["threads"] + ) + ) + result["all_observed_match"] = ( + result["boundary_complete"] + and requested_cpu is not None + and all(row["cpus"] == [requested_cpu] for row in result["threads"]) + ) + result["lifetime_affinity"] = "not_proven_by_boundary_sampling" + result["finished_monotonic_ns"] = time.monotonic_ns() + return result + + +def instrumentation_manifest(probe_path: Path, *input_paths: Path) -> dict: + paths = [ + probe_path, + Path(__file__), + Path(__file__).with_name("target_serve.py"), + *input_paths, + ] + return { + "schema": INSTRUMENTATION_SCHEMA, + "input_sha256": { + str(path.resolve()): hashlib.sha256(path.read_bytes()).hexdigest() + for path in paths + }, + "process_cpu_method": "proc_root_stat_utime_plus_stime", + "process_cpu_precision": "SC_CLK_TCK_quantized_not_schedstat_nanoseconds", + "thread_cpu_scope": "root_observed_boundary_threads", + "remote_process_clock": "unsupported_not_runtime_validated", + "resource_scope": "best_effort_non_atomic_process_tree", + "observer_clock": "time.monotonic_ns_boundaries_not_atomic_snapshot", } def _sanitize_cgroup_name(name: str) -> str: - return "".join(char if char.isalnum() or char in ("-", "_", ".") else "_" for char in name) + return "".join( + char if char.isalnum() or char in ("-", "_", ".") else "_" for char in name + ) def _read_pid_cgroups(pid: int) -> list[dict[str, Any]]: try: - lines = Path(f"/proc/{pid}/cgroup").read_text(encoding="utf-8", errors="replace").splitlines() + lines = ( + Path(f"/proc/{pid}/cgroup") + .read_text(encoding="utf-8", errors="replace") + .splitlines() + ) except OSError: return [] entries = [] @@ -177,7 +474,9 @@ def _read_pid_cgroups(pid: int) -> list[dict[str, Any]]: entries.append( { "hierarchy": hierarchy, - "controllers": [controller for controller in controllers.split(",") if controller], + "controllers": [ + controller for controller in controllers.split(",") if controller + ], "path": relative_path, } ) @@ -197,7 +496,9 @@ def collect_cgroup_artifacts(pid: int, output_dir: Path) -> dict[str, Any]: output_dir.mkdir(parents=True, exist_ok=True) entries = _read_pid_cgroups(pid) try: - raw_cgroup = Path(f"/proc/{pid}/cgroup").read_text(encoding="utf-8", errors="replace") + raw_cgroup = Path(f"/proc/{pid}/cgroup").read_text( + encoding="utf-8", errors="replace" + ) except OSError as error: raw_cgroup = f"unavailable: {error}\n" proc_cgroup_path = output_dir / "proc-cgroup.txt" @@ -235,7 +536,10 @@ def collect_cgroup_artifacts(pid: int, output_dir: Path) -> dict[str, Any]: source = root / file_name target = root_dir / file_name try: - target.write_text(source.read_text(encoding="utf-8", errors="replace"), encoding="utf-8") + target.write_text( + source.read_text(encoding="utf-8", errors="replace"), + encoding="utf-8", + ) copied.append(str(target)) except OSError as error: unavailable.append(f"{source}: {error}") @@ -262,24 +566,53 @@ class ResourceSampler: _thread: threading.Thread | None = None _started_monotonic: float | None = None _previous_monotonic: float | None = None - _previous_cpu: dict[tuple[int, int], int] = field(default_factory=dict) + _previous_cpu: dict[str, int] = field(default_factory=dict) _observer_error: str | None = None + _observer_error_evidence: dict | None = None + _clock_ticks: dict = field(default_factory=dict) + _previous_cpu_complete: bool = False + _samples_lock: threading.Lock = field(default_factory=threading.Lock) + _sealed_samples: list[dict[str, Any]] | None = None + _stop_requested_ns: int | None = None + _join_finished_ns: int | None = None + _thread_exited_ns: int | None = None + _last_committed_ns: int | None = None + _join_ever_timed_out: bool = False def start(self) -> None: if self.interval_seconds <= 0: raise ValueError("resource sample interval must be positive") if self._thread is not None: raise RuntimeError("resource sampler already started") + self._clock_ticks = clock_ticks_evidence() self._started_monotonic = time.perf_counter() - self._thread = threading.Thread(target=self._run, name="moli-resource-sampler", daemon=True) + self._thread = threading.Thread( + target=self._run, name="moli-resource-sampler", daemon=True + ) self._thread.start() def stop(self) -> dict[str, Any]: + if self._stop_requested_ns is None: + self._stop_requested_ns = time.monotonic_ns() self._stop.set() if self._thread is not None: self._thread.join(timeout=2) + alive = self._thread is not None and self._thread.is_alive() + self._join_finished_ns = time.monotonic_ns() + self._join_ever_timed_out |= alive + if not alive: + with self._samples_lock: + self._sealed_samples = copy.deepcopy(self.samples) return self.summary() + def export_samples(self) -> list[dict[str, Any]]: + with self._samples_lock: + return copy.deepcopy( + self._sealed_samples + if self._sealed_samples is not None + else self.samples + ) + def _run(self) -> None: try: next_capture = time.perf_counter() @@ -287,73 +620,155 @@ class ResourceSampler: capture_started = time.perf_counter() sample = snapshot_resources(self.root_pid, include_lifetime_cpu=False) captured = time.perf_counter() - identities = dict( - identity - for pid in sample.get("pids", []) - if (identity := _read_cpu_identity(int(pid))) is not None + rows = sample.get("processes", []) + identities = { + row["identity_after"]["identity"]: row["identity_after"][ + "total_ticks" + ] + for row in rows + if row["identity_valid"] + } + complete = bool(rows) and len(identities) == sample["process_count"] + common = identities.keys() & self._previous_cpu.keys() + regressed = sorted( + key for key in common if identities[key] < self._previous_cpu[key] ) - cpu_percent = None - if self._previous_monotonic is not None: - elapsed = captured - self._previous_monotonic - delta_ticks = sum( - max(0, ticks - self._previous_cpu[identity]) - for identity, ticks in identities.items() - if identity in self._previous_cpu + delta_ticks = ( + None + if regressed + else sum( + identities[key] - self._previous_cpu[key] for key in common ) - if elapsed > 0: - try: - clock_ticks = float(os.sysconf("SC_CLK_TCK")) - except (OSError, ValueError): - clock_ticks = 100.0 - if clock_ticks > 0: - cpu_percent = delta_ticks / clock_ticks / elapsed * 100.0 + ) + elapsed = ( + captured - self._previous_monotonic + if self._previous_monotonic is not None + else None + ) + stable = ( + complete + and self._previous_cpu_complete + and identities.keys() == self._previous_cpu.keys() + ) + rate = self._clock_ticks.get("ticks_per_second") + sample["cpu_percent"] = ( + delta_ticks / rate / elapsed * 100 + if stable + and delta_ticks is not None + and rate + and elapsed + and elapsed > 0 + else None + ) + sample["cpu_percent_method"] = "proc_tree_common_identity_tick_delta" + sample["cpu_evidence"] = { + "before": self._previous_cpu, + "after": identities, + "clock_ticks": self._clock_ticks, + "elapsed_seconds": elapsed, + "known_common_delta_ticks": delta_ticks, + "counter_regressions": regressed, + "created_processes": sorted( + identities.keys() - self._previous_cpu.keys() + ), + "retired_processes": sorted( + self._previous_cpu.keys() - identities.keys() + ), + "stable_observed_process_set": stable, + "lifetime_coverage": "not_proven_by_boundary_sampling", + } self._previous_monotonic = captured self._previous_cpu = identities - sample["cpu_percent"] = cpu_percent + self._previous_cpu_complete = complete sample["timestamp"] = time.time() sample["elapsed_ms"] = ( - (captured - self._started_monotonic) * 1000.0 + (captured - self._started_monotonic) * 1000 if self._started_monotonic is not None else 0.0 ) - sample["capture_duration_ms"] = (captured - capture_started) * 1000.0 + sample["capture_duration_ms"] = (captured - capture_started) * 1000 sample["kind"] = "periodic" - self.samples.append(sample) + with self._samples_lock: + sample["sequence"] = len(self.samples) + self.samples.append(sample) + self._last_committed_ns = time.monotonic_ns() next_capture += self.interval_seconds now = time.perf_counter() if next_capture <= now: missed = int((now - next_capture) / self.interval_seconds) + 1 next_capture += missed * self.interval_seconds - remaining = max(0.0, next_capture - now) - self._stop.wait(remaining) - except BaseException as error: + self._stop.wait(max(0.0, next_capture - now)) + except BaseException as error: # noqa: BLE001 -- Preserve observer termination as evidence. self._observer_error = f"{type(error).__name__}: {error}" + self._observer_error_evidence = { + **error_evidence(error), + "traceback": traceback.format_exc(), + } + finally: + self._thread_exited_ns = time.monotonic_ns() def summary(self) -> dict[str, Any]: - pss = [sample["pss_bytes"] for sample in self.samples if sample.get("pss_bytes") is not None] - rss = [sample["rss_bytes"] for sample in self.samples if sample.get("rss_bytes") is not None] - cpu = [sample["cpu_percent"] for sample in self.samples if sample.get("cpu_percent") is not None] - process_counts = [sample["process_count"] for sample in self.samples if sample.get("process_count") is not None] - thread_counts = [sample["thread_count"] for sample in self.samples if sample.get("thread_count") is not None] - fd_counts = [sample["fd_count"] for sample in self.samples if sample.get("fd_count") is not None] + samples = self.export_samples() + pss = [ + sample["pss_bytes"] + for sample in samples + if sample.get("pss_bytes") is not None + ] + rss = [ + sample["rss_bytes"] + for sample in samples + if sample.get("rss_bytes") is not None + ] + cpu = [ + sample["cpu_percent"] + for sample in samples + if sample.get("cpu_percent") is not None + ] + process_counts = [ + sample["process_count"] + for sample in samples + if sample.get("process_count") is not None + ] + thread_counts = [ + sample["thread_count"] + for sample in samples + if sample.get("thread_count") is not None + ] + fd_counts = [ + sample["fd_count"] + for sample in samples + if sample.get("fd_count") is not None + ] capture_durations = [ float(sample["capture_duration_ms"]) - for sample in self.samples + for sample in samples if sample.get("capture_duration_ms") is not None ] intervals = [ float(current["elapsed_ms"]) - float(previous["elapsed_ms"]) - for previous, current in zip(self.samples, self.samples[1:]) - if previous.get("elapsed_ms") is not None and current.get("elapsed_ms") is not None + for previous, current in pairwise(samples) + if previous.get("elapsed_ms") is not None + and current.get("elapsed_ms") is not None ] complete_pss_samples = sum( 1 - for sample in self.samples + for sample in samples if sample.get("pss_bytes") is not None and sample.get("pss_process_count") == sample.get("process_count") ) return { - "sample_count": len(self.samples), + "instrumentation_schema": INSTRUMENTATION_SCHEMA, + "sample_count": len(samples), + "sealed": self._sealed_samples is not None, + "stop_requested_monotonic_ns": self._stop_requested_ns, + "join_finished_monotonic_ns": self._join_finished_ns, + "thread_exited_monotonic_ns": self._thread_exited_ns, + "last_committed_monotonic_ns": self._last_committed_ns, + "last_committed_sample_sequence": len(samples) - 1 if samples else None, + "join_ever_timed_out": self._join_ever_timed_out, + "observer_error_evidence": self._observer_error_evidence, + "scope": "sampler_start_through_stop_request_inflight_completion", + "totals_are_final": self._sealed_samples is not None, "interval_seconds": self.interval_seconds, "peak_pss_bytes": max(pss) if pss else None, "peak_rss_bytes": max(rss) if rss else None, @@ -369,15 +784,25 @@ class ResourceSampler: "max": max(capture_durations) if capture_durations else None, }, "late_sample_count": sum( - 1 for interval in intervals if interval > self.interval_seconds * 1000.0 * 1.5 + 1 + for interval in intervals + if interval > self.interval_seconds * 1000.0 * 1.5 ), "observed_interval_ms": { "average": sum(intervals) / len(intervals) if intervals else None, "max": max(intervals) if intervals else None, }, + "resource_complete_samples": sum( + sample.get("selected_process_reads_complete") is True + for sample in samples + ), + "resource_reads_complete": bool(samples) + and all( + sample.get("selected_process_reads_complete") is True + for sample in samples + ), "pss_complete_samples": complete_pss_samples, - "pss_complete": bool(self.samples) - and complete_pss_samples == len(self.samples), + "pss_complete": bool(samples) and complete_pss_samples == len(samples), "observer_error": self._observer_error, "thread_alive_after_stop": bool( self._thread is not None and self._thread.is_alive() diff --git a/moli-benchmark/moli_benchmark/target_serve.py b/moli-benchmark/moli_benchmark/target_serve.py index 0986704d01..af38d48ed7 100644 --- a/moli-benchmark/moli_benchmark/target_serve.py +++ b/moli-benchmark/moli_benchmark/target_serve.py @@ -1,23 +1,26 @@ from __future__ import annotations +import asyncio import os import shutil import subprocess import tempfile import threading import time -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Any from .config import REPO_ROOT, ReservedPort, clear_proxy_env, reserve_port -from .sampling import ResourceSampler +from .sampling import ResourceSampler, _stat_read, error_evidence from .serve import probe_url from .synthetic_compare import target_enables_all_resource_fetch, target_metadata class TargetServeError(RuntimeError): - pass + def __init__(self, message: str, *, evidence: dict | None = None): + super().__init__(message) + self.evidence = evidence class TargetServeProcessExit(TargetServeError): @@ -35,11 +38,13 @@ class TargetServeHandle: endpoint: str command: list[str] logs: list[str] - sampler: ResourceSampler + sampler: ResourceSampler | None log_threads: list[threading.Thread] port_lease: ReservedPort temp_dir: Path | None = None ready_ms: float | None = None + root_identity: dict | None = None + log_state: dict[str, dict] = field(default_factory=dict) def _append_log(logs: list[str], line: str) -> None: @@ -48,39 +53,57 @@ def _append_log(logs: list[str], line: str) -> None: del logs[: len(logs) - 400] -def _read_available(stream: Any, label: str, logs: list[str]) -> None: - if stream is None: - return +def _read_available(stream: Any, label: str, logs: list[str], state: dict) -> None: + state.update( + eof=False, error=None, lines_read=0, started_monotonic_ns=time.monotonic_ns() + ) try: + if stream is None: + state["error"] = {"type": "MissingStream", "message": "pipe unavailable"} + return while True: line = stream.readline() if not line: + state["eof"] = True return - _append_log(logs, f"{label}: {line.decode('utf-8', errors='replace').rstrip()}") - except OSError: - return + state["lines_read"] += 1 + _append_log( + logs, f"{label}: {line.decode('utf-8', errors='replace').rstrip()}" + ) + except Exception as error: # noqa: BLE001 -- Record failure and continue bounded cleanup. + state["error"] = error_evidence(error) + finally: + state["finished_monotonic_ns"] = time.monotonic_ns() -def _terminate_process(process: subprocess.Popen[bytes]) -> bool: - if process.poll() is not None: - return True - try: - process.terminate() - except OSError: - return process.poll() is not None - try: - process.wait(timeout=2) - return True - except subprocess.TimeoutExpired: +def _terminate_process(process: subprocess.Popen[bytes]) -> dict: + record = {"actions": [], "returncode_before": process.poll()} + + def action(name: str, function: Any) -> bool: + item = {"action": name, "started_monotonic_ns": time.monotonic_ns()} + record["actions"].append(item) try: - process.kill() - except OSError: - return process.poll() is not None - try: - process.wait(timeout=2) + item["result"] = function() return True - except subprocess.TimeoutExpired: + except (OSError, subprocess.TimeoutExpired) as error: + item["error"] = error_evidence(error) return False + finally: + item["finished_monotonic_ns"] = time.monotonic_ns() + + if ( + record["returncode_before"] is None + and action("terminate", process.terminate) + and not action("wait_after_terminate", lambda: process.wait(timeout=2)) + and action("kill", process.kill) + ): + action("wait_after_kill", lambda: process.wait(timeout=2)) + record.update(returncode=process.poll(), exited=process.poll() is not None) + return record + + +def _root_state(handle: TargetServeHandle) -> dict: + return {"returncode": handle.process.poll(), "stat": _stat_read(handle.process.pid)} def _serve_command( @@ -136,7 +159,11 @@ def start_target_serve( sample_interval_seconds: float | None = None, ) -> TargetServeHandle: engine = target_metadata(target)["engine"] - temp_dir = Path(tempfile.mkdtemp(prefix=f"moli-benchmark-{target}-")) if engine == "chrome" else None + temp_dir = ( + Path(tempfile.mkdtemp(prefix=f"moli-benchmark-{target}-")) + if engine == "chrome" + else None + ) logs: list[str] = [] reserved_port = reserve_port() try: @@ -157,48 +184,109 @@ def start_target_serve( if temp_dir is not None: shutil.rmtree(temp_dir, ignore_errors=True) raise - log_threads = [ - threading.Thread(target=_read_available, args=(process.stdout, "stdout", logs), daemon=True), - threading.Thread(target=_read_available, args=(process.stderr, "stderr", logs), daemon=True), - ] - for thread in log_threads: - thread.start() - sampler = ( - ResourceSampler(process.pid) - if sample_interval_seconds is None - else ResourceSampler(process.pid, interval_seconds=sample_interval_seconds) - ) - sampler.start() - handle = TargetServeHandle( - target=target, - process=process, - endpoint=endpoint, - command=command, - logs=logs, - sampler=sampler, - log_threads=log_threads, - port_lease=reserved_port, - temp_dir=temp_dir, - ) - started = time.perf_counter() - deadline = started + timeout_seconds - version_url = endpoint + "/json/version" - while time.perf_counter() < deadline: - if process.poll() is not None: - stop_target_serve(handle) - raise TargetServeProcessExit( - f"{target} CDP server exited early with {process.returncode}: " - f"{'; '.join(logs[-20:])}" + handle = None + stage = "create_handle" + try: + # Publish cleanup ownership before any post-Popen startup operation. + handle = TargetServeHandle( + target=target, + process=process, + endpoint=endpoint, + command=command, + logs=logs, + sampler=None, + log_threads=[], + port_lease=reserved_port, + temp_dir=temp_dir, + ) + stage = "root_identity" + handle.root_identity = _stat_read(process.pid) + handle.log_state = {label: {} for label in ("stdout", "stderr")} + stage = "create_log_threads" + for stream, label in ((process.stdout, "stdout"), (process.stderr, "stderr")): + handle.log_threads.append( + threading.Thread( + target=_read_available, + args=(stream, label, logs, handle.log_state[label]), + daemon=True, + ) ) - if probe_url(version_url): - handle.ready_ms = (time.perf_counter() - started) * 1000.0 - return handle - time.sleep(0.025) - stop_target_serve(handle) - raise TargetServeReadinessTimeout( - f"timed out waiting for {target} CDP server at {version_url}: " - f"{'; '.join(logs[-20:])}" - ) + for label, thread in zip(("stdout", "stderr"), handle.log_threads): + stage = f"start_{label}_reader" + thread.start() + stage = "create_sampler" + handle.sampler = ( + ResourceSampler(process.pid) + if sample_interval_seconds is None + else ResourceSampler(process.pid, interval_seconds=sample_interval_seconds) + ) + stage = "start_sampler" + handle.sampler.start() + stage = "readiness" + started = time.perf_counter() + deadline = started + timeout_seconds + version_url = endpoint + "/json/version" + while time.perf_counter() < deadline: + if process.poll() is not None: + raise TargetServeProcessExit( + f"{target} CDP server exited early with {process.returncode}: " + f"{'; '.join(logs[-20:])}" + ) + if probe_url(version_url): + handle.ready_ms = (time.perf_counter() - started) * 1000.0 + return handle + time.sleep(0.025) + raise TargetServeReadinessTimeout( + f"timed out waiting for {target} CDP server at {version_url}: " + f"{'; '.join(logs[-20:])}" + ) + except BaseException as error: + # Returning a handle is not the boundary of resource ownership. Preserve + # this original startup error even when individual cleanup actions fail. + evidence = {"instrumentation_schema": 2, "events": [], "errors": []} + if handle is not None: + try: + evidence = stop_target_serve(handle, include_resource_samples=True) + except BaseException as cleanup_error: # noqa: BLE001 -- Preserve the original startup exception. + evidence["errors"].append( + {"action": "stop_target_serve", **error_evidence(cleanup_error)} + ) + if handle is None or "termination" not in evidence: + # Handle construction/cleanup itself failed. Use the same process + # termination policy and still attempt the independently owned lease. + def cleanup_action(name, function): + event = {"action": name, "started_monotonic_ns": time.monotonic_ns()} + evidence["events"].append(event) + try: + result = function() + event["result"] = result + return result + except BaseException as cleanup_error: # noqa: BLE001 -- Keep original startup failure. + detail = {"action": name, **error_evidence(cleanup_error)} + event["error"] = detail + evidence["errors"].append(detail) + return None + finally: + event["finished_monotonic_ns"] = time.monotonic_ns() + + evidence["termination"] = cleanup_action( + "terminate_root", lambda: _terminate_process(process) + ) + cleanup_action("release_port", reserved_port.close) + if temp_dir is not None: + cleanup_action("remove_profile", lambda: shutil.rmtree(temp_dir)) + evidence.update( + cleanup_complete=False, + stop_protocol_verified=False, + sampler_sealed_before_termination=False, + descendants_exited="unknown", + returncode=cleanup_action("root_returncode", process.poll), + ) + evidence.update( + startup_stage=stage, startup_error=error_evidence(error), command=command + ) + error.evidence = evidence + raise def stop_target_serve( @@ -208,24 +296,171 @@ def stop_target_serve( ) -> dict[str, Any]: if handle is None: return {} - process_exited = False - resources: dict[str, Any] = {} + report = {"instrumentation_schema": 2, "events": [], "errors": []} + + def call(name, function): + event = {"action": name, "started_monotonic_ns": time.monotonic_ns()} + report["events"].append(event) + try: + result = function() + event["result"] = result + return result + except Exception as error: # noqa: BLE001 -- Record failure and continue bounded cleanup. + evidence = {"action": name, **error_evidence(error)} + report["errors"].append(evidence) + event["error"] = evidence + return None + finally: + event["finished_monotonic_ns"] = time.monotonic_ns() + + before = call("root_before_sampler_stop", lambda: _root_state(handle)) + sampler = handle.sampler + report["sampler_constructed"] = sampler is not None + resources = ( + call("stop_and_seal_sampler", sampler.stop) if sampler is not None else None + ) + first_stop = resources + after = call("root_after_sampler_stop", lambda: _root_state(handle)) + # This changes only teardown ordering. Every already-started sample is kept; + # no periodic tail row is dropped or excused after an engine exit. + termination = call("terminate_root", lambda: _terminate_process(handle.process)) + if sampler is not None and ( + resources is None or resources.get("thread_alive_after_stop") + ): + resources = call("sampler_join_after_process_cleanup", sampler.stop) + resources = dict(resources or {}) + if include_resource_samples: + resources["samples"] = ( + call("snapshot_committed_samples", sampler.export_samples) + if sampler is not None + else None + ) + reader_states = [] + for label, thread in zip(("stdout", "stderr"), handle.log_threads): + call(f"join_{label}", lambda thread=thread: thread.join(timeout=0.2)) + reader_states.append( + { + "stream": label, + **handle.log_state.get(label, {}), + "thread_alive_after_join": thread.is_alive(), + } + ) + call("release_port", handle.port_lease.close) + if handle.temp_dir is not None: + call("remove_profile", lambda: shutil.rmtree(handle.temp_dir)) + expected = (handle.root_identity or {}).get("identity") + alive = expected is not None and all( + point + and point["returncode"] is None + and point["stat"].get("identity") == expected + and point["stat"].get("state") not in ("Z", "X") + for point in (before, after) + ) + sealed_before = bool( + first_stop + and first_stop.get("sealed") is True + and first_stop.get("thread_alive_after_stop") is False + and not first_stop.get("join_ever_timed_out") + ) + readers_complete = len(reader_states) == 2 and all( + state.get("eof") is True + and state.get("error") is None + and not state["thread_alive_after_join"] + for state in reader_states + ) + cleanup = bool( + termination + and termination["exited"] + and resources.get("thread_alive_after_stop") is False + and readers_complete + and not report["errors"] + ) + report.update( + root_identity=handle.root_identity, + root_observed_alive_at_stop_boundaries=alive, + sampler_sealed_before_termination=sealed_before, + stop_protocol_verified=alive + and sealed_before + and resources.get("observer_error") is None + and not report["errors"], + cleanup_complete=cleanup, + descendants_exited="unknown", + termination=termination, + log_readers=reader_states, + log_lines_dropped=max( + 0, + sum(state.get("lines_read", 0) for state in reader_states) + - len(handle.logs), + ), + returncode=handle.process.returncode, + resources=resources, + log_tail=list(handle.logs[-40:]), + ) + return report + + +async def finish_target_serve_run(client, handle, record, *, key="server") -> None: + """Preserve workload and close failures independently; always stop the target.""" + record["business_ok"] = bool(record.get("ok")) + errors = record.setdefault("cleanup_errors", []) + record.setdefault(key, {}) try: - process_exited = _terminate_process(handle.process) - if process_exited: - for thread in handle.log_threads: - thread.join(timeout=0.2) - else: - _append_log(handle.logs, "process did not exit after SIGKILL; skipped pipe drain") - resources = handle.sampler.stop() - if include_resource_samples: - resources["samples"] = list(handle.sampler.samples) + if client is not None: + try: + await client.websocket.close() + except asyncio.CancelledError as error: + errors.append({"action": "websocket_close", **error_evidence(error)}) + raise + except Exception as error: # noqa: BLE001 -- Record failure and continue bounded cleanup. + errors.append({"action": "websocket_close", **error_evidence(error)}) finally: - handle.port_lease.close() - if handle.temp_dir is not None: - shutil.rmtree(handle.temp_dir, ignore_errors=True) - return { - "returncode": handle.process.returncode, - "resources": resources, - "log_tail": handle.logs[-40:], - } + # Cancellation at the awaited close must not skip the synchronous + # sampler seal/engine stop attempts or overwrite the workload evidence. + try: + if handle is not None: + record[key] = stop_target_serve(handle, include_resource_samples=True) + else: + record.setdefault(key, {}) + except Exception as error: # noqa: BLE001 -- Record failure and continue bounded cleanup. + errors.append({"action": "stop_target_serve", **error_evidence(error)}) + stopped = record[key] + resources = stopped.get("resources", {}) + record["instrumentation_ok"] = bool( + stopped.get("stop_protocol_verified") + and stopped.get("cleanup_complete") + and resources.get("pss_complete") + and resources.get("resource_reads_complete") + and not errors + and record.get("checkpoint_reads_complete", False) + and record.get("cpu_evidence_complete", False) + and record.get("affinity_evidence_complete", False) + ) + record["ok"] = record["business_ok"] and record["instrumentation_ok"] + # Caller saves the complete failed sample before stopping later cohorts. + record["safe_to_continue"] = bool( + stopped.get("cleanup_complete") + and stopped.get("stop_protocol_verified") + and not resources.get("join_ever_timed_out") + and not resources.get("observer_error") + and not errors + ) + + +def stop_fixture_server(server, thread) -> dict: + record = {"actions": [], "errors": []} + for name, action in ( + ("shutdown", server.shutdown), + ("server_close", server.server_close), + ("join", lambda: thread.join(timeout=2)), + ): + event = {"action": name, "started_monotonic_ns": time.monotonic_ns()} + record["actions"].append(event) + try: + action() + except Exception as error: # noqa: BLE001 -- Record failure and continue bounded cleanup. + record["errors"].append({"action": name, **error_evidence(error)}) + finally: + event["finished_monotonic_ns"] = time.monotonic_ns() + record["thread_alive_after_join"] = thread.is_alive() + record["complete"] = not record["thread_alive_after_join"] and not record["errors"] + return record diff --git a/moli-benchmark/tests/test_cdp_failure_evidence.py b/moli-benchmark/tests/test_cdp_failure_evidence.py new file mode 100644 index 0000000000..b45a60b1c7 --- /dev/null +++ b/moli-benchmark/tests/test_cdp_failure_evidence.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +import asyncio +import errno +import json +import unittest +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from moli_benchmark import raw_cdp, target_serve + +EXCEPTION_EVENT = { + "method": "Runtime.exceptionThrown", + "params": {"exceptionDetails": {"text": "observed before command failure"}}, +} +BINDING_EVENT = {"method": "Runtime.bindingCalled", "params": {"payload": "observed"}} + + +class FakeWebSocket: + def __init__(self, incoming, *, close_error=None): + self.incoming = list(incoming) + self.close_error = close_error + self.sent = [] + self.read_count = 0 + self.close_count = 0 + + async def send(self, message): + self.sent.append(json.loads(message)) + + async def recv(self): + self.read_count += 1 + if not self.incoming: + raise AssertionError("unexpected extra receive") + item = self.incoming.pop(0) + if isinstance(item, BaseException): + raise item + return json.dumps(item) + + async def close(self): + self.close_count += 1 + if self.close_error is not None: + raise self.close_error + + +class SequentialCdpEvidenceTests(unittest.IsolatedAsyncioTestCase): + async def test_success_returns_same_ordered_events_without_extra_receive(self): + response = {"id": 7, "result": {"value": 42}} + websocket = FakeWebSocket([EXCEPTION_EVENT, BINDING_EVENT, response]) + reply, seen = await raw_cdp.RawCdpClient(websocket).recv_until_id(7) + self.assertEqual(reply, response) + self.assertEqual(seen, [EXCEPTION_EVENT, BINDING_EVENT, response]) + self.assertEqual(websocket.read_count, 3) + + async def test_cdp_error_retains_events_and_error_response(self): + failed_reply = {"id": 7, "error": {"code": -32000, "message": "rejected"}} + websocket = FakeWebSocket([EXCEPTION_EVENT, BINDING_EVENT, failed_reply]) + with self.assertRaises(raw_cdp.RawCdpError) as caught: + await raw_cdp.RawCdpClient(websocket).recv_until_id( + 7, method="Runtime.evaluate" + ) + self.assertIs(type(caught.exception), raw_cdp.RawCdpError) + self.assertEqual( + caught.exception.cdp_receive_evidence, + { + "message_id": 7, + "method": "Runtime.evaluate", + "messages": [EXCEPTION_EVENT, BINDING_EVENT, failed_reply], + }, + ) + self.assertEqual(websocket.read_count, 3) + self.assertIn("rejected", str(caught.exception)) + + async def test_transport_timeout_and_cancel_preserve_original_exception_object( + self, + ): + for failure in ( + EOFError("closed after event"), + TimeoutError("receive timed out after event"), + asyncio.CancelledError("receive canceled after event"), + ): + with self.subTest(kind=type(failure).__name__): + websocket = FakeWebSocket([EXCEPTION_EVENT, failure]) + with self.assertRaises(type(failure)) as caught: + await raw_cdp.RawCdpClient(websocket).recv_until_id( + 7, method="Runtime.evaluate" + ) + self.assertIs(caught.exception, failure) + self.assertEqual( + failure.cdp_receive_evidence["messages"], [EXCEPTION_EVENT] + ) + self.assertEqual(websocket.read_count, 2) + + async def test_deadline_expiry_retains_event_without_another_receive(self): + # A fake monotonic clock expires after the first received event; no sleep. + loop = SimpleNamespace(time=Mock(side_effect=[0, 0, 11])) + client = raw_cdp.RawCdpClient(FakeWebSocket([])) + + async def receive(): + return EXCEPTION_EVENT + + async def direct_wait(awaitable, *, timeout): + self.assertEqual(timeout, 10) + return await awaitable + + client.recv = receive + with ( + patch.object(raw_cdp.asyncio, "get_running_loop", return_value=loop), + patch.object(raw_cdp.asyncio, "wait_for", side_effect=direct_wait), + self.assertRaises(raw_cdp.RawCdpTimeoutError) as caught, + ): + await client.recv_until_id(7, method="Runtime.evaluate") + self.assertEqual( + caught.exception.cdp_receive_evidence["messages"], [EXCEPTION_EVENT] + ) + + +class CancelledCleanupEvidenceTests(unittest.IsolatedAsyncioTestCase): + async def test_cancelled_close_runs_real_stop_order_and_preserves_primary_error( + self, + ): + cancellation = asyncio.CancelledError("close canceled") + websocket = FakeWebSocket([], close_error=cancellation) + calls = [] + process = SimpleNamespace(pid=10, returncode=None) + process.poll = lambda: process.returncode + + def seal(): + self.assertIsNone(process.returncode) + calls.append("seal") + return {"sealed": True, "thread_alive_after_stop": False} + + def terminate(_process): + calls.append("terminate") + process.returncode = 143 + return {"exited": True, "returncode": 143} + + thread = SimpleNamespace(join=lambda timeout: None, is_alive=lambda: False) + handle = SimpleNamespace( + process=process, + root_identity={"identity": "10:100"}, + sampler=SimpleNamespace( + stop=seal, export_samples=lambda: [{"sequence": 1}] + ), + logs=[], + log_threads=[thread, thread], + log_state={ + name: {"eof": True, "error": None} for name in ("stdout", "stderr") + }, + port_lease=SimpleNamespace(close=lambda: calls.append("release")), + temp_dir=None, + ) + record = {"ok": False, "error": "original workload failure"} + with ( + patch.object( + target_serve, + "_stat_read", + return_value={"identity": "10:100", "state": "S"}, + ), + patch.object(target_serve, "_terminate_process", side_effect=terminate), + self.assertRaises(asyncio.CancelledError) as caught, + ): + await target_serve.finish_target_serve_run( + SimpleNamespace(websocket=websocket), handle, record + ) + self.assertIs(caught.exception, cancellation) + self.assertEqual(calls, ["seal", "terminate", "release"]) + self.assertEqual(record["error"], "original workload failure") + self.assertEqual(record["cleanup_errors"][0]["type"], "CancelledError") + self.assertEqual(record["server"]["resources"]["samples"], [{"sequence": 1}]) + self.assertTrue(record["server"]["cleanup_complete"]) + self.assertFalse(record["ok"]) + self.assertFalse(record["safe_to_continue"]) + + async def test_stop_error_does_not_replace_close_cancellation_or_primary_error( + self, + ): + cancellation = asyncio.CancelledError("close canceled") + record = {"ok": False, "error": "original workload failure"} + with ( + patch.object( + target_serve, + "stop_target_serve", + side_effect=OSError(errno.EIO, "stop failed"), + ) as stop, + self.assertRaises(asyncio.CancelledError) as caught, + ): + await target_serve.finish_target_serve_run( + SimpleNamespace(websocket=FakeWebSocket([], close_error=cancellation)), + object(), + record, + ) + self.assertIs(caught.exception, cancellation) + stop.assert_called_once() + self.assertEqual(record["error"], "original workload failure") + self.assertEqual( + [item["action"] for item in record["cleanup_errors"]], + ["websocket_close", "stop_target_serve"], + ) + self.assertEqual(record["cleanup_errors"][1]["errno"], errno.EIO) + self.assertFalse(record["safe_to_continue"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/moli-benchmark/tests/test_instrumentation_evidence.py b/moli-benchmark/tests/test_instrumentation_evidence.py new file mode 100644 index 0000000000..58c5b0979f --- /dev/null +++ b/moli-benchmark/tests/test_instrumentation_evidence.py @@ -0,0 +1,460 @@ +from __future__ import annotations + +import asyncio +import errno +import io +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from moli_benchmark import sampling, target_serve + + +def stat(pid=10, tid=None, start=100, ticks=5): + return { + "status": "ok", + "identity": f"{pid}:{start}" if tid is None else f"{pid}:{tid}:{start}", + "pid": pid, + "tid": tid, + "start_ticks": start, + "state": "S", + "ppid": 1, + "user_ticks": ticks, + "system_ticks": 0, + "total_ticks": ticks, + } + + +def cpu(threads, ticks=5, identity="10:100"): + root = {**stat(ticks=ticks), "identity": identity} + return { + "threads": threads, + "observations": [{"valid": True}], + "enumeration_error": None, + "root_before": root, + "root_after": root, + } + + +class InstrumentationEvidenceTests(unittest.TestCase): + def test_resource_error_keeps_errno_and_missing_is_not_zero(self): + with patch.object( + sampling.Path, + "read_text", + side_effect=OSError(errno.EACCES, "denied", "/proc/10/status"), + ): + read = sampling._text_read(sampling.Path("/proc/10/status")) + self.assertEqual(read["error"]["errno"], errno.EACCES) + self.assertEqual(sampling._field_read(read, "VmRSS:")["status"], "os_error") + self.assertEqual( + sampling._field_read({"status": "ok", "raw": "Pss: 0 kB"}, "Pss:", 1024)[ + "value" + ], + 0, + ) + self.assertEqual( + sampling._field_read({"status": "ok", "raw": "Rss: 1 kB"}, "Pss:")[ + "status" + ], + "missing_field", + ) + self.assertEqual( + sampling._field_read({"status": "ok", "raw": "Pss: -1 kB"}, "Pss:")[ + "status" + ], + "parse_error", + ) + + def test_stat_keeps_raw_and_uses_last_comm_parenthesis(self): + fields = ["S", "1"] + ["0"] * 18 + fields[11], fields[12], fields[19] = "7", "9", "123" + raw = "10 (name with ) spaces) " + " ".join(fields) + with patch.object(sampling.Path, "read_text", return_value=raw): + value = sampling._stat_read(10) + self.assertEqual(value["raw"], raw) + self.assertEqual(value["identity"], "10:123") + self.assertEqual(value["total_ticks"], 16) + with patch.object( + sampling.Path, "read_text", return_value=raw.replace("10 (", "11 (", 1) + ): + self.assertEqual(sampling._stat_read(10)["status"], "parse_error") + + def test_changed_root_identity_invalidates_process_total_and_thread_delta(self): + result = sampling.cpu_delta( + cpu({"10:7:100": 100}), + cpu({"10:7:100": 200}, identity="10:200"), + {"ticks_per_second": 100}, + "10:100", + ) + self.assertIsNone(result["cpu_ms"]) + self.assertIsNone(result["process_cpu"]["delta_ticks"]) + + def test_tid_reuse_and_retired_threads_do_not_corrupt_independent_process_total( + self, + ): + before = cpu({"10:7:100": 1_000_000}, ticks=10) + after = cpu({"10:7:200": 200_000}, ticks=13) + result = sampling.cpu_delta(before, after, {"ticks_per_second": 100}, "10:100") + self.assertEqual(result["retired_threads"], ["10:7:100"]) + self.assertEqual(result["created_threads"], ["10:7:200"]) + self.assertEqual(result["cpu_ms"], 0.2) + self.assertEqual(result["process_cpu"]["delta_ticks"], 3) + self.assertEqual(result["process_cpu"]["cpu_ms"], 30) + self.assertEqual(result["thread_cpu_observed"]["before"], before) + self.assertEqual(result["thread_cpu_observed"]["after"], after) + + def test_incomplete_thread_map_never_silently_substitutes_process_cpu(self): + before, after = cpu({"10:7:100": 1}), cpu({}, ticks=6) + after["observations"] = [{"valid": False}] + result = sampling.cpu_delta(before, after, {"ticks_per_second": 100}, "10:100") + self.assertIsNone(result["cpu_ms"]) + self.assertEqual(result["process_cpu"]["cpu_ms"], 10) + self.assertFalse(result["thread_cpu_observed"]["boundary_read_complete"]) + + def test_regression_is_not_clamped_and_clock_failure_is_not_100hz(self): + value = sampling.cpu_delta( + cpu({"10:7:100": 20}), + cpu({"10:7:100": 5}, ticks=4), + {"ticks_per_second": 100}, + "10:100", + ) + self.assertIsNone(value["cpu_ms"]) + self.assertIsNone(value["process_cpu"]["delta_ticks"]) + self.assertEqual( + value["thread_cpu_observed"]["counter_regressions"], ["10:7:100"] + ) + with patch.object( + sampling.os, "sysconf", side_effect=OSError(errno.EINVAL, "bad clock") + ): + clock = sampling.clock_ticks_evidence() + self.assertIsNone(clock["ticks_per_second"]) + self.assertEqual(clock["error"]["errno"], errno.EINVAL) + value = sampling.cpu_delta(cpu({}), cpu({}, ticks=6), clock, "10:100") + self.assertIsNone(value["process_cpu"]["cpu_ms"]) + self.assertEqual(value["process_cpu"]["delta_ticks"], 1) + + def test_zero_ticks_is_not_claimed_as_zero_execution_or_ns_precision(self): + value = sampling.cpu_delta( + cpu({"10:7:100": 100}), + cpu({"10:7:100": 200}), + {"ticks_per_second": 100, "counter_tick_ms": 10}, + "10:100", + ) + self.assertEqual(value["process_cpu"]["cpu_ms"], 0) + self.assertEqual(value["cpu_ms"], 0.0001) + self.assertEqual( + value["process_cpu"]["zero_ticks_mean"], + "no_exported_tick_change_not_zero_cpu", + ) + self.assertEqual( + value["process_cpu"]["remote_process_clock"]["status"], + "unsupported_not_runtime_validated", + ) + + def test_affinity_has_actual_values_identity_and_errno(self): + task = SimpleNamespace(name="7") + with ( + patch.object(sampling.Path, "iterdir", return_value=[task]), + patch.object( + sampling, "_stat_read", side_effect=lambda pid, tid=None: stat(pid, tid) + ), + patch.object(sampling.os, "sched_getaffinity", return_value={0}), + ): + result = sampling.affinity_snapshot(10, 0, "10:100") + self.assertTrue(result["all_observed_match"]) + self.assertEqual(result["threads"][0]["before"]["identity"], "10:7:100") + with ( + patch.object(sampling.Path, "iterdir", return_value=[task]), + patch.object( + sampling, "_stat_read", side_effect=lambda pid, tid=None: stat(pid, tid) + ), + patch.object( + sampling.os, + "sched_getaffinity", + side_effect=OSError(errno.ESRCH, "gone"), + ), + ): + result = sampling.affinity_snapshot(10, 0, "10:100") + self.assertFalse(result["boundary_complete"]) + self.assertEqual(result["threads"][0]["error"]["errno"], errno.ESRCH) + + def test_cpu_snapshot_retains_disappeared_thread_observation(self): + tasks = [SimpleNamespace(name="7"), SimpleNamespace(name="8")] + + def read(path): + if str(path).endswith("8/schedstat"): + return { + "status": "os_error", + "raw": None, + "error": {"errno": errno.ENOENT}, + } + return {"status": "ok", "raw": "123 0 0"} + + with ( + patch.object(sampling.Path, "iterdir", return_value=tasks), + patch.object( + sampling, "_stat_read", side_effect=lambda pid, tid=None: stat(pid, tid) + ), + patch.object(sampling, "_text_read", side_effect=read), + ): + result = sampling.cpu_snapshot(10) + self.assertEqual(result["enumerated_tids"], [7, 8]) + self.assertEqual(result["threads"], {"10:7:100": 123}) + self.assertFalse(result["observations"][1]["valid"]) + self.assertEqual( + result["observations"][1]["runtime"]["error"]["errno"], errno.ENOENT + ) + + def test_sampler_seals_after_inflight_sample_and_snapshots_are_independent(self): + entered, release = threading.Event(), threading.Event() + + def snapshot(*_args, **_kwargs): + entered.set() + self.assertTrue(release.wait(1)) + return { + "processes": [{"identity_valid": True, "identity_after": stat()}], + "process_count": 1, + "pss_bytes": 3, + "pss_process_count": 1, + "rss_bytes": 4, + "thread_count": 1, + "fd_count": 1, + } + + sampler = sampling.ResourceSampler(10, interval_seconds=0.01) + with ( + patch.object(sampling, "snapshot_resources", side_effect=snapshot), + patch.object( + sampling, "clock_ticks_evidence", return_value={"ticks_per_second": 100} + ), + ): + sampler.start() + self.assertTrue(entered.wait(1)) + sampler._stop.set() + release.set() + summary = sampler.stop() + self.assertTrue(summary["sealed"]) + self.assertFalse(summary["thread_alive_after_stop"]) + self.assertEqual(summary["sample_count"], 1) + sample = sampler.export_samples() + sample[0]["pss_bytes"] = 999 + self.assertEqual(sampler.export_samples()[0]["pss_bytes"], 3) + self.assertLessEqual( + summary["last_committed_monotonic_ns"], + summary["thread_exited_monotonic_ns"], + ) + self.assertLessEqual( + summary["thread_exited_monotonic_ns"], summary["join_finished_monotonic_ns"] + ) + + def test_join_timeout_remains_a_failure_after_later_success(self): + sampler = sampling.ResourceSampler(10) + thread = Mock() + thread.is_alive.return_value = True + sampler._thread = thread + first = sampler.stop() + self.assertFalse(first["sealed"]) + self.assertTrue(first["join_ever_timed_out"]) + thread.is_alive.return_value = False + last = sampler.stop() + self.assertTrue(last["sealed"]) + self.assertTrue(last["join_ever_timed_out"]) + + def test_log_read_errors_and_eof_are_distinct(self): + state, logs = {}, [] + target_serve._read_available(io.BytesIO(b"a\nb\n"), "stdout", logs, state) + self.assertTrue(state["eof"]) + self.assertEqual(state["lines_read"], 2) + target_serve._read_available( + SimpleNamespace( + readline=Mock(side_effect=OSError(errno.EIO, "read failed")) + ), + "stderr", + [], + state := {}, + ) + self.assertFalse(state["eof"]) + self.assertEqual(state["error"]["errno"], errno.EIO) + + def test_stop_seals_before_term_and_logs_require_eof(self): + calls = [] + process = SimpleNamespace( + pid=10, returncode=None, poll=lambda: process.returncode + ) + + def stop(): + calls.append("stop") + self.assertIsNone(process.returncode) + return { + "sealed": True, + "thread_alive_after_stop": False, + "join_ever_timed_out": False, + "pss_complete": True, + } + + def terminate(_): + calls.append("terminate") + process.returncode = 143 + return {"exited": True, "returncode": 143} + + thread = SimpleNamespace(join=lambda timeout: None, is_alive=lambda: False) + handle = SimpleNamespace( + process=process, + root_identity=stat(), + sampler=SimpleNamespace(stop=stop, export_samples=list), + logs=[], + log_threads=[thread, thread], + log_state={ + name: {"eof": True, "error": None, "lines_read": 0} + for name in ("stdout", "stderr") + }, + port_lease=SimpleNamespace(close=lambda: calls.append("release")), + temp_dir=None, + ) + with ( + patch.object(target_serve, "_stat_read", return_value=stat()), + patch.object(target_serve, "_terminate_process", side_effect=terminate), + ): + result = target_serve.stop_target_serve( + handle, include_resource_samples=True + ) + self.assertEqual(calls, ["stop", "terminate", "release"]) + self.assertTrue(result["stop_protocol_verified"]) + self.assertTrue(result["cleanup_complete"]) + self.assertEqual(result["returncode"], 143) + self.assertEqual(result["descendants_exited"], "unknown") + + def test_async_close_failure_does_not_skip_stop_or_overwrite_workload_error(self): + async def failed_close(): + raise OSError(errno.EIO, "socket close failed") + + record = {"ok": False, "error": "original workload failure"} + with patch.object( + target_serve, "stop_target_serve", return_value={"cleanup_complete": True} + ) as stop: + asyncio.run( + target_serve.finish_target_serve_run( + SimpleNamespace(websocket=SimpleNamespace(close=failed_close)), + object(), + record, + ) + ) + stop.assert_called_once() + self.assertEqual(record["error"], "original workload failure") + self.assertEqual(record["cleanup_errors"][0]["errno"], errno.EIO) + self.assertFalse(record["ok"]) + self.assertFalse(record["safe_to_continue"]) + + def test_unobserved_tree_root_stays_in_selected_set(self): + failed = { + "status": "os_error", + "identity": None, + "error": {"errno": errno.ENOENT}, + } + with ( + patch.object(sampling.Path, "glob", return_value=[]), + patch.object(sampling, "_stat_read", return_value=failed), + ): + pids, records, evidence = sampling._process_tree_evidence(10) + self.assertEqual(pids, [10]) + self.assertEqual(records[10], failed) + self.assertEqual(evidence["all_descendants_observed"], "unknown") + + def test_resource_identity_change_prevents_apparently_complete_totals(self): + with ( + patch.object( + sampling, + "_process_tree_evidence", + return_value=([10], {10: stat()}, {}), + ), + patch.object(sampling, "_stat_read", return_value=stat(start=200)), + patch.object( + sampling, + "_text_read", + return_value={ + "status": "ok", + "raw": "Pss: 1 kB\nVmRSS: 1 kB\nThreads: 1", + }, + ), + patch.object( + sampling, "_fd_read", return_value={"status": "ok", "value": 1} + ), + ): + value = sampling.snapshot_resources(10, include_lifetime_cpu=False) + self.assertEqual(value["process_count"], 1) + self.assertIsNone(value["pss_bytes"]) + self.assertEqual(value["pss_process_count"], 0) + self.assertFalse(value["processes"][0]["identity_valid"]) + + def test_readiness_failure_carries_stop_and_eof_evidence(self): + process = SimpleNamespace( + pid=10, returncode=None, stdout=io.BytesIO(), stderr=io.BytesIO() + ) + process.poll = lambda: process.returncode + process.terminate = lambda: setattr(process, "returncode", 143) + process.wait = lambda timeout: process.returncode + + class Thread: + def __init__(self, *, target, args, daemon): + self.target, self.args = target, args + + def start(self): + self.target(*self.args) + + def join(self, timeout): + pass + + def is_alive(self): + return False + + sampler = SimpleNamespace( + start=lambda: None, + stop=lambda: { + "sealed": True, + "thread_alive_after_stop": False, + "pss_complete": False, + }, + export_samples=list, + ) + lease = SimpleNamespace(port=23456, release_socket=Mock(), close=Mock()) + with ( + patch.object(target_serve, "reserve_port", return_value=lease), + patch.object(target_serve, "_stat_read", return_value=stat()), + patch.object(target_serve.subprocess, "Popen", return_value=process), + patch.object(target_serve.threading, "Thread", Thread), + patch.object(target_serve, "ResourceSampler", return_value=sampler), + self.assertRaises(target_serve.TargetServeReadinessTimeout) as caught, + ): + target_serve.start_target_serve("moli", Path("/unused"), 0) + value = caught.exception.evidence + self.assertTrue(value["sampler_sealed_before_termination"]) + self.assertTrue(value["cleanup_complete"]) + self.assertEqual(value["returncode"], 143) + self.assertEqual(len(value["log_readers"]), 2) + self.assertEqual(value["resources"]["samples"], []) + lease.close.assert_called_once() + + def test_fixture_cleanup_joins_even_when_close_fails(self): + calls = [] + + def failed(): + calls.append("close") + raise OSError(errno.EIO, "close") + + value = target_serve.stop_fixture_server( + SimpleNamespace( + shutdown=lambda: calls.append("shutdown"), server_close=failed + ), + SimpleNamespace( + join=lambda timeout: calls.append("join"), is_alive=lambda: False + ), + ) + self.assertEqual(calls, ["shutdown", "close", "join"]) + self.assertFalse(value["complete"]) + self.assertEqual(value["errors"][0]["errno"], errno.EIO) + + +if __name__ == "__main__": + unittest.main() diff --git a/moli-benchmark/tests/test_sampling.py b/moli-benchmark/tests/test_sampling.py index b131d4c145..6de70f03af 100644 --- a/moli-benchmark/tests/test_sampling.py +++ b/moli-benchmark/tests/test_sampling.py @@ -9,21 +9,41 @@ from moli_benchmark.sampling import ResourceSampler, snapshot_resources class ResourceSamplingTests(unittest.TestCase): def test_snapshot_does_not_report_partial_process_tree_totals(self) -> None: + def stat(pid): + return {"identity": f"{pid}:100", "status": "ok", "total_ticks": 0} + + def read(path): + raw = ( + "VmRSS: 1 kB\nThreads: 2" + if str(path).endswith("status") + else "Pss: 1 kB" + ) + if str(path) == "/proc/11/smaps_rollup": + return {"status": "os_error", "error": {"errno": 2}, "raw": None} + if str(path) == "/proc/11/status": + raw = "VmRSS: 1 kB" + return {"status": "ok", "raw": raw} + with ( - patch("moli_benchmark.sampling._process_tree", return_value=[10, 11]), - patch("moli_benchmark.sampling._read_pss_bytes", side_effect=[100, None]), - patch("moli_benchmark.sampling._read_rss_bytes", side_effect=[200, 300]), - patch("moli_benchmark.sampling._read_thread_count", side_effect=[2, None]), - patch("moli_benchmark.sampling._read_fd_count", side_effect=[3, 4]), + patch( + "moli_benchmark.sampling._process_tree_evidence", + return_value=([10, 11], {10: stat(10), 11: stat(11)}, {}), + ), + patch("moli_benchmark.sampling._text_read", side_effect=read), + patch("moli_benchmark.sampling._stat_read", side_effect=stat), + patch( + "moli_benchmark.sampling._fd_read", + return_value={"status": "ok", "value": 3}, + ), patch("moli_benchmark.sampling._read_cpu_percent", return_value=12.5), ): sample = snapshot_resources(10) self.assertIsNone(sample["pss_bytes"]) self.assertEqual(sample["pss_process_count"], 1) - self.assertEqual(sample["rss_bytes"], 500) + self.assertEqual(sample["rss_bytes"], 2048) self.assertIsNone(sample["thread_count"]) - self.assertEqual(sample["fd_count"], 7) + self.assertEqual(sample["fd_count"], 6) def test_summary_records_sampler_health_and_complete_pss_coverage(self) -> None: sampler = ResourceSampler(1, interval_seconds=0.5) @@ -71,7 +91,11 @@ class ResourceSamplingTests(unittest.TestCase): "moli_benchmark.sampling.snapshot_resources", side_effect=fail_snapshot, ): - sampler.start() + with patch( + "moli_benchmark.sampling.clock_ticks_evidence", + return_value={"ticks_per_second": 100}, + ): + sampler.start() self.assertTrue(attempted.wait(timeout=1)) summary = sampler.stop() diff --git a/moli-benchmark/tests/test_target_serve.py b/moli-benchmark/tests/test_target_serve.py index 21ff1fd7e1..ac1b6f6561 100644 --- a/moli-benchmark/tests/test_target_serve.py +++ b/moli-benchmark/tests/test_target_serve.py @@ -37,6 +37,9 @@ class _FakeSampler: def start(self) -> None: return None + def export_samples(self): + return list(self.samples) + def stop(self) -> dict[str, int]: return {"pid": self.pid} @@ -75,7 +78,9 @@ class TargetServeTests(unittest.TestCase): stop_target_serve(handle) self.assertNotIn(port, config._RESERVED_PORTS) - def test_start_target_serve_enables_layout_and_all_resource_fetch_for_full_moli(self) -> None: + def test_start_target_serve_enables_layout_and_all_resource_fetch_for_full_moli( + self, + ) -> None: process = _FakeProcess() with ( diff --git a/moli-benchmark/tests/test_target_start_failure_evidence.py b/moli-benchmark/tests/test_target_start_failure_evidence.py new file mode 100644 index 0000000000..2f3e615562 --- /dev/null +++ b/moli-benchmark/tests/test_target_start_failure_evidence.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import asyncio +import errno +import io +import subprocess +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from moli_benchmark import sampling, target_serve + + +class TargetStartFailureEvidenceTests(unittest.TestCase): + def run_start_failure( + self, stage, *, fail_release=False, startup_failure=None, term_timeout=False + ): + failure = startup_failure or RuntimeError(f"initial {stage} failure") + calls = [] + process = SimpleNamespace( + pid=10, returncode=None, stdout=io.BytesIO(), stderr=io.BytesIO() + ) + process.poll = lambda: process.returncode + + def terminate(): + calls.append("terminate") + if not term_timeout: + process.returncode = -15 + + def wait(timeout): + calls.append("wait") + if process.returncode is None: + raise subprocess.TimeoutExpired("fake-engine", timeout) + return process.returncode + + def kill(): + calls.append("kill") + process.returncode = -9 + + def release(): + calls.append("release_port") + if fail_release: + raise OSError(errno.EIO, "port release failed") + + process.terminate, process.wait, process.kill = terminate, wait, kill + lease = SimpleNamespace( + port=23456, + release_socket=lambda: calls.append("release_socket"), + close=release, + ) + + class Thread: + def __init__(self, *, target, args=(), daemon, name=None): + self.target, self.args = target, args + self.label = args[1] if args else "sampler" + self.started = False + + def start(self): + calls.append(f"start_{self.label}") + if stage == f"start_{self.label}": + raise failure + self.started = True + # Only fake stdout/stderr BytesIO readers may run. Never sample. + if self.label == "sampler": + raise AssertionError("test must fail before starting a sampler") + self.target(*self.args) + + def join(self, timeout): + calls.append(f"join_{self.label}") + if not self.started: + raise RuntimeError("cannot join thread before it is started") + + def is_alive(self): + return False + + class Sampler(sampling.ResourceSampler): + def stop(self): + calls.append("stop_sampler") + return super().stop() + + def sampler_factory(pid, **kwargs): + calls.append("create_sampler") + if stage == "create_sampler": + raise failure + return Sampler(pid, **kwargs) + + with ( + patch.object(target_serve, "reserve_port", return_value=lease), + patch.object( + target_serve, + "_stat_read", + return_value={"identity": "10:100", "state": "S"}, + ), + patch.object(target_serve.subprocess, "Popen", return_value=process), + patch.object(target_serve.threading, "Thread", Thread), + patch.object(target_serve, "ResourceSampler", side_effect=sampler_factory), + patch.object( + sampling, "clock_ticks_evidence", return_value={"ticks_per_second": 100} + ), + patch.object( + target_serve, + "probe_url", + side_effect=AssertionError("readiness must not run"), + ), + self.assertRaises(type(failure)) as caught, + ): + target_serve.start_target_serve("moli", Path("/never-executed"), 10) + self.assertIs(caught.exception, failure) + evidence = failure.evidence + self.assertEqual(evidence["startup_error"]["message"], str(failure)) + self.assertIn("terminate", calls) + self.assertIn("release_port", calls) + self.assertLess(calls.index("terminate"), calls.index("release_port")) + self.assertFalse(evidence["stop_protocol_verified"]) + self.assertFalse(evidence["sampler_sealed_before_termination"]) + self.assertFalse(evidence["cleanup_complete"]) + return failure, calls, evidence + + def test_sampler_start_failure_records_unstarted_join_but_terminates_and_releases( + self, + ): + _failure, calls, evidence = self.run_start_failure("start_sampler") + self.assertEqual(evidence["startup_stage"], "start_sampler") + self.assertTrue(evidence["sampler_constructed"]) + self.assertEqual(calls.count("stop_sampler"), 2) + self.assertLess(calls.index("stop_sampler"), calls.index("terminate")) + self.assertEqual(evidence["resources"]["samples"], []) + self.assertEqual(evidence["returncode"], -15) + self.assertEqual( + [item["action"] for item in evidence["errors"]], + ["stop_and_seal_sampler", "sampler_join_after_process_cleanup"], + ) + self.assertTrue( + all(item["type"] == "RuntimeError" for item in evidence["errors"]) + ) + + def test_partial_log_start_failure_records_missing_eof_and_does_not_invent_sampler( + self, + ): + _failure, calls, evidence = self.run_start_failure("start_stderr") + self.assertEqual(evidence["startup_stage"], "start_stderr_reader") + self.assertFalse(evidence["sampler_constructed"]) + self.assertNotIn("create_sampler", calls) + self.assertIsNone(evidence["resources"]["samples"]) + self.assertTrue(evidence["log_readers"][0]["eof"]) + self.assertNotIn("eof", evidence["log_readers"][1]) + self.assertEqual(evidence["errors"][0]["action"], "join_stderr") + self.assertIn("cannot join", evidence["errors"][0]["message"]) + + def test_sampler_constructor_failure_still_cleans_engine_and_both_readers(self): + _failure, calls, evidence = self.run_start_failure("create_sampler") + self.assertEqual(evidence["startup_stage"], "create_sampler") + self.assertNotIn("stop_sampler", calls) + self.assertFalse(evidence["sampler_constructed"]) + self.assertIsNone(evidence["resources"]["samples"]) + self.assertTrue(all(reader["eof"] for reader in evidence["log_readers"])) + + def test_cleanup_errors_and_term_timeout_do_not_mask_initial_failure(self): + _failure, calls, evidence = self.run_start_failure( + "start_sampler", fail_release=True, term_timeout=True + ) + self.assertIn("kill", calls) + self.assertEqual(evidence["returncode"], -9) + self.assertEqual(evidence["errors"][-1]["action"], "release_port") + self.assertEqual(evidence["errors"][-1]["errno"], errno.EIO) + self.assertEqual( + evidence["termination"]["actions"][1]["error"]["type"], "TimeoutExpired" + ) + + def test_startup_cancellation_keeps_same_object_with_cleanup_evidence(self): + cancellation = asyncio.CancelledError("startup canceled") + failure, _calls, evidence = self.run_start_failure( + "start_sampler", startup_failure=cancellation + ) + self.assertIs(failure, cancellation) + self.assertEqual(evidence["startup_error"]["type"], "CancelledError") + self.assertEqual(evidence["returncode"], -15) + + def test_handle_construction_failure_uses_existing_term_policy_and_records_cleanup_failure( + self, + ): + failure = RuntimeError("cannot initialize handle") + process = SimpleNamespace( + pid=10, poll=Mock(side_effect=OSError(errno.EIO, "poll failed")) + ) + lease = SimpleNamespace( + port=23456, + release_socket=Mock(), + close=Mock(side_effect=OSError(errno.EIO, "close failed")), + ) + with ( + patch.object(target_serve, "reserve_port", return_value=lease), + patch.object(target_serve.subprocess, "Popen", return_value=process), + patch.object(target_serve, "TargetServeHandle", side_effect=failure), + patch.object( + target_serve, + "_terminate_process", + return_value={"exited": True, "returncode": -15}, + ) as terminate, + self.assertRaises(RuntimeError) as caught, + ): + target_serve.start_target_serve("moli", Path("/never-executed"), 10) + self.assertIs(caught.exception, failure) + terminate.assert_called_once_with(process) + lease.close.assert_called_once() + self.assertEqual(failure.evidence["startup_stage"], "create_handle") + self.assertEqual( + [item["action"] for item in failure.evidence["errors"]], + ["release_port", "root_returncode"], + ) + self.assertIsNone(failure.evidence["returncode"]) + self.assertFalse(failure.evidence["cleanup_complete"]) + + +if __name__ == "__main__": + unittest.main()