more reliable wake events

Author: luke-beamcloud

fastloop/monitor.py

@@ -54,7 +54,7 @@ def is_healthy(self, max_stale_s: float = 30.0) -> bool:
 
     async def run(self) -> None:
         """Main loop."""
-        logger.info("LoopMonitor started")
+        logger.info(f"LoopMonitor started, queue_id={id(self.wake_queue)}")
         try:
             await self._app_start()
             await self._main_loop()
@@ -108,24 +108,47 @@ async def _main_loop(self) -> None:
             self._iteration += 1
             self._last_tick = time.time()
 
-            with contextlib.suppress(Exception):
+            try:
                 wakes = await self._drain_queue()
                 if wakes:
+                    logger.info(f"Processing {len(wakes)} wakes: {wakes}")
                     await self._process_wakes(wakes)
                 await self._maintenance()
+            except Exception as e:
+                logger.error(f"Monitor iteration error: {e}")
 
             if time.time() - last_heartbeat >= HEARTBEAT_S:
                 logger.info(
-                    f"LoopMonitor heartbeat: iter={self._iteration} queue={self.wake_queue.qsize()}"
+                    f"LoopMonitor heartbeat: iter={self._iteration} queue={self.wake_queue.qsize()} queue_id={id(self.wake_queue)}"
                 )
                 last_heartbeat = time.time()
 
     async def _drain_queue(self) -> list[str]:
-        """Get all pending wakes from the queue."""
+        """Drain wakes.
+
+        Prefer Redis-backed wake queue (cross-process), fallback to in-memory queue.
+        """
+        # Redis-backed: fixes multi-worker + multi-replica.
+        if hasattr(self.state_manager, "drain_wake_queue"):
+            try:
+                wakes = await self.state_manager.drain_wake_queue(  # type: ignore[attr-defined]
+                    timeout_s=WATCHDOG_INTERVAL_S
+                )
+                if wakes:
+                    logger.info(f"Got {len(wakes)} wakes from redis queue: {wakes[:5]}")
+                return wakes
+            except Exception as e:
+                logger.error(f"Error draining redis wake queue: {e}")
+
+        # Fallback: in-memory queue (single-process only).
         wakes: list[str] = []
         try:
