fix: judge telemetry failures by nature, not by step

The pool exhaustion in issue #15 was fatal only because min_size=10 forced
a transient error to surface at pool creation, and that step was hardcoded
to permanent death. Step is the wrong axis: it conflates "the DSN cannot
be parsed" with "someone else holds all the connections right now".

Failures are now classified by two rules. Fatal means the cause lies
entirely inside this process and cannot change, which only the
construction-time DSN satisfies. Everything else splits on whether the
failure has anything to do with this row's data: row-level failures drop
one row and keep trying, environment-level failures cool down for 60s and
then get exactly one retry, so a restarted database or a DBA creating the
table heals on its own.

42703 (missing column) is the single named exception and stays row-level
even though every row fails alike: issue #13 promised that the manual mode
trims the INSERT and exposes drift per row, and that promise outranks the
rule. Any future exception owes the same argument.

The _failed boolean is gone; the tracker is the only degradation state,
because two copies of the same fact drift apart. Closing stays outside
that state: it is the caller's own decision, not an anomaly to recover
from, so the snapshot reports it through dropped_rows and the drop reason
instead of raising the degraded flag on every clean shutdown.

tests/integration/test_postgres_telemetry.py

@@ -413,13 +413,13 @@ class TestLeastPrivilegeDeployment:
             await conn.close()
 
     async def test_records_land_without_schema_create_privilege(self, least_privilege_dsn):
-        """修复前: 建表被拒 → _failed → 整个进程一条不落(下游 150 次调用全丢)。"""
+        """修复前: 建表被拒 → 整体判死 → 整个进程一条不落(下游 150 次调用全丢)。"""
         low_dsn, schema = least_privilege_dsn
         recorder = _recorder(low_dsn, auto_migrate=True)
         try:
             await _record_minimal(recorder, call_id=_cid("lp1"))
             await _record_minimal(recorder, call_id=_cid("lp2"), cost=1.5)
