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 759b2f8..81fb7c1 100644 --- a/research-wiki/designs/2026-07-21-m25-resilience-design.md +++ b/research-wiki/designs/2026-07-21-m25-resilience-design.md @@ -94,6 +94,12 @@ **测试**: 单测削减/增长/上下限;RetryMW 集成——429 后准入收紧、被 paced 源跳过不耗预算、全 paced 走轮询非 CircuitOpen、在途数在异常/取消路径归零。 +### 3.36 迭代 2 补遗: 调用内降权加健康门槛(2026-07-21,P6 第三轮数据驱动) + +**动因**: AIMD 后 429 归零,但第三轮 750 次完成仍 21 败(2.8%)。根因: 无条件"失败 ≥2 让位"在**异构池**把第三次尝试推给已知坏源——健康源两次空补全(网关 17% 空补全率)后被降权,替补是 10% 成功率的看门狗源;且该源的"成功"是被 0.5s 看门狗筛出的短促退化响应(均值 128 token vs 健康 262),喂给结构化解析再炸一层。期望值算术: 第三次留在健康源成功率 ~83%,推给坏源 ~10%。 + +**修正**: `OutcomeAwareSelector` 协议增 `health(source_name) -> float`(EWMA 裸值);RetryMW 降权仅在存在**可信替代**(某未失败候选 health ≥ 0.5 × 失败源 health)时生效,否则原地第三试。无健康视图的选源器(round_robin 等)保持无条件降权(冷启动保护原语义)。 + ### 3.4 不做与预留(方案 C 组件的接入点) - 账号级 429 共享退避: 不做;预留 = 冷却备忘 key 从 source_name 换 account_key 即可接入(LiteLLM 先例,治理粒度=配额粒度原则记入 ARCHITECTURE)。 diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 64b528a..34ac8aa 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -74,13 +74,36 @@ def backoff_delay( def _demote_call_failures( - ordered: list[SourceConfig], attempt_fails: dict[str, int] + ordered: list[SourceConfig], + attempt_fails: dict[str, int], + health: Callable[[str], float] | None, ) -> list[SourceConfig]: - """调用内降权(设计 §3.3): 本调用失败 ≥2 次的源移尾;全失败则保持原序。""" + """调用内降权(设计 §3.3/§3.36): 失败 ≥2 次且存在可信替代才让位。 + + 可信替代 = 某未失败候选 health ≥ 0.5 × 失败源 health——异构池里健康源 + 偶发失败不该被推向已知坏源(第三轮教训: 期望成功率 83% vs 10%)。 + 无健康视图(round_robin 等)保持无条件降权(冷启动保护)。 + """ demoted = [s for s in ordered if attempt_fails.get(s.name, 0) >= 2] if not demoted or len(demoted) == len(ordered): return ordered - return [s for s in ordered if attempt_fails.get(s.name, 0) < 2] + demoted + if health is not None: + demoted = _credible_demotions(ordered, demoted, attempt_fails, health) + if not demoted: + return ordered + names = {d.name for d in demoted} + return [s for s in ordered if s.name not in names] + demoted + + +def _credible_demotions( + ordered: list[SourceConfig], + demoted: list[SourceConfig], + attempt_fails: dict[str, int], + health: Callable[[str], float], +) -> list[SourceConfig]: + """健康门槛过滤: 仅当存在"健康分 ≥ 失败源一半"的未失败候选,让位才有意义。""" + alts = [o for o in ordered if attempt_fails.get(o.name, 0) < 2] + return [s for s in demoted if any(health(o.name) >= 0.5 * health(s.name) for o in alts)] def _failure_reason(exc: PolyGatewayError) -> str: @@ -138,6 +161,7 @@ class RetryMW: self._memo = cooldown_memo or SourceCooldownMemo(now=now) # M2.5: 选源器可选健康喂数端口,构造期 isinstance 判定一次(设计 §3.2) self._outcome_sink = selector if isinstance(selector, OutcomeAwareSelector) else None + self._health_view = self._outcome_sink.health if self._outcome_sink else None # M2.5 §3.35: AIMD 自适应并发——429 收紧、成功回涨,超限调用排队不烧预算 self._pacer = pacer or AdaptivePacer(ceiling=64.0) self._emitter = emitter @@ -180,7 +204,9 @@ class RetryMW: ) -> tuple[tuple[SourceConfig, Permit, GateDecision] | None, int]: stats = {s.name: await self._quota.stats(s) for s in self._sources} gate_rejections = 0 - ordered = _demote_call_failures(self._selector.order(self._sources, stats), attempt_fails) + ordered = _demote_call_failures( + self._selector.order(self._sources, stats), attempt_fails, self._health_view + ) for cand in ordered: if self._memo.active(cand.name): # 冷却备忘跳过也计入拒绝数,保住 circuit_open 判据(CHS 同款) diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index a188e3d..2c894bd 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -194,6 +194,8 @@ class OutcomeAwareSelector(Protocol): def record_outcome(self, source_name: str, ok: bool) -> None: ... + def health(self, source_name: str) -> float: ... + @runtime_checkable class StructuredOutputStrategy(Protocol): diff --git a/src/polygateway/sources.py b/src/polygateway/sources.py index baf1a85..ca24736 100644 --- a/src/polygateway/sources.py +++ b/src/polygateway/sources.py @@ -63,6 +63,10 @@ class HealthAwareSelector: prev = self._ewma.get(source_name, 1.0) self._ewma[source_name] = prev + _EWMA_ALPHA * ((1.0 if ok else 0.0) - prev) + def health(self, source_name: str) -> float: + """EWMA 裸值(OutcomeAwareSelector 端口): RetryMW 降权门槛用。""" + return self._ewma.get(source_name, 1.0) + def _score(self, name: str, stats: dict[str, SourceStats]) -> float: ewma = max(self._ewma.get(name, 1.0), _SCORE_FLOOR) inflight = stats[name].inflight if name in stats else 0 diff --git a/tests/unit/test_health_selector.py b/tests/unit/test_health_selector.py index d0d9ce2..c4402e8 100644 --- a/tests/unit/test_health_selector.py +++ b/tests/unit/test_health_selector.py @@ -69,6 +69,13 @@ class TestHealthAwareSelector: assert order[0].name == "s1" # 两候选中 s1 胜出 assert order[-1].name == "s3" # 塌陷源垫底 + def test_health_exposes_raw_ewma(self): + sel = HealthAwareSelector(rng=lambda: 0.0) + assert sel.health("s1") == pytest.approx(1.0) # 乐观初始 + for _ in range(3): + sel.record_outcome("s1", ok=False) + assert sel.health("s1") == pytest.approx(1.0 * 0.8**3) + def test_missing_stats_defaults_to_zero_inflight(self): sel = HealthAwareSelector(rng=lambda: 0.0) assert [s.name for s in sel.order([_src("s1")], {})] == ["s1"] diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 2277a39..81165f0 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -370,11 +370,17 @@ class RecordingSelector(StaticSelector): def record_outcome(self, source_name, ok): self.outcomes.append((source_name, ok)) + def health(self, source_name): + return 1.0 + class ExplodingSelector(StaticSelector): def record_outcome(self, source_name, ok): raise RuntimeError("sink boom") + def health(self, source_name): + return 1.0 + class TestM25Orchestration: """M2.5 设计 §3.3: 调用内失败降权 + 健康喂数(先红后绿)。""" @@ -491,3 +497,43 @@ class TestAdaptivePacing: await mw(_REQ) assert pacer._inflight.get("a", 0) == 0 assert pacer._inflight.get("b", 0) == 0 + + +class HealthySink(StaticSelector): + """带健康视图的选源器桩(OutcomeAwareSelector 全量实现)。""" + + def __init__(self, health): + self._health = health + self.outcomes = [] + + def record_outcome(self, source_name, ok): + self.outcomes.append((source_name, ok)) + + def health(self, source_name): + return self._health.get(source_name, 1.0) + + +class TestHealthGatedDemotion: + """迭代 2(设计 §3.36): 降权需可信替代,否则原地重试。""" + + async def test_no_credible_alternative_stays_on_healthy(self): + # 替补健康分 0.08 < 0.5×0.9 → 不让位,第三次仍打 a + sel = HealthySink({"a": 0.9, "b": 0.08}) + mw, _, _, transport, _, _ = _harness( + [_src("a"), _src("b")], + [TransientError("1"), TransientError("2"), _ok()], + selector=sel, + ) + resp = await mw(_REQ) + assert [n for n, _ in transport.calls] == ["a", "a", "a"] + assert resp.source_name == "a" + + async def test_credible_alternative_still_yields(self): + sel = HealthySink({"a": 0.9, "b": 0.9}) + mw, _, _, transport, _, _ = _harness( + [_src("a"), _src("b")], + [TransientError("1"), TransientError("2"), _ok()], + selector=sel, + ) + await mw(_REQ) + assert [n for n, _ in transport.calls] == ["a", "a", "b"]