fix: bill only non-productive waiting against the chat stall budget
Issue #8: with timeout_s >= stall_window_s a single timed-out request exhausted the stall window before the second attempt was even dispatched, so LLM_MAX_RETRIES never applied and the whole scope was declared dead. Root cause is that real attempts and non-productive waiting charged the same wall clock, while the stall budget is the smaller of the two. The new StallClock subtracts attempt time from the stall account, leaving the two budgets orthogonal: attempts bill max_attempts, waiting bills stall_window_s. The dual-condition verdict, the inf semantics of progress_age_s, the 429 exemption and the error surface are untouched. The productive boundary is _attempt itself, telemetry included, so a slow recorder cannot push a call into a stalled verdict.
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user