diff --git a/research-wiki/designs/2026-07-21-m25-resilience-design.md b/research-wiki/designs/2026-07-21-m25-resilience-design.md index 9b9f47f..d27be92 100644 --- a/research-wiki/designs/2026-07-21-m25-resilience-design.md +++ b/research-wiki/designs/2026-07-21-m25-resilience-design.md @@ -110,6 +110,11 @@ **动因**: 第六轮实测网关限速是**按账号速率**而非并发——外部负载高峰时我方仅 26 req/min 仍吃 16% 429,AIMD(并发型)压不住速率型限流;429 每次消耗 1/3 重试预算,饱和窗口里调用被"429 链"干净杀死(16 retry_exhausted/336)。 **修正**: 携 Retry-After 的 429 是**服务端调度指令而非失败**(gRPC A6 pushback / SRE 语义): 照常退避(与 Retry-After 取大)、照常喂 AIMD/健康分,但**不消耗重试预算**;新增**调用级时间上限**(复用 `stall_window_s`,缺省 300s)在重试循环顶部兜底——持续饱和超窗即按既有 `stalled` 语义抛出,防无限循环。对下游的承诺变化: 饱和期调用最长等待 stall_window 而非快速失败(生产语义,迁移文档标注)。 +### 3.39 迭代 6 补遗: 健康证据抑制连续通道(2026-07-21,第八轮取证驱动) + +**动因**: 第八轮失败链取证发现健康源1 自己进 cooldown——空补全率 ~20% 下 5 连败每 ~3000 次尝试随机发生一次,连续通道(阈值 5)把唯一好源关 60s,期间调用全灭(链样本: `{'minimax_1': 'cooldown', 其余: 'cooldown/rate_limited'}`)。 +**修正**(Resilience4j 精神: 率证据优于连败直觉): 窗口样本 ≥ min_calls 且失败率 < fail_rate 时**抑制连续通道开路**——有充分健康证据的源上连败是统计噪声。连续通道本职(冷启动/低流量快杀死源)保留(窗口样本不足时照常开路);温热源猝死由率通道在失败累积后接管(检出延迟从 threshold 次升至率窗口收敛,契约测试⑩钉住);SourceDead force_open 不受影响。 + ### 3.4 不做与预留(方案 C 组件的接入点) - 账号级 429 共享退避: 不做;预留 = 冷却备忘 key 从 source_name 换 account_key 即可接入(LiteLLM 先例,治理粒度=配额粒度原则记入 ARCHITECTURE)。 diff --git a/src/polygateway/backends/memory/breaker.py b/src/polygateway/backends/memory/breaker.py index 72df5c4..dbf944c 100644 --- a/src/polygateway/backends/memory/breaker.py +++ b/src/polygateway/backends/memory/breaker.py @@ -81,6 +81,14 @@ class InMemoryGate: if failed: g.f0 += 1 + def _window_evidently_healthy(self, g: _SourceGate) -> bool: + """窗口样本充足且失败率低于阈值 = 有充分健康证据(迭代 6 抑制判据)。""" + self._rotate_window(g) + attempts = g.a0 + g.a1 + if attempts < self._cfg.min_calls: + return False + return (g.f0 + g.f1) / attempts < self._cfg.fail_rate + def _rate_channel_open(self, g: _SourceGate) -> bool: self._rotate_window(g) attempts = g.a0 + g.a1 @@ -227,8 +235,10 @@ class InMemoryGate: g.fails += 1 if self._rate_channel_open(g): self._open(g, reason, bump_streak=True) - elif g.fails >= self._cfg.fail_threshold: - self._open(g, reason, bump_streak=False) # 连续通道不递增(C1 封顶) + elif g.fails >= self._cfg.fail_threshold and not self._window_evidently_healthy(g): + # 连续通道不递增(C1 封顶);窗口证据充足且健康时连败是噪声, + # 不误熔"当前最好的源"(迭代 6,第八轮实证: 源1 被 5 连空补全误关 60s) + self._open(g, reason, bump_streak=False) return self._snapshot(g, applied=True) async def release_probe(self, entry: GateDecision) -> GateUpdate: diff --git a/src/polygateway/backends/redis/breaker.py b/src/polygateway/backends/redis/breaker.py index 58091ee..f632056 100644 --- a/src/polygateway/backends/redis/breaker.py +++ b/src/polygateway/backends/redis/breaker.py @@ -179,6 +179,7 @@ end local threshold = tonumber(ARGV[5]) local open_via_rate = false +local window_healthy = false if force_open == 0 and is_probe == 0 then rotate_window(KEYS[1], now, tonumber(ARGV[10])) redis.call('HINCRBY', KEYS[1], 'a0', 1) @@ -187,8 +188,13 @@ if force_open == 0 and is_probe == 0 then + tonumber(redis.call('HGET', KEYS[1], 'a1') or '0') local fails_w = tonumber(redis.call('HGET', KEYS[1], 'f0') or '0') + tonumber(redis.call('HGET', KEYS[1], 'f1') or '0') - if attempts >= tonumber(ARGV[8]) and fails_w / attempts >= tonumber(ARGV[9]) then - open_via_rate = true + if attempts >= tonumber(ARGV[8]) then + if fails_w / attempts >= tonumber(ARGV[9]) then + open_via_rate = true + else + -- 窗口证据充足且健康: 连败是噪声,抑制连续通道(迭代 6) + window_healthy = true + end end end @@ -197,7 +203,8 @@ if force_open == 1 or is_probe == 1 then else failures = failures + 1 end -if force_open == 1 or is_probe == 1 or open_via_rate or failures >= threshold then +local open_via_streak = failures >= threshold and not window_healthy +if force_open == 1 or is_probe == 1 or open_via_rate or open_via_streak then -- 递增 streak 的只有率通道开路与探针失败重开(C1: 连续通道/force_open 不递增) if open_via_rate or is_probe == 1 then redis.call('HINCRBY', KEYS[1], 'reopen_streak', 1) diff --git a/tests/contracts/test_breaker_contract.py b/tests/contracts/test_breaker_contract.py index 2ffabea..00c4275 100644 --- a/tests/contracts/test_breaker_contract.py +++ b/tests/contracts/test_breaker_contract.py @@ -290,3 +290,36 @@ class TestRetryAfter: gate = gate_factory(_CFG) await _open_gate(gate, "s1") # s1 开路;s2 健康 assert await gate.retry_after_s(("s1", "s2")) == 0.0 + + +class TestConsecutiveSuppression: + """迭代 6: 窗口证据充足且健康时,连败是噪声,不开路(设计 §3.39)。""" + + async def test_streak_suppressed_on_evidently_healthy_source(self, gate_factory): + # 20 成功垫底(窗口样本充足、失败率低)后 3 连败(阈值 3)→ 不开路 + cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=10) + gate = gate_factory(cfg) + for _ in range(20): + entry = await gate.try_enter("s1", "w") + await gate.record_success(entry) + await _fail(gate, reason="timeout", n=3) + assert (await gate.try_enter("s1", "w")).allowed + + async def test_streak_fires_when_window_insufficient(self, gate_factory): + # 冷启动(窗口样本不足)3 连败照常开路——连续通道本职保留 + cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=10) + gate = gate_factory(cfg) + update = await _fail(gate, reason="timeout", n=3) + assert update.state is GateState.OPEN + + async def test_warm_source_going_dead_caught_by_rate(self, gate_factory): + # 温热源猝死: 连败被抑制,但失败率窗口随失败累积必然接管 + cfg = BreakerConfig( + fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=4, fail_rate=0.6 + ) + gate = gate_factory(cfg) + for _ in range(3): + entry = await gate.try_enter("s1", "w") + await gate.record_success(entry) + update = await _fail(gate, reason="timeout", n=5) # 5/8 = 0.625 ≥ 0.6 + assert update.state is GateState.OPEN