test: prove on real PG that the pool never preconnects

The min_size=10 default survived to 1.2.4 because every PG test injected a
pool and thus skipped the pool-building path entirely. Unit tests now assert
the create_pool arguments, but "we passed min_size=0" and "the server really
opened that many backends" are two different claims, and only a real instance
can settle the second one. Count via a run-unique application_name carried on
the DSN: the instance is shared with other projects, so counting by database
or role would fold their connections into ours and make the case flaky by
construction.

Degradation is exercised through an unreachable DSN rather than by exhausting
the shared instance's connections. A refused connection lands in the same
class as exhaustion, and the fake clock lets the 60s cooldown be observed
without sleeping. retry_after_s is the signal that separates a real retry
(which renews the window) from the cheap short circuit (which does not).

Evidence: with create_pool reverted to its pre-fix form both cases go red
(observed 10 backends after a single write, and refusal surfacing at pool
creation instead of at prepare time).

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 删除;非任何真实凭据