Make platform HTTP write timeout unbounded and configurable

python/tests/platform/test_platform_utils.py

@@ -7,7 +7,12 @@
 import httpx
 import pytest
 from timbal.errors import PlatformError
-from timbal.platform.utils import _request, _resolve_url_and_headers, _stream
+from timbal.platform.utils import (
+    _default_timeout,
+    _request,
+    _resolve_url_and_headers,
+    _stream,
+)
 from timbal.state import set_run_context
 from timbal.state.config import PlatformAuth, PlatformAuthType, PlatformConfig, PlatformSubject
 from timbal.state.context import RunContext
@@ -55,6 +60,9 @@ def _isolated_env(monkeypatch):
         "TIMBAL_API_KEY",
         "TIMBAL_API_HOST",
         "TIMBAL_ORG_ID",
+        "TIMBAL_HTTP_TIMEOUT",
+        "TIMBAL_HTTP_WRITE_TIMEOUT",
+        "TIMBAL_HTTP_READ_TIMEOUT",
     ):
         monkeypatch.delenv(var, raising=False)
     # Reset config-loader cache so RunContext() doesn't pick up a stale hit
@@ -222,6 +230,40 @@ def _make_http_status_error(
     return httpx.HTTPStatusError(reason, request=request, response=response)
 
 
+# ===========================================================================
+# TestDefaultTimeout
+# ===========================================================================
+
+
+class TestDefaultTimeout:
+    """Tests for env-driven platform HTTP timeouts."""
+
+    def test_defaults_unbounded_write_and_read(self):
+        timeout = _default_timeout()
+        assert timeout.connect == 10.0
+        assert timeout.write is None
+        assert timeout.read is None
+
+    def test_env_overrides(self, monkeypatch):
+        monkeypatch.setenv("TIMBAL_HTTP_TIMEOUT", "5")
+        monkeypatch.setenv("TIMBAL_HTTP_WRITE_TIMEOUT", "30")
+        monkeypatch.setenv("TIMBAL_HTTP_READ_TIMEOUT", "none")
+        timeout = _default_timeout()
+        assert timeout.connect == 5.0
+        assert timeout.write == 30.0
+        assert timeout.read is None
+
+    def test_connect_none_falls_back_to_default(self, monkeypatch):
+        monkeypatch.setenv("TIMBAL_HTTP_TIMEOUT", "none")
+        timeout = _default_timeout()
+        assert timeout.connect == 10.0
+
+    def test_invalid_env_raises_value_error(self, monkeypatch):
+        monkeypatch.setenv("TIMBAL_HTTP_WRITE_TIMEOUT", "nope")
+        with pytest.raises(ValueError, match="TIMBAL_HTTP_WRITE_TIMEOUT"):
+            _default_timeout()
+
+
 # ===========================================================================
 # TestRequest
 # ===========================================================================
@@ -351,6 +393,99 @@ async def test_network_exception_retries_then_reraises(self):
                     await _request("GET", "health", service="api", max_retries=2)
         assert mock_http.request.call_count == 3
 
+    @pytest.mark.asyncio
+    async def test_default_timeout_passed_to_client(self):
+        mock_http = AsyncMock()
+        mock_client_cm = _mock_httpx_client(mock_http)
+        ok_response = MagicMock()
+        ok_response.raise_for_status = MagicMock()
+        mock_http.request = AsyncMock(return_value=ok_response)
+
+        with patch("timbal.platform.utils.httpx.AsyncClient", return_value=mock_client_cm) as mock_cls:
+            await _request("GET", "health", service="api", max_retries=0)
+
+        _, kwargs = mock_cls.call_args
+        timeout = kwargs["timeout"]
+        assert isinstance(timeout, httpx.Timeout)
+        assert timeout.connect == 10.0
+        assert timeout.write is None
+        assert timeout.read is None
+
+    @pytest.mark.asyncio
+    async def test_float_timeout_keeps_read_unbounded(self):
+        mock_http = AsyncMock()
+        mock_client_cm = _mock_httpx_client(mock_http)
+        ok_response = MagicMock()
+        ok_response.raise_for_status = MagicMock()
+        mock_http.request = AsyncMock(return_value=ok_response)
+
+        with patch("timbal.platform.utils.httpx.AsyncClient", return_value=mock_client_cm) as mock_cls:
+            await _request("GET", "health", service="api", max_retries=0, timeout=30)
+
+        _, kwargs = mock_cls.call_args
+        timeout = kwargs["timeout"]
+        assert timeout.connect == 30.0
+        assert timeout.write == 30.0
+        assert timeout.read is None
+
+    @pytest.mark.asyncio
+    async def test_write_timeout_raises_platform_error_with_hint(self):
+        mock_http = AsyncMock()
+        mock_client_cm = _mock_httpx_client(mock_http)
+        mock_http.request = AsyncMock(side_effect=httpx.WriteTimeout("write timed out"))
+
+        with patch("timbal.platform.utils.httpx.AsyncClient", return_value=mock_client_cm):
+            with patch("timbal.platform.utils.asyncio.sleep", new=AsyncMock()):
+                with pytest.raises(PlatformError) as ei:
+                    await _request(
+                        "POST",
+                        "upload",
+                        service="api",
+                        content=b"big",
+                        max_retries=0,
+                        timeout=httpx.Timeout(10.0, read=None, write=10.0),
+                    )
+        msg = str(ei.value)
+        assert "write timeout" in msg
+        assert "TIMBAL_HTTP_WRITE_TIMEOUT" in msg
+        assert "limit=10s" in msg
+
+    @pytest.mark.asyncio
+    async def test_read_timeout_raises_platform_error_with_hint(self):
+        mock_http = AsyncMock()
+        mock_client_cm = _mock_httpx_client(mock_http)
+        mock_http.request = AsyncMock(side_effect=httpx.ReadTimeout("read timed out"))
+
+        with patch("timbal.platform.utils.httpx.AsyncClient", return_value=mock_client_cm):
+            with patch("timbal.platform.utils.asyncio.sleep", new=AsyncMock()):
+                with pytest.raises(PlatformError) as ei:
+                    await _request(
+                        "GET",
+                        "health",
+                        service="api",
+                        max_retries=0,
+                        timeout=httpx.Timeout(10.0, read=5.0),
+                    )
+        msg = str(ei.value)
+        assert "read timeout" in msg
+        assert "TIMBAL_HTTP_READ_TIMEOUT" in msg
+        assert "limit=5s" in msg
+
+    @pytest.mark.asyncio
+    async def test_timeout_retries_then_succeeds(self):
+        mock_http = AsyncMock()
+        mock_client_cm = _mock_httpx_client(mock_http)
+        ok_response = MagicMock()
+        ok_response.raise_for_status = MagicMock()
+        mock_http.request = AsyncMock(side_effect=[httpx.ConnectTimeout("connect timed out"), ok_response])
+
+        with patch("timbal.platform.utils.httpx.AsyncClient", return_value=mock_client_cm):
+            with patch("timbal.platform.utils.asyncio.sleep", new=AsyncMock()) as mock_sleep:
+                result = await _request("GET", "health", service="api", max_retries=2)
+        assert result is ok_response
+        mock_sleep.assert_called_once()
+        assert mock_http.request.call_count == 2
+
     @pytest.mark.asyncio
     async def test_json_payload_forwarded(self):
         mock_http = AsyncMock()

python/timbal/platform/utils.py

@@ -12,6 +12,77 @@
 
 logger = structlog.get_logger("timbal.platform.utils")
 
+# Connect/pool default. Write/read default to unbounded so large uploads (KB
+# ingest, recording multipart, …) don't die with an opaque httpx.WriteTimeout
+# after 10s of body send. Override via TIMBAL_HTTP_*_TIMEOUT env vars.
+_DEFAULT_CONNECT_TIMEOUT = 10.0
+
+
+def _env_timeout_seconds(name: str, default: float | None) -> float | None:
+    """Parse a timeout env var: float seconds, or none/null/inf/unlimited → None."""
+    raw = os.environ.get(name)
+    if raw is None:
+        return default
+    value = raw.strip().lower()
+    if value in ("", "none", "null", "inf", "unlimited"):
+        return None
+    try:
+        return float(value)
+    except ValueError as exc:
+        raise ValueError(
+            f"Invalid {name}={raw!r}: expected a number of seconds, or 'none'."
+        ) from exc
+
+
+def _default_timeout() -> httpx.Timeout:
+    """Resolve the default httpx timeout for platform HTTP calls.
+
+    Env knobs (seconds; ``none`` = unbounded):
+      - ``TIMBAL_HTTP_TIMEOUT`` — connect + pool (default 10)
+      - ``TIMBAL_HTTP_WRITE_TIMEOUT`` — request body send (default unbounded)
+      - ``TIMBAL_HTTP_READ_TIMEOUT`` — response body read (default unbounded)
+    """
+    connect = _env_timeout_seconds("TIMBAL_HTTP_TIMEOUT", _DEFAULT_CONNECT_TIMEOUT)
+    if connect is None:
+        # Connect must stay finite — otherwise a dead host hangs forever.
+        connect = _DEFAULT_CONNECT_TIMEOUT
+    write = _env_timeout_seconds("TIMBAL_HTTP_WRITE_TIMEOUT", None)
+    read = _env_timeout_seconds("TIMBAL_HTTP_READ_TIMEOUT", None)
+    return httpx.Timeout(connect, read=read, write=write)
+
+
+def _timeout_error_message(exc: httpx.TimeoutException, url: str, timeout: httpx.Timeout) -> str:
+    """Human-readable PlatformError body for httpx timeout failures."""
+    if isinstance(exc, httpx.WriteTimeout):
+        phase = "write"
+        detail = "sending the request body"
+        knob = "TIMBAL_HTTP_WRITE_TIMEOUT"
+        configured = timeout.write
+    elif isinstance(exc, httpx.ReadTimeout):
+        phase = "read"
+        detail = "waiting for / reading the response"
+        knob = "TIMBAL_HTTP_READ_TIMEOUT"
+        configured = timeout.read
+    elif isinstance(exc, httpx.ConnectTimeout):
+        phase = "connect"
+        detail = "establishing the connection"
+        knob = "TIMBAL_HTTP_TIMEOUT"
+        configured = timeout.connect
+    else:
+        phase = "timeout"
+        detail = "completing the request"
+        knob = "TIMBAL_HTTP_TIMEOUT"
+        configured = timeout.connect
+
+    limit = "unbounded" if configured is None else f"{configured:g}s"
+    return (
+        f"\n"
+        f"  URL: {url}\n"
+        f"  Error: {phase} timeout while {detail} (limit={limit})\n"
+        f"  Hint: raise the limit via {knob}=<seconds> (or '{knob}=none' for "
+        f"unbounded), or pass timeout= to the platform request helper."
+    )
+
 
 def _resolve_url_and_headers(
     service: str | None,
@@ -113,13 +184,17 @@ async def _request(
 
     ``backoff`` overrides the retry delay: called with the 0-based attempt
     index, returns seconds to wait (default: 0.1s doubling). A ``Retry-After``
-    on 429 still wins when longer. ``timeout`` overrides the default
-    ``Timeout(10.0, read=None)`` — e.g. large multipart uploads need an
-    unbounded write timeout.
+    on 429 still wins when longer. ``timeout`` overrides
+    :func:`_default_timeout` (connect 10s, read/write unbounded; see
+    ``TIMBAL_HTTP_*_TIMEOUT``).
     """
     url, headers = _resolve_url_and_headers(service, path, headers)
     if timeout is None:
-        timeout = httpx.Timeout(10.0, read=None)
+        timeout = _default_timeout()
+    elif isinstance(timeout, (int, float)):
+        # Match httpx: a bare number is an all-phases timeout. Keep read
+        # unbounded so long-running responses still work.
+        timeout = httpx.Timeout(float(timeout), read=None)
     payload_kwargs = {}
     # `is not None` so an empty dict still sends a JSON body + Content-Type
     # (parameterless POSTs would otherwise 415 Unsupported Media Type).
@@ -181,8 +256,19 @@ async def _request(
                 status_code=exc.response.status_code,
             )
             await asyncio.sleep(wait_time)
+        except httpx.TimeoutException as exc:
+            if attempt == max_retries:
+                raise PlatformError(_timeout_error_message(exc, url, timeout)) from exc
+            wait_time = backoff(attempt) if backoff is not None else 0.1 * (2**attempt)
+            logger.warning(
+                f"Request timed out, retrying in {wait_time:.1f}s",
+                attempt=attempt + 1,
+                max_retries=max_retries,
+                error=type(exc).__name__,
+            )
+            await asyncio.sleep(wait_time)
         except Exception as exc:
-            # Retry on any other error (network, timeout, etc.)
+            # Retry on any other error (network, etc.)
             if attempt == max_retries:
                 raise
             wait_time = backoff(attempt) if backoff is not None else 0.1 * (2**attempt)
@@ -223,9 +309,10 @@ async def _stream(
     elif files:
         payload_kwargs["files"] = files
 
+    timeout = _default_timeout()
     for attempt in range(max_retries + 1):
         try:
-            async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, read=None)) as client:
+            async with httpx.AsyncClient(timeout=timeout) as client:
                 async with client.stream(method, url, headers=headers, params=params, **payload_kwargs) as response:
                     response.raise_for_status()
 
@@ -287,8 +374,19 @@ async def _stream(
                 status_code=exc.response.status_code,
             )
             await asyncio.sleep(wait_time)
+        except httpx.TimeoutException as exc:
+            if attempt == max_retries:
+                raise PlatformError(_timeout_error_message(exc, url, timeout)) from exc
+            wait_time = 0.1 * (2**attempt)
+            logger.warning(
+                f"Stream request timed out, retrying in {wait_time:.1f}s",
+                attempt=attempt + 1,
+                max_retries=max_retries,
+                error=type(exc).__name__,
+            )
+            await asyncio.sleep(wait_time)
         except Exception as exc:
-            # Retry on any other error (network, timeout, etc.)
+            # Retry on any other error (network, etc.)
             if attempt == max_retries:
                 raise
             wait_time = 0.1 * (2**attempt)