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.

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)