fix(sse): the server's own error event ends the stream

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
 

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", {}))))