fix: apply the non-productive stall budget to the ocr loop
Same failure path as the embedding loop: one timed-out attempt drains the wall-clock window, and the next round without a runnable source declares the scope dead in _on_no_runnable. All three governance loops now meter stall the same way.
This commit is contained in:
+10
-6
@@ -35,7 +35,7 @@ from polygateway.errors import (
|
||||
)
|
||||
from polygateway.middleware.breaker import BreakerGate
|
||||
from polygateway.middleware.ratelimit import QuotaGate
|
||||
from polygateway.middleware.retry import _failure_reason, backoff_delay
|
||||
from polygateway.middleware.retry import StallClock, _failure_reason, backoff_delay
|
||||
from polygateway.middleware.telemetry import TelemetryEmitter
|
||||
from polygateway.ports import OutcomeAwareSelector
|
||||
from polygateway.sources import SourceCooldownMemo
|
||||
@@ -204,13 +204,17 @@ class OcrClient:
|
||||
raise AllSourcesExhausted(scope=self._scope, reason="no_sources", retry_after_s=0.0)
|
||||
fails = 0
|
||||
reasons: dict[str, str] = {}
|
||||
entered_at = self._now()
|
||||
# 只计非生产性等待(issue #8): 真实尝试由重试预算治理,不重复烧 stall 预算
|
||||
clock = StallClock(self._now)
|
||||
while True:
|
||||
picked, gate_rejections = await self._pick_runnable(reasons)
|
||||
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(kind, image, *picked, reasons, session_id, parent_call_id)
|
||||
async with clock.attempting():
|
||||
outcome = await self._attempt(
|
||||
kind, image, *picked, reasons, session_id, parent_call_id
|
||||
)
|
||||
if isinstance(outcome, _AttemptOutcome):
|
||||
return outcome
|
||||
fails += 1
|
||||
@@ -253,7 +257,7 @@ class OcrClient:
|
||||
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)
|
||||
@@ -270,7 +274,7 @@ class OcrClient:
|
||||
per_source_reasons=reasons,
|
||||
)
|
||||
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,
|
||||
|
||||
@@ -80,6 +80,22 @@ class ScriptedOcrTransport:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class ClockAdvancingOcrTransport(ScriptedOcrTransport):
|
||||
"""按脚本 [(推进秒数, 动作), ...] 在一次尝试内部推进时钟(issue #8)。
|
||||
|
||||
stall 口径要区分"时间花在哪",故必须能让时钟只在 transport 内前进。
|
||||
"""
|
||||
|
||||
def __init__(self, script, clock):
|
||||
super().__init__([a for _, a in script])
|
||||
self._advances = [d for d, _ in script]
|
||||
self.clock = clock
|
||||
|
||||
async def _next(self, method, source, call_id):
|
||||
self.clock.advance(self._advances.pop(0))
|
||||
return await super()._next(method, source, call_id)
|
||||
|
||||
|
||||
class StaticSelector:
|
||||
def order(self, sources, stats):
|
||||
return list(sources)
|
||||
@@ -292,6 +308,38 @@ class TestBackpressure:
|
||||
await permit.settle(0)
|
||||
await permit.release()
|
||||
|
||||
async def test_single_timeout_does_not_exhaust_stall_budget(self):
|
||||
"""issue #8: 一次耗满 timeout 的尝试不得吃掉 stall 预算而使重试失效。
|
||||
|
||||
OCR 只有 `_on_no_runnable` 一处 stall 判定,故失效链条是"先超时一次
|
||||
(墙钟耗尽)→ 再遇到无可用源 → 判死"。此处正是这条路径。
|
||||
"""
|
||||
clock = FakeClock()
|
||||
limiter = held = None
|
||||
rounds = []
|
||||
|
||||
async def toggle_permit(_seconds):
|
||||
"""首次退避占满 permit,迫使下一轮走 _on_no_runnable;之后放行。"""
|
||||
nonlocal held
|
||||
rounds.append(_seconds)
|
||||
if len(rounds) > 10:
|
||||
raise RuntimeError("超过 10 次轮询仍未判死/未获 permit")
|
||||
if len(rounds) == 1:
|
||||
held = await limiter.acquire("m1", 0)
|
||||
else:
|
||||
await held.settle(0)
|
||||
await held.release()
|
||||
|
||||
transport = ClockAdvancingOcrTransport(
|
||||
[(300.1, TransientError("timeout", status_code=504)), (0.0, "text")], clock
|
||||
)
|
||||
client, limiter, _ = _client(
|
||||
[_src(max_concurrency=1)], [], now=clock, sleep=toggle_permit, transport=transport
|
||||
)
|
||||
r = await client.recognize_text(b"jpg")
|
||||
assert r.text == "LINE-1"
|
||||
assert len(transport.calls) == 2 # 第二次尝试确实发出了
|
||||
|
||||
|
||||
class FakeClock:
|
||||
def __init__(self, start=1000.0):
|
||||
|
||||
Reference in New Issue
Block a user