From 98773437f34dcc46fa46eea56dc9504dd90976ea Mon Sep 17 00:00:00 2001 From: Alex Cheema <41707476+AlexCheema@users.noreply.github.com> Date: Wed, 11 Feb 2026 15:07:36 -0800 Subject: [PATCH] Make info gatherer monitors resilient with retry loops and timeouts (#1448) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Motivation Info gatherer monitors could silently stop posting events, causing stale node state after rejoins. The macmon monitor was especially fragile — it had no retry loop, so a crash or parse error would kill it permanently. Worse, the unhandled exception would propagate to the TaskGroup and take down *all* sibling monitors. Additionally, none of the monitors had timeouts on their subprocess calls, so a hung `system_profiler` or `networksetup` could stall a monitor indefinitely. ## Changes - Wrap `_monitor_macmon` in a `while True` retry loop with `except Exception`, matching the pattern used by all other monitors - Add `fail_after` timeouts to all monitor loop bodies: - 10s for lightweight commands (`_monitor_misc`, `_watch_system_info`, `_gather_iface_map` init) - 30s for heavier commands (`_monitor_system_profiler_thunderbolt_data`, `_monitor_thunderbolt_bridge_status`) - Remove unused `CalledProcessError` and `cast` imports ## Why It Works All monitors now follow the same resilient pattern: `while True` → `try` with `fail_after` → `except Exception` (logs warning) → `sleep`. If a subprocess hangs, the timeout fires and `TimeoutError` is caught by the existing `except Exception` handler. If macmon crashes, it restarts after the interval instead of dying permanently. No single monitor failure can cascade to kill the others. ## Test Plan ### Manual Testing - Run exo, kill macmon process (`kill $(pgrep macmon)`), verify it restarts and metrics resume - Verify all monitors continue posting events after simulated hangs ### Automated Testing - All 188 existing tests pass - basedpyright: 0 errors - ruff: all checks passed 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 4.6 --- src/exo/utils/info_gatherer/info_gatherer.py | 92 ++++++++++++-------- 1 file changed, 55 insertions(+), 37 deletions(-) diff --git a/src/exo/utils/info_gatherer/info_gatherer.py b/src/exo/utils/info_gatherer/info_gatherer.py index 17f07aa1..8f31369b 100644 --- a/src/exo/utils/info_gatherer/info_gatherer.py +++ b/src/exo/utils/info_gatherer/info_gatherer.py @@ -420,7 +420,8 @@ class InfoGatherer: return while True: try: - await self.info_sender.send(await MiscData.gather()) + with fail_after(10): + await self.info_sender.send(await MiscData.gather()) except Exception as e: logger.warning(f"Error gathering misc data: {e}") await anyio.sleep(self.misc_poll_interval) @@ -428,20 +429,26 @@ class InfoGatherer: async def _monitor_system_profiler_thunderbolt_data(self): if self.system_profiler_interval is None: return - iface_map = await _gather_iface_map() - if iface_map is None: - return while True: try: - data = await ThunderboltConnectivity.gather() - assert data is not None + with fail_after(30): + iface_map = await _gather_iface_map() + if iface_map is None: + raise ValueError("Failed to gather interface map") - idents = [it for i in data if (it := i.ident(iface_map)) is not None] - await self.info_sender.send(MacThunderboltIdentifiers(idents=idents)) + data = await ThunderboltConnectivity.gather() + assert data is not None - conns = [it for i in data if (it := i.conn()) is not None] - await self.info_sender.send(MacThunderboltConnections(conns=conns)) + idents = [ + it for i in data if (it := i.ident(iface_map)) is not None + ] + await self.info_sender.send( + MacThunderboltIdentifiers(idents=idents) + ) + + conns = [it for i in data if (it := i.conn()) is not None] + await self.info_sender.send(MacThunderboltConnections(conns=conns)) except Exception as e: logger.warning(f"Error gathering Thunderbolt data: {e}") await anyio.sleep(self.system_profiler_interval) @@ -469,8 +476,9 @@ class InfoGatherer: return while True: try: - nics = await get_network_interfaces() - await self.info_sender.send(NodeNetworkInterfaces(ifaces=nics)) + with fail_after(10): + nics = await get_network_interfaces() + await self.info_sender.send(NodeNetworkInterfaces(ifaces=nics)) except Exception as e: logger.warning(f"Error gathering network interfaces: {e}") await anyio.sleep(self.interface_watcher_interval) @@ -480,9 +488,10 @@ class InfoGatherer: return while True: try: - curr = await ThunderboltBridgeInfo.gather() - if curr is not None: - await self.info_sender.send(curr) + with fail_after(30): + curr = await ThunderboltBridgeInfo.gather() + if curr is not None: + await self.info_sender.send(curr) except Exception as e: logger.warning(f"Error gathering Thunderbolt Bridge status: {e}") await anyio.sleep(self.thunderbolt_bridge_poll_interval) @@ -514,26 +523,35 @@ class InfoGatherer: if self.macmon_interval is None: return # macmon pipe --interval [interval in ms] - try: - async with await open_process( - [macmon_path, "pipe", "--interval", str(self.macmon_interval * 1000)] - ) as p: - if not p.stdout: - logger.critical("MacMon closed stdout") - return - async for text in TextReceiveStream( - BufferedByteReceiveStream(p.stdout) - ): - await self.info_sender.send(MacmonMetrics.from_raw_json(text)) - except CalledProcessError as e: - stderr_msg = "no stderr" - stderr_output = cast(bytes | str | None, e.stderr) - if stderr_output is not None: - stderr_msg = ( - stderr_output.decode() - if isinstance(stderr_output, bytes) - else str(stderr_output) + while True: + try: + async with await open_process( + [ + macmon_path, + "pipe", + "--interval", + str(self.macmon_interval * 1000), + ] + ) as p: + if not p.stdout: + logger.critical("MacMon closed stdout") + return + async for text in TextReceiveStream( + BufferedByteReceiveStream(p.stdout) + ): + await self.info_sender.send(MacmonMetrics.from_raw_json(text)) + except CalledProcessError as e: + stderr_msg = "no stderr" + stderr_output = cast(bytes | str | None, e.stderr) + if stderr_output is not None: + stderr_msg = ( + stderr_output.decode() + if isinstance(stderr_output, bytes) + else str(stderr_output) + ) + logger.warning( + f"MacMon failed with return code {e.returncode}: {stderr_msg}" ) - logger.warning( - f"MacMon failed with return code {e.returncode}: {stderr_msg}" - ) + except Exception as e: + logger.warning(f"Error in macmon monitor: {e}") + await anyio.sleep(self.macmon_interval)