-            wakes.append(
-                await asyncio.to_thread(self.wake_queue.get, True, WATCHDOG_INTERVAL_S)
+            item = await asyncio.to_thread(
+                self.wake_queue.get, True, WATCHDOG_INTERVAL_S
+            )
+            wakes.append(item)
+            logger.info(
+                f"Got wake from in-memory queue: {item}, queue_id={id(self.wake_queue)}"
             )
             while True:
                 try:
@@ -134,6 +157,8 @@ async def _drain_queue(self) -> list[str]:
                     break
         except Empty:
             pass
+        except Exception as e:
+            logger.error(f"Error draining in-memory queue: {e}")
         return wakes
 
     async def _process_wakes(self, wakes: list[str]) -> None:
@@ -153,15 +178,22 @@ async def _wake_loop(self, loop_id: str) -> None:
         if await self.state_manager.has_claim(loop_id):
             logger.debug(f"Loop {loop_id} already has claim, skipping wake")
             return
+
+        if not await self.state_manager.try_claim_loop_wake(loop_id):
+            logger.debug(f"Loop {loop_id} wake already claimed, skipping")
+            return
+
         logger.info(f"Waking loop: {loop_id}")
         if not await self.restart_callback(loop_id):
             logger.warning(f"Failed to restart loop: {loop_id}")
 
     async def _wake_workflow(self, run_id: str) -> None:
         if await self.state_manager.workflow_has_claim(run_id):
             return
+
         if not await self.fastloop.restart_workflow(run_id):
             await self.state_manager.update_workflow_status(run_id, LoopStatus.STOPPED)
+
         await self.state_manager.clear_workflow_wake_time(run_id)
 
     async def _maintenance(self) -> None:

fastloop/state/state.py

@@ -178,6 +178,11 @@ async def try_acquire_app_start_lock(self, loop_id: str) -> bool:
         """Try to acquire an app start lock for a loop."""
         pass
 
+    @abstractmethod
+    async def try_claim_loop_wake(self, loop_id: str) -> bool:
+        """Atomically try to claim a loop wake (dedupe across replicas)."""
+        pass
+
     @abstractmethod
     async def release_app_start_lock(self, loop_id: str) -> None:
         """Release the app start lock for a loop."""

fastloop/state/state_redis.py

@@ -50,10 +50,12 @@ class RedisKeys:
     LOOP_INITIAL_EVENT = f"{KEY_PREFIX}:{{app_name}}:initial_event:{{loop_id}}"
     LOOP_STATE = f"{KEY_PREFIX}:{{app_name}}:state:{{loop_id}}"
     LOOP_CLAIM = f"{KEY_PREFIX}:{{app_name}}:claim:{{loop_id}}"
+    LOOP_WAKE_CLAIM = f"{KEY_PREFIX}:{{app_name}}:wake_claim:{{loop_id}}"
     LOOP_CONTEXT = f"{KEY_PREFIX}:{{app_name}}:context:{{loop_id}}:{{key}}"
     LOOP_NONCE = f"{KEY_PREFIX}:{{app_name}}:nonce:{{loop_id}}"
     LOOP_EVENT_CHANNEL = f"{KEY_PREFIX}:{{app_name}}:events:{{loop_id}}:notify"
     LOOP_WAKE_SCHEDULE = f"{KEY_PREFIX}:{{app_name}}:wake_schedule"
+    WAKE_QUEUE = f"{KEY_PREFIX}:{{app_name}}:wake_queue"
     LOOP_MAPPING = f"{KEY_PREFIX}:{{app_name}}:mapping:{{external_ref_id}}"
     LOOP_CONNECTION_INDEX = f"{KEY_PREFIX}:{{app_name}}:connection_index:{{loop_id}}"
     LOOP_CONNECTION_KEY = (
@@ -140,6 +142,7 @@ def __init__(
         )
 
         self.wake_queue: Queue[str] = wake_queue
+        self._wake_queue_key = RedisKeys.WAKE_QUEUE.format(app_name=self.app_name)
         self._stop_wake_monitor = threading.Event()
         self.wake_thread: threading.Thread | None = None
 
@@ -216,34 +219,54 @@ def _process_due_wakes(self, rdb) -> int:
         # Process loop wakes
         loop_key = RedisKeys.LOOP_WAKE_SCHEDULE.format(app_name=self.app_name)
 
-        # Debug: check what's pending in the ZSET
-        all_pending = rdb.zrange(loop_key, 0, -1, withscores=True)
-        if all_pending:
-            first_id, first_score = all_pending[0]
-            logger.info(
-                f"Pending wakes: key={loop_key}, count={len(all_pending)}, "
-                f"first=({first_id.decode()}, {first_score}), now={now}, "
-                f"due_in={first_score - now:.1f}s"
-            )
+        # Keep the wake thread quiet; it runs continuously and can be noisy.
 
         for loop_id_bytes in rdb.zrangebyscore(loop_key, "-inf", now):
             loop_id = loop_id_bytes.decode("utf-8")
             if rdb.zrem(loop_key, loop_id):
-                self.wake_queue.put(loop_id)
                 processed += 1
-                logger.info(f"Queued wake: {loop_id}")
+                logger.info(f"Queued wake: {loop_id} -> redis:{self._wake_queue_key}")
+                # Redis-backed wake queue so any process/replica can consume.
+                rdb.rpush(self._wake_queue_key, loop_id)
 
         # Process workflow wakes
         wf_key = RedisKeys.WORKFLOW_WAKE_SCHEDULE.format(app_name=self.app_name)
         for wf_id_bytes in rdb.zrangebyscore(wf_key, "-inf", now):
             wf_id = wf_id_bytes.decode("utf-8")
             if rdb.zrem(wf_key, wf_id):
-                self.wake_queue.put(f"{WORKFLOW_WAKE_PREFIX}{wf_id}")
                 processed += 1
-                logger.info(f"Queued workflow wake: {wf_id}")
+                payload = f"{WORKFLOW_WAKE_PREFIX}{wf_id}"
+                logger.info(
+                    f"Queued workflow wake: {wf_id} -> redis:{self._wake_queue_key}"
+                )
+                rdb.rpush(self._wake_queue_key, payload)
 
         return processed
 
+    async def drain_wake_queue(
+        self, *, timeout_s: float, max_items: int = 100
+    ) -> list[str]:
+        """Drain wake events from the Redis-backed wake queue.
+
+        Uses BLPOP to wait for a single item, then drains remaining items with LPOP.
+        """
+        wakes: list[str] = []
+
+        item = await self.rdb.blpop(self._wake_queue_key, timeout=timeout_s)  # type: ignore
+        if not item:
+            return wakes
+
+        _key, value = item
+        wakes.append(value.decode("utf-8"))
+
+        for _ in range(max_items - 1):
+            v = await self.rdb.lpop(self._wake_queue_key)  # type: ignore
+            if not v:
+                break
+            wakes.append(v.decode("utf-8"))
+
+        return wakes
+
     async def set_loop_mapping(self, external_ref_id: str, loop_id: str):
         await self.rdb.set(
             RedisKeys.LOOP_MAPPING.format(
@@ -491,6 +514,14 @@ async def try_claim_loop_recovery(self, loop_id: str) -> bool:
         acquired = await self.rdb.set(claim_key, "1", nx=True, ex=60)
         return acquired is not None
 
+    async def try_claim_loop_wake(self, loop_id: str) -> bool:
+        """Atomically try to claim a loop wake. Returns True if this caller won."""
+        claim_key = RedisKeys.LOOP_WAKE_CLAIM.format(
+            app_name=self.app_name, loop_id=loop_id
+        )
+        acquired = await self.rdb.set(claim_key, "1", nx=True, ex=30)
+        return acquired is not None
+
     async def get_all_loop_ids(self) -> set[str]:
         members = await self.rdb.smembers(
             RedisKeys.LOOP_INDEX.format(app_name=self.app_name)
@@ -632,7 +663,8 @@ async def push_event(self, loop_id: str, event: "LoopEvent"):
             await pipe.execute()
 
         if event.sender == LoopEventSender.CLIENT:
-            self.wake_queue.put_nowait(loop_id)
+            # Wake via Redis-backed wake queue so any process/replica can consume it.
+            await self.rdb.rpush(self._wake_queue_key, loop_id)  # type: ignore
 
     async def get_context_value(self, loop_id: str, key: str) -> Any:
         value_str = await self.rdb.get(

fastloop/state/state_s3.py

@@ -559,6 +559,13 @@ async def try_acquire_app_start_lock(self, loop_id: str) -> bool:
                 return False
             raise
 
+    async def try_claim_loop_wake(self, loop_id: str) -> bool:
+        """Best-effort loop wake claim for S3 state.
+
+        Redis is the intended scalable state backend for multi-replica wakes.
+        """
+        return True
+
     async def release_app_start_lock(self, loop_id: str) -> None:
         """Release the app start lock."""
         lock_key = f"{self.prefix}/{self.app_name}/app_start_lock/{loop_id}.lock"

pyproject.toml

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