diff --git a/testing/chaos/scenarios/congestion-stress.yaml b/testing/chaos/scenarios/congestion-stress.yaml index 8b68c72..8dce07e 100644 --- a/testing/chaos/scenarios/congestion-stress.yaml +++ b/testing/chaos/scenarios/congestion-stress.yaml @@ -91,6 +91,44 @@ traffic: node_churn: enabled: false +# Three of the four success criteria above, encoded. Until now all four +# existed only as that comment and were checked by nothing, so the scenario +# could not fail on any of them. +# +# The floors are 1 node each because that is what the criteria say ("on at +# least one forwarding node", "on transit nodes", "on destination nodes"). +# They are deliberately not tightened to the observed counts: the criterion +# is the spec, and a floor invented from six runs would assert something +# nobody wrote down. +# +# What the floors are worth, from the archived corpus. The six runs that +# carry a status.txt -- the only ones provably completed, all between +# 2026-07-22 22:14 and 2026-07-23 02:02 -- meet all three with 5 to 9 nodes +# reporting each signal, so the margin over a floor of 1 is comfortable. +# The 176 older runs meet none of them. Those older runs did reach teardown +# (each wrote an analysis.txt), and ECN landed 2026-03-05, before all but +# one of them, so they are valid runs that observed no congestion rather +# than runs that died early. What changed on 2026-07-22 is not established: +# this file has not been touched since it was created, and neither have +# netem.py, traffic.py or control.py. Worth knowing before trusting a green +# result here, and worth its own investigation. +assertions: + congestion_signals: + min_nodes_detected: 1 + min_nodes_ce_forwarded: 1 + min_nodes_ce_received: 1 + +# The fourth criterion, "kernel_drop_events > 0 on at least one node", is +# NOT asserted, because the implementation does not meet it: across all 182 +# archived runs of this scenario, including the six that meet the other +# three, no node has ever reported a non-zero kernel_drop_events. Asserting +# it would red the scenario permanently, and asserting a weakened version +# would assert nothing. It stands here as an unmet criterion: either the +# 4 KB recv_buf_size no longer provokes SO_RXQ_OVFL under this traffic, or +# the counter does not reach the snapshot. Until that is settled it is a +# coverage gap, because the transport drop-detection path this scenario +# names as signal 2 is exercised by nothing. + logging: rust_log: "info" output_dir: "./sim-results" diff --git a/testing/chaos/sim/assertions.py b/testing/chaos/sim/assertions.py index 169189c..6a391d1 100644 --- a/testing/chaos/sim/assertions.py +++ b/testing/chaos/sim/assertions.py @@ -20,6 +20,7 @@ from dataclasses import dataclass from .control import snapshot_all_bloom from .scenario import ( BloomSendRateAssertion, + CongestionSignalsAssertion, MaxErrorsAssertion, MaxParentSwitchesAssertion, MinParentSwitchesAssertion, @@ -171,6 +172,70 @@ def evaluate_max_parent_switches( ) +_CONGESTION_FLOORS = ( + ("min_nodes_detected", "congestion_detected"), + ("min_nodes_ce_forwarded", "ce_forwarded"), + ("min_nodes_ce_received", "ce_received"), +) + + +def evaluate_congestion_signals( + cfg: CongestionSignalsAssertion, + snapshot: dict | None, +) -> AssertionOutcome: + """Floors on how many nodes observed each congestion counter. + + ``snapshot`` is the final congestion snapshot keyed by node id. None + means the snapshot never ran, which fails: a missing snapshot and a + mesh that observed no congestion produce the same zero counts, and + treating them alike is how an assertion comes to pass on an absence + of evidence. + """ + if not snapshot: + return AssertionOutcome( + name="congestion_signals", + passed=False, + detail=( + "FAIL congestion_signals: no final congestion snapshot was " + "taken, so no node was observed at all. This is a harness " + "failure, not a statement about congestion." + ), + ) + + parts, failures = [], [] + for attr, counter in _CONGESTION_FLOORS: + floor = getattr(cfg, attr) + if floor is None: + continue + hits = sorted( + nid for nid, data in snapshot.items() + if (data.get("congestion") or {}).get(counter, 0) > 0 + ) + parts.append(f"{counter}: {len(hits)} node(s) >0 (floor {floor})") + if len(hits) < floor: + failures.append( + f"{counter} non-zero on {len(hits)} node(s), need {floor}" + ) + else: + parts[-1] += f" [{', '.join(hits)}]" + + summary = "; ".join(parts) + if failures: + return AssertionOutcome( + name="congestion_signals", + passed=False, + detail=( + f"FAIL congestion_signals: {'; '.join(failures)} — across " + f"{len(snapshot)} node(s) sampled. Full counts: {summary}" + ), + ) + return AssertionOutcome( + name="congestion_signals", + passed=True, + detail=f"PASS congestion_signals: {summary}", + ) + + def evaluate_max_errors( cfg: MaxErrorsAssertion, errors: list[tuple[str, str]], diff --git a/testing/chaos/sim/runner.py b/testing/chaos/sim/runner.py index 6259eb9..97ad1a8 100644 --- a/testing/chaos/sim/runner.py +++ b/testing/chaos/sim/runner.py @@ -15,6 +15,7 @@ from datetime import datetime from .assertions import ( AssertionOutcome, BloomSendRateMonitor, + evaluate_congestion_signals, evaluate_max_errors, evaluate_max_parent_switches, evaluate_min_parent_switches, @@ -75,6 +76,10 @@ class SimRunner: # Post-run assertion monitors (sampled near end of run). self.bloom_rate_monitor: BloomSendRateMonitor | None = None self.assertion_outcomes: list[AssertionOutcome] = [] + # Set by the final snapshot. None means the snapshot never ran, + # which the congestion assertion must treat as a failure rather + # than as an absence of congestion. + self.final_congestion: dict | None = None def _evaluate_max_parent_switches( self, cfg, parent_switches: list[tuple[str, str]] @@ -617,6 +622,17 @@ class SimRunner: else: log.error("%s", outcome.detail) + cong_cfg = self.scenario.assertions.congestion_signals + if cong_cfg is not None: + outcome = evaluate_congestion_signals( + cong_cfg, self.final_congestion + ) + self.assertion_outcomes.append(outcome) + if outcome.passed: + log.info("%s", outcome.detail) + else: + log.error("%s", outcome.detail) + # Write assertion outcomes if self.assertion_outcomes: assertions_path = os.path.join(self.output_dir, "assertions.txt") @@ -677,6 +693,11 @@ class SimRunner: tree_path = os.path.join(self.output_dir, f"tree-snapshot-{label}.json") mmp_path = os.path.join(self.output_dir, f"mmp-snapshot-{label}.json") congestion_path = os.path.join(self.output_dir, f"congestion-snapshot-{label}.json") + # Retained for the congestion assertion, which needs the same + # responses the file gets. Reading the file back would let a write + # failure present as an assertion that saw nothing. + if label == "final": + self.final_congestion = congestion_snap os.makedirs(self.output_dir, exist_ok=True) with open(tree_path, "w") as f: json.dump(tree_snap, f, indent=2) diff --git a/testing/chaos/sim/scenario.py b/testing/chaos/sim/scenario.py index 9980d5e..39171a5 100644 --- a/testing/chaos/sim/scenario.py +++ b/testing/chaos/sim/scenario.py @@ -218,6 +218,26 @@ class MaxParentSwitchesAssertion: node: str | None = None +@dataclass +class CongestionSignalsAssertion: + """Floors on how many nodes observed each congestion signal. + + Each floor is the number of nodes whose final congestion snapshot + reports a non-zero counter, not the counter's own magnitude: the + scenario's written criteria are about *whether* a signal reached a + class of node, and a single node with a huge count would satisfy a + magnitude test while proving the signal never propagated. + + A floor left unset is not asserted. At least one must be set, since a + block that asserts nothing is the failure this assertion exists to + remove. + """ + + min_nodes_detected: int | None = None + min_nodes_ce_forwarded: int | None = None + min_nodes_ce_received: int | None = None + + @dataclass class MaxErrorsAssertion: """Ceiling on ERROR-level lines across every node's log. @@ -248,6 +268,7 @@ class AssertionsConfig: min_parent_switches: MinParentSwitchesAssertion | None = None max_parent_switches: MaxParentSwitchesAssertion | None = None max_errors: MaxErrorsAssertion | None = None + congestion_signals: CongestionSignalsAssertion | None = None @dataclass @@ -324,7 +345,7 @@ _SECTION_KEYS = { "link_swap.edges[]": {"edge", "policy"}, "assertions": { "bloom_send_rate", "min_parent_switches", "max_parent_switches", - "max_errors", + "max_errors", "congestion_signals", }, "logging": {"rust_log", "output_dir"}, } @@ -333,6 +354,9 @@ _ASSERTION_KEYS = { "min_parent_switches": {"min_total"}, "max_parent_switches": {"max_total", "node"}, "max_errors": {"max_total"}, + "congestion_signals": { + "min_nodes_detected", "min_nodes_ce_forwarded", "min_nodes_ce_received", + }, } _NETEM_POLICY_KEYS = { "delay_ms", "jitter_ms", "loss_pct", "duplicate_pct", "reorder_pct", @@ -642,6 +666,37 @@ def load_scenario(path: str) -> Scenario: # run means, and a scenario that has to opt in is a scenario that # can forget to. s.assertions.max_errors = MaxErrorsAssertion() + if "congestion_signals" in asrt: + cs = asrt["congestion_signals"] + _reject_unknown( + cs, _ASSERTION_KEYS["congestion_signals"], + "assertions.congestion_signals", + ) + floors = {} + for key in _ASSERTION_KEYS["congestion_signals"]: + if key not in cs: + continue + val = cs[key] + if isinstance(val, bool) or not isinstance(val, int): + raise ValueError( + f"assertions.congestion_signals.{key}: must be a positive " + f"integer number of nodes, got {val!r}" + ) + if val < 1: + raise ValueError( + f"assertions.congestion_signals.{key}: must be at least 1, " + f"got {val}. A floor of 0 is satisfied by a mesh that " + f"observed nothing, which is the case this assertion exists " + f"to catch; omit the key instead." + ) + floors[key] = val + if not floors: + raise ValueError( + "assertions.congestion_signals: set at least one floor " + "(min_nodes_detected, min_nodes_ce_forwarded, " + "min_nodes_ce_received); a block with none asserts nothing" + ) + s.assertions.congestion_signals = CongestionSignalsAssertion(**floors) # Logging section lg = raw.get("logging", {})