diff --git a/src/polygateway/client.py b/src/polygateway/client.py index 1bfbd0b..dac5f5c 100644 --- a/src/polygateway/client.py +++ b/src/polygateway/client.py @@ -134,6 +134,7 @@ class GatewayClient: retry: RetryPolicy, backpressure: BackpressurePolicy, quota_full: str = "wait", + circuit_open: str = "fail_fast", telemetry: TelemetryRecorder | None = None, pricing: PricingTable | None = None, text_cap: int | None = None, @@ -162,6 +163,7 @@ class GatewayClient: retry=retry, backpressure=backpressure, quota_full=quota_full, + circuit_open=circuit_open, cooldown_memo=SourceCooldownMemo(now=now), # AIMD ceiling 尊重源级静态并发上限(独立核验 I1: 不得静默钳制大于 64 的配置) pacer=AdaptivePacer( @@ -313,6 +315,7 @@ class GatewayClient: retry=settings.retry, backpressure=settings.backpressure, quota_full=settings.quota_full, + circuit_open=settings.circuit_open, telemetry=telemetry if telemetry is not None else _build_telemetry(settings), pricing=PricingTable.from_file(settings.pricing_path) if settings.pricing_path is not None diff --git a/src/polygateway/config.py b/src/polygateway/config.py index 8dfe5b2..4f1b839 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -50,6 +50,9 @@ _SOURCE_FIELDS: dict[str, tuple[str, str]] = { _RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE"}) _SELECTORS = frozenset({"round_robin", "least_inflight", "health_aware"}) _QUOTA_FULL = frozenset({"wait", "fail_fast"}) +# 熔断全拒时的处置(issue #14);值域与 _QUOTA_FULL 相同但语义不同——配额满是 +# "排队等自己的份额"(必然轮到),熔断开路是"等源恢复"(未必恢复),故分列两键 +_CIRCUIT_OPEN = frozenset({"wait", "fail_fast"}) # 后端合法域: env 解析与构造期校验共用一份定义,避免两处分叉 _LIMITER_BACKENDS = frozenset({"memory", "redis"}) _BREAKER_BACKENDS = frozenset({"memory", "redis"}) @@ -128,6 +131,9 @@ class GatewaySettings: backpressure: BackpressurePolicy selector: str quota_full: str + # 熔断全拒时是当场判死还是等冷却过去(issue #14);缺省 fail_fast 保持 + # 存量下游的控制流不变,单源 scope 应显式配 wait + circuit_open: str limiter_backend: str breaker_backend: str cache_backend: str @@ -203,6 +209,7 @@ class GatewaySettings: ("telemetry_backend", _TELEMETRY_BACKENDS), ("selector", _SELECTORS), ("quota_full", _QUOTA_FULL), + ("circuit_open", _CIRCUIT_OPEN), ): value = getattr(self, field) if value not in allowed: @@ -320,6 +327,9 @@ class GatewaySettings: backpressure=_load_backpressure(scope_u, env), selector=_load_choice(env, f"{scope_u}__SELECTOR", _SELECTORS, "health_aware"), quota_full=_load_choice(env, f"{scope_u}__QUOTA_FULL", _QUOTA_FULL, "wait"), + circuit_open=_load_choice( + env, f"{scope_u}__CIRCUIT_OPEN", _CIRCUIT_OPEN, "fail_fast" + ), **_load_pgw(env), ) diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py index dcf8c2a..f06b97c 100644 --- a/src/polygateway/embedding.py +++ b/src/polygateway/embedding.py @@ -101,6 +101,7 @@ class EmbeddingClient: retry: RetryPolicy, backpressure: BackpressurePolicy, quota_full: str = "wait", + circuit_open: str = "fail_fast", telemetry: TelemetryRecorder | None = None, pricing: PricingTable | None = None, text_cap: int | None = None, @@ -113,8 +114,6 @@ class EmbeddingClient: ) -> None: if batch_size < 1: raise ValueError("batch_size 必须 ≥ 1") - if quota_full not in ("wait", "fail_fast"): - raise ValueError(f"quota_full 必须是 wait|fail_fast: {quota_full!r}") if expected_dim is not None and expected_dim < 1: raise ValueError("expected_dim 必须 ≥ 1") self._scope = scope @@ -145,6 +144,7 @@ class EmbeddingClient: breaker=self._breaker, backpressure=backpressure, quota_full=quota_full, + circuit_open=circuit_open, now=now, sleep=sleep, rng=rng, @@ -500,6 +500,7 @@ class EmbeddingClient: retry=gw.retry, backpressure=gw.backpressure, quota_full=gw.quota_full, + circuit_open=gw.circuit_open, telemetry=telemetry if telemetry is not None else _build_telemetry(gw), pricing=PricingTable.from_file(gw.pricing_path) if gw.pricing_path is not None diff --git a/src/polygateway/middleware/admission.py b/src/polygateway/middleware/admission.py index a817523..8997784 100644 --- a/src/polygateway/middleware/admission.py +++ b/src/polygateway/middleware/admission.py @@ -29,6 +29,9 @@ from loguru import logger from polygateway.errors import AllSourcesExhausted, CircuitOpenError from polygateway.sources import SourceCooldownMemo +# 两个准入策略键共用的值域;校验只此一处,不在各客户端重复 +_POLICIES = frozenset({"wait", "fail_fast"}) + if TYPE_CHECKING: from collections.abc import Callable @@ -137,6 +140,7 @@ class SourceAdmission: breaker: BreakerGate, backpressure: BackpressurePolicy, quota_full: str, + circuit_open: str, memo: SourceCooldownMemo | None = None, pacer: AdaptivePacer | None = None, health_view: Callable[[str], float] | None = None, @@ -144,8 +148,9 @@ class SourceAdmission: 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}") + for name, value in (("quota_full", quota_full), ("circuit_open", circuit_open)): + if value not in _POLICIES: + raise ValueError(f"{name} 必须是 wait|fail_fast: {value!r}") self._scope = scope self._sources = sources self._selector = selector @@ -153,6 +158,7 @@ class SourceAdmission: self._breaker = breaker self._bp = backpressure self._quota_full = quota_full + self._circuit_open = circuit_open self._memo = memo or SourceCooldownMemo(now=now) self._pacer = pacer self._health_view = health_view diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index c30f2ab..0e9b00c 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -178,6 +178,7 @@ class RetryMW: retry: RetryPolicy, backpressure: BackpressurePolicy, quota_full: str = "wait", + circuit_open: str = "fail_fast", cooldown_memo: SourceCooldownMemo | None = None, pacer: AdaptivePacer | None = None, emitter: object | None = None, @@ -185,8 +186,6 @@ class RetryMW: sleep: Callable[[float], Awaitable[None]] = 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 = list(sources) # 记账写回与 pacer 结算仍在 `_attempt` 内,故这三者由本类持有并与 @@ -212,6 +211,7 @@ class RetryMW: breaker=self._breaker, backpressure=backpressure, quota_full=quota_full, + circuit_open=circuit_open, memo=cooldown_memo, pacer=self._pacer, health_view=self._outcome_sink.health if self._outcome_sink else None, diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index bd06daa..4f6da20 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -107,14 +107,13 @@ class OcrClient: retry: RetryPolicy, backpressure: BackpressurePolicy, quota_full: str = "wait", + circuit_open: str = "fail_fast", telemetry: TelemetryRecorder | None = None, text_cap: int | None = None, now: Callable[[], float] = time.monotonic, sleep: Callable[[float], Awaitable[None]] = 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 # MonkeyOCR 只发 multipart 表单,带 extra_body 的源必须先剥离,否则 # 遥测会记录一个从未发出的采样参数(issue #4 决策 G) @@ -139,6 +138,7 @@ class OcrClient: breaker=self._breaker, backpressure=backpressure, quota_full=quota_full, + circuit_open=circuit_open, now=now, sleep=sleep, rng=rng, @@ -513,6 +513,7 @@ class OcrClient: retry=gw.retry, backpressure=gw.backpressure, quota_full=gw.quota_full, + circuit_open=gw.circuit_open, telemetry=telemetry if telemetry is not None else _build_telemetry(gw), # OCR 行与 chat 行写同一张 llm_calls;漏传这一条,同表内就一半受控 # 一半不受控(issue #12) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 4f9772f..74011eb 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -170,6 +170,20 @@ class TestResilienceKeys: with pytest.raises(ValueError, match="probe"): GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "45"})) + def test_circuit_open_defaults_to_fail_fast(self): + """issue #14: 熔断拒绝的处置策略。 + + 缺省**不跟随** quota_full 的 wait——把最坏墙钟从毫秒抬到 stall 窗口 + 是"快速失败 → 长时间挂起"这个最危险的方向,不能强加给存量下游。 + """ + assert GatewaySettings.from_env("LLM", env=_env()).circuit_open == "fail_fast" + waiting = GatewaySettings.from_env( + "LLM", env=_env(**{"LLM__CIRCUIT_OPEN": "wait"}) + ) + assert waiting.circuit_open == "wait" + with pytest.raises(ValueError, match="CIRCUIT_OPEN"): + GatewaySettings.from_env("LLM", env=_env(**{"LLM__CIRCUIT_OPEN": "block"})) + def test_selector_and_quota_full(self): # M2.5: 缺省选源改 health_aware(生产级默认);显式配置者不变 s = GatewaySettings.from_env("LLM", env=_env()) @@ -589,6 +603,7 @@ class TestCrossFieldInvariants: ("telemetry_backend", "redis"), ("selector", "random"), ("quota_full", "block"), + ("circuit_open", "block"), ], ) def test_enum_field_rejects_value_outside_domain(self, field, bad_value):