fix(transport): classify timeouts per phase and map streamed body errors

Three adapter inconsistencies in how a request-versus-response failure is
reported, which matters because the retry policy must never blind-retry a
non-idempotent request whose response may already be in flight:

- The aiohttp client configured only a total timeout, so a connect-phase
  stall raised a bare TimeoutError indistinguishable from a read timeout and
  was misclassified as a response error. Configure sock_connect/sock_read so
  connect raises ConnectionTimeoutError (request error) and read raises
  SocketTimeoutError (response error), matching the per-operation timeouts
  the other adapters already use.
- The requests client did not wrap response-body streaming, so a mid-body
  ChunkedEncodingError, connection drop, or read timeout escaped as the raw
  library exception. Map a body-read Timeout to ServiceResponseTimeoutError
  and any other RequestException to ServiceResponseError.
- Document why the asyncio reference client deliberately classifies a
  post-connect OSError as a response error rather than the request error the
  other adapters use for their generic transport bucket.

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

Author: OmarAlJarrah

packages/dexpace-sdk-http-aiohttp/src/dexpace/sdk/http/aiohttp/client.py

@@ -53,8 +53,10 @@ class AiohttpHttpClient:
     components; the caller is then responsible for closing it.
 
     Attributes:
-        timeout: Total request timeout in seconds. Applied via
-            ``aiohttp.ClientTimeout(total=...)``. ``None`` disables the
+        timeout: Per-phase request timeout in seconds, applied to both the
+            connect and the socket-read phases via
+            ``aiohttp.ClientTimeout(sock_connect=..., sock_read=...)`` so the
+            two phases raise distinguishable exceptions. ``None`` disables the
             timeout entirely (not recommended).
     """
 
@@ -77,8 +79,15 @@ async def execute(self, request: Request) -> AsyncResponse:
         if self._closed:
             raise ServiceRequestError("AiohttpHttpClient is closed")
         session = await self._ensure_session()
+        # Per-phase budgets (not a single ``total=``) so aiohttp raises the
+        # distinguishable ``ConnectionTimeoutError`` for a connect-phase stall
+        # and ``SocketTimeoutError`` for a read-phase stall. This keeps the
+        # connect -> request-error / read -> response-error split consistent
+        # with the other transports, which all use per-operation timeouts.
         timeout_cfg = (
-            aiohttp.ClientTimeout(total=self.timeout) if self.timeout is not None else None
+            aiohttp.ClientTimeout(sock_connect=self.timeout, sock_read=self.timeout)
+            if self.timeout is not None
+            else None
         )
         data = _payload(request.body)
         try:

packages/dexpace-sdk-http-aiohttp/tests/test_aiohttp_client.py

@@ -175,7 +175,7 @@ async def test_content_length_extracted_from_response(base_url: str) -> None:
             assert response.body.content_length() > 0
 
 
-# ----------------------------------------------------------- unknown status (M21)
+# ----------------------------------------------------------------- unknown status
 
 
 class _FakeAioResponse:
@@ -201,7 +201,7 @@ def test_unknown_status_releases_connection_and_raises() -> None:
     assert "520" in str(exc_info.value)
 
 
-# ----------------------------------------------------------- post-close (M22)
+# ----------------------------------------------------------------- post-close
 
 
 async def test_execute_after_aclose_raises() -> None:
@@ -221,18 +221,17 @@ async def test_aclose_is_idempotent() -> None:
         await client.execute(Request(method=Method.GET, url=Url.parse("http://example.test/")))
 
 
-# ----------------------------------------------------- connect timeout (L36)
+# ----------------------------------------------------------- connect timeout
 
 
 class _ConnectTimeoutSession:
     """Stub session whose ``request`` raises aiohttp's connect-phase timeout.
 
