fix(trtllm): load embedding URLs through the shared HTTP client

components/src/dynamo/common/configuration/groups/http_args.py

@@ -105,8 +105,8 @@ class HttpConfigBase(ConfigBase):
     # Per-backend semantics:
     #   httpx → ``Timeout.read`` (``connect`` / ``pool`` stay independent
     #           so a stuck handshake or saturated pool still fast-fails).
-    #   aiohttp → ``ClientTimeout.total`` (aiohttp has no separate read
-    #             component; the override caps the whole request).
+    #   aiohttp → ``ClientTimeout.total``: the override caps the whole
+    #             request. It does not change the caller's ``read_timeout``.
     per_call_timeout_override: Optional[float]
 
     # TCP+TLS-handshake budget in seconds. Independent of the per-call /

components/src/dynamo/common/http/__init__.py

@@ -79,10 +79,12 @@ def get_default_client() -> HttpClient:
     return _default
 
 
-async def fetch_bytes(url, timeout, *, policy=None, max_bytes=None) -> bytes:
+async def fetch_bytes(
+    url, timeout, *, policy=None, max_bytes=None, read_timeout=None
+) -> bytes:
     """Singleton-backed convenience wrapper over :meth:`HttpClient.fetch_bytes`."""
     return await get_default_client().fetch_bytes(
-        url, timeout, policy=policy, max_bytes=max_bytes
+        url, timeout, policy=policy, max_bytes=max_bytes, read_timeout=read_timeout
     )
 
 

components/src/dynamo/common/http/aiohttp_client.py

@@ -61,18 +61,23 @@ def __init__(self, config=None) -> None:
         # session never closes it. Hold each one and close it ourselves.
         self._resolvers: dict[bool, BlocklistResolver] = {}
 
-    def _effective_timeout(self, timeout: float) -> aiohttp.ClientTimeout:
+    def _effective_timeout(
+        self, timeout: float, read_timeout: Optional[float] = None
+    ) -> aiohttp.ClientTimeout:
         # The override caps ``total`` (whole request). ``sock_connect``
         # bounds just the TCP+TLS handshake, so a stuck origin fast-fails at
         # the connect budget instead of burning the full ``total`` before any
-        # byte arrives.
+        # byte arrives. ``sock_read`` fails a server that stops sending, and
+        # the override does not change it.
         total = (
             self._config.per_call_timeout_override
             if self._config.per_call_timeout_override is not None
             else timeout
         )
         return aiohttp.ClientTimeout(
-            total=total, sock_connect=self._config.connect_timeout
+            total=total,
+            sock_connect=self._config.connect_timeout,
+            sock_read=read_timeout,
         )
 
     @staticmethod
@@ -164,14 +169,15 @@ async def _fetch_simple(
         *,
         max_bytes: Optional[int] = None,
         policy: Optional[UrlValidationPolicy] = None,
+        read_timeout: Optional[float] = None,
     ) -> bytes:
         allow_private = self._connect_allows_private(policy)
         # Only when the check is meant to bite. If private destinations are
         # already permitted for this fetch, the proxy gate protects nothing.
         if not allow_private:
             await self._require_trusted_egress_proxy(url)
         session = await self._get_session(allow_private)
