diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a91259a77..d2864fce79 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1397,6 +1397,14 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**. ### Fixed +- **persistence:** A PostgreSQL bootstrap failure inside an existing database + now reports its own error. The Gateway treated any error mentioning "does not + exist", such as a missing column or relation during a migration, as a missing + database and ran `CREATE DATABASE`. Startup then failed with `database ... + already exists` or `permission denied to create database`, and the original + error was missing from the traceback. Only a missing database (SQLSTATE + `3D000`) is auto-created now. ([#7410]) + - **mcp:** Ordinary users can now install DingTalk, WeCom and HubSpot from **Capability Center → Plugins**. Saving one of these bundled connections as a non-administrator, including editing or enabling it later, @@ -13060,3 +13068,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7362]: https://github.com/bytedance/deer-flow/pull/7362 [#7364]: https://github.com/bytedance/deer-flow/pull/7364 [#7400]: https://github.com/bytedance/deer-flow/pull/7400 +[#7410]: https://github.com/bytedance/deer-flow/pull/7410 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index b91b51eaa2..aeb750e6ae 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -1112,6 +1112,12 @@ ### 修复 +- **持久化：** 在已存在的数据库中 PostgreSQL 启动初始化失败时，现在会报告原始错误。 + 此前 Gateway 会把任何包含 "does not exist" 的错误（例如迁移时缺少列或表）当作数据库 + 不存在并执行 `CREATE DATABASE`，启动随之以 `database ... already exists` 或 + `permission denied to create database` 失败，原始错误也不会出现在 traceback 中。 + 现在只有数据库确实不存在（SQLSTATE `3D000`）时才会自动创建。([#7410]) + - **MCP：** 普通用户现在可以在“能力中心 → 插件”中安装钉钉、企业微信和 HubSpot。 此前非管理员保存这些内置连接（包括之后编辑或启用）会返回 `500 Internal Server Error`，因为个人连接检查读取了请求模型未声明的 `cwd` 字段。 @@ -10425,3 +10431,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7362]: https://github.com/bytedance/deer-flow/pull/7362 [#7364]: https://github.com/bytedance/deer-flow/pull/7364 [#7400]: https://github.com/bytedance/deer-flow/pull/7400 +[#7410]: https://github.com/bytedance/deer-flow/pull/7410 diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index e52f5ea18f..e83b230eca 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -4,6 +4,8 @@ Postgres bootstrap owns its session-scoped advisory lock until `pg_advisory_unlo Acquire that lock by polling `pg_try_advisory_lock`, never with a blocking `pg_advisory_lock`: the app engine's asyncpg `command_timeout` bounds every statement, so a blocking acquire fails a second instance's startup whenever a peer's migration outlasts it. +`init_engine` auto-creates a Postgres database only on SQLSTATE `3D000`. Missing relations, columns and roles also "do not exist"; a `CREATE DATABASE` retry would mask them. + When `database.postgres_schema` is configured, both async ORM connections and the synchronous SQLAlchemy connections used by DB-backed custom agents and managed subagents must use the same `search_path`; preserve this invariant when adding another persistence entry point. Alembic stamp/upgrade workers started inside `bootstrap_schema()` remain owned by the bootstrap critical section until the worker finishes. Drain those `asyncio.to_thread()` calls across host cancellation before releasing the in-process SQLite bootstrap lock or PostgreSQL advisory lock; otherwise another bootstrap can overlap a still-running migration worker. diff --git a/backend/packages/harness/deerflow/persistence/engine.py b/backend/packages/harness/deerflow/persistence/engine.py index 531ff70c0f..dd78051505 100644 --- a/backend/packages/harness/deerflow/persistence/engine.py +++ b/backend/packages/harness/deerflow/persistence/engine.py @@ -56,6 +56,34 @@ logger = logging.getLogger(__name__) _engine: AsyncEngine | None = None _session_factory: async_sessionmaker[AsyncSession] | None = None +# SQLSTATE invalid_catalog_name: the database named in the URL does not exist. +_MISSING_DATABASE_SQLSTATE = "3D000" + + +def _is_missing_database_error(exc: BaseException) -> bool: + """Return whether *exc* means the target PostgreSQL database is absent. + + Postgres words missing relations, columns and roles as "does not exist" + too, so the message cannot tell them apart; only the SQLSTATE can. + asyncpg raises this one unwrapped from connect(); follow ``orig``, + ``__cause__`` and ``__context__`` (as ``_is_unique_violation`` in + ``runtime/runs/manager.py`` does) so a wrapped copy is recognized as well. + """ + pending: list[BaseException] = [exc] + seen: set[int] = set() + while pending: + current = pending.pop() + if id(current) in seen: + continue + seen.add(id(current)) + if getattr(current, "sqlstate", None) == _MISSING_DATABASE_SQLSTATE: + return True + for attr in ("orig", "__cause__", "__context__"): + inner = getattr(current, attr, None) + if isinstance(inner, BaseException): + pending.append(inner) + return False + async def _auto_create_postgres_db(url: str) -> None: """Connect to the ``postgres`` maintenance DB and CREATE DATABASE. @@ -218,8 +246,11 @@ async def init_engine( await _ensure_postgres_schema() await bootstrap_schema(_engine, backend=backend, postgres_schema=postgres_schema) except Exception as exc: - if backend == "postgres" and "does not exist" in str(exc): + if backend == "postgres" and _is_missing_database_error(exc): # Database not yet created -- attempt to auto-create it, then retry. + # Any other bootstrap failure is re-raised untouched: a + # CREATE DATABASE attempt would replace it with "already exists" + # or "permission denied to create database". await _auto_create_postgres_db(url) # Rebuild engine against the now-existing database. The rebuilt # engine MUST keep the same connect_args so the retried bootstrap diff --git a/backend/tests/test_persistence_engine_auto_create_postgres.py b/backend/tests/test_persistence_engine_auto_create_postgres.py new file mode 100644 index 0000000000..d81c923bca --- /dev/null +++ b/backend/tests/test_persistence_engine_auto_create_postgres.py @@ -0,0 +1,94 @@ +"""Live PostgreSQL checks for init_engine's missing-database auto-create. + +Opt in with DEERFLOW_TEST_POSTGRES_URL or TEST_POSTGRES_URI (set by the +backend unit-test CI job). The role must be allowed to create databases. +""" + +from __future__ import annotations + +import os +import uuid +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +import pytest +import pytest_asyncio +from sqlalchemy import text +from sqlalchemy.engine.url import make_url +from sqlalchemy.ext.asyncio import create_async_engine + +from deerflow.persistence import engine as engine_mod + + +def _asyncpg_url(url: str | None) -> str | None: + if not url: + return url + parts = urlsplit(url) + scheme = "postgresql+asyncpg" if parts.scheme in {"postgres", "postgresql"} else parts.scheme + query = urlencode([(key, value) for key, value in parse_qsl(parts.query, keep_blank_values=True) if key not in {"sslmode", "channel_binding"}]) + return urlunsplit(parts._replace(scheme=scheme, query=query)) + + +POSTGRES_URL = _asyncpg_url(os.getenv("DEERFLOW_TEST_POSTGRES_URL") or os.getenv("TEST_POSTGRES_URI")) + +pytestmark = [ + pytest.mark.asyncio, + pytest.mark.skipif(not POSTGRES_URL, reason="set DEERFLOW_TEST_POSTGRES_URL or TEST_POSTGRES_URI to run live PostgreSQL tests"), +] + + +async def _execute_on_server(*statements: str, database: str | None = None) -> None: + url = make_url(POSTGRES_URL or "") + if database is not None: + url = url.set(database=database) + engine = create_async_engine(url, isolation_level="AUTOCOMMIT") + try: + async with engine.connect() as conn: + for statement in statements: + await conn.execute(text(statement)) + finally: + await engine.dispose() + + +@pytest_asyncio.fixture +async def missing_database(): + name = f"deerflow_autocreate_{uuid.uuid4().hex[:12]}" + url = make_url(POSTGRES_URL or "").set(database=name).render_as_string(hide_password=False) + try: + yield name, url + finally: + await engine_mod.close_engine() + await _execute_on_server(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)') + + +async def test_missing_database_is_created_and_bootstrapped(missing_database) -> None: + name, url = missing_database + + await engine_mod.init_engine("postgres", url=url) + + async with engine_mod.get_engine().connect() as conn: + assert (await conn.execute(text("SELECT current_database()"))).scalar_one() == name + assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalar_one() + + +async def test_bootstrap_failure_in_an_existing_database_is_not_masked(missing_database) -> None: + name, url = missing_database + await engine_mod.init_engine("postgres", url=url) + await engine_mod.close_engine() + # Leave an unguarded migration statement to fail: rewind to 0012 and drop + # the column 0013's backfill UPDATE reads. + await _execute_on_server( + "UPDATE alembic_version SET version_num = '0012_mcp_task_results'", + "ALTER TABLE mcp_tasks DROP COLUMN updated_at CASCADE", + database=name, + ) + + with pytest.raises(Exception) as raised: + await engine_mod.init_engine("postgres", url=url) + + # Before the fix, this tried CREATE DATABASE and raised DuplicateDatabaseError. + sqlstates = [] + error: BaseException | None = raised.value + while error is not None: + sqlstates.append(getattr(error, "sqlstate", None)) + error = error.__cause__ + assert "42703" in sqlstates, raised.value diff --git a/backend/tests/test_persistence_engine_postgres_config.py b/backend/tests/test_persistence_engine_postgres_config.py index 570b1d3825..21fec97979 100644 --- a/backend/tests/test_persistence_engine_postgres_config.py +++ b/backend/tests/test_persistence_engine_postgres_config.py @@ -16,6 +16,14 @@ from deerflow.config.database_config import DatabaseConfig from deerflow.persistence import engine as engine_mod +class _PostgresError(Exception): + """Driver error carrying a SQLSTATE, as asyncpg's exceptions do.""" + + def __init__(self, sqlstate: str, message: str) -> None: + super().__init__(message) + self.sqlstate = sqlstate + + def test_postgres_engine_kwargs_include_connection_hardening() -> None: kwargs = engine_mod._postgres_engine_kwargs(echo=False, pool_size=5) @@ -216,7 +224,7 @@ async def test_init_engine_postgres_retry_uses_hardened_kwargs(monkeypatch) -> N initial_engine.dispose = AsyncMock() retry_engine = MagicMock() retry_engine.dispose = AsyncMock() - bootstrap_schema = AsyncMock(side_effect=[Exception("database does not exist"), None]) + bootstrap_schema = AsyncMock(side_effect=[_PostgresError("3D000", 'database "deerflow" does not exist'), None]) auto_create = AsyncMock() monkeypatch.setitem(sys.modules, "asyncpg", ModuleType("asyncpg")) @@ -241,6 +249,100 @@ async def test_init_engine_postgres_retry_uses_hardened_kwargs(monkeypatch) -> N await engine_mod.close_engine() +def _wrapped_by_orig(missing: BaseException) -> BaseException: + from sqlalchemy.exc import OperationalError + + return OperationalError("connect", {}, missing) + + +def _wrapped_by_cause(missing: BaseException) -> BaseException: + wrapped = RuntimeError("bootstrap failed") + wrapped.__cause__ = missing + return wrapped + + +def _wrapped_by_context(missing: BaseException) -> BaseException: + wrapped = RuntimeError("bootstrap failed") + wrapped.__context__ = missing + return wrapped + + +@pytest.mark.asyncio +@pytest.mark.parametrize("wrap", [_wrapped_by_orig, _wrapped_by_cause, _wrapped_by_context], ids=["orig", "cause", "context"]) +async def test_init_engine_postgres_retries_a_wrapped_missing_database_error(monkeypatch, wrap) -> None: + url = "postgresql+asyncpg://user:password@localhost/deerflow" + engine = MagicMock() + engine.dispose = AsyncMock() + wrapped = wrap(_PostgresError("3D000", 'database "deerflow" does not exist')) + bootstrap_schema = AsyncMock(side_effect=[wrapped, None]) + auto_create = AsyncMock() + + monkeypatch.setitem(sys.modules, "asyncpg", ModuleType("asyncpg")) + with ( + patch.object(engine_mod, "create_async_engine", return_value=engine), + patch.object(engine_mod, "async_sessionmaker", return_value=MagicMock()), + patch.object(engine_mod, "_auto_create_postgres_db", new=auto_create), + patch("deerflow.persistence.bootstrap.bootstrap_schema", new=bootstrap_schema), + ): + try: + await engine_mod.init_engine(backend="postgres", url=url) + + auto_create.assert_awaited_once_with(url) + assert bootstrap_schema.await_count == 2 + finally: + await engine_mod.close_engine() + + +def _cyclic_chain() -> BaseException: + # The classifier must terminate on a chain that loops back on itself. + error = _PostgresError("42703", 'column "updated_at" does not exist') + wrapper = RuntimeError("bootstrap failed") + wrapper.__cause__ = error + error.__context__ = wrapper + return error + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + pytest.param(_PostgresError("42703", 'column "updated_at" does not exist'), id="undefined-column"), + pytest.param(_PostgresError("42P01", 'relation "runs" does not exist'), id="undefined-table"), + pytest.param(_PostgresError("42883", "function gen_random_uuid() does not exist"), id="undefined-function"), + pytest.param(_PostgresError("28000", 'role "deerflow" does not exist'), id="undefined-role"), + pytest.param(Exception('database "deerflow" does not exist'), id="no-sqlstate"), + pytest.param(_cyclic_chain(), id="cyclic-chain"), + ], +) +async def test_init_engine_postgres_reraises_other_does_not_exist_errors(monkeypatch, error) -> None: + # Postgres words all of these "does not exist". Only a missing database + # may trigger CREATE DATABASE; anything else must surface unchanged + # instead of as "already exists" or "permission denied to create database". + url = "postgresql+asyncpg://user:password@localhost/deerflow" + engine = MagicMock() + engine.dispose = AsyncMock() + bootstrap_schema = AsyncMock(side_effect=error) + auto_create = AsyncMock() + + monkeypatch.setitem(sys.modules, "asyncpg", ModuleType("asyncpg")) + with ( + patch.object(engine_mod, "create_async_engine", return_value=engine) as create_engine, + patch.object(engine_mod, "async_sessionmaker", return_value=MagicMock()), + patch.object(engine_mod, "_auto_create_postgres_db", new=auto_create), + patch("deerflow.persistence.bootstrap.bootstrap_schema", new=bootstrap_schema), + ): + try: + with pytest.raises(type(error)) as raised: + await engine_mod.init_engine(backend="postgres", url=url) + + assert raised.value is error + auto_create.assert_not_awaited() + create_engine.assert_called_once() + bootstrap_schema.assert_awaited_once() + finally: + await engine_mod.close_engine() + + @pytest.mark.asyncio async def test_init_engine_sqlite_omits_postgres_kwargs_and_keeps_wal_listener(tmp_path) -> None: url = f"sqlite+aiosqlite:///{tmp_path / 'deerflow.db'}" diff --git a/backend/tests/test_persistence_retry_dispose_cancellation.py b/backend/tests/test_persistence_retry_dispose_cancellation.py index b47222b807..e423bba462 100644 --- a/backend/tests/test_persistence_retry_dispose_cancellation.py +++ b/backend/tests/test_persistence_retry_dispose_cancellation.py @@ -31,8 +31,11 @@ async def test_postgres_auto_create_retry_drains_provisional_engine_dispose_acro monkeypatch.setattr(engine_mod, "create_async_engine", lambda *args, **kwargs: engine) monkeypatch.setattr(engine_mod, "async_sessionmaker", lambda *args, **kwargs: object()) + class MissingDatabaseError(Exception): + sqlstate = "3D000" + async def fail_bootstrap(*args, **kwargs) -> None: - raise RuntimeError("database does not exist") + raise MissingDatabaseError('database "deerflow" does not exist') async def fake_auto_create(_url: str) -> None: return None