fix: align soak scoreboard metrics with limiter semantics

RPM invariant now buckets by admit time on the Redis server clock,
exempts +/-2s boundary jitter, and excludes cache-hit rows that never
consumed a limiter slot. RSS sampling reports current ps RSS instead of
the monotonic ru_maxrss peak. Dispatch is paced by the concurrency
semaphore so --max-hours stays live and memory stays bounded. Add
--rescore RUN_ID to re-judge a finished run offline.

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("a", created_at="2026-07-20T10:00:10"),
-            _row("b", created_at="2026-07-20T10:00:59"),
+            _row("b", created_at="2026-07-20T10:00:50"),
-            _row("c", created_at="2026-07-20T10:01:01"),
+            _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

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
+    """当前 RSS(ps 口径,darwin/linux 皆 KB)。
-    return peak / 1e6 if sys.platform == "darwin" else peak / 1024
+
+    不用 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:
-            try:
+            # run 级双层隔离: FLUSHDB(入口)+ 每 run 独立 namespace(此处)
-                # run 级双层隔离: FLUSHDB(入口)+ 每 run 独立 namespace(此处)
+            kwargs.setdefault("cache_namespace", run_id)
-                kwargs.setdefault("cache_namespace", run_id)
+            resp = await client.chat(**kwargs)
-                resp = await client.chat(**kwargs)
+            stats["ok"] += 1
-                stats["ok"] += 1
+            stats["tokens"] += resp.prompt_tokens + resp.completion_tokens
-                stats["tokens"] += resp.prompt_tokens + resp.completion_tokens
+        except asyncio.CancelledError:
-            except asyncio.CancelledError:
+            stats["cancelled"] += 1
-                stats["cancelled"] += 1
+            raise
-                raise
+        except Exception as exc:  # 记分板从遥测读错误分布;这里只计数
-            except Exception as exc:  # 记分板从遥测读错误分布;这里只计数
+            stats["failed"] += 1
-                stats["failed"] += 1
+            print(f"[w{worker_idx}] 调用失败: {type(exc).__name__}: {exc}", flush=True)
-                print(f"[w{worker_idx}] 调用失败: {type(exc).__name__}: {exc}", flush=True)
 
-    async for kind, kwargs in generator:
+    def _should_stop() -> bool:
-        if (
+        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-calls", type=int)
-    parser.add_argument("--budget-tokens", type=int, required=True)
+    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"]))

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:
+def _admit_second(row: Row, clock_offset_s: float) -> float:
-    return int(datetime.fromisoformat(str(created_at)).timestamp()) // 60
+    """还原准入时刻(服务器钟): 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:
+def _movable_edge_rows(
-    """不变量 3: 按遥测时间戳重算,任一分钟桶内单源请求数 ≤ RPM 配置。
+    admits: list[float], buckets: Counter[int], minute: int, limit: int, slack_s: float
-
+) -> int:
-    口径 = 固定分钟窗口(与限流器 `int(sec/60)` 同源);滑动 60s 窗会对
+    """超限窗口内可归邻窗的贴边行数(受邻窗余量约束)。"""
-    "窗尾+窗头"的合法背靠背流量误报,不采用(findings §4 条 3 的执行口径)。
+    near_edge = sum(
-    """
+        1
-    buckets: Counter[tuple[str, int]] = Counter(
+        for a in admits
-        (r["source_name"], _minute_bucket(r["created_at"]))
+        if int(a // 60) == minute and min(a - minute * 60, (minute + 1) * 60 - a) <= slack_s
-        for r in rows
-        if per_source_rpm.get(r["source_name"], 0) > 0
     )
-    breaches = {key: n for key, n in buckets.items() if n > per_source_rpm[key[0]]}
+    room = max(0, limit - buckets.get(minute - 1, 0)) + max(0, limit - buckets.get(minute + 1, 0))
-    assert not breaches, f"RPM 击穿: {dict(list(breaches.items())[:5])}"
+    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: