From bc071c6f4145bbebd239fbe1dfac196d859fb309 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 24 Aug 2026 09:53:23 -0400 Subject: [PATCH] fix: make closing the telemetry pool bounded and final Closing was the last unbounded wait on the shutdown path: asyncpg's Pool.close() awaits wait_until_released() on every holder, so a single in-flight connection parks the caller forever (60s only buys a warning). It now runs under asyncio.wait_for and terminates the pool on timeout; external cancellation still propagates untouched. Closing is also final now. Clearing _pool used to leave the recorder free to build a fresh pool on the next write - worse in the injected case, where the owner believes it still holds every connection while the recorder quietly opened its own. Recovery is a runtime concern (cooldown retry), not a side effect of shutdown, so writes after aclose short out and count the dropped row with a reason of their own. Also covers the release/terminate fallback left untested by the pool work: the fake pool needed for the close cases makes it nearly free. --- src/polygateway/telemetry/postgres.py | 63 ++++++++-- tests/unit/test_telemetry.py | 168 +++++++++++++++++++++++++- 2 files changed, 217 insertions(+), 14 deletions(-) diff --git a/src/polygateway/telemetry/postgres.py b/src/polygateway/telemetry/postgres.py index cf8ed53..37947e7 100644 --- a/src/polygateway/telemetry/postgres.py +++ b/src/polygateway/telemetry/postgres.py @@ -47,6 +47,13 @@ _TABLE_EXISTS = "SELECT to_regclass('llm_calls')" # asyncpg 会在下次 acquire 时补一条新连接 _RELEASE_TIMEOUT_S = 1.0 +# 关闭池的独立上限(issue #15)。**不**复用写入预算: 关闭跑在收尾路径而非业务 +# 路径上,给它一个略宽的固定额度即可,但必须**有界**——asyncpg 的 +# `Pool.close()` 会 await 每个 holder 的 `wait_until_released()`,in-flight +# 连接不归还就无限等(`pool.py:939-948, 961-972`,60s 只发一条 warning), +# 其 docstring 自己写着 "advisable to use asyncio.wait_for to set a timeout" +_CLOSE_TIMEOUT_S = 5.0 + # 探测现有列;尊重 search_path(to_regclass 按当前 search_path 解析) _EXISTING_COLUMNS = ( "SELECT attname FROM pg_attribute " @@ -99,6 +106,7 @@ class PostgresRecorder: self._insert = insert_sql("postgres", COLUMNS) self._schema_ready = False self._failed = False # 结构性降级标志: 置位后所有写入短路 + self._closed = False # 关了就是关了: 置位后写入短路且**不重建池** # 降级的可编程出口与节流日志;`_failed` 与它并存是 issue #15 的过渡态, # 判据改造(冷却自愈)落地时状态收归 tracker 一处 self._status = TelemetryStatusTracker(backend="postgres", now=now) @@ -110,13 +118,18 @@ class PostgresRecorder: return self._status.snapshot() async def _ensure_ready(self) -> asyncpg.Pool | None: - """lazy 建池+备表;判死只认「确定写不进去」(issue #9),其余失败都留活路。""" - if self._failed: + """lazy 建池+备表;判死只认「确定写不进去」(issue #9),其余失败都留活路。 + + `_closed` 在锁内**必须复查**: 等锁期间发生的 `aclose` 否则会被这次 + 等待"绕过",等到锁时照旧建出一个没人负责关的池(注入档更隐蔽—— + 注入方以为自己管着全部连接,实际早已不是)。 + """ + if self._closed or self._failed: return None if self._schema_ready: return self._pool async with self._init_lock: - if self._failed: + if self._closed or self._failed: return None if self._schema_ready: return self._pool @@ -317,8 +330,10 @@ class PostgresRecorder: """ pool = await self._ensure_ready() if pool is None: - # 降级期间静默 return 就是 issue #15 的破口: 丢行必须计数且节流出声 - self._status.record_drop("遥测已降级") + # 降级期间静默 return 就是 issue #15 的破口: 丢行必须计数且节流出声。 + # 关闭后的丢行同样要计数,但原因不是降级——两者的处置完全不同 + # (一个等自愈,一个是调用方自己关了却还在写) + self._status.record_drop("遥测已关闭" if self._closed else "遥测已降级") return row = tuple(fields[col] for col in self._columns) conn = await pool.acquire(timeout=self._write_timeout_s) @@ -346,21 +361,43 @@ class PostgresRecorder: # 含 TimeoutError: 归还超时与归还出错的处置相同——断开而不是留一条 # 状态不明的连接在池里(asyncpg 的 reset 失败路径也是这么做的) logger.warning("Postgres 遥测连接归还失败(强制断开): {}", exc) - self._terminate(conn) + self._terminate(conn, label="连接") @staticmethod - def _terminate(conn: object) -> None: - """强制断开一条连接;断开本身再失败也只记 warning(遥测绝不冒泡)。""" + def _terminate(target: object, *, label: str) -> None: + """强制断开一条连接或整个池;断开本身再失败也只记 warning(遥测绝不冒泡)。 + + `label` 必填(不给默认值): 两个调用点的诊断价值全在"拆的是哪一层", + 默认值只会让其中一处悄悄报错成另一处。 + """ try: - conn.terminate() # type: ignore[attr-defined] + target.terminate() # type: ignore[attr-defined] except asyncio.CancelledError: raise except Exception as exc: - logger.warning("Postgres 遥测连接断开失败(交给池自行回收): {}", exc) + logger.warning("Postgres 遥测{}断开失败(交给上层自行回收): {}", label, exc) async def aclose(self) -> None: - """幂等关闭自建池;注入的池归注入方管理。""" + """幂等关闭自建池;**关了就是关了**,此后写入短路且不复活。注入的池归注入方管理。 + + 取消"关完还能自己重建池"的灰色状态(设计 §3.2 第 4 点): 关闭是所有权的 + 终结,而恢复是运行时行为(冷却重试),不该是关闭动作的副作用。 + + **关闭动作本身也有界**: `Pool.close()` 会 await 每个 holder 的 + `wait_until_released()`,in-flight 连接不归还就无限等——收尾路径上照样是 + "遥测拖垮业务"。超时即 `terminate()` 强拆: 关闭已在进行,留着一个关不掉的池 + 既不会自愈也没人再来收。外部取消照常穿透(铁律),不当成一次关闭超时。 + """ + self._closed = True pool, self._pool = self._pool, None self._schema_ready = False - if pool is not None and not self._external_pool: - await pool.close() + if pool is None or self._external_pool: + return + try: + await asyncio.wait_for(pool.close(), timeout=_CLOSE_TIMEOUT_S) + except asyncio.CancelledError: + raise + except Exception as exc: + # 含 TimeoutError: 关不掉与关出错的处置相同——强拆 + logger.warning("Postgres 遥测池关闭失败(强制断开): {}", exc) + self._terminate(pool, label="池") diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index b594633..84a722e 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -728,6 +728,7 @@ class _FakePgConn: fail_create: bool = False, probe_errors: int = 0, hang_insert: bool = False, + fail_terminate: bool = False, ): self.existing = existing self.fail_alter = fail_alter @@ -735,8 +736,15 @@ class _FakePgConn: self.probe_errors = probe_errors # 只挂 INSERT: 准备期照常完成,挂住的才是业务路径上那次内联 await self.hang_insert = hang_insert + self.fail_terminate = fail_terminate + self.terminated = False self.statements: list[str] = [] + def terminate(self): + self.terminated = True + if self.fail_terminate: + raise RuntimeError("connection is already closed") + async def execute(self, sql, *args): self.statements.append(sql) if sql.startswith("INSERT INTO") and self.hang_insert: @@ -769,11 +777,24 @@ class _FakePgPool: 故这里也不再提供上下文管理器。 """ - def __init__(self, conn, *, hang_acquire: bool = False): + def __init__( + self, + conn, + *, + hang_acquire: bool = False, + fail_release: bool = False, + hang_close: bool = False, + ): self._conn = conn self.hang_acquire = hang_acquire + self.fail_release = fail_release + # 模拟 asyncpg 的 `Pool.close()` 在 in-flight 连接未归还时**无限等** + # (pool.py:939-948, 961-972 只在 60s 发一条 warning) + self.hang_close = hang_close self.acquired = 0 self.released = 0 + self.close_calls = 0 + self.terminated = False self.acquire_timeouts: list[object] = [] self.release_timeouts: list[object] = [] @@ -787,8 +808,18 @@ class _FakePgPool: async def release(self, conn, *, timeout=None): assert conn is self._conn self.release_timeouts.append(timeout) + if self.fail_release: + raise RuntimeError("connection reset failed") self.released += 1 + async def close(self): + self.close_calls += 1 + if self.hang_close: + await asyncio.sleep(3600) + + def terminate(self): + self.terminated = True + class TestPostgresBackfillDiscipline: """PG 补列必须与 SQLite 侧对称: 失败只逐行降级,且稳态不抢排他锁(issue #3)。""" @@ -2014,3 +2045,138 @@ class TestPostgresPoolResourceSemantics: with pytest.raises(asyncio.CancelledError): await task assert pool.released == pool.acquired # 取消路径上也不许泄漏连接 + + +def _pg_recorder(*, pool=None, **overrides): + """本文件统一的 PG recorder 构造口: 池上限与写入预算取同一份测试常量。""" + from polygateway.telemetry.postgres import PostgresRecorder + + kwargs: dict[str, object] = { + "auto_migrate": True, + "pool_max": _TEST_POOL_MAX, + "write_timeout_s": _TEST_WRITE_TIMEOUT_S, + } + kwargs.update(overrides) + return PostgresRecorder(_PG_DSN, pool=pool, **kwargs) + + +class TestPostgresCloseIsBounded: + """issue #15 B 组: 关闭动作本身必须有界,且"关了就是关了"(设计 §3.2 第 4 点)。 + + 两个缺口各钉一次: ① `Pool.close()` 会 await 每个 holder 的 + `wait_until_released()`,in-flight 未归还时无限等 —— 收尾路径上照样是 + "遥测拖垮业务";② 关完还能自己重建池的灰色状态 —— 关闭是所有权终结, + 恢复归运行时的冷却机制管,不该是关闭动作的副作用。 + """ + + def _self_built(self, monkeypatch, pool): + """让 recorder 走**自建池**那条路,并交回建池次数(复活的唯一证据)。""" + import asyncpg + + created: list[str] = [] + + async def fake_create_pool(dsn, **kwargs): + created.append(dsn) + return pool + + monkeypatch.setattr(asyncpg, "create_pool", fake_create_pool) + return _pg_recorder(), created + + async def test_stuck_pool_close_falls_back_to_terminate(self, monkeypatch, captured_warnings): + """**主回归钉子**: 池关不掉时超时即 terminate,绝不无限期挂在收尾路径上。""" + from polygateway.telemetry import postgres + + monkeypatch.setattr(postgres, "_CLOSE_TIMEOUT_S", 0.05) + pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)), hang_close=True) + recorder, _ = self._self_built(monkeypatch, pool) + await _record_minimal(recorder) + + loop = asyncio.get_running_loop() + started = loop.time() + # 用例自带超时: 实现无界时这里要当场红,而不是挂死整个套件 + await asyncio.wait_for(recorder.aclose(), timeout=5) + + assert loop.time() - started < 1.0 + assert pool.close_calls == 1 and pool.terminated is True + assert captured_warnings # 强制拆池是异常路径,不许静默 + + async def test_writes_after_close_do_not_rebuild_the_pool(self, monkeypatch): + """关了就是关了: 后续写入短路丢行,**不**再建一个没人负责关的池。""" + pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS))) + recorder, created = self._self_built(monkeypatch, pool) + await _record_minimal(recorder) + assert len(created) == 1 + + await recorder.aclose() + dropped_before = recorder.telemetry_status.dropped_rows + await _record_minimal(recorder, call_id="c2") # 遥测绝不冒泡 + + assert len(created) == 1 # 复活的唯一证据: 第二次 create_pool + assert pool.acquired == 2 # 准备期 + 首次写入;关闭后一次都没有 + assert recorder.telemetry_status.dropped_rows == dropped_before + 1 + + async def test_aclose_is_idempotent(self, monkeypatch): + pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS))) + recorder, _ = self._self_built(monkeypatch, pool) + await _record_minimal(recorder) + await recorder.aclose() + await recorder.aclose() + assert pool.close_calls == 1 + + async def test_injected_pool_is_left_to_its_owner(self, monkeypatch): + """注入的池既不关也不拆(既有纪律),但 recorder 自己照样"关了就是关了"。 + + 注入档的复活更隐蔽: 关闭把 `_pool` 置 None 后,下一次写入会拿 DSN + **自建**一个池——注入方以为自己管着全部连接,实际早已不是。 + """ + import asyncpg + + pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS))) + created: list[str] = [] + + async def fake_create_pool(dsn, **kwargs): + # 不能直接 raise: recorder 会把它当建池失败吞掉,用例就白测了 + created.append(dsn) + return _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS))) + + monkeypatch.setattr(asyncpg, "create_pool", fake_create_pool) + recorder = _pg_recorder(pool=pool) + await _record_minimal(recorder) + await recorder.aclose() + + assert pool.close_calls == 0 and pool.terminated is False + await _record_minimal(recorder, call_id="c2") + assert created == [] # 注入档的复活: 拿 DSN 另起一个池 + assert pool.acquired == 2 # 关闭后也不再往注入的池上写 + + async def test_external_cancellation_during_close_is_not_swallowed(self, monkeypatch): + """铁律"取消可穿透": 有界关闭不得把外部取消吃成一次超时降级。""" + pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)), hang_close=True) + recorder, _ = self._self_built(monkeypatch, pool) + await _record_minimal(recorder) + + task = asyncio.create_task(recorder.aclose()) + await asyncio.sleep(0.05) # 让它跑进那次挂住的 close() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +class TestPostgresReleaseDegradation: + """归还连接失败时的防御路径(T3 未覆盖的缺口,借 T4 的假池顺带钉住)。 + + 留一条状态不明的连接在池里比断开更坏: 它会被下次 acquire 取到, + 把一次失败放大成持续失败。 + """ + + async def test_failed_release_terminates_the_connection(self, captured_warnings): + conn = _FakePgConn(list(_EXPECTED_COLUMNS)) + await _record_minimal(_pg_recorder(pool=_FakePgPool(conn, fail_release=True))) + assert conn.terminated is True + assert any("归还失败" in m for m in captured_warnings) + + async def test_terminate_failure_does_not_escape(self, captured_warnings): + """断开本身再失败也只记 warning: 遥测绝不冒泡,剩下的交给池自行回收。""" + conn = _FakePgConn(list(_EXPECTED_COLUMNS), fail_terminate=True) + await _record_minimal(_pg_recorder(pool=_FakePgPool(conn, fail_release=True))) + assert any("断开失败" in m for m in captured_warnings)