#!/usr/bin/env python3 """Fixed-rate mixed traffic, controlled client closes and cache hot-toggle soak. The default is a 30-minute live run. Short invocations must be explicitly marked --exploratory and cannot satisfy the release soak gate. No run is auto-scheduled. """ from __future__ import annotations import argparse from collections import Counter from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED import copy import hashlib from http.client import HTTPConnection import json import os from pathlib import Path import platform import statistics import sys import time import traceback from benchmark import counter_values from common import Fixture, http, seal_results, sha256, write_json from resources import Monitor from telemetry import traffic_analytics_snapshot, traffic_analytics_delta PAYLOADS = ["small.html", "large.bin", "huge.bin"] def fetch(index, rust, expected, abort_every): name = PAYLOADS[index % len(PAYLOADS)] source = "proxy" if (index // len(PAYLOADS)) % 2 else "static" port = rust.reverse_port if source == "proxy" else rust.static_port started = time.monotonic() case = {"index": index, "source": source, "file": name, "method": "GET", "intentional_abort": False} try: if abort_every and index % abort_every == 0: case.update({"file": "huge.bin", "intentional_abort": True}) connection = HTTPConnection("127.0.0.1", port, timeout=10) try: connection.request("GET", "/huge.bin", headers={"Accept-Encoding": "identity"}) response = connection.getresponse() body = response.read(128) case.update({"status": response.status, "bytes": len(body), "cache": response.getheader("X-Cache"), "passed": response.status == 200 and body == expected["huge.bin"][:128]}) finally: connection.close() else: headers = {"Accept-Encoding": "identity"} method = "HEAD" if index % 29 == 0 else "GET" if method == "GET" and index % 37 == 0: headers["Range"] = "bytes=0-127" expected_status, expected_body = 206, expected[name][:128] else: expected_status, expected_body = 200, b"" if method == "HEAD" else expected[name] status, response_headers, body = http(port, "/" + name, method, headers) case.update({"method": method, "status": status, "bytes": len(body), "cache": response_headers.get("x-cache"), "passed": status == expected_status and body == expected_body}) if not case["passed"]: case.update({"expected_status": expected_status, "expected_bytes": len(expected_body), "body_sha256": hashlib.sha256(body).hexdigest()}) except Exception as exc: case.update({"passed": False, "error": str(exc)}) case["seconds"] = time.monotonic() - started return case def probe_cache_mode(rust, expected, enabled, phase, out, observations): """Record two exact GETs per source, separate from the paced workload.""" phase_observations = [] for source, port in [("proxy", rust.reverse_port), ("static", rust.static_port)]: replies = [] for attempt in [1, 2]: try: status, headers, body = http(port, "/small.html?soak_cache_probe=1", headers={"Accept-Encoding": "identity"}) replies.append({"attempt": attempt, "status": status, "cache": headers.get("x-cache"), "bytes": len(body), "sha256": hashlib.sha256(body).hexdigest(), "body_passed": status == 200 and body == expected["small.html"]}) except Exception as error: replies.append({"attempt": attempt, "body_passed": False, "error": str(error)}) mode_passed = replies[1].get("cache") == "HIT" if enabled else all( reply.get("cache") is None for reply in replies) phase_observations.append({"phase": phase, "source": source, "cache_enabled": enabled, "responses": replies, "passed": all(reply["body_passed"] for reply in replies) and mode_passed}) observations.extend(phase_observations) write_json(out / "cache-probes.json", observations) if not all(observation["passed"] for observation in phase_observations): raise RuntimeError(f"cache mode/body probe failed after {phase}; see cache-probes.json") def backend_drain_snapshot(rust): status, _, body = rust.admin("/api/status") if status != 200: return {"passed": False, "error": f"admin status HTTP {status}"} backends = [] def walk(value): if isinstance(value, dict): if "id" in value and "active" in value and "status" in value and "requests" in value: backends.append({key: value[key] for key in ("id", "active", "status", "requests")}) for child in value.values(): walk(child) elif isinstance(value, list): for child in value: walk(child) walk(json.loads(body)) return {"passed": bool(backends) and all(backend["active"] == 0 for backend in backends), "backends": backends} def memory_trend(samples, label="server"): observations = [(sample["seconds"], sample["groups"][label]["rss_bytes"], sample["groups"][label]["fd"]) for sample in samples if sample["seconds"] >= 60] if len(observations) < 3: return {"available": False, "reason": "less than three samples after the first minute"} tail_seconds = observations[-1][0] early = [row[1] for row in observations if row[0] < observations[0][0] + 60] late = [row[1] for row in observations if row[0] >= tail_seconds - 60] x_mean = statistics.mean(row[0] for row in observations) y_mean = statistics.mean(row[1] for row in observations) denominator = sum((row[0] - x_mean) ** 2 for row in observations) slope = sum((row[0] - x_mean) * (row[1] - y_mean) for row in observations) / denominator if denominator else 0 return {"available": True, "rss_first_minute_after_warmup_median": statistics.median(early), "rss_last_minute_median": statistics.median(late), "rss_growth_bytes": statistics.median(late) - statistics.median(early), "rss_linear_bytes_per_hour": slope * 3600, "fd_first": observations[0][2], "fd_last": observations[-1][2], "fd_peak": max(row[2] for row in observations), "interpretation": "Descriptive only: cache toggles and allocator retention can change RSS without a leak"} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--binary", type=Path, required=True) parser.add_argument("--label", default="after") parser.add_argument("--nginx", type=Path, required=True) parser.add_argument("--out", type=Path, required=True) parser.add_argument("--duration", type=int, default=1800) parser.add_argument("--rate", type=float, default=100, help="offered requests per second, not a saturation benchmark") parser.add_argument("--clients", type=int, default=8) parser.add_argument("--workers", type=int, default=4) parser.add_argument("--origin-workers", type=int, default=2) parser.add_argument("--server-cpus") parser.add_argument("--origin-cpus") parser.add_argument("--toggle-seconds", type=int, default=120) parser.add_argument("--sample-seconds", type=int, default=5) parser.add_argument("--abort-every", type=int, default=101) parser.add_argument("--cooldown-seconds", type=int, default=15) parser.add_argument("--exploratory", action="store_true") args = parser.parse_args() if args.duration < 1800 and not args.exploratory: parser.error("the soak gate requires >=1800 seconds; label short diagnostics with --exploratory") if args.rate <= 0 or not 1 <= args.clients <= 128 or args.toggle_seconds < 1 or args.sample_seconds < 1: parser.error("rate, sample/toggle intervals and clients must be positive; clients <=128") args.out = args.out.resolve() if args.out.exists() and any(args.out.iterdir()): parser.error("--out must be a new or empty evidence directory") args.out.mkdir(parents=True, exist_ok=True) settings = {key: str(value) if isinstance(value, Path) else value for key, value in vars(args).items()} write_json(args.out / "environment.json", { "kind": "actual RustProxy fixed-rate mixed-traffic resource soak", "simulation": False, "settings": settings, "platform": platform.platform(), "binary_sha256": sha256(args.binary), "started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "frontend_connection_policy": "A new HTTP/1.1 TCP connection for each workload request and probe", "controlled_close_interpretation": "Client closes after reading 128 bytes; the server may already have sent the full body. No server-cancellation count is inferred.", "traffic_analytics_observation_policy": "Global counters are read before traffic, periodically with status diagnostics, and after final drain. Deltas include cache-mode probes and cooldown, and are separate from HTTP errors.", "not_covered": ["24-hour soak", "TLS/H2/H3 soak", "multi-backend failures", "PHP-FPM", "real application write load", "origin DNS/TLS rotation", "production traffic", "long-lived frontend keepalive connections"]}) totals = {"submitted": 0, "completed": 0, "passed": 0, "failed": 0, "controlled_client_closes": 0, "response_bytes": 0, "scheduler_late_events": 0, "peak_pending": 0} failure_examples, progress, toggles, counter_samples = [], [], [], [] traffic_samples, traffic_delta = [], None cache_probes, cache_status_counts = [], {"proxy": Counter(), "static": Counter()} drain_snapshot, configuration_restored = None, False samples, resources = [], {} fatal, runtime_seconds = None, 0 try: with Fixture(args.out / "fixture", workers=args.workers, server_cpus=args.server_cpus, origin_cpus=args.origin_cpus, origin_workers=args.origin_workers) as fixture: nginx = fixture.nginx(args.nginx, frontend=False) rust = fixture.rust(args.binary, args.label) rust.login() rust.wait_quiescent() expected = {name: (fixture.site / name).read_bytes() for name in PAYLOADS} config = copy.deepcopy(rust.base) config["cache"].update({"expose_status": True}) config["web_services"][0]["cache_control"] = "public, max-age=60" rust.apply(config, "soak-initial") probe_cache_mode(rust, expected, False, "initial-off", args.out, cache_probes) traffic_before = traffic_analytics_snapshot(rust) traffic_samples.append({"phase": "before", "seconds": 0, "snapshot": traffic_before}) write_json(args.out / "traffic-analytics.json", traffic_samples) monitor = Monitor({"server": rust.process.pid, "origin": nginx["origin"].pid}, args.sample_seconds) monitor.start() start = time.monotonic() deadline = start + args.duration next_toggle, next_snapshot, next_submit = start + args.toggle_seconds, start, start pending = set() def consume(done): for future in done: result = future.result() totals["completed"] += 1 totals["passed" if result["passed"] else "failed"] += 1 totals["controlled_client_closes"] += int(result["intentional_abort"]) totals["response_bytes"] += result.get("bytes", 0) cache_status = (result.get("cache") or "absent") if "status" in result else "no_response" cache_status_counts[result["source"]][cache_status] += 1 if not result["passed"] and len(failure_examples) < 100: failure_examples.append(result) try: with ThreadPoolExecutor(max_workers=args.clients) as pool: while time.monotonic() < deadline: now = time.monotonic() done = {future for future in pending if future.done()} consume(done) pending -= done if now >= next_toggle: config["cache"]["enabled"] = not config["cache"]["enabled"] rust.apply(config, f"soak-toggle-{len(toggles) + 1}") toggles.append({"seconds": time.monotonic() - start, "cache_enabled": config["cache"]["enabled"]}) write_json(args.out / "cache-toggles.json", toggles) probe_cache_mode(rust, expected, config["cache"]["enabled"], f"toggle-{len(toggles)}", args.out, cache_probes) next_toggle += args.toggle_seconds if now >= next_snapshot: status, headers, body = rust.admin("/api/status") if status != 200: raise RuntimeError(f"admin status became unavailable: HTTP {status}") counter_samples.append({"seconds": time.monotonic() - start, "cache_enabled": config["cache"]["enabled"], "counters": counter_values(json.loads(body))}) traffic_now = traffic_analytics_snapshot(rust) traffic_delta = traffic_analytics_delta(traffic_before, traffic_now) traffic_samples.append({"phase": "during", "seconds": time.monotonic() - start, "snapshot": traffic_now, "delta_from_start": traffic_delta}) row = {"seconds": time.monotonic() - start, **totals, "cache_enabled": config["cache"]["enabled"], "pending": len(pending), "observed_cache_status_counts": {source: dict(counts) for source, counts in cache_status_counts.items()}} progress.append(row) write_json(args.out / "progress.json", progress) write_json(args.out / "status-counters.json", counter_samples) write_json(args.out / "traffic-analytics.json", traffic_samples) write_json(args.out / "resources.json", list(monitor.samples)) print(json.dumps(row), flush=True) next_snapshot = time.monotonic() + args.sample_seconds now = time.monotonic() if len(pending) >= args.clients * 2: done, pending = wait(pending, timeout=.1, return_when=FIRST_COMPLETED) consume(done) continue if now < next_submit: time.sleep(min(.02, next_submit - now)) continue if now > next_submit + .25: totals["scheduler_late_events"] += 1 next_submit = now totals["submitted"] += 1 pending.add(pool.submit(fetch, totals["submitted"], rust, expected, args.abort_every)) totals["peak_pending"] = max(totals["peak_pending"], len(pending)) next_submit += 1 / args.rate done, pending = wait(pending, timeout=30) consume(done) if pending: raise RuntimeError("requests failed to settle during bounded soak drain") runtime_seconds = time.monotonic() - start rust.reset() configuration_restored = True cooldown_until = time.monotonic() + args.cooldown_seconds while time.monotonic() < cooldown_until: time.sleep(max(0, min(.5, cooldown_until - time.monotonic()))) probe_cache_mode(rust, expected, False, "restored-off", args.out, cache_probes) for _ in range(40): drain_snapshot = backend_drain_snapshot(rust) if drain_snapshot["passed"]: break time.sleep(.05) write_json(args.out / "backend-drain.json", drain_snapshot) if not drain_snapshot["passed"]: raise RuntimeError("backend active exchanges did not drain to zero") traffic_after = traffic_analytics_snapshot(rust) traffic_delta = traffic_analytics_delta(traffic_before, traffic_after) traffic_samples.append({"phase": "after_drain", "seconds": time.monotonic() - start, "snapshot": traffic_after, "delta_from_start": traffic_delta}) finally: monitor.stop() samples = monitor.samples resources = monitor.summary(totals["completed"]) write_json(args.out / "resources.json", samples) write_json(args.out / "status-counters.json", counter_samples) write_json(args.out / "traffic-analytics.json", traffic_samples) write_json(args.out / "failure-examples.json", failure_examples) except Exception as exc: fatal = {"error": str(exc), "traceback": traceback.format_exc()} write_json(args.out / "failure.json", fatal) report = {"completed": fatal is None, "simulation": False, "exploratory": args.exploratory, "duration_seconds": runtime_seconds, "soak_duration_gate": runtime_seconds >= 1800, "totals": totals, "resources": resources, "memory_trend": memory_trend(samples) if samples else None, "cache_toggles": len(toggles), "errors": failure_examples, "traffic_analytics": {"observations": len(traffic_samples), "delta": traffic_delta, "includes_cache_mode_probes_and_cooldown": True}, "cache_mode_probes": {"source_checks": len(cache_probes), "requests": sum(len(probe["responses"]) for probe in cache_probes), "passed": bool(cache_probes) and all(probe["passed"] for probe in cache_probes)}, "observed_cache_status_counts": cache_status_counts, "configuration_restored": configuration_restored, "backend_drain": drain_snapshot, "actual_completed_rps": totals["completed"] / runtime_seconds if runtime_seconds else None, "passed": fatal is None and totals["failed"] == 0 and runtime_seconds >= args.duration} write_json(args.out / "report.json", report) seal_results(args.out) print(json.dumps(report), flush=True) return 0 if report["passed"] else 1 if __name__ == "__main__": sys.exit(main())