feat: add the {SCOPE}__CIRCUIT_OPEN admission policy key

Limiter rejections have always chosen between waiting and failing fast;
breaker rejections had no such choice. The new key is the missing cell
of that matrix, shaped exactly like QUOTA_FULL so there is nothing new
to learn. It defaults to fail_fast: flipping the default would move
every existing deployment's worst-case wall clock from milliseconds to
the stall window, which is the wrong direction to impose on anyone.
Single-source scopes are the ones that want wait, and they now have a
way to say so.

The two keys stay separate despite sharing a domain, because a full
quota is "queue for your share" (your turn always comes) while an open
circuit is "wait for the source to recover" (it might not).

Policy validation collapses into SourceAdmission, the only consumer.
The three client constructors used to each carry their own copy of the
quota_full check; adding a second key there would have made eight
copies of the same two lines. Rejection timing and message are
unchanged -- admission is built inside those constructors.

This commit only wires the key through; the control flow that reads it
lands next.
This commit is contained in:
2026-08-20 00:17:46 -04:00
parent 8edd3fb2cd
commit eb956b2cdf
7 changed files with 44 additions and 8 deletions
+3
View File
@@ -134,6 +134,7 @@ class GatewayClient:
retry: RetryPolicy, retry: RetryPolicy,
backpressure: BackpressurePolicy, backpressure: BackpressurePolicy,
quota_full: str = "wait", quota_full: str = "wait",
circuit_open: str = "fail_fast",
telemetry: TelemetryRecorder | None = None, telemetry: TelemetryRecorder | None = None,
pricing: PricingTable | None = None, pricing: PricingTable | None = None,
text_cap: int | None = None, text_cap: int | None = None,
@@ -162,6 +163,7 @@ class GatewayClient:
retry=retry, retry=retry,
backpressure=backpressure, backpressure=backpressure,
quota_full=quota_full, quota_full=quota_full,
circuit_open=circuit_open,
cooldown_memo=SourceCooldownMemo(now=now), cooldown_memo=SourceCooldownMemo(now=now),
# AIMD ceiling 尊重源级静态并发上限(独立核验 I1: 不得静默钳制大于 64 的配置) # AIMD ceiling 尊重源级静态并发上限(独立核验 I1: 不得静默钳制大于 64 的配置)
pacer=AdaptivePacer( pacer=AdaptivePacer(
@@ -313,6 +315,7 @@ class GatewayClient:
retry=settings.retry, retry=settings.retry,
backpressure=settings.backpressure, backpressure=settings.backpressure,
quota_full=settings.quota_full, quota_full=settings.quota_full,
circuit_open=settings.circuit_open,
telemetry=telemetry if telemetry is not None else _build_telemetry(settings), telemetry=telemetry if telemetry is not None else _build_telemetry(settings),
pricing=PricingTable.from_file(settings.pricing_path) pricing=PricingTable.from_file(settings.pricing_path)
if settings.pricing_path is not None if settings.pricing_path is not None
+10
View File
@@ -50,6 +50,9 @@ _SOURCE_FIELDS: dict[str, tuple[str, str]] = {
_RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE"}) _RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE"})
_SELECTORS = frozenset({"round_robin", "least_inflight", "health_aware"}) _SELECTORS = frozenset({"round_robin", "least_inflight", "health_aware"})
_QUOTA_FULL = frozenset({"wait", "fail_fast"}) _QUOTA_FULL = frozenset({"wait", "fail_fast"})
# 熔断全拒时的处置(issue #14);值域与 _QUOTA_FULL 相同但语义不同——配额满是
# "排队等自己的份额"(必然轮到),熔断开路是"等源恢复"(未必恢复),故分列两键
_CIRCUIT_OPEN = frozenset({"wait", "fail_fast"})
# 后端合法域: env 解析与构造期校验共用一份定义,避免两处分叉 # 后端合法域: env 解析与构造期校验共用一份定义,避免两处分叉
_LIMITER_BACKENDS = frozenset({"memory", "redis"}) _LIMITER_BACKENDS = frozenset({"memory", "redis"})
_BREAKER_BACKENDS = frozenset({"memory", "redis"}) _BREAKER_BACKENDS = frozenset({"memory", "redis"})
@@ -128,6 +131,9 @@ class GatewaySettings:
backpressure: BackpressurePolicy backpressure: BackpressurePolicy
selector: str selector: str
quota_full: str quota_full: str
# 熔断全拒时是当场判死还是等冷却过去(issue #14);缺省 fail_fast 保持
# 存量下游的控制流不变,单源 scope 应显式配 wait
circuit_open: str
limiter_backend: str limiter_backend: str
breaker_backend: str breaker_backend: str
cache_backend: str cache_backend: str
@@ -203,6 +209,7 @@ class GatewaySettings:
("telemetry_backend", _TELEMETRY_BACKENDS), ("telemetry_backend", _TELEMETRY_BACKENDS),
("selector", _SELECTORS), ("selector", _SELECTORS),
("quota_full", _QUOTA_FULL), ("quota_full", _QUOTA_FULL),
("circuit_open", _CIRCUIT_OPEN),
): ):
value = getattr(self, field) value = getattr(self, field)
if value not in allowed: if value not in allowed:
@@ -320,6 +327,9 @@ class GatewaySettings:
backpressure=_load_backpressure(scope_u, env), backpressure=_load_backpressure(scope_u, env),
selector=_load_choice(env, f"{scope_u}__SELECTOR", _SELECTORS, "health_aware"), selector=_load_choice(env, f"{scope_u}__SELECTOR", _SELECTORS, "health_aware"),
quota_full=_load_choice(env, f"{scope_u}__QUOTA_FULL", _QUOTA_FULL, "wait"), 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), **_load_pgw(env),
) )
+3 -2
View File
@@ -101,6 +101,7 @@ class EmbeddingClient:
retry: RetryPolicy, retry: RetryPolicy,
backpressure: BackpressurePolicy, backpressure: BackpressurePolicy,
quota_full: str = "wait", quota_full: str = "wait",
circuit_open: str = "fail_fast",
telemetry: TelemetryRecorder | None = None, telemetry: TelemetryRecorder | None = None,
pricing: PricingTable | None = None, pricing: PricingTable | None = None,
text_cap: int | None = None, text_cap: int | None = None,
@@ -113,8 +114,6 @@ class EmbeddingClient:
) -> None: ) -> None:
if batch_size < 1: if batch_size < 1:
raise ValueError("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: if expected_dim is not None and expected_dim < 1:
raise ValueError("expected_dim 必须 ≥ 1") raise ValueError("expected_dim 必须 ≥ 1")
self._scope = scope self._scope = scope
@@ -145,6 +144,7 @@ class EmbeddingClient:
breaker=self._breaker, breaker=self._breaker,
backpressure=backpressure, backpressure=backpressure,
quota_full=quota_full, quota_full=quota_full,
circuit_open=circuit_open,
now=now, now=now,
sleep=sleep, sleep=sleep,
rng=rng, rng=rng,
@@ -500,6 +500,7 @@ class EmbeddingClient:
retry=gw.retry, retry=gw.retry,
backpressure=gw.backpressure, backpressure=gw.backpressure,
quota_full=gw.quota_full, quota_full=gw.quota_full,
circuit_open=gw.circuit_open,
telemetry=telemetry if telemetry is not None else _build_telemetry(gw), telemetry=telemetry if telemetry is not None else _build_telemetry(gw),
pricing=PricingTable.from_file(gw.pricing_path) pricing=PricingTable.from_file(gw.pricing_path)
if gw.pricing_path is not None if gw.pricing_path is not None
+8 -2
View File
@@ -29,6 +29,9 @@ from loguru import logger
from polygateway.errors import AllSourcesExhausted, CircuitOpenError from polygateway.errors import AllSourcesExhausted, CircuitOpenError
from polygateway.sources import SourceCooldownMemo from polygateway.sources import SourceCooldownMemo
# 两个准入策略键共用的值域;校验只此一处,不在各客户端重复
_POLICIES = frozenset({"wait", "fail_fast"})
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Callable from collections.abc import Callable
@@ -137,6 +140,7 @@ class SourceAdmission:
breaker: BreakerGate, breaker: BreakerGate,
backpressure: BackpressurePolicy, backpressure: BackpressurePolicy,
quota_full: str, quota_full: str,
circuit_open: str,
memo: SourceCooldownMemo | None = None, memo: SourceCooldownMemo | None = None,
pacer: AdaptivePacer | None = None, pacer: AdaptivePacer | None = None,
health_view: Callable[[str], float] | None = None, health_view: Callable[[str], float] | None = None,
@@ -144,8 +148,9 @@ class SourceAdmission:
sleep: Callable[[float], object] = asyncio.sleep, sleep: Callable[[float], object] = asyncio.sleep,
rng: Callable[[], float] = random.random, rng: Callable[[], float] = random.random,
) -> None: ) -> None:
if quota_full not in ("wait", "fail_fast"): for name, value in (("quota_full", quota_full), ("circuit_open", circuit_open)):
raise ValueError(f"quota_full 必须是 wait|fail_fast: {quota_full!r}") if value not in _POLICIES:
raise ValueError(f"{name} 必须是 wait|fail_fast: {value!r}")
self._scope = scope self._scope = scope
self._sources = sources self._sources = sources
self._selector = selector self._selector = selector
@@ -153,6 +158,7 @@ class SourceAdmission:
self._breaker = breaker self._breaker = breaker
self._bp = backpressure self._bp = backpressure
self._quota_full = quota_full self._quota_full = quota_full
self._circuit_open = circuit_open
self._memo = memo or SourceCooldownMemo(now=now) self._memo = memo or SourceCooldownMemo(now=now)
self._pacer = pacer self._pacer = pacer
self._health_view = health_view self._health_view = health_view
+2 -2
View File
@@ -178,6 +178,7 @@ class RetryMW:
retry: RetryPolicy, retry: RetryPolicy,
backpressure: BackpressurePolicy, backpressure: BackpressurePolicy,
quota_full: str = "wait", quota_full: str = "wait",
circuit_open: str = "fail_fast",
cooldown_memo: SourceCooldownMemo | None = None, cooldown_memo: SourceCooldownMemo | None = None,
pacer: AdaptivePacer | None = None, pacer: AdaptivePacer | None = None,
emitter: object | None = None, emitter: object | None = None,
@@ -185,8 +186,6 @@ class RetryMW:
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
rng: Callable[[], float] = random.random, rng: Callable[[], float] = random.random,
) -> None: ) -> None:
if quota_full not in ("wait", "fail_fast"):
raise ValueError(f"quota_full 必须是 wait|fail_fast: {quota_full!r}")
self._scope = scope self._scope = scope
self._sources = list(sources) self._sources = list(sources)
# 记账写回与 pacer 结算仍在 `_attempt` 内,故这三者由本类持有并与 # 记账写回与 pacer 结算仍在 `_attempt` 内,故这三者由本类持有并与
@@ -212,6 +211,7 @@ class RetryMW:
breaker=self._breaker, breaker=self._breaker,
backpressure=backpressure, backpressure=backpressure,
quota_full=quota_full, quota_full=quota_full,
circuit_open=circuit_open,
memo=cooldown_memo, memo=cooldown_memo,
pacer=self._pacer, pacer=self._pacer,
health_view=self._outcome_sink.health if self._outcome_sink else None, health_view=self._outcome_sink.health if self._outcome_sink else None,
+3 -2
View File
@@ -107,14 +107,13 @@ class OcrClient:
retry: RetryPolicy, retry: RetryPolicy,
backpressure: BackpressurePolicy, backpressure: BackpressurePolicy,
quota_full: str = "wait", quota_full: str = "wait",
circuit_open: str = "fail_fast",
telemetry: TelemetryRecorder | None = None, telemetry: TelemetryRecorder | None = None,
text_cap: int | None = None, text_cap: int | None = None,
now: Callable[[], float] = time.monotonic, now: Callable[[], float] = time.monotonic,
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
rng: Callable[[], float] = random.random, rng: Callable[[], float] = random.random,
) -> None: ) -> None:
if quota_full not in ("wait", "fail_fast"):
raise ValueError(f"quota_full 必须是 wait|fail_fast: {quota_full!r}")
self._scope = scope self._scope = scope
# MonkeyOCR 只发 multipart 表单,带 extra_body 的源必须先剥离,否则 # MonkeyOCR 只发 multipart 表单,带 extra_body 的源必须先剥离,否则
# 遥测会记录一个从未发出的采样参数(issue #4 决策 G) # 遥测会记录一个从未发出的采样参数(issue #4 决策 G)
@@ -139,6 +138,7 @@ class OcrClient:
breaker=self._breaker, breaker=self._breaker,
backpressure=backpressure, backpressure=backpressure,
quota_full=quota_full, quota_full=quota_full,
circuit_open=circuit_open,
now=now, now=now,
sleep=sleep, sleep=sleep,
rng=rng, rng=rng,
@@ -513,6 +513,7 @@ class OcrClient:
retry=gw.retry, retry=gw.retry,
backpressure=gw.backpressure, backpressure=gw.backpressure,
quota_full=gw.quota_full, quota_full=gw.quota_full,
circuit_open=gw.circuit_open,
telemetry=telemetry if telemetry is not None else _build_telemetry(gw), telemetry=telemetry if telemetry is not None else _build_telemetry(gw),
# OCR 行与 chat 行写同一张 llm_calls;漏传这一条,同表内就一半受控 # OCR 行与 chat 行写同一张 llm_calls;漏传这一条,同表内就一半受控
# 一半不受控(issue #12) # 一半不受控(issue #12)
+15
View File
@@ -170,6 +170,20 @@ class TestResilienceKeys:
with pytest.raises(ValueError, match="probe"): with pytest.raises(ValueError, match="probe"):
GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "45"})) 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): def test_selector_and_quota_full(self):
# M2.5: 缺省选源改 health_aware(生产级默认);显式配置者不变 # M2.5: 缺省选源改 health_aware(生产级默认);显式配置者不变
s = GatewaySettings.from_env("LLM", env=_env()) s = GatewaySettings.from_env("LLM", env=_env())
@@ -589,6 +603,7 @@ class TestCrossFieldInvariants:
("telemetry_backend", "redis"), ("telemetry_backend", "redis"),
("selector", "random"), ("selector", "random"),
("quota_full", "block"), ("quota_full", "block"),
("circuit_open", "block"),
], ],
) )
def test_enum_field_rejects_value_outside_domain(self, field, bad_value): def test_enum_field_rejects_value_outside_domain(self, field, bad_value):