diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index fb5378d177..00a3305dc2 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -56,6 +56,36 @@ def _browser_tools_enabled_in_config(config: AppConfig) -> bool: return any(getattr(tool, "name", None) == "browser_navigate" for tool in (getattr(config, "tools", None) or [])) +# ``GATEWAY_WORKERS`` is the knob this project documents and the one +# ``docker/docker-compose.yaml`` forwards as ``--workers``. ``backend/Dockerfile`` +# and ``scripts/serve.sh`` launch uvicorn with no worker count at all, and uvicorn +# then takes the count from ``WEB_CONCURRENCY``. Both must be read, otherwise a +# multi-process deployment starts with the safety gates inert. +# A blank value means "unset", as the compose command's ``${GATEWAY_WORKERS:-1}`` treats it. +_WORKER_COUNT_ENV_VARS = ("GATEWAY_WORKERS", "WEB_CONCURRENCY") + + +def _gateway_worker_count() -> tuple[int, str]: + """Return the Gateway worker-process count and the environment variable that set it. + + The name belongs in the refusal text: an operator who only ever set + ``WEB_CONCURRENCY`` cannot act on a message about ``GATEWAY_WORKERS``. + """ + for name in _WORKER_COUNT_ENV_VARS: + raw = os.environ.get(name) + if not raw or not raw.strip(): + continue + try: + return int(raw), name + except (TypeError, ValueError): + # Uvicorn rejects a non-numeric count itself, so this is not the place + # to fail the launch -- but an unparsable documented knob must not hide + # the count uvicorn takes from ``WEB_CONCURRENCY`` on the launches that + # pass no ``--workers``. Keep looking; report 1 only after the loop. + continue + return 1, _WORKER_COUNT_ENV_VARS[0] + + def _enforce_postgres_for_multi_worker(config: AppConfig) -> None: """Refuse unsafe multi-process configurations before persistence starts. @@ -80,10 +110,7 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None: 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 + workers, worker_env = _gateway_worker_count() scheduler = getattr(config, "scheduler", None) multi_instance_requested = bool(getattr(scheduler, "multi_instance", False)) @@ -104,24 +131,24 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None: return if config.scheduler.enabled and not multi_instance_scheduler: - raise SystemExit(f"GATEWAY_WORKERS={workers} cannot run with scheduler.enabled=true because each worker starts its own scheduler. Set GATEWAY_WORKERS=1, scheduler.multi_instance=true, or scheduler.enabled=false.") + raise SystemExit(f"{worker_env}={workers} cannot run with scheduler.enabled=true because each worker starts its own scheduler. Set {worker_env}=1, scheduler.multi_instance=true, or scheduler.enabled=false.") if _browser_tools_enabled_in_config(config): raise SystemExit(browser_multi_worker_error(workers)) 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.") + raise SystemExit(f"{worker_env}={workers} requires database.backend='postgres', but database.backend is '{backend}'. SQLite cannot support concurrent multi-process access. Set {worker_env}=1 or switch to Postgres.") if run_events_backend != "db": raise SystemExit( - f"GATEWAY_WORKERS={workers} requires run_events.backend='db', but run_events.backend is '{run_events_backend}'. " + f"{worker_env}={workers} requires run_events.backend='db', but run_events.backend is '{run_events_backend}'. " "Memory and JSONL event stores are process-local, so delivery receipt singleton guarantees cannot hold across workers. " - "Set GATEWAY_WORKERS=1 or configure run_events.backend: db." + f"Set {worker_env}=1 or configure run_events.backend: db." ) if run_ownership is None or not run_ownership.heartbeat_enabled: raise SystemExit( - f"GATEWAY_WORKERS={workers} requires run_ownership.heartbeat_enabled=true. " + f"{worker_env}={workers} requires run_ownership.heartbeat_enabled=true. " "Without heartbeat, every run has a NULL lease, so reconciliation " "treats all inflight runs as orphans — Worker B would kill Worker A's " "live runs on every rolling update or scale-up. " @@ -151,15 +178,13 @@ def _validate_agent_storage(config: AppConfig) -> None: f"but database.backend is '{db_backend}'. A 'memory' database is per-process and cannot " "share agent definitions across nodes. Set database.backend, or use agent_storage.backend='file'." ) - try: - workers = int(os.environ.get("GATEWAY_WORKERS", "1")) - except (TypeError, ValueError): - workers = 1 + workers, worker_env = _gateway_worker_count() if workers > 1 and db_backend == "postgres" and backend == "file": logger.warning( - "GATEWAY_WORKERS=%s with database.backend='postgres' but agent_storage.backend='file': " + "%s=%s with database.backend='postgres' but agent_storage.backend='file': " "custom agents and managed subagents are stored per-node on local disk and are not visible " "across workers/nodes. Set agent_storage.backend='db' to share them.", + worker_env, workers, ) diff --git a/backend/tests/test_multi_worker_postgres_gate.py b/backend/tests/test_multi_worker_postgres_gate.py index b4b1f1bcad..9d2801c5f3 100644 --- a/backend/tests/test_multi_worker_postgres_gate.py +++ b/backend/tests/test_multi_worker_postgres_gate.py @@ -18,7 +18,7 @@ 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 app.gateway.deps import _enforce_postgres_for_multi_worker, _validate_agent_storage, langgraph_runtime from app.gateway.routers.browser import _browser_tools_enabled from deerflow.config.database_config import DatabaseConfig from deerflow.config.run_ownership_config import RunOwnershipConfig @@ -44,6 +44,20 @@ def _config_with_backend( ) +@pytest.fixture(autouse=True) +def isolated_worker_env(monkeypatch): + """Keep the suite independent of the invoking shell's worker count. + + Several tests here assert that the gate stays inert, which only holds when + no worker count is set at all. ``WEB_CONCURRENCY`` is the count uvicorn takes + on the launches that pass no ``--workers``, so an exported value in the + invoking shell would otherwise turn these inert-gate expectations into + refusals. + """ + monkeypatch.delenv("GATEWAY_WORKERS", raising=False) + monkeypatch.delenv("WEB_CONCURRENCY", raising=False) + + # --------------------------------------------------------------------------- # Unit tests of the gate function itself # --------------------------------------------------------------------------- @@ -52,6 +66,7 @@ def _config_with_backend( def test_gate_noop_when_gateway_workers_unset(monkeypatch): """With GATEWAY_WORKERS unset, every backend must be accepted.""" monkeypatch.delenv("GATEWAY_WORKERS", raising=False) + monkeypatch.delenv("WEB_CONCURRENCY", raising=False) for backend in ("sqlite", "memory", "postgres"): _enforce_postgres_for_multi_worker(_config_with_backend(backend)) @@ -63,6 +78,107 @@ def test_gate_noop_for_single_worker(monkeypatch): _enforce_postgres_for_multi_worker(_config_with_backend(backend)) +# --------------------------------------------------------------------------- +# Uvicorn's WEB_CONCURRENCY fallback also starts several worker processes +# --------------------------------------------------------------------------- + + +def test_gate_rejects_multi_worker_from_uvicorn_worker_fallback(monkeypatch): + """WEB_CONCURRENCY=N starts N workers even with GATEWAY_WORKERS unset. + + ``backend/Dockerfile`` and ``scripts/serve.sh`` launch uvicorn with no + ``--workers``, so uvicorn takes the count from ``WEB_CONCURRENCY``. + """ + monkeypatch.delenv("GATEWAY_WORKERS", raising=False) + monkeypatch.setenv("WEB_CONCURRENCY", "2") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + assert "requires database.backend='postgres'" in str(exc_info.value) + + +def test_gate_rejects_scheduler_duplication_from_uvicorn_worker_fallback(monkeypatch): + """Each of those workers starts its own scheduler, which the gate exists to refuse.""" + monkeypatch.delenv("GATEWAY_WORKERS", raising=False) + monkeypatch.setenv("WEB_CONCURRENCY", "3") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite", scheduler_enabled=True)) + assert "each worker starts its own scheduler" in str(exc_info.value) + + +def test_gate_rejects_process_local_browser_from_uvicorn_worker_fallback(monkeypatch): + monkeypatch.delenv("GATEWAY_WORKERS", raising=False) + monkeypatch.setenv("WEB_CONCURRENCY", "2") + with pytest.raises(SystemExit, match="process-local"): + _enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True, browser_enabled=True)) + + +def test_gate_reads_uvicorn_worker_fallback_when_gateway_workers_is_blank(monkeypatch): + """A blank GATEWAY_WORKERS means unset, exactly as the compose default treats it.""" + monkeypatch.setenv("GATEWAY_WORKERS", "") + monkeypatch.setenv("WEB_CONCURRENCY", "2") + with pytest.raises(SystemExit): + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + + +def test_unparsable_gateway_workers_does_not_mask_uvicorn_worker_fallback(monkeypatch): + """A non-numeric documented knob must not report one worker while uvicorn starts four. + + The launchers that pass no ``--workers`` (``backend/Dockerfile``, ``scripts/serve.sh``, + ``backend/Makefile gateway``) take the count from ``WEB_CONCURRENCY`` alone, so there + ``GATEWAY_WORKERS=abc`` never reaches uvicorn at all. + """ + monkeypatch.setenv("GATEWAY_WORKERS", "abc") + monkeypatch.setenv("WEB_CONCURRENCY", "4") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + assert "WEB_CONCURRENCY=4" in str(exc_info.value), "must name the variable that actually set the count" + + +def test_unparsable_gateway_workers_does_not_mask_the_browser_gate(monkeypatch): + monkeypatch.setenv("GATEWAY_WORKERS", "auto") + monkeypatch.setenv("WEB_CONCURRENCY", "2") + config = _config_with_backend("postgres", heartbeat_enabled=True, browser_enabled=True) + with pytest.raises(SystemExit, match="process-local"): + _enforce_postgres_for_multi_worker(config) + + +def test_agent_storage_warning_names_the_fallback_when_the_knob_is_unparsable(monkeypatch, caplog): + monkeypatch.setenv("GATEWAY_WORKERS", "auto") + monkeypatch.setenv("WEB_CONCURRENCY", "2") + with caplog.at_level("WARNING"): + _validate_agent_storage(_config_with_backend("postgres", heartbeat_enabled=True)) + messages = [r.message for r in caplog.records if "not visible across workers" in r.message] + assert messages and "WEB_CONCURRENCY=2" in messages[0] + + +def test_gate_accepts_single_worker_from_uvicorn_worker_fallback(monkeypatch): + """WEB_CONCURRENCY=1 is a single worker, so the gate stays inert.""" + monkeypatch.delenv("GATEWAY_WORKERS", raising=False) + for value in ("1", "0", ""): + monkeypatch.setenv("WEB_CONCURRENCY", value) + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite", scheduler_enabled=True)) + + +def test_gate_prefers_gateway_workers_over_uvicorn_worker_fallback(monkeypatch): + """An explicit GATEWAY_WORKERS wins: uvicorn ignores WEB_CONCURRENCY when --workers is passed. + + ``docker/docker-compose.yaml`` always passes ``--workers ${GATEWAY_WORKERS:-1}``. + """ + monkeypatch.setenv("GATEWAY_WORKERS", "1") + monkeypatch.setenv("WEB_CONCURRENCY", "4") + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite", scheduler_enabled=True)) + + +def test_agent_storage_warning_names_the_variable_that_set_the_count(monkeypatch, caplog): + """The divergence warning must be actionable for a WEB_CONCURRENCY deployment.""" + monkeypatch.delenv("GATEWAY_WORKERS", raising=False) + monkeypatch.setenv("WEB_CONCURRENCY", "2") + with caplog.at_level("WARNING"): + _validate_agent_storage(_config_with_backend("postgres", heartbeat_enabled=True)) + messages = [r.message for r in caplog.records if "not visible across workers" in r.message] + assert messages and "WEB_CONCURRENCY=2" in messages[0] + + def test_gate_allows_multi_worker_with_postgres_and_heartbeat(monkeypatch): monkeypatch.setenv("GATEWAY_WORKERS", "2") _enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True)) @@ -244,7 +360,8 @@ 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. + that with its own crash. The count is then taken from the remaining + spellings, and with no other variable set the gate stays inert. """ for invalid in ("", "auto", "1.5", "abc", "0x4"): monkeypatch.setenv("GATEWAY_WORKERS", invalid)