mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
9a92d2e9b7
Two Gateways booting the same missing database both reach CREATE DATABASE and only one wins; the loser aborted startup with duplicate_database (42P04) even though the winner's catalog row was already committed and the target was usable. Follow the SQLSTATE, like the missing-database check beside it does, so the loser falls through to its retry. Every other failure -- permission denied included -- still propagates untouched.
118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
"""Two Gateways booting the same missing PostgreSQL database: only one can create it.
|
|
|
|
``_auto_create_postgres_db`` runs when bootstrap fails with SQLSTATE 3D000. With two
|
|
processes starting against the same server, the loser's ``CREATE DATABASE`` is refused with
|
|
SQLSTATE 42P04 *after* the winner has committed the catalog row, so the target exists and is
|
|
usable — the boot must continue to the retry rather than fail on a race it effectively won.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from deerflow.persistence import engine as engine_mod
|
|
|
|
_TARGET_URL = "postgresql+asyncpg://deerflow:secret@127.0.0.1:5432/deerflow_target"
|
|
|
|
|
|
class _SqlstateError(Exception):
|
|
"""Stands in for the asyncpg exception, which the driver attaches ``sqlstate`` to."""
|
|
|
|
def __init__(self, sqlstate: str | None, message: str = "server said no") -> None:
|
|
super().__init__(message)
|
|
if sqlstate is not None:
|
|
self.sqlstate = sqlstate
|
|
|
|
|
|
class _FailingMaintenanceEngine:
|
|
"""Fake throwaway ``postgres``-database engine whose CREATE DATABASE raises *error*."""
|
|
|
|
def __init__(self, error: BaseException) -> None:
|
|
self.error = error
|
|
self.statements: list[str] = []
|
|
self.disposed = False
|
|
|
|
def connect(self) -> _FailingMaintenanceEngine:
|
|
return self
|
|
|
|
async def __aenter__(self) -> _FailingMaintenanceEngine:
|
|
return self
|
|
|
|
async def __aexit__(self, *exc_info: object) -> None:
|
|
return None
|
|
|
|
async def execute(self, statement: object) -> None:
|
|
self.statements.append(str(statement))
|
|
raise self.error
|
|
|
|
async def dispose(self) -> None:
|
|
self.disposed = True
|
|
|
|
|
|
@pytest.fixture
|
|
def maintenance(monkeypatch: pytest.MonkeyPatch):
|
|
installed: list[_FailingMaintenanceEngine] = []
|
|
|
|
def _install(error: BaseException) -> _FailingMaintenanceEngine:
|
|
engine = _FailingMaintenanceEngine(error)
|
|
installed.append(engine)
|
|
monkeypatch.setattr(engine_mod, "create_async_engine", lambda *args, **kwargs: engine)
|
|
return engine
|
|
|
|
_install.installed = installed # type: ignore[attr-defined]
|
|
return _install
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_database_from_a_peer_continues_the_boot(maintenance, caplog) -> None:
|
|
engine = maintenance(_SqlstateError("42P04", 'database "deerflow_target" already exists'))
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
await engine_mod._auto_create_postgres_db(_TARGET_URL)
|
|
|
|
assert 'CREATE DATABASE "deerflow_target"' in engine.statements[0]
|
|
assert "was created by a peer process" in caplog.text
|
|
assert engine.disposed, "the throwaway maintenance engine must still be disposed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wrapped_duplicate_database_is_recognised(maintenance) -> None:
|
|
"""SQLSTATE lives on the driver exception, which reaches us wrapped in ``orig``."""
|
|
wrapper = RuntimeError("bootstrap failed")
|
|
wrapper.__cause__ = _SqlstateError("42P04", 'database "deerflow_target" already exists')
|
|
engine = maintenance(wrapper)
|
|
|
|
await engine_mod._auto_create_postgres_db(_TARGET_URL)
|
|
|
|
assert engine.disposed
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_permission_denied_still_propagates(maintenance) -> None:
|
|
"""The other CREATE DATABASE failure mode is a real error and must not be swallowed."""
|
|
engine = maintenance(_SqlstateError("42501", "permission denied to create database"))
|
|
|
|
with pytest.raises(_SqlstateError):
|
|
await engine_mod._auto_create_postgres_db(_TARGET_URL)
|
|
|
|
assert engine.disposed
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unrelated_failure_still_propagates(maintenance) -> None:
|
|
engine = maintenance(OSError("connection refused"))
|
|
|
|
with pytest.raises(OSError):
|
|
await engine_mod._auto_create_postgres_db(_TARGET_URL)
|
|
|
|
assert engine.disposed
|
|
|
|
|
|
def test_missing_and_duplicate_codes_are_distinct() -> None:
|
|
"""Guards against collapsing the two SQLSTATEs into one constant."""
|
|
assert engine_mod._MISSING_DATABASE_SQLSTATE != engine_mod._DUPLICATE_DATABASE_SQLSTATE
|
|
assert not engine_mod._is_missing_database_error(_SqlstateError("42P04"))
|
|
assert not engine_mod._is_duplicate_database_error(_SqlstateError("3D000"))
|