-            assert recorder._failed is False  # 判死开关不得被建表权限触发
+            assert recorder.telemetry_status.degraded is False  # 建表权限不得触发降级
             rows = await _fetch(
                 low_dsn,
                 "SELECT call_id, cost FROM llm_calls WHERE call_id LIKE $1 ORDER BY call_id",
@@ -665,15 +665,15 @@ class TestCallerDimensionsAcceptance:
     async def test_backfill_failure_degrades_per_row_not_wholesale(
         self, least_privilege_pre_tenant_dsn, captured_warnings
     ):
-        """补列失败的降级方向: 记 warning、不置 `_failed`、后续 INSERT 仍照发。
+        """补列失败的降级方向: 记 warning、不整体降级、后续 INSERT 仍照发。
 
-        置 `_failed` 会让整个进程从此一条遥测都不写(比逐行丢弃严重得多),
-        且一旦 DBA 补上列也不会自愈——必须等重启。
+        整体降级会让整个进程停写(比逐行丢弃严重得多),而缺列(SQLSTATE 42703)
+        是判据的唯一具名例外: 必须逐行暴露,好让下游看见 schema 漂移(issue #13)。
         """
         recorder = _recorder(least_privilege_pre_tenant_dsn, auto_migrate=True)
         try:
             await _record_minimal(recorder, call_id=_cid("lpp1"))  # 不得抛
-            assert recorder._failed is False
+            assert recorder.telemetry_status.degraded is False
             assert any("补列失败" in m for m in captured_warnings)
             # 缺列的表上 INSERT 必然失败;逐行 warning 正是"INSERT 照发了"的证据
             assert any("写入失败" in m for m in captured_warnings)
@@ -867,7 +867,7 @@ class TestManualSchemaModeAcceptance:
 
             assert [m for m in captured_warnings if "补列失败" in m] == []
             assert [m for m in captured_warnings if "写入失败" in m] == []
-            assert recorder._failed is False
+            assert recorder.telemetry_status.degraded is False
             notices = [m for m in captured_warnings if "auto_migrate=False" in m]
             assert len(notices) == 1  # 准备期一次,第二行不再重复
             assert "以下维度不会被记录: tenant_id, meta" in notices[0]

tests/unit/test_telemetry.py

@@ -729,6 +729,7 @@ class _FakePgConn:
         probe_errors: int = 0,
         hang_insert: bool = False,
         fail_terminate: bool = False,
+        insert_error: BaseException | None = None,
     ):
         self.existing = existing
         self.fail_alter = fail_alter
@@ -737,6 +738,8 @@ class _FakePgConn:
         # 只挂 INSERT: 准备期照常完成,挂住的才是业务路径上那次内联 await
         self.hang_insert = hang_insert
         self.fail_terminate = fail_terminate
+        # INSERT 阶段抛出的真实 PG 异常(带 SQLSTATE),用来钉失败三分的边界
+        self.insert_error = insert_error
         self.terminated = False
         self.statements: list[str] = []
 
@@ -747,8 +750,11 @@ class _FakePgConn:
 
     async def execute(self, sql, *args):
         self.statements.append(sql)
-        if sql.startswith("INSERT INTO") and self.hang_insert:
-            await asyncio.sleep(3600)
+        if sql.startswith("INSERT INTO"):
+            if self.hang_insert:
+                await asyncio.sleep(3600)
+            if self.insert_error is not None:
+                raise self.insert_error
         if sql.startswith("ALTER TABLE") and self.fail_alter:
             raise RuntimeError("must be owner of table llm_calls")
         if sql.lstrip().startswith("CREATE TABLE"):
@@ -784,8 +790,12 @@ class _FakePgPool:
         hang_acquire: bool = False,
         fail_release: bool = False,
         hang_close: bool = False,
+        acquire_error: BaseException | None = None,
     ):
         self._conn = conn
+        # 取连接阶段抛出的真实异常: 连接耗尽/DSN 非法都在这一步现形
+        # (`min_size=0` 之后建池不触库,实测 create_pool 连 DSN 都不解析)
+        self.acquire_error = acquire_error
         self.hang_acquire = hang_acquire
         self.fail_release = fail_release
         # 模拟 asyncpg 的 `Pool.close()` 在 in-flight 连接未归还时**无限等**
@@ -802,6 +812,8 @@ class _FakePgPool:
         self.acquire_timeouts.append(timeout)
         if self.hang_acquire:
             await asyncio.sleep(3600)
+        if self.acquire_error is not None:
+            raise self.acquire_error
         self.acquired += 1
         return self._conn
 
@@ -849,11 +861,11 @@ class TestPostgresBackfillDiscipline:
         )
 
     async def test_alter_failure_does_not_disable_the_recorder(self):
-        """ALTER 失败(如账号只有 INSERT 权限)不得置 _failed —— 那会让遥测全灭。"""
+        """ALTER 失败(如账号只有 INSERT 权限)不得让 recorder 降级 —— 那会让遥测全灭。"""
         conn = _FakePgConn(self._LEGACY, fail_alter=True)
         recorder = self._recorder(conn)
         await _record_minimal(recorder)  # 不得抛
-        assert recorder._failed is False
+        assert recorder.telemetry_status.degraded is False
         assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)
 
     async def test_no_alter_when_columns_already_exist(self):
@@ -922,7 +934,7 @@ class TestPostgresTableProbe:
         conn = _FakePgConn(self._CURRENT, fail_create=True)
         recorder = self._recorder(conn)
         await _record_minimal(recorder)  # 不得抛
-        assert recorder._failed is False
+        assert recorder.telemetry_status.degraded is False
         assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)
 
     async def test_missing_table_is_created_and_not_backfilled(self):
@@ -932,15 +944,16 @@ class TestPostgresTableProbe:
         await _record_minimal(recorder)
         assert len(self._created(conn)) == 1
         assert not [s for s in conn.statements if s.startswith("ALTER TABLE")]
-        assert recorder._failed is False
+        assert recorder.telemetry_status.degraded is False
         assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)
 
-    async def test_create_failure_on_missing_table_degrades_to_noop(self):
-        """表确定不存在且建不出来 = 确定写不进去: 此时才允许永久 no-op。"""
+    async def test_create_failure_on_missing_table_enters_cooldown(self):
+        """表确定不存在且建不出来 = 环境级(DBA 建了表就该好): 冷却降级,不判死。"""
         conn = _FakePgConn([], fail_create=True)
         recorder = self._recorder(conn)
         await _record_minimal(recorder)  # 不得抛
-        assert recorder._failed is True
+        status = recorder.telemetry_status
+        assert status.degraded is True and status.fatal is False
         assert not [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
 
     async def test_probe_failure_is_transient_not_terminal(self):
@@ -948,7 +961,8 @@ class TestPostgresTableProbe:
         conn = _FakePgConn(self._CURRENT, probe_errors=1)
         recorder = self._recorder(conn)
         await _record_minimal(recorder, call_id="first")  # 不得抛
-        assert recorder._failed is False
+        # 不认识的失败不给升级: 探测抖动只丢本行,绝不进 60s 冷却(issue #9)
+        assert recorder.telemetry_status.degraded is False
         assert not [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
         await _record_minimal(recorder, call_id="second")
         assert [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
@@ -1914,7 +1928,7 @@ class TestSQLiteStatusVisibility:
 
 
 class TestPostgresStatusVisibility:
-    """PG 侧的判死本任务不改判据,只让它经 tracker 变得可见(计划 T2)。"""
+    """PG 侧的降级必须能被下游查到(计划 T2 建立可见性,T5 改判据)。"""
 
     def _recorder(self, conn):
         from polygateway.telemetry.postgres import PostgresRecorder
@@ -1928,13 +1942,15 @@ class TestPostgresStatusVisibility:
         )
 
     async def test_unusable_table_shows_up_in_the_status(self, captured_warnings):
-        """表确定不存在且建不出来 = 既有的判死档;现在它要能被下游查到。"""
+        """表确定不存在且建不出来: 降级可见,且是**可自愈**的环境级而非永久判死。"""
+        from polygateway.telemetry.postgres import _DEGRADE_COOLDOWN_S
+
         recorder = self._recorder(_FakePgConn([], fail_create=True))
         await _record_minimal(recorder)
         status = recorder.telemetry_status
-        assert status.degraded is True and status.fatal is True
-        assert status.dropped_rows == 1  # 判死那一次调用本身也丢了一行
-        assert recorder._failed is True  # 过渡期两份状态并存(T5 收掉 `_failed`)
+        assert status.degraded is True and status.fatal is False  # 环境级: 建了表就该自愈
+        assert status.dropped_rows == 1  # 降级那一次调用本身也丢了一行
+        assert status.retry_after_s == pytest.approx(_DEGRADE_COOLDOWN_S)
 
     async def test_healthy_recorder_is_not_degraded(self):
         recorder = self._recorder(_FakePgConn(list(_EXPECTED_COLUMNS)))
@@ -2115,6 +2131,29 @@ class TestPostgresCloseIsBounded:
         assert pool.acquired == 2  # 准备期 + 首次写入;关闭后一次都没有
         assert recorder.telemetry_status.dropped_rows == dropped_before + 1
 
+    async def test_close_is_not_degradation(self, monkeypatch, captured_warnings):
+        """**关闭 ≠ 降级**(T4 留下的语义问题,T5 收口)。
+
+        `degraded` 的含义是"后端本该可写却写不进去,库正在设法恢复"。关闭是调用方
+        自己的决定,没有异常、也按设计不会自愈——把它记成降级,等于让每一次正常
+        收尾都发一次降级信号,下游"degraded 就告警"的规则会被每次退出打穿。
+        关闭后真正要对账的是"还有多少行没落地",那由 `dropped_rows` 与逐条原因
+        承担,不必污染 `degraded`。
+        """
+        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)))
+        recorder, _ = self._self_built(monkeypatch, pool)
+        await _record_minimal(recorder)
+        await recorder.aclose()
+        captured_warnings.clear()
+
+        await _record_minimal(recorder, call_id="c2")
+
+        status = recorder.telemetry_status
+        assert status.degraded is False and status.fatal is False
+        assert status.reason is None and status.retry_after_s is None
+        assert status.dropped_rows == 1  # 丢了多少行照样可对账
+        assert any("遥测已关闭" in m for m in captured_warnings)  # 且分得清是哪一种丢
+
     async def test_aclose_is_idempotent(self, monkeypatch):
         pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)))
         recorder, _ = self._self_built(monkeypatch, pool)
@@ -2180,3 +2219,164 @@ class TestPostgresReleaseDegradation:
         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)
+
+
+class TestPostgresFailureClassification:
+    """issue #15 B 组: 判死判据从"哪一步失败"改为"失败是什么性质"(设计 §3.2)。
+
+    判据两句: ①**致命 = 失败原因完全在进程内部且不可变**;②**行级 vs 环境级看
+    失败与这一行的数据有没有关系**。三档边界两侧各钉一次——按 SQLSTATE 前两位
+    一刀切正是本 issue 之前的错法,回归会当场红。
+    """
+
+    def _self_built(self, monkeypatch, *, outcomes, clock):
+        """走**自建池**那条路;`outcomes` 逐次消费,元素是异常就抛出。
+
+        必须自建而非注入: 注入档走 `_external_pool=True` 分支,完全绕过建池,
+        而 issue 现场的失败恰恰发生在建池那一步。
+        """
+        import asyncpg
+
+        created: list[str] = []
+
+        async def fake_create_pool(dsn, **kwargs):
+            created.append(dsn)
+            outcome = outcomes[min(len(created) - 1, len(outcomes) - 1)]
+            if isinstance(outcome, BaseException):
+                raise outcome
+            return outcome
+
+        monkeypatch.setattr(asyncpg, "create_pool", fake_create_pool)
+        return _pg_recorder(now=clock), created
+
+    async def test_pool_exhaustion_degrades_with_cooldown_and_self_heals(self, monkeypatch):
+        """**issue 场景直接回归**: 53300 落在准备期,过去 = 整进程永久失遥测。
+
+        `too many clients` 是外部状态,别人还连接就该好——它永远不满足"原因完全
+        在进程内部且不可变",故绝不许判死,只许冷却重试。
+        """
+        import asyncpg
+
+        from polygateway.telemetry.postgres import _DEGRADE_COOLDOWN_S
+
+        clock = _FakeClock()
+        conn = _FakePgConn(list(_EXPECTED_COLUMNS))
+        recorder, created = self._self_built(
+            monkeypatch,
+            outcomes=[
+                asyncpg.exceptions.TooManyConnectionsError("sorry, too many clients already"),
+                _FakePgPool(conn),
+            ],
+            clock=clock,
+        )
+
+        await _record_minimal(recorder, call_id="c1")  # 不得抛
+        status = recorder.telemetry_status
+        assert status.degraded is True
+        assert status.fatal is False  # ← 现状在这里判死,整进程从此一条不落
+        assert status.retry_after_s == pytest.approx(_DEGRADE_COOLDOWN_S)
+        assert not conn.statements
+
+        await _record_minimal(recorder, call_id="c2")  # 冷却期内零成本短路
+        assert len(created) == 1  # 不再内联吞一次 connect 超时
+
+        clock.advance(_DEGRADE_COOLDOWN_S)
+        await _record_minimal(recorder, call_id="c3")
+
+        assert len(created) == 2  # 到期放行一次重新准备
+        assert [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
+        assert recorder.telemetry_status.degraded is False  # 自愈,无需重启进程
+        assert recorder.telemetry_status.dropped_rows == 2  # 降级期间那两行确实丢了
+
+    async def test_unparseable_dsn_is_fatal_and_costs_nothing_afterwards(self, captured_warnings):
+        """DSN 是构造期定死的字符串: 唯一"进程内不可能变好"的东西,故唯一的致命档。"""
+        import asyncpg
+
+        clock = _FakeClock()
+        pool = _FakePgPool(
+            _FakePgConn(list(_EXPECTED_COLUMNS)),
+            acquire_error=asyncpg.exceptions.ClientConfigurationError("invalid DSN: bad scheme"),
+        )
+        recorder = _pg_recorder(pool=pool, now=clock)
+
+        await _record_minimal(recorder, call_id="c1")  # 不得抛
+        status = recorder.telemetry_status
+        assert status.degraded is True and status.fatal is True
+        assert status.retry_after_s is None  # 本进程内不会自愈
+        assert any("重启" in m for m in captured_warnings)  # 恢复条件必须写在日志里
+
+        clock.advance(1_000_000.0)
+        attempts = len(pool.acquire_timeouts)
+        await _record_minimal(recorder, call_id="c2")
+        assert len(pool.acquire_timeouts) == attempts  # 此后零成本短路,不再触库
+
+    @pytest.mark.parametrize("error_name", ["InsufficientPrivilegeError", "UndefinedTableError"])
+    async def test_environment_level_sqlstates_enter_cooldown(self, error_name):
+        """`42501`/`42P01` 与这一行的数据无关(每一行都会同样失败)→ 环境级。
+
+        它们与 `42703` 同属 SQLSTATE `42` 类却分属两档: 判据看的是"失败与这一行的
+        数据有没有关系",不是前两位。
+        """
+        import asyncpg
+
+        from polygateway.telemetry.postgres import _DEGRADE_COOLDOWN_S
+
+        clock = _FakeClock()
+        conn = _FakePgConn(
+            list(_EXPECTED_COLUMNS), insert_error=getattr(asyncpg.exceptions, error_name)("boom")
+        )
+        recorder = _pg_recorder(pool=_FakePgPool(conn), now=clock)
+
+        await _record_minimal(recorder, call_id="c1")
+        status = recorder.telemetry_status
+        assert status.degraded is True and status.fatal is False
+        assert status.retry_after_s == pytest.approx(_DEGRADE_COOLDOWN_S)
+
+        inserts = len([s for s in conn.statements if s.startswith("INSERT INTO llm_calls")])
+        await _record_minimal(recorder, call_id="c2")
+        # 冷却期内不再每行内联付一次往返;权限/建表修好后由冷却到期自动恢复
+        assert len([s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]) == inserts
+
+    async def test_missing_column_stays_row_level(self, captured_warnings):
+        """`42703` 是判据的**唯一具名例外**,由 issue #13 定死: 缺列要逐行暴露。
+
+        按判据第 2 句它本该是环境级(缺列时每行都失败),归行级是因为 manual 档
+        会裁剪 INSERT 继续写,"部分列写进去了 + 逐行 warning"本身有价值,
+        不该被冷却掉——下游正是靠这条 warning 发现 schema 漂移的。
+        """
+        import asyncpg
+
+        conn = _FakePgConn(
+            list(_EXPECTED_COLUMNS),
+            insert_error=asyncpg.exceptions.UndefinedColumnError('column "meta" does not exist'),
+        )
+        recorder = _pg_recorder(pool=_FakePgPool(conn))
+
+        await _record_minimal(recorder, call_id="c1")
+        await _record_minimal(recorder, call_id="c2")
+
+        status = recorder.telemetry_status
+        assert status.degraded is False  # 不进冷却
+        assert status.dropped_rows == 2
+        # 每一行都照发 INSERT,每一行都出声: schema 漂移必须持续可见
+        assert len([s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]) == 2
+        # 逐行那条不节流(与累计计数那条区分开): 每丢一行都要出声
+        assert len([m for m in captured_warnings if "写入失败(丢弃该行" in m]) == 2
+
+    async def test_uncreatable_table_recovers_once_the_dba_creates_it(self):
+        """表建不出来是环境级: DBA 建完表,冷却到期就该自己好,不必重启进程。"""
+        from polygateway.telemetry.postgres import _DEGRADE_COOLDOWN_S
+
+        clock = _FakeClock()
+        conn = _FakePgConn([], fail_create=True)
+        recorder = _pg_recorder(pool=_FakePgPool(conn), now=clock)
+
+        await _record_minimal(recorder, call_id="c1")
+        assert recorder.telemetry_status.degraded is True
+
+        conn.existing = list(_EXPECTED_COLUMNS)  # DBA 手工建了表
+        clock.advance(_DEGRADE_COOLDOWN_S)
+        await _record_minimal(recorder, call_id="c2")
+
+        assert [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
+        assert recorder.telemetry_status.degraded is False