diff --git a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py index 0865fcdc..5b1e0e7c 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/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 - if dialect == "mysql": - await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}") + dialect = conn.dialect + name = dialect.name + 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}") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/store/async_provider.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/store/async_provider.py index 25156b01..94966e0d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/store/async_provider.py +++ b/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 diff --git a/offline-backend-20260512/config.yaml b/offline-backend-20260512/config.yaml index aa098e5c..751c0156 100644 --- a/offline-backend-20260512/config.yaml +++ b/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