diff --git a/src/polygateway/telemetry/schema.py b/src/polygateway/telemetry/schema.py index 99ab544..274b5aa 100644 --- a/src/polygateway/telemetry/schema.py +++ b/src/polygateway/telemetry/schema.py @@ -155,6 +155,13 @@ def insert_sql(backend: str, columns: Sequence[str]) -> str: 字符串拼进 SQL(占位符只保护值,保护不了列名)。sqlite 用 `?`、postgres 用 `$n`, 两端的重复键处理都不绑定具体约束名(`INSERT OR IGNORE` / `ON CONFLICT`)。 + **PG 的 `ON CONFLICT` 一律不带冲突目标,不得"顺手"补回 `(call_id)`**: PG 要求 + 分区表的唯一约束必须包含分区键,按 `created_at` 分区(issue #12 的保留期方案)后 + 主键变成 `(call_id, created_at)`,带目标的语句匹配不到任何约束,PG 直接拒收 + ("there is no unique or exclusion constraint matching the ON CONFLICT + specification"),而遥测写失败只逐行 warning——分区部署下会全线静默丢数据。 + 无目标版本在两种表形态上都合法,普通表上语义逐字等价(表上只有主键一个唯一约束)。 + Args: backend: `"sqlite"` 或 `"postgres"`。 columns: 要写入的列,顺序即占位符顺序(调用方须按同序取值)。 @@ -176,7 +183,7 @@ def insert_sql(backend: str, columns: Sequence[str]) -> str: placeholders = ", ".join("?" for _ in selected) return f"INSERT OR IGNORE INTO {TABLE} ({names}) VALUES ({placeholders})" placeholders = ", ".join(f"${i + 1}" for i in range(len(selected))) - return f"INSERT INTO {TABLE} ({names}) VALUES ({placeholders}) ON CONFLICT (call_id) DO NOTHING" + return f"INSERT INTO {TABLE} ({names}) VALUES ({placeholders}) ON CONFLICT DO NOTHING" def telemetry_schema_sql(backend: str) -> str: diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index a9b3f9e..48fd4c9 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -14,6 +14,7 @@ import asyncio import json import os import re +from datetime import UTC, datetime, timedelta from uuid import uuid4 import pytest @@ -628,3 +629,123 @@ class TestCallerDimensionsAcceptance: assert any("写入失败" in m for m in captured_warnings) finally: await recorder.aclose() + + +# issue #12 的目标表形态: 按 created_at 做 RANGE 分区(过期清理 DROP PARTITION 而非 DELETE)。 +# PG 强制分区表的唯一约束必须包含分区键,故主键只能是 (call_id, created_at) —— +# 这正是带目标的 `ON CONFLICT (call_id)` 再也匹配不到约束的现场。 +_PARTITIONED_DDL = """ +CREATE TABLE {schema}.llm_calls ( + call_id TEXT NOT NULL, + parent_call_id TEXT, + session_id TEXT, + model TEXT NOT NULL, + provider TEXT NOT NULL, + source_name TEXT NOT NULL, + messages TEXT NOT NULL, + response TEXT NOT NULL, + thinking TEXT NOT NULL DEFAULT '', + prompt_tokens INTEGER NOT NULL, + completion_tokens INTEGER NOT NULL, + usage_source TEXT NOT NULL, + latency_ms INTEGER NOT NULL, + ttft_ms DOUBLE PRECISION, + max_inter_token_ms DOUBLE PRECISION, + cache_hit BOOLEAN NOT NULL DEFAULT FALSE, + error TEXT, + cost DOUBLE PRECISION, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + cached_prompt_tokens INTEGER, + model_reported TEXT, + sampling TEXT, + reasoning_tokens INTEGER, + tenant_id TEXT NOT NULL DEFAULT '', + meta JSONB NOT NULL DEFAULT '{{}}'::jsonb, + PRIMARY KEY (call_id, created_at) +) PARTITION BY RANGE (created_at) +""" + +_PARTITION_DDL = ( + "CREATE TABLE {schema}.llm_calls_current PARTITION OF {schema}.llm_calls " + "FOR VALUES FROM ('{start}') TO ('{end}')" +) + + +def _current_month_bounds() -> tuple[str, str]: + """当前月的 [月初, 下月初) 边界字面量;分区键落在区间外会因找不到分区而写失败。""" + now = datetime.now(UTC) + start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) + end = (start + timedelta(days=32)).replace(day=1) + fmt = "%Y-%m-%d %H:%M:%S%z" + return start.strftime(fmt), end.strftime(fmt) + + +@pytest.fixture +async def partitioned_schema(dsn): + """自建临时 schema 里造一张按 created_at RANGE 分区的表 + 覆盖当前月的分区。 + + 与 legacy_schema 同款隔离: 绝不碰共享的 public.llm_calls,teardown 只 DROP + 自己建的 schema(CASCADE 连分区一并删)。 + """ + import asyncpg + + name = f"pgwtest_part_{uuid4().hex[:8]}" + start, end = _current_month_bounds() + conn = await asyncpg.connect(dsn, timeout=10) + try: + await conn.execute(f"CREATE SCHEMA {name}") + await conn.execute(_PARTITIONED_DDL.format(schema=name)) + await conn.execute(_PARTITION_DDL.format(schema=name, start=start, end=end)) + finally: + await conn.close() + yield _search_path_dsn(dsn, name), name + conn = await asyncpg.connect(dsn, timeout=10) + try: + await conn.execute(f"DROP SCHEMA {name} CASCADE") + finally: + await conn.close() + + +class TestConflictTargetFreeInsert: + """issue #13: INSERT 不绑定冲突目标,普通表与分区表两种形态都写得进去。""" + + async def test_plain_table_still_dedupes_by_call_id(self, fresh_schema, captured_warnings): + """普通表上语义不变: 重复 call_id 仍只落一行,且不是被拒后丢弃。 + + 表上只有主键这一个唯一约束,故无目标的 DO NOTHING 与 `(call_id)` 逐字等价; + 断言"无写入失败 warning"是为了区分"冲突被忽略"与"整条被 PG 拒收"。 + """ + fresh_dsn, _ = fresh_schema + recorder = PostgresRecorder(fresh_dsn) + try: + await _record_minimal(recorder, call_id=_cid("nodup")) + await _record_minimal(recorder, call_id=_cid("nodup"), response="second") + assert [m for m in captured_warnings if "写入失败" in m] == [] + rows = await _fetch( + fresh_dsn, "SELECT response FROM llm_calls WHERE call_id = $1", _cid("nodup") + ) + assert [r["response"] for r in rows] == ["ok"] # 首行胜出,写入幂等 + finally: + await recorder.aclose() + + async def test_partitioned_table_accepts_writes(self, partitioned_schema, captured_warnings): + """分区表上写入成功且能读回——改动前这里必红。 + + 带目标的 `ON CONFLICT (call_id)` 在主键为 `(call_id, created_at)` 的表上 + 匹配不到任何约束,PG 报 "there is no unique or exclusion constraint matching + the ON CONFLICT specification";该错误被逐行降级吞成 warning,于是分区部署下 + 遥测全线写不进去却一声不吭,只能靠"读不回来"暴露。 + """ + part_dsn, _ = partitioned_schema + recorder = PostgresRecorder(part_dsn) + try: + await _record_minimal(recorder, call_id=_cid("part"), tenant_id="tenant-p") + assert [m for m in captured_warnings if "写入失败" in m] == [] + rows = await _fetch( + part_dsn, + "SELECT call_id, tenant_id FROM llm_calls WHERE call_id = $1", + _cid("part"), + ) + assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(_cid("part"), "tenant-p")] + finally: + await recorder.aclose() diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index f9a5083..c3a43b8 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -131,7 +131,9 @@ _FROZEN_PG_INSERT = ( "sampling, reasoning_tokens, tenant_id, meta) " "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, " "$19, $20, $21, $22, $23, $24) " - "ON CONFLICT (call_id) DO NOTHING" + # 无冲突目标(issue #13 Task 2): 带 `(call_id)` 的版本在按 created_at 分区、 + # 主键为 (call_id, created_at) 的表上匹配不到约束,PG 直接拒收整条写入 + "ON CONFLICT DO NOTHING" ) @@ -174,16 +176,17 @@ class TestSchemaModule: assert all("IF NOT EXISTS" not in stmt for _, stmt in PG_BACKFILL) def test_insert_sql_reproduces_the_frozen_statements(self): - """`insert_sql(backend, COLUMNS)` 必须与搬迁前的 `_INSERT` 逐字节相同。""" + """`insert_sql(backend, COLUMNS)` 与搬迁前的 `_INSERT` 一致(PG 侧去掉冲突目标)。""" from polygateway.telemetry.schema import COLUMNS, insert_sql assert insert_sql("sqlite", COLUMNS) == _FROZEN_SQLITE_INSERT assert insert_sql("postgres", COLUMNS) == _FROZEN_PG_INSERT # 裁剪列表按位置占位符重新编号,不留空洞 assert insert_sql("postgres", ["call_id", "model"]) == ( - "INSERT INTO llm_calls (call_id, model) VALUES ($1, $2) " - "ON CONFLICT (call_id) DO NOTHING" + "INSERT INTO llm_calls (call_id, model) VALUES ($1, $2) ON CONFLICT DO NOTHING" ) + # 冲突目标不得被"顺手"补回: 分区表上它会让每一条遥测都被 PG 拒收 + assert "ON CONFLICT (" not in insert_sql("postgres", COLUMNS) def test_insert_sql_rejects_foreign_columns_and_backends(self): """列名来自数据库探测结果而非常量,子集校验是唯一的注入面闸门。"""