fix(persistence): wait out a peer's Postgres bootstrap past command_timeout (#6306)

* fix(persistence): wait out a peer's Postgres bootstrap past command_timeout

The bootstrap advisory lock was taken with a blocking pg_advisory_lock on
the app engine, whose asyncpg command_timeout (30s by default) also bounds
that statement. A second Gateway instance therefore failed startup with
TimeoutError whenever another instance's migration outlasted the timeout.

Poll the non-blocking pg_try_advisory_lock instead: each attempt returns
immediately and stays bounded by command_timeout, while the wait lasts as
long as the holder's migration. The wait is logged once.

* docs(changelog): reference #6306 in the bootstrap lock entry

* docs(persistence): describe polled bootstrap lock in the migrations guide

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -466,6 +466,15 @@ This release closes that milestone with **301 merged pull requests**.
 
 ### Fixed
 
+- **persistence:** A second Gateway instance no longer fails startup with
+  `TimeoutError` while another instance runs a PostgreSQL schema migration. The
+  bootstrap advisory lock was taken with a blocking `pg_advisory_lock` on the
+  app engine, whose asyncpg `database.command_timeout` (30s by default) also
+  applies to that statement, so any migration longer than the timeout aborted
+  the waiting instance. Acquisition now polls the non-blocking
+  `pg_try_advisory_lock`: the wait lasts as long as the holder's migration,
+  each attempt stays bounded by `command_timeout`, and the wait is logged once.
+  ([#6306])
 - **projects:** Reading a shelf document for the first time no longer blocks
   every other database write on SQLite while the document converts. Lazy
   conversion ran pymupdf/markitdown inside the `BEGIN IMMEDIATE` transaction
@@ -7746,3 +7755,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6263]: https://github.com/bytedance/deer-flow/pull/6263
 [#6282]: https://github.com/bytedance/deer-flow/pull/6282
 [#6305]: https://github.com/bytedance/deer-flow/pull/6305
+[#6306]: https://github.com/bytedance/deer-flow/pull/6306

CHANGELOG_zh.md

@@ -423,6 +423,12 @@
 
 ### 修复
 
+- **持久化：** 另一个实例正在执行 PostgreSQL 模式迁移时，第二个 Gateway 实例不再
+  因 `TimeoutError` 启动失败。引导期 advisory lock 此前在应用引擎上以阻塞的
+  `pg_advisory_lock` 获取，而该引擎的 asyncpg `database.command_timeout`（默认
+  30 秒）同样作用于这条语句，因此任何超过该时限的迁移都会让等待中的实例中止。
+  现在获取改为轮询非阻塞的 `pg_try_advisory_lock`：等待时长与持锁方的迁移一致，
+  每次尝试仍受 `command_timeout` 约束，等待只记录一次日志。([#6306])
 - **项目：** 在 SQLite 上首次读取书架文档时，文档转换期间不再阻塞所有其他数据库
   写入。此前懒转换在 `BEGIN IMMEDIATE` 事务内运行 pymupdf/markitdown，以便与移入
   回收站和彻底删除串行化发布；而 SQLite 的这把锁作用于整个数据库，运行状态、线程
@@ -6463,3 +6469,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6263]: https://github.com/bytedance/deer-flow/pull/6263
 [#6282]: https://github.com/bytedance/deer-flow/pull/6282
 [#6305]: https://github.com/bytedance/deer-flow/pull/6305
+[#6306]: https://github.com/bytedance/deer-flow/pull/6306

backend/packages/harness/deerflow/persistence/AGENTS.md

@@ -2,6 +2,8 @@
 
 Postgres bootstrap owns its session-scoped advisory lock until `pg_advisory_unlock` completes. Drain that unlock across host cancellation before leaving the SQLAlchemy connection context; repeated cancellation must not return a pooled session while it still holds the bootstrap mutex. Ordinary database errors remain best-effort and are logged.
 
+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.
+
 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.

backend/packages/harness/deerflow/persistence/bootstrap.py

@@ -50,10 +50,12 @@ Layered, with different guarantees per backend. Postgres has true
 cross-process serialisation. SQLite is single-process safe and cross-process
 best-effort; multi-instance deployments should use Postgres.
 
-* **Postgres -- true cross-process serialisation.** ``pg_advisory_lock`` runs
-  the whole reflect-and-act sequence under an exclusive lock that survives
-  cross-process. Concurrent Gateway instances queue cleanly and the second
-  one observes head as a no-op.
+* **Postgres -- true cross-process serialisation.** A session-level advisory
+  lock runs the whole reflect-and-act sequence under an exclusive lock that
+  survives cross-process. Concurrent Gateway instances wait for it however
+  long the holder's migration takes (polling, so the engine's
+  ``command_timeout`` never cuts the wait short) and the second one observes
+  head as a no-op.
 
 * **SQLite -- single-process serialisation, best-effort cross-process.**
   SQLite is single-node by deployment, so the realistic concurrency case is
@@ -97,7 +99,7 @@ from alembic.script import ScriptDirectory
 from alembic.util.exc import CommandError
 from sqlalchemy import inspect as sa_inspect
 from sqlalchemy import text
-from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 from deerflow.utils.file_io import await_drained
 
@@ -184,6 +186,11 @@ _BASELINE_REVISION = "0001_baseline"
 # releases the prior lock).
 _PG_LOCK_KEY = 0x0DEE_12F1_0BEE_3682
 
+# Delay between ``pg_try_advisory_lock`` attempts while another instance holds
+# the bootstrap lock. Waiting is a loop of short statements, so the engine's
+# ``command_timeout`` bounds each attempt instead of the whole wait.
+_PG_LOCK_POLL_INTERVAL_SECONDS = 1.0
+
 
 # Tables created by ``0001_baseline.upgrade()``. The legacy branch restricts
 # its ``create_all`` backfill to this set so it does NOT pre-empt later
@@ -521,6 +528,26 @@ def _upgrade(cfg: AlembicConfig, revision: str) -> None:
 # ---------------------------------------------------------------------------
 
 
+async def _acquire_postgres_lock(conn: AsyncConnection) -> None:
+    """Take the session-level bootstrap advisory lock on *conn*, waiting as
+    long as another instance holds it.
+
+    Polls ``pg_try_advisory_lock`` instead of blocking in
+    ``pg_advisory_lock``. The app engine sets asyncpg's ``command_timeout``
+    (30s by default), which applies to every statement without an explicit
+    timeout -- a blocking acquire would raise ``TimeoutError`` and fail
+    startup whenever a peer's migration outlasts it. Each poll returns
+    immediately, so ``command_timeout`` still bounds a stalled round trip
+    while the wait itself stays unbounded, like the holder's migration.
+    """
+    logged_wait = False
+    while not (await conn.execute(text("SELECT pg_try_advisory_lock(:k)"), {"k": _PG_LOCK_KEY})).scalar_one():
+        if not logged_wait:
+            logger.info("bootstrap: postgres advisory lock key=0x%x is held by another instance; waiting", _PG_LOCK_KEY)
+            logged_wait = True
+        await asyncio.sleep(_PG_LOCK_POLL_INTERVAL_SECONDS)
+
+
 @asynccontextmanager
 async def _postgres_lock(engine: AsyncEngine):
     """Hold a Postgres session-level advisory lock for the body of the block.
@@ -547,12 +574,14 @@ async def _postgres_lock(engine: AsyncEngine):
     the kill **for this transaction only** (no global / role-level effect).
     Self-hosted Postgres usually ships with the timeout off, so this is a
     no-op there; on managed PG it is what keeps the lock alive while DDL
-    runs. Must execute *before* ``pg_advisory_lock`` so a slow lock acquire
-    on a heavily-contended cluster is itself protected.
+    runs. Must execute *before* the lock is acquired so a slow acquire on a
+    heavily-contended cluster is itself protected.
+
+    Acquisition polls rather than blocks -- see ``_acquire_postgres_lock``.
     """
     async with engine.connect() as conn:
         await conn.execute(text("SET LOCAL idle_in_transaction_session_timeout = 0"))
-        await conn.execute(text("SELECT pg_advisory_lock(:k)"), {"k": _PG_LOCK_KEY})
+        await _acquire_postgres_lock(conn)
         try:
             logger.info("bootstrap: acquired postgres advisory lock key=0x%x", _PG_LOCK_KEY)
             yield

backend/packages/harness/deerflow/persistence/migrations/AGENTS.md

@@ -86,7 +86,7 @@ the extra nullable columns and their data remain intact. A regression exercises
 that procedure from the original schema and verifies repository reads/inserts
 and preservation of incarnation data.
 
-**Concurrency safety**: Postgres uses `pg_advisory_lock` to serialise concurrent Gateway instances. SQLite uses a per-engine `asyncio.Lock` for same-process startup and is best-effort across processes via SQLite's file-level write lock + `PRAGMA busy_timeout`; multi-instance deployments should use Postgres. Column revisions in `versions/` additionally use idempotent helpers (`_helpers.py::safe_add_column`, `safe_drop_column`) so repeated post-baseline changes and retries are no-ops when the change is already present.
+**Concurrency safety**: Postgres serialises concurrent Gateway instances with a session-level advisory lock, acquired by polling `pg_try_advisory_lock` so the app engine's `command_timeout` never cuts the wait short. SQLite uses a per-engine `asyncio.Lock` for same-process startup and is best-effort across processes via SQLite's file-level write lock + `PRAGMA busy_timeout`; multi-instance deployments should use Postgres. Column revisions in `versions/` additionally use idempotent helpers (`_helpers.py::safe_add_column`, `safe_drop_column`) so repeated post-baseline changes and retries are no-ops when the change is already present.
 
 **Authoring a new revision**:
 ```bash

backend/tests/test_persistence_bootstrap_pg_lock.py

@@ -13,37 +13,63 @@ transaction.
 This test pins:
 
 1. The ``SET LOCAL`` is emitted at all (no silent regression).
-2. It runs **before** ``pg_advisory_lock`` -- otherwise a slow lock acquire
+2. It runs **before** the lock is acquired -- otherwise a slow lock acquire
    on a heavily-contended cluster would itself be vulnerable.
 3. The ``pg_advisory_unlock`` still fires on the way out (the new SQL must
    not break the release path).
 
+It also pins that waiting for a peer's migration is not cut short by the app
+engine's asyncpg ``command_timeout``: acquisition polls the non-blocking
+``pg_try_advisory_lock`` so no single statement outlives that deadline.
+
 We mock the engine instead of standing up a real Postgres because the only
 behaviour worth pinning here is the SQL execution order; the timeout's
-runtime effect is Postgres's contract, not ours.
+runtime effect is Postgres's contract, not ours. The one exception drives
+Gateway startup (``init_engine_from_config``) against a live server and is opt-in via
+``DEERFLOW_TEST_POSTGRES_URL``.
 """
 
 from __future__ import annotations
 
 import asyncio
+import logging
+import os
+import uuid
 
 import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.schema import DropSchema
 
+from deerflow.config.database_config import DatabaseConfig
 from deerflow.persistence import bootstrap as bootstrap_mod
+from deerflow.persistence.engine import close_engine, get_engine, init_engine_from_config
+
+POSTGRES_URL = os.getenv("DEERFLOW_TEST_POSTGRES_URL")
+
+
+class _FakeResult:
+    def __init__(self, value: object = None) -> None:
+        self._value = value
+
+    def scalar_one(self) -> object:
+        return self._value
 
 
 class _FakeAsyncConn:
     """Async-context-manager stand-in for SQLAlchemy's ``AsyncConnection``.
 
     Records every ``execute(stmt, params)`` so the test can assert SQL order.
+    The advisory lock is always free, so ``pg_try_advisory_lock`` succeeds.
     """
 
     def __init__(self) -> None:
         self.executed: list[tuple[str, dict | None]] = []
 
     async def execute(self, stmt, params=None):
-        self.executed.append((str(stmt), params))
-        return None
+        sql = str(stmt)
+        self.executed.append((sql, params))
+        return _FakeResult(True if "pg_try_advisory_lock" in sql else None)
 
     async def __aenter__(self) -> _FakeAsyncConn:
         return self
@@ -77,10 +103,10 @@ async def test_postgres_lock_disables_idle_in_transaction_kill_before_locking()
     assert set_local_idx is not None, f"SET LOCAL never executed; saw: {sqls}"
     assert "0" in sqls[set_local_idx], f"SET LOCAL did not target value 0: {sqls[set_local_idx]!r}"
 
-    # 2. SET LOCAL precedes pg_advisory_lock.
-    lock_idx = next((i for i, s in enumerate(sqls) if "pg_advisory_lock" in s), None)
-    assert lock_idx is not None, f"pg_advisory_lock never executed; saw: {sqls}"
-    assert set_local_idx < lock_idx, f"SET LOCAL must run before pg_advisory_lock; got order {sqls}"
+    # 2. SET LOCAL precedes the lock acquire.
+    lock_idx = next((i for i, s in enumerate(sqls) if "pg_try_advisory_lock" in s), None)
+    assert lock_idx is not None, f"pg_try_advisory_lock never executed; saw: {sqls}"
+    assert set_local_idx < lock_idx, f"SET LOCAL must run before pg_try_advisory_lock; got order {sqls}"
 
     # 3. pg_advisory_unlock still fires on exit.
     assert any("pg_advisory_unlock" in s for s in sqls), f"pg_advisory_unlock missing; saw: {sqls}"
@@ -114,7 +140,7 @@ class _BlockingUnlockConn(_FakeAsyncConn):
             self.unlock_started.set()
             await self.allow_unlock.wait()
             self.unlock_finished.set()
-        return None
+        return _FakeResult(True if "pg_try_advisory_lock" in sql else None)
 
 
 class _BlockingUnlockEngine:
@@ -158,3 +184,101 @@ async def test_postgres_lock_drains_unlock_across_repeated_cancellation() -> Non
         if not task.done():
             task.cancel()
             await asyncio.gather(task, return_exceptions=True)
+
+
+class _ContendedLockConn(_FakeAsyncConn):
+    """A peer instance holds the bootstrap lock until ``peer_done`` is set.
+
+    Every statement runs under ``command_timeout``, the way asyncpg applies the
+    connection default to statements that pass no explicit timeout, so a
+    blocking ``pg_advisory_lock`` that outlives it raises ``TimeoutError``.
+    """
+
+    def __init__(self, command_timeout: float) -> None:
+        super().__init__()
+        self.command_timeout = command_timeout
+        self.peer_done = asyncio.Event()
+
+    async def execute(self, stmt, params=None):
+        return await asyncio.wait_for(self._execute(str(stmt), params), timeout=self.command_timeout)
+
+    async def _execute(self, sql: str, params: dict | None) -> _FakeResult:
+        self.executed.append((sql, params))
+        if "pg_try_advisory_lock" in sql:
+            return _FakeResult(self.peer_done.is_set())
+        if "pg_advisory_lock" in sql:
+            await self.peer_done.wait()
+        return _FakeResult()
+
+
+class _ContendedLockEngine:
+    def __init__(self, command_timeout: float) -> None:
+        self.conn = _ContendedLockConn(command_timeout)
+
+    def connect(self) -> _ContendedLockConn:
+        return self.conn
+
+
+@pytest.mark.asyncio
+async def test_postgres_lock_outwaits_command_timeout_while_a_peer_migrates(monkeypatch, caplog) -> None:
+    monkeypatch.setattr(bootstrap_mod, "_PG_LOCK_POLL_INTERVAL_SECONDS", 0.01)
+    engine = _ContendedLockEngine(command_timeout=0.05)
+    acquired = asyncio.Event()
+
+    async def bootstrap() -> None:
+        async with bootstrap_mod._postgres_lock(engine):  # type: ignore[arg-type]
+            acquired.set()
+
+    with caplog.at_level(logging.INFO, logger=bootstrap_mod.__name__):
+        task = asyncio.create_task(bootstrap())
+        try:
+            # The peer's migration outlasts command_timeout several times over.
+            await asyncio.sleep(0.25)
+            assert not task.done(), f"bootstrap stopped waiting while the peer held the lock: {task.exception()!r}"
+
+            engine.conn.peer_done.set()
+            await asyncio.wait_for(task, timeout=1)
+        finally:
+            if not task.done():
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+
+    assert acquired.is_set()
+    sqls = [stmt for stmt, _ in engine.conn.executed]
+    assert sum("pg_try_advisory_lock" in s for s in sqls) > 2, f"expected repeated polls; saw: {sqls}"
+    assert any("pg_advisory_unlock" in s for s in sqls), f"unlock missing; saw: {sqls}"
+    waiting_logs = [r for r in caplog.records if "held by another instance" in r.getMessage()]
+    assert len(waiting_logs) == 1, "the wait should be logged once, not on every poll"
+
+
+@pytest.mark.skipif(not POSTGRES_URL, reason="set DEERFLOW_TEST_POSTGRES_URL to run the live PostgreSQL bootstrap-lock test")
+@pytest.mark.asyncio
+async def test_init_engine_outwaits_command_timeout_on_live_postgres(monkeypatch) -> None:
+    """Gateway startup waits out a peer's migration that outlasts ``command_timeout``."""
+    monkeypatch.setattr(bootstrap_mod, "_PG_LOCK_POLL_INTERVAL_SECONDS", 0.1)
+    schema = f"deerflow_test_{uuid.uuid4().hex[:12]}"
+    db_config = DatabaseConfig(backend="postgres", postgres_url=POSTGRES_URL or "", postgres_schema=schema, command_timeout=1)
+    peer = create_async_engine(db_config.app_sqlalchemy_url)
+    try:
+        async with peer.connect() as peer_conn:
+            await peer_conn.execute(text("SELECT pg_advisory_lock(:k)"), {"k": bootstrap_mod._PG_LOCK_KEY})
+            task = asyncio.create_task(init_engine_from_config(db_config))
+            try:
+                # The peer's migration outlasts command_timeout more than twice over.
+                await asyncio.sleep(2.5)
+                assert not task.done(), f"startup stopped waiting while the peer held the lock: {task.exception()!r}"
+            finally:
+                await peer_conn.execute(text("SELECT pg_advisory_unlock(:k)"), {"k": bootstrap_mod._PG_LOCK_KEY})
+                await asyncio.wait_for(asyncio.gather(task, return_exceptions=True), timeout=60)
+            await task
+
+        engine = get_engine()
+        assert engine is not None
+        async with engine.connect() as conn:
+            revision = (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalar_one()
+        assert revision
+    finally:
+        await close_engine()
+        async with peer.begin() as conn:
+            await conn.execute(DropSchema(schema, cascade=True, if_exists=True))
+        await peer.dispose()