From b3a1fb464f9d899abc5f24f536c0dee20a8458f0 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 8 May 2026 14:23:08 +0000 Subject: [PATCH] testing: add bloom-storm chaos scenario Six-node depth-4 mesh with an induced upstream parent flap. Asserts a trailing-window ceiling on per-node `stats.bloom.sent` and a sanity floor on parent-switch count over a ~3-4 min observation window. Guards against the regression class where a spanning-tree update that changes only an internal path edge (no root or depth delta) fails to be properly contained and instead propagates to leaves as a sustained bloom-traffic oscillation, visible only at fleet scale and only after several minutes of uptime. Adds a new chaos primitive (`link_swap`) for deterministic asymmetric link-cost flapping and a post-run assertion framework with two checks: - `bloom_send_rate.max_per_node`: trailing-window ceiling on the `show_bloom` stats counter delta. Calibrated against the post-mortem reproduction harness data (per-variant counter table against pre-fix vs post-fix binaries). - `min_parent_switches.min_total`: sanity guard against a misconfigured harness where the flap inducer fires but the topology never produces a real parent-switch event (e.g., wrong root election from a different seed). Without this, the bloom-rate assertion would trivially pass on any binary including a regressed one. The runner exits 3 on assertion failure (alongside 0 success and 2 panic-detected). Threshold derivation is documented in the scenario README; the seed pin is also documented there since smallest-NodeAddr root election is sensitive to the pubkey hash ordering. Wired into ci-local.sh's chaos pool and the GitHub CI chaos matrix. --- .github/workflows/ci.yml | 3 + testing/chaos/scenarios/bloom-storm.README.md | 145 ++++++++++++++++ testing/chaos/scenarios/bloom-storm.yaml | 142 +++++++++++++++ testing/chaos/sim/__main__.py | 2 + testing/chaos/sim/assertions.py | 163 ++++++++++++++++++ testing/chaos/sim/control.py | 21 +++ testing/chaos/sim/link_swap.py | 151 ++++++++++++++++ testing/chaos/sim/runner.py | 117 +++++++++++++ testing/chaos/sim/scenario.py | 123 +++++++++++++ testing/ci-local.sh | 2 + 10 files changed, 869 insertions(+) create mode 100644 testing/chaos/scenarios/bloom-storm.README.md create mode 100644 testing/chaos/scenarios/bloom-storm.yaml create mode 100644 testing/chaos/sim/assertions.py create mode 100644 testing/chaos/sim/link_swap.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e5ebed1..7361f04 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -366,6 +366,9 @@ jobs: - suite: congestion-stress type: chaos scenario: congestion-stress + - suite: bloom-storm + type: chaos + scenario: bloom-storm # ── Sidecar deployment ────────────────────────────────────────── - suite: sidecar type: sidecar diff --git a/testing/chaos/scenarios/bloom-storm.README.md b/testing/chaos/scenarios/bloom-storm.README.md new file mode 100644 index 0000000..346305e --- /dev/null +++ b/testing/chaos/scenarios/bloom-storm.README.md @@ -0,0 +1,145 @@ +# bloom-storm — bloom announce storm regression scenario + +Six-node depth-4 mesh with an induced upstream parent flap. Asserts a +trailing-window ceiling on per-node `stats.bloom.sent` to catch a +regression class where a localized spanning-tree update at an +internal mid-chain edge fails to be properly contained and instead +propagates as a bloom announce storm down to the leaves. + +## Topology + +```text + n01 (root, depth 0) + / \ + n02 n03 (parent candidates pa/pb, depth 1) + \ / + n04 (flap, depth 2 — induced parent flap) + | + n05 (leaf, depth 3) + | + n06 (tail, depth 4) +``` + +n01 is expected to win the root election under the deterministic key +derivation used by the chaos runner; the run's tree snapshots +(`tree-snapshot-warmup.json`, `tree-snapshot-final.json`) record the +actual assignment. + +## Bug class guarded against + +A spanning-tree update that changes only an internal path edge — no +root change, no depth change — must not produce a sustained bloom +announce storm at downstream nodes. The original regression +(rolled-back `0caef2a`, fixed in master `4cdf382`) had this property: +in the field, a single mid-chain ancestor swap on an upstream node +caused every downstream node in its subtree to issue a bloom +announce on every parent re-evaluation tick of the upstream node, +resulting in a ~480x mesh-wide elevation in `FilterAnnounce` traffic +and a perfect bimodal flip on `est_entries`. + +## Mechanism + +The new `link_swap` chaos primitive deterministically alternates the +netem delay on `n02-n04` (5ms vs 100ms) and `n03-n04` (100ms vs 5ms) +every 4 seconds. Combined with the FIPS overrides +(`parent_hysteresis: 0.0`, `reeval_interval_secs: 1`, +`flap_threshold: 9999`, `hold_down_secs: 0`), this forces n04 to +switch parents on every swap. The whole point of the scenario is to +*force* sustained parent flapping at n04 and assert that the bloom +layer doesn't amplify it. + +## Assertions + +```yaml +assertions: + bloom_send_rate: + window_secs: 30 + max_per_node: 30 + min_parent_switches: + min_total: 10 +``` + +`bloom_send_rate` is the load-bearing assertion: per-node delta of +`stats.bloom.sent` over the trailing 30s of the run must be at most +30. Per-node deltas and the offending node IDs are written to +`assertions.txt` and the runner exits 3 on failure. + +`min_parent_switches` is a sanity guard. It fails if the run did not +record at least 10 parent switches across all nodes, which would +mean the harness fired its link swaps but the topology never produced +a real parent-switch event (e.g., wrong root election made the flap +target's parent candidates structurally non-equivalent). Without this +guard, the bloom-rate assertion would trivially pass on any binary, +including a regressed one. + +## Threshold derivation + +The original `issues/2026-0019-repro/` reproduction harness measured +(90s flap window, ~21 induced parent switches at the mid-chain +node): + +| binary | tail bloom_sent / 90s | rate scaled to 30s | +| ----------- | --------------------: | -----------------: | +| pre-fix | 21 | ~7 | +| current fix | 0 | 0 | + +Observed on this scenario at master `db5b6b1` (180s run, 35 parent +switches, 41 link swaps), per-node `bloom_sent` deltas over the +trailing 30s: + +```text +n01=5 n02=5 n03=4 n04=12 n05=6 n06=0 +``` + +n04 (the flapping node) is the highest because it is legitimately +re-sending its filter on its own parent changes. n06 (the depth-4 +"tail") sees 0, matching the calm post-fix behavior recorded in +`issues/2026-0019-repro/RESULTS.md` for the `fix2` variant. + +In the field, the regression's mesh-wide rate scaled ~480x above +steady state. A `30 / 30s / node` ceiling sits ~2.5x above the +observed maximum on fixed master and well below the +deployment-scale storm rate, giving headroom for harness jitter +without losing the ability to fail loud on the regression class. + +If `link_swap.interval_secs` or the netem delta is changed, +recalibrate. The threshold is calibrated against the values in +`scenarios/bloom-storm.yaml` as committed and the `seed: 31` pin. + +## Limitations + +- The bloom-storm regression has not been confirmed-failing here + on a regressed binary in this harness directly; the threshold is + inferred from the values measured in the dedicated + `issues/2026-0019-repro/` post-mortem harness against + `0caef2a`. To gain that confirmation, check out `0caef2a` + (or the `backup-broadcast-gate-bloom-storm` branch if still + retained), build, copy binaries into `testing/docker/`, and rerun + this scenario; the bloom-rate assertion is expected to fail loud + with n05/n06 deltas well above 30. + +- Root-election outcome is sensitive to the seed (smallest + `NodeAddr` wins, where `NodeAddr = SHA-256(pubkey)[..16]`). The + seed value `31` is pinned for this reason. The + `min_parent_switches` assertion catches drift if the seed is + changed without re-validating the topology. + +## Running locally + +```bash +# From the source repo root, with binaries already built and copied +# into testing/docker/ (see testing/scripts/build.sh). +./testing/chaos/scripts/chaos.sh bloom-storm +``` + +Run output is in `testing/chaos/sim-results/-bloom-storm/`. +Key artifacts: + +- `analysis.txt` — log analysis (panics, errors, parent switches). +- `assertions.txt` — per-assertion pass/fail with per-node deltas. +- `tree-snapshot-warmup.json`, `tree-snapshot-final.json` — control + socket tree state at warmup end and at run end. +- `runner.log` — full orchestration log. + +Total runtime: ~3.5 minutes (25s warmup + 180s scenario + ~30s +teardown). diff --git a/testing/chaos/scenarios/bloom-storm.yaml b/testing/chaos/scenarios/bloom-storm.yaml new file mode 100644 index 0000000..b819605 --- /dev/null +++ b/testing/chaos/scenarios/bloom-storm.yaml @@ -0,0 +1,142 @@ +# Bloom-storm regression scenario. +# +# Six-node depth-4 mesh with an induced upstream parent flap. Guards +# against a regression class where a spanning-tree update that changes +# only an internal path edge (no root or depth delta) fails to +# propagate to leaves, producing a sustained bloom-traffic +# oscillation visible only at fleet scale and only after several +# minutes of uptime. See README.md alongside this file for the +# bug-class description and threshold derivation. +# +# Topology (n01 should win the root election under deterministic +# key derivation; the validation snapshot at end-of-run will confirm +# the assigned root): +# +# n01 (root, depth 0) +# / \ +# n02 n03 (parent candidates pa/pb, depth 1) +# \ / +# n04 (flap, depth 2 — induced parent flap) +# | +# n05 (leaf, depth 3 — observes mid-chain swap) +# | +# n06 (tail, depth 4 — sees the cascade) +# +# The flap is induced by alternating the netem delay on n02-n04 and +# n03-n04 every 4 seconds. Because the FIPS overrides below disable +# parent-flap dampening, n04 switches parents on each swap. The bug +# class under test is whether this localized swap at depth 2 leaks +# downstream as a bloom-announce storm at depth 3 and depth 4. +# +# Assertions (bloom_send_rate.max_per_node_per_window) are evaluated +# over the trailing 30s of the run, after sustained flapping has had +# time to expose any cascade behavior. + +scenario: + name: "bloom-storm" + # Seed pinned so deterministic key derivation makes n01 win the + # root election and orders the other NodeAddrs as n02 < n03 < + # ... < n04 (so n04 is at the bottom of the tiebreak and its + # parent candidates n02/n03 are both at depth 1). The warmup tree + # snapshot in sim-results/ is the authoritative check. + # + # Other seeds may elect a different node as root or order n04 + # ahead of n02/n03, in which case the induced flap will not + # exercise the cascade path the assertion is written to catch. + seed: 31 + duration_secs: 180 + +topology: + algorithm: explicit + num_nodes: 6 + params: + adjacency: + - [n01, n02] + - [n01, n03] + - [n02, n04] + - [n03, n04] + - [n04, n05] + - [n05, n06] + subnet: "172.20.0.0/24" + ip_start: 10 + +netem: + enabled: true + default_policy: + # Calm baseline so steady-state bloom traffic is at its floor. + delay_ms: [1, 1] + jitter_ms: [0, 0] + loss_pct: [0, 0] + +# Asymmetric link-cost flap on n04's two upstream candidates. Every +# `interval_secs`, the policies on the two listed edges are swapped. +# This drives n04 to alternate parents between n02 and n03 each +# round. The 4s cadence and 5ms-vs-100ms delta come from the +# original ISSUE-2026-0019 reproduction harness — they are +# calibrated to produce a parent switch per round under the +# zero-hysteresis FIPS overrides below. +link_swap: + enabled: true + interval_secs: 4 + policies: + fast: + delay_ms: [5, 5] + jitter_ms: [0, 0] + loss_pct: [0, 0] + slow: + delay_ms: [100, 100] + jitter_ms: [0, 0] + loss_pct: [0, 0] + # Initial assignment: n02-n04 fast, n03-n04 slow. After each + # interval the assignments are swapped. + edges: + - edge: "n02-n04" + policy: fast + - edge: "n03-n04" + policy: slow + +link_flaps: + enabled: false + +traffic: + enabled: false + +logging: + rust_log: "info" + output_dir: "./sim-results" + +# Per-node assertions evaluated at the end of the run. Each asserts +# on the delta of a control-socket counter over a trailing window. +assertions: + bloom_send_rate: + # The trailing window is the last `window_secs` of the run. + # `max_per_node` is the ceiling on stats.bloom.sent delta over + # that window, per node. The threshold is set ~5x above the + # observed steady-state ceiling on fixed master and well below + # the storm-state value on the regressed binary. + window_secs: 30 + max_per_node: 30 + min_parent_switches: + # Sanity guard: detect a misconfigured harness where the flap + # inducer fires but the topology never produces a real parent + # switch (e.g., wrong root election made n04's parent + # candidates structurally non-equivalent). Without this, the + # bloom-rate assertion would trivially pass on any binary, + # including a regressed one. + # + # Floor calibrated against expected behavior under the chosen + # seed: 41 link swaps over 180s should produce >= 10 parent + # switches at n04 alone if the topology is right. + min_total: 10 + +# FIPS overrides to suppress parent-flap dampening so the asymmetric +# link-cost flap reliably produces a parent switch per round. The +# whole point of the scenario is to *force* sustained parent +# flapping at n04 and assert that bloom layer doesn't amplify it. +fips_overrides: + node: + tree: + parent_hysteresis: 0.0 + reeval_interval_secs: 1 + flap_threshold: 9999 + hold_down_secs: 0 diff --git a/testing/chaos/sim/__main__.py b/testing/chaos/sim/__main__.py index f88bcd8..b8e2684 100644 --- a/testing/chaos/sim/__main__.py +++ b/testing/chaos/sim/__main__.py @@ -54,6 +54,8 @@ def main(): if result and result.panics: sys.exit(2) + if runner.assertions_failed: + sys.exit(3) sys.exit(0) diff --git a/testing/chaos/sim/assertions.py b/testing/chaos/sim/assertions.py new file mode 100644 index 0000000..de05141 --- /dev/null +++ b/testing/chaos/sim/assertions.py @@ -0,0 +1,163 @@ +"""Post-run scenario assertions evaluated via control-socket data. + +Assertions are declared in the scenario YAML under ``assertions:`` and +are evaluated near the end of the simulation, before teardown begins. +Each failing assertion is recorded with a clear pass/fail message; the +runner exits non-zero when any assertion fails. + +Currently supported assertions: + +- ``bloom_send_rate``: per-node trailing-window ceiling on + ``stats.bloom.sent`` delta. Calibrated for the bloom-storm + regression scenario but generally usable. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +from .control import snapshot_all_bloom +from .scenario import BloomSendRateAssertion, MinParentSwitchesAssertion +from .topology import SimTopology + +log = logging.getLogger(__name__) + + +@dataclass +class AssertionOutcome: + name: str + passed: bool + detail: str + + +def _bloom_sent_total(node_data: dict) -> int | None: + """Extract stats.bloom.sent from a show_bloom response.""" + stats = node_data.get("stats") or {} + sent = stats.get("sent") + if sent is None: + return None + try: + return int(sent) + except (TypeError, ValueError): + return None + + +class BloomSendRateMonitor: + """Samples per-node ``stats.bloom.sent`` to evaluate a trailing-window + ceiling assertion at end-of-run. + + Usage: + m = BloomSendRateMonitor(topology, cfg) + m.sample_window_start() # called window_secs before scenario end + ... + m.sample_end() # called at scenario end + outcome = m.evaluate() + """ + + def __init__(self, topology: SimTopology, cfg: BloomSendRateAssertion): + self.topology = topology + self.cfg = cfg + self.window_start: dict[str, int] = {} + self.window_end: dict[str, int] = {} + + def sample_window_start(self) -> None: + snap = snapshot_all_bloom(self.topology) + for nid, data in snap.items(): + v = _bloom_sent_total(data) + if v is not None: + self.window_start[nid] = v + + def sample_end(self) -> None: + snap = snapshot_all_bloom(self.topology) + for nid, data in snap.items(): + v = _bloom_sent_total(data) + if v is not None: + self.window_end[nid] = v + + def evaluate(self) -> AssertionOutcome: + max_per_node = self.cfg.max_per_node + window_secs = self.cfg.window_secs + + if not self.window_start or not self.window_end: + return AssertionOutcome( + name="bloom_send_rate", + passed=False, + detail=( + f"FAIL bloom_send_rate: failed to sample window endpoints " + f"(start={len(self.window_start)} nodes, " + f"end={len(self.window_end)} nodes)" + ), + ) + + per_node_deltas: dict[str, int] = {} + for nid, end_v in self.window_end.items(): + start_v = self.window_start.get(nid) + if start_v is None: + continue + per_node_deltas[nid] = end_v - start_v + + offenders = { + nid: d for nid, d in per_node_deltas.items() if d > max_per_node + } + max_obs = max(per_node_deltas.values()) if per_node_deltas else 0 + + if offenders: + sorted_off = sorted(offenders.items(), key=lambda kv: -kv[1]) + details = ", ".join(f"{nid}={d}" for nid, d in sorted_off) + detail = ( + f"FAIL bloom_send_rate: {len(offenders)} node(s) exceeded " + f"ceiling of {max_per_node} bloom_sent over trailing " + f"{window_secs}s — offenders: {details} " + f"(all per-node deltas: " + f"{', '.join(f'{n}={v}' for n, v in sorted(per_node_deltas.items()))})" + ) + return AssertionOutcome( + name="bloom_send_rate", + passed=False, + detail=detail, + ) + + detail = ( + f"PASS bloom_send_rate: max per-node delta {max_obs} <= " + f"ceiling {max_per_node} over trailing {window_secs}s " + f"(per-node: " + f"{', '.join(f'{n}={v}' for n, v in sorted(per_node_deltas.items()))})" + ) + return AssertionOutcome( + name="bloom_send_rate", + passed=True, + detail=detail, + ) + + +def evaluate_min_parent_switches( + cfg: MinParentSwitchesAssertion, + parent_switch_count: int, +) -> AssertionOutcome: + """Sanity guard: fail the scenario if the harness-induced flap did + not produce at least ``cfg.min_total`` parent switches across the + run. Detects misconfiguration (e.g., wrong root election) where + the bloom-rate assertion would otherwise trivially pass on any + binary including the regressed one. + """ + if parent_switch_count >= cfg.min_total: + return AssertionOutcome( + name="min_parent_switches", + passed=True, + detail=( + f"PASS min_parent_switches: {parent_switch_count} switches " + f">= floor {cfg.min_total}" + ), + ) + return AssertionOutcome( + name="min_parent_switches", + passed=False, + detail=( + f"FAIL min_parent_switches: {parent_switch_count} switches " + f"< floor {cfg.min_total} — harness did not induce sufficient " + f"parent flapping; bloom-rate assertion would be trivially " + f"true. Check tree-snapshot-warmup.json: did the expected " + f"node win the root election?" + ), + ) diff --git a/testing/chaos/sim/control.py b/testing/chaos/sim/control.py index cb2fc01..f2415dc 100644 --- a/testing/chaos/sim/control.py +++ b/testing/chaos/sim/control.py @@ -149,6 +149,27 @@ def snapshot_all_mmp(topology: SimTopology) -> dict[str, dict]: return result +def query_bloom(container: str) -> dict | None: + """Query a node's bloom filter state and stats.""" + return query_node(container, "show_bloom") + + +def snapshot_all_bloom(topology: SimTopology) -> dict[str, dict]: + """Query show_bloom on all nodes, return {node_id: bloom_data}. + + Nodes that fail to respond are omitted from the result. + """ + result = {} + for node_id in sorted(topology.nodes): + container = topology.container_name(node_id) + data = query_bloom(container) + if data is not None: + result[node_id] = data + else: + log.warning("No bloom data from %s", node_id) + return result + + def query_routing(container: str) -> dict | None: """Query a node's routing stats (includes congestion counters).""" return query_node(container, "show_routing") diff --git a/testing/chaos/sim/link_swap.py b/testing/chaos/sim/link_swap.py new file mode 100644 index 0000000..76dfeda --- /dev/null +++ b/testing/chaos/sim/link_swap.py @@ -0,0 +1,151 @@ +"""Deterministic asymmetric link-cost flapping. + +Periodically rotates a fixed assignment of netem policies across a +fixed set of edges. Unlike ``LinkManager`` (random link-down events) +or the ``netem.mutation`` block (random per-edge policy mutation), +this is a deterministic, periodic flip between named policies on +named edges — useful for forcing a downstream node to switch parents +on a fixed cadence. + +Rotation is a simple one-position cyclic shift: the policy assigned +to edges[0] moves to edges[1], edges[1] -> edges[2], ..., edges[N-1] +-> edges[0]. With two edges this degenerates to a swap, which is the +intended use for the bloom-storm scenario. +""" + +from __future__ import annotations + +import logging +import random +import time + +from .netem import NetemManager, NetemParams +from .scenario import LinkSwapConfig, NetemPolicy +from .topology import SimTopology + +log = logging.getLogger(__name__) + + +def _canonical_edge(edge_str: str) -> tuple[str, str]: + """Parse "nXX-nYY" into a canonical (a, b) tuple sorted alphabetically.""" + parts = edge_str.split("-") + if len(parts) != 2: + raise ValueError(f"link_swap edge '{edge_str}' is not 'nXX-nYY' form") + a, b = sorted(parts) + return a, b + + +class LinkSwapManager: + """Manages deterministic netem-policy rotation across a fixed edge set.""" + + def __init__( + self, + topology: SimTopology, + config: LinkSwapConfig, + netem_mgr: NetemManager, + rng: random.Random, + ): + self.topology = topology + self.config = config + self.netem_mgr = netem_mgr + self.rng = rng + + # Resolve edges and validate they exist in the topology + topo_edges = {tuple(sorted([a, b])) for a, b in topology.edges} + self._edges: list[tuple[str, str]] = [] + for entry in config.edges: + edge = _canonical_edge(entry.edge) + if edge not in topo_edges: + raise ValueError( + f"link_swap edge {entry.edge} not present in topology" + ) + self._edges.append(edge) + + # Current per-position policy name. Index i in this list is + # the policy currently applied to self._edges[i]. Each rotate() + # cyclically shifts these by one position. + self._current_policies: list[str] = [e.policy for e in config.edges] + + self.swap_count = 0 + # Last apply timestamp; the runner uses this together with + # config.interval_secs to schedule rotations. + self.last_swap_at: float | None = None + + def policies(self) -> dict[str, NetemPolicy]: + return self.config.policies + + def setup_initial(self) -> None: + """Apply the initial policy assignment to each edge. + + Called once after NetemManager.setup_initial() so the link's + per-direction tc class state already exists. The initial + assignment is taken straight from the config (no rotation). + """ + for edge, policy_name in zip(self._edges, self._current_policies): + policy = self.config.policies[policy_name] + self._apply(edge, policy) + self.last_swap_at = time.time() + log.info( + "Link swap initialized: %d edges, interval %.1fs, policies=%s", + len(self._edges), + self.config.interval_secs, + list(self.config.policies.keys()), + ) + + def maybe_swap(self, now: float) -> bool: + """If the swap interval has elapsed, rotate policies. Return whether + a swap occurred so the runner can reschedule. + """ + if self.last_swap_at is None: + self.last_swap_at = now + return False + if (now - self.last_swap_at) < self.config.interval_secs: + return False + + # Cyclic shift by one position + self._current_policies = ( + [self._current_policies[-1]] + self._current_policies[:-1] + ) + for edge, policy_name in zip(self._edges, self._current_policies): + policy = self.config.policies[policy_name] + self._apply(edge, policy) + self.swap_count += 1 + self.last_swap_at = now + log.debug( + "Link swap #%d: %s", + self.swap_count, + ", ".join( + f"{a}-{b}={p}" + for (a, b), p in zip(self._edges, self._current_policies) + ), + ) + return True + + def _apply(self, edge: tuple[str, str], policy: NetemPolicy) -> None: + """Apply a policy to both directions of an edge using NetemManager.""" + params = self._sample_policy(policy) + # Drive both directions through NetemManager's per-link state so + # the rotation persists across mutate() runs and survives node + # restarts via setup_node(). + self.netem_mgr._update_link(edge[0], edge[1], params) + + def _sample_policy(self, policy: NetemPolicy) -> NetemParams: + """Sample concrete params from a policy's ranges (deterministic + when min == max, which is the bloom-storm scenario's contract). + """ + return NetemParams( + delay_ms=int(self.rng.uniform(policy.delay_ms[0], policy.delay_ms[1])), + jitter_ms=int(self.rng.uniform(policy.jitter_ms[0], policy.jitter_ms[1])), + loss_pct=round( + self.rng.uniform(policy.loss_pct[0], policy.loss_pct[1]), 1 + ), + duplicate_pct=round( + self.rng.uniform(policy.duplicate_pct[0], policy.duplicate_pct[1]), 1 + ), + reorder_pct=round( + self.rng.uniform(policy.reorder_pct[0], policy.reorder_pct[1]), 1 + ), + corrupt_pct=round( + self.rng.uniform(policy.corrupt_pct[0], policy.corrupt_pct[1]), 1 + ), + ) diff --git a/testing/chaos/sim/runner.py b/testing/chaos/sim/runner.py index a259f9d..dc4960b 100644 --- a/testing/chaos/sim/runner.py +++ b/testing/chaos/sim/runner.py @@ -12,10 +12,12 @@ import sys import time from datetime import datetime +from .assertions import AssertionOutcome, BloomSendRateMonitor, evaluate_min_parent_switches from .compose import generate_compose from .config_gen import write_configs from .control import snapshot_all_congestion, snapshot_all_mmp, snapshot_all_trees from .docker_exec import docker_compose +from .link_swap import LinkSwapManager from .links import LinkManager from .logs import AnalysisResult, analyze_logs, collect_logs, write_sim_metadata from .netem import NetemManager @@ -46,10 +48,15 @@ class SimRunner: self.veth_mgr: VethManager | None = None self.netem_mgr: NetemManager | None = None self.link_mgr: LinkManager | None = None + self.link_swap_mgr: LinkSwapManager | None = None self.traffic_mgr: TrafficManager | None = None self.node_mgr: NodeManager | None = None self.peer_churn_mgr: PeerChurnManager | None = None + # Post-run assertion monitors (sampled near end of run). + self.bloom_rate_monitor: BloomSendRateMonitor | None = None + self.assertion_outcomes: list[AssertionOutcome] = [] + @staticmethod def _resolve_output_dir(scenario: Scenario) -> str: """Build a timestamped output directory path. @@ -190,6 +197,20 @@ class SimRunner: self.topology, s.link_flaps, self.rng, netem_mgr=self.netem_mgr ) + if s.link_swap.enabled: + if not self.netem_mgr: + raise RuntimeError( + "link_swap requires netem.enabled (depends on per-link tc state)" + ) + self.link_swap_mgr = LinkSwapManager( + self.topology, s.link_swap, self.netem_mgr, self.rng, + ) + + if s.assertions.bloom_send_rate is not None: + self.bloom_rate_monitor = BloomSendRateMonitor( + self.topology, s.assertions.bloom_send_rate, + ) + if s.traffic.enabled: self.traffic_mgr = TrafficManager( self.topology, s.traffic, self.rng, down_nodes=self._down_nodes @@ -226,6 +247,12 @@ class SimRunner: if self.peer_churn_mgr: self.peer_churn_mgr.refresh_all_npubs() + # Initial link swap policy assignment (after warmup so the netem + # tc state is fully set up and the daemons have already + # discovered each other under the calm baseline). + if self.link_swap_mgr: + self.link_swap_mgr.setup_initial() + def _handle_node_restart(self, node_id: str): """Called after a node container is restarted. @@ -260,12 +287,38 @@ class SimRunner: next_churn = self._schedule_next(start, s.node_churn.interval_secs) if self.node_mgr else float("inf") next_peer_churn = self._schedule_next(start, s.peer_churn.interval_secs) if self.peer_churn_mgr else float("inf") + # Bloom-send-rate assertion: sample at window_secs before end. + bloom_window_start_at = float("inf") + if self.bloom_rate_monitor is not None: + bloom_window_start_at = ( + start + duration - s.assertions.bloom_send_rate.window_secs + ) + bloom_window_started = False + while not self._interrupted: now = time.time() elapsed = now - start if elapsed >= duration: break + # Bloom-rate window-start sampling + if ( + self.bloom_rate_monitor is not None + and not bloom_window_started + and now >= bloom_window_start_at + ): + log.info( + "Sampling bloom-rate window start (last %ds of run)...", + s.assertions.bloom_send_rate.window_secs, + ) + self.bloom_rate_monitor.sample_window_start() + bloom_window_started = True + + # Deterministic link swap (before netem mutation so a + # mutation round can't clobber the swap mid-tick). + if self.link_swap_mgr: + self.link_swap_mgr.maybe_swap(now) + # Netem mutation if self.netem_mgr and now >= next_netem: self.netem_mgr.mutate() @@ -304,6 +357,8 @@ class SimRunner: active = self.traffic_mgr.active_count if self.traffic_mgr else 0 peer_churns = self.peer_churn_mgr.churn_count if self.peer_churn_mgr else 0 status_extra = f" peer_churns={peer_churns}" if self.peer_churn_mgr else "" + if self.link_swap_mgr: + status_extra += f" swaps={self.link_swap_mgr.swap_count}" print( f"\r [{elapsed:.0f}s/{duration}s] " f"nodes={len(self.topology.nodes)} " @@ -320,11 +375,49 @@ class SimRunner: print() # Clear status line + # Bloom-rate window-end sampling. + # Done before teardown so containers are still running. + if self.bloom_rate_monitor is not None: + if not bloom_window_started: + # Loop exited before the window-start mark (e.g., + # interrupted). Take both samples now so we still + # produce a finite outcome rather than an empty dict. + log.warning( + "Bloom-rate window did not start during run; " + "sampling both endpoints at end (delta will be 0)." + ) + self.bloom_rate_monitor.sample_window_start() + log.info("Sampling bloom-rate window end...") + self.bloom_rate_monitor.sample_end() + + def _evaluate_assertions(self) -> None: + """Evaluate post-run assertions and stash outcomes on self. + + Called from teardown while containers are still running so the + assertion outcomes (which include per-node detail) can be + written alongside the run artifacts. + """ + if self.bloom_rate_monitor is not None: + outcome = self.bloom_rate_monitor.evaluate() + self.assertion_outcomes.append(outcome) + if outcome.passed: + log.info("%s", outcome.detail) + else: + log.error("%s", outcome.detail) + + @property + def assertions_failed(self) -> bool: + return any(not o.passed for o in self.assertion_outcomes) + def _teardown(self) -> AnalysisResult | None: """Stop dynamic elements, collect logs, analyze, stop containers.""" result = None if self.topology and self.compose_file: + # Evaluate post-run assertions before doing any teardown so + # control sockets are still reachable. + self._evaluate_assertions() + # Stop traffic if self.traffic_mgr: log.info("Stopping traffic sessions...") @@ -366,6 +459,30 @@ class SimRunner: f.write(result.summary()) print(result.summary()) + # Log-derived assertions (evaluated after analyze_logs so + # parent_switches and similar are populated). + mps_cfg = self.scenario.assertions.min_parent_switches + if mps_cfg is not None: + outcome = evaluate_min_parent_switches( + mps_cfg, len(result.parent_switches) + ) + 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") + with open(assertions_path, "w") as f: + for o in self.assertion_outcomes: + f.write(o.detail + "\n") + print("=== Assertions ===") + for o in self.assertion_outcomes: + print(o.detail) + print() + # Write metadata write_sim_metadata( self.output_dir, diff --git a/testing/chaos/sim/scenario.py b/testing/chaos/sim/scenario.py index 6374ac5..79a7228 100644 --- a/testing/chaos/sim/scenario.py +++ b/testing/chaos/sim/scenario.py @@ -150,6 +150,67 @@ class IngressConfig: burst_bytes: int = 32000 +@dataclass +class LinkSwapEdge: + """One edge in a link-swap rotation. + + Edge is a canonical "nXX-nYY" string. ``policy`` names a policy + in ``LinkSwapConfig.policies``. + """ + + edge: str = "" + policy: str = "" + + +@dataclass +class LinkSwapConfig: + """Deterministic asymmetric link-cost flapping. + + On each ``interval_secs``, the policies on every pair of edges in + ``edges`` are swapped (cyclically rotated by one position). This + differs from ``link_flaps`` (random link-down events) and + ``netem.mutation`` (random per-edge policy mutation) by being a + deterministic, periodic flip between two named netem policies on + a fixed set of edges. + + Used to drive a downstream node to repeatedly switch parents on + a fixed cadence — exercises the spanning-tree rebalance path + without the noise of a random mutation walk. + """ + + enabled: bool = False + interval_secs: float = 4.0 + policies: dict[str, NetemPolicy] = field(default_factory=dict) + edges: list[LinkSwapEdge] = field(default_factory=list) + + +@dataclass +class BloomSendRateAssertion: + """Trailing-window ceiling on per-node ``stats.bloom.sent`` delta.""" + + window_secs: int = 30 + max_per_node: int = 30 + + +@dataclass +class MinParentSwitchesAssertion: + """Sanity guard: total parent switches across the run must be at + least ``min_total``. Used to detect a misconfigured harness where + the flap inducer is firing but the topology never produces a real + parent-switch event (e.g., wrong root election). + """ + + min_total: int = 1 + + +@dataclass +class AssertionsConfig: + """Optional post-run assertions evaluated against control-socket data.""" + + bloom_send_rate: BloomSendRateAssertion | None = None + min_parent_switches: MinParentSwitchesAssertion | None = None + + @dataclass class LoggingConfig: rust_log: str = "info" @@ -169,6 +230,8 @@ class Scenario: peer_churn: PeerChurnConfig = field(default_factory=PeerChurnConfig) bandwidth: BandwidthConfig = field(default_factory=BandwidthConfig) ingress: IngressConfig = field(default_factory=IngressConfig) + link_swap: LinkSwapConfig = field(default_factory=LinkSwapConfig) + assertions: AssertionsConfig = field(default_factory=AssertionsConfig) logging: LoggingConfig = field(default_factory=LoggingConfig) # Raw YAML dict appended to each generated FIPS node config. # Allows scenarios to override any FIPS config parameter @@ -320,6 +383,40 @@ def load_scenario(path: str) -> Scenario: s.ingress.tiers_kbps = [int(t) for t in tiers] s.ingress.burst_bytes = int(ig.get("burst_bytes", 32000)) + # Link swap section (deterministic asymmetric link-cost flapping). + ls = raw.get("link_swap", {}) + s.link_swap.enabled = ls.get("enabled", False) + if "interval_secs" in ls: + s.link_swap.interval_secs = float(ls["interval_secs"]) + if "policies" in ls: + s.link_swap.policies = { + name: _parse_netem_policy(pdata) + for name, pdata in ls["policies"].items() + } + if "edges" in ls: + for edata in ls["edges"]: + if not isinstance(edata, dict): + raise ValueError("link_swap.edges entries must be dicts") + edge = str(edata.get("edge", "")) + policy = str(edata.get("policy", "")) + if not edge or not policy: + raise ValueError("link_swap.edges entries require 'edge' and 'policy'") + s.link_swap.edges.append(LinkSwapEdge(edge=edge, policy=policy)) + + # Assertions section (post-run control-socket-based checks). + asrt = raw.get("assertions", {}) + if "bloom_send_rate" in asrt: + bsr = asrt["bloom_send_rate"] + s.assertions.bloom_send_rate = BloomSendRateAssertion( + window_secs=int(bsr.get("window_secs", 30)), + max_per_node=int(bsr.get("max_per_node", 30)), + ) + if "min_parent_switches" in asrt: + mps = asrt["min_parent_switches"] + s.assertions.min_parent_switches = MinParentSwitchesAssertion( + min_total=int(mps.get("min_total", 1)), + ) + # Logging section lg = raw.get("logging", {}) s.logging.rust_log = lg.get("rust_log", "info") @@ -435,3 +532,29 @@ def _validate(s: Scenario): raise ValueError(f"ingress.tiers_kbps: all values must be > 0, got {tier}") if s.ingress.burst_bytes <= 0: raise ValueError(f"ingress.burst_bytes must be > 0, got {s.ingress.burst_bytes}") + + # Validate link_swap + if s.link_swap.enabled: + if s.link_swap.interval_secs <= 0: + raise ValueError("link_swap.interval_secs must be > 0") + if len(s.link_swap.edges) < 2: + raise ValueError("link_swap.edges must list at least 2 edges to swap") + if not s.link_swap.policies: + raise ValueError("link_swap.policies must not be empty when link_swap.enabled") + for entry in s.link_swap.edges: + if entry.policy not in s.link_swap.policies: + raise ValueError( + f"link_swap.edges: policy '{entry.policy}' not in link_swap.policies" + ) + + # Validate assertions + if s.assertions.bloom_send_rate is not None: + bsr = s.assertions.bloom_send_rate + if bsr.window_secs < 1: + raise ValueError("assertions.bloom_send_rate.window_secs must be >= 1") + if bsr.max_per_node < 0: + raise ValueError("assertions.bloom_send_rate.max_per_node must be >= 0") + if bsr.window_secs > s.duration_secs: + raise ValueError( + "assertions.bloom_send_rate.window_secs must not exceed scenario duration" + ) diff --git a/testing/ci-local.sh b/testing/ci-local.sh index 9afb0ef..34661e5 100755 --- a/testing/ci-local.sh +++ b/testing/ci-local.sh @@ -23,6 +23,7 @@ # chaos-ethernet-only, chaos-tcp-mesh, chaos-bottleneck-parent, # chaos-cost-avoidance, chaos-cost-reeval, chaos-cost-stability, # chaos-depth-vs-cost, chaos-mixed-technology, chaos-congestion-stress, +# chaos-bloom-storm, # sidecar, dns-resolver, deb-install # # Opt-in (require --with-tor; depend on live Tor network): @@ -70,6 +71,7 @@ CHAOS_SUITES=( "depth-vs-cost depth-vs-cost" "mixed-technology mixed-technology" "congestion-stress congestion-stress" + "bloom-storm bloom-storm" ) GATEWAY_SUITES=(gateway) SIDECAR_SUITES=(sidecar)