[object Object]

← back to Exo

Catch ClosedResourceError when forwarding chunks to client queues (#1856)

62570227ff74d5e88ce0795e399d2e68f9103838 · 2026-04-08 11:50:04 +0300 · mlpy0

While stress-testing inference with rapid client cancels mid-stream, I
hit a reproducible crash where the entire exo process exits.

When a client cancels a streaming chat completion partway through, its
receive stream gets closed cleanly via its context manager. The producer
in `API._apply_state` then calls `queue.send(event.chunk)`, which raises
`anyio.ClosedResourceError` rather than `BrokenResourceError`. The
existing handler only catches `BrokenResourceError`, so the exception
propagates through the API task group, kills the Node task group, and
the process exits with `EXO Shutdown complete`.

Trace from one of the crashes:

```
File "exo/api/main.py", line 1818, in _apply_state
    await queue.send(event.chunk)
File "anyio/streams/memory.py", line 212, in send_nowait
    raise ClosedResourceError
anyio.ClosedResourceError
```

The fix is to catch `ClosedResourceError` alongside
`BrokenResourceError` in both queue handlers (text and image), so the
dead queue gets dropped and `_apply_state` keeps running for other
in-flight requests.

Files touched

Diff

commit 62570227ff74d5e88ce0795e399d2e68f9103838
Author: mlpy0 <96022931+mlpy0@users.noreply.github.com>
Date:   Wed Apr 8 11:50:04 2026 +0300

    Catch ClosedResourceError when forwarding chunks to client queues (#1856)
    
    While stress-testing inference with rapid client cancels mid-stream, I
    hit a reproducible crash where the entire exo process exits.
    
    When a client cancels a streaming chat completion partway through, its
    receive stream gets closed cleanly via its context manager. The producer
    in `API._apply_state` then calls `queue.send(event.chunk)`, which raises
    `anyio.ClosedResourceError` rather than `BrokenResourceError`. The
    existing handler only catches `BrokenResourceError`, so the exception
    propagates through the API task group, kills the Node task group, and
    the process exits with `EXO Shutdown complete`.
    
    Trace from one of the crashes:
    
    ```
    File "exo/api/main.py", line 1818, in _apply_state
        await queue.send(event.chunk)
    File "anyio/streams/memory.py", line 212, in send_nowait
        raise ClosedResourceError
    anyio.ClosedResourceError
    ```
    
    The fix is to catch `ClosedResourceError` alongside
    `BrokenResourceError` in both queue handlers (text and image), so the
    dead queue gets dropped and `_apply_state` keeps running for other
    in-flight requests.
---
 src/exo/api/main.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/exo/api/main.py b/src/exo/api/main.py
index 685281a3..3c183f4f 100644
--- a/src/exo/api/main.py
+++ b/src/exo/api/main.py
@@ -12,7 +12,7 @@ from typing import Annotated, Any, Literal, cast
 from uuid import uuid4
 
 import anyio
-from anyio import BrokenResourceError
+from anyio import BrokenResourceError, ClosedResourceError
 from fastapi import FastAPI, File, Form, HTTPException, Query, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
@@ -1808,7 +1808,7 @@ class API:
                         assert isinstance(event.chunk, ImageChunk)
                         try:
                             await queue.send(event.chunk)
-                        except BrokenResourceError:
+                        except (BrokenResourceError, ClosedResourceError):
                             self._image_generation_queues.pop(event.command_id, None)
                     if queue := self._text_generation_queues.get(
                         event.command_id, None
@@ -1816,7 +1816,7 @@ class API:
                         assert not isinstance(event.chunk, ImageChunk)
                         try:
                             await queue.send(event.chunk)
-                        except BrokenResourceError:
+                        except (BrokenResourceError, ClosedResourceError):
                             self._text_generation_queues.pop(event.command_id, None)
                 if isinstance(event, TracesMerged):
                     self._save_merged_trace(event)

← 645bc209 Add Fast Synch Enabled toggle to macOS app settings (#1852)  ·  back to Exo  ·  Cancel SSE keep-alive when instance is deleted (#1828) b12cd1b1 →