feat: derive the schema mode from the telemetry backend

src/polygateway/config.py

@@ -55,6 +55,9 @@ _LIMITER_BACKENDS = frozenset({"memory", "redis"})
 _BREAKER_BACKENDS = frozenset({"memory", "redis"})
 _CACHE_BACKENDS = frozenset({"redis", "memory", "none"})
 _TELEMETRY_BACKENDS = frozenset({"sqlite", "postgres", "none"})
+# 遥测 schema 档位(issue #13): auto 允许 recorder 给旧表 ALTER 补列,manual 不发 DDL
+_SCHEMA_MODES = frozenset({"auto", "manual"})
+_SCHEMA_MODE_KEY = "PGW_TELEMETRY_SCHEMA_MODE"
 _REDIS_DEPENDENT_BACKENDS = ("limiter_backend", "breaker_backend", "cache_backend")
 # 背压默认(M1 仅 poll 生效;CHS _BACKOFF_S=0.05 同源)
 _DEFAULT_STALL_WINDOW_S = 300.0
@@ -131,8 +134,10 @@ class GatewaySettings:
     telemetry_backend: str
     telemetry_sqlite_path: str | None
     telemetry_pg_dsn: str | None
-    # 是否允许 recorder 给已存在的旧表自动 ALTER 补列(issue #13);
-    # 派生规则只写在 `_load_pgw` 一处,不与 recorder 的类签名漂移
+    # 是否允许 recorder 给已存在的旧表自动 ALTER 补列(issue #13);env 的三态
+    # 派生只写在 `_load_schema_mode` 一处,不与 recorder 的类签名漂移。
+    # backend=none 时恒 False 这条跨字段不变量则由 `_validate_telemetry`
+    # 把关,对直接构造与 `dataclasses.replace` 同样生效
     telemetry_auto_migrate: bool
     redis_url: str | None
     pricing_path: str | None
@@ -214,7 +219,14 @@ class GatewaySettings:
         剥而不是拒: 两条装配路对同一 DSN 应产出同一结果。但不静默——`from_env`
         那条路在 `_load_pg_dsn` 就剥干净了,能走到这里的只有手工构造的调用方,
         他有权知道库动了他给的值。
+
+        `telemetry_auto_migrate` 同理归一化而非报错: backend=none 时根本没有
+        recorder 消费它,True 是个自相矛盾却无害的状态。`from_env` 那条路的派生
+        已经给出 False,归一化是为了直接构造与 `dataclasses.replace` 也一致——
+        不变量挂在构造期,才不用每加一个装配工厂就多一处要同步。
         """
+        if self.telemetry_backend == "none" and self.telemetry_auto_migrate:
+            object.__setattr__(self, "telemetry_auto_migrate", False)
         if self.telemetry_backend == "sqlite" and not self.telemetry_sqlite_path:
             raise ValueError("telemetry_backend=sqlite 时必须提供 telemetry_sqlite_path")
         if self.telemetry_backend != "postgres":
@@ -437,6 +449,7 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]:
     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")
+    auto_migrate = _load_schema_mode(env, telemetry_backend)
     return {
         "limiter_backend": limiter_backend,
         "breaker_backend": breaker_backend,
@@ -447,11 +460,7 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]:
         if telemetry_backend == "sqlite"
         else None,
         "telemetry_pg_dsn": _load_pg_dsn(env) if telemetry_backend == "postgres" else None,
-        # 按后端不对称派生(issue #13): SQLite 是下游自己的本地文件(无 DBA、无迁移
-        # 工具),补列是毫秒级元数据操作;PG 是共享生产表,ALTER 取 ACCESS EXCLUSIVE
-        # 锁会阻塞该表其后所有查询,而遥测是业务路径上的内联 await。
-        # backend=none 时无 recorder 消费该值,派生结果恒 False。
-        "telemetry_auto_migrate": telemetry_backend == "sqlite",
+        "telemetry_auto_migrate": auto_migrate,
         "redis_url": redis_url,
         "pricing_path": env.get("PGW_PRICING_PATH") or None,
         "structured_max_retries": _load_structured_retries(env),
@@ -459,6 +468,34 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]:
     }
 
 
+def _load_schema_mode(env: Mapping[str, str], telemetry_backend: str) -> bool:
+    """把 `PGW_TELEMETRY_SCHEMA_MODE` 的三态解成 `telemetry_auto_migrate`(issue #13)。
+
+    三态: 键未设 → 按后端**不对称**派生;显式 auto/manual → 两侧都可覆盖。
+    不对称的理由是两个后端的风险量级不同: SQLite 是下游自己的本地文件(没有
+    DBA、没有迁移工具、没有第二个系统碰它),ALTER 是毫秒级元数据操作,要求
+    手工跑 SQL 是给零运维场景强加运维步骤;PG 是共享的生产表,ALTER 取
+    ACCESS EXCLUSIVE 锁会排在长事务后阻塞该表其后的所有查询,而遥测是业务
+    路径上的内联 await。
+
+    `_load_choice` 带 default,不能直接用来读这个键——default 会把"未设"和
+    "设成默认值"抹平成同一种,三态就塌回两态,后端派生也就再没机会生效。故
+    先用 `_first` 探"设没设",确认设了才交给 `_load_choice` 做值域校验(错误
+    信息点出 env 键名这件事仍由它负责)。
+
+    Args:
+        env: 已合并的环境映射。
+        telemetry_backend: 已校验过值域的遥测后端名。
+
+    Returns:
+        recorder 是否获准给旧表自动 ALTER 补列;backend=none 时无人消费,
+        构造期守卫会再把它归一化为 False。
+    """
+    if _first(env, _SCHEMA_MODE_KEY) is None:
+        return telemetry_backend == "sqlite"
+    return _load_choice(env, _SCHEMA_MODE_KEY, _SCHEMA_MODES, "auto") == "auto"
+
+
 def _strip_dsn_driver(dsn: str) -> str:
     """剥 SQLAlchemy 风格的 `+driver` 后缀(asyncpg 不认);已干净的原样返回。"""
     scheme, sep, rest = dsn.partition("://")