fix(sse): chain the underlying transport error on reconnect-budget exhaustion

When the reconnect budget was exhausted, both the sync and async SSE clients
raised a bare ServiceResponseError("SSE reconnect budget exhausted"), discarding
the last transient transport error that caused the final reconnect. The error
was caught and swallowed mid-stream, leaving callers without the root cause.

Capture the last transient error caught while streaming and chain it as the
exception cause (raise ... from last_error) when the budget is hit. The message
and exception type are unchanged, so existing matchers still pass; the cause now
carries the original transport failure for diagnosis.

Also adds the spec-listed tests the suite was missing: empty-id clearing of the
replay header, upward-only backoff jitter, async clean-EOF reconnect, and async
retry-hint-then-budget exhaustion.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: OmarAlJarrah

packages/dexpace-sdk-core/src/dexpace/sdk/core/http/sse/connection.py

@@ -186,15 +186,18 @@ def __exit__(
 
     def __iter__(self) -> Iterator[SseEvent]:
         failures = 0
+        last_error: BaseException | None = None
         try:
             while True:
                 response = self._connect(_resume_request(self._initial, self._last_event_id))
-                progressed = yield from self._stream(response)
+                progressed, error = yield from self._stream(response)
+                if error is not None:
+                    last_error = error
                 self.close()
                 if progressed:
                     failures = 0
                 if self._max_reconnects is not None and failures >= self._max_reconnects:
-                    raise ServiceResponseError("SSE reconnect budget exhausted")
+                    raise ServiceResponseError("SSE reconnect budget exhausted") from last_error
                 self._clock.sleep(
                     _next_backoff(
                         self._retry_base, failures, self._max_backoff, self._jitter, self._rand
@@ -212,22 +215,22 @@ def _connect(self, request: Request) -> Response:
             raise HttpResponseError(response=response)
         return response
 
-    def _stream(self, response: Response) -> Generator[SseEvent, None, bool]:
-        """Yield one connection's events; return whether any were yielded.
+    def _stream(
+        self, response: Response
+    ) -> Generator[SseEvent, None, tuple[bool, BaseException | None]]:
+        """Yield one connection's events; return (progressed, transient_error).
 
-        A transient transport error mid-stream ends the generator normally
-        (so the caller reconnects); other exceptions propagate.
-
-        ``retry:`` directives that arrive without an accompanying ``data:``
-        field (so no ``SseEvent`` is emitted) are still picked up from the
-        parser's accumulated state after iteration ends, ensuring the
-        server-supplied reconnect delay is honoured even on pure
-        retry-hint frames.
+        A transient transport error mid-stream ends the generator normally (so
+        the caller reconnects) and is returned so the caller can chain it as the
+        cause if the reconnect budget is later exhausted; other exceptions
+        propagate. A ``retry:`` hint that arrived without a data event is picked
+        up from the parser's sticky state after iteration.
         """
         progressed = False
+        error: BaseException | None = None
         body = response.body
         if body is None:
-            return progressed
+            return progressed, error
         parser = SseParser()
         try:
             for chunk in body.iter_bytes():
@@ -240,13 +243,11 @@ def _stream(self, response: Response) -> Generator[SseEvent, None, bool]:
                 self._last_event_id = _last_event_id_of(event, self._last_event_id)
                 progressed = True
                 yield event
-        except _TRANSIENT:
-            pass
-        # Capture the server's sticky reconnect hint (including a retry:-only frame
-        # that emitted no event) for the next reconnection's backoff base.
+        except _TRANSIENT as exc:
+            error = exc
         if parser.retry is not None:
             self._retry_base = parser.retry / 1000.0
-        return progressed
+        return progressed, error
 
 
 class AsyncSseConnection:
@@ -350,6 +351,7 @@ def __aiter__(self) -> AsyncIterator[SseEvent]:
 
     async def _events(self) -> AsyncIterator[SseEvent]:
         failures = 0
+        last_error: BaseException | None = None
         try:
             while True:
                 response = await self._connect(_resume_request(self._initial, self._last_event_id))
@@ -363,15 +365,15 @@ async def _events(self) -> AsyncIterator[SseEvent]:
                                 self._last_event_id = _last_event_id_of(event, self._last_event_id)
                                 progressed = True
                                 yield event
-                    except _TRANSIENT:
-                        pass
+                    except _TRANSIENT as exc:
+                        last_error = exc
                     if stream.retry is not None:
                         self._retry_base = stream.retry / 1000.0
                 await self.aclose()
                 if progressed:
                     failures = 0
                 if self._max_reconnects is not None and failures >= self._max_reconnects:
-                    raise ServiceResponseError("SSE reconnect budget exhausted")
+                    raise ServiceResponseError("SSE reconnect budget exhausted") from last_error
                 await self._clock.sleep(
                     _next_backoff(
                         self._retry_base, failures, self._max_backoff, self._jitter, self._rand

packages/dexpace-sdk-core/tests/sse/test_async_connection.py

@@ -145,3 +145,37 @@ async def consume() -> None:
 
     assert seen == ["hi"]
     assert body.closed is True
+
+
+async def test_async_reconnects_after_clean_eof() -> None:
+    first = _AsyncDropBody([b"data: a\n\n"])
+    second = _AsyncDropBody([b"data: b\n\n"])
+    script = _AsyncScript([_response(first), _response(second)])
+    conn = AsyncSseConnection(script, _request(), clock=_RecordingAsyncClock(), rand=_LowJitter())
+
+    received: list[str] = []
+    async with conn as events:
+        async for event in events:
+            received.append(event.data)
+            if len(received) == 2:
+                break
+
+    assert received == ["a", "b"]
+    assert first.closed is True
+
+
+async def test_async_honours_retry_then_budget() -> None:
+    retry_then_drop = _response(
+        _AsyncDropBody([b"retry: 1000\n\n"], error=ServiceResponseError("x"))
+    )
+    drop1 = _response(_AsyncDropBody([], error=ServiceResponseError("x")))
+    drop2 = _response(_AsyncDropBody([], error=ServiceResponseError("x")))
+    script = _AsyncScript([retry_then_drop, drop1, drop2])
+    clock = _RecordingAsyncClock()
+    conn = AsyncSseConnection(script, _request(), clock=clock, rand=_LowJitter(), max_reconnects=2)
+
+    with pytest.raises(ServiceResponseError, match="reconnect budget"):
+        async for _ in conn:
+            pass
+
+    assert clock.sleeps == [1.0, 2.0]

packages/dexpace-sdk-core/tests/sse/test_connection.py

@@ -90,6 +90,13 @@ def uniform(self, a: float, b: float) -> float:
         return a
 
 
+class _HighJitter(random.Random):
+    """``uniform`` returns its high bound — exercises upward jitter."""
+
+    def uniform(self, a: float, b: float) -> float:
+        return b
+
+
 def test_yields_events_and_closes_on_caller_stop() -> None:
     body = _DropBody([b"data: one\n\n", b"data: two\n\n"])
     script = _Script([_response(body)])
@@ -190,3 +197,59 @@ def test_non_success_status_raises_without_reconnect() -> None:
 
     # No reconnect attempted: send called exactly once.
     assert len(script.requests) == 1
+
+
+def test_empty_id_clears_replay_header() -> None:
+    # id:5 then an explicit empty id (clears), so the reconnect omits the header.
+    first = _DropBody([b"id: 5\ndata: a\n\nid:\ndata: b\n\n"], error=ServiceResponseError("x"))
+    second = _DropBody([b"data: c\n\n"])
+    script = _Script([_response(first), _response(second)])
+    conn = SseConnection(script, _request(), clock=_RecordingClock(), rand=_LowJitter())
+
+    received: list[str] = []
+    with conn as events:
+        for event in events:
+            received.append(event.data)
+            if len(received) == 3:
+                break
+
+    assert received == ["a", "b", "c"]
+    assert script.requests[1].headers.get("last-event-id") is None
+
+
+def test_backoff_jitter_is_upward() -> None:
+    first = _DropBody([b"data: a\n\n"], error=ServiceResponseError("x"))
+    second = _DropBody([b"data: b\n\n"])
+    script = _Script([_response(first), _response(second)])
+    clock = _RecordingClock()
+    conn = SseConnection(script, _request(), clock=clock, rand=_HighJitter(), jitter=0.1)
+
+    received: list[str] = []
+    with conn as events:
+        for event in events:
+            received.append(event.data)
+            if len(received) == 2:
+                break
+
+    # First connection progressed -> failures reset to 0 -> base 3.0; high
+    # jitter multiplies by 1.0 + 0.1, never below the base.
+    assert clock.sleeps == [pytest.approx(3.3)]
+    assert clock.sleeps[0] > 3.0
+
+
+def test_budget_exhaustion_chains_last_transport_error() -> None:
+    boom = ServiceResponseError("connection reset")
+    script = _Script(
+        [
+            _response(_DropBody([], error=boom)),
+            _response(_DropBody([], error=boom)),
+        ]
+    )
+    conn = SseConnection(
+        script, _request(), clock=_RecordingClock(), rand=_LowJitter(), max_reconnects=1
+    )
+
+    with pytest.raises(ServiceResponseError, match="reconnect budget") as exc_info:
+        for _ in conn:
+            pass
+    assert exc_info.value.__cause__ is boom