From acc1bcc18ad54c29564481bcc813bc756c6172f5 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 21 Jul 2026 00:18:42 -0400 Subject: [PATCH] feat: unlock redis/postgres backend config with assembly guards --- src/polygateway/config.py | 42 ++++++++++++++++++--- tests/unit/test_config.py | 77 ++++++++++++++++++++++++++++++++++++--- 2 files changed, 107 insertions(+), 12 deletions(-) diff --git a/src/polygateway/config.py b/src/polygateway/config.py index 92662b8..8d3f141 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -105,7 +105,9 @@ class GatewaySettings: cache_ttl_s: int | None telemetry_backend: str telemetry_sqlite_path: str | None + telemetry_pg_dsn: str | None redis_url: str | None + pricing_path: str | None structured_max_retries: int lease_ttl_s: float @@ -139,6 +141,7 @@ class GatewaySettings: **_load_pgw(env), ) _guard_lease(settings) + _guard_stall(settings) return settings @@ -214,12 +217,20 @@ 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 = _first(env, f"{scope}__BREAKER__PROBE_TTL_S") if probe is not None: 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 * max(s.timeout_s for s in sources), cooldown_s) + # 派生规则: 探针租约须撑过一次最慢调用,且不短于冷却期(第三项保证守卫恒成立) + probe_ttl_s = max(2 * slowest, cooldown_s, probe_floor) return BreakerConfig(fail_threshold=threshold, cooldown_s=cooldown_s, probe_ttl_s=probe_ttl_s) @@ -251,15 +262,15 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]: breaker_backend = _load_choice( env, "PGW_BREAKER_BACKEND", frozenset({"memory", "redis"}), "memory" ) - if "redis" in (limiter_backend, breaker_backend): - raise ValueError("限流/熔断 Redis 后端在 M2 交付;M1 仅支持 memory") _, cache_backend = _require(env, "PGW_CACHE_BACKEND") _, telemetry_backend = _require(env, "PGW_TELEMETRY_BACKEND") if cache_backend not in ("redis", "memory", "none"): raise ValueError(f"PGW_CACHE_BACKEND 非法值 {cache_backend!r}") - if telemetry_backend not in ("sqlite", "none"): - raise ValueError(f"PGW_TELEMETRY_BACKEND 非法值 {telemetry_backend!r}(postgres 在 M2)") + if telemetry_backend not in ("sqlite", "postgres", "none"): + raise ValueError(f"PGW_TELEMETRY_BACKEND 非法值 {telemetry_backend!r}") redis_url = env.get("REDIS_URL") or None + if "redis" in (limiter_backend, breaker_backend) and redis_url is None: + raise ValueError("缺关键配置: 限流/熔断后端取 redis 需设置 REDIS_URL") return { "limiter_backend": limiter_backend, "breaker_backend": breaker_backend, @@ -269,12 +280,21 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]: "telemetry_sqlite_path": _require(env, "PGW_TELEMETRY_SQLITE_PATH")[1] if telemetry_backend == "sqlite" else None, + "telemetry_pg_dsn": _load_pg_dsn(env) if telemetry_backend == "postgres" else None, "redis_url": redis_url, + "pricing_path": env.get("PGW_PRICING_PATH") or None, "structured_max_retries": _load_structured_retries(env), "lease_ttl_s": _load_lease_ttl(env), } +def _load_pg_dsn(env: Mapping[str, str]) -> str: + """读取 Postgres DSN 并剥 SQLAlchemy 风格驱动后缀(asyncpg 不认 `+driver`)。""" + _, dsn = _require(env, "PGW_TELEMETRY_PG_DSN") + scheme, sep, rest = dsn.partition("://") + return f"{scheme.partition('+')[0]}{sep}{rest}" + + def _load_cache_keys( env: Mapping[str, str], cache_backend: str, redis_url: str | None ) -> dict[str, object]: @@ -308,6 +328,16 @@ def _guard_lease(settings: GatewaySettings) -> None: ) +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 9908277..824986b 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -106,10 +106,24 @@ class TestResilienceKeys: def test_probe_ttl_derived_when_absent(self): s = GatewaySettings.from_env("LLM", env=_env()) - # 派生规则: max(2 × 最大源 timeout, cooldown) - assert s.breaker.probe_ttl_s == max(2 * 120.0, 60.0) - s2 = GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "45"})) - assert s2.breaker.probe_ttl_s == 45.0 + # 派生规则(M2 补第三项): max(2 × 最大源 timeout, cooldown, 最大源 timeout + 5) + assert s.breaker.probe_ttl_s == max(2 * 120.0, 60.0, 120.0 + 5) + s2 = GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "300"})) + assert s2.breaker.probe_ttl_s == 300.0 + + def test_probe_ttl_derivation_third_term_wins(self): + # timeout=4、cooldown=2 → max(8, 2, 9) = 9;守卫 9 ≥ 4+5 恰好成立不报错 + env = _env( + LLM_CIRCUIT_BREAKER_COOLDOWN="2", + **{"LLM__QWEN__1__TIMEOUT_S": "4"}, + ) + s = GatewaySettings.from_env("LLM", env=env) + assert s.breaker.probe_ttl_s == 9.0 + + def test_explicit_probe_ttl_below_guard_rejected(self): + # 守卫: probe_ttl_s ≥ max(timeout_s) + 5(CHS container 语义,M2 设计 §3) + with pytest.raises(ValueError, match="probe"): + GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "45"})) def test_selector_and_quota_full(self): s = GatewaySettings.from_env("LLM", env=_env()) @@ -161,6 +175,57 @@ class TestAssemblyGuards: # 有效阈值 = max(配置值 5, 并发 8 × 2) = 16(.env 注释约定入库) assert s.breaker.fail_threshold == 16 - def test_m1_only_memory_governance_backends(self): - with pytest.raises(ValueError, match="M2"): + def test_redis_governance_backend_requires_url(self): + """M2 解禁 redis 后端: 取 redis 时 REDIS_URL 必在,缺则装配报错。""" + env = _env( + PGW_LIMITER_BACKEND="redis", + PGW_BREAKER_BACKEND="redis", + REDIS_URL="redis://:pw@10.0.0.1:6379/3", + ) + s = GatewaySettings.from_env("LLM", env=env) + assert s.limiter_backend == "redis" and s.breaker_backend == "redis" + with pytest.raises(ValueError, match="REDIS_URL"): GatewaySettings.from_env("LLM", env=_env(PGW_LIMITER_BACKEND="redis")) + + def test_postgres_telemetry_requires_dsn_and_strips_driver_suffix(self): + """M2: telemetry=postgres 需 PGW_TELEMETRY_PG_DSN;SQLAlchemy 风格驱动后缀剥离。""" + with pytest.raises(ValueError, match="PG_DSN"): + GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_BACKEND="postgres")) + env = _env( + PGW_TELEMETRY_BACKEND="postgres", + PGW_TELEMETRY_PG_DSN="postgresql+psycopg://u:p@h:5432/polygateway", + ) + s = GatewaySettings.from_env("LLM", env=env) + assert s.telemetry_pg_dsn == "postgresql://u:p@h:5432/polygateway" + env2 = _env( + PGW_TELEMETRY_BACKEND="postgres", + PGW_TELEMETRY_PG_DSN="postgresql+asyncpg://u:p@h:5432/polygateway", + ) + assert ( + GatewaySettings.from_env("LLM", env=env2).telemetry_pg_dsn + == "postgresql://u:p@h:5432/polygateway" + ) + + def test_telemetry_backend_whitelist(self): + with pytest.raises(ValueError, match="TELEMETRY_BACKEND"): + GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_BACKEND="mysql")) + + def test_pricing_path_optional(self): + assert GatewaySettings.from_env("LLM", env=_env()).pricing_path is None + s = GatewaySettings.from_env("LLM", env=_env(PGW_PRICING_PATH="conf/prices.json")) + assert s.pricing_path == "conf/prices.json" + + def test_stall_window_must_cover_ttft(self): + """守卫: stall_window_s ≥ 最大 ttft_timeout_s(防误判卡死,ARCH §7.3)。""" + env = _env( + **{ + "LLM__QWEN__1__TTFT_TIMEOUT_S": "30", + "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15", + "LLM__BACKPRESSURE__STALL_WINDOW_S": "20", + } + ) + with pytest.raises(ValueError, match="stall"): + GatewaySettings.from_env("LLM", env=env) + env_ok = dict(env) + env_ok["LLM__BACKPRESSURE__STALL_WINDOW_S"] = "60" + assert GatewaySettings.from_env("LLM", env=env_ok).backpressure.stall_window_s == 60.0