fix(transport): correct timeout classification and release connections on error

Classify connect-phase and read-phase timeouts consistently across the
adapters: a read timeout maps to a response timeout and a connect timeout to a
request timeout, so the retry policy charges them to the right budget and never
auto-retries a non-idempotent request whose response may already be in flight.
Map the asyncio transport's read-phase timeouts and stream errors into the SDK
error hierarchy instead of leaking raw stdlib exceptions, and stop overwriting a
caller-supplied Host header.

Release the underlying connection when a response carries a status code outside
the known set before raising, so an unrecognised code no longer leaks a pooled
connection, and make the aiohttp client refuse use after close instead of
silently opening a fresh session.

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

Author: OmarAlJarrah

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

@@ -1,16 +1,16 @@
 # Copyright (c) 2026 dexpace and Omar Aljarrah.
 # Licensed under the MIT License. See LICENSE.md in the repository root for details.
 
-"""``AsyncHttpClient`` implementation backed by :mod:`aiohttp`.
+"""``AsyncHttpClient`` implementation backed by `aiohttp`.
 
 ``aiohttp`` exposes only an async API; this package therefore ships an
 async transport without a sync twin. The adapter is a thin pass-through:
 
 - Request bodies are forwarded to ``aiohttp`` via an async-iterable shim
-  over :meth:`RequestBody.iter_bytes`, so uploads stream without buffering
+  over `RequestBody.iter_bytes`, so uploads stream without buffering
   the full payload into memory.
 - Response content streams through ``aiohttp.StreamReader``; we wrap it as
-  an :class:`AsyncResponseBody` so the SDK's body lifecycle (deferred
+  an `AsyncResponseBody` so the SDK's body lifecycle (deferred
   read, deterministic close) is preserved.
 - Transport exceptions are mapped to the SDK's typed error hierarchy.
 
@@ -27,6 +27,7 @@
 import aiohttp
 from dexpace.sdk.core.errors import (
     ServiceRequestError,
+    ServiceRequestTimeoutError,
     ServiceResponseError,
     ServiceResponseTimeoutError,
 )
@@ -45,7 +46,7 @@
 
 
 class AiohttpHttpClient:
-    """Async ``HttpClient`` over an :class:`aiohttp.ClientSession`.
+    """Async ``HttpClient`` over an `aiohttp.ClientSession`.
 
     The client owns the session by default and releases it on ``aclose``.
     Pass an existing ``session`` to share connection pooling with other
