fix(scheduler-sidecar): host AsyncIOScheduler inside a real event loop

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

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)