fix: address M2 verifier findings in soak harness

tools/soak/run_soak.py

@@ -36,17 +36,33 @@ def _merged_env() -> dict[str, str]:
     return {k: v for k, v in {**dotenv_values(_ROOT / ".env"), **os.environ}.items() if v}
 
 
-def _guard(env: dict[str, str], workers: int) -> None:
+_SIGNED_MAX_CONCURRENCY = 100  # 网关保护签字值(设计 §8.1,2026-07-20 人类)
+_SIGNED_MAX_RPM = 600
+
+
+def _guard(env: dict[str, str], workers: int, scope: str) -> None:
     redis_url = env.get("REDIS_URL", "")
     if not redis_url.rstrip("/").endswith("/3"):
         raise SystemExit(f"拒跑: REDIS_URL 必须指向专用 db3,当前 {redis_url!r}")
     pg = env.get("PGW_TELEMETRY_PG_DSN", "")
     if pg and not pg.rstrip("/").endswith("/polygateway"):
-        raise SystemExit(f"拒跑: PG DSN 必须指向 polygateway 专用库,当前库名不符")
+        raise SystemExit("拒跑: PG DSN 必须指向 polygateway 专用库,当前库名不符")
     if workers > 1 and (
         env.get("PGW_LIMITER_BACKEND") != "redis" or env.get("PGW_BREAKER_BACKEND") != "redis"
     ):
         raise SystemExit("拒跑: --workers>1 需要 PGW_LIMITER_BACKEND/PGW_BREAKER_BACKEND=redis")
+    # 网关保护(签字值 100/600): 压测 scope 必须配全局闸且不超签字上限
+    conc = env.get(f"{scope}__GLOBAL__MAX_CONCURRENCY")
+    rpm = env.get(f"{scope}__GLOBAL__RPM")
+    if not conc or not rpm:
+        raise SystemExit(
+            f"拒跑: 压测须配网关保护 {scope}__GLOBAL__MAX_CONCURRENCY(≤{_SIGNED_MAX_CONCURRENCY})"
+            f" 与 {scope}__GLOBAL__RPM(≤{_SIGNED_MAX_RPM})——设计 §8.1 签字值"
+        )
+    if int(conc) > _SIGNED_MAX_CONCURRENCY or int(rpm) > _SIGNED_MAX_RPM:
+        raise SystemExit(
+            f"拒跑: 全局闸 {conc}/{rpm} 超签字上限 {_SIGNED_MAX_CONCURRENCY}/{_SIGNED_MAX_RPM}"
+        )
 
 
 async def _flush_db3(redis_url: str) -> None:
@@ -86,8 +102,10 @@ async def _worker_async(args: argparse.Namespace, worker_idx: int) -> None:
     sem = asyncio.Semaphore(args.concurrency)
     stats = {"calls": 0, "ok": 0, "failed": 0, "cancelled": 0, "tokens": 0}
     rss_samples = [_rss_mb()]
+    from tools.soak.scoreboard import capped_budget
+
     deadline = time.monotonic() + args.max_hours * 3600
-    budget_calls = args.budget_calls // args.workers
+    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()
 
@@ -175,6 +193,17 @@ def _scoreboard(args: argparse.Namespace, env: dict[str, str]) -> None:
     _check("RPM 从未击穿(分钟桶)", sb.inv_rpm_never_exceeded, rows, rpm_conf)
     for r in results:
         _check(f"RSS 平稳(w)", sb.inv_rss_stable, r["rss_mb"], 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))
+    else:
+        verdicts.append(("记账归零/gate 可再准入", "SKIP — memory 后端跨进程不可查"))
+    # 不变量 2c: P5/P6 且配置了故障源 → 故障必须真实发生
+    if args.scenario in ("P5", "P6"):
+        fault_names = [
+            name for name in rpm_conf if rpm_conf[name] <= 10
+        ]  # 紧闸源;坏 key/黑洞源由错误分布人工核对(findings §4 条 2 校准留待 P5 实跑)
+        _check("故障混编生效(P5/P6)", sb.inv_fault_errors_present, rows, fault_source_names=fault_names)
     rate = sb.structured_success_rate(rows, session_suffix="-p3")
     verdicts.append(("P3 结构化成功率", f"{rate:.3f}(基线首跑建立)"))
     report = sb.render_report(args.run_id, rows, verdicts)
@@ -185,6 +214,41 @@ def _scoreboard(args: argparse.Namespace, env: dict[str, str]) -> None:
         raise SystemExit("硬不变量被击穿,见报告")
 
 
+async def _live_checks(
+    args: argparse.Namespace, env: dict[str, str], verdicts: list[tuple[str, str]]
+) -> None:
+    """不变量 1 活后端复查(跑后): 限流 inflight 归零 + 熔断门可再准入。"""
+    from polygateway.backends.redis.breaker import RedisGate
+    from polygateway.backends.redis.limiter import RedisLimiter
+    from polygateway.config import GatewaySettings
+
+    from tools.soak import scoreboard as sb
+
+    settings = GatewaySettings.from_env(args.scope, env=env)
+    names = [s.name for s in settings.sources]
+    limiter = RedisLimiter.from_url(
+        env["REDIS_URL"],
+        scope=settings.scope,
+        sources={s.name: s for s in settings.sources},
+        global_limits=settings.global_limits,
+        lease_ttl_s=settings.lease_ttl_s,
+    )
+    gate = RedisGate.from_url(env["REDIS_URL"], config=settings.breaker, scope=settings.scope)
+    try:
+        for name, invariant in (
+            ("记账归零(inflight)", sb.inv_accounting_zeroed(limiter, names)),
+            ("gate 可再准入(探针不悬挂)", sb.inv_gate_reenterable(gate, names)),
+        ):
+            try:
+                await invariant
+                verdicts.append((name, "PASS"))
+            except AssertionError as exc:
+                verdicts.append((name, f"FAIL — {exc}"))
+    finally:
+        await limiter.aclose()
+        await gate.aclose()
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description="PolyGateway 真实数据压测")
     parser.add_argument("--scenario", required=True, choices=["P1", "P2", "P3", "P4", "P5", "P6"])
@@ -200,7 +264,7 @@ def main() -> None:
     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)
+    _guard(env, args.workers, args.scope)
     print(f"run_id={args.run_id}: FLUSHDB db3 + namespace 隔离")
     asyncio.run(_flush_db3(env["REDIS_URL"]))
     if args.workers == 1: