fix: Postgres store 自愈连接池 + 启用 PG checkpointer + 分析查询超时按服务端适配

- async_provider: AsyncPostgresStore 改用连接池(check_connection 校验)，
  避免单条长连接在服务重启/空闲超时后永久失效；池大小可经
  DEER_FLOW_PG_POOL_MIN/MAX 环境变量调整
- admin_users: 排行榜分析查询的每语句超时按服务端区分
  (MySQL>=5.7.4 max_execution_time / MariaDB max_statement_time /
  MySQL<=5.6 跳过 / PostgreSQL statement_timeout)
- config.yaml: checkpointer.use_postgres 置为 true

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

offline-backend-20260512/backend/app/gateway/routers/admin_users.py

@@ -458,13 +458,36 @@ async def _try_set_session_value(session, sql: str) -> None:
             logger.debug("Failed to rollback after analytics session setting failure", exc_info=True)
 
 
+def _is_mariadb(dialect) -> bool:
+    """SQLAlchemy reports ``dialect.name == 'mysql'`` for MariaDB too."""
+    if getattr(dialect, "_is_mariadb", False) or getattr(dialect, "is_mariadb", False):
+        return True
+    version_info = getattr(dialect, "server_version_info", None) or ()
+    return any("mariadb" in str(part).lower() for part in version_info)
+
+
 async def _set_analytics_read_timeout(session) -> None:
+    """Best-effort per-statement read timeout for the analytics queries.
+
+    The right knob differs by server, and some servers have none at all:
+      - MySQL >= 5.7.4: ``max_execution_time`` (milliseconds)
+      - MariaDB:        ``max_statement_time`` (seconds)
+      - MySQL <= 5.6:   no per-statement timeout variable — skip it entirely
+                        (5.6 raises "Unknown system variable" otherwise)
+      - PostgreSQL:     ``statement_timeout`` (milliseconds)
+    """
     conn = await session.connection()
-    dialect = conn.dialect.name
+    dialect = conn.dialect
-    if dialect == "mysql":
+    name = dialect.name
-        await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}")
+    if name == "mysql":
+        if _is_mariadb(dialect):
+            # MariaDB: max_statement_time is in SECONDS (fractional ok).
+            await _try_set_session_value(session, f"SET SESSION max_statement_time = {_LEADERBOARD_QUERY_TIMEOUT_MS / 1000}")
+        elif (getattr(dialect, "server_version_info", None) or ()) >= (5, 7, 4):
+            await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}")
+        # else: MySQL <= 5.6 has no statement-timeout variable — skip silently.
         await _try_set_session_value(session, f"SET SESSION innodb_lock_wait_timeout = {_LEADERBOARD_LOCK_WAIT_SECONDS}")
-    elif dialect == "postgresql":
+    elif name == "postgresql":
         await _try_set_session_value(session, f"SET LOCAL statement_timeout = {_LEADERBOARD_QUERY_TIMEOUT_MS}")
 
 

offline-backend-20260512/backend/packages/harness/deerflow/runtime/store/async_provider.py

@@ -28,6 +28,42 @@ from deerflow.runtime.store.provider import POSTGRES_CONN_REQUIRED, POSTGRES_STO
 
 logger = logging.getLogger(__name__)
 
+
+def _pool_size_env(name: str, default: int) -> int:
+    import os
+
+    raw = os.getenv(name, "").strip()
+    if not raw:
+        return default
+    try:
+        return max(0, int(raw))
+    except ValueError:
+        logger.warning("Invalid %s=%r; using default %d", name, raw, default)
+        return default
+
+
+def _postgres_pool_config() -> dict:
+    """Pool config for the Postgres store.
+
+    A single long-lived connection (the default ``from_conn_string`` behavior)
+    dies permanently on a server restart / idle timeout / network blip — every
+    later store call then fails with "server closed the connection
+    unexpectedly". A pool with ``check=check_connection`` validates each
+    connection on checkout and replaces dead ones, so the runtime self-heals.
+    Extra keys (``check``/``max_idle``/``max_lifetime``) are passed straight
+    through to ``AsyncConnectionPool`` by ``from_conn_string``.
+    """
+    from psycopg_pool import AsyncConnectionPool
+
+    return {
+        "min_size": _pool_size_env("DEER_FLOW_PG_POOL_MIN", 1),
+        "max_size": _pool_size_env("DEER_FLOW_PG_POOL_MAX", 10),
+        "check": AsyncConnectionPool.check_connection,
+        "max_idle": 300.0,
+        "max_lifetime": 3600.0,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Internal backend factory
 # ---------------------------------------------------------------------------
@@ -71,9 +107,9 @@ async def _async_store(config) -> AsyncIterator[BaseStore]:
         if not config.connection_string:
             raise ValueError(POSTGRES_CONN_REQUIRED)
 
-        async with AsyncPostgresStore.from_conn_string(config.connection_string) as store:
+        async with AsyncPostgresStore.from_conn_string(config.connection_string, pool_config=_postgres_pool_config()) as store:
             await store.setup()
-            logger.info("Store: using AsyncPostgresStore")
+            logger.info("Store: using AsyncPostgresStore (pooled, self-healing)")
             yield store
         return
 

offline-backend-20260512/config.yaml

@@ -200,7 +200,7 @@ checkpointer:
   type: postgres
   # true: 新 checkpoint 写入 Postgres；false: 保留 PG 配置但继续使用 SQLite。
   # 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。
-  use_postgres: false
+  use_postgres: true
   connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow
   sqlite_connection_string: backend/.deer-flow/data/deerflow.db
   dual_read_enabled: true