"""Postgres 遥测后端(M2 设计 §5): asyncpg lazy 池 + 两级降级。 参考仓无先例(三项目遥测全 SQLite);asyncpg 工程写法取 GovDoc `taskrun/postgres_store.py`($n 占位、`ON CONFLICT DO NOTHING`),但其 "失败冒泡"方向按遥测铁律**有意反转**: ① 结构性失败 → warning 一次后永久降级(所有写入短路); ② 运行时单条写失败 → 逐条 warning 丢弃,不降级不重试(连接抖动由 asyncpg 池自恢复;避免浸泡开头一次抖动导致后续全程失遥测)。 构造不连库(lazy),24 列 schema 与 SQLite 版同名同序。 **"结构性"的判据是「确定写不进去」,不是「初始化时出过错」**(issue #9): 只有建池失败(重试要在业务路径上内联吞掉 connect 超时)与"表确定不存在 且建不出来"(后续 INSERT 必然全败)才判死;探测失败、补列失败、取连接 失败一律只 warning,让写入照常尝试或下次调用重试。 """ from __future__ import annotations import asyncio from typing import TYPE_CHECKING from loguru import logger from polygateway.telemetry.schema import ( COLUMNS, PG_BACKFILL, PG_DDL, insert_sql, missing_columns_warning, ) if TYPE_CHECKING: import asyncpg # 探测表是否存在;不需要任何权限,且与 INSERT 走同一套 search_path 解析 _TABLE_EXISTS = "SELECT to_regclass('llm_calls')" # 探测现有列;尊重 search_path(to_regclass 按当前 search_path 解析) _EXISTING_COLUMNS = ( "SELECT attname FROM pg_attribute " "WHERE attrelid = to_regclass('llm_calls') AND attnum > 0 AND NOT attisdropped" ) class PostgresRecorder: """TelemetryRecorder 端口的 Postgres 实现;asyncpg 原生异步,无线程桥接。""" def __init__(self, dsn: str, *, pool: asyncpg.Pool | None = None, auto_migrate: bool) -> None: """记下装配参数(不连库);列与 INSERT 语句在首次准备期定型。 Args: dsn: asyncpg 连接串(已剥驱动后缀)。 pool: 外部注入的池;注入方自己负责关闭。 auto_migrate: True 则给已存在的旧表自动补列;False(PG 侧的缺省档) 则一条 ALTER 都不发——`ALTER TABLE ADD COLUMN` 取 ACCESS EXCLUSIVE 锁,会排在长事务后阻塞该表其后所有查询,而遥测是业务路径上的内联 await。keyword-only **必填**: 缺省规则只写在 config 一处,不与本类 签名漂移(设计 D-c)。 """ try: import asyncpg # noqa: F401 - 仅探测 extra 是否安装 except ImportError as exc: raise ImportError( "Postgres 遥测未启用: 安装 pip install 'polygateway[postgres]' 后重试" ) from exc self._dsn = dsn self._pool: asyncpg.Pool | None = pool self._external_pool = pool is not None self._auto_migrate = auto_migrate # 先按全量列定型: 准备期探测失败时保守沿用全量(今天的行为) self._columns: tuple[str, ...] = COLUMNS self._insert = insert_sql("postgres", COLUMNS) self._schema_ready = False self._failed = False # 结构性降级标志: 置位后所有写入短路 self._init_lock = asyncio.Lock() async def _ensure_ready(self) -> asyncpg.Pool | None: """lazy 建池+备表;判死只认「确定写不进去」(issue #9),其余失败都留活路。""" if self._failed: return None if self._schema_ready: return self._pool async with self._init_lock: if self._failed: return None if self._schema_ready: return self._pool pool = await self._open_pool() if pool is None: return None return await self._prepare_schema(pool) async def _open_pool(self) -> asyncpg.Pool | None: """建池;失败即永久降级(唯一一处「无条件判死」)。""" if self._pool is not None: return self._pool try: import asyncpg self._pool = await asyncpg.create_pool(self._dsn, timeout=10) except asyncio.CancelledError: raise except Exception as exc: # 池建不出来 = 确定写不进去;且每次调用重试都要内联吞掉 connect # 超时,而遥测是业务路径上的 await —— 此处必须永久降级 self._failed = True logger.warning("Postgres 遥测建池失败,后续记录降级为 no-op: {}", exc) return None return self._pool async def _prepare_schema(self, pool: asyncpg.Pool) -> asyncpg.Pool | None: """备好表并交回可用的池;瞬时失败只跳过本次,确定写不进去才判死。""" try: async with pool.acquire() as conn: columns = await self._prepare_table(conn) except asyncio.CancelledError: raise except Exception as exc: # 池已在手,取连接/探测失败多为瞬时抖动: 不判死也不标就绪, # 只跳过本次记录,下次调用重新准备 logger.warning("Postgres 遥测建表探测失败(跳过本条,下次重试): {}", exc) return None if columns is None: self._failed = True return None # 写入列、语句与就绪标志必须**一起**生效: `_ensure_ready` 只看 `_schema_ready` # 就绕开 `_init_lock` 直接返回池,先置就绪会开出"已就绪但语句还是旧的"的窗口 self._columns = columns self._insert = insert_sql("postgres", columns) self._schema_ready = True return pool async def _prepare_table(self, conn: object) -> tuple[str, ...] | None: """备好 `llm_calls` 并返回本实例要写的列;**表存在就绝不发 DDL**。 返回 None 仅表示表确定不存在且建不出来(唯一允许判死的情形)。 `CREATE TABLE IF NOT EXISTS` 不能无条件发: PostgreSQL 对 schema 的 CREATE 权限检查**早于** `IF NOT EXISTS` 的存在性判断(PG 16.14 实测: 只授 `SELECT, INSERT ON llm_calls` 的角色,表明明在、也写得进去,这一句 照样被拒 `permission denied for schema`)。这与 `_backfill_columns` 撞的 是同一类问题(issue #3/#9),故守卫也必须同款: 先探测,后 DDL。 探测走 `to_regclass`,不需要任何权限,且与 INSERT 的 search_path 解析 口径一致——比裸 DDL 更准(裸 `CREATE TABLE` 落在首个**可建**的 schema, 可能与 INSERT 命中的不是同一张表)。 """ exists = await conn.fetchval(_TABLE_EXISTS) is not None # type: ignore[attr-defined] if exists: return await self._resolve_columns(conn) # 旧表可能缺列 try: await conn.execute(PG_DDL) # type: ignore[attr-defined] except asyncio.CancelledError: raise except Exception as exc: logger.warning("Postgres 遥测建表失败(表不存在,记录无处可落): {}", exc) return None return COLUMNS # 新建表列已齐全,无需再走补列 async def _resolve_columns(self, conn: object) -> tuple[str, ...]: """探测旧表现有列并定型写入列: auto 档先补齐,manual 档改为裁剪(issue #13)。 **先探测**的理由(两档共用): `ADD COLUMN IF NOT EXISTS` 即便列已存在,也会 **先取 ACCESS EXCLUSIVE 锁**再判存在性(实测会被一个开着的读事务阻塞)。遥测是 内联 await,让每个进程的首次写入都去抢共享审计表的排他锁,等于用记录基础设施 拖垮业务调用。探测走 ACCESS SHARE,稳态下一条 ALTER 都不会发。 探测失败保守沿用全量列(今天的行为): 猜不出真实列集合时,让写入照常尝试。 """ try: existing = {row["attname"] for row in await conn.fetch(_EXISTING_COLUMNS)} # type: ignore[attr-defined] except asyncio.CancelledError: raise except Exception as exc: logger.warning("Postgres 遥测列探测失败(沿用全量列,写入将逐行降级): {}", exc) return COLUMNS if self._auto_migrate: await self._backfill_columns(conn, existing) return COLUMNS return self._trim_columns(existing) def _trim_columns(self, existing: set[str]) -> tuple[str, ...]: """manual 档: 按现有列裁剪写入列,并把缺列一次讲清楚。 裁剪是关掉 ALTER 的**前提**而非增强: 旧表缺列时仍发全量 INSERT,每一行 都会因未知列被拒 → 遥测彻底丢失,比自动 ALTER 更严重地违反"遥测必录"。 探测结果与 `COLUMNS` 毫无交集时视同探测异常保守回落全量: 空列集拼不出合法 INSERT,`insert_sql` 会 ValueError,而 `_prepare_schema` 里那次调用在 try **之外**,异常会顺着 `record_llm_call` 一路冒给业务调用方(遥测绝不冒泡) ——回落必须发生在把空列集交给它之前。 """ effective = tuple(column for column in COLUMNS if column in existing) if not effective: logger.warning( "Postgres 遥测表 llm_calls 没有任何本库认识的列(沿用全量列,写入将逐行降级);" "现有列: {}", sorted(existing), ) return COLUMNS missing = [column for column in COLUMNS if column not in existing] if missing: # 单参数传入: 补列 SQL 里带 `'{}'::jsonb` 字面量,拼进 format 模板会被当占位符 logger.warning( "{}", missing_columns_warning("postgres", missing, alien_table="call_id" not in existing), ) return effective async def _backfill_columns(self, conn: object, existing: set[str]) -> None: """auto 档: 给已存在的旧表补新列(issue #3);**失败绝不置 `_failed`**。 不置 `_failed` 的实测理由: 应用账号只有 INSERT 权限时,`ALTER TABLE` 的 ownership 检查早于 `IF NOT EXISTS` 的存在性判断——列明明齐全也会失败。置位会让 整个 recorder 永久 no-op,与「补列失败只降级为逐行丢弃」的承诺相悖 (SQLite 侧同款守卫,两侧必须对称)。补列失败后写入沿用全量列(今天的行为): auto 档承诺的是"把列补上",补不上就让缺列以逐行 warning 暴露;要降级写入 请显式选 manual。 """ try: for column, statement in PG_BACKFILL: if column not in existing: await conn.execute(statement) # type: ignore[attr-defined] except asyncio.CancelledError: raise except Exception as exc: logger.warning("Postgres 遥测补列失败(写入将逐行降级): {}", exc) async def record_llm_call(self, **fields: object) -> None: """写一行遥测;单条失败逐条 warning 丢弃(两级降级之二),绝不冒泡。 取值按 `self._columns`(manual 档可能已被裁剪),与 `self._insert` 的 占位符同序——两者必须一起改,分开改就是把值写进错位的列。 """ pool = await self._ensure_ready() if pool is None: return row = tuple(fields[col] for col in self._columns) try: async with pool.acquire() as conn: await conn.execute(self._insert, *row) except asyncio.CancelledError: raise except Exception as exc: # 遥测铁律: 丢一条 < 拖垮调用;仅记 warning(非 pass),池自恢复 logger.warning("Postgres 遥测写入失败(丢弃该行): {}", exc) async def aclose(self) -> None: """幂等关闭自建池;注入的池归注入方管理。""" pool, self._pool = self._pool, None self._schema_ready = False if pool is not None and not self._external_pool: await pool.close()