From 4dd824b6dc29b59f361c4b76920861973ce7bbc5 Mon Sep 17 00:00:00 2001 From: Nishant Date: Tue, 4 Aug 2026 16:12:07 +0530 Subject: [PATCH 1/2] fix: gate background loops behind enable_background_workers Uvicorn runs lifespan once per worker, so multi-worker deployments were starting N schedulers/dispatchers/watchdogs against the same DB. Add AUTOMATION_ENABLE_BACKGROUND_WORKERS (default true) so request-serving replicas can leave the loops to a single dedicated process. Fixes #286 --- openhands/automation/app.py | 118 ++++++++++++-------- openhands/automation/config.py | 9 ++ tests/test_background_workers.py | 182 +++++++++++++++++++++++++++++++ 3 files changed, 263 insertions(+), 46 deletions(-) create mode 100644 tests/test_background_workers.py diff --git a/openhands/automation/app.py b/openhands/automation/app.py index 0d04a26..cb78da5 100644 --- a/openhands/automation/app.py +++ b/openhands/automation/app.py @@ -40,6 +40,75 @@ logger = logging.getLogger("automation.app") +def start_background_worker_tasks( + app: FastAPI, + settings, + shutdown_event: asyncio.Event, +) -> list[tuple[str, asyncio.Task]]: + """Start scheduler/dispatcher/watchdog when this process owns them. + + Returns the started ``(name, task)`` pairs so lifespan can await them on + shutdown. When ``settings.enable_background_workers`` is False, no tasks + are created (request-serving replicas in a multi-worker deployment). + """ + app.state.scheduler_task = None + app.state.dispatcher_task = None + app.state.watchdog_task = None + + if not settings.enable_background_workers: + logger.info( + "Background workers disabled " + "(AUTOMATION_ENABLE_BACKGROUND_WORKERS=false); " + "scheduler/dispatcher/watchdog will not start in this process" + ) + return [] + + # Scheduler: polls automations and creates PENDING runs + scheduler_task = asyncio.create_task( + scheduler_loop( + app.state.session_factory, + interval_seconds=settings.scheduler_interval_seconds, + shutdown_event=shutdown_event, + ) + ) + app.state.scheduler_task = scheduler_task + logger.info("Background scheduler started") + + # Dispatcher: picks up PENDING runs and dispatches them + if not settings.base_url: + logger.warning( + "AUTOMATION_BASE_URL not set — using localhost. " + "Sandboxes in the cloud won't be able to reach this URL." + ) + dispatcher_task = asyncio.create_task( + dispatcher_loop( + app.state.session_factory, + settings=settings, + interval_seconds=settings.dispatcher_interval_seconds, + shutdown_event=shutdown_event, + ) + ) + app.state.dispatcher_task = dispatcher_task + logger.info("Background dispatcher started") + + # Watchdog: marks stale RUNNING runs as FAILED + watchdog_task = asyncio.create_task( + watchdog_loop( + app.state.session_factory, + settings=settings, + shutdown_event=shutdown_event, + ) + ) + app.state.watchdog_task = watchdog_task + logger.info("Background watchdog started") + + return [ + ("scheduler", scheduler_task), + ("dispatcher", dispatcher_task), + ("watchdog", watchdog_task), + ] + + @asynccontextmanager async def lifespan(app: FastAPI): """Application startup/shutdown lifecycle.""" @@ -117,48 +186,9 @@ async def lifespan(app: FastAPI): msg = f"SQLite migration failed. Database may be inconsistent: {e}" raise RuntimeError(msg) from e - # Start the background scheduler and dispatcher shutdown_event = asyncio.Event() app.state.shutdown_event = shutdown_event - - # Scheduler: polls automations and creates PENDING runs - scheduler_task = asyncio.create_task( - scheduler_loop( - app.state.session_factory, - interval_seconds=settings.scheduler_interval_seconds, - shutdown_event=shutdown_event, - ) - ) - app.state.scheduler_task = scheduler_task - logger.info("Background scheduler started") - - # Dispatcher: picks up PENDING runs and dispatches them - if not settings.base_url: - logger.warning( - "AUTOMATION_BASE_URL not set — using localhost. " - "Sandboxes in the cloud won't be able to reach this URL." - ) - dispatcher_task = asyncio.create_task( - dispatcher_loop( - app.state.session_factory, - settings=settings, - interval_seconds=settings.dispatcher_interval_seconds, - shutdown_event=shutdown_event, - ) - ) - app.state.dispatcher_task = dispatcher_task - logger.info("Background dispatcher started") - - # Watchdog: marks stale RUNNING runs as FAILED - watchdog_task = asyncio.create_task( - watchdog_loop( - app.state.session_factory, - settings=settings, - shutdown_event=shutdown_event, - ) - ) - app.state.watchdog_task = watchdog_task - logger.info("Background watchdog started") + background_tasks = start_background_worker_tasks(app, settings, shutdown_event) yield @@ -166,12 +196,8 @@ async def lifespan(app: FastAPI): logger.info("Shutting down background tasks...") shutdown_event.set() - # Wait for all tasks to exit gracefully - for task_name, task in [ - ("scheduler", scheduler_task), - ("dispatcher", dispatcher_task), - ("watchdog", watchdog_task), - ]: + # Wait for started tasks to exit gracefully (no-op when workers disabled) + for task_name, task in background_tasks: try: await asyncio.wait_for(task, timeout=5.0) except TimeoutError: diff --git a/openhands/automation/config.py b/openhands/automation/config.py index 0b8727d..26ac139 100644 --- a/openhands/automation/config.py +++ b/openhands/automation/config.py @@ -327,6 +327,9 @@ class ServiceSettings(BaseSettings): AUTOMATION_WORKSPACE_BASE: Base workspace directory (local mode default) # Background workers + AUTOMATION_ENABLE_BACKGROUND_WORKERS: Start scheduler/dispatcher/watchdog + in this process (default: True). Set False on request-serving replicas + when a dedicated background process owns the loops (see #286). AUTOMATION_SCHEDULER_INTERVAL_SECONDS: Scheduler poll interval (default: 60) AUTOMATION_SCHEDULER_BATCH_SIZE: Scheduler batch size (default: 50) AUTOMATION_DISPATCHER_INTERVAL_SECONDS: Dispatcher poll interval (default: 10) @@ -410,6 +413,12 @@ class ServiceSettings(BaseSettings): openhands_api_base_url: str = "https://app.all-hands.dev" # Background workers + # When True (default), this process runs scheduler/dispatcher/watchdog. + # Uvicorn runs lifespan once per worker process, so multi-worker / multi-replica + # deployments should enable this on exactly one dedicated process and disable + # it on request-serving instances to avoid duplicate polling and sandbox-API + # fan-out (see OpenHands/automation#286). + enable_background_workers: bool = True scheduler_interval_seconds: int = 60 scheduler_batch_size: int = 50 dispatcher_interval_seconds: int = 10 diff --git a/tests/test_background_workers.py b/tests/test_background_workers.py new file mode 100644 index 0000000..351906e --- /dev/null +++ b/tests/test_background_workers.py @@ -0,0 +1,182 @@ +"""Regression tests for single-owner background workers (#286). + +Uvicorn runs lifespan once per worker process. Without a gate, N workers start +N schedulers / dispatchers / watchdogs against the same DB. These tests pin the +``enable_background_workers`` ownership model from #286. +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import FastAPI + +from openhands.automation.app import start_background_worker_tasks +from openhands.automation.config import ServiceSettings + + +async def _idle_loop(*_args, shutdown_event: asyncio.Event, **_kwargs) -> None: + """Stand-in for scheduler/dispatcher/watchdog that exits on shutdown.""" + await shutdown_event.wait() + + +def _settings(*, enable: bool) -> ServiceSettings: + return ServiceSettings( + enable_background_workers=enable, + base_url="https://example.test", + scheduler_interval_seconds=60, + dispatcher_interval_seconds=10, + watchdog_interval_seconds=60, + ) + + +def _app_with_session_factory() -> FastAPI: + app = FastAPI() + app.state.session_factory = MagicMock(name="session_factory") + return app + + +@pytest.mark.asyncio +async def test_disabled_starts_no_background_workers() -> None: + """Request-serving workers must not start scheduler/dispatcher/watchdog.""" + app = _app_with_session_factory() + shutdown_event = asyncio.Event() + + with ( + patch("openhands.automation.app.scheduler_loop", new=_idle_loop), + patch("openhands.automation.app.dispatcher_loop", new=_idle_loop), + patch("openhands.automation.app.watchdog_loop", new=_idle_loop), + ): + tasks = start_background_worker_tasks( + app, _settings(enable=False), shutdown_event + ) + + assert tasks == [] + assert app.state.scheduler_task is None + assert app.state.dispatcher_task is None + assert app.state.watchdog_task is None + + +@pytest.mark.asyncio +async def test_enabled_starts_exactly_one_of_each_worker() -> None: + """The dedicated background process owns one of each loop.""" + app = _app_with_session_factory() + shutdown_event = asyncio.Event() + + with ( + patch("openhands.automation.app.scheduler_loop", new=_idle_loop), + patch("openhands.automation.app.dispatcher_loop", new=_idle_loop), + patch("openhands.automation.app.watchdog_loop", new=_idle_loop), + ): + tasks = start_background_worker_tasks( + app, _settings(enable=True), shutdown_event + ) + + try: + names = [name for name, _task in tasks] + assert names == ["scheduler", "dispatcher", "watchdog"] + assert app.state.scheduler_task is tasks[0][1] + assert app.state.dispatcher_task is tasks[1][1] + assert app.state.watchdog_task is tasks[2][1] + assert all(not task.done() for _name, task in tasks) + finally: + shutdown_event.set() + await asyncio.gather(*(task for _name, task in tasks)) + + +@pytest.mark.asyncio +async def test_multi_worker_simulation_single_owner() -> None: + """Simulate replicas×workers: only the process with the flag owns loops. + + Three "workers" start; two are request-serving (flag off) and one is the + dedicated background owner (flag on). Across the deployment there must be + exactly one scheduler, one dispatcher, and one watchdog. + """ + worker_flags = (False, False, True) + started: list[tuple[str, asyncio.Task]] = [] + shutdown_events: list[asyncio.Event] = [] + + with ( + patch("openhands.automation.app.scheduler_loop", new=_idle_loop), + patch("openhands.automation.app.dispatcher_loop", new=_idle_loop), + patch("openhands.automation.app.watchdog_loop", new=_idle_loop), + ): + for enable in worker_flags: + app = _app_with_session_factory() + shutdown_event = asyncio.Event() + shutdown_events.append(shutdown_event) + started.extend( + start_background_worker_tasks( + app, _settings(enable=enable), shutdown_event + ) + ) + + try: + by_name: dict[str, list[asyncio.Task]] = { + "scheduler": [], + "dispatcher": [], + "watchdog": [], + } + for name, task in started: + by_name[name].append(task) + + assert len(by_name["scheduler"]) == 1 + assert len(by_name["dispatcher"]) == 1 + assert len(by_name["watchdog"]) == 1 + assert len(started) == 3 + finally: + for event in shutdown_events: + event.set() + if started: + await asyncio.gather(*(task for _name, task in started)) + + +@pytest.mark.asyncio +async def test_multi_worker_all_enabled_still_duplicates() -> None: + """Document the operator invariant: enabling on every worker still fans out. + + The gate does not elect a leader. If every uvicorn worker leaves the flag + at its default (True), each still starts a full set of loops — same as + before #286 for single-process deploys, but wrong for multi-worker. + """ + started: list[tuple[str, asyncio.Task]] = [] + shutdown_events: list[asyncio.Event] = [] + + with ( + patch("openhands.automation.app.scheduler_loop", new=_idle_loop), + patch("openhands.automation.app.dispatcher_loop", new=_idle_loop), + patch("openhands.automation.app.watchdog_loop", new=_idle_loop), + ): + for _ in range(3): + app = _app_with_session_factory() + shutdown_event = asyncio.Event() + shutdown_events.append(shutdown_event) + started.extend( + start_background_worker_tasks( + app, _settings(enable=True), shutdown_event + ) + ) + + try: + assert len(started) == 9 # 3 workers × 3 loops + assert sum(1 for name, _ in started if name == "scheduler") == 3 + assert sum(1 for name, _ in started if name == "dispatcher") == 3 + assert sum(1 for name, _ in started if name == "watchdog") == 3 + finally: + for event in shutdown_events: + event.set() + await asyncio.gather(*(task for _name, task in started)) + + +def test_enable_background_workers_defaults_true() -> None: + """Local / single-process deploys keep prior behavior without new env.""" + assert ServiceSettings().enable_background_workers is True + + +def test_enable_background_workers_env_false(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("AUTOMATION_ENABLE_BACKGROUND_WORKERS", "false") + # ServiceSettings reads env at construction via pydantic-settings + settings = ServiceSettings() + assert settings.enable_background_workers is False From e9fcc34387464f52286106a13793b1022c0aa7c5 Mon Sep 17 00:00:00 2001 From: Nishant Date: Tue, 4 Aug 2026 16:17:36 +0530 Subject: [PATCH 2/2] docs: note AUTOMATION_ENABLE_BACKGROUND_WORKERS for multi-worker deploys --- README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/README.md b/README.md index 58c3ca2..eaa47c1 100644 --- a/README.md +++ b/README.md @@ -95,3 +95,7 @@ containers/ # Docker configuration ## Deployment This service is deployed via the [deploy repository](https://github.com/All-Hands-AI/deploy). Docker images are automatically built and pushed to `ghcr.io/openhands/automation` on every push to main and on tags. + +### Multi-worker / multi-replica + +Uvicorn starts the cron scheduler, run dispatcher, and staleness watchdog once per worker process. For multi-worker deployments, run those loops in a single dedicated process (`AUTOMATION_ENABLE_BACKGROUND_WORKERS=true`, the default) and set `AUTOMATION_ENABLE_BACKGROUND_WORKERS=false` on request-serving replicas so they only handle HTTP.