From ca6e86e15ca0e8731ae5b82c4d35d9980813e5ba Mon Sep 17 00:00:00 2001 From: wshallwshall Date: Thu, 20 Aug 2026 09:31:37 -0500 Subject: [PATCH 1/3] feat(probe): a connscale gap now records WHICH path degraded it, and the smoke test asserts only what it can The FD probe returned the same all-None sample from THREE different paths and the record could not say which one fired. That is why `main`'s tip red was mis-attributed by four sessions today: the failing assertion was `tests/test_connscale_smoke.py:128`, `assert r.fd_count_peak is not None`, and the record's repr carried `sent=36 acked=24` -- so the attribution followed the NUMBERS in the dump to the item about sent/acked, rather than the ASSERTION, which was about the probe. The artifact did not carry the information needed to attribute it. WHAT CHANGED. `ProbeDegraded` names each cause distinctly -- the subtree walk timing out, erroring fast, returning zero rows, or returning a snapshot with no root; and the per-PID read timing out, erroring or parsing nothing. `ProcSample.degraded` carries it, `_gap(cause)` replaces the single shared `_EMPTY_PROC`, and `_resolve_errored` becomes a property over the recorded cause so a boolean can no longer be set without saying why. `is_budget_exhausted()` is the discriminator, stated once: a path that SPENT its budget measures the runner; one that failed without spending it is a broken probe. Those get opposite readings and they were previously indistinguishable. THE SMOKE ASSERTION IS NARROWED, NOT RELAXED, AND THE DIFFERENCE IS THE WHOLE POINT. It used to demand `fd_count_peak is not None` of EVERY record, with no tolerance for a probe that honestly could not measure -- so a degraded probe red a REQUIRED context as though the ENGINE were at fault. It now asserts `> 0` only where a value is present AND still requires that AT LEAST ONE record measured something. A run where the probe never measured at all still fails. Buying a green by deleting the requirement was the available wrong answer here and it is not what this does. Twelve cases in a new tests/test_connscale_probe_degradation.py pin each cause to its path, including the two that must record NO cause on success, and one asserting a resolution that fails without naming a cause is still given one -- so the enum cannot be bypassed by a future path that forgets to classify. Does NOT fix the underlying walk timeout, and does not claim to. It makes the next occurrence attributable, which is what four sessions lacked today. VERIFIED: tests/test_connscale_probe_degradation.py + tests/test_connscale_smoke.py -> 32 passed; tests/test_sandbox.py unaffected; ruff 0.15.22 clean on all five files; no cp1252-unsafe character introduced (census identical against HEAD). harness/ is not in CI's mypy scope (`mypy messagefoundry messagefoundry_webconsole --exclude 'messagefoundry/tray/'`), and the 3-file mypy invocation reports the same single package-discovery error at HEAD as with this change, so it is pre-existing and not introduced here. ITS ADVERSARIAL REVIEW HAD NOT REPORTED WHEN THIS WAS COMMITTED. Committed anyway to protect the work across a usage-window boundary, on the explicit understanding that a review finding is fixed FORWARD -- which is exactly what happened to the sibling #1290 commit earlier on this branch, where the review found a blocking defect and e2ce84ad corrected it before anything left this tree. Nothing here is pushed. --- harness/load/connscale/probe.py | 159 +++++- harness/load/connscale/report.py | 33 +- harness/load/connscale/runner.py | 76 ++- tests/test_connscale_probe_degradation.py | 596 ++++++++++++++++++++++ tests/test_connscale_smoke.py | 83 ++- 5 files changed, 898 insertions(+), 49 deletions(-) create mode 100644 tests/test_connscale_probe_degradation.py diff --git a/harness/load/connscale/probe.py b/harness/load/connscale/probe.py index 7a691a4c..80888188 100644 --- a/harness/load/connscale/probe.py +++ b/harness/load/connscale/probe.py @@ -22,7 +22,8 @@ statm`` (resident pages). Where the launching interpreter spawns no shim the subtree is just the one process — byte-identical to single-process sampling. Every field is ``None`` when nothing in the subtree could be read (a dead tree / a missing tool), so the runner records a gap rather than - crashing. + crashing — and that gap CARRIES ITS CAUSE (:class:`ProbeDegraded`), because a gap that cannot say + which of the probe's seven degrade paths produced it is not attributable to anything. The walk is **provenance-checked** (BACKLOG #1210): a candidate that PREDATES the root is not a descendant of it, so it is rejected along with its subtree. Windows never rewrites @@ -42,6 +43,7 @@ import sys import time from dataclasses import dataclass +from enum import StrEnum from pathlib import Path from messagefoundry.apiclient import ApiError, EngineClient @@ -75,6 +77,58 @@ type ProcRow = tuple[int, int, float | None] +class ProbeDegraded(StrEnum): + """WHY a :class:`ProcSample` carries no reading — the path that degraded this tick. + + A gap that does not name its cause is UNATTRIBUTABLE, and this probe has seven distinct ways to + produce one. They were previously indistinguishable: every path returned the same all-``None`` + sample, so a record could say only THAT the probe did not read, never WHICH mechanism stopped it — + and a CI red was mis-attributed to unrelated work three times over precisely because the artifact + did not carry the information needed to attribute it. + + The one distinction a CONSUMER has to draw is BUDGET-EXHAUSTED vs not (:attr:`is_budget_exhausted`), + because the two earn OPPOSITE verdicts: a shell-out that spent its whole ``_PROBE_TIMEOUT_S`` + measures the RUNNER (a starved host could not answer in time), while every other member means the + probe RAN and produced nothing usable, which is a defect in the probe. That line is stated here + ONCE. ``tests/test_connscale_cpu_probe.py`` draws the same line at the walk level from the seconds a + failed walk actually spent (``_BUDGET_CONSUMED_FRACTION``), and the two must stay one vocabulary: + "budget exhausted" is could-not-measure, anything faster is measured-and-broken.""" + + # --- the subtree walk: one process-table snapshot, per `FdSampler._resolve_pids` --- + # The walk spent its whole timeout. This measures the runner, not the engine. + WALK_TIMEOUT = "walk_timeout" + # The walk's shell-out raised something OTHER than a timeout (OSError / a non-timeout + # SubprocessError), so it failed without using its budget. + WALK_ERROR = "walk_error" + # The walk COMPLETED and yielded zero usable rows. A live host always has many processes, so this + # is a silent enumeration failure (truncated output / a walk that never really ran), never a + # genuine empty result -- see `_enumerate_windows`. + WALK_EMPTY = "walk_empty" + # The snapshot carried no row for the ROOT pid, so no candidate could be validated against the + # root's creation instant. Fail closed (`_validated_descendants`) rather than walk unchecked. + WALK_NO_ROOT = "walk_no_root" + + # --- the per-PID read: `FdSampler._sample_windows` / `._sample_posix` --- + # The per-PID read spent its whole timeout. Measures the runner, as WALK_TIMEOUT does. + READ_TIMEOUT = "read_timeout" + # The per-PID read raised something other than a timeout, without using its budget. + READ_ERROR = "read_error" + # The per-PID read RAN and returned zero usable rows across the whole subtree. Not a timeout at + # all -- the enumeration happened and produced nothing. + READ_EMPTY = "read_empty" + + @property + def is_budget_exhausted(self) -> bool: + """True when this cause is a shell-out that SPENT its whole ``_PROBE_TIMEOUT_S``. + + The discriminator a consumer needs, and the only one: budget-exhausted says the host was too + slow to answer, so the probe reports COULD NOT MEASURE and the subject under test is not + implicated. Every other member says the probe ran and produced nothing usable, which IS a + finding. Kept as a property on the vocabulary itself so no consumer re-derives the split from a + member list of its own that could then drift member-by-member.""" + return self in (ProbeDegraded.WALK_TIMEOUT, ProbeDegraded.READ_TIMEOUT) + + @dataclass(frozen=True) class ProcSample: """One OS-side reading of the engine process (all ``None`` when unreadable — a poll tick gap). @@ -88,15 +142,28 @@ class ProcSample: to derive utilisation, and that difference is only a clean CPU delta when the summed-over PID set is unchanged — A3's periodic subtree re-resolution can add a joining ``serve --shard`` worker or drop a departing one mid-window, so the runner uses this set to sum only same-set intervals and - degrade the rest to a gap (BACKLOG #220).""" + degrade the rest to a gap (BACKLOG #220). + * ``degraded`` — WHY this tick measured nothing, when it measured nothing. Set **iff** the tick is a + FULL gap (every field above ``None``); a tick that read anything at all carries ``None`` here. A + partial POSIX read (handles present, CPU absent) is NOT a degradation — the gauges that read still + read, and the ones that did not are visible as their own ``None``.""" handles: int | None cpu_seconds: float | None working_set_bytes: int | None cpu_pids: frozenset[int] | None = None + degraded: ProbeDegraded | None = None -_EMPTY_PROC = ProcSample(handles=None, cpu_seconds=None, working_set_bytes=None, cpu_pids=None) +def _gap(cause: ProbeDegraded) -> ProcSample: + """A full-gap sample that NAMES the path that produced it. + + Every degrade site goes through here, so a gap cannot be constructed without stating its cause — + which is the whole point: an unattributed gap is what let one starved-runner timeout and one + genuinely broken enumeration render as the same artifact.""" + return ProcSample( + handles=None, cpu_seconds=None, working_set_bytes=None, cpu_pids=None, degraded=cause + ) class FdSampler: @@ -107,17 +174,21 @@ class FdSampler: resolved periodically and cached in between; each :meth:`sample_proc` sums a cheap per-PID read across it. :meth:`sample` keeps the legacy handle-count-only shape (``int | None``). Every field is ``None`` when nothing in the subtree could be read (a dead tree / a missing tool) so a poll tick - records a gap, never raises.""" + records a gap, never raises — and that gap names the path that produced it + (:attr:`ProcSample.degraded`).""" def __init__(self, pid: int, *, resolve_every: int = _RESOLVE_EVERY_TICKS) -> None: self._pid = pid self._pids: list[int] | None = None # [root, *descendants], re-resolved every N ticks - # True while the last subtree resolution ERRORED (Windows enumeration failed/timed out, or the - # root's own creation instant was absent so nothing could be validated against it) — as opposed - # to a genuine no-descendants result. An errored resolution is NOT cached (so a later tick - # retries) and its samples are reported probe-degraded (all None) rather than measuring a root - # that may be only the launcher shim. - self._resolve_errored = False + # WHY the last subtree resolution failed, or None when it succeeded (Windows enumeration + # timed out / errored / came back empty, or the root's own creation instant was absent so + # nothing could be validated against it) — as opposed to a genuine no-descendants result. An + # errored resolution is NOT cached (so a later tick retries) and its samples are reported + # probe-degraded rather than measuring a root that may be only the launcher shim. + # + # This holds the CAUSE, not a bool, because the bool was the defect: four different resolution + # failures set it identically and the tick they degraded could not say which had fired. + self._resolve_degraded: ProbeDegraded | None = None # A3: the subtree is NOT stable for a SHARDED engine — ADR 0037's supervisor spawns one # `serve --shard` subprocess per shard, and a subtree cached before they appear measures an idle # supervisor forever (a flat CPU counter that used to render as a plausible 0.00). Re-resolve @@ -129,6 +200,12 @@ def __init__(self, pid: int, *, resolve_every: int = _RESOLVE_EVERY_TICKS) -> No def pid(self) -> int: return self._pid + @property + def _resolve_errored(self) -> bool: + """Did the last subtree resolution fail? DERIVED from :attr:`_resolve_degraded` so the fact is + stored once — a separate bool beside the cause is two statements of one fact, and they drift.""" + return self._resolve_degraded is not None + def sample(self) -> int | None: """The current handle/fd count across the engine subtree, or ``None`` if it can't be read (legacy shape). Delegates to :meth:`sample_proc` so it stays one cheap read per PID.""" @@ -139,12 +216,13 @@ def sample_proc(self) -> ProcSample: each field ``None`` when nothing could be read. Runs the OS probe synchronously — the runner calls it in ``run_in_executor`` (off the event loop), like the rest of the sampling.""" pids = self._resolve_pids() - if self._resolve_errored: - # Subtree resolution ERRORED (a failed/timed-out Windows enumeration, or no row for the - # root). Reading the root PID alone would report a launcher shim's footprint as the + if self._resolve_degraded is not None: + # Subtree resolution FAILED (a timed-out / errored / empty Windows enumeration, or no row + # for the root). Reading the root PID alone would report a launcher shim's footprint as the # engine's — worse than a gap, because it's a plausible-looking WRONG number that could flip - # a footprint delta. Record a probe-degraded gap (all None) and let a later tick retry. - return _EMPTY_PROC + # a footprint delta. Record a probe-degraded gap CARRYING WHICH of those fired, and let a + # later tick retry. + return _gap(self._resolve_degraded) if _WINDOWS: return self._sample_windows(pids) return self._sample_posix(pids) @@ -173,15 +251,21 @@ def _resolve_pids(self) -> list[int]: # Serving a previously-VALIDATED subtree. If the last re-resolve errored, that error # applied to that tick only — the cached subtree is still the best known truth, and # degrading every tick until the next re-walk would turn one transient enumeration - # failure into a run-long blackout. Clear the flag so this tick reports a real reading. - self._resolve_errored = False + # failure into a run-long blackout. Clear the cause so this tick reports a real reading. + self._resolve_degraded = None return self._pids self._ticks_since_resolve = 0 + # Cleared BEFORE the walk: the walk itself records why it failed, and a stale cause from the + # previous walk would otherwise be attributed to this one. + self._resolve_degraded = None descendants = self._descendants_windows() if _WINDOWS else self._descendants_posix() if descendants is None: - self._resolve_errored = True + if self._resolve_degraded is None: + # Defensive: `_descendants_windows` names every failure it returns None for. A stand-in + # that returns None without naming one still gets a cause rather than an unattributed + # gap — an unnamed cause is the exact condition this field exists to remove. + self._resolve_degraded = ProbeDegraded.WALK_ERROR return [self._pid] # transient (this tick only), not cached — retry next tick - self._resolve_errored = False ordered = [self._pid] for pid in descendants: if pid not in ordered: @@ -218,8 +302,14 @@ def _enumerate_windows(self) -> list[ProcRow] | None: text=True, timeout=_PROBE_TIMEOUT_S, ) + # TimeoutExpired is caught FIRST because it is a SubprocessError subclass, and it is the one + # failure here that measures the RUNNER rather than the probe (see ProbeDegraded). + except subprocess.TimeoutExpired: + self._resolve_degraded = ProbeDegraded.WALK_TIMEOUT + return None # spent its whole budget — NOT "no descendants" except (OSError, subprocess.SubprocessError): - return None # errored/timed out — NOT "no descendants" + self._resolve_degraded = ProbeDegraded.WALK_ERROR + return None # errored without using its budget — NOT "no descendants" # Parse whatever rows came back regardless of the exit code (a partial result is still usable). rows: list[ProcRow] = [] for line in out.stdout.splitlines(): @@ -236,6 +326,7 @@ def _enumerate_windows(self) -> list[ProcRow] | None: # (a silent failure / truncated output). Signal errored so the caller retries + degrades rather # than caching root-only and reporting the launcher shim's footprint as the engine's. if not rows: + self._resolve_degraded = ProbeDegraded.WALK_EMPTY return None return rows @@ -267,8 +358,14 @@ def _enumerate_posix(self) -> list[ProcRow]: def _descendants_windows(self) -> list[int] | None: rows = self._enumerate_windows() if rows is None: - return None - return _validated_descendants(rows, self._pid) + return None # `_enumerate_windows` recorded WHICH enumeration failure this was + walked = _validated_descendants(rows, self._pid) + if walked is None: + # The enumeration itself SUCCEEDED; what failed is validation — the snapshot carried no row + # for the root, so nothing could be checked against its creation instant. A distinct + # mechanism from any enumeration failure, and it must not be reported as one. + self._resolve_degraded = ProbeDegraded.WALK_NO_ROOT + return walked def _descendants_posix(self) -> list[int]: walked = _validated_descendants(self._enumerate_posix(), self._pid) @@ -294,8 +391,12 @@ def _sample_windows(self, pids: list[int]) -> ProcSample: text=True, timeout=_PROBE_TIMEOUT_S, ) + # TimeoutExpired first (it subclasses SubprocessError): a read that spent its whole budget + # measures the runner, an immediate error is the probe failing. Opposite verdicts downstream. + except subprocess.TimeoutExpired: + return _gap(ProbeDegraded.READ_TIMEOUT) except (OSError, subprocess.SubprocessError): - return _EMPTY_PROC + return _gap(ProbeDegraded.READ_ERROR) # NB: ignore the exit code. `Get-Process -Id a,b` where one PID has since exited emits a # non-terminating error (exit 1) EVEN under -ErrorAction SilentlyContinue, yet still writes the # live processes' rows to stdout. Trust the parsed rows; only zero rows ⇒ a genuine gap. @@ -322,7 +423,9 @@ def _sample_windows(self, pids: list[int]) -> ProcSample: cpu_pids.add(pid) rows += 1 if rows == 0: - return _EMPTY_PROC + # The read RAN and parsed nothing. NOT a timeout — no budget was exhausted, the command + # completed and produced no usable row for any PID in the subtree. + return _gap(ProbeDegraded.READ_EMPTY) return ProcSample( handles=handles, cpu_seconds=cpu_ticks / _WIN_CPU_TICKS_PER_S, @@ -350,6 +453,14 @@ def _sample_posix(self, pids: list[int]) -> ProcSample: if r is not None: rss_sum += r r_seen += 1 + if not (h_seen or c_seen or r_seen): + # Nothing in the whole subtree was readable — the reads RAN and produced no usable row, so + # this is READ_EMPTY for the same reason the Windows zero-rows branch is. The POSIX side + # does not split out a timeout: /proc reads are file reads with no budget to exhaust, and + # the one budgeted call (the lsof fallback) is per-PID, so a subtree-wide gap here is not + # attributable to any single PID's timeout. Claiming a timeout we did not observe would be + # exactly the fabricated cause this vocabulary exists to prevent. + return _gap(ProbeDegraded.READ_EMPTY) return ProcSample( handles=handles_sum if h_seen else None, cpu_seconds=cpu_sum if c_seen else None, diff --git a/harness/load/connscale/report.py b/harness/load/connscale/report.py index 4adecb9a..0c0a7c8e 100644 --- a/harness/load/connscale/report.py +++ b/harness/load/connscale/report.py @@ -149,6 +149,19 @@ class ConnScaleRecord: # non-batching record deserializes unchanged. The batch comparison pairs B0 vs B1 by this tag (it # reuses the fusion comparator's verdict path keyed on this field instead of ``fuse_thread_hops``). batch_handoff_statements: bool = False + # --- wall #4 probe provenance: why `fd_count_peak` is None, when it is None --- + # `fd_count_peak = None` used to be the whole story, and it is the same value whether the host was + # too starved to enumerate, the process tree was gone, or the enumerator ran and returned zero + # rows. Those warrant different verdicts, so the OS probe's own account of the window travels with + # the gauge. `fd_probe_ticks` is the SCOPE for `fd_probe_degraded_ticks` (a degraded count without + # its denominator is not readable), and `fd_probe_degraded` holds the DISTINCT causes as strings + # (`harness.load.connscale.probe.ProbeDegraded` values) — strings, so this module keeps its + # independence from the probe. All three default so an older artifact deserializes unchanged; + # an empty `fd_probe_degraded` alongside `fd_probe_ticks == 0` means the probe did not run at all, + # which is itself distinct from having run and failed. + fd_probe_ticks: int = 0 + fd_probe_degraded_ticks: int = 0 + fd_probe_degraded: tuple[str, ...] = () def to_json_dict(self) -> dict[str, object]: return { @@ -212,7 +225,16 @@ def to_json_dict(self) -> dict[str, object]: else round(self.empty_claims_per_msg, 3) ), }, - "wall4_fd": {"count_peak": self.fd_count_peak}, + "wall4_fd": { + "count_peak": self.fd_count_peak, + # The gap's account of itself. Present even on a clean window (0 degraded of N ticks), + # because "the probe measured every tick" is a fact worth reading in an artifact too. + "probe": { + "ticks": self.fd_probe_ticks, + "degraded_ticks": self.fd_probe_degraded_ticks, + "degraded": list(self.fd_probe_degraded), + }, + }, "wall5_reload": {"seconds": self.reload_seconds}, "wall6_ack_ms": { "p50": round(self.ack_p50_ms, 3), @@ -370,6 +392,15 @@ def render_console(self) -> str: f"{_na(r.fd_count_peak):>7}{_na(_round_or_none(r.cpu_seconds_total, 1)):>8}" f"{_na(r.reload_seconds):>8}{r.ack_p99_ms:>9.1f}" ) + # The `fd` column renders `n/a` on a gap, and nobody can act on `n/a`. Name the mechanism beside + # it, with the scope of the count, so the console says which of the probe's degrade paths fired. + for r in self.records: + if r.fd_probe_degraded_ticks: + causes = ", ".join(r.fd_probe_degraded) or "cause not recorded" + lines.append( + f"fd probe: {r.sweep_mode}@N={r.count} -- {r.fd_probe_degraded_ticks} of " + f"{r.fd_probe_ticks} tick(s) measured nothing [{causes}]" + ) lines.append("") lines.append("SLOs:") if not self.slos: diff --git a/harness/load/connscale/runner.py b/harness/load/connscale/runner.py index 9b162db9..a465a760 100644 --- a/harness/load/connscale/runner.py +++ b/harness/load/connscale/runner.py @@ -823,6 +823,11 @@ def _build_record( wake_fanout_per_s=wake_per_s, empty_claims_per_msg=empty_per_msg, fd_count_peak=proc.handles_peak, + # Carried BESIDE fd_count_peak, never instead of it: a consumer that sees None needs to know + # whether the probe could not measure or measured nothing, and those are different verdicts. + fd_probe_ticks=proc.probe_ticks, + fd_probe_degraded_ticks=proc.probe_degraded_ticks, + fd_probe_degraded=proc.probe_degraded, reload_seconds=reload_seconds, ack_p50_ms=ack.p50_ms, ack_p95_ms=ack.p95_ms, @@ -987,13 +992,27 @@ def _throughput_rates(samples: list[EngineSample]) -> tuple[float, float]: @dataclass(frozen=True) class _ProcDerived: - """Derived process-footprint gauges over the window (each None where the OS probe couldn't read).""" + """Derived process-footprint gauges over the window (each None where the OS probe couldn't read). + + The last three fields are the window's probe PROVENANCE, not gauges: they say how much of the + window the OS probe actually measured and, where it did not, WHY. Without them a ``None`` gauge is + indistinguishable across a starved host, a dead process tree and a broken enumerator — three + conditions with three different verdicts.""" handles_peak: int | None cpu_seconds_total: float | None cpu_util_cores_peak: float | None cpu_util_cores_mean: float | None working_set_peak_bytes: int | None + #: How many probe ticks this window collected at all. The SCOPE for the count below: "2 degraded" + #: means nothing without it, and 2-of-3 and 2-of-200 warrant opposite reactions. + probe_ticks: int + #: How many of those ticks were full gaps (:attr:`ProcSample.degraded` set). + probe_degraded_ticks: int + #: The DISTINCT degradation causes seen across the window, sorted. Values are + #: :class:`ProbeDegraded` members, carried as plain strings so the report layer need not import + #: the probe. Empty when every tick read something. + probe_degraded: tuple[str, ...] def _drain_proc(samples: list[EngineSample]) -> _ProcDerived: @@ -1011,7 +1030,12 @@ def _drain_proc(samples: list[EngineSample]) -> _ProcDerived: the total and its elapsed span to ``covered_span``; a set-change interval is degraded to a gap (it contributes nothing), and the peak-cores loop is gated the same way. The flat-CPU-gap guard and ``cpu_mean`` are recomputed over ``covered_span``, and the CPU gauges degrade to ``None`` when zero - clean intervals contributed (or fewer than two CPU readings exist). See BACKLOG #220.""" + clean intervals contributed (or fewer than two CPU readings exist). See BACKLOG #220. + + It also carries the window's probe PROVENANCE through to the record: how many ticks were collected, + how many of those were full gaps, and the distinct causes behind them. The gauges alone cannot + express the difference between "the host was too slow to answer" and "the enumerator returned zero + rows" — both are ``None`` — and a consumer that has to choose a verdict needs exactly that.""" readings: list[tuple[float, ProcSample]] = [] for s in samples: proc = _PROC_BY_SAMPLE.pop(id(s), None) @@ -1030,6 +1054,30 @@ def _drain_proc(samples: list[EngineSample]) -> _ProcDerived: handles_peak = max(handles) if handles else None ws_peak = max(working_set) if working_set else None + # Probe PROVENANCE for the window. A gauge that reads None because the host was starved and one + # that reads None because the enumerator is broken are the same value and opposite findings, so the + # causes the probe recorded per tick are carried through to the record rather than discarded here. + probe_ticks = len(readings) + degraded = [p.degraded for _, p in readings if p.degraded is not None] + causes = tuple(sorted({str(c) for c in degraded})) + + def _derived( + cpu_total: float | None, cpu_peak: float | None, cpu_mean: float | None + ) -> _ProcDerived: + """Assemble the result, filling the provenance fields in ONE place. The CPU gauges have three + legitimate exit paths; routing them all through here means no path can ship a gap that has + dropped its cause on the way out.""" + return _ProcDerived( + handles_peak=handles_peak, + cpu_seconds_total=cpu_total, + cpu_util_cores_peak=cpu_peak, + cpu_util_cores_mean=cpu_mean, + working_set_peak_bytes=ws_peak, + probe_ticks=probe_ticks, + probe_degraded_ticks=len(degraded), + probe_degraded=causes, + ) + cpu_total: float | None = None cpu_mean: float | None = None cpu_peak: float | None = None @@ -1054,13 +1102,7 @@ def _drain_proc(samples: list[EngineSample]) -> _ProcDerived: if clean_intervals == 0: # Every interval was a membership change (or non-advancing time): no clean CPU measurement # survived, so report a gap rather than a fabricated 0.00. - return _ProcDerived( - handles_peak=handles_peak, - cpu_seconds_total=None, - cpu_util_cores_peak=None, - cpu_util_cores_mean=None, - working_set_peak_bytes=ws_peak, - ) + return _derived(None, None, None) # A3 / B-class guard, recomposed over the summed clean span: a FLAT cumulative CPU counter over # a non-trivial covered span is not a physical "0% CPU" — the counter's unit is 100 ns (Windows # ticks) / a clock tick (POSIX), and we only got here because the process was READABLE. A live @@ -1069,24 +1111,12 @@ def _drain_proc(samples: list[EngineSample]) -> _ProcDerived: # spawned). Reporting 0.00 here is the signature defect of this harness: a plausible number where # there is no measurement. Emit an explicit gap (None) so it renders "n/a" and draws no verdict. if total == 0.0 and covered_span >= _CPU_FLAT_GAP_SPAN_S: - return _ProcDerived( - handles_peak=handles_peak, - cpu_seconds_total=None, - cpu_util_cores_peak=None, - cpu_util_cores_mean=None, - working_set_peak_bytes=ws_peak, - ) + return _derived(None, None, None) cpu_total = total cpu_peak = peak if covered_span > 0.0: cpu_mean = total / covered_span - return _ProcDerived( - handles_peak=handles_peak, - cpu_seconds_total=cpu_total, - cpu_util_cores_peak=cpu_peak, - cpu_util_cores_mean=cpu_mean, - working_set_peak_bytes=ws_peak, - ) + return _derived(cpu_total, cpu_peak, cpu_mean) def _peak_int(values: list[int | None]) -> int | None: diff --git a/tests/test_connscale_probe_degradation.py b/tests/test_connscale_probe_degradation.py new file mode 100644 index 00000000..ec29a39b --- /dev/null +++ b/tests/test_connscale_probe_degradation.py @@ -0,0 +1,596 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""A wall #4 gap must carry its CAUSE, and the smoke must assert only what that cause entitles it to. + +Before this module the connscale FD probe had SEVEN paths that produced a reading-less +:class:`ProcSample`, and all seven produced the SAME one — every field ``None``, nothing naming the +mechanism. Measured before the fix by driving the real ``FdSampler`` down four of them: four distinct +causes, ONE distinct ``repr``. The consequences were both real: + +* A CI red carrying ``fd_count_peak=None`` was mis-attributed to unrelated work three times over, + because the artifact did not contain the information needed to attribute it. +* ``tests/test_connscale_smoke.py`` asserted ``fd_count_peak is not None`` with no tolerance, so a + process-table walk that timed out on a starved runner reddened a REQUIRED context as though the + ENGINE were at fault — indistinguishably from an enumerator that ran and returned zero rows. + +The two halves are covered here together because they are one contract: the probe records which path +degraded it, and the test reads that cause to pick a verdict. Neither half is worth anything alone — +a cause nobody reads changes no outcome, and a verdict with no cause to read is the tolerance-only +"fix" that buys a green by discarding the evidence. + +**The verdict split is the probe's own** (``ProbeDegraded.is_budget_exhausted``) and it is the same +line ``tests/test_connscale_cpu_probe.py`` already draws from the seconds a failed walk spent +(``_BUDGET_CONSUMED_FRACTION``): a shell-out that spent its whole budget measures the RUNNER +(could-not-measure), anything faster means the probe ran and produced nothing (measured-and-broken). +One vocabulary, stated once, read in both places. +""" + +from __future__ import annotations + +import dataclasses +import json +import os +import subprocess +from typing import Any + +import pytest + +from harness.load.connscale import probe as probe_module +from harness.load.connscale.probe import FdSampler, ProbeDegraded, ProcSample, _gap +from harness.load.connscale.report import ConnScaleRecord, ConnScaleReport, NoLoss, SloCheck +from harness.load.connscale.runner import _PROC_BY_SAMPLE, _drain_proc +from harness.load.enginepoll import EngineSample + +# The smoke test owns the verdict helper, beside the assertion it serves and the prose explaining it. +# Imported here rather than re-stated, so this file cannot encode a second, quietly different rule +# about which gaps are tolerable -- that duplication is the failure being fixed, in miniature. +from tests.test_connscale_smoke import _assert_fd_probe, _is_budget_exhausted + +#: An implausible PID: nothing can be read for it on any platform, so the POSIX read path is exercised +#: without depending on a live process. +_DEAD_PID = 2**31 - 1 + +#: The two causes that mean the probe SPENT its timeout budget rather than failing fast. +_BUDGET_EXHAUSTED = (ProbeDegraded.WALK_TIMEOUT, ProbeDegraded.READ_TIMEOUT) + + +class _FakeSubprocess: + """Stand-in for the ``subprocess`` module INSIDE ``probe.py`` only. + + Injected by replacing ``probe.subprocess``, not by patching ``subprocess.run`` globally: the probe + is the only caller under test, and a global patch would also intercept anything pytest or the + runtime shells out during the test.""" + + TimeoutExpired = subprocess.TimeoutExpired + SubprocessError = subprocess.SubprocessError + + def __init__(self, behaviour: Any) -> None: + self._behaviour = behaviour + + def run(self, cmd: Any, **kwargs: Any) -> Any: + return self._behaviour(cmd, **kwargs) + + +class _Completed: + """The two attributes ``probe.py`` reads off a completed shell-out.""" + + def __init__(self, stdout: str = "", returncode: int = 0) -> None: + self.stdout = stdout + self.returncode = returncode + + +def _spent_its_budget(cmd: Any, **kwargs: Any) -> Any: + raise subprocess.TimeoutExpired(cmd, kwargs.get("timeout", 5.0)) + + +def _failed_fast(cmd: Any, **kwargs: Any) -> Any: + raise OSError("the tool is not on PATH") + + +def _returned_nothing(cmd: Any, **kwargs: Any) -> Any: + return _Completed(stdout="") + + +def _with_subprocess(monkeypatch: pytest.MonkeyPatch, behaviour: Any) -> None: + monkeypatch.setattr(probe_module, "subprocess", _FakeSubprocess(behaviour)) + + +# --- the vocabulary itself ------------------------------------------------------------------------- + + +def test_every_cause_is_classified_and_only_a_spent_budget_is_tolerable() -> None: + # The split decides FAIL vs TOLERATE, so pin the whole membership rather than spot-checking two + # members: a cause added later without being classified must show up here, not as a silent + # tolerance. Classification is fail-closed by construction (`is_budget_exhausted` names the + # tolerable members explicitly), so an unclassified newcomer reads as broken, which is the safe + # direction -- it fails loudly instead of arriving pre-excused. + tolerable = {c for c in ProbeDegraded if c.is_budget_exhausted} + assert tolerable == set(_BUDGET_EXHAUSTED) + # Every member is a real classification, and the not-tolerable side is not empty (a split where + # everything landed on one side would pass a membership check while deciding nothing). + assert all(isinstance(c.is_budget_exhausted, bool) for c in ProbeDegraded) + assert set(ProbeDegraded) - tolerable + + +def test_a_gap_carries_its_cause_and_no_reading() -> None: + # `_gap` is the ONLY way a degrade site builds a sample, so this invariant is what makes "every + # field None" and "a cause is recorded" inseparable -- the pair that used to come apart. + for cause in ProbeDegraded: + g = _gap(cause) + assert g == ProcSample(None, None, None, None, cause) + assert (g.handles, g.cpu_seconds, g.working_set_bytes, g.cpu_pids) == (None,) * 4 + assert g.degraded is cause + + +# --- the subtree walk records which walk failure fired ---------------------------------------------- +# +# These drive the Windows walk/read methods DIRECTLY on any platform. That is not a shortcut around a +# platform guard: `_enumerate_windows` / `_sample_windows` are pure shell-out-and-parse with no +# OS-specific syscall, so with `subprocess` injected they behave identically everywhere -- which means +# the Linux CI legs cover the Windows paths where the observed red actually happened, instead of +# skipping exactly the code under test. + + +def test_a_walk_that_spends_its_budget_records_walk_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _with_subprocess(monkeypatch, _spent_its_budget) + sampler = FdSampler(_DEAD_PID, resolve_every=1) + assert sampler._enumerate_windows() is None + assert sampler._resolve_degraded is ProbeDegraded.WALK_TIMEOUT + + +def test_a_walk_that_errors_fast_records_walk_error(monkeypatch: pytest.MonkeyPatch) -> None: + # TimeoutExpired subclasses SubprocessError, so a single `except (OSError, SubprocessError)` would + # collapse this case into the timeout one. The two earn opposite verdicts; keep them apart. + _with_subprocess(monkeypatch, _failed_fast) + sampler = FdSampler(_DEAD_PID, resolve_every=1) + assert sampler._enumerate_windows() is None + assert sampler._resolve_degraded is ProbeDegraded.WALK_ERROR + + +def test_a_walk_that_returns_zero_rows_records_walk_empty(monkeypatch: pytest.MonkeyPatch) -> None: + # A live host always has many processes, so a COMPLETED walk with zero usable rows is a silent + # enumeration failure -- and, per the scope of this work, the case that must still fail. + _with_subprocess(monkeypatch, _returned_nothing) + sampler = FdSampler(_DEAD_PID, resolve_every=1) + assert sampler._enumerate_windows() is None + assert sampler._resolve_degraded is ProbeDegraded.WALK_EMPTY + assert not ProbeDegraded.WALK_EMPTY.is_budget_exhausted + + +def test_a_snapshot_without_the_root_records_walk_no_root(monkeypatch: pytest.MonkeyPatch) -> None: + # The enumeration SUCCEEDED here; validation is what failed. Reporting it as an enumeration failure + # would point a reader at the wrong half of the walk. + sampler = FdSampler(500, resolve_every=1) + monkeypatch.setattr(sampler, "_enumerate_windows", lambda: [(900, 1, 10_000.0)]) + assert sampler._descendants_windows() is None + assert sampler._resolve_degraded is ProbeDegraded.WALK_NO_ROOT + + +def test_a_resolution_that_fails_without_naming_a_cause_is_still_given_one( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # `_resolve_pids` has a defensive fallback for a descendants-resolver that returns None without + # recording why. The shipped resolvers always record one, so this branch is reachable only through + # a stand-in -- which is exactly why it needs pinning: nothing else demonstrates that a resolution + # failure can never reach `sample_proc` as an UNNAMED gap, and an unnamed gap is the defect. + monkeypatch.setattr(probe_module, "_WINDOWS", True) + sampler = FdSampler(500, resolve_every=1) + monkeypatch.setattr(sampler, "_descendants_windows", lambda: None) + assert sampler.sample_proc().degraded is ProbeDegraded.WALK_ERROR + + +def test_a_walk_that_succeeds_records_no_cause(monkeypatch: pytest.MonkeyPatch) -> None: + # THE POSITIVE CONTROL for the four above: the causes are set by failure, not by merely walking. + # Without this, a `_resolve_degraded` wired to a constant would pass every rejection test here. + sampler = FdSampler(500, resolve_every=1) + monkeypatch.setattr( + sampler, "_enumerate_windows", lambda: [(500, 1, 10_000.0), (600, 500, 10_001.0)] + ) + assert sampler._descendants_windows() == [600] + assert sampler._resolve_degraded is None + assert sampler._resolve_errored is False + + +# --- the per-PID read records which read failure fired ---------------------------------------------- + + +def test_a_read_that_spends_its_budget_records_read_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _with_subprocess(monkeypatch, _spent_its_budget) + got = FdSampler(_DEAD_PID)._sample_windows([_DEAD_PID]) + assert got.degraded is ProbeDegraded.READ_TIMEOUT + + +def test_a_read_that_errors_fast_records_read_error(monkeypatch: pytest.MonkeyPatch) -> None: + _with_subprocess(monkeypatch, _failed_fast) + got = FdSampler(_DEAD_PID)._sample_windows([_DEAD_PID]) + assert got.degraded is ProbeDegraded.READ_ERROR + + +def test_a_read_that_returns_zero_rows_records_read_empty(monkeypatch: pytest.MonkeyPatch) -> None: + # The site the scope of this work calls out by name: the enumeration RAN and returned rows == 0, + # which is NOT a timeout. It must be distinguishable from one, and it must not be tolerable. + _with_subprocess(monkeypatch, _returned_nothing) + got = FdSampler(_DEAD_PID)._sample_windows([_DEAD_PID]) + assert got.degraded is ProbeDegraded.READ_EMPTY + assert not ProbeDegraded.READ_EMPTY.is_budget_exhausted + + +def test_a_read_that_parses_rows_records_no_cause(monkeypatch: pytest.MonkeyPatch) -> None: + # POSITIVE CONTROL for the read path: a parseable row yields a real reading and NO cause, so the + # three assertions above are discriminating rather than vacuous. + _with_subprocess(monkeypatch, lambda cmd, **kw: _Completed(stdout="61 20000000 6500000 4242\n")) + got = FdSampler(4242)._sample_windows([4242]) + assert got.degraded is None + assert (got.handles, got.working_set_bytes, got.cpu_pids) == (61, 6_500_000, frozenset({4242})) + + +def test_a_posix_read_with_nothing_readable_records_read_empty( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The POSIX side degrades when no PID in the subtree yields any field. It reports READ_EMPTY and + # deliberately does NOT claim a timeout: /proc reads have no budget to exhaust, and inventing a + # cause we did not observe is the same defect as recording none. + _with_subprocess(monkeypatch, _failed_fast) # no /proc and no lsof + got = FdSampler(_DEAD_PID)._sample_posix([_DEAD_PID]) + assert got.degraded is ProbeDegraded.READ_EMPTY + assert got.handles is None + + +# --- the defect itself: four causes that used to be one artifact ------------------------------------- + + +def _sample_proc_under(monkeypatch: pytest.MonkeyPatch, behaviour: Any) -> ProcSample: + """One full ``sample_proc()`` through the WINDOWS branch, with ``subprocess`` injected. + + ``_WINDOWS`` is forced so the Windows walk+read path runs on every platform. The observed CI red + was on a Windows leg, and gating this on ``sys.platform`` would skip the reproduction on the very + legs that are cheapest and most numerous.""" + monkeypatch.setattr(probe_module, "_WINDOWS", True) + _with_subprocess(monkeypatch, behaviour) + return FdSampler(_DEAD_PID, resolve_every=1).sample_proc() + + +def _is_walk(cmd: Any) -> bool: + return any("Win32_Process" in str(a) for a in cmd) + + +def test_four_degradation_paths_that_were_indistinguishable_are_now_distinct( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """THE regression test for the measured defect. + + Driven through the real ``sample_proc()`` entry point, exactly as the pre-fix reproduction was: + four distinct causes then produced one ``ProcSample`` value, so a record could not say which had + fired. Asserting the four are now distinct FAILS on the pre-fix probe, where every branch returned + the same ``_EMPTY_PROC``.""" + live = f"61 20000000 6500000 {_DEAD_PID}\n" + walk_rows = f"{_DEAD_PID} 1 10000000000\n" + + def walk_timeout(cmd: Any, **kw: Any) -> Any: + return _spent_its_budget(cmd, **kw) if _is_walk(cmd) else _Completed(stdout=live) + + def walk_zero_rows(cmd: Any, **kw: Any) -> Any: + return _Completed(stdout="") if _is_walk(cmd) else _Completed(stdout=live) + + def read_timeout(cmd: Any, **kw: Any) -> Any: + return _Completed(stdout=walk_rows) if _is_walk(cmd) else _spent_its_budget(cmd, **kw) + + def read_zero_rows(cmd: Any, **kw: Any) -> Any: + return _Completed(stdout=walk_rows if _is_walk(cmd) else "") + + got = { + "walk timed out": _sample_proc_under(monkeypatch, walk_timeout), + "walk ran, zero rows": _sample_proc_under(monkeypatch, walk_zero_rows), + "read timed out": _sample_proc_under(monkeypatch, read_timeout), + "read ran, zero rows": _sample_proc_under(monkeypatch, read_zero_rows), + } + assert {k: v.degraded for k, v in got.items()} == { + "walk timed out": ProbeDegraded.WALK_TIMEOUT, + "walk ran, zero rows": ProbeDegraded.WALK_EMPTY, + "read timed out": ProbeDegraded.READ_TIMEOUT, + "read ran, zero rows": ProbeDegraded.READ_EMPTY, + } + # The pre-fix measurement was "4 causes, 1 distinct sample". State the falsifier as that count. + assert len({repr(v) for v in got.values()}) == 4 + # Every one of them is still a full gap -- naming the cause must not have invented a reading. + assert all(v.handles is None for v in got.values()) + # And the two halves of the split land on opposite sides, which is what makes it a verdict. + assert got["walk timed out"].degraded is not None + assert got["walk timed out"].degraded.is_budget_exhausted + assert got["read ran, zero rows"].degraded is not None + assert not got["read ran, zero rows"].degraded.is_budget_exhausted + + +def test_a_live_sample_records_a_cause_exactly_when_it_reads_nothing() -> None: + """The invariant on the REAL probe against the REAL OS, with no injection anywhere. + + A null result needs a mechanism: every test above supplies its own failure, so all of them would + still pass against a probe that could not read anything at all on this host. This one runs the + unmodified probe against a live process and pins the biconditional -- a cause is recorded exactly + when nothing was read -- so it reports honestly whichever way this runner behaves.""" + got = FdSampler(os.getpid(), resolve_every=1).sample_proc() + read_something = any( + v is not None for v in (got.handles, got.cpu_seconds, got.working_set_bytes) + ) + assert (got.degraded is None) is read_something, got + + +# --- the cause reaches the record -------------------------------------------------------------------- + + +def _engine_sample(elapsed: float) -> EngineSample: + return EngineSample( + elapsed_s=elapsed, + pending=0, + inflight=0, + done=0, + dead=0, + read=0, + written=0, + out_dead=0, + queue_depth=0, + in_pipeline=0, + db_size_bytes=0, + journal_mode="wal", + synchronous="normal", + uptime_s=elapsed, + ) + + +def _drain(procs: list[ProcSample]) -> Any: + samples = [] + for i, p in enumerate(procs): + s = _engine_sample(float(i)) + _PROC_BY_SAMPLE[id(s)] = p + samples.append(s) + return _drain_proc(samples) + + +def _reading(handles: int, pid: int = 1234) -> ProcSample: + return ProcSample(handles, 1.0, 6_000_000, frozenset({pid})) + + +def test_a_fully_degraded_window_reports_its_causes_with_their_scope() -> None: + d = _drain([_gap(ProbeDegraded.WALK_TIMEOUT), _gap(ProbeDegraded.READ_EMPTY)]) + assert d.handles_peak is None # the gauge is still a gap... + assert d.probe_degraded == ("read_empty", "walk_timeout") # ...but it now says why + # A degraded count is unreadable without its denominator: 2-of-2 and 2-of-200 differ. + assert (d.probe_degraded_ticks, d.probe_ticks) == (2, 2) + + +def test_a_partially_degraded_window_keeps_its_gauges_and_still_reports_the_gap() -> None: + # The common real shape: some ticks read, some do not. `handles_peak` reads from the good ticks, and + # the degraded ones must not vanish just because the gauge survived -- a peak drawn from 1 of 3 + # ticks is a different quality of evidence from one drawn from 3 of 3, and only these counts say so. + d = _drain([_reading(61), _gap(ProbeDegraded.WALK_TIMEOUT), _reading(75)]) + assert d.handles_peak == 75 + assert d.probe_degraded == ("walk_timeout",) + assert (d.probe_degraded_ticks, d.probe_ticks) == (1, 3) + + +def test_a_clean_window_records_no_causes() -> None: + # POSITIVE CONTROL for the two above: provenance is recorded from what happened, not stamped on. + d = _drain([_reading(61), _reading(75)]) + assert d.handles_peak == 75 + assert d.probe_degraded == () + assert (d.probe_degraded_ticks, d.probe_ticks) == (0, 2) + + +def test_the_no_clean_interval_cpu_gap_path_keeps_the_provenance() -> None: + # `_drain_proc` has THREE exits and two of them are CPU-gap early returns. Each is a separate + # `return`, so each can independently forget to carry the cause -- which would be this change + # failing precisely where a gap is being reported. Exit 1: every interval crossed a subtree + # membership change, so no clean CPU delta survives. + d = _drain( + [ + ProcSample(61, 10.0, 6_000_000, frozenset({100}), None), + ProcSample(61, 30.0, 6_000_000, frozenset({100, 200}), None), + _gap(ProbeDegraded.READ_ERROR), + ] + ) + assert d.cpu_seconds_total is None # confirms this exit was actually taken, not the tail one + assert d.probe_degraded == ("read_error",) + assert (d.probe_degraded_ticks, d.probe_ticks) == (1, 3) + + +def test_the_flat_counter_cpu_gap_path_keeps_the_provenance() -> None: + # Exit 2: a flat cumulative CPU counter across a span past the guard (a wrong PID binding). The + # elapsed values must exceed `_CPU_FLAT_GAP_SPAN_S` or this silently falls through to the tail + # return and stops testing the branch it names. + samples = [ + _engine_sample(0.0), + _engine_sample(10.0), + _engine_sample(20.0), + ] + flat = [ + ProcSample(61, 12.5, 6_000_000, frozenset({1}), None), + ProcSample(61, 12.5, 6_000_000, frozenset({1}), None), + _gap(ProbeDegraded.WALK_TIMEOUT), + ] + for s, p in zip(samples, flat, strict=True): + _PROC_BY_SAMPLE[id(s)] = p + d = _drain_proc(samples) + assert d.cpu_seconds_total is None # confirms the flat-counter exit was taken + assert d.probe_degraded == ("walk_timeout",) + assert (d.probe_degraded_ticks, d.probe_ticks) == (1, 3) + + +# --- the record and its artifacts -------------------------------------------------------------------- + + +def _record( + *, + count: int = 12, + fd: int | None = 100, + ticks: int = 6, + degraded_ticks: int = 0, + degraded: tuple[str, ...] = (), + mode: str = "fixed_aggregate", +) -> ConnScaleRecord: + return ConnScaleRecord( + sweep_mode=mode, + count=count, + offered_aggregate_rate=24.0, + sent=100, + acked=100, + nak=0, + deferred=0, + no_loss=NoLoss(True, 100, 100, 100, 100, 0, "ok"), + in_pipeline_peak=1, + drain_seconds=0.5, + executor_queue_depth_peak=1, + executor_busy_peak=1, + pool_wait_p50_ms=None, + pool_wait_p95_ms=None, + pool_wait_p99_ms=None, + pool_wait_max_ms=None, + pool_idle_min=None, + pool_size_max=None, + empty_claims_per_s=1.0, + idle_poll_per_s=0.5, + wake_fanout_per_s=0.5, + empty_claims_per_msg=None, + fd_count_peak=fd, + reload_seconds=0.01, + ack_p50_ms=1.0, + ack_p95_ms=1.0, + ack_p99_ms=1.0, + fd_probe_ticks=ticks, + fd_probe_degraded_ticks=degraded_ticks, + fd_probe_degraded=degraded, + ) + + +def test_the_json_artifact_carries_the_probe_provenance_beside_the_gauge() -> None: + # The artifact is what a later reader attributes a red from, so the cause has to survive into it -- + # a cause that lives only in memory attributes nothing after the job ends. + d = _record(fd=None, ticks=6, degraded_ticks=6, degraded=("walk_timeout",)).to_json_dict() + wall4 = d["wall4_fd"] + assert isinstance(wall4, dict) + assert wall4["count_peak"] is None + assert wall4["probe"] == {"ticks": 6, "degraded_ticks": 6, "degraded": ["walk_timeout"]} + # It must round-trip as JSON (a tuple would serialize, an enum member would not have). + assert json.loads(json.dumps(d))["wall4_fd"]["probe"]["degraded"] == ["walk_timeout"] + + +def test_an_older_artifact_without_the_provenance_still_builds_a_record() -> None: + # The three fields default, so a record built from an artifact predating them deserializes + # unchanged -- and reads as "the probe did not run", which is distinct from "it ran and failed". + # Constructed by OMITTING them rather than by reading their declared defaults, because the property + # that matters is that such a call still succeeds, not that a default is written down somewhere. + added = ("fd_probe_ticks", "fd_probe_degraded_ticks", "fd_probe_degraded") + legacy: dict[str, Any] = {k: v for k, v in vars(_record()).items() if k not in added} + r = ConnScaleRecord(**legacy) + assert (r.fd_probe_ticks, r.fd_probe_degraded_ticks, r.fd_probe_degraded) == (0, 0, ()) + + +def test_the_console_names_the_mechanism_beside_the_n_a() -> None: + # `fd` renders `n/a` on a gap and nobody can act on `n/a`. The console has to say which path failed + # and over how many ticks, or the operator-facing view reproduces the original defect. + report = ConnScaleReport( + profile="t", + engine_url="http://127.0.0.1:8800", + db_backend=None, + shim_installed=True, + records=[_record(fd=None, ticks=6, degraded_ticks=5, degraded=("read_empty",))], + slos=[SloCheck("zero_loss", True, True, True)], + notes=[], + result_ok=True, + exit_code=0, + ) + text = report.render_console() + assert "read_empty" in text + assert "5 of 6 tick(s) measured nothing" in text + # And a clean record adds no noise. + clean = dataclasses.replace(report, records=[_record()]) + assert "measured nothing" not in clean.render_console() + + +# --- the smoke test's verdict ------------------------------------------------------------------------ + + +def test_a_measured_record_is_asserted_exactly_as_strictly_as_before() -> None: + # The pre-existing strength is preserved where the probe worked: a positive count passes, and a + # non-positive one still fails. The change adds a verdict for gaps; it does not relax readings. + _assert_fd_probe([_record(fd=100, ticks=6)]) + with pytest.raises(AssertionError): + _assert_fd_probe([_record(fd=0, ticks=6)]) + + +def test_a_spent_budget_gap_is_tolerated_when_another_step_measured() -> None: + # The starved-runner case that was reddening a required context. Tolerated -- but only alongside a + # step that actually measured, which is what stops the tolerance becoming a blanket green. + _assert_fd_probe( + [ + _record(count=12, fd=None, ticks=6, degraded_ticks=6, degraded=("walk_timeout",)), + _record(count=24, fd=140, ticks=6), + ] + ) + + +def test_a_broken_probe_still_fails_and_the_message_names_the_cause() -> None: + # The scope of this work is explicit that a probe which enumerated and returned zero rows must + # STILL FAIL. A tolerance that swallowed this would buy a green by deleting the finding. + with pytest.raises(AssertionError, match="PROBE DEFECT") as e: + _assert_fd_probe( + [ + _record(count=12, fd=None, ticks=6, degraded_ticks=6, degraded=("read_empty",)), + _record(count=24, fd=140, ticks=6), + ] + ) + assert "read_empty" in str(e.value) + assert "6 of 6 probe tick(s) degraded" in str(e.value) # the count, with its scope + + +def test_a_mixed_gap_fails_on_its_broken_cause_even_beside_a_tolerable_one() -> None: + # One tolerable cause must not launder a broken one sharing the window. + with pytest.raises(AssertionError, match="PROBE DEFECT"): + _assert_fd_probe( + [ + _record( + fd=None, ticks=6, degraded_ticks=6, degraded=("read_empty", "walk_timeout") + ), + _record(count=24, fd=140, ticks=6), + ] + ) + + +def test_an_unattributed_gap_fails_rather_than_being_tolerated() -> None: + # A gap naming no cause is precisely the pre-fix artifact. It is not evidence in either direction, + # so it must not land on the tolerated side -- otherwise the fix would have made the ORIGINAL + # ambiguous artifact the one thing that always passes. + with pytest.raises(AssertionError, match="UNATTRIBUTED GAP"): + _assert_fd_probe([_record(fd=None, ticks=6, degraded_ticks=6, degraded=())]) + + +def test_an_unrecognised_cause_is_treated_as_broken_not_tolerable() -> None: + # Fail closed on a cause this test does not know: a future degrade path must be classified + # deliberately rather than arriving pre-excused by a permissive default. + assert _is_budget_exhausted("walk_timeout") is True + assert _is_budget_exhausted("something_new") is False + with pytest.raises(AssertionError, match="PROBE DEFECT"): + _assert_fd_probe( + [ + _record(fd=None, ticks=6, degraded_ticks=6, degraded=("something_new",)), + _record(count=24, fd=140, ticks=6), + ] + ) + + +def test_a_run_that_never_measured_wall_four_fails_despite_every_cause_being_tolerable() -> None: + # The bound on the tolerance, and the reason this is not "a tolerance alone". Each step in + # isolation is excusable; a whole run that measured wall #4 zero times proves nothing about wall + # #4, so it fails rather than passing on the strength of four excuses. + with pytest.raises(AssertionError, match="WALL #4 UNMEASURED") as e: + _assert_fd_probe( + [ + _record(count=12, fd=None, ticks=6, degraded_ticks=6, degraded=("walk_timeout",)), + _record(count=24, fd=None, ticks=6, degraded_ticks=6, degraded=("read_timeout",)), + ] + ) + # The verdict has to carry the evidence forward, or it repeats the message-nobody-can-act-on defect. + assert "walk_timeout" in str(e.value) and "read_timeout" in str(e.value) diff --git a/tests/test_connscale_smoke.py b/tests/test_connscale_smoke.py index aa6e5456..955fdefc 100644 --- a/tests/test_connscale_smoke.py +++ b/tests/test_connscale_smoke.py @@ -19,10 +19,13 @@ from __future__ import annotations import sys +from collections.abc import Sequence import pytest +from harness.load.connscale.probe import ProbeDegraded from harness.load.connscale.profile import load_connscale_profile_text +from harness.load.connscale.report import ConnScaleRecord from harness.load.connscale.runner import run_connscale from tests._connscale_ports import ( INBOUND_PORT_HI, @@ -69,6 +72,82 @@ def _smoke_profile(base_port: int) -> object: """) +def _is_budget_exhausted(cause: str) -> bool: + """Is this recorded cause one where the probe SPENT its whole timeout budget? + + Delegates to the probe's own :class:`ProbeDegraded` rather than listing members here, so the + tolerable set cannot drift member-by-member away from the definition. An UNRECOGNISED cause is + treated as not-tolerable: a new degrade path must be classified deliberately, and defaulting the + unknown to "tolerate" is how a fresh probe defect would arrive already excused.""" + try: + return ProbeDegraded(cause).is_budget_exhausted + except ValueError: + return False + + +def _assert_fd_probe(records: Sequence[ConnScaleRecord]) -> None: + """Assert wall #4 to the strength the OS probe's contract actually supports — no more, no less. + + This used to be a bare ``assert r.fd_count_peak is not None and r.fd_count_peak > 0, r`` inside the + loop above, with no tolerance for a probe that honestly could not measure. ``fd_count_peak`` is + ``None`` on every one of the probe's degrade paths, and only some of them implicate anything under + test — so a starved runner whose process-table walk timed out reddened a REQUIRED context as though + the ENGINE were at fault, and did it with a message that named no mechanism. The record now carries + the cause, so this can state a verdict instead of a value check. + + The tolerable/not line is the probe's own (``ProbeDegraded.is_budget_exhausted``), NOT a second + vocabulary invented here. It is the same line ``tests/test_connscale_cpu_probe.py`` draws from the + seconds a failed walk actually spent: budget exhausted is COULD NOT MEASURE, anything faster is + MEASURED AND BROKEN. + + Four properties, and none is weaker than the old assertion wherever the probe worked: + + 1. Where the probe READ, the count must be positive — the wall exists. Unchanged. + 2. Where it did not, the record must NAME a cause. An unattributable gap FAILS. A tolerance that + swallowed it would buy a green by destroying the only evidence of what went wrong, which is the + failure mode this whole change exists to remove. + 3. A cause that is not budget-exhausted FAILS — a walk that returned zero rows, a failed + enumeration, a snapshot with no root row, a per-PID read that ran and parsed nothing. A broken + probe must not hide behind the tolerance written for a slow one. + 4. A budget-exhausted cause is tolerated PER RECORD but not for the run: at least one step must have + measured, so a probe that never works anywhere still cannot pass. That is the same bound + assertion (5) below already places on the reload probe.""" + for r in records: + if r.fd_count_peak is not None: + assert r.fd_count_peak > 0, r + continue + causes = tuple(r.fd_probe_degraded) + scope = f"{r.fd_probe_degraded_ticks} of {r.fd_probe_ticks} probe tick(s) degraded" + assert causes, ( + f"UNATTRIBUTED GAP -- wall #4 read nothing at {r.sweep_mode}@N={r.count} and the record " + f"names no cause ({scope}). A gap that cannot say which probe path produced it is not " + f"evidence about the engine in either direction, and is deliberately NOT tolerated: " + f"tolerating it is what made one starved runner and one broken enumerator the same " + f"artifact. Record: {r}" + ) + broken = tuple(c for c in causes if not _is_budget_exhausted(c)) + assert not broken, ( + f"PROBE DEFECT -- wall #4 read nothing at {r.sweep_mode}@N={r.count} because {broken} " + f"({scope}; all causes {causes}). None of those is an exhausted timeout budget: the probe " + f"RAN and produced nothing usable, which is a defect in the probe itself and is reported " + f"as a failure rather than downgraded to a tolerated gap. Record: {r}" + ) + # Every step degraded, and (given the loop above) every cause was an exhausted budget. Each such + # step alone measures the RUNNER rather than this tree, but a run in which the FD probe never once + # read is not evidence that wall #4 exists -- so the tolerance stops here instead of buying a green + # over zero measurements. + detail = "; ".join( + f"{r.sweep_mode}@N={r.count} {r.fd_probe_degraded_ticks}/{r.fd_probe_ticks} " + f"{tuple(r.fd_probe_degraded)}" + for r in records + ) + assert any(r.fd_count_peak is not None for r in records), ( + f"WALL #4 UNMEASURED -- all {len(records)} step(s) degraded on an exhausted probe budget " + f"[{detail}]. Any single step timing out is tolerated; a whole run measuring wall #4 zero " + f"times is not, because nothing here then covers it." + ) + + async def test_connscale_smoke_end_to_end() -> None: # Reserve a contiguous inbound-port block (BACKLOG #1014). The sweep's max connection count # needs that many contiguous inbound ports, and the engine binds base_port + i for each. A @@ -125,10 +204,12 @@ async def test_connscale_smoke_end_to_end() -> None: for r in report.records: assert r.executor_queue_depth_peak is not None, r # the shim installed the default executor assert r.executor_busy_peak is not None, r - assert r.fd_count_peak is not None and r.fd_count_peak > 0, r # Wall #3 is separated, never summed into one number; both halves are non-negative. assert r.idle_poll_per_s >= 0.0 and r.wake_fanout_per_s >= 0.0 + # (4b) Wall #4 (FD), asserted only as far as this test is ENTITLED to. See _assert_fd_probe. + _assert_fd_probe(report.records) + # (5) The reload-latency probe (wall #5) times an O(connections) quiesce-and-swap. Like the other # OS-side probes it is best-effort and gap-tolerant BY DESIGN: a reload fired mid-hold at the highest # connection count can occasionally exceed the client timeout under peak load, and the probe records a From 7da16ea70456d0b9318a32304342a71e80b05b96 Mon Sep 17 00:00:00 2001 From: wshallwshall Date: Thu, 20 Aug 2026 09:32:00 -0500 Subject: [PATCH 2/3] fix(test): the orphan-reap assertion asked for an ordering the engine does not guarantee `test_worker_kill_reaps_the_whole_process_tree` asserted the LATER of two events at the instant the EARLIER one landed, and it is currently reding a REQUIRED context: PR #458's ubuntu leg failed at tests/test_sandbox.py:998, "the grandchild survived the worker kill", and the same test failed on `main` in run 32206563674. THE DOCSTRING WAS THE DEFECT, NOT ONLY THE TIMING. It claimed pipe-EOF is equivalent to "grandchild reaped". It is not, and the two observables are not the same event: - pipe EOF fires when the last holder of the write end releases its fds -- at process EXIT; - `os.kill(pid, 0)` raises ESRCH only once the pid is REAPED; - the grandchild is ORPHANED, because the sandbox worker (its parent) is killed and `proc.wait()`ed first, so pytest cannot `waitpid` it and reaping falls to PID 1 or the nearest subreaper, asynchronously. So t_reap is STRICTLY AFTER t_exit == t_EOF, always, and the test asserted the later one with no wait, poll or deadline between them. It passed only when the reap won a race it was never entitled to win. TIMING-DEPENDENT AND WRONG ARE NOT ALTERNATIVES HERE -- the ordering is guaranteed by the mechanism. A sleep or a retry would have made it green while leaving the false equivalence in place for the next reader to rely on again, which is why the docstring is part of the change rather than a footnote to it. The two senses of "reap" that collide here are now stated: `_reap_process_tree` reaps in this codebase's sense -- TERMINATE every process in the tree, which is what the engine guarantees and what the pipe-EOF assertion genuinely proves -- while POSIX `waitpid` reaping is a different act on a different schedule, and the test now waits for TERMINATION only. It asks for nothing the engine does not promise. VERIFIED: tests/test_sandbox.py -> 25 passed; the false-equivalence sentence is gone (grep returns 0); ruff 0.15.22 clean; no cp1252-unsafe character introduced. ITS ADVERSARIAL REVIEW HAD NOT REPORTED WHEN THIS WAS COMMITTED -- committed to protect the work across a usage-window boundary, with any finding to be fixed forward. Nothing is pushed. The platform split matters for whoever reads this next: the ordering argument is POSIX-shaped, and this box is Windows, so the CI red is the behavioural evidence rather than a local reproduction. --- tests/test_sandbox.py | 117 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 100 insertions(+), 17 deletions(-) diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py index 60acac8d..86b1475f 100644 --- a/tests/test_sandbox.py +++ b/tests/test_sandbox.py @@ -904,8 +904,40 @@ def _orphan_graph(tmp_path: Path) -> tuple[Registry, str, Path]: return load_config(tmp_path), str(tmp_path), pidfile -def _pid_alive(pid: int) -> bool: - """Whether ``pid`` names a live process — cross-platform, no third-party deps.""" +def _proc_state(pid: int) -> str | None: + """The Linux ``/proc//stat`` process-state field, or ``None`` where ``/proc`` is absent. + + Only ``Z`` is load-bearing here: exited, every fd already released, but the pid is still in the + table because nobody has ``waitpid``-ed it yet.""" + try: + with open(f"/proc/{pid}/stat", encoding="utf-8") as fh: + text = fh.read() + # UnicodeDecodeError is not an OSError, and a non-Linux /proc need not be text at all — a + # platform we cannot read is a "cannot tell", never a crash in a helper the asserts depend on. + except (OSError, UnicodeDecodeError): + return None + if ")" not in text: # pragma: no cover - a Linux /proc always parenthesises comm + return None + # Field 2 (`comm`) can contain spaces and parens, so everything after the LAST ')' is field 3 + # onward and field N sits at index N-3 — the idiom `_posix_stat_ppid_starttime` already uses in + # harness/load/connscale/probe.py. State is field 3, hence index 0. + after = text.rpartition(")")[2].split() + return after[0] if after else None + + +def _pid_running(pid: int) -> bool | None: + """Whether ``pid`` names a process that is still RUNNING. ``None`` = this platform cannot tell a + runner from an exited-but-unreaped pid. + + Deliberately NOT "does this pid exist", because on POSIX those are different questions and the + difference is the whole point of the caller below. ``os.kill(pid, 0)`` keeps succeeding for a + ZOMBIE: a process that has already exited and released every fd it held, but whose pid stays in + the table until its reaper calls ``waitpid``. Terminating a process is something a caller can + demand; retiring its pid afterwards is the reaper's business and on its schedule. + + Windows has no zombie state — there is no exited-but-unretired pid to be fooled by, and + ``GetExitCodeProcess`` stops reporting ``STILL_ACTIVE`` once the process has terminated — so the + plain liveness check already IS the running/not-running answer there.""" if sys.platform == "win32": import ctypes @@ -914,7 +946,7 @@ def _pid_alive(pid: int) -> bool: kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) handle = kernel32.OpenProcess(process_query_limited_information, False, pid) if not handle: - return False # no such pid (or already fully reaped) + return False # no such pid (or already fully gone) try: code = ctypes.c_ulong() if not kernel32.GetExitCodeProcess(handle, ctypes.byref(code)): @@ -925,10 +957,28 @@ def _pid_alive(pid: int) -> bool: try: os.kill(pid, 0) except ProcessLookupError: - return False + return False # reaped: out of the process table entirely except PermissionError: - return True # exists but owned by someone else — still "alive" - return True + return True # exists but owned by someone else — still "running" as far as we can see + state = _proc_state(pid) + if state is None: # POSIX without /proc (e.g. macOS): a zombie and a runner look identical + return None + return state != "Z" + + +def _wait_until_not_running(pid: int, timeout: float) -> bool | None: + """Poll :func:`_pid_running` until ``pid`` stops running, returning its last answer. + + Bounded rather than instantaneous because a POSIX exit is not atomic: the kernel closes the + dying process's fds — which is what releases a pipe — BEFORE it marks the task a zombie, so a + single check taken at the instant of pipe-EOF can still land inside that tail. This waits for + TERMINATION only and never for the reap, so it asks for nothing the engine does not guarantee.""" + deadline = time.monotonic() + timeout + while True: + running = _pid_running(pid) + if running is not True or time.monotonic() >= deadline: + return running + time.sleep(0.01) def _best_effort_kill_pid(pid: int) -> None: @@ -953,21 +1003,42 @@ def _best_effort_kill_pid(pid: int) -> None: def test_worker_kill_reaps_the_whole_process_tree(tmp_path: Path) -> None: - """Killing the worker must reap the WHOLE tree, not just the immediate child (BACKLOG #342). + """Killing the worker must take down the WHOLE tree, not just the immediate child. A Handler spawns a grandchild that inherits fd 1 (the response pipe). Before the fix a bare ``proc.kill()`` terminated only the worker, leaving the grandchild alive — an orphan still holding - the pipe, so the pipe never reached EOF and the kill was incomplete. The fix reaps the tree: a - Windows ``JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`` job object (exercised locally, this host is + the pipe, so the pipe never reached EOF and the kill was incomplete. The fix takes down the tree: + a Windows ``JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`` job object (exercised locally, this host is Windows) or a POSIX new-session process group killed with ``killpg`` (exercised by the CI ubuntu-latest leg). - The observable is platform-neutral: for THIS grandchild — which holds fd 1 until it exits — - pipe-EOF is equivalent to "grandchild reaped", so the primary assert covers BOTH halves of the - defect (pipe released AND no lingering process). ``_pid_alive`` re-checks the process half - directly. See the FALSIFICATION recorded in the lane report: forcing - ``_assign_kill_on_close_job`` to return ``None`` degrades the Windows path to a bare - ``proc.kill()``, the grandchild survives, the pipe never EOFs, and this test goes red.""" + Note the two senses of "reap" that meet here. ``_reap_process_tree`` reaps in this codebase's + sense — TERMINATE every process in the tree. POSIX ``waitpid`` reaping is a different act — + RETIRE an already-exited pid from the process table — and the engine neither performs nor can + bound it for an orphan. The asserts below are split along exactly that line: + + * PRIMARY, pipe-EOF: proves every holder of fd 1 has EXITED and released it — the worker AND the + grandchild. That is precisely what ``_reap_process_tree`` guarantees, so it is the pass/fail + proof of the fix. + * SECONDARY, the process half: proves the grandchild is NOT RUNNING. It must not ask whether the + pid was ``waitpid``-ed, which an earlier version of this test did by asserting + ``os.kill(pid, 0)`` raises the instant EOF landed. That ordering is not available: this + grandchild is an ORPHAN — its parent was killed and ``proc.wait()``-ed first, so pytest cannot + ``waitpid`` it and the reap falls to PID 1 or the nearest ``PR_SET_CHILD_SUBREAPER`` ancestor. + A reap is therefore STRICTLY AFTER the exit that produced the EOF, by an interval nothing in + this repo controls. Measured on Linux with the same shape: microseconds under systemd, and + never at all inside a 5s busy-poll under a subreaper that is not in a ``waitpid`` loop — which + is what a containerised CI leg or a ``systemd --user`` session supplies. Windows has no zombie + state to be caught by, and measured on this host it never once reported ``STILL_ACTIVE`` at + pipe-EOF (0 of 30, plain and slow-teardown grandchildren both), so the exposure is a POSIX one. + + The secondary is not redundant with the primary: it is the guard against the primary going + VACUOUSLY green if someone later edits ``_ORPHAN_GRAPH`` so the grandchild no longer holds fd 1, + in which case EOF would fire on the worker's death alone and say nothing about the tree. + + See the FALSIFICATION recorded in the lane report: forcing ``_assign_kill_on_close_job`` to + return ``None`` degrades the Windows path to a bare ``proc.kill()``, the grandchild survives, + the pipe never EOFs, and this test goes red.""" registry, config_dir, pidfile = _orphan_graph(tmp_path) session = _session(config_dir) grandchild_pid: int | None = None @@ -994,8 +1065,20 @@ def test_worker_kill_reaps_the_whole_process_tree(tmp_path: Path) -> None: "response pipe never reached EOF -- a grandchild still holds it; the worker tree " "was not reaped" ) - # SECONDARY: the process half, asserted directly. - assert not _pid_alive(grandchild_pid), "the grandchild survived the worker kill" + # SECONDARY: the process half, asserted directly — NOT-RUNNING, not reaped (see docstring). + # The 5s bound is derived from the FALSIFICATION MARGIN, not from any reaper's latency: the + # grandchild sleeps 30s, so one that genuinely survived the kill is still running through + # every one of these seconds and the assert stays red. It also matches `_kill`'s own + # `proc.wait(timeout=5)`. Waiting here can therefore only absorb a process's exit tail; it + # can never convert the defect into a pass. + running = _wait_until_not_running(grandchild_pid, timeout=5.0) + # `None` means the platform cannot separate a zombie from a runner (POSIX without /proc, + # e.g. macOS) and there is no sound process-half assertion to make there. Neither CI + # platform is one — Windows has no zombie state, Linux has /proc — so pin that: a `None` + # on either is `_pid_running` having broken, and must not slip through as "nothing to say". + if sys.platform == "win32" or sys.platform.startswith("linux"): + assert running is not None, "_pid_running went blind on a platform CI actually runs" + assert running is not True, "the grandchild is still running after the worker kill" finally: if grandchild_pid is not None: _best_effort_kill_pid(grandchild_pid) From bc809aa61310f7eca8f0aad6ae5585f9a888ed0c Mon Sep 17 00:00:00 2001 From: wshallwshall Date: Thu, 20 Aug 2026 09:48:50 -0500 Subject: [PATCH 3/3] fix(test): the run-level FD bound let a real collapse pass, and the POSIX probe called a success blind Two defects, both found by the adversarial reviews of 937803a9 and 1a514e14, both proven end-to-end by the reviewers rather than argued. Fixed forward; neither commit had left this tree. FINDING 1 (CRITICAL) -- THE GREEN MEANT LESS, IN EXACTLY THE WAY THE BRIEF PREDICTED. The one-line assertion 937803a9 replaced was doing DOUBLE DUTY: besides checking each reading, it was the only thing guaranteeing `fd_count_monotonic` had two readings in a group to COMPARE. `_monotonic_slo` SKIPS None readings by design, so with a run-wide `any(measured)` bound it can return ok=True observed='monotonic' having compared NOTHING. Measured on the real smoke: forcing every step after the first to WALK_TIMEOUT left one group holding a single reading, PAIRS ACTUALLY COMPARED=0, and the SLO still reported monotonic -- and a genuine 1000-handle FD collapse passed the same way, while the unforced control compared 2 pairs and a real 1000->100 drop correctly failed. So the bound is now at the SLO's GRANULARITY, not the run's: some (sweep_mode, claim_mode) group must hold two measured readings, which is the same key `_monotonic_slo` groups on (BACKLOG #1101). Grouping on sweep_mode alone would reintroduce the bug one level up, since two readings split across claim modes are never compared with each other. AND THE JUSTIFICATION I SHIPPED FOR THE WEAK BOUND WAS ITSELF WRONG. It claimed the run-wide form was "the same bound assertion (5) already places on the reload probe". Same SHAPE, wrong ANALOGY: `reload_seconds` feeds no monotonicity SLO, so one reading there costs nothing, while `fd_count_peak` feeds one. Borrowing a bound from the one probe where the analogy is safe is how this got through. FINDING 2 -- THE POSIX PATH CALLED A SUCCESS BLINDNESS. `_pid_running` did `os.kill(pid, 0)` and then read /proc, two syscalls with a gap. If the reaper retired the pid inside that gap the /proc read raised FileNotFoundError, `_proc_state` returned None, and the test's blindness guard turned a SUCCESSFUL reap into a red on Linux -- a new false failure introduced by 1a514e14 on the platform where the original red actually fires. A missing /proc/ on a kernel that HAS /proc is a definite answer, "gone", not an inability to answer. Only a platform with no /proc at all (macOS) genuinely cannot separate a zombie from a runner, and None is now reserved for that. The check reads the DIRECTORY rather than `sys.platform`, because the question is whether this kernel exposes the interface, which is what the code actually depends on. THREE TESTS MOVED WITH THE SEMANTICS, and one is renamed rather than merely edited: `test_a_spent_budget_gap_is_tolerated_when_another_step_measured` becomes `..._is_tolerated_beside_a_comparable_pair`, because "another step measured" was never the condition -- two readings split so that no group holds two are never compared. A test whose NAME asserts the wrong rule is a second source of truth for it. ADDED test_one_measured_reading_in_a_group_is_not_a_comparison as the regression test for finding 1, carrying the measurement that proves it: a green resting on an empty comparison is what this file exists to prevent, and it had just shipped one. VERIFIED: 31 passed in test_connscale_probe_degradation.py; 27 passed across test_connscale_smoke.py and test_sandbox.py; the corrected bound rejects the reviewer's failing shape (per-group={...: 1}) and accepts the control ({...: 2}). ruff 0.15.22 clean. Windows host: the POSIX branch of finding 2 is argued from the code and the reviewer's WSL2 measurement, not reproduced locally, and I am not claiming otherwise. --- tests/test_connscale_probe_degradation.py | 36 +++++++++++++++++++---- tests/test_connscale_smoke.py | 33 ++++++++++++++++++--- tests/test_sandbox.py | 16 ++++++++-- 3 files changed, 73 insertions(+), 12 deletions(-) diff --git a/tests/test_connscale_probe_degradation.py b/tests/test_connscale_probe_degradation.py index ec29a39b..193ee2f2 100644 --- a/tests/test_connscale_probe_degradation.py +++ b/tests/test_connscale_probe_degradation.py @@ -516,22 +516,46 @@ def test_the_console_names_the_mechanism_beside_the_n_a() -> None: def test_a_measured_record_is_asserted_exactly_as_strictly_as_before() -> None: # The pre-existing strength is preserved where the probe worked: a positive count passes, and a # non-positive one still fails. The change adds a verdict for gaps; it does not relax readings. - _assert_fd_probe([_record(fd=100, ticks=6)]) + # TWO measured readings, because the run-level bound requires a comparable PAIR -- see + # test_one_measured_reading_in_a_group_is_not_a_comparison for why one is not enough. + _assert_fd_probe([_record(count=12, fd=100, ticks=6), _record(count=24, fd=110, ticks=6)]) with pytest.raises(AssertionError): - _assert_fd_probe([_record(fd=0, ticks=6)]) + _assert_fd_probe([_record(count=12, fd=0, ticks=6), _record(count=24, fd=110, ticks=6)]) -def test_a_spent_budget_gap_is_tolerated_when_another_step_measured() -> None: - # The starved-runner case that was reddening a required context. Tolerated -- but only alongside a - # step that actually measured, which is what stops the tolerance becoming a blanket green. +def test_a_spent_budget_gap_is_tolerated_beside_a_comparable_pair() -> None: + # The starved-runner case that was reddening a required context. Tolerated -- but only where the + # SLO that covers wall #4 still had a pair to compare. Renamed from "when another step measured": + # one other measuring step is NOT the condition, because two readings split so that no group holds + # two are never compared with each other. _assert_fd_probe( [ _record(count=12, fd=None, ticks=6, degraded_ticks=6, degraded=("walk_timeout",)), _record(count=24, fd=140, ticks=6), + _record(count=48, fd=150, ticks=6), ] ) +def test_one_measured_reading_in_a_group_is_not_a_comparison() -> None: + # THE REGRESSION TEST FOR A DEFECT THAT SHIPPED HERE. The first version of the run-level bound was + # `any(fd_count_peak is not None)`, which passes on a run where exactly one step measured -- and + # `_monotonic_slo` SKIPS None readings, so `fd_count_monotonic` then reports ok=True having + # compared ZERO pairs. Measured on the real smoke at the time: forcing every step after the first + # to WALK_TIMEOUT gave PAIRS ACTUALLY COMPARED=0 and the SLO still said "monotonic", and a genuine + # 1000-handle collapse passed the same way. A green that rests on an empty comparison is exactly + # what this file exists to prevent. + with pytest.raises(AssertionError, match="WALL #4 NEVER COMPARED") as e: + _assert_fd_probe( + [ + _record(count=12, fd=1000, ticks=6), + _record(count=24, fd=None, ticks=6, degraded_ticks=6, degraded=("walk_timeout",)), + ] + ) + # The verdict must show WHY it is not a comparison, not merely that it failed. + assert "per group" in str(e.value) + + def test_a_broken_probe_still_fails_and_the_message_names_the_cause() -> None: # The scope of this work is explicit that a probe which enumerated and returned zero rows must # STILL FAIL. A tolerance that swallowed this would buy a green by deleting the finding. @@ -585,7 +609,7 @@ def test_a_run_that_never_measured_wall_four_fails_despite_every_cause_being_tol # The bound on the tolerance, and the reason this is not "a tolerance alone". Each step in # isolation is excusable; a whole run that measured wall #4 zero times proves nothing about wall # #4, so it fails rather than passing on the strength of four excuses. - with pytest.raises(AssertionError, match="WALL #4 UNMEASURED") as e: + with pytest.raises(AssertionError, match="WALL #4 NEVER COMPARED") as e: _assert_fd_probe( [ _record(count=12, fd=None, ticks=6, degraded_ticks=6, degraded=("walk_timeout",)), diff --git a/tests/test_connscale_smoke.py b/tests/test_connscale_smoke.py index 955fdefc..18ed8a85 100644 --- a/tests/test_connscale_smoke.py +++ b/tests/test_connscale_smoke.py @@ -141,10 +141,35 @@ def _assert_fd_probe(records: Sequence[ConnScaleRecord]) -> None: f"{tuple(r.fd_probe_degraded)}" for r in records ) - assert any(r.fd_count_peak is not None for r in records), ( - f"WALL #4 UNMEASURED -- all {len(records)} step(s) degraded on an exhausted probe budget " - f"[{detail}]. Any single step timing out is tolerated; a whole run measuring wall #4 zero " - f"times is not, because nothing here then covers it." + # THE BOUND IS AT THE SLO'S GRANULARITY, NOT THE RUN'S, AND THE DIFFERENCE IS A REAL DEFECT THAT + # SHIPPED HERE FIRST. The obvious form -- "at least one record measured" -- is not enough, because + # the per-record assertion this replaced was doing DOUBLE DUTY: it was also the only thing + # guaranteeing `fd_count_monotonic` had two readings in a group to COMPARE. `_monotonic_slo` + # SKIPS None readings (runner.py, "Missing readings (None) are skipped, not failed"), so with a + # run-wide bound it can return ok=True observed='monotonic' having compared NOTHING -- and a real + # FD collapse then passes. Measured on this smoke: forcing every step after the first to + # WALK_TIMEOUT left one group with a single reading, PAIRS ACTUALLY COMPARED=0, and the SLO still + # reported monotonic; a 1000-handle collapse at N=12 with both N=24 steps degraded passed the same + # way, while the unforced control compared 2 pairs and a genuine 1000->100 drop correctly failed. + # + # So require that some (sweep_mode, claim_mode) group -- the SAME key `_monotonic_slo` groups on + # (BACKLOG #1101) -- actually has a pair. Grouping on sweep_mode alone would re-introduce the bug + # one level up: two readings split across claim modes are never compared with each other. + # + # NOT MIRRORED FROM assertion (5). An earlier version of this justified the run-wide bound as "the + # same bound the reload probe already has". Same SHAPE, wrong ANALOGY: `reload_seconds` feeds no + # monotonicity SLO, so a single reading there costs nothing. `fd_count_peak` feeds one. + measured_per_group: dict[tuple[str, str], int] = {} + for r in records: + if r.fd_count_peak is not None: + key = (r.sweep_mode, r.claim_mode) + measured_per_group[key] = measured_per_group.get(key, 0) + 1 + assert any(n >= 2 for n in measured_per_group.values()), ( + f"WALL #4 NEVER COMPARED -- no (sweep_mode, claim_mode) group has two measured readings, so " + f"`fd_count_monotonic` compared zero pairs and its green says nothing. Measured readings per " + f"group: {measured_per_group or '{}'} [{detail}]. A single step timing out is tolerated; a run " + f"in which the FD wall is never compared against itself is not, because the SLO that is " + f"supposed to cover it silently passes over an empty set." ) diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py index 86b1475f..fb64f56a 100644 --- a/tests/test_sandbox.py +++ b/tests/test_sandbox.py @@ -961,8 +961,20 @@ def _pid_running(pid: int) -> bool | None: except PermissionError: return True # exists but owned by someone else — still "running" as far as we can see state = _proc_state(pid) - if state is None: # POSIX without /proc (e.g. macOS): a zombie and a runner look identical - return None + if state is None: + # `_proc_state` returns None for TWO situations that must not be collapsed, and collapsing + # them turns a SUCCESS into a red on Linux. `os.kill(pid, 0)` above and the `/proc` read here + # are two syscalls with a gap between them, and a reaper can retire the pid inside that gap -- + # so a MISSING `/proc/` on a kernel that HAS `/proc` is a definite answer, "gone", not an + # inability to answer. Only a platform with no `/proc` at all (macOS) genuinely cannot tell a + # zombie from a runner, and that is the case `None` is reserved for. + # + # Reading the directory rather than the platform string: `sys.platform` says which OS, and the + # question here is whether THIS kernel exposes the interface. They agree today and the second + # is what the code actually depends on. + if os.path.isdir("/proc"): + return False # /proc exists and this pid is not in it: retired between the two calls + return None # no /proc at all: a zombie and a runner are indistinguishable here return state != "Z"