-    aiohttp only surfaces ``ConnectionTimeoutError`` for a connect-scoped
-    timeout (``connect=`` / ``sock_connect=``); a plain ``total=`` timeout that
-    happens to expire mid-connect raises a bare ``TimeoutError`` instead, which
-    is indistinguishable from a read-phase timeout. We therefore drive the
-    branch directly with the exception aiohttp actually raises for a connect
-    timeout, keeping the test hermetic.
+    aiohttp raises ``ConnectionTimeoutError`` for a connect-scoped timeout
+    (``connect=`` / ``sock_connect=``) and ``SocketTimeoutError`` for a read
+    timeout (``sock_read=``); the client configures both so the two phases stay
+    distinguishable. We drive the connect branch directly with the exception
+    aiohttp raises so the test stays hermetic (no real unreachable-host connect).
     """
 
     def request(self, **_kwargs: object) -> _ConnectTimeoutSession:
@@ -250,8 +249,43 @@ async def test_connect_timeout_maps_to_request_timeout() -> None:
         await client.execute(Request(method=Method.GET, url=Url.parse("http://example.test/")))
 
 
-async def test_total_timeout_maps_to_response_timeout(base_url: str) -> None:
-    """A bare total timeout (read phase) still maps to ServiceResponseTimeoutError."""
+class _CaptureTimeoutSession:
+    """Captures the ``ClientTimeout`` passed to ``request`` then fails the connect."""
+
+    def __init__(self) -> None:
+        self.captured: aiohttp.ClientTimeout | None = None
+
+    def request(
+        self, *, timeout: aiohttp.ClientTimeout, **_kwargs: object
+    ) -> _CaptureTimeoutSession:
+        self.captured = timeout
+        return self
+
+    def __await__(self) -> object:
+        raise aiohttp.ConnectionTimeoutError("connect timed out")
+        yield  # pragma: no cover - makes this an awaitable generator
+
+
+async def test_timeout_configured_per_phase_so_connect_is_distinguishable() -> None:
+    """The client asks aiohttp for per-phase sock_connect/sock_read, not a total budget.
+
+    A total-only budget makes a connect-phase timeout raise a bare
+    ``TimeoutError`` indistinguishable from a read timeout; per-phase config
+    makes connect raise ``ConnectionTimeoutError`` so it maps to a request error.
+    """
+    session = _CaptureTimeoutSession()
+    client = AiohttpHttpClient(timeout=5.0, session=session)  # type: ignore[arg-type]
+    with pytest.raises(ServiceRequestTimeoutError):
+        await client.execute(Request(method=Method.GET, url=Url.parse("http://example.test/")))
+    cfg = session.captured
+    assert cfg is not None
+    assert cfg.sock_connect == 5.0
+    assert cfg.sock_read == 5.0
+    assert cfg.total is None
+
+
+async def test_read_timeout_maps_to_response_timeout(base_url: str) -> None:
+    """A read-phase (sock_read) timeout maps to ServiceResponseTimeoutError."""
     async with AiohttpHttpClient(timeout=0.25) as client:
         with pytest.raises(ServiceResponseTimeoutError):
             await client.execute(Request(method=Method.GET, url=Url.parse(f"{base_url}/slow")))

packages/dexpace-sdk-http-requests/src/dexpace/sdk/http/requests/client.py

@@ -14,6 +14,11 @@
 - ``requests.ReadTimeout`` -> `ServiceResponseTimeoutError`
 - ``requests.ConnectionError`` -> `ServiceRequestError`
 - ``requests.RequestException`` (catch-all) -> `ServiceRequestError`
+
+Failures that surface later, while the response body is being streamed, are
+classified as response-side errors (the request was already sent): a
+``requests.Timeout`` -> `ServiceResponseTimeoutError` and any other
+``requests.RequestException`` -> `ServiceResponseError`.
 """
 
 from __future__ import annotations
@@ -184,16 +189,15 @@ def read(self, size: int = -1) -> bytes:
         if self._iter is None:
             self._iter = self._response.iter_content(chunk_size=_CHUNK_SIZE)
         if size < 0:
-            for chunk in self._iter:
+            while (chunk := self._next_chunk()) is not None:
                 if chunk:
                     self._buf.extend(chunk)
             out = bytes(self._buf)
             self._buf.clear()
             return out
         while len(self._buf) < size:
-            try:
-                chunk = next(self._iter)
-            except StopIteration:
+            chunk = self._next_chunk()
+            if chunk is None:
                 break
             if chunk:
                 self._buf.extend(chunk)
@@ -204,6 +208,24 @@ def read(self, size: int = -1) -> bytes:
         del self._buf[:take]
         return out
 
