From 8cde7f258ef6ef7839c9ffcfbd8835d54f936443 Mon Sep 17 00:00:00 2001 From: heart-scalpel Date: Mon, 6 Jul 2026 15:20:27 +0800 Subject: [PATCH] feat(gateway): refuse multi-worker startup on non-Postgres backends (#3960) Why: issue #3948 identifies four correctness breakers when GATEWAY_WORKERS > 1. Work item 1 adds a startup gate that refuses to boot when database.backend is not postgres, giving operators a clear error instead of silent SQLite write-lock corruption. The gate runs inside langgraph_runtime() before init_engine_from_config, so a misconfigured deploy never opens a listener or writes to disk. Non-integer env values fall back to 1 so uvicorn's own validation is unaffected. --- backend/app/gateway/deps.py | 28 ++++ .../tests/test_multi_worker_postgres_gate.py | 142 ++++++++++++++++++ 2 files changed, 170 insertions(+) create mode 100644 backend/tests/test_multi_worker_postgres_gate.py diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index c3d935696..93197f2ff 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -19,6 +19,7 @@ from __future__ import annotations import asyncio import logging +import os from collections.abc import AsyncGenerator, Callable from contextlib import AsyncExitStack, asynccontextmanager from typing import TYPE_CHECKING, TypeVar, cast @@ -45,6 +46,27 @@ logger = logging.getLogger(__name__) _RUN_DRAIN_TIMEOUT_SECONDS = 5.0 +def _enforce_postgres_for_multi_worker(config: AppConfig) -> None: + """Refuse to start when GATEWAY_WORKERS > 1 and the DB backend is not Postgres. + + SQLite write-locks cannot support concurrent multi-process access. + This gate runs once at startup before any persistence engine is + initialised so the error message is clear and the process exits + immediately. + """ + try: + workers = int(os.environ.get("GATEWAY_WORKERS", "1")) + except (TypeError, ValueError): + workers = 1 + + if workers <= 1: + return + + backend = getattr(config.database, "backend", None) + if backend != "postgres": + raise SystemExit(f"GATEWAY_WORKERS={workers} requires database.backend='postgres', but database.backend is '{backend}'. SQLite cannot support concurrent multi-process access. Set GATEWAY_WORKERS=1 or switch to Postgres.") + + async def _drain_inflight_runs(run_manager: RunManager) -> None: """Drain in-flight runs before the checkpointer is torn down (issue #3373). @@ -209,6 +231,12 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen from deerflow.runtime.checkpointer.async_provider import make_checkpointer from deerflow.runtime.events.store import make_run_event_store + # ------------------------------------------------------------------ + # Multi-worker safety gate: reject SQLite when GATEWAY_WORKERS > 1. + # SQLite write-locks cannot support concurrent multi-process access. + # ------------------------------------------------------------------ + _enforce_postgres_for_multi_worker(startup_config) + async with AsyncExitStack() as stack: config = startup_config diff --git a/backend/tests/test_multi_worker_postgres_gate.py b/backend/tests/test_multi_worker_postgres_gate.py new file mode 100644 index 000000000..f87e5e7b9 --- /dev/null +++ b/backend/tests/test_multi_worker_postgres_gate.py @@ -0,0 +1,142 @@ +"""Tests for the multi-worker Postgres startup gate. + +Pins the contract documented in ``docs/multi_worker.md`` work item 1 +(issue #3948): when ``GATEWAY_WORKERS > 1`` and the configured +database backend is not Postgres, the Gateway must refuse to start. +The gate runs inside :func:`langgraph_runtime` *before* any +persistence engine is initialised so operators see a clear error +instead of intermittent SQLite ``database is locked`` failures in +production. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import FastAPI + +from app.gateway.deps import _enforce_postgres_for_multi_worker, langgraph_runtime +from deerflow.config.database_config import DatabaseConfig + + +def _config_with_backend(backend: str) -> SimpleNamespace: + return SimpleNamespace(database=DatabaseConfig(backend=backend)) + + +# --------------------------------------------------------------------------- +# Unit tests of the gate function itself +# --------------------------------------------------------------------------- + + +def test_gate_noop_when_gateway_workers_unset(monkeypatch): + """With GATEWAY_WORKERS unset, every backend must be accepted.""" + monkeypatch.delenv("GATEWAY_WORKERS", raising=False) + for backend in ("sqlite", "memory", "postgres"): + _enforce_postgres_for_multi_worker(_config_with_backend(backend)) + + +def test_gate_noop_for_single_worker(monkeypatch): + """GATEWAY_WORKERS=1 preserves the historical single-worker behavior.""" + monkeypatch.setenv("GATEWAY_WORKERS", "1") + for backend in ("sqlite", "memory", "postgres"): + _enforce_postgres_for_multi_worker(_config_with_backend(backend)) + + +def test_gate_allows_multi_worker_with_postgres(monkeypatch): + monkeypatch.setenv("GATEWAY_WORKERS", "2") + _enforce_postgres_for_multi_worker(_config_with_backend("postgres")) + + +def test_gate_rejects_multi_worker_with_sqlite(monkeypatch): + monkeypatch.setenv("GATEWAY_WORKERS", "2") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + msg = str(exc_info.value) + assert "GATEWAY_WORKERS=2" in msg + assert "postgres" in msg.lower() + assert "sqlite" in msg.lower() + + +def test_gate_rejects_multi_worker_with_memory(monkeypatch): + """The gate is not sqlite-specific: memory is also unsafe across processes.""" + monkeypatch.setenv("GATEWAY_WORKERS", "2") + with pytest.raises(SystemExit): + _enforce_postgres_for_multi_worker(_config_with_backend("memory")) + + +def test_gate_rejects_high_worker_counts(monkeypatch): + """The threshold is >1, not ==2; prod-scale counts must also be gated.""" + monkeypatch.setenv("GATEWAY_WORKERS", "4") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + assert "GATEWAY_WORKERS=4" in str(exc_info.value) + + +def test_gate_treats_invalid_env_as_single_worker(monkeypatch): + """Non-integer GATEWAY_WORKERS values must not crash startup. + + Uvicorn itself rejects these later; the gate should not preempt + that with its own crash. Falling back to 1 keeps the gate inert. + """ + for invalid in ("", "auto", "1.5", "abc", "0x4"): + monkeypatch.setenv("GATEWAY_WORKERS", invalid) + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + + +def test_gate_treats_zero_and_negatives_as_single_worker(monkeypatch): + """GATEWAY_WORKERS <= 1 (including 0 and negatives) skips the gate.""" + for value in ("0", "-1", "-999"): + monkeypatch.setenv("GATEWAY_WORKERS", value) + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + + +def test_gate_error_message_lists_both_remediations(monkeypatch): + """Operators must see both fix options without reading docs.""" + monkeypatch.setenv("GATEWAY_WORKERS", "2") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + msg = str(exc_info.value) + assert "GATEWAY_WORKERS=1" in msg, "must mention the rollback knob" + assert "Postgres" in msg, "must mention the alternative backend" + + +# --------------------------------------------------------------------------- +# Integration: the gate is wired into langgraph_runtime before init_engine +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_langgraph_runtime_invokes_gate_before_persistence_setup(monkeypatch): + """When the gate trips, no persistence / stream-bridge setup may run. + + Guards against regressions that reorder the gate behind + ``init_engine_from_config`` (or any other expensive startup step). + """ + monkeypatch.setenv("GATEWAY_WORKERS", "2") + + init_engine_from_config = AsyncMock(name="init_engine_from_config") + + @asynccontextmanager + async def _noop_stream_bridge(_config): + yield MagicMock() + + with ( + patch( + "deerflow.persistence.engine.init_engine_from_config", + init_engine_from_config, + ), + patch("deerflow.runtime.make_stream_bridge", side_effect=_noop_stream_bridge) as make_stream_bridge, + patch("deerflow.runtime.make_store", side_effect=_noop_stream_bridge) as make_store, + ): + app = FastAPI() + startup_config = _config_with_backend("sqlite") + with pytest.raises(SystemExit): + async with langgraph_runtime(app, startup_config): + pass + + init_engine_from_config.assert_not_called() + make_stream_bridge.assert_not_called() + make_store.assert_not_called()