iomgaa/PolyGateway
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases 9 Wiki Activity

feat: unlock redis/postgres backend config with assembly guards

Browse Source
This commit is contained in:
iomgaa
2026-07-21 00:18:42 -04:00
parent f9677fbd8e
commit acc1bcc18a
2 changed files with 107 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/polygateway/config.py
+36 -6
View File
@@ -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
tests/unit/test_config.py
+71 -6
View File
@@ -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