diff --git a/tests/unit/test_soak_corpus.py b/tests/unit/test_soak_corpus.py index d95ce33..896a0bb 100644 --- a/tests/unit/test_soak_corpus.py +++ b/tests/unit/test_soak_corpus.py @@ -145,13 +145,24 @@ from tools.soak.scoreboard import ( # noqa: E402 ) -def _row(call_id, *, source="s1", created_at="2026-07-20T10:00:00", error=None, session="r-p3"): +def _row( + call_id, + *, + source="s1", + created_at="2026-07-20T10:00:00", + error=None, + session="r-p3", + latency_ms=0, + cache_hit=0, +): return { "call_id": call_id, "source_name": source, "created_at": created_at, "error": error, "session_id": session, + "latency_ms": latency_ms, + "cache_hit": cache_hit, } @@ -169,11 +180,11 @@ class TestInvariants: def test_rpm_minute_bucket(self): # 固定分钟窗口口径(与限流器 int(sec/60) 同源;滑动窗会对合法的 - # 跨窗背靠背流量误报,不采用) + # 跨窗背靠背流量误报,不采用)。击穿用例取窗口中部,避免贴边歧义。 rows = [ - _row("a", created_at="2026-07-20T10:00:01"), - _row("b", created_at="2026-07-20T10:00:59"), - _row("c", created_at="2026-07-20T10:01:01"), + _row("a", created_at="2026-07-20T10:00:10"), + _row("b", created_at="2026-07-20T10:00:50"), + _row("c", created_at="2026-07-20T10:01:10"), ] inv_rpm_never_exceeded(rows, {"s1": 2}) rows.append(_row("d", created_at="2026-07-20T10:00:30")) @@ -183,6 +194,49 @@ class TestInvariants: def test_rpm_ignores_unlimited_sources(self): inv_rpm_never_exceeded([_row(str(i)) for i in range(100)], {"s1": 0}) + def test_rpm_buckets_by_admit_time_not_completion(self): + # 准入相隔 100s(各窗 1 次),长延迟让完成时刻挤进同一分钟—— + # 完成时刻聚桶会误报(P6 伪击穿教训之一) + rows = [ + _row("a", created_at="2026-07-20T10:02:30", latency_ms=120_000), # 准入 10:00:30 + _row("b", created_at="2026-07-20T10:02:35", latency_ms=25_000), # 准入 10:02:10 + ] + inv_rpm_never_exceeded(rows, {"s1": 1}) + + def test_rpm_server_clock_offset_dedistorts(self): + # P6 实跑复现: Redis 服务器钟快 31s,本机分钟内 10 次准入 + # 实为两个服务器窗口各 5 次(+31 后: 37..59 前窗 | 63..90 后窗) + secs = [6, 19, 21, 26, 28, 32, 42, 47, 52, 59] + rows = [_row(str(i), created_at=f"2026-07-20T10:00:{s:02d}") for i, s in enumerate(secs)] + inv_rpm_never_exceeded(rows, {"s1": 5}, clock_offset_s=31.0) + + def test_rpm_true_breach_still_caught(self): + # 6 次准入全落窗口中部(无贴边歧义),限额 5 → 必须击穿 + rows = [ + _row(str(i), created_at=f"2026-07-20T10:00:{s:02d}") + for i, s in enumerate([10, 20, 25, 30, 40, 50]) + ] + with pytest.raises(AssertionError): + inv_rpm_never_exceeded(rows, {"s1": 5}) + + def test_rpm_excludes_cache_hits(self): + # P6 实跑教训之二: 缓存命中在限流闸之前返回,未耗 RPM 名额也未打 + # 网关,但遥测按"遥测必录"记行且署原源名——不得计入 RPM 口径 + rows = [ + _row(str(i), created_at=f"2026-07-20T10:00:{10 + i * 5:02d}", cache_hit=1) + for i in range(8) + ] + rows.append(_row("real", created_at="2026-07-20T10:00:30")) + inv_rpm_never_exceeded(rows, {"s1": 1}) + + def test_rpm_boundary_jitter_exempted(self): + # 贴边行(±2s 采样噪声)在邻窗有余量时可归邻窗,不算击穿 + rows = [ + _row(str(i), created_at=f"2026-07-20T10:00:{s:02d}") + for i, s in enumerate([59, 10, 20, 30, 40, 50]) + ] + inv_rpm_never_exceeded(rows, {"s1": 5}) + def test_rss_stable(self): inv_rss_stable([100.0, 105.0, 110.0], max_growth_mb=50.0) with pytest.raises(AssertionError): @@ -197,6 +251,54 @@ class TestInvariants: assert structured_success_rate(rows, session_suffix="-p3") == pytest.approx(0.5) +class TestHarnessMeasurement: + """P6 伪击穿修复(2026-07-21): RSS 采当前值、分发有界。""" + + def test_rss_mb_reports_current_not_peak(self): + import resource + import sys as _sys + + from tools.soak.run_soak import _rss_mb + + ballast = bytearray(300 * 1024 * 1024) + ballast[::4096] = b"x" * len(ballast[::4096]) # 触页,确保计入 RSS + divisor = 1e6 if _sys.platform == "darwin" else 1024 + peak_mb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / divisor + del ballast + # ru_maxrss 含 300MB 压舱物峰值;当前 RSS 应已显著回落 + assert _rss_mb() < peak_mb - 100 + + async def test_paced_dispatch_bounds_inflight_and_honors_stop(self): + import asyncio + + from tools.soak.run_soak import _paced_dispatch + + gauge = {"now": 0, "max": 0, "done": 0} + dispatched: list[int] = [] + + async def _work(): + gauge["now"] += 1 + gauge["max"] = max(gauge["max"], gauge["now"]) + await asyncio.sleep(0.02) + gauge["now"] -= 1 + gauge["done"] += 1 + + async def _gen(): + for i in range(10): + yield "chat", {"i": i} + + inflight = await _paced_dispatch( + _gen(), + sem=asyncio.Semaphore(2), + spawn=lambda kind, kwargs: _work(), + should_stop=lambda: len(dispatched) >= 6, + on_dispatched=lambda: dispatched.append(1), + ) + await asyncio.gather(*inflight, return_exceptions=True) + assert gauge["max"] <= 2 # 并发名额先占后建任务 + assert gauge["done"] == 6 # should_stop 在预算命中处截停分发 + + class TestLiveInvariantsAndCaps: def test_capped_budget_clamps_to_signed_values(self): from tools.soak.scoreboard import capped_budget diff --git a/tools/soak/run_soak.py b/tools/soak/run_soak.py index 3838222..55bbc50 100644 --- a/tools/soak/run_soak.py +++ b/tools/soak/run_soak.py @@ -18,7 +18,7 @@ import asyncio import json import multiprocessing as mp import os -import resource +import subprocess import sys import time from pathlib import Path @@ -76,8 +76,40 @@ async def _flush_db3(redis_url: str) -> None: def _rss_mb() -> float: - peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss - return peak / 1e6 if sys.platform == "darwin" else peak / 1024 + """当前 RSS(ps 口径,darwin/linux 皆 KB)。 + + 不用 ru_maxrss——那是历史峰值只增不减,2026-07-21 P6 曾把分发期 + 瞬时缓冲峰值误判为 780MB "泄漏"(实测跑中稳定 36MB)。 + """ + out = subprocess.run( + ["ps", "-o", "rss=", "-p", str(os.getpid())], capture_output=True, text=True, check=True + ) + return int(out.stdout.strip()) / 1024.0 + + +async def _paced_dispatch(generator, *, sem, spawn, should_stop, on_dispatched) -> set[asyncio.Task]: + """有界分发: 先占并发名额再建任务,名额由任务收尾释放。 + + 2026-07-21 P6 教训: 无界 create_task 曾在 15s 内入队全部预算, + 内存峰值 ~800MB,且 --max-hours 截止检查随分发结束而失效。 + """ + inflight: set[asyncio.Task] = set() + + async def _run(coro) -> None: + try: + await coro + finally: + sem.release() + + async for kind, kwargs in generator: + if should_stop(): + break + await sem.acquire() + on_dispatched() + task = asyncio.create_task(_run(spawn(kind, kwargs))) + inflight.add(task) + task.add_done_callback(inflight.discard) + return inflight async def _worker_async(args: argparse.Namespace, worker_idx: int) -> None: @@ -107,42 +139,42 @@ async def _worker_async(args: argparse.Namespace, worker_idx: int) -> None: deadline = time.monotonic() + args.max_hours * 3600 budget_calls = capped_budget(args.scenario, args.budget_calls) // args.workers budget_tokens = min(args.budget_tokens, TOKEN_HARD_CAP) // args.workers - inflight: set[asyncio.Task] = set() async def _one(kind: str, kwargs: dict) -> None: - async with sem: - try: - # run 级双层隔离: FLUSHDB(入口)+ 每 run 独立 namespace(此处) - kwargs.setdefault("cache_namespace", run_id) - resp = await client.chat(**kwargs) - stats["ok"] += 1 - stats["tokens"] += resp.prompt_tokens + resp.completion_tokens - except asyncio.CancelledError: - stats["cancelled"] += 1 - raise - except Exception as exc: # 记分板从遥测读错误分布;这里只计数 - stats["failed"] += 1 - print(f"[w{worker_idx}] 调用失败: {type(exc).__name__}: {exc}", flush=True) + try: + # run 级双层隔离: FLUSHDB(入口)+ 每 run 独立 namespace(此处) + kwargs.setdefault("cache_namespace", run_id) + resp = await client.chat(**kwargs) + stats["ok"] += 1 + stats["tokens"] += resp.prompt_tokens + resp.completion_tokens + except asyncio.CancelledError: + stats["cancelled"] += 1 + raise + except Exception as exc: # 记分板从遥测读错误分布;这里只计数 + stats["failed"] += 1 + print(f"[w{worker_idx}] 调用失败: {type(exc).__name__}: {exc}", flush=True) - async for kind, kwargs in generator: - if ( + def _should_stop() -> bool: + return ( stats["calls"] >= budget_calls or stats["tokens"] >= budget_tokens or time.monotonic() > deadline - ): - break + ) + + def _on_dispatched() -> None: stats["calls"] += 1 - task = asyncio.create_task(_one(kind, kwargs)) - inflight.add(task) - task.add_done_callback(inflight.discard) if stats["calls"] % 20 == 0: rss_samples.append(_rss_mb()) print(f"[w{worker_idx}] {stats}", flush=True) + + inflight = await _paced_dispatch( + generator, sem=sem, spawn=_one, should_stop=_should_stop, on_dispatched=_on_dispatched + ) if inflight: await asyncio.gather(*inflight, return_exceptions=True) # 优雅收尾 in-flight rss_samples.append(_rss_mb()) await client.aclose() - result = {"stats": stats, "rss_mb": rss_samples, "telemetry": str(telemetry_path)} + result = {"stats": stats, "rss_current_mb": rss_samples, "telemetry": str(telemetry_path)} (_ROOT / f"data/soak/result_{run_id}_{worker_idx}.json").write_text( json.dumps(result), encoding="utf-8" ) @@ -192,9 +224,25 @@ def _scoreboard(args: argparse.Namespace, env: dict[str, str]) -> None: # 仅四段源键 {SCOPE}__{PROVIDER}__{N}__RPM(排除 {SCOPE}__GLOBAL__RPM) if len(parts) == 4 and parts[0] == args.scope and parts[3] == "RPM": rpm_conf[f"{parts[1].lower()}_{parts[2]}"] = int(val) - _check("RPM 从未击穿(分钟桶)", sb.inv_rpm_never_exceeded, rows, rpm_conf) + # RPM 窗口口径 = 限流器所用时钟: redis 后端为服务器钟(需测偏移),memory 为本机钟 + clock_offset_s = 0.0 + if env.get("PGW_LIMITER_BACKEND") == "redis": + clock_offset_s = asyncio.run(_redis_clock_offset_s(env["REDIS_URL"])) + print(f"Redis 服务器钟偏移: {clock_offset_s:+.2f}s(RPM 窗口口径校正)") + _check( + "RPM 从未击穿(限流器窗口口径)", + sb.inv_rpm_never_exceeded, + rows, + rpm_conf, + clock_offset_s=clock_offset_s, + ) for r in results: - _check(f"RSS 平稳(w)", sb.inv_rss_stable, r["rss_mb"], max_growth_mb=args.max_rss_growth_mb) + samples = r.get("rss_current_mb") + if samples is None: + # 2026-07-21 前的旧结果只存 ru_maxrss 峰值序列,不可判泄漏 + verdicts.append(("RSS 平稳(w)", "SKIP — 旧格式峰值采样不可判(P6 伪击穿教训)")) + else: + _check("RSS 平稳(w)", sb.inv_rss_stable, samples, max_growth_mb=args.max_rss_growth_mb) # 不变量 1(记账归零 + gate 可再准入): 活后端检查,仅 redis 后端可跨进程复查 if env.get("PGW_LIMITER_BACKEND") == "redis": asyncio.run(_live_checks(args, env, verdicts)) @@ -215,6 +263,18 @@ def _scoreboard(args: argparse.Namespace, env: dict[str, str]) -> None: raise SystemExit("硬不变量被击穿,见报告") +async def _redis_clock_offset_s(redis_url: str) -> float: + """Redis 服务器钟相对本机钟的偏移(秒);限流窗口 id 以服务器钟为准。""" + import redis.asyncio as aioredis + + client = aioredis.from_url(redis_url) + try: + sec, usec = await client.time() + return float(sec) + float(usec) / 1e6 - time.time() + finally: + await client.aclose() + + async def _live_checks( args: argparse.Namespace, env: dict[str, str], verdicts: list[tuple[str, str]] ) -> None: @@ -253,18 +313,35 @@ async def _live_checks( def main() -> None: parser = argparse.ArgumentParser(description="PolyGateway 真实数据压测") parser.add_argument("--scenario", required=True, choices=["P1", "P2", "P3", "P4", "P5", "P6"]) - parser.add_argument("--budget-calls", type=int, required=True) - parser.add_argument("--budget-tokens", type=int, required=True) + parser.add_argument("--budget-calls", type=int) + parser.add_argument("--budget-tokens", type=int) parser.add_argument("--workers", type=int, default=1) parser.add_argument("--concurrency", type=int, default=8, help="单 worker 并发上限") parser.add_argument("--scope", default="SOAK") parser.add_argument("--run-id", default=None) parser.add_argument("--max-hours", type=float, default=3.0) parser.add_argument("--max-rss-growth-mb", type=float, default=500.0) + parser.add_argument( + "--rescore", + metavar="RUN_ID", + default=None, + help="对已完成 run 离线重跑记分板(不产流量、不 FLUSHDB;须带原 env 覆盖)", + ) args = parser.parse_args() + env = _merged_env() + if args.rescore: + args.run_id = args.rescore + found = sorted((_ROOT / "data/soak").glob(f"result_{args.run_id}_*.json")) + if not found: + raise SystemExit(f"拒跑: 找不到 data/soak/result_{args.run_id}_*.json") + args.workers = len(found) + _guard(env, args.workers, args.scope) + _scoreboard(args, env) + return + if args.budget_calls is None or args.budget_tokens is None: + parser.error("--budget-calls/--budget-tokens 为实跑必填") if args.run_id is None: args.run_id = time.strftime("soak_%Y%m%d_%H%M%S") - env = _merged_env() _guard(env, args.workers, args.scope) print(f"run_id={args.run_id}: FLUSHDB db3 + namespace 隔离") asyncio.run(_flush_db3(env["REDIS_URL"])) diff --git a/tools/soak/scoreboard.py b/tools/soak/scoreboard.py index b7a47e1..13680bf 100644 --- a/tools/soak/scoreboard.py +++ b/tools/soak/scoreboard.py @@ -45,23 +45,58 @@ def inv_call_ids_unique(rows: list[Row]) -> None: assert not dupes, f"call_id 重复: {dupes[:5]}" -def _minute_bucket(created_at: str) -> int: - return int(datetime.fromisoformat(str(created_at)).timestamp()) // 60 +def _admit_second(row: Row, clock_offset_s: float) -> float: + """还原准入时刻(服务器钟): created_at 是完成落库时刻,减去调用延迟。""" + done = datetime.fromisoformat(str(row["created_at"])).timestamp() + latency_ms = row.get("latency_ms") or 0 + return done - float(latency_ms) / 1000.0 + clock_offset_s -def inv_rpm_never_exceeded(rows: list[Row], per_source_rpm: dict[str, int]) -> None: - """不变量 3: 按遥测时间戳重算,任一分钟桶内单源请求数 ≤ RPM 配置。 - - 口径 = 固定分钟窗口(与限流器 `int(sec/60)` 同源);滑动 60s 窗会对 - "窗尾+窗头"的合法背靠背流量误报,不采用(findings §4 条 3 的执行口径)。 - """ - buckets: Counter[tuple[str, int]] = Counter( - (r["source_name"], _minute_bucket(r["created_at"])) - for r in rows - if per_source_rpm.get(r["source_name"], 0) > 0 +def _movable_edge_rows( + admits: list[float], buckets: Counter[int], minute: int, limit: int, slack_s: float +) -> int: + """超限窗口内可归邻窗的贴边行数(受邻窗余量约束)。""" + near_edge = sum( + 1 + for a in admits + if int(a // 60) == minute and min(a - minute * 60, (minute + 1) * 60 - a) <= slack_s ) - breaches = {key: n for key, n in buckets.items() if n > per_source_rpm[key[0]]} - assert not breaches, f"RPM 击穿: {dict(list(breaches.items())[:5])}" + room = max(0, limit - buckets.get(minute - 1, 0)) + max(0, limit - buckets.get(minute + 1, 0)) + return min(near_edge, room) + + +def inv_rpm_never_exceeded( + rows: list[Row], + per_source_rpm: dict[str, int], + *, + clock_offset_s: float = 0.0, + boundary_slack_s: float = 2.0, +) -> None: + """不变量 3: 任一限流器分钟窗口内单源准入数 ≤ RPM 配置。 + + 口径与限流器同源(`backends/redis/limiter.py` `_window_id`): 窗口 = + **Redis 服务器钟**的固定分钟;准入时刻 = created_at(完成落库)− latency。 + 滑动 60s 窗会对"窗尾+窗头"的合法背靠背流量误报,不采用(findings §4 条 3)。 + created_at 秒级截断给准入时刻 ±秒级噪声,距窗口边界 ≤ boundary_slack_s 的 + 行允许归入有余量的邻窗。缓存命中行不计: 缓存在限流闸之前返回,未耗名额 + 也未打网关(遥测必录使其带原源名落库)。2026-07-21 P6 教训: 本机钟聚桶 + + 完成时刻口径 + 计入缓存行,三重口径偏差曾把合规流量误判为击穿。 + """ + per_source: dict[str, list[float]] = {} + for r in rows: + if per_source_rpm.get(r["source_name"], 0) > 0 and not r.get("cache_hit"): + per_source.setdefault(r["source_name"], []).append(_admit_second(r, clock_offset_s)) + breaches: dict[tuple[str, int], int] = {} + for source, admits in per_source.items(): + limit = per_source_rpm[source] + buckets = Counter(int(a // 60) for a in admits) + for minute, n in buckets.items(): + if n <= limit: + continue + movable = _movable_edge_rows(admits, buckets, minute, limit, boundary_slack_s) + if n - movable > limit: + breaches[(source, minute)] = n + assert not breaches, f"RPM 击穿(准入时刻+服务器钟口径): {dict(list(breaches.items())[:5])}" def inv_rss_stable(samples_mb: list[float], *, max_growth_mb: float) -> None: