From 46ac9a03b9d1f0e12b36b104e0e357f3a72a7b0a Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 17:11:43 +0200 Subject: [PATCH] Add forced TCP bench mode and verify observed source timestamp wraps --- tests/stream_netns/run.py | 11 ++++++++++- tests/stream_netns/scenario.py | 28 +++++++++++++++++++++++----- 2 files changed, 33 insertions(+), 6 deletions(-) diff --git a/tests/stream_netns/run.py b/tests/stream_netns/run.py index 1b1f039b..e9a5888d 100644 --- a/tests/stream_netns/run.py +++ b/tests/stream_netns/run.py @@ -91,7 +91,9 @@ def main(): parser.add_argument("--duration", type=int, default=120, help="Baseline duration in real seconds") parser.add_argument("--faults", action="store_true", help="Degrade WAN, recover and check unsubscribe") parser.add_argument("--fault-duration", type=int, default=30) - parser.add_argument("--tcp", action="store_true", help="Also expose TCP sockets, permitting ETCP multilink") + transport = parser.add_mutually_exclusive_group() + transport.add_argument("--tcp", action="store_true", help="Also expose TCP sockets, permitting ETCP multilink") + transport.add_argument("--tcp-only", action="store_true", help="Expose only TCP, requiring the real STCP transport") parser.add_argument("--inside", action="store_true", help=argparse.SUPPRESS) parser.add_argument("--host-netns", help=argparse.SUPPRESS) args = parser.parse_args() @@ -120,6 +122,8 @@ def main(): command.append("--faults") if args.tcp: command.append("--tcp") + if args.tcp_only: + command.append("--tcp-only") print(f"Artifacts: {directory}", flush=True) child = subprocess.Popen(command) try: @@ -135,6 +139,11 @@ def main(): (directory / "host-after.json").write_text(json.dumps(after, indent=2) + "\n") unchanged = stable_snapshot(before) == stable_snapshot(after) and os.readlink("/proc/self/ns/net") == original_namespace (directory / "host-check.json").write_text(json.dumps({"unchanged": unchanged}, indent=2) + "\n") + result_path = directory / "result.json" + result = json.loads(result_path.read_text()) if result_path.exists() else {"pass": False, "error": "Isolated runner did not complete"} + result.update(host_unchanged=unchanged, launcher_exit=status) + result["pass"] = result["pass"] and unchanged and status == 0 + result_path.write_text(json.dumps(result, indent=2) + "\n") if not unchanged: print("FAIL: main network namespace snapshot changed during the run", file=sys.stderr) return 1 diff --git a/tests/stream_netns/scenario.py b/tests/stream_netns/scenario.py index bca2f4a8..c97a5a2b 100644 --- a/tests/stream_netns/scenario.py +++ b/tests/stream_netns/scenario.py @@ -11,7 +11,7 @@ from topology import NODES, WANS class Agent: - def __init__(self, network, node, binary, seed, tcp): + def __init__(self, network, node, binary, seed, tcp, tcp_only): self.node = node directory = network.directory / node.name directory.mkdir() @@ -20,7 +20,7 @@ class Agent: self.identity = identity["node"] kind = "nat" if node.nat else "public" sockets = "".join(f"[server:{transport}]\naddr={node.ip}:32000\ntype={kind}\ntransport={transport}\n" - for transport in (["udp", "tcp"] if tcp else ["udp"])) + for transport in (["tcp"] if tcp_only else ["udp", "tcp"] if tcp else ["udp"])) config = (f"[global]\nmy_node_name={node.name}\nmy_private_key={identity['private']}\n" f"my_public_key={identity['public']}\nclient_type={node.client_type}\nauto_sockets=no\n" f"db_path={directory / 'db'}\ndebug_level=warn\n" @@ -81,6 +81,9 @@ class Scenario: self.started = time.monotonic() self.phases = [] self.latest = {} + self.last_metadata = {} + self.clock_previous = {} + self.clock_wraps = 0 def sample(self, phase): snapshot = {name: agent.rpc("stats") for name, agent in self.agents.items()} @@ -91,6 +94,16 @@ class Scenario: raise RuntimeError(f"{name}: payload/order validation failed; see samples.jsonl and utun.log") if any(s.get("stored_bytes", 0) > 32 * 1024 * 1024 for s in statistics["streams"]): raise RuntimeError(f"{name}: stream data buffer exceeded 32 MiB") + for stream in statistics["streams"]: + self.last_metadata.setdefault((name, stream["source"]), {}).update(stream) + path = next((p for p in stream.get("paths", []) if p["sub"] == 0), {}) + if stream["publisher"] and "source_timestamp_ms" in path: + key = (name, stream["source"]) + current = (path["seq"], path["source_timestamp_ms"]) + previous = self.clock_previous.get(key) + if previous and current[0] > previous[0] and current[1] < previous[1]: + self.clock_wraps += 1 + self.clock_previous[key] = current return snapshot def wait(self, phase, predicate, timeout): @@ -121,7 +134,7 @@ class Scenario: def run(self): binary = Path(self.args.binary).resolve() for node in NODES: - self.agents[node.name] = Agent(self.network, node, binary, self.args.seed, self.args.tcp) + self.agents[node.name] = Agent(self.network, node, binary, self.args.seed, self.args.tcp, self.args.tcp_only) owner = self.agents["n01"] group = owner.rpc("create")["group"] print(f"Joining real CHAT group {group}", flush=True) @@ -181,6 +194,8 @@ class Scenario: generated = self.stream(final[self.args.publisher], source_id)["generated"] if generated < 10: raise RuntimeError("Publisher generated insufficient data") + if self.args.duration >= 66 and not self.clock_wraps: + raise RuntimeError("Full-duration run did not observe a source timestamp wrap") for name in names: stream = self.stream(final[name], source_id) if stream["next_seq"] != generated: @@ -190,17 +205,20 @@ class Scenario: f"start={stream['start_seq']}, generated={generated}") self.report(final, source_id) return {"pass": True, "group": group, "publisher": self.args.publisher, "generated": generated, - "phases": self.phases, "subscribers_checked": len(names)} + "phases": self.phases, "subscribers_checked": len(names), "source_timestamp_wraps": self.clock_wraps, + "transport": "tcp" if self.args.tcp_only else "udp+tcp" if self.args.tcp else "udp", + "rate_bps": self.args.rate} def report(self, final, source): rows = [] for node in NODES: stream = self.stream(final[node.name], source) + metadata = self.last_metadata[(node.name, source)] rows.append(dict(node=node.name, country=node.country, nat=node.nat or "public", received=stream["received"], gap_batches=stream["gap_batches"], errors=stream["errors"], p50_ms=stream["latency_p50_ms"], p95_ms=stream["latency_p95_ms"], p99_ms=stream["latency_p99_ms"], maximum_ms=stream["latency_max_ms"], - migrations=stream.get("migrations", 0), probes=stream.get("probes", 0))) + migrations=metadata.get("migrations", 0), probes=metadata.get("probes", 0))) with (self.network.directory / "summary.csv").open("w", newline="") as output: writer = csv.DictWriter(output, fieldnames=list(rows[0])) writer.writeheader(); writer.writerows(rows)