@@ -57,7 +58,7 @@ class AiohttpHttpClient:
             timeout entirely (not recommended).
     """
 
-    __slots__ = ("_owns_session", "_session", "_session_factory", "timeout")
+    __slots__ = ("_closed", "_owns_session", "_session", "timeout")
 
     def __init__(
         self,
@@ -70,9 +71,11 @@ def __init__(
         self.timeout = timeout
         self._session = session
         self._owns_session = session is None
-        self._session_factory = aiohttp.ClientSession
+        self._closed = False
 
     async def execute(self, request: Request) -> AsyncResponse:
+        if self._closed:
+            raise ServiceRequestError("AiohttpHttpClient is closed")
         session = await self._ensure_session()
         timeout_cfg = (
             aiohttp.ClientTimeout(total=self.timeout) if self.timeout is not None else None
@@ -90,6 +93,10 @@ async def execute(self, request: Request) -> AsyncResponse:
             aio_response = await ctx
         except aiohttp.ClientConnectorError as err:
             raise ServiceRequestError(f"Connect failed: {err}", error=err) from err
+        except aiohttp.ConnectionTimeoutError as err:
+            raise ServiceRequestTimeoutError(
+                f"Connection to {request.url} timed out", error=err
+            ) from err
         except TimeoutError as err:
             raise ServiceResponseTimeoutError(
                 f"Request to {request.url} timed out", error=err
@@ -101,6 +108,9 @@ async def execute(self, request: Request) -> AsyncResponse:
         return _wrap_response(request, aio_response)
 
     async def aclose(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
         if self._session is not None and self._owns_session:
             await self._session.close()
             self._session = None
@@ -119,7 +129,8 @@ async def __aexit__(
 
     async def _ensure_session(self) -> aiohttp.ClientSession:
         if self._session is None:
-            self._session = self._session_factory()
+            # Construct lazily so the session binds the running event loop.
+            self._session = aiohttp.ClientSession()
             self._owns_session = True
         return self._session
 
@@ -159,6 +170,9 @@ def _wrap_response(request: Request, aio_response: aiohttp.ClientResponse) -> As
     try:
         status = Status(aio_response.status)
     except ValueError as err:
+        # Release the handle before bailing so the connection returns to the
+        # pool instead of leaking (aiohttp's release() is synchronous).
+        aio_response.release()
         raise ServiceResponseError(
             f"Unknown status code: {aio_response.status}", error=err
         ) from err
@@ -179,7 +193,7 @@ def _wrap_response(request: Request, aio_response: aiohttp.ClientResponse) -> As
 
 
 class _StreamReaderAdapter:
-    """``SupportsAsyncRead`` adapter over an :class:`aiohttp.ClientResponse`.
+    """``SupportsAsyncRead`` adapter over an `aiohttp.ClientResponse`.
 
     Owns the response handle; closing the adapter releases the connection
     back to the pool.

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

@@ -13,13 +13,16 @@
 
 from dexpace.sdk.core.errors import (
     ServiceRequestError,
+    ServiceRequestTimeoutError,
+    ServiceResponseError,
     ServiceResponseTimeoutError,
 )
 from dexpace.sdk.core.http.common import Url
 from dexpace.sdk.core.http.request import Method, Request
 from dexpace.sdk.core.http.request.request_body import RequestBody
 from dexpace.sdk.core.http.response import Status
 from dexpace.sdk.http.aiohttp import AiohttpHttpClient
+from dexpace.sdk.http.aiohttp.client import _wrap_response
 
 # ---------------------------------------------------------------------- handlers
 
@@ -170,3 +173,85 @@ async def test_content_length_extracted_from_response(base_url: str) -> None:
             assert response.body is not None
             # /ok returns a small JSON; Content-Length should be set by aiohttp.
             assert response.body.content_length() > 0
+
+
+# ----------------------------------------------------------- unknown status (M21)
+
+
+class _FakeAioResponse:
+    """Minimal stand-in for an ``aiohttp.ClientResponse`` carrying a raw status."""
+
+    def __init__(self, status: int) -> None:
+        self.status = status
+        self.released = False
+
+    def release(self) -> None:  # aiohttp's release() is synchronous
+        self.released = True
+
+
+def test_unknown_status_releases_connection_and_raises() -> None:
+    """An unregistered status code must release the response before raising."""
+    request = Request(method=Method.GET, url=Url.parse("http://example.test/"))
+    fake = _FakeAioResponse(520)  # Cloudflare 'Web Server Returned an Unknown Error'
+
+    with pytest.raises(ServiceResponseError) as exc_info:
+        _wrap_response(request, fake)  # type: ignore[arg-type]
+
+    assert fake.released, "connection leaked: release() was not called"
+    assert "520" in str(exc_info.value)
+
+
+# ----------------------------------------------------------- post-close (M22)
+
+
+async def test_execute_after_aclose_raises() -> None:
+    """A closed client must not resurrect; execute() raises ServiceRequestError."""
+    client = AiohttpHttpClient(timeout=5.0)
+    await client.aclose()
+    with pytest.raises(ServiceRequestError, match="closed"):
+        await client.execute(Request(method=Method.GET, url=Url.parse("http://example.test/")))
+
+
+async def test_aclose_is_idempotent() -> None:
+    """Calling aclose() twice is a no-op and stays in the closed state."""
+    client = AiohttpHttpClient(timeout=5.0)
+    await client.aclose()
+    await client.aclose()  # must not raise
+    with pytest.raises(ServiceRequestError, match="closed"):
+        await client.execute(Request(method=Method.GET, url=Url.parse("http://example.test/")))
+
+
+# ----------------------------------------------------- connect timeout (L36)
+
+
+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.
+    """
+
+    def request(self, **_kwargs: object) -> _ConnectTimeoutSession:
+        return self
+
+    def __await__(self) -> object:
+        raise aiohttp.ConnectionTimeoutError("connect timed out")
+        yield  # pragma: no cover - makes this an awaitable generator
+
+
+async def test_connect_timeout_maps_to_request_timeout() -> None:
+    """A connect-phase timeout maps to ServiceRequestTimeoutError, not a response timeout."""
+    client = AiohttpHttpClient(timeout=5.0, session=_ConnectTimeoutSession())  # type: ignore[arg-type]
+    with pytest.raises(ServiceRequestTimeoutError):
+        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."""
+    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-httpx/src/dexpace/sdk/http/httpx/async_.py

@@ -1,15 +1,15 @@
 # Copyright (c) 2026 dexpace and Omar Aljarrah.
 # Licensed under the MIT License. See LICENSE.md in the repository root for details.
 
-"""Async ``HttpClient`` implementation built on :class:`httpx.AsyncClient`.
+"""Async ``HttpClient`` implementation built on `httpx.AsyncClient`.
 
 Supports streaming uploads and downloads, per-phase timeouts (connect /
-read / write / pool), HTTP/2 (opt-in via :mod:`httpx`), and proxies.
+read / write / pool), HTTP/2 (opt-in via `httpx`), and proxies.
 Production-grade alternative to the asyncio reference client shipped in
 ``dexpace-sdk-http-stdlib``.
 
-The transport delegates to :class:`httpx.AsyncClient`, streaming the
-request body via :meth:`AsyncRequestBody.aiter_bytes` (when provided) or
+The transport delegates to `httpx.AsyncClient`, streaming the
+request body via `AsyncRequestBody.aiter_bytes` (when provided) or
 the sync ``RequestBody.iter_bytes`` (drained eagerly into bytes; httpx
 accepts both). The response exposes ``AsyncResponseBody.from_async_stream``
 wrapping httpx's ``aiter_bytes`` iterator.
@@ -40,11 +40,11 @@
 
 
 class AsyncHttpxHttpClient:
-    """Async transport over :class:`httpx.AsyncClient`.
+    """Async transport over `httpx.AsyncClient`.
 
     Per-phase timeouts (``connect_timeout``, ``read_timeout``,
     ``write_timeout``, ``pool_timeout``) are forwarded to
-    :class:`httpx.Timeout`. ``None`` disables the phase's timeout.
+    `httpx.Timeout`. ``None`` disables the phase's timeout.
 
     Args:
         connect_timeout: Seconds allowed for connection establishment.
@@ -54,9 +54,9 @@ class AsyncHttpxHttpClient:
             request body.
         pool_timeout: Seconds allowed to acquire a connection from the
             pool.
-        transport: Optional :class:`httpx.AsyncBaseTransport`; the primary
+        transport: Optional `httpx.AsyncBaseTransport`; the primary
             extension hook for tests (``httpx.MockTransport``).
-        client: Optional pre-built :class:`httpx.AsyncClient` — overrides
+        client: Optional pre-built `httpx.AsyncClient` — overrides
             the timeout / transport kwargs entirely. Ownership transfers
             to this transport.
     """

