From 7f2c0fbe0c09a74d4f2dad04768951d5cb845afd Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Mon, 5 Oct 2026 16:33:29 +0900 Subject: [PATCH] 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 --- CHANGELOG.md | 10 ++ CHANGELOG_zh.md | 7 + .../harness/deerflow/persistence/AGENTS.md | 2 + .../harness/deerflow/persistence/bootstrap.py | 45 +++++- .../deerflow/persistence/migrations/AGENTS.md | 2 +- .../test_persistence_bootstrap_pg_lock.py | 142 ++++++++++++++++-- 6 files changed, 190 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 71229300dd..e1328690a7 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index b9d42e5bc7..ec8b5bf691 100644 --- a/CHANGELOG_zh.md +++ b/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 diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index aeff416cf2..bb6170be3f 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/persistence/bootstrap.py b/backend/packages/harness/deerflow/persistence/bootstrap.py index df066a4708..a542140c87 100644 --- a/backend/packages/harness/deerflow/persistence/bootstrap.py +++ b/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 diff --git a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md index 60e1535186..dfddf267b4 100644 --- a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md +++ b/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 diff --git a/backend/tests/test_persistence_bootstrap_pg_lock.py b/backend/tests/test_persistence_bootstrap_pg_lock.py index c4087fbf6c..3e3ca109e3 100644 --- a/backend/tests/test_persistence_bootstrap_pg_lock.py +++ b/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()