diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py index 2616f6c..dcf8c2a 100644 --- a/src/polygateway/embedding.py +++ b/src/polygateway/embedding.py @@ -28,7 +28,6 @@ from loguru import logger from polygateway.config import EmbeddingSettings from polygateway.errors import ( AllSourcesExhausted, - CircuitOpenError, GovernanceBackendError, PolyGatewayError, RequestRejectedError, @@ -37,11 +36,11 @@ from polygateway.errors import ( SourceNotConfiguredError, TransientError, ) +from polygateway.middleware.admission import SourceAdmission, settle_and_release from polygateway.middleware.breaker import BreakerGate from polygateway.middleware.ratelimit import QuotaGate from polygateway.middleware.retry import StallClock, _failure_reason, backoff_delay from polygateway.middleware.telemetry import TelemetryEmitter -from polygateway.sources import SourceCooldownMemo from polygateway.types import ( ChatRequest, EmbeddingResponse, @@ -122,13 +121,10 @@ class EmbeddingClient: # embed payload 硬编码 {model, input},带 extra_body 的源必须先剥离, # 否则遥测会记录一个从未发出的采样参数(issue #4 决策 G) self._sources = strip_unsupported_extra_body(list(sources), path="embedding") - self._selector = selector self._quota = QuotaGate(limiter, scope=self._scope) self._breaker = BreakerGate(breaker, scope=self._scope) self._transport = transport self._retry = retry - self._bp = backpressure - self._quota_full = quota_full self._emitter = ( TelemetryEmitter(telemetry, pricing=pricing, text_cap=text_cap) if telemetry else None ) @@ -137,10 +133,22 @@ class EmbeddingClient: self._batch_size = batch_size self._normalize = normalize self._expected_dim = expected_dim - self._memo = SourceCooldownMemo(now=now) self._now = now self._sleep = sleep self._rng = rng + # 准入编排三条循环共用一份(issue #14);冷却备忘由它独占 + self._admission = SourceAdmission( + scope=self._scope, + sources=self._sources, + selector=selector, + quota=self._quota, + breaker=self._breaker, + backpressure=backpressure, + quota_full=quota_full, + now=now, + sleep=sleep, + rng=rng, + ) self._closed = False async def embed( @@ -207,9 +215,9 @@ class EmbeddingClient: # 只计非生产性等待(issue #8): 真实尝试由重试预算治理,不重复烧 stall 预算 clock = StallClock(self._now) while True: - picked, gate_rejections = await self._pick_runnable(reasons) + picked, gate_rejections = await self._admission.pick(reasons, {}) if picked is None: - await self._on_no_runnable(gate_rejections, reasons, clock) + await self._admission.on_no_runnable(gate_rejections, reasons, clock) continue async with clock.attempting(): outcome = await self._attempt( @@ -228,62 +236,6 @@ class EmbeddingClient: if not outcome.immediate: await self._sleep(backoff_delay(self._retry, fails, outcome.exc, self._rng)) - async def _pick_runnable( - self, reasons: dict[str, str] - ) -> tuple[tuple[SourceConfig, Permit, GateDecision] | None, int]: - stats = {s.name: await self._quota.stats(s) for s in self._sources} - gate_rejections = 0 - for cand in self._selector.order(self._sources, stats): - if self._memo.active(cand.name): - gate_rejections += 1 - reasons[cand.name] = "cooldown" - continue - permit = await self._quota.try_acquire(cand) - if permit is None: - reasons.setdefault(cand.name, "rate_limited") - continue - entry = None - try: - entry = await self._breaker.try_enter(cand, uuid.uuid4().hex) - finally: - if entry is None: - await self._settle_and_release(permit, 0) - if entry.allowed: - return (cand, permit, entry), gate_rejections - gate_rejections += 1 - reasons[cand.name] = "circuit_open" - self._memo.set_until(cand.name, self._now() + entry.retry_after_s) - await self._settle_and_release(permit, 0) - return None, gate_rejections - - async def _on_no_runnable( - 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) - raise CircuitOpenError( - scope=self._scope, - retry_after_s=await self._breaker.retry_after_s(names), - per_source_reasons=reasons, - ) - if self._quota_full == "fail_fast": - raise AllSourcesExhausted( - scope=self._scope, - reason="quota_exhausted", - retry_after_s=self._bp.poll_interval_s, - per_source_reasons=reasons, - ) - stall = self._bp.stall_window_s - 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, - reason="stalled", - retry_after_s=await self._breaker.retry_after_s(names), - per_source_reasons=reasons, - ) - await self._sleep(self._bp.poll_interval_s * (0.5 + 0.5 * self._rng())) - async def _attempt( self, batch: list[str], @@ -377,7 +329,7 @@ class EmbeddingClient: ) return _FailedBatch(exc, immediate=dead) finally: - await self._settle_and_release(permit, actual) + await settle_and_release(permit, actual) # —— 辅助 —— @@ -399,17 +351,6 @@ class EmbeddingClient: except (GovernanceBackendError, SourceNotConfiguredError) as exc: logger.warning("embedding 治理记账写回降级(不冒泡): {}", exc) - async def _settle_and_release(self, permit: Permit, actual: int) -> None: - try: - try: - await permit.settle(actual) - finally: - await permit.release() - except asyncio.CancelledError: - raise - except Exception as exc: - logger.warning("embedding permit 结算/释放失败(不掩盖主异常): {}", exc) - async def _emit( self, batch: list[str], diff --git a/src/polygateway/middleware/admission.py b/src/polygateway/middleware/admission.py new file mode 100644 index 0000000..a817523 --- /dev/null +++ b/src/polygateway/middleware/admission.py @@ -0,0 +1,245 @@ +"""SourceAdmission: 一次尝试的准入编排,三条治理循环(chat/embedding/ocr)共用一份。 + +**收敛缘由(issue #14)**: 本模块的两个方法此前在 `middleware/retry.py`、 +`embedding.py`、`ocr.py` 各存一份逐字复制(后两份是第一份的子集)。准入语义 +一直在演进——issue #8 改过 stall 口径、M2.5 加过 AIMD pacer、issue #14 要加 +熔断等待档——每演进一次就要三处同步,漏一处即行为分叉。三份复制正是库铁律 +痛斥的那种模式(遥测"三项目 4 处复制"的教训),只不过这次发生在库内部。 + +**职责边界**: 只管"挑出一个可跑的源"与"一个都挑不出来时怎么办";一次尝试 +本身(transport 调用、记账写回、逐次遥测)仍归各循环的 `_attempt`。 + +**共享而非持有**: `QuotaGate`/`BreakerGate`/`AdaptivePacer`/`SourceSelector` 由 +调用方构造后传入**同一实例**——三处 `_attempt` 仍要用它们做记账写回与 +`pacer.leave()`。pacer 尤其不能各建一个: 它有在途计数,分裂成两个计数器会让 +`admit`/`enter` 与 `leave` 记到不同账上。`SourceCooldownMemo` 只被准入消费, +由本类独占。 +""" + +from __future__ import annotations + +import asyncio +import random +import time +import uuid +from typing import TYPE_CHECKING + +from loguru import logger + +from polygateway.errors import AllSourcesExhausted, CircuitOpenError +from polygateway.sources import SourceCooldownMemo + +if TYPE_CHECKING: + from collections.abc import Callable + + from polygateway.middleware.breaker import BreakerGate + from polygateway.middleware.ratelimit import QuotaGate + from polygateway.middleware.retry import StallClock + from polygateway.ports import GateDecision, Permit, SourceSelector + from polygateway.sources import AdaptivePacer + from polygateway.types import BackpressurePolicy, SourceConfig + + +async def settle_and_release(permit: Permit, actual: int) -> None: + """finally 专用: settle 后必 release;失败降级 warning,绝不掩盖主异常/取消。 + + 三条循环的 `_attempt` 与本模块的准入拒绝路径共用这一份(此前三处逐字复制, + 仅 warning 文案不同)。 + """ + try: + try: + await permit.settle(actual) + finally: + await permit.release() + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning("permit 结算/释放失败(不掩盖主异常): {}", exc) + + +def _demote_call_failures( + ordered: list[SourceConfig], + attempt_fails: dict[str, int], + health: Callable[[str], float] | None, +) -> list[SourceConfig]: + """调用内降权(设计 §3.3/§3.36): 失败 ≥2 次且存在可信替代才让位。 + + 可信替代 = 某未失败候选 health ≥ 0.5 × 失败源 health——异构池里健康源 + 偶发失败不该被推向已知坏源(第三轮教训: 期望成功率 83% vs 10%)。 + 无健康视图(round_robin 等)保持无条件降权(冷启动保护)。 + + `attempt_fails` 为空时恒等返回原列表对象——embedding/ocr 不维护调用内 + 失败计数,故对它们这一步是零成本的空操作,无需在调用侧加分支。 + """ + demoted = [s for s in ordered if attempt_fails.get(s.name, 0) >= 2] + if not demoted or len(demoted) == len(ordered): + return ordered + if health is None: + return _move_to_tail(ordered, demoted) + return _health_gated_reorder(ordered, demoted, attempt_fails, health) + + +def _move_to_tail(ordered: list[SourceConfig], demoted: list[SourceConfig]) -> list[SourceConfig]: + """无健康视图: 无条件移尾(冷启动保护原语义)。""" + names = {d.name for d in demoted} + return [s for s in ordered if s.name not in names] + demoted + + +def _health_gated_reorder( + ordered: list[SourceConfig], + demoted: list[SourceConfig], + attempt_fails: dict[str, int], + health: Callable[[str], float], +) -> list[SourceConfig]: + """健康门槛降权: 无可信替代则原地重试;有则插到可信替代之后。""" + demoted = _credible_demotions(ordered, demoted, attempt_fails, health) + if not demoted: + return ordered + names = {d.name for d in demoted} + rest = [s for s in ordered if s.name not in names] + return _insert_after_credible(rest, demoted, health) + + +def _insert_after_credible( + rest: list[SourceConfig], + demoted: list[SourceConfig], + health: Callable[[str], float], +) -> list[SourceConfig]: + """插入位置(第四轮教训): 被降权源排在可信替代之后、不可信源之前—— + 可信替代被限流闸/熔断跳过时,下一候选是失败源本身而非垃圾源。""" + bar = 0.5 * max(health(d.name) for d in demoted) + credible = [s for s in rest if health(s.name) >= bar] + junk = [s for s in rest if health(s.name) < bar] + return credible + demoted + junk + + +def _credible_demotions( + ordered: list[SourceConfig], + demoted: list[SourceConfig], + attempt_fails: dict[str, int], + health: Callable[[str], float], +) -> list[SourceConfig]: + """健康门槛过滤: 仅当存在"健康分 ≥ 失败源一半"的未失败候选,让位才有意义。""" + alts = [o for o in ordered if attempt_fails.get(o.name, 0) < 2] + return [s for s in demoted if any(health(o.name) >= 0.5 * health(s.name) for o in alts)] + + +class SourceAdmission: + """准入编排器(CHS `governance.py:107-285` 同款);时钟/睡眠/随机全部注入。""" + + def __init__( + self, + *, + scope: str, + sources: list[SourceConfig], + selector: SourceSelector, + quota: QuotaGate, + breaker: BreakerGate, + backpressure: BackpressurePolicy, + quota_full: str, + memo: SourceCooldownMemo | None = None, + pacer: AdaptivePacer | None = None, + health_view: Callable[[str], float] | None = None, + now: Callable[[], float] = time.monotonic, + sleep: Callable[[float], object] = asyncio.sleep, + rng: Callable[[], float] = random.random, + ) -> None: + if quota_full not in ("wait", "fail_fast"): + raise ValueError(f"quota_full 必须是 wait|fail_fast: {quota_full!r}") + self._scope = scope + self._sources = sources + self._selector = selector + self._quota = quota + self._breaker = breaker + self._bp = backpressure + self._quota_full = quota_full + self._memo = memo or SourceCooldownMemo(now=now) + self._pacer = pacer + self._health_view = health_view + self._now = now + self._sleep = sleep + self._rng = rng + + # —— 选源与准入(CHS _pick_runnable 120-167)—— + + async def pick( + self, reasons: dict[str, str], attempt_fails: dict[str, int] + ) -> tuple[tuple[SourceConfig, Permit, GateDecision] | None, int]: + """挑出第一个过闸的候选;返回 (选中三元组 | None, 熔断类拒绝计数)。""" + stats = {s.name: await self._quota.stats(s) for s in self._sources} + gate_rejections = 0 + ordered = _demote_call_failures( + self._selector.order(self._sources, stats), attempt_fails, self._health_view + ) + for cand in ordered: + if self._memo.active(cand.name): + # 冷却备忘跳过也计入拒绝数,保住 circuit_open 判据(CHS 同款) + gate_rejections += 1 + reasons[cand.name] = "cooldown" + continue + if self._pacer is not None and not self._pacer.admit(cand.name): + # AIMD 超限: 不计 gate_rejections → 走 quota-wait 排队,不误判熔断 + reasons.setdefault(cand.name, "adaptive_paced") + continue + permit = await self._quota.try_acquire(cand) + if permit is None: + reasons.setdefault(cand.name, "rate_limited") + continue + entry = None + try: + entry = await self._breaker.try_enter(cand, uuid.uuid4().hex) + finally: + # try_enter 未归还 entry(异常/取消)→ 释放已占 permit,不吞任何异常 + if entry is None: + await settle_and_release(permit, 0) + if entry.allowed: + if self._pacer is not None: + self._pacer.enter(cand.name) + return (cand, permit, entry), gate_rejections + gate_rejections += 1 + reasons[cand.name] = "circuit_open" + # 开路源本地记冷却,避免每轮白烧 RPM 探测(CHS governance.py:107) + self._memo.set_until(cand.name, self._now() + entry.retry_after_s) + await settle_and_release(permit, 0) + return None, gate_rejections + + # —— 背压与 stall 判死(CHS governance.py:270-285)—— + + 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: + """一个源都挑不出来时的处置: 判死或等待一轮再来。""" + if gate_rejections == len(self._sources): + names = tuple(s.name for s in self._sources) + raise CircuitOpenError( + scope=self._scope, + retry_after_s=await self._breaker.retry_after_s(names), + per_source_reasons=reasons, + ) + if self._quota_full == "fail_fast": + raise AllSourcesExhausted( + scope=self._scope, + reason="quota_exhausted", + retry_after_s=self._bp.poll_interval_s, + per_source_reasons=reasons, + ) + if await self.stalled(clock): + names = tuple(s.name for s in self._sources) + raise AllSourcesExhausted( + scope=self._scope, + reason="stalled", + retry_after_s=await self._breaker.retry_after_s(names), + per_source_reasons=reasons, + ) + # jitter ∈ [0.5p, 1.0p] 防惊群(CHS governance.py:283-285) + await self._sleep(self._bp.poll_interval_s * (0.5 + 0.5 * self._rng())) diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index e46df63..c30f2ab 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -23,7 +23,6 @@ from loguru import logger from polygateway.errors import ( AllSourcesExhausted, - CircuitOpenError, GovernanceBackendError, PolyGatewayError, RequestRejectedError, @@ -32,10 +31,11 @@ from polygateway.errors import ( SourceNotConfiguredError, TransientError, ) +from polygateway.middleware.admission import SourceAdmission, settle_and_release from polygateway.middleware.breaker import BreakerGate from polygateway.middleware.ratelimit import QuotaGate from polygateway.ports import OutcomeAwareSelector -from polygateway.sources import AdaptivePacer, SourceCooldownMemo +from polygateway.sources import AdaptivePacer from polygateway.streaming import StreamLivenessTimeout from polygateway.types import LLMResponse @@ -50,6 +50,7 @@ if TYPE_CHECKING: SourceSelector, Transport, ) + from polygateway.sources import SourceCooldownMemo from polygateway.types import ( BackpressurePolicy, ChatRequest, @@ -134,70 +135,6 @@ class StallClock: self._productive_s += self._now() - started -def _demote_call_failures( - ordered: list[SourceConfig], - attempt_fails: dict[str, int], - health: Callable[[str], float] | None, -) -> list[SourceConfig]: - """调用内降权(设计 §3.3/§3.36): 失败 ≥2 次且存在可信替代才让位。 - - 可信替代 = 某未失败候选 health ≥ 0.5 × 失败源 health——异构池里健康源 - 偶发失败不该被推向已知坏源(第三轮教训: 期望成功率 83% vs 10%)。 - 无健康视图(round_robin 等)保持无条件降权(冷启动保护)。 - """ - demoted = [s for s in ordered if attempt_fails.get(s.name, 0) >= 2] - if not demoted or len(demoted) == len(ordered): - return ordered - if health is None: - return _move_to_tail(ordered, demoted) - return _health_gated_reorder(ordered, demoted, attempt_fails, health) - - -def _move_to_tail(ordered: list[SourceConfig], demoted: list[SourceConfig]) -> list[SourceConfig]: - """无健康视图: 无条件移尾(冷启动保护原语义)。""" - names = {d.name for d in demoted} - return [s for s in ordered if s.name not in names] + demoted - - -def _health_gated_reorder( - ordered: list[SourceConfig], - demoted: list[SourceConfig], - attempt_fails: dict[str, int], - health: Callable[[str], float], -) -> list[SourceConfig]: - """健康门槛降权: 无可信替代则原地重试;有则插到可信替代之后。""" - demoted = _credible_demotions(ordered, demoted, attempt_fails, health) - if not demoted: - return ordered - names = {d.name for d in demoted} - rest = [s for s in ordered if s.name not in names] - return _insert_after_credible(rest, demoted, health) - - -def _insert_after_credible( - rest: list[SourceConfig], - demoted: list[SourceConfig], - health: Callable[[str], float], -) -> list[SourceConfig]: - """插入位置(第四轮教训): 被降权源排在可信替代之后、不可信源之前—— - 可信替代被限流闸/熔断跳过时,下一候选是失败源本身而非垃圾源。""" - bar = 0.5 * max(health(d.name) for d in demoted) - credible = [s for s in rest if health(s.name) >= bar] - junk = [s for s in rest if health(s.name) < bar] - return credible + demoted + junk - - -def _credible_demotions( - ordered: list[SourceConfig], - demoted: list[SourceConfig], - attempt_fails: dict[str, int], - health: Callable[[str], float], -) -> list[SourceConfig]: - """健康门槛过滤: 仅当存在"健康分 ≥ 失败源一半"的未失败候选,让位才有意义。""" - alts = [o for o in ordered if attempt_fails.get(o.name, 0) < 2] - return [s for s in demoted if any(health(o.name) >= 0.5 * health(s.name) for o in alts)] - - def _failure_reason(exc: PolyGatewayError) -> str: """失败原因归类(CHS governance.py:169 同款)。""" if isinstance(exc, SourceDeadError): @@ -252,23 +189,36 @@ class RetryMW: raise ValueError(f"quota_full 必须是 wait|fail_fast: {quota_full!r}") self._scope = scope self._sources = list(sources) - self._selector = selector + # 记账写回与 pacer 结算仍在 `_attempt` 内,故这三者由本类持有并与 + # `SourceAdmission` **共享同一实例**(pacer 有在途计数,不可分裂) self._quota = QuotaGate(limiter, scope=self._scope) self._breaker = BreakerGate(gate, scope=self._scope) self._transport = transport self._retry = retry - self._bp = backpressure - self._quota_full = quota_full - self._memo = cooldown_memo or SourceCooldownMemo(now=now) # M2.5: 选源器可选健康喂数端口,构造期 isinstance 判定一次(设计 §3.2) self._outcome_sink = selector if isinstance(selector, OutcomeAwareSelector) else None - self._health_view = self._outcome_sink.health if self._outcome_sink else None # M2.5 §3.35: AIMD 自适应并发——429 收紧、成功回涨,超限调用排队不烧预算 self._pacer = pacer or AdaptivePacer(ceiling=64.0) self._emitter = emitter self._now = now self._sleep = sleep self._rng = rng + # 准入编排三条循环共用一份(issue #14);冷却备忘由它独占 + self._admission = SourceAdmission( + scope=self._scope, + sources=self._sources, + selector=selector, + quota=self._quota, + breaker=self._breaker, + backpressure=backpressure, + quota_full=quota_full, + memo=cooldown_memo, + pacer=self._pacer, + health_view=self._outcome_sink.health if self._outcome_sink else None, + now=now, + sleep=sleep, + rng=rng, + ) async def __call__(self, request: ChatRequest) -> LLMResponse: """执行治理调用;scope 级失败按 §6.1 携结构化字段上抛。""" @@ -283,16 +233,16 @@ class RetryMW: clock = StallClock(self._now) while True: # 调用级时间上限(迭代 5): 429 免预算后的兜底,防饱和期无限循环 - if await self._stalled(clock): + if await self._admission.stalled(clock): raise AllSourcesExhausted( scope=self._scope, reason="stalled", retry_after_s=self._retry.backoff_base_s, per_source_reasons=reasons, ) - picked, gate_rejections = await self._pick_runnable(reasons, attempt_fails) + picked, gate_rejections = await self._admission.pick(reasons, attempt_fails) if picked is None: - await self._on_no_runnable(gate_rejections, reasons, clock) + await self._admission.on_no_runnable(gate_rejections, reasons, clock) continue async with clock.attempting() as attempt: outcome = await self._attempt(request, *picked, reasons, attempt_fails) @@ -314,87 +264,6 @@ class RetryMW: if not outcome.immediate: await self._sleep(self._backoff_delay(max(fails, 1), outcome.exc)) - # —— 选源与准入(CHS _pick_runnable 120-167)—— - - async def _pick_runnable( - self, reasons: dict[str, str], attempt_fails: dict[str, int] - ) -> tuple[tuple[SourceConfig, Permit, GateDecision] | None, int]: - stats = {s.name: await self._quota.stats(s) for s in self._sources} - gate_rejections = 0 - ordered = _demote_call_failures( - self._selector.order(self._sources, stats), attempt_fails, self._health_view - ) - for cand in ordered: - if self._memo.active(cand.name): - # 冷却备忘跳过也计入拒绝数,保住 circuit_open 判据(CHS 同款) - gate_rejections += 1 - reasons[cand.name] = "cooldown" - continue - if not self._pacer.admit(cand.name): - # AIMD 超限: 不计 gate_rejections → 走 quota-wait 排队,不误判熔断 - reasons.setdefault(cand.name, "adaptive_paced") - continue - permit = await self._quota.try_acquire(cand) - if permit is None: - reasons.setdefault(cand.name, "rate_limited") - continue - entry = None - try: - entry = await self._breaker.try_enter(cand, uuid.uuid4().hex) - finally: - # try_enter 未归还 entry(异常/取消)→ 释放已占 permit,不吞任何异常 - if entry is None: - await self._settle_and_release(permit, 0) - if entry.allowed: - self._pacer.enter(cand.name) - return (cand, permit, entry), gate_rejections - gate_rejections += 1 - reasons[cand.name] = "circuit_open" - # 开路源本地记冷却,避免每轮白烧 RPM 探测(CHS governance.py:107) - self._memo.set_until(cand.name, self._now() + entry.retry_after_s) - await self._settle_and_release(permit, 0) - return None, gate_rejections - - # —— 背压与 stall 判死(CHS governance.py:270-285)—— - - 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: - if gate_rejections == len(self._sources): - names = tuple(s.name for s in self._sources) - raise CircuitOpenError( - scope=self._scope, - retry_after_s=await self._breaker.retry_after_s(names), - per_source_reasons=reasons, - ) - if self._quota_full == "fail_fast": - raise AllSourcesExhausted( - scope=self._scope, - reason="quota_exhausted", - retry_after_s=self._bp.poll_interval_s, - per_source_reasons=reasons, - ) - if await self._stalled(clock): - names = tuple(s.name for s in self._sources) - raise AllSourcesExhausted( - scope=self._scope, - reason="stalled", - retry_after_s=await self._breaker.retry_after_s(names), - per_source_reasons=reasons, - ) - # jitter ∈ [0.5p, 1.0p] 防惊群(CHS governance.py:283-285) - await self._sleep(self._bp.poll_interval_s * (0.5 + 0.5 * self._rng())) - # —— 单次尝试(CHS run 200-268)—— async def _attempt( @@ -460,7 +329,7 @@ class RetryMW: return _Failed(exc, immediate=dead) finally: self._pacer.leave(source.name) - await self._settle_and_release(permit, actual) + await settle_and_release(permit, actual) async def _on_rejected( self, exc: RequestRejectedError, source: SourceConfig, entry: GateDecision @@ -523,18 +392,6 @@ class RetryMW: reasoning_tokens=result.reasoning_tokens, ) - async def _settle_and_release(self, permit: Permit, actual: int) -> None: - """finally 专用: settle 后必 release;失败降级 warning,绝不掩盖主异常/取消。""" - try: - try: - await permit.settle(actual) - finally: - await permit.release() - except asyncio.CancelledError: - raise - except Exception as exc: - logger.warning("permit 结算/释放失败(不掩盖主异常): {}", exc) - async def _emit( self, request: ChatRequest, diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index 7f1b151..bd06daa 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -24,7 +24,6 @@ from loguru import logger from polygateway.errors import ( AllSourcesExhausted, - CircuitOpenError, GovernanceBackendError, PolyGatewayError, RequestRejectedError, @@ -33,12 +32,12 @@ from polygateway.errors import ( SourceNotConfiguredError, TransientError, ) +from polygateway.middleware.admission import SourceAdmission, settle_and_release from polygateway.middleware.breaker import BreakerGate from polygateway.middleware.ratelimit import QuotaGate 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 from polygateway.types import ( ChatRequest, LLMResponse, @@ -126,14 +125,24 @@ class OcrClient: self._breaker = BreakerGate(breaker, scope=self._scope) self._transport = transport self._retry = retry - self._bp = backpressure - self._quota_full = quota_full self._emitter = TelemetryEmitter(telemetry, text_cap=text_cap) if telemetry else None self._telemetry = telemetry - self._memo = SourceCooldownMemo(now=now) self._now = now self._sleep = sleep self._rng = rng + # 准入编排三条循环共用一份(issue #14);冷却备忘由它独占 + self._admission = SourceAdmission( + scope=self._scope, + sources=self._sources, + selector=selector, + quota=self._quota, + breaker=self._breaker, + backpressure=backpressure, + quota_full=quota_full, + now=now, + sleep=sleep, + rng=rng, + ) self._closed = False # —— 公共端口(OcrTextPort / OcrLayoutPort)—— @@ -234,9 +243,9 @@ class OcrClient: # 只计非生产性等待(issue #8): 真实尝试由重试预算治理,不重复烧 stall 预算 clock = StallClock(self._now) while True: - picked, gate_rejections = await self._pick_runnable(reasons) + picked, gate_rejections = await self._admission.pick(reasons, {}) if picked is None: - await self._on_no_runnable(gate_rejections, reasons, clock) + await self._admission.on_no_runnable(gate_rejections, reasons, clock) continue async with clock.attempting(): outcome = await self._attempt( @@ -255,62 +264,6 @@ class OcrClient: if not outcome.immediate: await self._sleep(backoff_delay(self._retry, fails, outcome.exc, self._rng)) - async def _pick_runnable( - self, reasons: dict[str, str] - ) -> tuple[tuple[SourceConfig, Permit, GateDecision] | None, int]: - stats = {s.name: await self._quota.stats(s) for s in self._sources} - gate_rejections = 0 - for cand in self._selector.order(self._sources, stats): - if self._memo.active(cand.name): - gate_rejections += 1 - reasons[cand.name] = "cooldown" - continue - permit = await self._quota.try_acquire(cand) - if permit is None: - reasons.setdefault(cand.name, "rate_limited") - continue - entry = None - try: - entry = await self._breaker.try_enter(cand, uuid.uuid4().hex) - finally: - if entry is None: - await self._settle_and_release(permit) - if entry.allowed: - return (cand, permit, entry), gate_rejections - gate_rejections += 1 - reasons[cand.name] = "circuit_open" - self._memo.set_until(cand.name, self._now() + entry.retry_after_s) - await self._settle_and_release(permit) - return None, gate_rejections - - async def _on_no_runnable( - 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) - raise CircuitOpenError( - scope=self._scope, - retry_after_s=await self._breaker.retry_after_s(names), - per_source_reasons=reasons, - ) - if self._quota_full == "fail_fast": - raise AllSourcesExhausted( - scope=self._scope, - reason="quota_exhausted", - retry_after_s=self._bp.poll_interval_s, - per_source_reasons=reasons, - ) - stall = self._bp.stall_window_s - 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, - reason="stalled", - retry_after_s=await self._breaker.retry_after_s(names), - per_source_reasons=reasons, - ) - await self._sleep(self._bp.poll_interval_s * (0.5 + 0.5 * self._rng())) - async def _attempt( self, kind: _OcrKind, @@ -398,7 +351,7 @@ class OcrClient: ) return _FailedAttempt(exc, immediate=dead) finally: - await self._settle_and_release(permit) + await settle_and_release(permit, 0) async def _invoke( self, kind: _OcrKind, image: bytes, source: SourceConfig, call_id: str @@ -435,18 +388,6 @@ class OcrClient: except (GovernanceBackendError, SourceNotConfiguredError) as exc: logger.warning("OCR 治理记账写回降级(不冒泡): {}", exc) - async def _settle_and_release(self, permit: Permit) -> None: - """settle 恒 0: OCR 无 token 计费(设计 §5 差异①)。""" - try: - try: - await permit.settle(0) - finally: - await permit.release() - except asyncio.CancelledError: - raise - except Exception as exc: - logger.warning("OCR permit 结算/释放失败(不掩盖主异常): {}", exc) - async def _emit( self, kind: _OcrKind, diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index affce44..8f83f19 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -629,7 +629,7 @@ class TestDemotionInsertPosition: async def test_demoted_lands_before_junk_sources(self): # a 失败 2 次;b 可信(0.9)但会被跳过时,第三候选应是 a 而非垃圾源 c - from polygateway.middleware.retry import _demote_call_failures + from polygateway.middleware.admission import _demote_call_failures srcs = [_src("a"), _src("b"), _src("c")] health = {"a": 0.9, "b": 0.9, "c": 0.05}.__getitem__ @@ -637,7 +637,7 @@ class TestDemotionInsertPosition: assert [s.name for s in out] == ["b", "a", "c"] async def test_health_blind_demotion_still_tail(self): - from polygateway.middleware.retry import _demote_call_failures + from polygateway.middleware.admission import _demote_call_failures srcs = [_src("a"), _src("b"), _src("c")] out = _demote_call_failures(srcs, {"a": 2}, None)