fix: return 429 attempt time to the stall budget
Independent verification found the first cut had swapped one bug for a worse one. The budgets were split by "did we send a request", so a 429 attempt counted as productive — but 429 is exempt from the retry budget, so its time burned neither budget. Against a queueing gateway that holds the request for the full timeout before answering 429, a call could hang for 301 attempts / 25.2 hours, measured, versus 301 seconds before the change. The split is now by which budget the time consumes: time that burns max_attempts is excluded from stall, time that does not (429 attempts included) belongs to stall. Measured again: back to one attempt / 301s. Only the chat loop needs this — embedding and ocr count 429 against max_attempts unconditionally, so the gap never existed there. The stall verdict moved into _stalled(), which both call sites had duplicated, to keep __call__ under the complexity gate.
This commit is contained in:
@@ -75,11 +75,24 @@ def backoff_delay(
|
||||
return max(delay, retry_after)
|
||||
|
||||
|
||||
class _Attempt:
|
||||
"""一次尝试的计时句柄;`refund()` 把它退还给 stall 账(见 `StallClock`)。"""
|
||||
|
||||
__slots__ = ("productive",)
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.productive = True
|
||||
|
||||
def refund(self) -> None:
|
||||
"""该次尝试不消耗重试预算(429),故其耗时归 stall 治理而非重试治理。"""
|
||||
self.productive = False
|
||||
|
||||
|
||||
class StallClock:
|
||||
"""调用级 stall 计时器: 只累计非生产性等待(issue #8 设计 §3.1)。
|
||||
|
||||
stall 预算治理的是"无人治理的等待"——429 退避、配额 wait 轮询、熔断冷却;
|
||||
真实尝试的耗时已由重试预算 `max_attempts` 治理,必须从 stall 账里扣除。
|
||||
**划分依据是"谁消耗重试预算"**,不是"是否发出了请求"。消耗 `max_attempts`
|
||||
的时间已被重试预算治理,从 stall 账扣除;不消耗它的时间无人治理,归 stall。
|
||||
两者重叠计费正是 issue #8 的根因: stall 预算(默认 300s)小于重试预算
|
||||
(3 × timeout_s),必然先耗尽,于是重试预算在超时场景下永远用不上。
|
||||
|
||||
@@ -87,7 +100,13 @@ class StallClock:
|
||||
"尝试已有结论"之后的动作,不是在等待重试机会;把它们计入 stall 会让遥测
|
||||
抖动参与判死。
|
||||
|
||||
每次调用创建一个实例。严禁提升为实例属性: 并发调用共享会互相污染计时。
|
||||
**例外: 429 尝试须 `refund()`**。429 免重试预算(饱和期等待而非死亡),若其
|
||||
耗时又算生产性,就掉进两个预算的缝隙——排队型网关持满 timeout 才回 429 时,
|
||||
每轮只有退避那一两秒进 stall 账,调用可挂满 `stall_window/backoff_base` 轮
|
||||
(实测 timeout=300/base=2 时达 25 小时)。退还后缝隙闭合。
|
||||
|
||||
每次调用创建一个实例。严禁提升为实例属性: `_entered_at` 会固定在进程启动
|
||||
时刻,使 `stalled_s()` 随进程运行时长单调增长,最终所有调用被误判 stalled。
|
||||
模块级共享单元, EmbeddingClient 与 OcrClient 复用(同 `backoff_delay`)。
|
||||
"""
|
||||
|
||||
@@ -99,18 +118,20 @@ class StallClock:
|
||||
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]:
|
||||
"""包裹一次真实尝试,其耗时记为生产性。"""
|
||||
async def attempting(self) -> AsyncIterator[_Attempt]:
|
||||
"""包裹一次真实尝试,其耗时默认记为生产性(除非被 `refund()`)。"""
|
||||
handle = _Attempt()
|
||||
started = self._now()
|
||||
try:
|
||||
yield
|
||||
yield handle
|
||||
finally:
|
||||
# 只做算术, 不吞任何异常——CancelledError 逐字穿透(库铁律)
|
||||
self._productive_s += self._now() - started
|
||||
# 只做算术与取值, 不吞任何异常——CancelledError 逐字穿透(库铁律)
|
||||
if handle.productive:
|
||||
self._productive_s += self._now() - started
|
||||
|
||||
|
||||
def _demote_call_failures(
|
||||
@@ -196,6 +217,15 @@ class _Failed:
|
||||
immediate: bool
|
||||
|
||||
|
||||
def _is_rate_limited(outcome: LLMResponse | _Failed) -> bool:
|
||||
"""429 = 服务端调度指令(gRPC pushback 语义,迭代 5): 按 Retry-After 退避但
|
||||
**不消耗重试预算**——饱和窗口里等待而非死亡;其余失败照常计数。
|
||||
|
||||
因其免重试预算,该次尝试的耗时必须归 stall 治理(`StallClock` 的 refund)。
|
||||
"""
|
||||
return isinstance(outcome, _Failed) and _failure_reason(outcome.exc) == "rate_limited"
|
||||
|
||||
|
||||
class RetryMW:
|
||||
"""尝试编排器;时钟/睡眠/随机全部注入,纯确定性可测(P6)。"""
|
||||
|
||||
@@ -252,10 +282,8 @@ class RetryMW:
|
||||
# 非生产性等待——真实尝试由重试预算治理,不再重复烧 stall 预算
|
||||
clock = StallClock(self._now)
|
||||
while True:
|
||||
# 调用级时间上限(迭代 5): 429 免预算后的兜底,防饱和期无限循环。
|
||||
# 与 _on_no_runnable 同款双条件(CHS 口径): 本地超窗且全局无进展才判死
|
||||
stall = self._bp.stall_window_s
|
||||
if clock.stalled_s() > stall and await self._quota.progress_age_s() > stall:
|
||||
# 调用级时间上限(迭代 5): 429 免预算后的兜底,防饱和期无限循环
|
||||
if await self._stalled(clock):
|
||||
raise AllSourcesExhausted(
|
||||
scope=self._scope,
|
||||
reason="stalled",
|
||||
@@ -266,13 +294,15 @@ class RetryMW:
|
||||
if picked is None:
|
||||
await self._on_no_runnable(gate_rejections, reasons, clock)
|
||||
continue
|
||||
async with clock.attempting():
|
||||
async with clock.attempting() as attempt:
|
||||
outcome = await self._attempt(request, *picked, reasons, attempt_fails)
|
||||
rate_limited = _is_rate_limited(outcome)
|
||||
if rate_limited:
|
||||
# 免了重试预算就得进 stall 账,否则这段耗时无人治理(见 StallClock)
|
||||
attempt.refund()
|
||||
if isinstance(outcome, LLMResponse):
|
||||
return outcome
|
||||
# 429 = 服务端调度指令(gRPC pushback 语义,迭代 5): 按 Retry-After
|
||||
# 退避但不消耗重试预算——饱和窗口里等待而非死亡;其余失败照常计数
|
||||
if _failure_reason(outcome.exc) != "rate_limited":
|
||||
if not rate_limited:
|
||||
fails += 1
|
||||
if fails >= self._retry.max_attempts:
|
||||
raise AllSourcesExhausted(
|
||||
@@ -325,6 +355,16 @@ class RetryMW:
|
||||
await self._settle_and_release(permit, 0)
|
||||
return None, gate_rejections
|
||||
|
||||
async def _stalled(self, clock: StallClock) -> bool:
|
||||
"""双条件 stall 判死(CHS governance.py:270-281): 本地累计等待与全局
|
||||
无进展**同时**超窗才判死——本地 monotonic 与后端时钟刻意不混用。
|
||||
|
||||
本地一侧只计非生产性等待(issue #8,见 `StallClock`)。短路顺序有意为之:
|
||||
本地未超窗就不问后端,省一次 Redis 往返。
|
||||
"""
|
||||
stall = self._bp.stall_window_s
|
||||
return clock.stalled_s() > stall and await self._quota.progress_age_s() > stall
|
||||
|
||||
async def _on_no_runnable(
|
||||
self, gate_rejections: int, reasons: dict[str, str], clock: StallClock
|
||||
) -> None:
|
||||
@@ -342,10 +382,7 @@ class RetryMW:
|
||||
retry_after_s=self._bp.poll_interval_s,
|
||||
per_source_reasons=reasons,
|
||||
)
|
||||
# 双条件 stall 判死(CHS governance.py:270-281): 本地累计等待与全局
|
||||
# 无进展**同时**超窗才判死——本地 monotonic 与后端时钟刻意不混用。
|
||||
stall = self._bp.stall_window_s
|
||||
if clock.stalled_s() > stall and await self._quota.progress_age_s() > stall:
|
||||
if await self._stalled(clock):
|
||||
names = tuple(s.name for s in self._sources)
|
||||
raise AllSourcesExhausted(
|
||||
scope=self._scope,
|
||||
|
||||
Reference in New Issue
Block a user