iomgaa
a57a5cea72
Independent verification caught that the split shipped in the previous commit did not actually hold on the only path production uses. The gate wrappers re-raise GovernanceBackendError but nothing else, so SourceNotConfiguredError fell into the following `except Exception` and came back out as a governance_backend_down failure with retry_after_s=5.0. A misconfigured source name would still retry forever and never surface. The existing tests missed it because both of them call the private _cfg() directly, one layer below the wrapper the governance loops actually go through. The regression test goes through QuotaGate. telemetry.py has to widen its terminal catch in the same commit: once the wrapper stops relabeling the error, it is no longer a GovernanceBackendError, and it is raised before any attempt exists, so the path would have recorded no telemetry at all. Also corrects the leak path count from three to five. QuotaGate.stats and BreakerGate.retry_after_s are not wrapped by _record_quietly either.
348 lines
14 KiB
Python
348 lines
14 KiB
Python
"""背压 stall 双条件判死 + poll jitter + 记账侧降级(M2 设计 §4/§10)。
|
|
|
|
保真蓝本 CHS governance.py:200-285: local_waited(本地 monotonic,调用级
|
|
累计不重置)与 progress_age(全局活性)**同时**超 stall_window 才判死;
|
|
poll 间隔带 [0.5p, 1.0p] jitter 防惊群。
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from polygateway.backends.memory.breaker import InMemoryGate
|
|
from polygateway.backends.memory.limiter import InMemoryLimiter
|
|
from polygateway.errors import (
|
|
AllSourcesExhausted,
|
|
GatewayUnavailableError,
|
|
GovernanceBackendError,
|
|
SourceNotConfiguredError,
|
|
TransientError,
|
|
)
|
|
from polygateway.middleware.ratelimit import QuotaGate
|
|
from polygateway.middleware.retry import RetryMW, backoff_delay
|
|
from polygateway.sources import RoundRobinSelector, SourceCooldownMemo
|
|
from polygateway.types import (
|
|
BackpressurePolicy,
|
|
BreakerConfig,
|
|
ChatRequest,
|
|
GlobalLimits,
|
|
RetryPolicy,
|
|
)
|
|
from tests.contracts.conftest import FakeClock, make_source
|
|
from tests.unit.test_retry import FakeTransport, _ok
|
|
|
|
_BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
|
|
_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
|
|
_REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}])
|
|
_STALL = 300.0
|
|
|
|
|
|
class BoundedSleep:
|
|
"""每次 sleep 执行注入的副作用;超过上限仍在轮询 = 判死逻辑失效,炸出而非死循环。"""
|
|
|
|
def __init__(self, side_effect=None, limit: int = 10):
|
|
self.delays: list[float] = []
|
|
self._side_effect = side_effect
|
|
self._limit = limit
|
|
|
|
async def __call__(self, seconds: float) -> None:
|
|
self.delays.append(seconds)
|
|
if len(self.delays) > self._limit:
|
|
raise RuntimeError(f"超过 {self._limit} 次轮询仍未判死/未获 permit")
|
|
if self._side_effect is not None:
|
|
await self._side_effect(len(self.delays))
|
|
|
|
|
|
def _mw(sources, limiter, script, *, clock, sleep, rng=lambda: 0.0, quota_full="wait", gate=None):
|
|
return RetryMW(
|
|
scope="llm",
|
|
sources=sources,
|
|
selector=RoundRobinSelector(),
|
|
limiter=limiter,
|
|
gate=gate or InMemoryGate(config=_BREAKER, now=clock),
|
|
transport=FakeTransport(script),
|
|
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,
|
|
cooldown_memo=SourceCooldownMemo(now=clock),
|
|
emitter=None,
|
|
now=clock,
|
|
sleep=sleep,
|
|
rng=rng,
|
|
)
|
|
|
|
|
|
def _blocked_limiter(clock):
|
|
"""单并发源被外部占满 → RetryMW 进入 wait 轮询分支。"""
|
|
src = make_source(max_concurrency=1)
|
|
limiter = InMemoryLimiter(
|
|
scope="llm",
|
|
sources={"s1": src},
|
|
global_limits=_NO_GLOBAL,
|
|
lease_ttl_s=10_000.0,
|
|
now=clock,
|
|
)
|
|
return src, limiter
|
|
|
|
|
|
class TestStallQuadrants:
|
|
async def test_both_windows_exceeded_raises_stalled(self):
|
|
"""双超窗: 本地等待与全局无进展同时 > stall_window → stalled。"""
|
|
clock = FakeClock()
|
|
src, limiter = _blocked_limiter(clock)
|
|
_held = await limiter.try_acquire("s1", 0)
|
|
|
|
async def advance(_n):
|
|
clock.advance(_STALL + 100)
|
|
|
|
mw = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(advance))
|
|
with pytest.raises(AllSourcesExhausted) as ei:
|
|
await mw(_REQ)
|
|
assert ei.value.reason == "stalled"
|
|
assert ei.value.per_source_reasons == {"s1": "rate_limited"}
|
|
|
|
async def test_local_exceeded_but_global_fresh_keeps_waiting(self):
|
|
"""仅本地超窗: 别人一直在出餐 → 不判死,等到 permit 后正常成功。"""
|
|
clock = FakeClock()
|
|
src, limiter = _blocked_limiter(clock)
|
|
held = await limiter.try_acquire("s1", 0)
|
|
|
|
async def advance_and_feed(n):
|
|
clock.advance(_STALL + 100) # 本地远超窗
|
|
await limiter.mark_progress() # 但全局刚出过餐
|
|
if n >= 3:
|
|
await held.release() # 第 3 轮让出 permit
|
|
|
|
mw = _mw([src], limiter, [_ok()], clock=clock, sleep=BoundedSleep(advance_and_feed))
|
|
resp = await mw(_REQ)
|
|
assert resp.content == "ok"
|
|
|
|
async def test_global_stale_but_local_fresh_keeps_waiting(self):
|
|
"""仅全局超窗(从未出餐 age=inf): 本地才刚开始等 → 不判死。"""
|
|
clock = FakeClock()
|
|
src, limiter = _blocked_limiter(clock)
|
|
held = await limiter.try_acquire("s1", 0)
|
|
|
|
async def release_soon(n):
|
|
if n >= 2: # 本地累计 poll 极短,未超窗
|
|
await held.release()
|
|
|
|
mw = _mw([src], limiter, [_ok()], clock=clock, sleep=BoundedSleep(release_soon))
|
|
resp = await mw(_REQ)
|
|
assert resp.content == "ok"
|
|
|
|
async def test_poll_jitter_bounds(self):
|
|
"""wait 轮询间隔 ∈ [0.5p, 1.0p](CHS governance.py:283-285)。"""
|
|
clock = FakeClock()
|
|
src, limiter = _blocked_limiter(clock)
|
|
held = await limiter.try_acquire("s1", 0)
|
|
|
|
async def release_soon(n):
|
|
if n >= 2:
|
|
await held.release()
|
|
|
|
for rng_v, expected in ((0.0, 0.005), (1.0, 0.01)):
|
|
clock2 = FakeClock()
|
|
src2, limiter2 = _blocked_limiter(clock2)
|
|
held2 = await limiter2.try_acquire("s1", 0)
|
|
|
|
async def release2(n, _h=held2):
|
|
if n >= 2:
|
|
await _h.release()
|
|
|
|
sleep = BoundedSleep(release2)
|
|
mw = _mw([src2], limiter2, [_ok()], clock=clock2, sleep=sleep, rng=lambda v=rng_v: v)
|
|
await mw(_REQ)
|
|
assert sleep.delays[0] == pytest.approx(expected)
|
|
|
|
async def test_fail_fast_unaffected(self):
|
|
"""fail_fast 路径回归: 不进入 stall 判定,立即 quota_exhausted。"""
|
|
clock = FakeClock()
|
|
src, limiter = _blocked_limiter(clock)
|
|
_held = await limiter.try_acquire("s1", 0)
|
|
mw = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), quota_full="fail_fast")
|
|
with pytest.raises(AllSourcesExhausted) as ei:
|
|
await mw(_REQ)
|
|
assert ei.value.reason == "quota_exhausted"
|
|
|
|
async def test_cancellation_pierces_wait_loop(self):
|
|
"""stall 等待中的取消穿透(铁律): sleep 可取消,任务立即终止。"""
|
|
clock = FakeClock()
|
|
src, limiter = _blocked_limiter(clock)
|
|
_held = await limiter.try_acquire("s1", 0)
|
|
mw = _mw([src], limiter, [], clock=clock, sleep=asyncio.sleep)
|
|
task = asyncio.create_task(mw(_REQ))
|
|
await asyncio.sleep(0.03)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
|
|
|
|
class _GateSuccessBroken(InMemoryGate):
|
|
async def record_success(self, entry):
|
|
raise GovernanceBackendError("redis 抖动", scope="llm")
|
|
|
|
|
|
class _GateFailureBroken(InMemoryGate):
|
|
async def record_failure(self, entry, reason, force_open):
|
|
raise GovernanceBackendError("redis 抖动", scope="llm")
|
|
|
|
|
|
class _LimiterProgressBroken(InMemoryLimiter):
|
|
async def mark_progress(self):
|
|
raise GovernanceBackendError("redis 抖动", scope="llm")
|
|
|
|
|
|
class TestAccountingDegradation:
|
|
"""记账侧降级(设计 §10,ARCH §7.3 勘误): 调用已完成,写回失败不冒泡。"""
|
|
|
|
async def test_record_success_failure_does_not_lose_response(self):
|
|
clock = FakeClock()
|
|
src = make_source()
|
|
limiter = InMemoryLimiter(
|
|
scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock
|
|
)
|
|
gate = _GateSuccessBroken(config=_BREAKER, now=clock)
|
|
mw = _mw([src], limiter, [_ok()], clock=clock, sleep=BoundedSleep(), gate=gate)
|
|
resp = await mw(_REQ)
|
|
assert resp.content == "ok" # 真实成功响应不因记账失败被丢弃
|
|
|
|
async def test_mark_progress_failure_does_not_lose_response(self):
|
|
clock = FakeClock()
|
|
src = make_source()
|
|
limiter = _LimiterProgressBroken(
|
|
scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock
|
|
)
|
|
mw = _mw([src], limiter, [_ok()], clock=clock, sleep=BoundedSleep())
|
|
resp = await mw(_REQ)
|
|
assert resp.content == "ok"
|
|
|
|
async def test_record_failure_failure_does_not_mask_retry(self):
|
|
"""失败记账挂掉 → 原始尝试异常不被掩盖,重试照常换发并成功。"""
|
|
clock = FakeClock()
|
|
src = make_source()
|
|
limiter = InMemoryLimiter(
|
|
scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock
|
|
)
|
|
gate = _GateFailureBroken(config=_BREAKER, now=clock)
|
|
script = [TransientError("boom", status_code=500), _ok()]
|
|
mw = _mw([src], limiter, script, clock=clock, sleep=BoundedSleep(), gate=gate)
|
|
resp = await mw(_REQ)
|
|
assert resp.content == "ok"
|
|
|
|
|
|
class TestBackoffExtraction:
|
|
"""T5 提取的模块级纯函数(T9 Embedding 复用)。"""
|
|
|
|
def test_formula_matches_policy(self):
|
|
policy = RetryPolicy(max_attempts=3, backoff_base_s=2.0, backoff_max_s=30.0)
|
|
exc = TransientError("x", status_code=500)
|
|
assert backoff_delay(policy, 1, exc, lambda: 0.0) == pytest.approx(1.0) # 2*0.5
|
|
assert backoff_delay(policy, 1, exc, lambda: 1.0) == pytest.approx(3.0) # 2*1.5
|
|
assert backoff_delay(policy, 10, exc, lambda: 1.0) == pytest.approx(45.0) # 封顶 30*1.5
|
|
|
|
def test_retry_after_hint_wins_when_larger(self):
|
|
policy = RetryPolicy(max_attempts=3, backoff_base_s=2.0, backoff_max_s=30.0)
|
|
exc = TransientError("x", status_code=429, retry_after_s=7.5)
|
|
assert backoff_delay(policy, 1, exc, lambda: 0.0) == pytest.approx(7.5)
|
|
|
|
|
|
class TestQuotaGateProgressAge:
|
|
async def test_passthrough_and_wrap(self):
|
|
from polygateway.middleware.ratelimit import QuotaGate
|
|
|
|
class _L:
|
|
async def progress_age_s(self):
|
|
return 12.5
|
|
|
|
class _Broken:
|
|
async def progress_age_s(self):
|
|
raise OSError("down")
|
|
|
|
assert await QuotaGate(_L(), scope="llm").progress_age_s() == 12.5
|
|
with pytest.raises(GovernanceBackendError):
|
|
await QuotaGate(_Broken(), scope="llm").progress_age_s()
|
|
|
|
|
|
class TestUnknownSourceIsAssemblyDefect:
|
|
"""未知源 = 限流后端的源名单与治理循环对不上,是装配缺陷不是后端故障。
|
|
|
|
两个后端行为必须一致(Redis 版对应用例在 `test_redis_key_layout.py::
|
|
TestConversions::test_unknown_source_rejected`);内存版此前无覆盖,
|
|
该分支从未被测过(issue #7 §3.4)。
|
|
"""
|
|
|
|
def test_memory_limiter_rejects_unknown_source(self):
|
|
limiter = InMemoryLimiter(
|
|
scope="llm", sources={"s1": make_source("s1")}, global_limits=_NO_GLOBAL
|
|
)
|
|
with pytest.raises(SourceNotConfiguredError) as ei:
|
|
limiter._cfg("nope")
|
|
# 关键: 若归入 scope 级家族,配置写错的任务会永远延期重投、永不进死信
|
|
assert not isinstance(ei.value, GatewayUnavailableError)
|
|
|
|
@pytest.mark.parametrize("method", ["try_acquire", "stats"])
|
|
async def test_survives_the_quota_gate_wrapper(self, method):
|
|
"""必须穿透 QuotaGate,否则整个拆分在生产路径上等于没做。
|
|
|
|
上面两条(以及 redis 版)打的都是私有 `_cfg`,绕过了包装器。而治理循环
|
|
只经 QuotaGate 访问后端,包装器的 `except Exception` 会把装配缺陷重新
|
|
包成 `GovernanceBackendError`——下游又拿到可重投异常,永远重投不告警。
|
|
"""
|
|
src = make_source("s1")
|
|
# 限流后端的源名单与治理循环拿到的源对不上 = 装配缺陷
|
|
limiter = InMemoryLimiter(
|
|
scope="llm", sources={"other": src}, global_limits=_NO_GLOBAL
|
|
)
|
|
gate = QuotaGate(limiter, scope="llm")
|
|
with pytest.raises(SourceNotConfiguredError) as ei:
|
|
await getattr(gate, method)(src)
|
|
assert not isinstance(ei.value, GatewayUnavailableError)
|
|
|
|
|
|
class TestGateFailuresReachCallersAsScopeLevel:
|
|
"""三条闸门泄漏路径必须以 scope 级不可用的形态到达调用方(issue #7)。
|
|
|
|
记账路径由 `_record_quietly` 降级为 warning,但闸门路径没有那层包裹,会一路
|
|
抛给调用方。只写 `except GatewayUnavailableError` 的调用方此前接不住,后果
|
|
是 Redis 抖一下就让积压任务烧掉业务失败预算进死信——而那是运维重启即可恢复
|
|
的故障。三条路径逐一钉住,防止将来任何一条被漏掉。
|
|
"""
|
|
|
|
async def test_try_acquire_failure_is_scope_level(self):
|
|
from polygateway.middleware.ratelimit import QuotaGate
|
|
|
|
class _Broken:
|
|
async def try_acquire(self, name, est):
|
|
raise OSError("down")
|
|
|
|
with pytest.raises(GatewayUnavailableError) as ei:
|
|
await QuotaGate(_Broken(), scope="LLM").try_acquire(make_source("s1"))
|
|
assert ei.value.scope == "llm"
|
|
assert ei.value.reason == "governance_backend_down"
|
|
assert ei.value.retry_after_s > 0 # 0 会让积压任务零延迟冲击已挂的后端
|
|
|
|
async def test_try_enter_failure_is_scope_level(self):
|
|
from polygateway.middleware.breaker import BreakerGate
|
|
|
|
class _Broken:
|
|
async def try_enter(self, name, owner):
|
|
raise OSError("down")
|
|
|
|
with pytest.raises(GatewayUnavailableError) as ei:
|
|
await BreakerGate(_Broken(), scope="LLM").try_enter(make_source("s1"), "owner")
|
|
assert ei.value.scope == "llm"
|
|
assert ei.value.reason == "governance_backend_down"
|
|
|
|
async def test_progress_age_failure_is_scope_level(self):
|
|
from polygateway.middleware.ratelimit import QuotaGate
|
|
|
|
class _Broken:
|
|
async def progress_age_s(self):
|
|
raise OSError("down")
|
|
|
|
with pytest.raises(GatewayUnavailableError) as ei:
|
|
await QuotaGate(_Broken(), scope="LLM").progress_age_s()
|
|
assert ei.value.scope == "llm"
|
|
assert ei.value.reason == "governance_backend_down"
|