← back to Exo
fix: handle BrokenResourceError in download progress callback (#1846)
ee2e505b3c6d9a94b5f0fac76736d925e3fc6c70 · 2026-04-10 17:43:42 +0600 · kaiisfree
## Summary
- Wraps progress callback `send()` in try/except to gracefully handle
`BrokenResourceError` when the memory stream is closed
- Prevents unhandled `ExceptionGroup` from crashing the process when the
download consumer disconnects during transfer
## Root Cause
The download progress callback sends updates through an anyio memory
object stream. When the receiving end closes (e.g., client disconnect,
timeout, or task cancellation), `send()` raises `BrokenResourceError`.
Inside an anyio `TaskGroup`, this unhandled exception becomes an
`ExceptionGroup` that propagates up and crashes the coordinator.
## Fix
Catch `BrokenResourceError` (and `ClosedResourceError` for completeness)
in the progress callback and handle gracefully — the download continues
but progress updates are silently dropped for disconnected consumers.
Fixes #1844
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Files touched
M src/exo/download/coordinator.py
Diff
commit ee2e505b3c6d9a94b5f0fac76736d925e3fc6c70
Author: kaiisfree <letkaibefree@yahoo.com>
Date: Fri Apr 10 17:43:42 2026 +0600
fix: handle BrokenResourceError in download progress callback (#1846)
## Summary
- Wraps progress callback `send()` in try/except to gracefully handle
`BrokenResourceError` when the memory stream is closed
- Prevents unhandled `ExceptionGroup` from crashing the process when the
download consumer disconnects during transfer
## Root Cause
The download progress callback sends updates through an anyio memory
object stream. When the receiving end closes (e.g., client disconnect,
timeout, or task cancellation), `send()` raises `BrokenResourceError`.
Inside an anyio `TaskGroup`, this unhandled exception becomes an
`ExceptionGroup` that propagates up and crashes the coordinator.
## Fix
Catch `BrokenResourceError` (and `ClosedResourceError` for completeness)
in the progress callback and handle gracefully — the download continues
but progress updates are silently dropped for disconnected consumers.
Fixes #1844
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---
src/exo/download/coordinator.py | 69 ++++++++++++++++++++++-------------------
1 file changed, 37 insertions(+), 32 deletions(-)
diff --git a/src/exo/download/coordinator.py b/src/exo/download/coordinator.py
index 39741d94..bfb9ddf8 100644
--- a/src/exo/download/coordinator.py
+++ b/src/exo/download/coordinator.py
@@ -4,7 +4,7 @@ from dataclasses import dataclass, field
from pathlib import Path
import anyio
-from anyio import current_time, to_thread
+from anyio import BrokenResourceError, ClosedResourceError, current_time, to_thread
from loguru import logger
from exo.download.download_utils import (
@@ -86,42 +86,47 @@ class DownloadCoordinator:
model_id = callback_shard.model_card.model_id
throttle_interval_secs = 1.0
- if progress.status == "complete":
- found = await to_thread.run_sync(resolve_existing_model, model_id)
- if found is not None:
- completed = self._completed_from_path(
- callback_shard, found, progress.total
+ try:
+ if progress.status == "complete":
+ found = await to_thread.run_sync(resolve_existing_model, model_id)
+ if found is not None:
+ completed = self._completed_from_path(
+ callback_shard, found, progress.total
+ )
+ else:
+ completed = DownloadCompleted(
+ shard_metadata=callback_shard,
+ node_id=self.node_id,
+ total=progress.total,
+ model_directory=self._default_model_dir(model_id),
+ )
+ self.download_status[model_id] = completed
+ await self.event_sender.send(
+ NodeDownloadProgress(download_progress=completed)
)
- else:
- completed = DownloadCompleted(
- shard_metadata=callback_shard,
+ self._last_progress_time.pop(model_id, None)
+ elif (
+ progress.status == "in_progress"
+ and current_time() - self._last_progress_time.get(model_id, 0.0)
+ > throttle_interval_secs
+ ):
+ ongoing = DownloadOngoing(
node_id=self.node_id,
- total=progress.total,
+ shard_metadata=callback_shard,
+ download_progress=map_repo_download_progress_to_download_progress_data(
+ progress
+ ),
model_directory=self._default_model_dir(model_id),
)
- self.download_status[model_id] = completed
- await self.event_sender.send(
- NodeDownloadProgress(download_progress=completed)
- )
- self._last_progress_time.pop(model_id, None)
- elif (
- progress.status == "in_progress"
- and current_time() - self._last_progress_time.get(model_id, 0.0)
- > throttle_interval_secs
- ):
- ongoing = DownloadOngoing(
- node_id=self.node_id,
- shard_metadata=callback_shard,
- download_progress=map_repo_download_progress_to_download_progress_data(
- progress
- ),
- model_directory=self._default_model_dir(model_id),
- )
- self.download_status[model_id] = ongoing
- await self.event_sender.send(
- NodeDownloadProgress(download_progress=ongoing)
+ self.download_status[model_id] = ongoing
+ await self.event_sender.send(
+ NodeDownloadProgress(download_progress=ongoing)
+ )
+ self._last_progress_time[model_id] = current_time()
+ except (BrokenResourceError, ClosedResourceError):
+ logger.debug(
+ f"Event stream closed while sending download progress for {model_id}, skipping update"
)
- self._last_progress_time[model_id] = current_time()
async def run(self) -> None:
logger.info(
← f2e6b1ef prevent some crash loops (#1827)
·
back to Exo
·
Truncate long logs with repr (#1854) abd75ae0 →