diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 7fbe421..b4941a9 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -11,6 +11,7 @@ httpx 是库的核心依赖而非实现层内部件,不违反"middleware 只依 from __future__ import annotations import asyncio +import contextlib import random import time import uuid @@ -39,7 +40,7 @@ from polygateway.streaming import StreamLivenessTimeout from polygateway.types import LLMResponse if TYPE_CHECKING: - from collections.abc import Awaitable, Callable + from collections.abc import AsyncIterator, Awaitable, Callable from polygateway.ports import ( GateDecision, @@ -74,6 +75,44 @@ def backoff_delay( return max(delay, retry_after) +class StallClock: + """调用级 stall 计时器: 只累计非生产性等待(issue #8 设计 §3.1)。 + + stall 预算治理的是"无人治理的等待"——429 退避、配额 wait 轮询、熔断冷却; + 真实尝试的耗时已由重试预算 `max_attempts` 治理,必须从 stall 账里扣除。 + 两者重叠计费正是 issue #8 的根因: stall 预算(默认 300s)小于重试预算 + (3 × timeout_s),必然先耗尽,于是重试预算在超时场景下永远用不上。 + + "生产性"的边界即 `_attempt` 的边界,含该次尝试的记账与遥测收尾——它们是 + "尝试已有结论"之后的动作,不是在等待重试机会;把它们计入 stall 会让遥测 + 抖动参与判死。 + + 每次调用创建一个实例。严禁提升为实例属性: 并发调用共享会互相污染计时。 + 模块级共享单元, EmbeddingClient 与 OcrClient 复用(同 `backoff_delay`)。 + """ + + __slots__ = ("_now", "_entered_at", "_productive_s") + + def __init__(self, now: Callable[[], float]) -> None: + self._now = now + self._entered_at = now() + self._productive_s = 0.0 + + def stalled_s(self) -> float: + """非生产性等待累计秒数 = 调用总耗时 - 真实尝试耗时。""" + return self._now() - self._entered_at - self._productive_s + + @contextlib.asynccontextmanager + async def attempting(self) -> AsyncIterator[None]: + """包裹一次真实尝试,其耗时记为生产性。""" + started = self._now() + try: + yield + finally: + # 只做算术, 不吞任何异常——CancelledError 逐字穿透(库铁律) + self._productive_s += self._now() - started + + def _demote_call_failures( ordered: list[SourceConfig], attempt_fails: dict[str, int], @@ -209,12 +248,14 @@ class RetryMW: reasons: dict[str, str] = {} # 调用内失败计数(设计 §3.3): 局部状态,调用结束即弃;严禁实例属性(并发共享) attempt_fails: dict[str, int] = {} - entered_at = self._now() # 调用级累计计时,循环内不重置(CHS governance.py:207) + # 调用级累计计时,循环内不重置(CHS governance.py:207);issue #8 起只计 + # 非生产性等待——真实尝试由重试预算治理,不再重复烧 stall 预算 + clock = StallClock(self._now) while True: # 调用级时间上限(迭代 5): 429 免预算后的兜底,防饱和期无限循环。 # 与 _on_no_runnable 同款双条件(CHS 口径): 本地超窗且全局无进展才判死 stall = self._bp.stall_window_s - if self._now() - entered_at > stall and await self._quota.progress_age_s() > stall: + if clock.stalled_s() > stall and await self._quota.progress_age_s() > stall: raise AllSourcesExhausted( scope=self._scope, reason="stalled", @@ -223,9 +264,10 @@ class RetryMW: ) picked, gate_rejections = await self._pick_runnable(reasons, attempt_fails) if picked is None: - await self._on_no_runnable(gate_rejections, reasons, entered_at) + await self._on_no_runnable(gate_rejections, reasons, clock) continue - outcome = await self._attempt(request, *picked, reasons, attempt_fails) + async with clock.attempting(): + outcome = await self._attempt(request, *picked, reasons, attempt_fails) if isinstance(outcome, LLMResponse): return outcome # 429 = 服务端调度指令(gRPC pushback 语义,迭代 5): 按 Retry-After @@ -284,7 +326,7 @@ class RetryMW: return None, gate_rejections async def _on_no_runnable( - self, gate_rejections: int, reasons: dict[str, str], entered_at: float + self, gate_rejections: int, reasons: dict[str, str], clock: StallClock ) -> None: if gate_rejections == len(self._sources): names = tuple(s.name for s in self._sources) @@ -303,7 +345,7 @@ class RetryMW: # 双条件 stall 判死(CHS governance.py:270-281): 本地累计等待与全局 # 无进展**同时**超窗才判死——本地 monotonic 与后端时钟刻意不混用。 stall = self._bp.stall_window_s - if self._now() - entered_at > stall and await self._quota.progress_age_s() > stall: + if clock.stalled_s() > stall and await self._quota.progress_age_s() > stall: names = tuple(s.name for s in self._sources) raise AllSourcesExhausted( scope=self._scope, diff --git a/tests/unit/test_backpressure.py b/tests/unit/test_backpressure.py index 302de45..4bd4ee2 100644 --- a/tests/unit/test_backpressure.py +++ b/tests/unit/test_backpressure.py @@ -53,19 +53,31 @@ class BoundedSleep: await self._side_effect(len(self.delays)) -def _mw(sources, limiter, script, *, clock, sleep, rng=lambda: 0.0, quota_full="wait", gate=None): +def _mw( + sources, + limiter, + script, + *, + clock, + sleep, + rng=lambda: 0.0, + quota_full="wait", + gate=None, + transport=None, + emitter=None, +): return RetryMW( scope="llm", sources=sources, selector=RoundRobinSelector(), limiter=limiter, gate=gate or InMemoryGate(config=_BREAKER, now=clock), - transport=FakeTransport(script), + transport=transport or 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, + emitter=emitter, now=clock, sleep=sleep, rng=rng, @@ -178,6 +190,169 @@ class TestStallQuadrants: await task +class ClockAdvancingTransport: + """按脚本 [(推进秒数, 动作), ...] 执行: 在一次尝试内部推进时钟, 模拟真实耗时。 + + 动作语义同 `FakeTransport`(异常即抛、"hang" 即挂起、其余为返回值)。 + stall 口径的关键区分在于"时间花在哪", 故必须能让时钟只在 transport 内前进。 + """ + + def __init__(self, script, clock): + self.script = list(script) + self.clock = clock + self.calls = [] + + async def complete(self, *, messages, source, stream, overlay, call_id): + self.calls.append((source.name, call_id)) + advance, action = self.script.pop(0) + self.clock.advance(advance) + if isinstance(action, Exception): + raise action + if action == "hang": + await asyncio.Event().wait() + return action + + +class _SlowEmitter: + """遥测收尾中推进时钟: 钉住"遥测耗时属生产性"(设计 §3.1 边界声明)。""" + + def __init__(self, clock, advance): + self._clock = clock + self._advance = advance + + async def emit_attempt(self, *args, **kwargs): + self._clock.advance(self._advance) + + +class TestStallBudget: + """stall 预算只计非生产性等待(issue #8 设计 §3.1)。 + + 根因是两个预算重叠计费: 真实尝试的耗时同时烧重试预算与 stall 预算, + 而 stall 预算更小必然先耗尽, 于是 max_attempts 在超时场景下永不生效。 + """ + + def _free_limiter(self, clock): + src = make_source() + limiter = InMemoryLimiter( + scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock + ) + return src, limiter + + async def test_single_timeout_does_not_exhaust_stall_budget(self): + """timeout_s == stall_window_s 时, 一次超时不得判死——重试预算须真实可用。""" + clock = FakeClock() + src, limiter = self._free_limiter(clock) + # 第一次尝试耗满 300s 超时后失败, 第二次立即成功 + transport = ClockAdvancingTransport( + [(_STALL + 1, TransientError("timeout", status_code=504)), (0.0, _ok())], clock + ) + mw = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), transport=transport) + resp = await mw(_REQ) + assert resp.content == "ok" + assert len(transport.calls) == 2 # 第二次尝试确实发出了 + + async def test_productive_time_excluded_from_stall(self): + """连续多次长尝试也不烧 stall 预算: 它们烧的是重试预算。""" + clock = FakeClock() + src, limiter = self._free_limiter(clock) + transport = ClockAdvancingTransport( + [ + (_STALL + 100, TransientError("slow", status_code=500)), + (_STALL + 100, TransientError("slow", status_code=500)), + (0.0, _ok()), + ], + clock, + ) + mw = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), transport=transport) + resp = await mw(_REQ) + assert resp.content == "ok" + + async def test_telemetry_time_counts_as_productive(self): + """遥测收尾属 `_attempt` 边界内: 遥测抖动不得参与判死(设计 §3.1)。 + + 必须走**失败**路径才有判别力: 成功后直接 return, 循环开头的 stall + 判定根本不会再执行。此处让首次尝试快速失败、而遥测收尾慢得超窗, + 下一轮循环开头即检验遥测耗时有没有被算进 stall 账。 + """ + clock = FakeClock() + src, limiter = self._free_limiter(clock) + transport = ClockAdvancingTransport( + [(0.1, TransientError("boom", status_code=500)), (0.0, _ok())], clock + ) + mw = _mw( + [src], + limiter, + [], + clock=clock, + sleep=BoundedSleep(), + transport=transport, + emitter=_SlowEmitter(clock, _STALL + 100), + ) + resp = await mw(_REQ) + assert resp.content == "ok" + + async def test_nonproductive_wait_still_triggers_stall(self): + """兜底未被削弱: 纯轮询等待累满窗口仍判死。""" + 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" + + async def test_saturation_429_still_stalls(self): + """429 免预算不烧 fails, 主循环兜底须仍能判死而非无限循环(设计 §3.5)。""" + clock = FakeClock() + src, limiter = self._free_limiter(clock) + # 429 往返本身极快(生产性可忽略), 退避 sleep 才是非生产性的大头 + transport = ClockAdvancingTransport( + [(0.1, TransientError("429", status_code=429)) for _ in range(10)], clock + ) + + async def advance(_n): + clock.advance(_STALL) + + mw = _mw( + [src], limiter, [], clock=clock, sleep=BoundedSleep(advance), transport=transport + ) + with pytest.raises(AllSourcesExhausted) as ei: + await mw(_REQ) + assert ei.value.reason == "stalled" # 不是 retry_exhausted: 429 确实没烧重试预算 + + async def test_cancel_inside_attempt_pierces(self): + """取消发生在 `attempting()` 包裹内仍逐字穿透(库铁律)。""" + clock = FakeClock() + src, limiter = self._free_limiter(clock) + transport = ClockAdvancingTransport([(0.0, "hang")], clock) + mw = _mw([src], limiter, [], clock=clock, sleep=asyncio.sleep, transport=transport) + task = asyncio.create_task(mw(_REQ)) + while not transport.calls: + await asyncio.sleep(0.01) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert (await limiter.source_stats("s1")).inflight == 0 # permit 在 finally 释放 + + async def test_concurrent_calls_do_not_share_clock(self): + """StallClock 必须是调用级局部状态: 一路长尝试不得污染另一路的 stall 账。""" + clock = FakeClock() + src = make_source(max_concurrency=2) + limiter = InMemoryLimiter( + scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock + ) + slow = ClockAdvancingTransport([(_STALL + 100, _ok("slow"))], clock) + fast = ClockAdvancingTransport([(0.0, _ok("fast"))], clock) + mw_slow = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), transport=slow) + mw_fast = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), transport=fast) + results = await asyncio.gather(mw_slow(_REQ), mw_fast(_REQ)) + assert {r.content for r in results} == {"slow", "fast"} + + class _GateSuccessBroken(InMemoryGate): async def record_success(self, entry): raise GovernanceBackendError("redis 抖动", scope="llm")