-        client_timeout = self._effective_timeout(timeout)
+        client_timeout = self._effective_timeout(timeout, read_timeout)
         try:
             async with session.get(
                 url, timeout=client_timeout, allow_redirects=True
@@ -206,14 +212,15 @@ async def _fetch_body_or_redirect(
         *,
         max_bytes: Optional[int] = None,
         policy: Optional[UrlValidationPolicy] = None,
+        read_timeout: Optional[float] = None,
     ) -> tuple[bytes | None, str | None]:
         allow_private = self._connect_allows_private(policy)
         # Only when the check is meant to bite. If private destinations are
         # already permitted for this fetch, the proxy gate protects nothing.
         if not allow_private:
             await self._require_trusted_egress_proxy(url)
         session = await self._get_session(allow_private)
-        client_timeout = self._effective_timeout(timeout)
+        client_timeout = self._effective_timeout(timeout, read_timeout)
         try:
             async with session.get(
                 url, timeout=client_timeout, allow_redirects=False

components/src/dynamo/common/http/base.py

@@ -121,6 +121,7 @@ async def fetch_bytes(
         *,
         policy: Optional[UrlValidationPolicy] = None,
         max_bytes: Optional[int] = None,
+        read_timeout: Optional[float] = None,
     ) -> bytes:
         """Fetch ``url`` and return the response body.
 
@@ -134,15 +135,21 @@ async def fetch_bytes(
         Raises :class:`UrlValidationError` — it is a verdict on a
         client-supplied source, like the redirect cap below.
 
+        ``read_timeout`` set: raise :class:`HttpTimeoutError` when the server
+        sends nothing for that many seconds. ``timeout`` still bounds the
+        whole request. ``None`` sets no such limit.
+
         ``policy`` set: follow redirects manually and revalidate each
         hop against the policy via :func:`url_validator.validate_url`.
         This is the SSRF-safe path; raises :class:`UrlValidationError`
         if any hop fails or the chain exceeds ``_MAX_REDIRECTS``.
         """
         if policy is None:
-            return await self._fetch_simple(url, timeout, max_bytes=max_bytes)
+            return await self._fetch_simple(
+                url, timeout, max_bytes=max_bytes, read_timeout=read_timeout
+            )
         return await self._fetch_with_revalidation(
-            url, timeout, policy, max_bytes=max_bytes
+            url, timeout, policy, max_bytes=max_bytes, read_timeout=read_timeout
         )
 
     async def _fetch_with_revalidation(
@@ -152,6 +159,7 @@ async def _fetch_with_revalidation(
         policy: UrlValidationPolicy,
         *,
         max_bytes: Optional[int] = None,
+        read_timeout: Optional[float] = None,
     ) -> bytes:
         """Manual redirect loop with per-hop SSRF validation (backend-neutral)."""
         current = url
@@ -162,7 +170,11 @@ async def _fetch_with_revalidation(
             visited.append(current)
 
             body, redirect_to = await self._fetch_body_or_redirect(
-                current, timeout, max_bytes=max_bytes, policy=policy
+                current,
+                timeout,
+                max_bytes=max_bytes,
+                policy=policy,
+                read_timeout=read_timeout,
             )
 
             if redirect_to is None:
@@ -190,6 +202,7 @@ async def _fetch_simple(
         *,
         max_bytes: Optional[int] = None,
         policy: Optional[UrlValidationPolicy] = None,
+        read_timeout: Optional[float] = None,
     ) -> bytes:
         """Backend's native redirect-following GET (no SSRF policy applied).
 
@@ -206,6 +219,7 @@ async def _fetch_body_or_redirect(
         *,
         max_bytes: Optional[int] = None,
         policy: Optional[UrlValidationPolicy] = None,
+        read_timeout: Optional[float] = None,
     ) -> tuple[bytes | None, str | None]:
         """Single hop with redirects disabled.
 

components/src/dynamo/common/tests/http/test_aiohttp_client.py

@@ -16,6 +16,10 @@
 from __future__ import annotations
 
 import asyncio
+import http.server
+import threading
+import time
+from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import aiohttp
@@ -443,3 +447,136 @@ async def test_the_proxy_gate_is_awaitable() -> None:
     import inspect
 
     assert inspect.iscoroutinefunction(AiohttpClient._require_trusted_egress_proxy)
+
+
+# --- Read timeout: a stalled server fails early, a slow one still finishes ---
+
+
+@pytest.mark.parametrize("policy", [None, _PERMISSIVE], ids=["simple", "policy"])
+@pytest.mark.parametrize(
+    ("override", "kwargs", "expected"),
+    [
+        # No read timeout: what every caller that passes none gets, unchanged.
+        (None, {}, {"total": 30.0}),
+        (None, {"read_timeout": 2.0}, {"total": 30.0, "sock_read": 2.0}),
+        # DYN_HTTP_TIMEOUT replaces the total budget only.
+        ("7", {"read_timeout": 2.0}, {"total": 7.0, "sock_read": 2.0}),
+    ],
+    ids=["default", "read-timeout", "override"],
+)
+async def test_the_request_timeout_carries_the_read_timeout(
+    monkeypatch, policy, override, kwargs, expected
+) -> None:
+    if override is None:
+        monkeypatch.delenv("DYN_HTTP_TIMEOUT", raising=False)
+    else:
+        monkeypatch.setenv("DYN_HTTP_TIMEOUT", override)
+    seen = []
+    respond = _cm_returning(_FakeResponse(status=200, body=b"ok"))
+
+    def _get(url, **get_kwargs):
+        seen.append(get_kwargs["timeout"])
+        return respond(url)
+
+    session = MagicMock(spec=aiohttp.ClientSession)
+    session.closed = False
+    session.get = _get
+    client = _make_client_with_session(session)
+
+    body = await client.fetch_bytes("https://h/x.png", 30.0, policy=policy, **kwargs)
+
+    assert body == b"ok"
+    connect = client._config.connect_timeout
+    assert seen == [aiohttp.ClientTimeout(sock_connect=connect, **expected)]
+
+
+@pytest.fixture
+def paced_server():
+    """Serve ``state.body`` on loopback.
+
+    ``chunks`` and ``gap`` send the body in pieces, ``gap`` seconds apart.
+    ``stall`` sends one byte and then nothing more.
+    """
+    state = SimpleNamespace(body=bytes(range(256)) * 4, chunks=1, gap=0.0, stall=False)
+    done = threading.Event()
+
+    class Handler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):  # noqa: N802 - http.server API
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(state.body)))
+            self.end_headers()
+            try:
+                if state.stall:
+                    self.wfile.write(state.body[:1])
+                    done.wait(10)
+                    return
+                size = -(-len(state.body) // state.chunks)
+                for start in range(0, len(state.body), size):
+                    time.sleep(state.gap)
+                    self.wfile.write(state.body[start : start + size])
+            except (BrokenPipeError, ConnectionResetError):
+                pass  # the client gave up
+
+        def log_message(self, *args):
+            pass
+
+    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    threading.Thread(
+        target=server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
+    ).start()
+    state.url = f"http://127.0.0.1:{server.server_address[1]}/x.bin"
+    try:
+        yield state
+    finally:
+        done.set()
+        server.shutdown()
+        server.server_close()
+
+
+# The timeouts below stay under 5 s, the value from which aiohttp rounds a
+# timeout up to a whole second.
+
+
+@_allows_cleanup_closed_notice
+@pytest.mark.parametrize("policy", [None, _PERMISSIVE], ids=["simple", "policy"])
+async def test_a_stalled_server_fails_at_the_read_timeout(
+    monkeypatch, paced_server, policy
+) -> None:
+    """A server that stops sending fails at ``read_timeout``, long before the
+    whole-request ``timeout``."""
+    monkeypatch.delenv("DYN_HTTP_TIMEOUT", raising=False)
+    paced_server.stall = True
+    client = AiohttpClient()
+    try:
+        start = time.monotonic()
+        with pytest.raises(mm_http.HttpTimeoutError):
+            await client.fetch_bytes(
+                paced_server.url, 4.0, policy=policy, read_timeout=1.0
+            )
+        elapsed = time.monotonic() - start
+    finally:
+        await client.close()
+
+    assert elapsed < 2.5, f"waited {elapsed:.2f} s, the whole 4 s budget"
+
+
+@_allows_cleanup_closed_notice
+async def test_a_slow_steady_server_completes_within_the_total_budget(
+    monkeypatch, paced_server
+) -> None:
+    """``read_timeout`` bounds each wait for bytes, not the whole download."""
+    monkeypatch.delenv("DYN_HTTP_TIMEOUT", raising=False)
+    paced_server.chunks, paced_server.gap = 15, 0.1
+    client = AiohttpClient()
+    try:
+        start = time.monotonic()
+        body = await client.fetch_bytes(
+            paced_server.url, 4.0, policy=_PERMISSIVE, read_timeout=1.0
+        )
+        elapsed = time.monotonic() - start
+    finally:
+        await client.close()
+
+    assert body == paced_server.body
+    # Longer than the 1 s read timeout, and within the 4 s budget.
+    assert elapsed > 1.0

components/src/dynamo/common/tests/http/test_http_facade.py

@@ -94,7 +94,7 @@ async def test_fetch_with_policy_returns_first_response(
 
     call_count = {"n": 0}
 
-    async def _fake(url, timeout, *, max_bytes=None, policy=None):
+    async def _fake(url, timeout, *, max_bytes=None, policy=None, read_timeout=None):
         call_count["n"] += 1
         return b"body-bytes", None
 
@@ -115,7 +115,7 @@ async def test_fetch_with_policy_follows_safe_redirect(
 
     hops: list[str] = []
 
-    async def _fake(url, timeout, *, max_bytes=None, policy=None):
+    async def _fake(url, timeout, *, max_bytes=None, policy=None, read_timeout=None):
         hops.append(url)
         if url == "https://example.com/x.png":
             return None, "https://example.com/final.png"
@@ -138,7 +138,7 @@ async def test_fetch_with_policy_blocks_redirect_to_private_ip(
 
     strict = UrlValidationPolicy(allow_private_ips=False)
 
-    async def _fake(url, timeout, *, max_bytes=None, policy=None):
+    async def _fake(url, timeout, *, max_bytes=None, policy=None, read_timeout=None):
         return None, "http://169.254.169.254/latest/meta-data/"
 
     with patch.object(client, "_fetch_body_or_redirect", _fake):
@@ -161,14 +161,36 @@ async def test_fetch_with_policy_enforces_redirect_limit(
         "https://example.com/d": "https://example.com/e",
     }
 
-    async def _fake(url, timeout, *, max_bytes=None, policy=None):
+    async def _fake(url, timeout, *, max_bytes=None, policy=None, read_timeout=None):
         return None, chain[url]
 
     with patch.object(client, "_fetch_body_or_redirect", _fake):
         with pytest.raises(UrlValidationError, match="Too many redirects"):
             await mm_http.fetch_bytes("https://example.com/a", 30.0, policy=_PERMISSIVE)
 
 
+@pytest.mark.parametrize("policy", [None, _PERMISSIVE], ids=["simple", "policy"])
+async def test_read_timeout_reaches_the_backend(policy) -> None:
+    client = mm_http.get_default_client()
+    seen = []
+
+    async def _simple(url, timeout, *, max_bytes=None, policy=None, read_timeout=None):
+        seen.append(read_timeout)
+        return b"body"
+
+    async def _hop(url, timeout, *, max_bytes=None, policy=None, read_timeout=None):
+        seen.append(read_timeout)
+        return b"body", None
+
+    with patch.object(client, "_fetch_simple", _simple), patch.object(
+        client, "_fetch_body_or_redirect", _hop
+    ):
+        await mm_http.fetch_bytes(
+            "https://example.com/x.png", 30.0, policy=policy, read_timeout=2.5
+        )
+    assert seen == [2.5]
+
+
 # --- Download cap (collect_capped) ---
 #
 # The cap exists because moving the diffusion download out of SGLang left its

components/src/dynamo/common/tests/http/test_url_validator.py

@@ -547,11 +547,13 @@ async def test_redirect_chain_in_the_limit_message_is_bounded() -> None:
     long_hop = "https://example.com/" + "A" * 200_000
 
     class _Client(HttpClient):
-        async def _fetch_simple(self, url, timeout, *, max_bytes=None, policy=None):
+        async def _fetch_simple(
+            self, url, timeout, *, max_bytes=None, policy=None, read_timeout=None
+        ):
             raise AssertionError("unused")
 
         async def _fetch_body_or_redirect(
-            self, url, timeout, *, max_bytes=None, policy=None
+            self, url, timeout, *, max_bytes=None, policy=None, read_timeout=None
         ):
             return None, long_hop