diff --git a/src/polygateway/middleware/admission.py b/src/polygateway/middleware/admission.py index 8997784..ca71563 100644 --- a/src/polygateway/middleware/admission.py +++ b/src/polygateway/middleware/admission.py @@ -224,28 +224,66 @@ class SourceAdmission: async def on_no_runnable( self, gate_rejections: int, reasons: dict[str, str], clock: StallClock ) -> None: - """一个源都挑不出来时的处置: 判死或等待一轮再来。""" + """一个源都挑不出来时的处置: **按拒绝原因分派**到各自的策略。 + + 分派而非串行是硬要求(issue #14): 串行写法下 `circuit_open=wait` 不抛 + 之后会径直掉进配额分支,`quota_full=fail_fast` 的调用方于是收到一个 + `reason=quota_exhausted` 的异常——而配额其实是满的,坏的是熔断门。 + """ + names = tuple(s.name for s in self._sources) if gate_rejections == len(self._sources): - names = tuple(s.name for s in self._sources) - raise CircuitOpenError( - scope=self._scope, - retry_after_s=await self._breaker.retry_after_s(names), - per_source_reasons=reasons, - ) - if self._quota_full == "fail_fast": - raise AllSourcesExhausted( - scope=self._scope, - reason="quota_exhausted", - retry_after_s=self._bp.poll_interval_s, - per_source_reasons=reasons, - ) + # 全部因熔断类原因(门开路 / 本地冷却备忘)被拒 + if self._circuit_open == "fail_fast": + raise CircuitOpenError( + scope=self._scope, + retry_after_s=await self._breaker.retry_after_s(names), + per_source_reasons=reasons, + ) + # wait: 保护作用完整保留(这一轮照样一个请求都不发),改变的只是 + # 调用方当场死还是排队等——多源可换源故 fail-fast 对,单源无源可换 + hint = await self._breaker.retry_after_s(names) + else: + # 至少一个源是被配额/AIMD 挡的,归 quota_full 管 + if self._quota_full == "fail_fast": + raise AllSourcesExhausted( + scope=self._scope, + reason="quota_exhausted", + retry_after_s=self._bp.poll_interval_s, + per_source_reasons=reasons, + ) + hint = 0.0 if await self.stalled(clock): - names = tuple(s.name for s in self._sources) raise AllSourcesExhausted( scope=self._scope, reason="stalled", retry_after_s=await self._breaker.retry_after_s(names), per_source_reasons=reasons, ) - # jitter ∈ [0.5p, 1.0p] 防惊群(CHS governance.py:283-285) - await self._sleep(self._bp.poll_interval_s * (0.5 + 0.5 * self._rng())) + nap = self._nap(hint, clock) + if hint > 0: + logger.info( + "熔断开路等待 {:.1f}s 后重试(scope={}, 原因={})", nap, self._scope, reasons + ) + await self._sleep(nap) + + def _nap(self, hint: float, clock: StallClock) -> float: + """本轮等待多久。**必须在 `stalled()` 判定之后调用**(预算可能已耗尽)。 + + `hint > 0`(熔断开路有确定的冷却截止)时睡到那个时刻,而不是按 + `poll_interval` 空转——60 秒冷却用 10ms 轮询是 6000 次空转,内存后端 + 只是查字典,Redis 后端则是 6000 次往返 × 每个在途调用。抖动**上**加 + 而非缩放(既有 quota 路径是 `[0.5p, 1.0p]`): 对一个确定的截止时刻提前 + 醒来必然被再拒一次,白跑一趟。 + + 两档都夹到剩余 stall 预算,故单次调用的最坏墙钟是 `stall_window_s` + 加一个 poll 间隔,不随 `max_cooldown_s` 漂移。多加的那一格是因为 + `stalled()` 判据是 `>` 而非 `>=`——恰好睡到窗口边界不判死,留这一格 + 让下一轮必定判死。`hint == 0` 时整个式子退化为既有的 jitter 轮询。 + """ + jitter = self._bp.poll_interval_s * (0.5 + 0.5 * self._rng()) + budget = self._bp.stall_window_s - clock.stalled_s() + self._bp.poll_interval_s + wait = hint + jitter if hint > 0 else jitter + # 下界取 jitter 而非 poll_interval: 既有 quota 轮询是 [0.5p, 1.0p],用 + # poll_interval 兜底会把 rng→0 那半边抬上去。预算为负时(本地已超窗但 + # 全局仍在出餐,故 stalled() 不判死)靠它退回正常轮询节奏,不忙循环。 + return max(jitter, min(wait, budget)) diff --git a/tests/unit/test_backpressure.py b/tests/unit/test_backpressure.py index a8d1d1a..df2e858 100644 --- a/tests/unit/test_backpressure.py +++ b/tests/unit/test_backpressure.py @@ -13,6 +13,7 @@ from polygateway.backends.memory.breaker import InMemoryGate from polygateway.backends.memory.limiter import InMemoryLimiter from polygateway.errors import ( AllSourcesExhausted, + CircuitOpenError, GatewayUnavailableError, GovernanceBackendError, SourceNotConfiguredError, @@ -62,6 +63,7 @@ def _mw( sleep, rng=lambda: 0.0, quota_full="wait", + circuit_open="fail_fast", gate=None, transport=None, emitter=None, @@ -76,6 +78,7 @@ def _mw( retry=RetryPolicy(max_attempts=3, backoff_base_s=2.0, backoff_max_s=30.0), backpressure=BackpressurePolicy(stall_window_s=_STALL, poll_interval_s=0.01), quota_full=quota_full, + circuit_open=circuit_open, cooldown_memo=SourceCooldownMemo(now=clock), emitter=emitter, now=clock, @@ -593,3 +596,143 @@ class TestGateFailuresReachCallersAsScopeLevel: await QuotaGate(_Broken(), scope="LLM").progress_age_s() assert ei.value.scope == "llm" assert ei.value.reason == "governance_backend_down" + + +class TestCircuitOpenPolicy: + """issue #14: 熔断全拒时是当场判死还是等冷却过去。 + + 缺省 fail_fast 即历史行为(TestStallQuadrants 等既有用例照旧覆盖); + 本类钉的是 wait 档,以及两条策略互不串线。 + """ + + @staticmethod + async def _opened_gate(clock, cfg=_BREAKER): + gate = InMemoryGate(config=cfg, now=clock) + for _ in range(cfg.fail_threshold): + entry = await gate.try_enter("s1", "w") + await gate.record_failure(entry, "network_error", False) + return gate + + @staticmethod + def _free_limiter(clock, src): + return InMemoryLimiter( + scope="llm", + sources={"s1": src}, + global_limits=_NO_GLOBAL, + lease_ttl_s=10_000.0, + now=clock, + ) + + async def test_fail_fast_is_the_default(self): + """缺省档逐字保持历史行为: 全源开路当场抛 CircuitOpenError。""" + clock = FakeClock() + src = make_source() + mw = _mw( + [src], self._free_limiter(clock, src), [], clock=clock, + sleep=BoundedSleep(), gate=await self._opened_gate(clock), + ) + with pytest.raises(CircuitOpenError) as ei: + await mw(_REQ) + assert ei.value.reason == "circuit_open" + + async def test_wait_sleeps_out_the_cooldown_instead_of_dying(self): + """wait 档: 睡到冷却结束再来一轮,拿到探针后正常返回。 + + 睡的是**冷却剩余**而不是 poll_interval——60 秒冷却用 10ms 轮询要空转 + 6000 次,memory 后端只是查字典,Redis 后端则是 6000 次往返 × 每个在途调用。 + """ + clock = FakeClock() + src = make_source() + sleep = BoundedSleep() + + async def advance(_n): + clock.advance(sleep.delays[-1]) + + sleep._side_effect = advance + mw = _mw( + [src], self._free_limiter(clock, src), [_ok()], clock=clock, + sleep=sleep, gate=await self._opened_gate(clock), circuit_open="wait", + ) + resp = await mw(_REQ) + assert resp.content == "ok" + # 一觉睡到冷却结束(jitter 上加,rng=0 → +0.5×poll),不是 poll 空转 + assert sleep.delays[0] == pytest.approx(_BREAKER.cooldown_s + 0.005) + + async def test_wait_does_not_leak_into_the_quota_branch(self): + """两条策略互不串线: circuit_open=wait 配 quota_full=fail_fast 时, + 熔断等待**不得**被当成配额耗尽上报——串线会让调用方拿到一个 + reason=quota_exhausted 的异常,而配额其实是满的。""" + clock = FakeClock() + src = make_source() + sleep = BoundedSleep() + + async def advance(_n): + clock.advance(sleep.delays[-1]) + + sleep._side_effect = advance + mw = _mw( + [src], self._free_limiter(clock, src), [_ok()], clock=clock, sleep=sleep, + gate=await self._opened_gate(clock), quota_full="fail_fast", circuit_open="wait", + ) + assert (await mw(_REQ)).content == "ok" + + async def test_wait_still_dies_when_cooldown_outlasts_the_stall_budget(self): + """等待有可解释的上界: 冷却比 stall 预算还长时,在窗口耗尽处判死。 + + 单次睡眠夹到剩余 stall 预算,故最坏墙钟 = stall_window + 一个 poll, + 不随 max_cooldown_s 漂移。 + """ + clock = FakeClock() + src = make_source() + long_cooldown = BreakerConfig( + fail_threshold=3, cooldown_s=1000.0, probe_ttl_s=2000.0, max_cooldown_s=1000.0 + ) + sleep = BoundedSleep() + + async def advance(_n): + clock.advance(sleep.delays[-1]) + + sleep._side_effect = advance + mw = _mw( + [src], self._free_limiter(clock, src), [], clock=clock, sleep=sleep, + gate=await self._opened_gate(clock, long_cooldown), circuit_open="wait", + ) + with pytest.raises(AllSourcesExhausted) as ei: + await mw(_REQ) + assert ei.value.reason == "stalled" + assert ei.value.per_source_reasons == {"s1": "circuit_open"} + assert sleep.delays[0] == pytest.approx(_STALL + 0.01) # 夹到预算 + 一个 poll + + async def test_wait_loop_stays_cancellable(self): + """取消穿透(铁律): 熔断等待中的取消不得被吞。""" + clock = FakeClock() + src = make_source() + mw = _mw( + [src], self._free_limiter(clock, src), [], clock=clock, + sleep=asyncio.sleep, gate=await self._opened_gate(clock), circuit_open="wait", + ) + task = asyncio.create_task(mw(_REQ)) + await asyncio.sleep(0.03) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + async def test_half_open_rejection_does_not_blacklist_a_recovered_source(self): + """issue #14 §1.3 回归: 探针成功后本进程立即可再选该源。 + + 此前 HALF_OPEN 拒绝把探针租约(派生自 2 × timeout,现场 600s)写进冷却 + 备忘,而 `set_until` 取更晚者、不可回退——门恢复 CLOSED 之后本进程仍 + 跳过该源整整一个租约,单源下每次调用照旧判死。多源部署同样中招,只是 + 被别的源接住流量掩盖了。 + """ + clock = FakeClock() + cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=600.0) + gate = await self._opened_gate(clock, cfg) + memo = SourceCooldownMemo(now=clock) + clock.advance(cfg.cooldown_s + 1) + probe = await gate.try_enter("s1", "w1") + blocked = await gate.try_enter("s1", "w2") # 并发调用撞上在途探针 + assert not blocked.allowed + memo.set_until("s1", clock() + blocked.retry_after_s) # 准入路径的写法 + await gate.record_success(probe) # 探针成功 → 门恢复 CLOSED + assert not memo.active("s1")