From 12dced55feb7774127cb302bec256d5dcc711ef5 Mon Sep 17 00:00:00 2001 From: "Joseph T. French" Date: Fri, 13 Nov 2026 00:06:37 -0600 Subject: [PATCH] fix(sse): the server's own error event ends the stream A stream answered with `event: error` was reopened without limit, since the event counted as progress; a caller with no timeout never got a verdict. --- robosystems_client/clients/sse_client.py | 15 ++++++++-- tests/test_sse_client.py | 36 ++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/robosystems_client/clients/sse_client.py b/robosystems_client/clients/sse_client.py index 9ddbc0c..be75734 100644 --- a/robosystems_client/clients/sse_client.py +++ b/robosystems_client/clients/sse_client.py @@ -77,6 +77,9 @@ class EventType(Enum): # Sent on every open and while idle, so neither shows the stream got anywhere. _NO_PROGRESS_EVENTS = {"connected", "keepalive"} +# The server ends a stream it will not serve with an event of this name. +_SERVER_ERROR = "error" + class _StreamInterrupted(Exception): """The stream ended or failed before the server said it was finished.""" @@ -262,11 +265,15 @@ def _dispatch_event(self, event_buffer: Dict[str, Any]) -> None: self.emit(event_type, parsed_data) # Check for completion events - just set flag, don't close from within loop - # The loop will break on next iteration and close() will be called in finally + # The loop will break on next iteration and close() will be called in finally. + # The server's own `error` event is its answer for this stream (the + # per-user connection cap, a stream it refuses): reconnecting would only + # be answered the same way again. if event_type in [ EventType.OPERATION_COMPLETED.value, EventType.OPERATION_ERROR.value, EventType.OPERATION_CANCELLED.value, + _SERVER_ERROR, ]: self.closed = True @@ -523,11 +530,15 @@ def _dispatch_event(self, event_buffer: Dict[str, Any]): self.emit(event_type, parsed_data) # Check for completion events - just set flag, don't close from within loop - # The loop will break on next iteration and close() will be called in finally + # The loop will break on next iteration and close() will be called in finally. + # The server's own `error` event is its answer for this stream (the + # per-user connection cap, a stream it refuses): reconnecting would only + # be answered the same way again. if event_type in [ EventType.OPERATION_COMPLETED.value, EventType.OPERATION_ERROR.value, EventType.OPERATION_CANCELLED.value, + _SERVER_ERROR, ]: self.closed = True diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py index 2b6db38..a169433 100644 --- a/tests/test_sse_client.py +++ b/tests/test_sse_client.py @@ -658,6 +658,42 @@ def test_stream_end_stops_without_reconnecting(self, fast_config): assert len(seen) == 1 assert events["reconnecting"] == [] + def test_the_servers_own_error_is_final(self, fast_config): + """A stream the server answers with `event: error` is not reopened: every + reconnect would get the same answer, and a caller with no timeout would + never get a verdict.""" + body = _sse(("connected", {}), ("error", {"error": "Too many connections"})) + handler, seen = _responses(httpx.Response(200, content=body)) + + events = _run_sync(fast_config, handler) + + assert len(seen) == 1 + assert events["errors"] == [{"error": "Too many connections"}] + assert events["reconnecting"] == [] + assert events["exhausted"] == [] + + def test_the_servers_own_error_is_final_for_the_async_client(self, fast_config): + body = _sse(("connected", {}), ("error", {"error": "Too many connections"})) + handler, seen = _responses(httpx.Response(200, content=body)) + real_client = httpx.AsyncClient + errors = [] + + async def run(): + with patch( + "robosystems_client.clients.sse_client.httpx.AsyncClient", + side_effect=lambda **kw: real_client( + transport=httpx.MockTransport(handler), **kw + ), + ): + client = AsyncSSEClient(fast_config) + client.on("error", errors.append) + await client.connect("op_1") + + asyncio.run(run()) + + assert len(seen) == 1 + assert errors == [{"error": "Too many connections"}] + def test_reopening_without_progress_does_not_reset_the_budget(self, fast_config): handler, seen = _responses( httpx.Response(200, stream=_DroppingStream(_sse(("connected", {}))))