[object Object]

← back to Exo

fix: log exceptions causing silent node shutdown (#1621)

190e63e56daf90cde22a845231fd95a11e29dca0 · 2026-02-25 08:24:18 -0800 · Alex Cheema

## Motivation

Nodes silently shut down mid-inference with no exception logged. The
logs show a clean-looking shutdown cascade (unsubscribe all topics →
stop worker → runner communication closed) but no error explaining
*why*. This makes debugging cluster issues extremely difficult.

## Changes

**`src/exo/routing/router.py`** — Added `try/except Exception` around
`_networking_recv()` and `_networking_recv_connection_messages()` loops.
Logs the root cause at ERROR level via
`logger.opt(exception=...).error(...)` before re-raising. Uses
`Exception` (not `BaseException`) so clean SIGTERM cancellation is
unaffected.

**`src/exo/main.py`** — Wrapped `anyio.run(node.run)` in
`try/except/finally`:
- `except BaseException`: logs the fatal exception at CRITICAL level
through loguru
- `finally`: ensures `logger.info("EXO Shutdown complete")` and
`logger_cleanup()` always run, guaranteeing the async log queue is
flushed before process exit

## Why It Works

The Router's recv loops (`_networking_recv`,
`_networking_recv_connection_messages`) are infinite `while True` loops
calling Rust bindings with **no exception handling**. When the Rust
networking layer raises `ConnectionError` (e.g. channel closed), the
exception cascades silently through anyio task groups: Router → Node →
all components shut down. The exception was never logged because (1) no
try-except anywhere in the cascade, and (2) `logger_cleanup()` was
skipped when `anyio.run()` raised, so loguru's async queue was never
flushed.

## Test Plan

### Manual Testing
- Run `uv run exo`, send SIGTERM → should see clean shutdown with "EXO
Shutdown complete", no error/critical logs
- Next time the silent shutdown reproduces, logs should now show the
actual exception with full traceback at ERROR level from the recv loop,
plus CRITICAL level from main()

### Automated Testing
- All 222 existing tests pass (1 pre-existing failure in
`test_python.py::test_sleep_on_multiple_items` unrelated to this change)
- `uv run basedpyright` — 0 errors
- `uv run ruff check` — all checks passed
- `nix fmt` — applied

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Files touched

Diff