packages/dexpace-sdk-http-httpx/src/dexpace/sdk/http/httpx/sync.py

@@ -1,16 +1,16 @@
 # Copyright (c) 2026 dexpace and Omar Aljarrah.
 # Licensed under the MIT License. See LICENSE.md in the repository root for details.
 
-"""Synchronous ``HttpClient`` implementation built on :mod:`httpx`.
+"""Synchronous ``HttpClient`` implementation built on `httpx`.
 
 Supports streaming uploads and downloads, per-phase timeouts (connect /
-read / write / pool), HTTP/2 (opt-in via :mod:`httpx`), and proxies.
+read / write / pool), HTTP/2 (opt-in via `httpx`), and proxies.
 Production-grade alternative to the urllib reference client shipped in
 ``dexpace-sdk-http-stdlib``.
 
-The transport delegates to :class:`httpx.Client`, streaming the request
-body via :meth:`RequestBody.iter_bytes` and exposing the response as a
-:class:`ResponseBody` whose ``iter_bytes`` walks the httpx response's own
+The transport delegates to `httpx.Client`, streaming the request
+body via `RequestBody.iter_bytes` and exposing the response as a
+`ResponseBody` whose ``iter_bytes`` walks the httpx response's own
 ``iter_bytes`` iterator. Closing the SDK response closes the underlying
 httpx response and releases the connection back to the pool.
 """
@@ -40,15 +40,15 @@
 
 
 class HttpxHttpClient:
-    """Synchronous transport over :class:`httpx.Client`.
+    """Synchronous transport over `httpx.Client`.
 
     Behaves as a structural ``HttpClient`` (single ``execute`` method) plus
     a context-manager surface so a ``Pipeline`` can take ownership and call
     ``close`` on exit.
 
     Per-phase timeouts (``connect_timeout``, ``read_timeout``,
     ``write_timeout``, ``pool_timeout``) are forwarded to
-    :class:`httpx.Timeout`. ``None`` disables the phase's timeout.
+    `httpx.Timeout`. ``None`` disables the phase's timeout.
 
     Args:
         connect_timeout: Seconds allowed for connection establishment.
@@ -58,9 +58,9 @@ class HttpxHttpClient:
             request body.
         pool_timeout: Seconds allowed to acquire a connection from the
             pool.
-        transport: Optional :class:`httpx.BaseTransport`; the primary
+        transport: Optional `httpx.BaseTransport`; the primary
             extension hook for tests (``httpx.MockTransport``).
-        client: Optional pre-built :class:`httpx.Client` — overrides the
+        client: Optional pre-built `httpx.Client` — overrides the
             timeout / transport kwargs entirely. Ownership transfers to
             this transport.
     """

