← back to Exo
Make info gatherer monitors resilient with retry loops and timeouts (#1448)
98773437f34dcc46fa46eea56dc9504dd90976ea · 2026-02-11 15:07:36 -0800 · Alex Cheema
## 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
<!-- Hardware: macOS with macmon installed -->
<!-- What you did: -->
- 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 <noreply@anthropic.com>
Files touched
M src/exo/utils/info_gatherer/info_gatherer.py
Diff
commit 98773437f34dcc46fa46eea56dc9504dd90976ea
Author: Alex Cheema <41707476+AlexCheema@users.noreply.github.com>
Date: Wed Feb 11 15:07:36 2026 -0800
Make info gatherer monitors resilient with retry loops and timeouts (#1448)
## 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
<!-- Hardware: macOS with macmon installed -->
<!-- What you did: -->
- 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 <noreply@anthropic.com>
---
src/exo/utils/info_gatherer/info_gatherer.py | 94 +++++++++++++++++-----------
1 file changed, 56 insertions(+), 38 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
-
- idents = [it for i in data if (it := i.ident(iface_map)) is not None]
- await self.info_sender.send(MacThunderboltIdentifiers(idents=idents))
+ with fail_after(30):
+ iface_map = await _gather_iface_map()
+ if iface_map is None:
+ raise ValueError("Failed to gather interface map")
+
+ data = await ThunderboltConnectivity.gather()
+ assert data is not None
+
+ 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))
+ 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)
← a8acb3ca dashboard: show available disk space on downloads page
·
back to Exo
·
fix: prevent DownloadModel TaskCreated event flood (#1452) 62e8110e →