commit 190e63e56daf90cde22a845231fd95a11e29dca0
Author: Alex Cheema <41707476+AlexCheema@users.noreply.github.com>
Date:   Wed Feb 25 08:24:18 2026 -0800

    fix: log exceptions causing silent node shutdown (#1621)
    
    ## Motivation
    
    Nodes silently shut down mid-inference with no exception logged. The
    logs show a clean-looking shutdown cascade (unsubscribe all topics →
    stop worker → runner communication closed) but no error explaining
    *why*. This makes debugging cluster issues extremely difficult.
    
    ## Changes
    
    **`src/exo/routing/router.py`** — Added `try/except Exception` around
    `_networking_recv()` and `_networking_recv_connection_messages()` loops.
    Logs the root cause at ERROR level via
    `logger.opt(exception=...).error(...)` before re-raising. Uses
    `Exception` (not `BaseException`) so clean SIGTERM cancellation is
    unaffected.
    
    **`src/exo/main.py`** — Wrapped `anyio.run(node.run)` in
    `try/except/finally`:
    - `except BaseException`: logs the fatal exception at CRITICAL level
    through loguru
    - `finally`: ensures `logger.info("EXO Shutdown complete")` and
    `logger_cleanup()` always run, guaranteeing the async log queue is
    flushed before process exit
    
    ## Why It Works
    
    The Router's recv loops (`_networking_recv`,
    `_networking_recv_connection_messages`) are infinite `while True` loops
    calling Rust bindings with **no exception handling**. When the Rust
    networking layer raises `ConnectionError` (e.g. channel closed), the
    exception cascades silently through anyio task groups: Router → Node →
    all components shut down. The exception was never logged because (1) no
    try-except anywhere in the cascade, and (2) `logger_cleanup()` was
    skipped when `anyio.run()` raised, so loguru's async queue was never
    flushed.
    
    ## Test Plan
    
    ### Manual Testing
    - Run `uv run exo`, send SIGTERM → should see clean shutdown with "EXO
    Shutdown complete", no error/critical logs
    - Next time the silent shutdown reproduces, logs should now show the
    actual exception with full traceback at ERROR level from the recv loop,
    plus CRITICAL level from main()
    
    ### Automated Testing
    - All 222 existing tests pass (1 pre-existing failure in
    `test_python.py::test_sleep_on_multiple_items` unrelated to this change)
    - `uv run basedpyright` — 0 errors
    - `uv run ruff check` — all checks passed
    - `nix fmt` — applied
    
    🤖 Generated with [Claude Code](https://claude.com/claude-code)
    
    Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
---
 src/exo/main.py           | 13 +++++++++---
 src/exo/routing/router.py | 50 ++++++++++++++++++++++++++++++-----------------
 2 files changed, 42 insertions(+), 21 deletions(-)

diff --git a/src/exo/main.py b/src/exo/main.py
index 1735a9eb..b27b108a 100644
--- a/src/exo/main.py
+++ b/src/exo/main.py
@@ -270,9 +270,16 @@ def main():
         logger.info("FAST_SYNCH forced OFF")
 
     node = anyio.run(Node.create, args)
-    anyio.run(node.run)
-    logger.info("EXO Shutdown complete")
-    logger_cleanup()
+    try:
+        anyio.run(node.run)
+    except BaseException as exception:
+        logger.opt(exception=exception).critical(
+            "EXO terminated due to unhandled exception"
+        )
+        raise
+    finally:
+        logger.info("EXO Shutdown complete")
+        logger_cleanup()
 
 
 class Args(CamelCaseModel):
diff --git a/src/exo/routing/router.py b/src/exo/routing/router.py
index 9e84b83c..1950e4c8 100644
--- a/src/exo/routing/router.py
+++ b/src/exo/routing/router.py
@@ -174,28 +174,42 @@ class Router:
         logger.info(f"Unsubscribed from {topic}")
 
     async def _networking_recv(self):
-        while True:
-            topic, data = await self._net.gossipsub_recv()
-            logger.trace(f"Received message on {topic} with payload {data}")
-            if topic not in self.topic_routers:
-                logger.warning(f"Received message on unknown or inactive topic {topic}")
-                continue
+        try:
+            while True:
+                topic, data = await self._net.gossipsub_recv()
+                logger.trace(f"Received message on {topic} with payload {data}")
+                if topic not in self.topic_routers:
+                    logger.warning(
+                        f"Received message on unknown or inactive topic {topic}"
+                    )
+                    continue
 
-            router = self.topic_routers[topic]
-            await router.publish_bytes(data)
+                router = self.topic_routers[topic]
+                await router.publish_bytes(data)
+        except Exception as exception:
+            logger.opt(exception=exception).error(
+                "Gossipsub receive loop terminated unexpectedly"
+            )
+            raise
 
     async def _networking_recv_connection_messages(self):
-        while True:
-            update = await self._net.connection_update_recv()
-            message = ConnectionMessage.from_update(update)
-            logger.trace(
-                f"Received message on connection_messages with payload {message}"
+        try:
+            while True:
+                update = await self._net.connection_update_recv()
+                message = ConnectionMessage.from_update(update)
+                logger.trace(
+                    f"Received message on connection_messages with payload {message}"
+                )
+                if CONNECTION_MESSAGES.topic in self.topic_routers:
+                    router = self.topic_routers[CONNECTION_MESSAGES.topic]
+                    assert router.topic.model_type == ConnectionMessage
+                    router = cast(TopicRouter[ConnectionMessage], router)
+                    await router.publish(message)
+        except Exception as exception:
+            logger.opt(exception=exception).error(
+                "Connection update receive loop terminated unexpectedly"
             )
-            if CONNECTION_MESSAGES.topic in self.topic_routers:
-                router = self.topic_routers[CONNECTION_MESSAGES.topic]
-                assert router.topic.model_type == ConnectionMessage
-                router = cast(TopicRouter[ConnectionMessage], router)
-                await router.publish(message)
+            raise
 
     async def _networking_publish(self):
         with self.networking_receiver as networked_items:

← 76608935 fix: replace flaky internet checks with explicit offline mod  ·  back to Exo  ·  Address Mac Mini pipeline GPU timeouts (#1620) e23c3a30 →