fix: probe for the telemetry table before creating it
PostgreSQL checks the schema CREATE privilege before the IF NOT EXISTS existence test, so an account with only table-level INSERT was denied on CREATE TABLE IF NOT EXISTS even though the table was right there and writable. The denial set _failed and the whole recorder went no-op for the process lifetime, silently: 150+ calls downstream lost their latency, token and cost rows with nothing but one warning to show for it. The probe is the direct fix. The larger fix is the criterion: structural degradation now means "provably cannot write" (pool creation failed, or the table is absent and cannot be created), not "something threw during init" -- a probe or acquire failure just skips the row and retries on the next call. SQLite stays as it is on purpose. Measured: it short-circuits the statement at parse time, so it passes even under another connection's EXCLUSIVE lock or on a read-only file. A probe there would buy nothing; the docstring now says so to keep symmetry-minded future edits away.
This commit is contained in:
@@ -1,12 +1,17 @@
|
||||
"""Postgres 遥测后端(M2 设计 §5): asyncpg lazy 池 + 两级降级。
|
||||
|
||||
参考仓无先例(三项目遥测全 SQLite);asyncpg 工程写法取 GovDoc
|
||||
`taskrun/postgres_store.py`($n 占位、`CREATE TABLE IF NOT EXISTS`、
|
||||
`ON CONFLICT DO NOTHING`),但其"失败冒泡"方向按遥测铁律**有意反转**:
|
||||
① 结构性失败(建池/建表)→ warning 一次后永久降级(池置 None 短路);
|
||||
`taskrun/postgres_store.py`($n 占位、`ON CONFLICT DO NOTHING`),但其
|
||||
"失败冒泡"方向按遥测铁律**有意反转**:
|
||||
① 结构性失败 → warning 一次后永久降级(所有写入短路);
|
||||
② 运行时单条写失败 → 逐条 warning 丢弃,不降级不重试(连接抖动由
|
||||
asyncpg 池自恢复;避免浸泡开头一次抖动导致后续全程失遥测)。
|
||||
构造不连库(lazy),20 列 schema 与 SQLite 版同名同序。
|
||||
构造不连库(lazy),22 列 schema 与 SQLite 版同名同序。
|
||||
|
||||
**"结构性"的判据是「确定写不进去」,不是「初始化时出过错」**(issue #9):
|
||||
只有建池失败(重试要在业务路径上内联吞掉 connect 超时)与"表确定不存在
|
||||
且建不出来"(后续 INSERT 必然全败)才判死;探测失败、补列失败、取连接
|
||||
失败一律只 warning,让写入照常尝试或下次调用重试。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -55,6 +60,9 @@ _BACKFILL = (
|
||||
("reasoning_tokens", "ALTER TABLE llm_calls ADD COLUMN reasoning_tokens INTEGER"),
|
||||
)
|
||||
|
||||
# 探测表是否存在;不需要任何权限,且与 INSERT 走同一套 search_path 解析
|
||||
_TABLE_EXISTS = "SELECT to_regclass('llm_calls')"
|
||||
|
||||
# 探测现有列;尊重 search_path(to_regclass 按当前 search_path 解析)
|
||||
_EXISTING_COLUMNS = (
|
||||
"SELECT attname FROM pg_attribute "
|
||||
@@ -111,30 +119,81 @@ class PostgresRecorder:
|
||||
self._init_lock = asyncio.Lock()
|
||||
|
||||
async def _ensure_ready(self) -> asyncpg.Pool | None:
|
||||
"""lazy 建池+建表;结构性失败 warning 一次后永久降级(设计 §5 两级之一)。"""
|
||||
"""lazy 建池+备表;判死只认「确定写不进去」(issue #9),其余失败都留活路。"""
|
||||
if self._failed:
|
||||
return None
|
||||
if self._schema_ready:
|
||||
return self._pool
|
||||
async with self._init_lock:
|
||||
if self._failed or self._schema_ready:
|
||||
return None if self._failed else self._pool
|
||||
try:
|
||||
if self._pool is None:
|
||||
import asyncpg
|
||||
|
||||
self._pool = await asyncpg.create_pool(self._dsn, timeout=10)
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.execute(_DDL)
|
||||
await self._backfill_columns(conn)
|
||||
self._schema_ready = True
|
||||
return self._pool
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
self._failed = True
|
||||
logger.warning("Postgres 遥测初始化失败,后续记录降级为 no-op: {}", exc)
|
||||
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:
|
||||
writable = await self._prepare_table(conn)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
# 池已在手,取连接/探测失败多为瞬时抖动: 不判死也不标就绪,
|
||||
# 只跳过本次记录,下次调用重新准备
|
||||
logger.warning("Postgres 遥测建表探测失败(跳过本条,下次重试): {}", exc)
|
||||
return None
|
||||
if not writable:
|
||||
self._failed = True
|
||||
return None
|
||||
self._schema_ready = True
|
||||
return pool
|
||||
|
||||
async def _prepare_table(self, conn: object) -> bool:
|
||||
"""备好 `llm_calls`;**表存在就绝不发 DDL**。返回 False 仅表示表确定不存在。
|
||||
|
||||
`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:
|
||||
await self._backfill_columns(conn) # 旧表可能缺列;失败只逐行降级
|
||||
return True
|
||||
try:
|
||||
await conn.execute(_DDL) # type: ignore[attr-defined]
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.warning("Postgres 遥测建表失败(表不存在,记录无处可落): {}", exc)
|
||||
return False
|
||||
return True # 新建表列已齐全,无需再走补列
|
||||
|
||||
async def _backfill_columns(self, conn: object) -> None:
|
||||
"""给已存在的旧表补新列(issue #3);**先探测再 ALTER,失败绝不置 `_failed`**。
|
||||
|
||||
Reference in New Issue
Block a user