ensure restarts

Author: luke-beamcloud

fastloop/fastloop.py

@@ -9,7 +9,7 @@
 
 import asyncio
 from collections.abc import Callable
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 from enum import Enum
 from http import HTTPStatus
 from queue import Queue
@@ -71,19 +71,18 @@ def __init__(
     ):
         @asynccontextmanager
         async def lifespan(_: FastAPI):
-            self._monitor_task = asyncio.create_task(
-                LoopMonitor(
-                    state_manager=self.state_manager,
-                    loop_manager=self.loop_manager,
-                    restart_callback=self.restart_loop,
-                    wake_queue=self.wake_queue,
-                    fastloop_instance=self,
-                ).run()
-            )
+            self._stopping = False
+            self._start_monitor(reason="lifespan")
 
             yield
 
-            self._monitor_task.cancel()
+            self._stopping = True
+            if self._monitor_restart_task:
+                self._monitor_restart_task.cancel()
+            if self._monitor_task:
+                self._monitor_task.cancel()
+                with suppress(asyncio.CancelledError):
+                    await self._monitor_task
             await self.loop_manager.stop_all()
             await self.workflow_manager.stop_all()
             await self.task_manager.stop_all()
@@ -108,6 +107,9 @@ async def lifespan(_: FastAPI):
         self.workflow_manager: WorkflowManager = WorkflowManager(self.state_manager)
         self.task_manager: TaskManager = TaskManager(self.state_manager)
         self._monitor_task: asyncio.Task[None] | None = None
+        self._monitor_restart_task: asyncio.Task[None] | None = None
+        self._monitor_restart_delay_s: float = 0.5
+        self._stopping: bool = False
         self._loop_start_func: Callable[[LoopContext], None] | None = None
         self._loop_metadata: dict[str, dict[str, Any]] = {}
         self._workflow_metadata: dict[str, dict[str, Any]] = {}
@@ -137,6 +139,58 @@ async def events_history_endpoint(entity_id: str):  # type: ignore
         async def events_sse_endpoint(entity_id: str):  # type: ignore
             return await self.loop_manager.events_sse(entity_id)
 
+        @self.middleware("http")
+        async def _ensure_monitor_running(request, call_next):  # type: ignore
+            if self._monitor_task is None or self._monitor_task.done():
+                self._start_monitor(reason="middleware_safety_net")
+            return await call_next(request)
+
+    def _start_monitor(self, *, reason: str) -> None:
+        if self._stopping:
+            return
+        if self._monitor_task is not None and not self._monitor_task.done():
+            return
+        logger.info("Starting LoopMonitor", extra={"reason": reason})
+        self._monitor_task = asyncio.create_task(
+            LoopMonitor(
+                state_manager=self.state_manager,
+                loop_manager=self.loop_manager,
+                restart_callback=self.restart_loop,
+                wake_queue=self.wake_queue,
+                fastloop_instance=self,
+            ).run()
+        )
+        self._monitor_task.add_done_callback(self._on_monitor_done)
+
+    def _on_monitor_done(self, task: asyncio.Task[Any]) -> None:
+        if self._stopping:
+            return
+        with suppress(asyncio.CancelledError):
+            exc = task.exception()
+        if exc is None:
+            logger.warning("LoopMonitor stopped unexpectedly; restarting")
+        else:
+            logger.error("LoopMonitor crashed; restarting", extra={"error": str(exc)})
+        self._schedule_monitor_restart()
+
+    def _schedule_monitor_restart(self) -> None:
+        if self._stopping:
+            return
+        if (
+            self._monitor_restart_task is not None
+            and not self._monitor_restart_task.done()
+        ):
+            return
+
+        delay = self._monitor_restart_delay_s
+        self._monitor_restart_delay_s = min(self._monitor_restart_delay_s * 2, 10.0)
+
+        async def _restart() -> None:
+            await asyncio.sleep(delay)
+            self._start_monitor(reason="restart_after_crash")
+
+        self._monitor_restart_task = asyncio.create_task(_restart())
+
     @property
     def config(self) -> BaseConfig:
         return self.config_manager.get_config()

fastloop/monitor.py

@@ -292,42 +292,66 @@ async def _register_schedules(self) -> None:
 
     async def run(self):
         """Main monitor loop."""
+        logger.info("LoopMonitor started")
+
         if not self._app_start_processed:
-            await self._process_app_start_callbacks()
-            await self._register_schedules()
-            self._app_start_processed = True
+            try:
+                await self._process_app_start_callbacks()
+                await self._register_schedules()
+            except Exception as e:
+                logger.error(
+                    "LoopMonitor startup failed (continuing without app-start processing)",
+                    extra={"error": str(e)},
+                )
+            finally:
+                self._app_start_processed = True
 
         while not self._stop_event.is_set():
             try:
-                # Process all pending wakes, handling errors individually
-                # Use get_nowait in a try/except to avoid race between empty() and get()
-                while True:
-                    try:
-                        wake_id = self.wake_queue.get_nowait()
+                # Block for up to WATCHDOG_INTERVAL_S to avoid polling the event loop.
+                # This is a threadsafe Queue fed by the Redis wake thread.
+                wake_batch: list[str] = []
+                try:
+                    first = await asyncio.to_thread(
+                        self.wake_queue.get, True, WATCHDOG_INTERVAL_S
+                    )
+                    wake_batch.append(first)
+                except Empty:
+                    pass
+
+                # Drain any additional wakes immediately.
+                if wake_batch:
+                    while True:
                         try:
-                            await self._process_wake(wake_id)
-                        except Exception as e:
-                            logger.error(
-                                "Error processing wake",
-                                extra={"wake_id": wake_id, "error": str(e)},
-                            )
-                    except Empty:
-                        break
+                            wake_batch.append(self.wake_queue.get_nowait())
+                        except Empty:
+                            break
+
+                processed_wakes = 0
+                for wake_id in wake_batch:
+                    try:
+                        await self._process_wake(wake_id)
+                        processed_wakes += 1
+                    except Exception as e:
+                        logger.error(
+                            "Error processing wake",
+                            extra={"wake_id": wake_id, "error": str(e)},
+                        )
+                if processed_wakes:
+                    logger.info(
+                        "Processed wakes from queue",
+                        extra={
+                            "count": processed_wakes,
+                            "queue_size": self.wake_queue.qsize(),
+                        },
+                    )
 
                 await self._check_orphaned_loops()
                 await self._check_orphaned_workflows()
                 await self._check_orphaned_tasks()
                 await self._check_scheduled_workflows()
                 await self._check_scheduled_tasks()
                 await self._check_disconnect_stops()
-
-                try:
-                    await asyncio.wait_for(
-                        self._stop_event.wait(), timeout=WATCHDOG_INTERVAL_S
-                    )
-                    break
-                except TimeoutError:
-                    pass
             except asyncio.CancelledError:
                 break
             except Exception as e:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastloop"
-version = "0.1.106"
+version = "0.1.107"
 description = "A Python package for deploying stateful loops"
 readme = "README.md"
 requires-python = ">=3.12"