diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index 6d5bd53..72d3d2a 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -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, diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index 1c737d6..2c0cecb 100644 --- a/tests/unit/test_ocr_client.py +++ b/tests/unit/test_ocr_client.py @@ -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):