chore(db): 降低数据库诊断与背压 skip 日志噪音

健康态 pg pool diag 改为 DEBUG 并默认 5 分钟汇总；接话热路径 skip 日志同步降级。

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: TogetsuDo

src/features/corpus/find_cache.py

@@ -61,7 +61,7 @@ async def _cached_find(
     now = time.monotonic()
     if _reply_db_fail_active(key, now=now):
         if for_reply:
-            logger.info(
+            logger.debug(
                 "corpus_find_reply.skip reply_db_fail_cooldown kw_len={}",
                 len(key),
             )
@@ -95,7 +95,7 @@ async def _cached_find(
         if is_pg_pool_timeout_error(exc):
             mark_reply_db_fail(key)
             if for_reply:
-                logger.info(
+                logger.debug(
                     "corpus_find_reply.skip db_timeout kw_len={}",
                     len(key),
                 )

src/foundation/db/pool_diagnostics.py

@@ -1,4 +1,4 @@
-"""PG 连接池诊断：周期汇总 + 慢持连统计（默认 INFO，便于长期观测背压）。"""
+"""PG 连接池诊断：周期汇总 + 慢持连统计（健康态 DEBUG，异常态 INFO）。"""
 
 from __future__ import annotations
 
@@ -12,7 +12,7 @@
 
 from src.foundation.config.repo_settings import repo_env_raw_value
 
-_TICK_SEC = 60.0
+_TICK_SEC_DEFAULT = 300.0
 _diag_task: asyncio.Task[None] | None = None
 _bound = False
 
@@ -34,6 +34,37 @@ def session_hold_warn_ms() -> float:
     return 500.0
 
 
+def pool_diag_tick_sec() -> float:
+    raw = repo_env_raw_value("PG_POOL_DIAG_TICK_SEC")
+    if raw is not None:
+        try:
+            return max(30.0, float(str(raw).strip()))
+        except ValueError:
+            pass
+    return _TICK_SEC_DEFAULT
+
+
+def pool_diag_tick_notable(
+    *,
+    under_pressure: bool,
+    idle_in_tx: int | None,
+    slow_sessions: int,
+    remote_skipped_pressure: int,
+    remote_skipped_busy: int,
+    mirror_skip: int,
+    learn_pool_wait: int,
+) -> bool:
+    if under_pressure or slow_sessions > 0:
+        return True
+    if idle_in_tx and idle_in_tx > 0:
+        return True
+    if remote_skipped_pressure > 0 or remote_skipped_busy > 0:
+        return True
+    if mirror_skip > 0 or learn_pool_wait > 0:
+        return True
+    return False
+
+
 def _is_ignored_caller_path(path: str) -> bool:
     return path.endswith(_CALLER_SCAN_SKIP_SUFFIXES) or "/site-packages/" in path
 
@@ -161,7 +192,20 @@ async def emit_pool_diagnostics_tick() -> None:
     if not slow_top:
         slow_top = "-"
 
-    logger.info(
+    skipped_pressure = int(remote.get("skipped_pressure", 0))
+    skipped_busy = int(remote.get("skipped_busy", 0))
+    learn_pool_wait = int(learn.get("learn_pool_wait_spins", 0) or 0)
+    notable = pool_diag_tick_notable(
+        under_pressure=bool(budget.get("under_pressure")),
+        idle_in_tx=idle_tx,
+        slow_sessions=_slow_session_total,
+        remote_skipped_pressure=skipped_pressure,
+        remote_skipped_busy=skipped_busy,
+        mirror_skip=_mirror_skipped_pressure,
+        learn_pool_wait=learn_pool_wait,
+    )
+    diag_log = logger.info if notable else logger.debug
+    diag_log(
         "pg pool diag: checked_out={}/{} util={} idle_in_tx={} pg_wait=[{}] "
         "remote_skip_pressure={} remote_skip_busy={} mirror_skip={} "
         "slow_sessions={} slow_max_ms={:.0f} learn_q={} learn_pool_wait={} slow_top=[{}]",
@@ -170,13 +214,13 @@ async def emit_pool_diagnostics_tick() -> None:
         util_pct,
         idle_tx if idle_tx is not None else "?",
         wait_s,
-        remote.get("skipped_pressure", 0),
-        remote.get("skipped_busy", 0),
+        skipped_pressure,
+        skipped_busy,
         _mirror_skipped_pressure,
         _slow_session_total,
         _slow_hold_max_ms,
         learn.get("learn_queue_size", "?"),
-        learn.get("learn_pool_wait_spins", 0),
+        learn_pool_wait,
         slow_top,
     )
 
@@ -195,13 +239,13 @@ async def emit_pool_diagnostics_tick() -> None:
 
 
 async def pool_diagnostics_loop() -> None:
-    await asyncio.sleep(_TICK_SEC)
+    await asyncio.sleep(pool_diag_tick_sec())
     while True:
         try:
             await emit_pool_diagnostics_tick()
         except Exception as e:
             logger.warning("pg pool diagnostics tick failed: {}", e)
-        await asyncio.sleep(_TICK_SEC)
+        await asyncio.sleep(pool_diag_tick_sec())
 
 
 def bind_pg_pool_diagnostics() -> None:
@@ -228,7 +272,7 @@ def start_pg_pool_diagnostics_task() -> None:
     _diag_task = asyncio.create_task(pool_diagnostics_loop(), name="pg_pool_diagnostics")
     logger.debug(
         "pg pool diagnostics started (tick={}s, session_hold_warn={}ms)",
-        int(_TICK_SEC),
+        int(pool_diag_tick_sec()),
         int(session_hold_warn_ms()),
     )
 

src/foundation/db/repository_pg.py

@@ -586,13 +586,13 @@ async def cached_reply_query_snapshot(
     if not key:
         return None
     if pg_pool_under_pressure(threshold=0.55):
-        logger.info(
+        logger.debug(
             "reply_query_snapshot.skip pg_pool_pressure kw_len={}",
             len(key),
         )
         return None
     if reply_db_fail_active(key):
-        logger.info(
+        logger.debug(
             "reply_query_snapshot.skip reply_db_fail_cooldown kw_len={}",
             len(key),
         )
@@ -620,7 +620,7 @@ async def cached_reply_query_snapshot(
                 _reply_query_snapshot_inflight.pop(key, None)
         if is_pg_pool_timeout_error(exc):
             mark_reply_db_fail(key)
-            logger.info(
+            logger.debug(
                 "reply_query_snapshot.skip db_timeout kw_len={}",
                 len(key),
             )

tests/common/test_pool_diagnostics.py

@@ -33,3 +33,24 @@ class _WantedFrame:
     monkeypatch.setattr(pool_diagnostics.inspect, "stack", lambda: [None, None, _FrameInfo(), _WantedFrame()])
 
     assert pool_diagnostics.pg_session_caller_hint_entry() == "fallback_wrapper@common/test_pool_diagnostics.py:27"
+
+
+def test_pool_diag_tick_notable_only_when_anomaly() -> None:
+    base = dict(
+        under_pressure=False,
+        idle_in_tx=0,
+        slow_sessions=0,
+        remote_skipped_pressure=0,
+        remote_skipped_busy=0,
+        mirror_skip=0,
+        learn_pool_wait=0,
+    )
+    assert pool_diagnostics.pool_diag_tick_notable(**base) is False
+    assert pool_diagnostics.pool_diag_tick_notable(**{**base, "under_pressure": True}) is True
+    assert pool_diagnostics.pool_diag_tick_notable(**{**base, "remote_skipped_busy": 3}) is True
+
+
+def test_pool_diag_tick_sec_default(monkeypatch) -> None:
+    monkeypatch.delenv("PG_POOL_DIAG_TICK_SEC", raising=False)
+    monkeypatch.setattr(pool_diagnostics, "repo_env_raw_value", lambda _k: None)
+    assert pool_diagnostics.pool_diag_tick_sec() == 300.0