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>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed 2026-06-08 18:01:55 +08:00
1 parent 50d70a5050
commit 40206dc94d
3 files changed
+66 -7

No files matched your search

@@ -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) 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: 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() 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}") 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}") await _try_set_session_value(session, f"SET LOCAL statement_timeout = {_LEADERBOARD_QUERY_TIMEOUT_MS}")
@@ -28,6 +28,42 @@ from deerflow.runtime.store.provider import POSTGRES_CONN_REQUIRED, POSTGRES_STO
logger = logging.getLogger(__name__) 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 # Internal backend factory
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -71,9 +107,9 @@ async def _async_store(config) -> AsyncIterator[BaseStore]:
if not config.connection_string: if not config.connection_string:
raise ValueError(POSTGRES_CONN_REQUIRED) 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() await store.setup()
logger.info("Store: using AsyncPostgresStore") logger.info("Store: using AsyncPostgresStore (pooled, self-healing)")
yield store yield store
return return
+1 -1
View File
@@ -200,7 +200,7 @@ checkpointer:
type: postgres type: postgres
# true: 新 checkpoint 写入 Postgres;false: 保留 PG 配置但继续使用 SQLite。 # true: 新 checkpoint 写入 Postgres;false: 保留 PG 配置但继续使用 SQLite。
# 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。 # 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。
use_postgres: false use_postgres: true
connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow
sqlite_connection_string: backend/.deer-flow/data/deerflow.db sqlite_connection_string: backend/.deer-flow/data/deerflow.db
dual_read_enabled: true dual_read_enabled: true