packages/dexpace-sdk-http-httpx/tests/test_async_httpx_client.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2026 dexpace and Omar Aljarrah.
 # Licensed under the MIT License. See LICENSE.md in the repository root for details.
 
-"""Tests for ``AsyncHttpxHttpClient`` using :class:`httpx.MockTransport`."""
+"""Tests for ``AsyncHttpxHttpClient`` using `httpx.MockTransport`."""
 
 from __future__ import annotations
 

packages/dexpace-sdk-http-httpx/tests/test_httpx_client.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2026 dexpace and Omar Aljarrah.
 # Licensed under the MIT License. See LICENSE.md in the repository root for details.
 
-"""Tests for ``HttpxHttpClient`` using :class:`httpx.MockTransport`."""
+"""Tests for ``HttpxHttpClient`` using `httpx.MockTransport`."""
 
 from __future__ import annotations
 

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

@@ -5,15 +5,15 @@
 
 ``RequestsHttpClient`` wraps a ``requests.Session`` configured with
 ``stream=True`` so response bodies are read lazily and surfaced through the
-SDK's :class:`ResponseBody` streaming API. Request bodies are produced via
-:meth:`RequestBody.iter_bytes` in 8 KiB chunks.
+SDK's `ResponseBody` streaming API. Request bodies are produced via
+`RequestBody.iter_bytes` in 8 KiB chunks.
 
 Exception mapping (``requests`` -> SDK):
 
-- ``requests.ConnectTimeout`` -> :class:`ServiceRequestTimeoutError`
-- ``requests.ReadTimeout`` -> :class:`ServiceResponseTimeoutError`
-- ``requests.ConnectionError`` -> :class:`ServiceRequestError`
-- ``requests.RequestException`` (catch-all) -> :class:`ServiceRequestError`
+- ``requests.ConnectTimeout`` -> `ServiceRequestTimeoutError`
+- ``requests.ReadTimeout`` -> `ServiceResponseTimeoutError`
+- ``requests.ConnectionError`` -> `ServiceRequestError`
+- ``requests.RequestException`` (catch-all) -> `ServiceRequestError`
 """
 
 from __future__ import annotations
@@ -50,7 +50,7 @@ class RequestsHttpClient:
     a context-manager surface so a ``Pipeline`` can take ownership and call
     ``close`` on exit. Each call sends one ``requests`` request with
     ``stream=True`` and wraps the streamed response into a
-    :class:`ResponseBody`.
+    `ResponseBody`.
 
     Attributes:
         timeout: Single timeout in seconds applied to ``Session.request``;
@@ -145,6 +145,7 @@ def _build_response(request: Request, raw: requests.Response) -> Response:
     try:
         status = Status(raw.status_code)
     except ValueError as err:
+        raw.close()
         raise ServiceResponseError(f"Unknown status code: {raw.status_code}", error=err) from err
     headers = Headers(list(raw.headers.items()))
     body = ResponseBody.from_stream(_IterContentStream(raw))  # type: ignore[arg-type]
@@ -162,7 +163,7 @@ def _build_response(request: Request, raw: requests.Response) -> Response:
 class _IterContentStream:
     """Adapter that exposes ``requests.Response.iter_content`` as a stream.
 
-    :class:`ResponseBody.from_stream` calls ``read(chunk_size)`` and
+    `ResponseBody.from_stream` calls ``read(chunk_size)`` and
     ``close()`` on its argument. ``requests`` doesn't expose a file-like
     object that honours chunk-size hints once decoded, but ``iter_content``
     does — this adapter buffers what the iterator yields and serves it in
@@ -182,7 +183,7 @@ def read(self, size: int = -1) -> bytes:
             return b""
         if self._iter is None:
             self._iter = self._response.iter_content(chunk_size=_CHUNK_SIZE)
-        if size is None or size < 0:
+        if size < 0:
             for chunk in self._iter:
                 if chunk:
                     self._buf.extend(chunk)