+    def _next_chunk(self) -> bytes | None:
+        """Pull the next body chunk, mapping read-phase failures to SDK errors.
+
+        Returns the chunk, or ``None`` at end of stream. The request is already
+        on the wire, so a read-phase failure is a response-side error: a
+        ``requests`` read timeout becomes ``ServiceResponseTimeoutError`` and
+        any other transport failure mid-body becomes ``ServiceResponseError``.
+        """
+        assert self._iter is not None
+        try:
+            return next(self._iter)
+        except StopIteration:
+            return None
+        except requests.Timeout as err:
+            raise ServiceResponseTimeoutError("Response body read timed out", error=err) from err
+        except requests.RequestException as err:
+            raise ServiceResponseError(f"Response body read failed: {err}", error=err) from err
+
     def close(self) -> None:
         if self._closed:
             return

packages/dexpace-sdk-http-requests/tests/test_requests_client.py

@@ -236,3 +236,63 @@ def test_unknown_status_closes_response() -> None:
     with client, pytest.raises(ServiceResponseError):
         client.execute(request)
     assert closed["yes"], "Response should be closed when status mapping fails"
+
+
+class _BodyFailureAdapter(requests.adapters.BaseAdapter):
+    """Returns a 200 whose body stream raises ``exc`` partway through the read."""
+
+    def __init__(self, exc: Exception) -> None:
+        super().__init__()
+        self._exc = exc
+
+    def send(  # signature mirrors BaseAdapter.send
+        self,
+        request: requests.PreparedRequest,
+        stream: bool = False,
+        timeout: float | tuple[float | None, float | None] | None = None,
+        verify: bool | str = True,
+        cert: str | tuple[str, str] | None = None,
+        proxies: dict[str, str] | None = None,
+    ) -> requests.Response:
+        response = requests.Response()
+        response.status_code = 200
+        response.reason = "OK"
+        response.url = request.url or ""
+        response.raw = io.BytesIO(b"")
+        exc = self._exc
+
+        def _raising_iter(chunk_size: int = 1, decode_unicode: bool = False) -> Iterator[bytes]:
+            yield b"partial"
+            raise exc
+
+        response.iter_content = _raising_iter  # type: ignore[method-assign, assignment]
+        return response
+
+    def close(self) -> None:  # pragma: no cover - nothing to release
+        pass
+
+
+def test_body_read_chunked_error_maps_to_service_response_error() -> None:
+    """A mid-body ``ChunkedEncodingError`` surfaces as ServiceResponseError."""
+    session = requests.Session()
+    session.mount("http://", _BodyFailureAdapter(requests.exceptions.ChunkedEncodingError("boom")))
+    client = RequestsHttpClient(session=session)
+    request = Request(method=Method.GET, url=Url.parse("http://example.test/"))
+    with client:
+        response = client.execute(request)
+        assert response.body is not None
+        with pytest.raises(ServiceResponseError):
+            response.body.bytes()
+
+
+def test_body_read_timeout_maps_to_service_response_timeout_error() -> None:
+    """A mid-body ``ReadTimeout`` surfaces as ServiceResponseTimeoutError."""
+    session = requests.Session()
+    session.mount("http://", _BodyFailureAdapter(requests.exceptions.ReadTimeout("slow")))
+    client = RequestsHttpClient(session=session)
+    request = Request(method=Method.GET, url=Url.parse("http://example.test/"))
+    with client:
+        response = client.execute(request)
+        assert response.body is not None
+        with pytest.raises(ServiceResponseTimeoutError):
+            response.body.bytes()

packages/dexpace-sdk-http-stdlib/src/dexpace/sdk/http/stdlib/asyncio_http_client.py

@@ -114,7 +114,12 @@ async def execute(self, request: Request) -> AsyncResponse:
             raise
         except OSError as err:
             # A mid-exchange socket error after a successful connect is a
-            # response-side failure, not a connect failure.
+            # response-side failure, not a connect failure: the request may
+            # already be on the wire, so it is deliberately classified as a
+            # response error (not retry-safe) rather than the request error the
+            # httpx / aiohttp / requests adapters use for their generic
+            # transport bucket. The conservative choice avoids blind-retrying a
+            # potentially-received non-idempotent request.
             raise ServiceResponseError(
                 f"Exchange with {host}:{port} failed: {err}", error=err
             ) from err