258 lines
9.9 KiB
Python
258 lines
9.9 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, GovernanceBackendError, TransientError
|
|
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 抖动")
|
|
|
|
|
|
class _GateFailureBroken(InMemoryGate):
|
|
async def record_failure(self, entry, reason, force_open):
|
|
raise GovernanceBackendError("redis 抖动")
|
|
|
|
|
|
class _LimiterProgressBroken(InMemoryLimiter):
|
|
async def mark_progress(self):
|
|
raise GovernanceBackendError("redis 抖动")
|
|
|
|
|
|
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()).progress_age_s() == 12.5
|
|
with pytest.raises(GovernanceBackendError):
|
|
await QuotaGate(_Broken()).progress_age_s()
|