diff --git a/src/polygateway/config.py b/src/polygateway/config.py index 6908c66..6483337 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -51,6 +51,7 @@ _QUOTA_FULL = frozenset({"wait", "fail_fast"}) _DEFAULT_STALL_WINDOW_S = 300.0 _DEFAULT_POLL_INTERVAL_S = 0.05 _DEFAULT_LEASE_TTL_S = 1500.0 # CHS _DEFAULT_LEASE_TTL_MS 同源 +_PROBE_GRACE_S = 5.0 # 半开探针租约相对最慢调用的清理宽限(CHS container.py:274-275) def _cast(raw: str, kind: str, key: str) -> object: @@ -88,7 +89,16 @@ def _require(env: Mapping[str, str], *keys: str) -> tuple[str, str]: @dataclass(frozen=True) class GatewaySettings: - """一个 scope 的完整装配配置;构造经 from_env 聚合并通过全部守卫。""" + """一个 scope 的完整装配配置;**任何**构造路径都通过全部装配守卫(ARCH §7.3)。 + + 守卫校验的是**跨字段**不变量: 单看一个字段都合法,组合起来才会在运行时 + 咬人(租约先于请求过期、正常慢首包被误判卡死、半开探针在途被接管)。 + types.py 各子配置的 `__post_init__` 只看得见自己的字段,故由本类把关。 + + 放在 `__post_init__` 而非某个工厂里: 这些约束是本类定义的一部分,不是 + 某个入口的输入检查。挂在构造期,直接构造、`dataclasses.replace` 与全部 + 装配工厂一并覆盖;挂在工厂里则每加一个工厂就多一处要同步。 + """ scope: str sources: tuple[SourceConfig, ...] @@ -111,6 +121,44 @@ class GatewaySettings: structured_max_retries: int lease_ttl_s: float + def __post_init__(self) -> None: + self._validate_sources() + self._validate_lease() + self._validate_stall() + self._validate_probe() + + def _validate_sources(self) -> None: + """零源的配置装出来选源必然失败,构造期即拒。""" + if not self.sources: + raise ValueError("GatewaySettings.sources 不能为空: 至少一个源") + + def _validate_lease(self) -> None: + """调用超时须 ≤ permit 租约 TTL,防租约先于请求过期使并发超出配额。""" + slowest = max(s.timeout_s for s in self.sources) + if slowest > self.lease_ttl_s: + raise ValueError( + f"源最大 timeout_s({slowest})超过 permit 租约 lease_ttl_s" + f"({self.lease_ttl_s});调大 lease_ttl_s 或调小源的 timeout_s" + ) + + def _validate_stall(self) -> None: + """stall 窗口须 ≥ 最慢源 TTFT 上限,防把正常慢首包误判为卡死。""" + ttfts = [s.ttft_timeout_s for s in self.sources if s.ttft_timeout_s is not None] + if ttfts and self.backpressure.stall_window_s < max(ttfts): + raise ValueError( + f"backpressure.stall_window_s({self.backpressure.stall_window_s})须 ≥ " + f"最大源 ttft_timeout_s({max(ttfts)});调大 stall_window_s 或调小 ttft_timeout_s" + ) + + def _validate_probe(self) -> None: + """半开探针租约须撑过一次最慢调用,否则探针在途即被接管(M2 设计 §3)。""" + floor = max(s.timeout_s for s in self.sources) + _PROBE_GRACE_S + if self.breaker.probe_ttl_s < floor: + raise ValueError( + f"breaker.probe_ttl_s({self.breaker.probe_ttl_s})须 ≥ 最慢源 " + f"timeout_s + {_PROBE_GRACE_S}({floor});调大 probe_ttl_s 或调小源的 timeout_s" + ) + @classmethod def from_env( cls, @@ -119,7 +167,7 @@ class GatewaySettings: *, env_file: str = ".env", ) -> GatewaySettings: - """聚合 env(缺省 .env + os.environ,后者优先)并执行装配守卫。""" + """聚合 env(缺省 .env + os.environ,后者优先);守卫由 `__post_init__` 执行。""" if env is None: env = { k: v for k, v in {**dotenv_values(env_file), **os.environ}.items() if v is not None @@ -129,7 +177,7 @@ class GatewaySettings: global_limits = _load_global_limits(scope_u, env) retry = _load_retry(scope_u, env) breaker = _load_breaker(scope_u, env, sources, global_limits) - settings = cls( + return cls( scope=scope_u.lower(), sources=tuple(sources), global_limits=global_limits, @@ -140,9 +188,6 @@ class GatewaySettings: quota_full=_load_choice(env, f"{scope_u}__QUOTA_FULL", _QUOTA_FULL, "wait"), **_load_pgw(env), ) - _guard_lease(settings) - _guard_stall(settings) - return settings def _load_sources(scope: str, env: Mapping[str, str]) -> list[SourceConfig]: @@ -217,16 +262,12 @@ def _load_breaker( if concurrency > 0: threshold = max(threshold, concurrency * 2) slowest = max(s.timeout_s for s in sources) - probe_floor = slowest + 5.0 # CHS container.py:274-275: 最慢调用 + 清理宽限 + probe_floor = slowest + _PROBE_GRACE_S probe = _first(env, f"{scope}__BREAKER__PROBE_TTL_S") if probe is not None: + # 配置值不在此校验: 探针租约下限是跨字段不变量,由 GatewaySettings._validate_probe + # 统一把关(否则直接构造那条装配路会绕过) probe_ttl_s = float(_cast(probe[1], "float", probe[0])) - # 装配守卫(M2 设计 §3): 探针租约必须撑过一次最慢调用,否则半开探针在途即被接管 - if probe_ttl_s < probe_floor: - raise ValueError( - f"probe_ttl_s({probe_ttl_s})须 ≥ 最大源 timeout_s + 5({probe_floor});" - f"调大 {probe[0]} 或调小源超时" - ) else: # 派生规则: 探针租约须撑过一次最慢调用,且不短于冷却期(第三项保证守卫恒成立) probe_ttl_s = max(2 * slowest, cooldown_s, probe_floor) @@ -339,26 +380,6 @@ def _load_structured_retries(env: Mapping[str, str]) -> int: return value -def _guard_lease(settings: GatewaySettings) -> None: - """装配守卫: 调用超时须 ≤ permit 租约 TTL,防租约先于请求过期(ARCH §7.3)。""" - slowest = max(s.timeout_s for s in settings.sources) - if slowest > settings.lease_ttl_s: - raise ValueError( - f"源最大 timeout_s({slowest})超过 permit 租约 TTL({settings.lease_ttl_s});" - f"调大 PGW_LEASE_TTL_S 或调小超时" - ) - - -def _guard_stall(settings: GatewaySettings) -> None: - """装配守卫: stall 窗口须 ≥ 最慢源 TTFT 上限,防把正常慢首包误判为卡死(ARCH §7.3)。""" - ttfts = [s.ttft_timeout_s for s in settings.sources if s.ttft_timeout_s is not None] - if ttfts and settings.backpressure.stall_window_s < max(ttfts): - raise ValueError( - f"stall_window_s({settings.backpressure.stall_window_s})须 ≥ 最大源 " - f"ttft_timeout_s({max(ttfts)});调大 BACKPRESSURE__STALL_WINDOW_S 或调小 TTFT" - ) - - def _load_lease_ttl(env: Mapping[str, str]) -> float: found = _first(env, "PGW_LEASE_TTL_S") return float(_cast(found[1], "float", found[0])) if found else _DEFAULT_LEASE_TTL_S diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 5d513ad..b7fd079 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -1,8 +1,11 @@ """config.py 配置聚合测试(设计 §8): 多源命名、键优先级、缺失报错。""" +import dataclasses + import pytest -from polygateway.config import GatewaySettings +from polygateway.client import GatewayClient +from polygateway.config import GatewaySettings, OcrSettings _BASE_ENV = { "LLM__QWEN__1__BASE_URL": "https://gw-a.example/v1", @@ -319,3 +322,94 @@ class TestOcrSettings: env = {k: v for k, v in self._OCR_ENV.items() if k != "OCR__MONKEY__1__BASE_URL"} with pytest.raises(ValueError): OcrSettings.from_env("OCR", env=env) + + +class TestCrossFieldInvariants: + """四条跨字段不变量必须在**任何**构造路径上生效(设计 2026-07-29)。 + + 这些约束单看一个字段都合法,组合起来才非法,因此 types.py 各子配置的 + __post_init__ 看不见——只能由聚合层 GatewaySettings 把关。守卫若只挂在 + from_env 上,from_settings 这条同等官方的装配路(CLAUDE.md §4.5)就能 + 装出违反不变量的配置,类会存在于自己 docstring 声称不可能的状态。 + + 每条不变量测两侧: 越界必拒、边界值(恰好相等)必过——收紧的是错的组合, + 不是所有直接构造。 + """ + + def _base(self, **overrides) -> GatewaySettings: + return GatewaySettings.from_env("LLM", env=_env(**overrides)) + + def _with_watchdog(self) -> GatewaySettings: + """带看门狗的基准: TTFT/inter-token 成对配置才满足 SourceConfig 不变式。""" + return self._base( + **{ + "LLM__QWEN__1__TTFT_TIMEOUT_S": "30", + "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15", + "LLM__BACKPRESSURE__STALL_WINDOW_S": "300", + } + ) + + # —— 源超时 ≤ permit 租约 TTL(ARCH §7.3: 防租约先于请求过期,并发悄悄超配额)—— + + def test_lease_rejects_timeout_above_ttl_on_direct_construction(self): + base = self._base() # 源 timeout_s=120 + with pytest.raises(ValueError, match="lease_ttl_s"): + dataclasses.replace(base, lease_ttl_s=1.0) + + def test_lease_accepts_timeout_equal_to_ttl(self): + base = self._base() + assert dataclasses.replace(base, lease_ttl_s=120.0).lease_ttl_s == 120.0 + + # —— stall 窗口 ≥ 最大源 TTFT(ARCH §7.3: 防正常慢首包被误判卡死掐断)—— + + def test_stall_rejects_window_below_max_ttft_on_direct_construction(self): + base = self._with_watchdog() # 源 ttft_timeout_s=30 + narrowed = dataclasses.replace(base.backpressure, stall_window_s=20.0) + with pytest.raises(ValueError, match="stall_window_s"): + dataclasses.replace(base, backpressure=narrowed) + + def test_stall_accepts_window_equal_to_max_ttft(self): + base = self._with_watchdog() + exact = dataclasses.replace(base.backpressure, stall_window_s=30.0) + assert dataclasses.replace(base, backpressure=exact).backpressure.stall_window_s == 30.0 + + # —— 探针租约 ≥ 最慢源超时 + 5(M2 设计 §3: 防半开探针在途即被接管)—— + + def test_probe_rejects_ttl_below_floor_on_direct_construction(self): + base = self._base() # 最慢 timeout_s=120,故下限 125 + shortened = dataclasses.replace(base.breaker, probe_ttl_s=100.0) + with pytest.raises(ValueError, match="probe_ttl_s"): + dataclasses.replace(base, breaker=shortened) + + def test_probe_accepts_ttl_at_floor(self): + base = self._base() + at_floor = dataclasses.replace(base.breaker, probe_ttl_s=125.0) + assert dataclasses.replace(base, breaker=at_floor).breaker.probe_ttl_s == 125.0 + + # —— sources 非空 —— + + def test_empty_sources_rejected_with_actionable_message(self): + """零源装出来的 client 选源必然失败;消息须点明原因,不能泄漏 max() 的内置异常。""" + base = self._base() + with pytest.raises(ValueError) as exc: + dataclasses.replace(base, sources=()) + assert "至少一个源" in str(exc.value) + assert "empty sequence" not in str(exc.value) + + # —— 装配路径覆盖 —— + + def test_factory_cannot_receive_invalid_settings(self): + """from_settings 这条路吃不到非法配置。 + + 异常实际抛在实参求值(构造 settings)那一刻,而不是工厂内部——这正是 + 把守卫放构造期换来的性质: 非法实例根本不存在,无需每个工厂各自设防。 + """ + base = self._base() + with pytest.raises(ValueError, match="lease_ttl_s"): + GatewayClient.from_settings(dataclasses.replace(base, lease_ttl_s=1.0)) + + def test_ocr_settings_cannot_wrap_invalid_gateway(self): + """OcrSettings/EmbeddingSettings 只是包一层 GatewaySettings,自动继承同一把关。""" + base = self._base() + with pytest.raises(ValueError, match="lease_ttl_s"): + OcrSettings(gateway=dataclasses.replace(base, lease_ttl_s=1.0))