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

@@ -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
 

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()