diff --git a/research-wiki/designs/2026-08-26-issue18-pg-test-isolation-design.md b/research-wiki/designs/2026-08-26-issue18-pg-test-isolation-design.md index 0eff45c..1ac6c47 100644 --- a/research-wiki/designs/2026-08-26-issue18-pg-test-isolation-design.md +++ b/research-wiki/designs/2026-08-26-issue18-pg-test-isolation-design.md @@ -60,7 +60,7 @@ L3 不是测试的问题，是脚本契约的问题——它同时是生产风 | 仅 `--backend postgres` 接受 | sqlite 给了 `--table` → 退出 **1** | 与 `--batch-size` 同款；SQLite 库文件即目标，无 schema 概念，无歧义可消 | | 必须是**两段**限定名 | `--table llm_calls` → 退出 **1**，提示写成 `schema.表名` | 单段等于没声明，隐式性原样保留 | | **表名段必须逐字等于 `llm_calls`** | `--table audit.events` → 退出 **1**，消息点明本脚本只清理 `llm_calls` | 见 §4.4：不加这条，`--table` 会把本脚本从"遥测表清理器"扩成"任意同形表删除工具" | -| 两段均非空、均不含 `.` 与 `"` | 不合法 → 退出 **1** | 复杂标识符（含点/引号的表名）不支持，此时退回不给 `--table` 的路径；写进 `--help` | +| 两段均非空、均不含 `"` | 不合法 → 退出 **1** | 复杂标识符（含引号的表名）不支持，此时退回不给 `--table` 的路径；写进 `--help`。**本行原写作"均不含 `.` 与 `\"`"，实现阶段核出"段内含 `.`"是不可达分支**——按 `.` 切分后恰好两段是前置条件，`a.b.c` 走的是"不是恰好两段"那条消息，故删去该半句 | | **逐字比较，不做大小写折叠** | 传 `_quote()` 包裹的限定名给 `to_regclass` | catalog 里存的是真实标识符；未加引号建的表在 catalog 中是小写。折叠会与"引号标识符区分大小写"的真实语义打架 | | 解析不到 | 退出 **2**，消息点名"显式指定的表 X 不存在"，并附一句"PG 中未加引号建的标识符在 catalog 里是小写" | 与 `search_path` 找不到的消息**分开写**：诊断方向不同。**退出码维持 2 而非 1**：`Public.llm_calls` 格式合法，找不到是环境事实而非参数非法——把它归成 1 会让"schema 真的不存在"这类该告警的情形被调度器当成不必重试的参数错误。大小写这类高频手误由消息文本消化，不由退出码 | | 无权限 | 后续 `COUNT` 抛 `PostgresError` → 既有 except → 退出 **2** | 无需新增分支 | diff --git a/tests/integration/test_retention_tool_pg.py b/tests/integration/test_retention_tool_pg.py index 7c95ba4..6e7eeb1 100644 --- a/tests/integration/test_retention_tool_pg.py +++ b/tests/integration/test_retention_tool_pg.py @@ -1,11 +1,12 @@ """`tools/telemetry_retention.py` 的 PostgreSQL 分支测试(issue #12 Task 3,真实 PG)。 -DSN 走 .env `PGW_TELEMETRY_PG_DSN`,缺则 skip。 +隔离纪律(issue #18): 本文件跑的是一个**会删数据的脚本**,而实例上的共享表 `llm_calls` +与真实批跑共用。故**凡启动脚本的用例一律用 `pg_sandbox(role="owner")` 的临时角色跑**: +该角色对共享表一无所有,越界不是"会被发现",而是数据库层面做不到。 -隔离纪律(M4 事故教训): `public.llm_calls` 是与真实批跑共享的表,而本测试跑的是 -一个**会删数据的脚本**——一律在自建的临时 schema 里操作(DSN 挂 search_path), -teardown 只 `DROP SCHEMA ... CASCADE`;分批删除那例另行断言 `public.llm_calls` -的行数前后不变,把"search_path 没生效"这种最坏情况钉成红灯而不是静默删库。 +这条纪律取代了此前那条"跑完对比共享表行数"的安全网——行数快照守的是安全属性,却把它 +编码成对全局可变量的观测: 外部进程一写就假红,外部插入与脚本误删互相抵消则假阴。 +权限边界两个方向都没有。 """ from __future__ import annotations @@ -16,10 +17,8 @@ import subprocess import sys from datetime import UTC, datetime, timedelta from pathlib import Path -from uuid import uuid4 import pytest -from dotenv import dotenv_values from polygateway.telemetry.schema import PG_DDL @@ -55,20 +54,6 @@ def _partitioned_ddl() -> str: ) -def _dsn_value() -> str | None: - merged = {**dotenv_values(".env"), **os.environ} - raw = merged.get("PGW_TELEMETRY_PG_DSN") - if not raw: - return None - scheme, sep, rest = raw.partition("://") - return f"{scheme.partition('+')[0]}{sep}{rest}" - - -def _search_path_dsn(dsn: str, schema: str) -> str: - sep = "&" if "?" in dsn else "?" - return f"{dsn}{sep}options=-csearch_path%3D{schema}" - - def _stamp(delta: timedelta) -> datetime: return datetime.now(UTC) + delta @@ -84,43 +69,6 @@ def _run(*args: str, env: dict[str, str] | None = None) -> subprocess.CompletedP ) -@pytest.fixture -async def dsn(): - value = _dsn_value() - if value is None: - pytest.skip("PGW_TELEMETRY_PG_DSN 未配置") - # 隔离守卫: 该实例有 app/chs_prod 等在用库,只许打 polygateway 专用库 - if not value.rstrip("/").endswith("/polygateway"): - pytest.fail(f"保留期脚本测试只允许连 polygateway 专用库,当前 DSN 库名不符: {value!r}") - return value - - -async def _make_schema(dsn_value: str, prefix: str, ddl: str, extra: tuple[str, ...] = ()) -> str: - import asyncpg - - name = f"pgwret_{prefix}_{uuid4().hex[:8]}" - conn = await asyncpg.connect(dsn_value, timeout=10) - try: - await conn.execute(f"CREATE SCHEMA {name}") - await conn.execute(f"SET search_path = {name}") - await conn.execute(ddl) - for statement in extra: - await conn.execute(statement) - finally: - await conn.close() - return name - - -async def _drop_schema(dsn_value: str, name: str) -> None: - import asyncpg - - conn = await asyncpg.connect(dsn_value, timeout=10) - try: - await conn.execute(f"DROP SCHEMA {name} CASCADE") - finally: - await conn.close() - - async def _seed(schema_dsn: str, rows: list[tuple[str, str, datetime]]) -> None: import asyncpg @@ -142,50 +90,31 @@ async def _call_ids(schema_dsn: str) -> list[str]: return [r["call_id"] for r in rows] -async def _public_count(dsn_value: str) -> int: - """共享表的行数;本测试全程不得让它变动一行。""" - import asyncpg - - conn = await asyncpg.connect(dsn_value, timeout=10) - try: - if await conn.fetchval("SELECT to_regclass('public.llm_calls')") is None: - return -1 - return await conn.fetchval("SELECT COUNT(*) FROM public.llm_calls") - finally: - await conn.close() - - @pytest.fixture -async def partitioned_schema(dsn): +async def partitioned_sandbox(pg_sandbox): """临时 schema 内的**分区表**: 脚本必须认出它并让路给 DROP PARTITION。""" - name = await _make_schema( - dsn, - "part", - _partitioned_ddl(), + return await pg_sandbox( + ddl=_partitioned_ddl(), extra=( "CREATE TABLE llm_calls_all PARTITION OF llm_calls " "FOR VALUES FROM ('2000-01-01') TO ('2100-01-01')", ), + role="owner", ) - yield _search_path_dsn(dsn, name), name - await _drop_schema(dsn, name) @pytest.fixture -async def plain_schema(dsn): +async def plain_sandbox(pg_sandbox): """临时 schema 内的普通表: 存量场景,脚本的分批 DELETE 兜底路径。""" - name = await _make_schema(dsn, "plain", PG_DDL) - yield _search_path_dsn(dsn, name), name - await _drop_schema(dsn, name) + return await pg_sandbox(ddl=PG_DDL, role="owner") class TestPartitionedTarget: async def test_partitioned_table_exits_three_without_deleting_anything( - self, partitioned_schema + self, partitioned_sandbox ): - schema_dsn, schema = partitioned_schema await _seed( - schema_dsn, + partitioned_sandbox.dsn, [ ("part-old-1", "", _stamp(timedelta(days=-30))), ("part-old-2", "acme", _stamp(timedelta(days=-20))), @@ -194,24 +123,28 @@ class TestPartitionedTarget: # 带 --apply 跑: 危险的那条路径必须在真正删之前就被分区探测拦住 result = _run( - "--backend", "postgres", "--dsn", schema_dsn, "--older-than-days", "7", "--apply" + "--backend", + "postgres", + "--dsn", + partitioned_sandbox.dsn, + "--older-than-days", + "7", + "--apply", ) assert result.returncode == 3, (result.stdout, result.stderr) combined = result.stdout + result.stderr assert "DROP PARTITION" in combined assert "DETACH" in combined - assert await _call_ids(schema_dsn) == ["part-old-1", "part-old-2"] + assert await _call_ids(partitioned_sandbox.dsn) == ["part-old-1", "part-old-2"] # 脚本必须报出它解析到的**限定表名**: 这是"我删的到底是哪张表"的唯一凭据 - assert f"{schema}.llm_calls" in result.stdout + assert f"{partitioned_sandbox.schema}.llm_calls" in result.stdout class TestPlainTableBatches: - async def test_apply_deletes_only_expired_rows_in_batches(self, plain_schema, dsn): - schema_dsn, schema = plain_schema - before_public = await _public_count(dsn) + async def test_apply_deletes_only_expired_rows_in_batches(self, plain_sandbox): await _seed( - schema_dsn, + plain_sandbox.dsn, [ ("old-1", "", _stamp(timedelta(days=-40))), ("old-2", "acme", _stamp(timedelta(days=-30))), @@ -227,7 +160,7 @@ class TestPlainTableBatches: "--backend", "postgres", "--dsn", - schema_dsn, + plain_sandbox.dsn, "--older-than-days", "7", "--apply", @@ -236,8 +169,8 @@ class TestPlainTableBatches: ) assert result.returncode == 0, (result.stdout, result.stderr) - assert await _call_ids(schema_dsn) == ["fresh-1", "fresh-2"] - assert f"{schema}.llm_calls" in result.stdout + assert await _call_ids(plain_sandbox.dsn) == ["fresh-1", "fresh-2"] + assert f"{plain_sandbox.schema}.llm_calls" in result.stdout assert "将删除行数: 5" in result.stdout assert "'acme': 3" in result.stdout # 5 行 / 每批 2 行 = 3 批,每批各自提交;批次行必须真的出现三条 @@ -245,30 +178,200 @@ class TestPlainTableBatches: assert "批次 3" in result.stdout assert "批次 4" not in result.stdout assert "已删除 5 行" in result.stdout - assert await _public_count(dsn) == before_public - async def test_dry_run_on_a_plain_table_deletes_nothing(self, plain_schema): - schema_dsn, _ = plain_schema - await _seed(schema_dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) + async def test_dry_run_on_a_plain_table_deletes_nothing(self, plain_sandbox): + await _seed(plain_sandbox.dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) - result = _run("--backend", "postgres", "--dsn", schema_dsn, "--older-than-days", "7") + result = _run("--backend", "postgres", "--dsn", plain_sandbox.dsn, "--older-than-days", "7") assert result.returncode == 0, (result.stdout, result.stderr) assert "将删除行数: 1" in result.stdout assert "dry-run" in result.stdout - assert await _call_ids(schema_dsn) == ["old-1"] + assert await _call_ids(plain_sandbox.dsn) == ["old-1"] + + +class TestExplicitTable: + """`--table SCHEMA.NAME` 的真实解析行为(issue #18 设计 §4.1;判据 1b)。 + + 单测只能验参数分类,验不了 `to_regclass` 的语义——schema 不存在返 NULL 而非抛错、 + 引号限定名区分大小写、无权限落在 `COUNT` 而非解析,这三条都必须真连库才成立。 + """ + + async def test_explicit_table_deletes_exactly_like_the_implicit_path(self, plain_sandbox): + await _seed( + plain_sandbox.dsn, + [ + ("old-1", "", _stamp(timedelta(days=-40))), + ("old-2", "acme", _stamp(timedelta(days=-30))), + ("old-3", "acme", _stamp(timedelta(days=-20))), + ("old-4", "acme", _stamp(timedelta(days=-15))), + ("old-5", "", _stamp(timedelta(days=-10))), + ("fresh-1", "acme", _stamp(timedelta(days=-1))), + ("fresh-2", "", _stamp(timedelta(hours=-1))), + ], + ) + + result = _run( + "--backend", + "postgres", + "--dsn", + plain_sandbox.dsn, + "--older-than-days", + "7", + "--apply", + "--batch-size", + "2", + "--table", + f"{plain_sandbox.schema}.llm_calls", + ) + + # 与不给 --table 的那条用例逐条同款: 显式声明只改"怎么找到表",不改任何行为 + assert result.returncode == 0, (result.stdout, result.stderr) + assert await _call_ids(plain_sandbox.dsn) == ["fresh-1", "fresh-2"] + assert f"{plain_sandbox.schema}.llm_calls" in result.stdout + assert "将删除行数: 5" in result.stdout + assert "'acme': 3" in result.stdout + assert "批次 1" in result.stdout + assert "批次 3" in result.stdout + assert "批次 4" not in result.stdout + assert "已删除 5 行" in result.stdout + + async def test_table_in_a_nonexistent_schema_exits_two(self, plain_sandbox): + """schema 不存在时 `to_regclass` 返 NULL(不抛错),故落进既有的"目标不可用"。""" + await _seed(plain_sandbox.dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) + missing = "pgw_s_nosuchxxxxxxxx" + + result = _run( + "--backend", + "postgres", + "--dsn", + plain_sandbox.dsn, + "--older-than-days", + "7", + "--apply", + "--table", + f"{missing}.llm_calls", + ) + + assert result.returncode == 2, (result.stdout, result.stderr) + assert missing in result.stderr + assert "llm_calls" in result.stderr + assert await _call_ids(plain_sandbox.dsn) == ["old-1"] + + async def test_table_owned_by_another_role_exits_two(self, pg_sandbox): + """拿 A 的连接指 B 的表: 权限拒绝,两张表都不能少一行。""" + sandbox_a = await pg_sandbox(ddl=PG_DDL, role="owner") + sandbox_b = await pg_sandbox(ddl=PG_DDL, role="owner") + await _seed(sandbox_a.dsn, [("a-old", "acme", _stamp(timedelta(days=-30)))]) + await _seed(sandbox_b.dsn, [("b-old", "acme", _stamp(timedelta(days=-30)))]) + + result = _run( + "--backend", + "postgres", + "--dsn", + sandbox_a.dsn, + "--older-than-days", + "7", + "--apply", + "--table", + f"{sandbox_b.schema}.llm_calls", + ) + + assert result.returncode == 2, (result.stdout, result.stderr) + assert await _call_ids(sandbox_a.dsn) == ["a-old"] + assert await _call_ids(sandbox_b.dsn) == ["b-old"] + + async def test_explicit_partitioned_table_still_exits_three(self, partitioned_sandbox): + await _seed(partitioned_sandbox.dsn, [("part-old-1", "acme", _stamp(timedelta(days=-30)))]) + + result = _run( + "--backend", + "postgres", + "--dsn", + partitioned_sandbox.dsn, + "--older-than-days", + "7", + "--apply", + "--table", + f"{partitioned_sandbox.schema}.llm_calls", + ) + + assert result.returncode == 3, (result.stdout, result.stderr) + combined = result.stdout + result.stderr + assert "DROP PARTITION" in combined + assert "DETACH" in combined + assert await _call_ids(partitioned_sandbox.dsn) == ["part-old-1"] + + +class TestInferredTargetHint: + async def test_apply_without_table_warns_that_the_target_was_inferred(self, plain_sandbox): + """未钉死目标时必须当场说清"这张表是猜出来的"(设计 §4.4;判据 2)。 + + 该提示行只在 PG 分支打印,不连库的单测触发不到它,故验收落在集成层。 + """ + await _seed(plain_sandbox.dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) + + result = _run( + "--backend", + "postgres", + "--dsn", + plain_sandbox.dsn, + "--older-than-days", + "7", + "--apply", + ) + + assert result.returncode == 0, (result.stdout, result.stderr) + assert "search_path" in result.stdout + assert "--table" in result.stdout + + async def test_dry_run_does_not_print_the_hint(self, plain_sandbox): + """dry-run 不可逆性为零,它本就以"看清楚再决定"为用途,多一行提示是噪音。""" + await _seed(plain_sandbox.dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) + + result = _run("--backend", "postgres", "--dsn", plain_sandbox.dsn, "--older-than-days", "7") + + assert result.returncode == 0, (result.stdout, result.stderr) + assert "--table" not in result.stdout + + +class TestSearchPathFallsThrough: + async def test_bare_search_path_cannot_touch_the_shared_table(self, plain_sandbox): + """最坏情况: `search_path` 没生效,脚本落到共享表 `llm_calls` 上(设计 §5.3)。 + + 用沙箱角色的**裸** DSN 跑(search_path 回落 `"$user", public`,而角色名与 schema + 名有意错开,故 `"$user"` 命不中沙箱),不给 `--table`,带 `--apply`。角色对共享表 + 无任何权限,于是两条可能的路都收敛到退出码 2: 库里有那张表则 `COUNT` 被权限拒绝, + 没有则解析不到。**不断言 PG 的英文原文**——服务端 `lc_messages` 不由测试掌握。 + """ + await _seed(plain_sandbox.dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) + + result = _run( + "--backend", + "postgres", + "--dsn", + plain_sandbox.bare_dsn, + "--older-than-days", + "7", + "--apply", + ) + + assert result.returncode == 2, (result.stdout, result.stderr) + assert result.stderr.strip() + assert "llm_calls" in result.stderr + # 沙箱表一行不少: 脚本既没删共享表,也没绕回来删自己这张 + assert await _call_ids(plain_sandbox.dsn) == ["old-1"] class TestMissingAsyncpg: - async def test_missing_asyncpg_exits_two_without_touching_rows(self, plain_schema, tmp_path): + async def test_missing_asyncpg_exits_two_without_touching_rows(self, plain_sandbox, tmp_path): """缺 asyncpg 必须明确报错退出(码 2),不静默降级——这是运维工具不是库路径。 用一个只 `raise ImportError` 的临时 `asyncpg.py` 挂进子进程的 PYTHONPATH 构造该 场景: 脚本跑在子进程里,monkeypatch 对它无效。DSN 用**真实可连**的临时 schema, 这样"没有导入守卫"的实现会走通并退出 0,而不是碰巧也退出 2 而假绿。 """ - schema_dsn, _ = plain_schema - await _seed(schema_dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) + await _seed(plain_sandbox.dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) stub = tmp_path / "stub" stub.mkdir() (stub / "asyncpg.py").write_text( @@ -285,7 +388,7 @@ class TestMissingAsyncpg: "--backend", "postgres", "--dsn", - schema_dsn, + plain_sandbox.dsn, "--older-than-days", "7", "--apply", @@ -295,4 +398,4 @@ class TestMissingAsyncpg: assert result.returncode == 2, (result.stdout, result.stderr) assert "asyncpg" in result.stderr assert "pip install" in result.stderr - assert await _call_ids(schema_dsn) == ["old-1"] + assert await _call_ids(plain_sandbox.dsn) == ["old-1"] diff --git a/tests/unit/test_retention_tool.py b/tests/unit/test_retention_tool.py index 249ea5a..94ef3e3 100644 --- a/tests/unit/test_retention_tool.py +++ b/tests/unit/test_retention_tool.py @@ -261,6 +261,104 @@ class TestUsageErrors: assert result.returncode == 1 + # --- --table 的参数分类(issue #18 设计 §4.2);真实解析行为在集成层验 --- + + def test_sqlite_with_table_exits_one(self, tmp_path): + """SQLite 库文件即目标,无 schema 概念,故 `--table` 在该分支无歧义可消。""" + result = _run( + "--backend", + "sqlite", + "--path", + str(tmp_path / "x.db"), + "--older-than-days", + "7", + "--table", + "some_schema.llm_calls", + ) + + assert result.returncode == 1 + assert "--table" in result.stderr + # 单看退出码与 "--table" 字样会被 argparse 的 "unrecognized arguments" 蒙混 + # 过去(它也是退出 1、也回显参数名)。必须钉住"参数已被识别、因规则被拒"。 + assert "unrecognized" not in result.stderr + + def test_table_without_schema_qualifier_exits_one(self): + """单段等于没声明: 目标仍由 `search_path` 决定,隐式性原样保留,故拒绝。""" + result = _run( + "--backend", + "postgres", + "--dsn", + "postgresql://x/y", + "--older-than-days", + "7", + "--table", + "llm_calls", + ) + + assert result.returncode == 1 + assert "--table" in result.stderr + # 单看退出码与 "--table" 字样会被 argparse 的 "unrecognized arguments" 蒙混 + # 过去(它也是退出 1、也回显参数名)。必须钉住"参数已被识别、因规则被拒"。 + assert "unrecognized" not in result.stderr + + def test_table_with_empty_segment_exits_one(self): + result = _run( + "--backend", + "postgres", + "--dsn", + "postgresql://x/y", + "--older-than-days", + "7", + "--table", + ".llm_calls", + ) + + assert result.returncode == 1 + assert "--table" in result.stderr + # 单看退出码与 "--table" 字样会被 argparse 的 "unrecognized arguments" 蒙混 + # 过去(它也是退出 1、也回显参数名)。必须钉住"参数已被识别、因规则被拒"。 + assert "unrecognized" not in result.stderr + + def test_table_with_quote_in_a_segment_exits_one(self): + """含引号的复杂标识符不支持: 此时退回不给 `--table` 的路径(见 epilog)。""" + result = _run( + "--backend", + "postgres", + "--dsn", + "postgresql://x/y", + "--older-than-days", + "7", + "--table", + 'sch"ema.llm_calls', + ) + + assert result.returncode == 1 + assert "--table" in result.stderr + # 单看退出码与 "--table" 字样会被 argparse 的 "unrecognized arguments" 蒙混 + # 过去(它也是退出 1、也回显参数名)。必须钉住"参数已被识别、因规则被拒"。 + assert "unrecognized" not in result.stderr + + def test_table_naming_another_table_exits_one(self): + """表名段锁死: 不加这条,`--table` 会把本脚本扩成"任意同形表删除工具"。""" + result = _run( + "--backend", + "postgres", + "--dsn", + "postgresql://x/y", + "--older-than-days", + "7", + "--table", + "audit.events", + ) + + assert result.returncode == 1 + assert "--table" in result.stderr + # 单看退出码与 "--table" 字样会被 argparse 的 "unrecognized arguments" 蒙混 + # 过去(它也是退出 1、也回显参数名)。必须钉住"参数已被识别、因规则被拒"。 + assert "unrecognized" not in result.stderr + # 错误消息要当场把边界说清: 本脚本的作用域到 llm_calls 为止 + assert "llm_calls" in result.stderr + class TestHelp: def test_help_names_the_maintenance_role_and_the_recommended_path(self): @@ -271,3 +369,12 @@ class TestHelp: assert "维护角色" in result.stdout assert "REVOKE" in result.stdout assert "PARTITION" in result.stdout + + def test_help_states_the_table_name_is_fixed(self): + """`--table` 只有 schema 一段可变,这条边界必须写在运维会读到的地方。""" + result = _run("--help") + + assert result.returncode == 0 + assert "--table" in result.stdout + assert "只清理" in result.stdout + assert "llm_calls" in result.stdout diff --git a/tools/telemetry_retention.py b/tools/telemetry_retention.py index 72cd8a3..ff50000 100644 --- a/tools/telemetry_retention.py +++ b/tools/telemetry_retention.py @@ -59,6 +59,11 @@ _EPILOG = """\ 时间口径: 截止时刻 = 当前 UTC 时刻 - N 天,删除 created_at < 截止时刻 的行; --older-than-days 0 即"删除此刻之前的全部行"。 +--table: 本脚本只清理表 llm_calls,故 --table 只有 schema 一段可变(写成 + --table .llm_calls)。给了它,目标就由参数精确解析、不再经 + search_path 推断。含点或引号的复杂标识符不支持,此时请不给 --table, + 退回 search_path 解析那条路径。 + 示例: python tools/telemetry_retention.py --backend sqlite --path runs/telemetry.db \\ --older-than-days 90 # dry-run,只看会删什么 @@ -114,6 +119,11 @@ def _build_parser() -> _Parser: action="store_true", help="仅 sqlite: 删除后执行 VACUUM 回收文件空间;须与 --apply 同时给", ) + parser.add_argument( + "--table", + metavar="SCHEMA.NAME", + help=f"仅 postgres: 把目标钉死为 .{TABLE},绕开 search_path 推断", + ) return parser @@ -150,10 +160,16 @@ def _validate_sqlite(parser: _Parser, args: argparse.Namespace) -> None: parser.error("--backend sqlite 不接受 --dsn") if args.batch_size is not None: parser.error("--batch-size 仅用于 --backend postgres") + if args.table is not None: + parser.error("--table 仅用于 --backend postgres:SQLite 的库文件即目标,无 schema 可消歧") def _validate_postgres(parser: _Parser, args: argparse.Namespace) -> None: - """Postgres 分支: 必须有 --dsn,拒绝 sqlite 专属参数,并在此落 --batch-size 缺省值。""" + """Postgres 分支: 必须有 --dsn,拒绝 sqlite 专属参数,并在此落 --batch-size 缺省值。 + + `--table` 在此解析成 `args.table_schema`(未给则 None): 校验与解析放在同一处, + 后面的执行路径就只面对一个已经合法的 schema 名,不必再重复判断。 + """ if args.dsn is None: parser.error("--backend postgres 需要 --dsn") if args.path is not None: @@ -164,6 +180,35 @@ def _validate_postgres(parser: _Parser, args: argparse.Namespace) -> None: args.batch_size = 1000 elif args.batch_size < 1: parser.error("--batch-size 必须 >= 1") + args.table_schema = None if args.table is None else _parse_table(parser, args.table) + + +def _parse_table(parser: _Parser, value: str) -> str: + """校验 `--table SCHEMA.NAME` 并返回 schema 段;任何不合法形态退出 1。 + + **表名段为什么不可变**: 只校验"两段、非空"的话,一次手误 `--table audit.events` + 就会让本脚本对一张恰好也有 `created_at` / `tenant_id` 的业务表跑同一套分批 DELETE。 + 脚本的名字、退出码 3 的分区提示、README 的定位全都围绕遥测表写,它从未声称自己 + 是通用清理器;把这条校验去掉等于在一个拿 DELETE 权限跑的脚本上开静默的口子。 + """ + segments = value.split(".") + if len(segments) != 2: + parser.error(f"--table 必须是 .{TABLE} 这样的两段限定名,当前: {value!r}") + schema, name = segments + # 段内不可能再含 "." (上面按 "." 切成恰好两段),故此处只需查引号 + if not schema or not name: + parser.error(f"--table 的 schema 段与表名段都不得为空,当前: {value!r}") + if '"' in schema or '"' in name: + parser.error( + f"--table 不支持含引号的复杂标识符,当前: {value!r};" + "这种情形请不给 --table,退回 search_path 解析那条路径。" + ) + if name != TABLE: + parser.error( + f"--table 的表名段必须逐字等于 {TABLE}:本脚本只清理遥测表 {TABLE}," + f"不是通用清理器,当前: {value!r}" + ) + return schema def _print_stats(total: int, low: object, high: object, tenants: Sequence[tuple[str, int]]) -> None: @@ -240,7 +285,9 @@ def _quote(identifier: str) -> str: return f'"{escaped}"' -async def _run_postgres(dsn: str, cutoff: datetime, apply_: bool, batch_size: int) -> int: +async def _run_postgres( + dsn: str, cutoff: datetime, apply_: bool, batch_size: int, table_schema: str | None +) -> int: """PostgreSQL 分支: 分区表让路,普通表分批 DELETE(每批一个事务)。""" try: import asyncpg @@ -257,7 +304,7 @@ async def _run_postgres(dsn: str, cutoff: datetime, apply_: bool, batch_size: in print(f"连接 PostgreSQL 失败: {exc}", file=sys.stderr) return EXIT_BACKEND try: - return await _purge_postgres(conn, cutoff, apply_, batch_size) + return await _purge_postgres(conn, cutoff, apply_, batch_size, table_schema) except asyncpg.PostgresError as exc: print(f"PostgreSQL 操作失败: {exc}", file=sys.stderr) return EXIT_BACKEND @@ -265,23 +312,41 @@ async def _run_postgres(dsn: str, cutoff: datetime, apply_: bool, batch_size: in await conn.close() -async def _purge_postgres(conn: Any, cutoff: datetime, apply_: bool, batch_size: int) -> int: +async def _purge_postgres( + conn: Any, cutoff: datetime, apply_: bool, batch_size: int, table_schema: str | None +) -> int: """已连上后的清理主体(conn 是 asyncpg.Connection,不 import 类型以免脚本硬依赖)。""" - # 先解析目标: to_regclass 走连接自己的 search_path,故必须把解析结果打出来—— - # "我删的到底是哪张表"是这个脚本唯一不能猜的事(共享库里另有同名表的场景常见)。 + # 给了 --table 就用引号限定名精确解析(绕开 search_path),否则维持裸表名解析—— + # 后者走连接自己的 search_path,故无论哪条路都必须把解析结果打出来:"我删的到底是 + # 哪张表"是这个脚本唯一不能猜的事(共享库里另有同名表的场景常见)。 + lookup = TABLE if table_schema is None else f"{_quote(table_schema)}.{_quote(TABLE)}" target = await conn.fetchrow( "SELECT n.nspname AS schema, c.relname AS name, " "EXISTS (SELECT 1 FROM pg_partitioned_table p WHERE p.partrelid = c.oid) AS partitioned " "FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " "WHERE c.oid = to_regclass($1)", - TABLE, + lookup, ) if target is None: - print(f"目标库的 search_path 下找不到表 {TABLE}", file=sys.stderr) + # 两条路的诊断方向不同,消息分开写: 显式指定找不到多半是名字/大小写写错了, + # search_path 找不到则是连接配置的事。 + if table_schema is None: + print(f"目标库的 search_path 下找不到表 {TABLE}", file=sys.stderr) + else: + print( + f"显式指定的表 {table_schema}.{TABLE} 不存在或当前角色不可见。" + "注意: PG 中未加引号建的标识符在 catalog 里是小写。", + file=sys.stderr, + ) return EXIT_BACKEND schema, name = target["schema"], target["name"] qualified = f"{_quote(schema)}.{_quote(name)}" print(f"目标表: {schema}.{name}") + if apply_ and table_schema is None: + # 只在 --apply 时提示: dry-run 不可逆性为零,且它本就以"看清楚再决定"为用途。 + print( + "注意: 目标表由连接的 search_path 推断得到。要把目标钉死,请加 --table .<表名>。" + ) if target["partitioned"]: print( f"{schema}.{name} 是分区表: 本脚本拒绝对分区表执行 DELETE。\n" @@ -347,7 +412,9 @@ def main(argv: Sequence[str] | None = None) -> int: print(f"模式: {'apply(将真正删除)' if args.apply else 'dry-run(只统计,不删除)'}") if args.backend == "sqlite": return _run_sqlite(args.path, cutoff.strftime(_SQLITE_TIME_FORMAT), args.apply, args.vacuum) - return asyncio.run(_run_postgres(args.dsn, cutoff, args.apply, args.batch_size)) + return asyncio.run( + _run_postgres(args.dsn, cutoff, args.apply, args.batch_size, args.table_schema) + ) if __name__ == "__main__":