diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index 68b6d29..b645e51 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 +import time from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path @@ -319,14 +320,74 @@ class TestSchema: await recorder.aclose() +class _FakeClock: + """可手动推进的单调时钟: 冷却窗口靠它测,用例里绝不真睡 60 秒。""" + + def __init__(self, start: float = 1_000.0) -> None: + self.t = start + + def __call__(self) -> float: + return self.t + + def advance(self, seconds: float) -> None: + self.t += seconds + + class TestDegradation: async def test_unreachable_server_degrades_silently(self): - """结构性失败(建池不通)→ warning 一次后永久降级,业务零感知。""" + """服务端连不上 → warning 一次后降级,业务零感知(不抛、不拖)。""" recorder = _recorder("postgresql://u:p@127.0.0.1:1/x", auto_migrate=True) await _record_minimal(recorder) # 不抛 await _record_minimal(recorder, call_id=_cid("c2")) # 已降级短路,同样不抛 await recorder.aclose() + async def test_refused_connection_cools_down_and_retries_after_cooldown(self): + """连接被拒 → 冷却降级(**非 fatal**)→ 冷却期内零成本短路 → 到期真的重试。 + + 走**不可达 DSN** 而不是把共享实例的连接打满: 那台 PG 上还有 app/chs_prod + 等在用库,制造连接耗尽会伤到别人;而"连接被拒"与"连接耗尽"落的是同一档 + (环境级,`_classify_failure`),这条路验的是同一段状态机。 + + **时序前提**(避免间歇红): 假时钟只驱动 tracker 的冷却窗口,与真实网络耗时 + 完全无关,故三段断言都不依赖墙钟。`retry_after_s` 是"有没有真的重试过"的 + 唯一外部信号——重试失败会给冷却窗口续期,而短路不会碰它。 + """ + clock = _FakeClock() + recorder = PostgresRecorder( + "postgresql://u:p@127.0.0.1:1/x", + auto_migrate=True, + pool_max=_POOL_MAX, + write_timeout_s=_WRITE_TIMEOUT_S, + now=clock, + ) + try: + await _record_minimal(recorder, call_id=_cid("deg1")) + first = recorder.telemetry_status + # 非 fatal 正是 issue #15 的核心: 连接被拒过去在建池那一步被一刀判死, + # 整进程从此一行遥测都不落、只有重启能恢复 + assert (first.degraded, first.fatal) == (True, False) + assert first.retry_after_s == pytest.approx(60.0) + assert first.dropped_rows == 1 + # min_size=0 之后建池不再触库,连接被拒因此暴露在准备期而不是建池期 + assert "建表探测失败" in (first.reason or "") + + clock.advance(30.0) + await _record_minimal(recorder, call_id=_cid("deg2")) + mid = recorder.telemetry_status + # 冷却窗口没被刷新 = 这次调用压根没去连库(降级期间零成本短路) + assert mid.retry_after_s == pytest.approx(30.0) + assert mid.dropped_rows == 2 + + clock.advance(30.1) + await _record_minimal(recorder, call_id=_cid("deg3")) + after = recorder.telemetry_status + # 冷却窗口被重新拉满 = 真的重连了一次(照旧被拒,故仍降级但仍可自愈) + assert after.retry_after_s == pytest.approx(60.0) + assert (after.degraded, after.fatal) == (True, False) + assert after.dropped_rows == 3 + finally: + await recorder.aclose() + async def test_row_failure_does_not_poison_later_rows(self, dsn): """运行时单条写失败(NUL 字节文本被 PG 拒)→ 丢该行,后续行照常落库。""" recorder = _recorder(dsn, auto_migrate=True) @@ -349,6 +410,77 @@ class TestDegradation: await recorder.aclose() +def _tagged(dsn: str, app_name: str) -> str: + """给 DSN 挂上 `application_name` 查询参数,让本池的连接在服务端可被点名。 + + 走 DSN 参数而不是给 recorder 加 `server_settings` 入口: 纯测试便利不值得 + 扩公共 API(P1)。也**不能**改成"测试自建池后以 `pool=` 注入"——那会走 + `_external_pool` 分支、完全绕过被测的建池路径,而本节要验的恰恰是它。 + """ + sep = "&" if "?" in dsn else "?" + return f"{dsn}{sep}application_name={app_name}" + + +async def _pool_backend_count(dsn: str, app_name: str) -> int: + """数**本池**在服务端的连接数(只读查询,不改实例任何状态)。 + + 只按 run 级唯一的 `application_name` 过滤: 这台实例被多项目共用,按库名或 + 用户名计数会把别人的连接算进来,做出的是设计上就会间歇红的用例 + (CLAUDE.md §4.6)。本查询自己那条连接走未打 tag 的 DSN,故不会数到自己。 + """ + rows = await _fetch( + dsn, "SELECT count(*) AS n FROM pg_stat_activity WHERE application_name = $1", app_name + ) + return rows[0]["n"] + + +async def _settled_backend_count(dsn: str, app_name: str, *, timeout_s: float = 5.0) -> int: + """等本 tag 的连接数归零并返回最终值;超时则返回当下值,交给断言去红。 + + 轮询而不是一次采样: 客户端 `close()` 返回与服务端后台进程从 + `pg_stat_activity` 消失之间没有同步保证(实测立即归零,5s 余量只是不赌它)。 + """ + deadline = time.monotonic() + timeout_s + while True: + count = await _pool_backend_count(dsn, app_name) + if count == 0 or time.monotonic() >= deadline: + return count + await asyncio.sleep(0.1) + + +class TestPoolFootprint: + """issue #15 的直接回归钉子: 池不预连接,占用不超过库自己声明的上限。 + + 单元层断的是"`min_size`/`max_size` 传对了",这里断的是"服务端真的只开了 + 那么多连接"——两件事,只有真实 PG 能证后者。 + """ + + async def test_pool_does_not_preconnect_and_stays_within_pool_max(self, dsn): + app_name = f"{_RUN_PREFIX}-pool" # run 级唯一,与并跑的其他运行互不可见 + recorder = _recorder(_tagged(dsn, app_name), auto_migrate=True) + try: + # 构造只记参数、不触库: 这一条与下一条合起来才是钉子——修复前 + # `create_pool` 继承 asyncpg 的 min_size=10,首次写入后下面会是 10 + assert await _pool_backend_count(dsn, app_name) == 0 + + await _record_minimal(recorder, call_id=_cid("fp1")) + # **时序前提**: 写入已 await 到返回,连接必然已建立(没建立就写不成功), + # 归还只是还进池而不断开,asyncpg 空闲回收是 300s 不会在用例内触发。 + # 故这是个确定值,不是"某一刻恰好的采样" + assert await _pool_backend_count(dsn, app_name) == 1 + + await asyncio.gather( + *(_record_minimal(recorder, call_id=_cid(f"fp{i}")) for i in range(2, 22)) + ) + steady = await _pool_backend_count(dsn, app_name) + # 上界由 max_size 保证;下界 ≥1 不是凑数——它确保过滤条件真的命中了本池, + # 否则 tag 一旦拼错,上面那条 ==0 会以"永远绿"的形态通过 + assert 1 <= steady <= _POOL_MAX + finally: + await recorder.aclose() + assert await _settled_backend_count(dsn, app_name) == 0 # 关闭即归还全部连接 + + _PROBE_PASSWORD = "pgw_issue9_probe" # 临时角色,teardown 删除;非任何真实凭据