From 1e27cdbdd7eb0d1a5ee89436d352f2ab01ca8a43 Mon Sep 17 00:00:00 2001 From: Ruby Wolff Date: Sat, 11 Jul 2026 08:33:10 +1000 Subject: [PATCH] fix(scheduler-sidecar): host AsyncIOScheduler inside a real event loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prod crash-loop: RuntimeError: no running event loop from apscheduler/schedulers/asyncio.py, raised inside start_scheduler() -> AsyncIOScheduler.start(). It ran fine in-process before because it bound to uvicorn's already-running loop; the sidecar's sync _run() called it with no loop at all, so nothing scheduled in prod and the container restarted forever. DB wait stays synchronous (plain blocking I/O, no loop needed). Everything from start_scheduler() onward now runs inside asyncio.run(_amain()): signal handlers install via loop.add_signal_handler (falling back to signal.signal + call_soon_threadsafe where add_signal_handler isn't implemented), start_scheduler() runs with a loop bound, an asyncio.Event gates shutdown, then stop_scheduler(). core/scheduler.py is untouched. New regression test stubs start_scheduler() to call asyncio.get_running_loop() itself and asserts it doesn't raise — verified against the previous synchronous implementation that this reproduces the exact prod RuntimeError, and passes against the fix. Co-Authored-By: Claude Fable 5 --- rivaflow/rivaflow/scheduler_main.py | 68 +++++++++++---- rivaflow/tests/unit/test_scheduler_sidecar.py | 84 +++++++++++++++---- 2 files changed, 121 insertions(+), 31 deletions(-) diff --git a/rivaflow/rivaflow/scheduler_main.py b/rivaflow/rivaflow/scheduler_main.py index 8842c94..29b3c80 100644 --- a/rivaflow/rivaflow/scheduler_main.py +++ b/rivaflow/rivaflow/scheduler_main.py @@ -10,16 +10,25 @@ ``RIVAFLOW_RUN_SCHEDULER=0`` (see ``rivaflow.api.main``), and docker-compose.prod.yml runs this module as its own `scheduler` service. +``core/scheduler.py`` uses APScheduler's AsyncIOScheduler, which binds to +whatever asyncio loop is running when ``.start()`` is called — it ran fine +inside the old in-process setup because uvicorn's own loop was already +running. A bare synchronous ``_run()`` has no loop at all, so +``start_scheduler()`` raised ``RuntimeError: no running event loop`` and the +sidecar crash-looped. The DB wait stays synchronous (it has nothing to do +with asyncio); everything from ``start_scheduler()`` onward runs inside +``asyncio.run(_amain())``. + Does NOT run database migrations (start.sh / the api container already does) and does NOT import or serve the FastAPI app — this process only imports what start_scheduler()/stop_scheduler() need. """ +import asyncio import logging import os import signal import sys -import threading import time from collections.abc import Callable from types import FrameType @@ -45,7 +54,7 @@ def _wait_for_db( Returns True once the DB responds, False if timeout_seconds is exceeded without a successful query. Backs off between attempts (1s, 2s, 4s, ... capped at _DB_WAIT_MAX_BACKOFF) so a slow-starting Postgres container - isn't hammered. + isn't hammered. Runs before the event loop exists — plain blocking I/O. """ deadline = time.monotonic() + timeout_seconds backoff = _DB_WAIT_INITIAL_BACKOFF @@ -78,37 +87,66 @@ def _wait_for_db( backoff = min(backoff * 2, _DB_WAIT_MAX_BACKOFF) -def _install_signal_handlers(stop_event: threading.Event) -> None: - """Register SIGTERM/SIGINT to signal a clean shutdown via stop_event.""" +def _install_signal_handlers( + loop: asyncio.AbstractEventLoop, stop_event: asyncio.Event +) -> None: + """Register SIGTERM/SIGINT to flip stop_event, inside the running loop. + + Prefers loop.add_signal_handler (asyncio-native, Unix only — this is what + actually runs in prod). Falls back to signal.signal + call_soon_threadsafe + on platforms where add_signal_handler isn't implemented (e.g. Windows dev + machines), since a plain signal.signal handler fires on an arbitrary OS + thread and can't touch the loop's Event directly. + """ - def _handle(signum: int, _frame: FrameType | None) -> None: - logger.info("Received signal %s — shutting down scheduler", signum) + def _set_stop() -> None: + logger.info("Received shutdown signal — stopping scheduler") stop_event.set() - signal.signal(signal.SIGTERM, _handle) - signal.signal(signal.SIGINT, _handle) + try: + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, _set_stop) + except NotImplementedError: + def _handle(signum: int, _frame: FrameType | None) -> None: + logger.info("Received signal %s — stopping scheduler", signum) + loop.call_soon_threadsafe(stop_event.set) + + signal.signal(signal.SIGTERM, _handle) + signal.signal(signal.SIGINT, _handle) -def _run(stop_event: threading.Event | None = None) -> int: - """Start the scheduler and block until a shutdown signal (or stop_event) fires.""" - if not _wait_for_db(): - return 1 - event = stop_event if stop_event is not None else threading.Event() - _install_signal_handlers(event) +async def _amain(stop_event: asyncio.Event | None = None) -> int: + """Start the scheduler inside a running loop and block until shutdown. + + AsyncIOScheduler.start() must be called with a loop already running + (it binds to it) — hence this being async and driven via asyncio.run(), + rather than start_scheduler() being called from plain sync code. + """ + loop = asyncio.get_running_loop() + event = stop_event if stop_event is not None else asyncio.Event() + _install_signal_handlers(loop, event) start_scheduler() logger.info( "Scheduler sidecar running (pid=%d) — waiting for shutdown signal", os.getpid() ) - event.wait() + await event.wait() stop_scheduler() logger.info("Scheduler sidecar exiting cleanly") return 0 +def _run() -> int: + """Sync entrypoint: bounded DB wait, then host the scheduler in a real event loop.""" + if not _wait_for_db(): + return 1 + + return asyncio.run(_amain()) + + def main() -> None: sys.exit(_run()) diff --git a/rivaflow/tests/unit/test_scheduler_sidecar.py b/rivaflow/tests/unit/test_scheduler_sidecar.py index bf5b722..4e13a56 100644 --- a/rivaflow/tests/unit/test_scheduler_sidecar.py +++ b/rivaflow/tests/unit/test_scheduler_sidecar.py @@ -15,8 +15,9 @@ from __future__ import annotations +import asyncio +import os import signal -import threading from contextlib import contextmanager from pathlib import Path @@ -149,6 +150,17 @@ def test_wait_for_db_failure_retries_then_gives_up(monkeypatch): # --------------------------------------------------------------------------- # 2b. rivaflow.scheduler_main — run loop (start → block → stop) +# +# Regression context (2026-07): the sidecar crash-looped in prod with +# `RuntimeError: no running event loop` from apscheduler/schedulers/asyncio.py +# — AsyncIOScheduler.start() binds to whatever loop is running when it's +# called, and a bare synchronous _run() has no loop at all. The fix moved +# everything from start_scheduler() onward into asyncio.run(_amain()). +# test_run_hosts_start_scheduler_inside_a_running_event_loop below pins that +# contract directly by having the stubbed start_scheduler() call +# asyncio.get_running_loop() itself — verified to fail against the prior +# synchronous implementation (see PR description / commit message for the +# before/after repro). # --------------------------------------------------------------------------- @@ -164,41 +176,81 @@ def test_run_exits_without_starting_scheduler_when_db_unreachable(monkeypatch): assert exit_code == 1 assert called == [] + # No event loop should even be spun up when the DB never comes up. -def test_run_starts_then_stops_scheduler_in_order(monkeypatch): +def test_run_hosts_start_scheduler_inside_a_running_event_loop(monkeypatch): + """start_scheduler() must observe a running asyncio loop (AsyncIOScheduler.start() + binds to it) — this is exactly the contract the sync-_run() bug violated.""" import rivaflow.scheduler_main as sched_main monkeypatch.setattr(sched_main, "_wait_for_db", lambda: True) + order: list[str] = [] + + def _fake_start_scheduler() -> None: + asyncio.get_running_loop() # raises RuntimeError outside asyncio.run() + order.append("start") + + monkeypatch.setattr(sched_main, "start_scheduler", _fake_start_scheduler) + monkeypatch.setattr(sched_main, "stop_scheduler", lambda: order.append("stop")) + + def _fast_stop(loop: asyncio.AbstractEventLoop, stop_event: asyncio.Event) -> None: + # Simulate a shutdown signal arriving shortly after startup, without + # touching the real process signal handlers from inside a test run. + loop.call_later(0.01, stop_event.set) + + monkeypatch.setattr(sched_main, "_install_signal_handlers", _fast_stop) + + exit_code = sched_main._run() + + assert order == ["start", "stop"] + assert exit_code == 0 + + +def test_amain_starts_then_stops_scheduler_in_order(monkeypatch): + """Drive _amain directly with a pre-built stop_event set shortly after start.""" + import rivaflow.scheduler_main as sched_main + order: list[str] = [] monkeypatch.setattr(sched_main, "start_scheduler", lambda: order.append("start")) monkeypatch.setattr(sched_main, "stop_scheduler", lambda: order.append("stop")) - # Don't touch the real process signal handlers from inside a test run. - monkeypatch.setattr(sched_main, "_install_signal_handlers", lambda _event: None) + monkeypatch.setattr( + sched_main, "_install_signal_handlers", lambda _loop, _event: None + ) - stop_event = threading.Event() - stop_event.set() # already set — event.wait() returns immediately, no real blocking + async def _drive() -> int: + event = asyncio.Event() + asyncio.get_running_loop().call_later(0.01, event.set) + return await sched_main._amain(stop_event=event) - exit_code = sched_main._run(stop_event=stop_event) + exit_code = asyncio.run(_drive()) assert order == ["start", "stop"] assert exit_code == 0 -def test_signal_handler_sets_stop_event(): - """SIGTERM (or SIGINT) delivered to the sidecar must flip the stop event, not exit abruptly.""" +def test_install_signal_handlers_sets_event_on_real_sigterm(): + """SIGTERM (or SIGINT) delivered to the sidecar must flip the asyncio stop event.""" import rivaflow.scheduler_main as sched_main - event = threading.Event() original_sigterm = signal.getsignal(signal.SIGTERM) original_sigint = signal.getsignal(signal.SIGINT) - try: - sched_main._install_signal_handlers(event) - handler = signal.getsignal(signal.SIGTERM) - assert not event.is_set() - handler(signal.SIGTERM, None) - assert event.is_set() + async def _probe() -> None: + loop = asyncio.get_running_loop() + event = asyncio.Event() + sched_main._install_signal_handlers(loop, event) + try: + assert not event.is_set() + os.kill(os.getpid(), signal.SIGTERM) + await asyncio.wait_for(event.wait(), timeout=2) + assert event.is_set() + finally: + loop.remove_signal_handler(signal.SIGTERM) + loop.remove_signal_handler(signal.SIGINT) + + try: + asyncio.run(_probe()) finally: signal.signal(signal.SIGTERM, original_sigterm) signal.signal(signal.SIGINT, original_sigint)