From 44e6b62534f06880fec69dc9d74e0aaec99974b8 Mon Sep 17 00:00:00 2001 From: Reflex Date: Sun, 16 Aug 2026 20:33:34 +0000 Subject: [PATCH 1/4] fix(transport): normalize lifecycle incident failures --- src/runloop_api_client/_base_client.py | 57 +++- src/runloop_api_client/_exceptions.py | 76 +++++- src/runloop_api_client/lib/error_contract.py | 97 +++++++ .../lib/tunnel_readiness.py | 79 ++++++ src/runloop_api_client/sdk/async_devbox.py | 23 ++ src/runloop_api_client/sdk/devbox.py | 27 ++ tests/test_transport_error_contract.py | 248 ++++++++++++++++++ 7 files changed, 587 insertions(+), 20 deletions(-) create mode 100644 src/runloop_api_client/lib/error_contract.py create mode 100644 src/runloop_api_client/lib/tunnel_readiness.py create mode 100644 tests/test_transport_error_contract.py diff --git a/src/runloop_api_client/_base_client.py b/src/runloop_api_client/_base_client.py index da359063d..b91507aec 100644 --- a/src/runloop_api_client/_base_client.py +++ b/src/runloop_api_client/_base_client.py @@ -94,6 +94,7 @@ APIResponseValidationError, ) from ._utils._json import openapi_dumps +from .lib.error_contract import is_safe_transport_retry log: logging.Logger = logging.getLogger(__name__) @@ -665,7 +666,10 @@ def _enforce_trailing_slash(self, url: URL) -> URL: def _make_status_error_from_response( self, response: httpx.Response, + *, + attempts: int = 1, ) -> APIStatusError: + body: object | None if response.is_closed and not response.is_stream_consumed: # We can't read the response body as it has been closed # before it was read. This can happen if an event hook @@ -677,12 +681,19 @@ def _make_status_error_from_response( body = err_text try: - body = json.loads(err_text) - err_msg = f"Error code: {response.status_code} - {body}" + body = cast(object, json.loads(err_text)) + body_mapping = cast(Mapping[str, object], body) if isinstance(body, dict) else None + body_message = body_mapping.get("message") if body_mapping is not None else None + if isinstance(body_message, str): + err_msg = body_message + else: + err_msg = f"Error code: {response.status_code} - {body}" except Exception: err_msg = err_text or f"Error code: {response.status_code}" - return self._make_status_error(err_msg, body=body, response=response) + error = self._make_status_error(err_msg, body=cast(object, body), response=response) + error.attempts = attempts + return error def _make_status_error( self, @@ -1045,6 +1056,26 @@ def _should_retry(self, response: httpx.Response) -> bool: log.debug("Not retrying as header `x-should-retry` is set to `false`") return False + # These failures happen after a request or response may have been + # partially transferred. Retrying them implicitly can duplicate an + # execute or replay a multipart stream. Servers may explicitly opt in + # with X-Should-Retry when an idempotency record makes that safe. + try: + raw_payload = response.json() + except Exception: + raw_payload = None + payload = cast(Mapping[str, object], raw_payload) if isinstance(raw_payload, dict) else None + code = response.headers.get("x-runloop-error-code") + if code is None and isinstance(payload, dict) and isinstance(payload.get("error"), str): + code = payload["error"] + if code in { + "upload_request_body_idle_timeout", + "tunnel_backend_idle_timeout", + "tunnel_backend_connection_reset", + }: + log.debug("Not retrying ambiguous transfer failure %s", code) + return False + # Retry on request timeouts. if response.status_code == 408: log.debug("Retrying due to status code %i", response.status_code) @@ -1366,7 +1397,7 @@ def request( except httpx.TimeoutException as err: log.debug("Encountered httpx.TimeoutException", exc_info=True) - if remaining_retries > 0: + if remaining_retries > 0 and is_safe_transport_retry(err): self._sleep_for_retry( retries_taken=retries_taken, max_retries=max_retries, @@ -1377,11 +1408,11 @@ def request( continue log.debug("Raising timeout error") - raise APITimeoutError(request=request) from err + raise APITimeoutError(request=request, cause=err, attempts=retries_taken + 1) from err except Exception as err: log.debug("Encountered Exception", exc_info=True) - if remaining_retries > 0: + if remaining_retries > 0 and is_safe_transport_retry(err): self._sleep_for_retry( retries_taken=retries_taken, max_retries=max_retries, @@ -1392,7 +1423,7 @@ def request( continue log.debug("Raising connection error") - raise APIConnectionError(request=request) from err + raise APIConnectionError(request=request, cause=err, attempts=retries_taken + 1) from err log.debug( 'HTTP Response: %s %s "%i %s" %s', @@ -1424,7 +1455,7 @@ def request( err.response.read() log.debug("Re-raising status error") - raise self._make_status_error_from_response(err.response) from None + raise self._make_status_error_from_response(err.response, attempts=retries_taken + 1) from None break @@ -2076,7 +2107,7 @@ async def request( except httpx.TimeoutException as err: log.debug("Encountered httpx.TimeoutException", exc_info=True) - if remaining_retries > 0: + if remaining_retries > 0 and is_safe_transport_retry(err): await self._sleep_for_retry( retries_taken=retries_taken, max_retries=max_retries, @@ -2087,11 +2118,11 @@ async def request( continue log.debug("Raising timeout error") - raise APITimeoutError(request=request) from err + raise APITimeoutError(request=request, cause=err, attempts=retries_taken + 1) from err except Exception as err: log.debug("Encountered Exception", exc_info=True) - if remaining_retries > 0: + if remaining_retries > 0 and is_safe_transport_retry(err): await self._sleep_for_retry( retries_taken=retries_taken, max_retries=max_retries, @@ -2102,7 +2133,7 @@ async def request( continue log.debug("Raising connection error") - raise APIConnectionError(request=request) from err + raise APIConnectionError(request=request, cause=err, attempts=retries_taken + 1) from err log.debug( 'HTTP Response: %s %s "%i %s" %s', @@ -2134,7 +2165,7 @@ async def request( await err.response.aread() log.debug("Re-raising status error") - raise self._make_status_error_from_response(err.response) from None + raise self._make_status_error_from_response(err.response, attempts=retries_taken + 1) from None break diff --git a/src/runloop_api_client/_exceptions.py b/src/runloop_api_client/_exceptions.py index dd4f7d276..2a3986df5 100644 --- a/src/runloop_api_client/_exceptions.py +++ b/src/runloop_api_client/_exceptions.py @@ -6,6 +6,8 @@ import httpx +from .lib.error_contract import status_error_details, transport_error_details + __all__ = [ "BadRequestError", "AuthenticationError", @@ -37,11 +39,39 @@ class APIError(RunloopError): If there was no response associated with this error then it will be `None`. """ - def __init__(self, message: str, request: httpx.Request, *, body: object | None) -> None: # noqa: ARG002 + code: str + phase: str + retryable: bool + request_id: str | None + retry_after: float | None + attempts: int + cause: BaseException | None + + def __init__( + self, + message: str, + request: httpx.Request, + *, + body: object | None, + code: str = "runloop_error", + phase: str = "unknown", + retryable: bool = False, + request_id: str | None = None, + retry_after: float | None = None, + attempts: int = 1, + cause: BaseException | None = None, + ) -> None: super().__init__(message) self.request = request self.message = message self.body = body + self.code = code + self.phase = phase + self.retryable = retryable + self.request_id = request_id + self.retry_after = retry_after + self.attempts = attempts + self.cause = cause class APIResponseValidationError(APIError): @@ -60,20 +90,52 @@ class APIStatusError(APIError): response: httpx.Response status_code: int - def __init__(self, message: str, *, response: httpx.Response, body: object | None) -> None: - super().__init__(message, response.request, body=body) + def __init__(self, message: str, *, response: httpx.Response, body: object | None, attempts: int = 1) -> None: + details = status_error_details(response, body) + super().__init__( + message, + response.request, + body=body, + code=details.code, + phase=details.phase, + retryable=details.retryable, + request_id=details.request_id, + retry_after=details.retry_after, + attempts=attempts, + ) self.response = response self.status_code = response.status_code class APIConnectionError(APIError): - def __init__(self, *, message: str = "Connection error.", request: httpx.Request) -> None: - super().__init__(message, request, body=None) + def __init__( + self, + *, + message: str = "Connection error.", + request: httpx.Request, + cause: BaseException | None = None, + attempts: int = 1, + ) -> None: + details = transport_error_details(cause) if cause is not None else transport_error_details(Exception()) + super().__init__( + message, + request, + body=None, + code=details.code, + phase=details.phase, + retryable=details.retryable, + attempts=attempts, + cause=cause, + ) class APITimeoutError(APIConnectionError): - def __init__(self, request: httpx.Request) -> None: - super().__init__(message="Request timed out.", request=request) + def __init__(self, request: httpx.Request, *, cause: BaseException | None = None, attempts: int = 1) -> None: + super().__init__(message="Request timed out.", request=request, cause=cause, attempts=attempts) + if cause is None: + self.code = "connection_timeout" + self.phase = "connect" + self.retryable = True class BadRequestError(APIStatusError): diff --git a/src/runloop_api_client/lib/error_contract.py b/src/runloop_api_client/lib/error_contract.py new file mode 100644 index 000000000..92473344d --- /dev/null +++ b/src/runloop_api_client/lib/error_contract.py @@ -0,0 +1,97 @@ +"""Stable normalization for Runloop API and HTTPX transport failures. + +This module is handwritten and intentionally lives under ``lib`` so generated +client updates only need a small integration point. +""" + +from __future__ import annotations + +from typing import Mapping, cast +from dataclasses import dataclass + +import httpx + + +@dataclass(frozen=True) +class ErrorDetails: + code: str + phase: str + retryable: bool + request_id: str | None = None + retry_after: float | None = None + + +def _number(value: object) -> float | None: + try: + parsed = float(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return None + return parsed if parsed >= 0 else None + + +def parse_retry_after(headers: httpx.Headers, body: object = None) -> float | None: + """Parse Retry-After while accepting the SDK's millisecond extension.""" + milliseconds = _number(headers.get("retry-after-ms")) + if milliseconds is not None: + return milliseconds / 1000 + seconds = _number(headers.get("retry-after")) + if seconds is not None: + return seconds + if isinstance(body, Mapping): + payload = cast(Mapping[str, object], body) + details = payload.get("details") + if isinstance(details, Mapping): + return _number(cast(Mapping[str, object], details).get("retry_after")) + return None + + +def status_error_details(response: httpx.Response, body: object) -> ErrorDetails: + payload: Mapping[str, object] = cast(Mapping[str, object], body) if isinstance(body, Mapping) else {} + header_code = response.headers.get("x-runloop-error-code") + body_code = payload.get("error") + code = header_code or (body_code if isinstance(body_code, str) else None) or f"http_{response.status_code}" + body_phase = payload.get("phase") + phase = body_phase if isinstance(body_phase, str) else "api" + retryable_value = payload.get("retryable") + retryable = retryable_value if isinstance(retryable_value, bool) else response.status_code in {408, 409, 429} + if response.status_code >= 500 and not isinstance(retryable_value, bool): + retryable = True + request_id: str | None = response.headers.get("x-runloop-request-id") + body_request_id = payload.get("request_id") + if request_id is None and isinstance(body_request_id, str): + request_id = body_request_id + return ErrorDetails( + code=code, + phase=phase, + retryable=retryable, + request_id=request_id, + retry_after=parse_retry_after(response.headers, cast(object, body)), + ) + + +def transport_error_details(error: BaseException) -> ErrorDetails: + if isinstance(error, httpx.ConnectTimeout): + return ErrorDetails("connection_timeout", "connect", True) + if isinstance(error, httpx.WriteTimeout): + return ErrorDetails("request_write_timeout", "request_write", False) + if isinstance(error, httpx.WriteError): + return ErrorDetails("request_write_failed", "request_write", False) + if isinstance(error, httpx.ReadTimeout): + return ErrorDetails("response_read_timeout", "response_read", False) + if isinstance(error, httpx.RemoteProtocolError): + if "idle_timeout" in str(error).lower(): + return ErrorDetails("http2_idle_timeout", "response_read", False) + return ErrorDetails("http2_protocol_error", "transport", False) + if isinstance(error, httpx.TimeoutException): + return ErrorDetails("connection_timeout", "connect", True) + return ErrorDetails("connection_failed", "connect", isinstance(error, httpx.ConnectError)) + + +def is_safe_transport_retry(error: BaseException) -> bool: + """Only retry failures that prove the request body was not partially sent.""" + # Preserve the generated client's handling of non-HTTPX exceptions (for + # example a pre-send auth hook failure). HTTPX errors carry enough phase + # information for the stricter partial-write audit below. + if not isinstance(error, httpx.HTTPError): + return True + return isinstance(error, (httpx.ConnectTimeout, httpx.ConnectError)) diff --git a/src/runloop_api_client/lib/tunnel_readiness.py b/src/runloop_api_client/lib/tunnel_readiness.py new file mode 100644 index 000000000..e0857ee1d --- /dev/null +++ b/src/runloop_api_client/lib/tunnel_readiness.py @@ -0,0 +1,79 @@ +"""Bounded retry helpers for tunnel service readiness.""" + +from __future__ import annotations + +import time +import inspect +from typing import TypeVar, Callable, Awaitable + +from .._exceptions import APIStatusError + +T = TypeVar("T") + + +def _timeout(error: APIStatusError, *, port: int, path: str, timeout_seconds: float, attempts: int) -> None: + message = f"Tunnel service was not ready for port {port} path {path!r} within {timeout_seconds:g} seconds." + error.message = message + error.args = (message,) + error.attempts = attempts + raise error + + +def wait_for_tunnel_service( + operation: Callable[[], T], + *, + port: int, + path: str = "/", + timeout_seconds: float = 30.0, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, +) -> T: + """Retry only ``tunnel_service_not_ready`` until a bounded deadline.""" + if timeout_seconds <= 0: + raise ValueError("timeout_seconds must be greater than zero") + deadline = clock() + timeout_seconds + attempts = 0 + while True: + attempts += 1 + try: + return operation() + except APIStatusError as error: + error.attempts = attempts + if error.code != "tunnel_service_not_ready": + raise + remaining = deadline - clock() + if remaining <= 0 or attempts >= 1000: + _timeout(error, port=port, path=path, timeout_seconds=timeout_seconds, attempts=attempts) + delay = error.retry_after if error.retry_after is not None else 0.5 + sleep(min(max(delay, 0), remaining)) + + +async def async_wait_for_tunnel_service( + operation: Callable[[], Awaitable[T]], + *, + port: int, + path: str = "/", + timeout_seconds: float = 30.0, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], Awaitable[None]], +) -> T: + """Async counterpart to :func:`wait_for_tunnel_service`.""" + if timeout_seconds <= 0: + raise ValueError("timeout_seconds must be greater than zero") + deadline = clock() + timeout_seconds + attempts = 0 + while True: + attempts += 1 + try: + return await operation() + except APIStatusError as error: + error.attempts = attempts + if error.code != "tunnel_service_not_ready": + raise + remaining = deadline - clock() + if remaining <= 0 or attempts >= 1000: + _timeout(error, port=port, path=path, timeout_seconds=timeout_seconds, attempts=attempts) + delay = error.retry_after if error.retry_after is not None else 0.5 + result = sleep(min(max(delay, 0), remaining)) + if inspect.isawaitable(result): + await result diff --git a/src/runloop_api_client/sdk/async_devbox.py b/src/runloop_api_client/sdk/async_devbox.py index bed785a7d..1146dde33 100644 --- a/src/runloop_api_client/sdk/async_devbox.py +++ b/src/runloop_api_client/sdk/async_devbox.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time import asyncio import logging from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence, Awaitable, cast @@ -37,6 +38,7 @@ from ..lib.polling import PollingConfig from ..types.devboxes import ExecutionUpdateChunk from .async_execution import AsyncExecution, _AsyncStreamingGroup +from ..lib.tunnel_readiness import async_wait_for_tunnel_service from .async_execution_result import AsyncExecutionResult from ..types.devbox_execute_async_params import DevboxNiceExecuteAsyncParams from ..types.devboxes.devbox_logs_list_view import DevboxLogsListView @@ -830,6 +832,27 @@ async def enable_tunnel( **params, ) + async def wait_for_tunnel_ready( + self, + port: int, + path: str = "/", + *, + timeout_seconds: float = 30.0, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, + **params: Unpack[SDKDevboxEnableTunnelParams], + ) -> TunnelView: + """Enable a tunnel, waiting through transient service readiness failures.""" + client = self._devbox._client.with_options(max_retries=0) + return await async_wait_for_tunnel_service( + lambda: client.devboxes.enable_tunnel(self._devbox.id, **params), + port=port, + path=path, + timeout_seconds=timeout_seconds, + clock=clock, + sleep=sleep, + ) + async def remove_tunnel( self, **params: Unpack[SDKDevboxRemoveTunnelParams], diff --git a/src/runloop_api_client/sdk/devbox.py b/src/runloop_api_client/sdk/devbox.py index c97ea2682..f8569d820 100644 --- a/src/runloop_api_client/sdk/devbox.py +++ b/src/runloop_api_client/sdk/devbox.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time import logging import threading from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence @@ -39,6 +40,7 @@ from ..lib.polling import PollingConfig from ..types.devboxes import ExecutionUpdateChunk from .execution_result import ExecutionResult +from ..lib.tunnel_readiness import wait_for_tunnel_service from ..types.devbox_execute_async_params import DevboxNiceExecuteAsyncParams from ..types.devboxes.devbox_logs_list_view import DevboxLogsListView from ..types.devbox_async_execution_detail_view import DevboxAsyncExecutionDetailView @@ -833,6 +835,31 @@ def enable_tunnel( **params, ) + def wait_for_tunnel_ready( + self, + port: int, + path: str = "/", + *, + timeout_seconds: float = 30.0, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + **params: Unpack[SDKDevboxEnableTunnelParams], + ) -> TunnelView: + """Enable a tunnel, waiting through transient service readiness failures. + + The generated client's own retries are disabled so this bounded helper + owns the deadline and preserves the final normalized error. + """ + client = self._devbox._client.with_options(max_retries=0) + return wait_for_tunnel_service( + lambda: client.devboxes.enable_tunnel(self._devbox.id, **params), + port=port, + path=path, + timeout_seconds=timeout_seconds, + clock=clock, + sleep=sleep, + ) + def remove_tunnel( self, **params: Unpack[SDKDevboxRemoveTunnelParams], diff --git a/tests/test_transport_error_contract.py b/tests/test_transport_error_contract.py new file mode 100644 index 000000000..26a965b02 --- /dev/null +++ b/tests/test_transport_error_contract.py @@ -0,0 +1,248 @@ +"""Hermetic reproduction of the 2026-08-16 transport incident families. + +Run with: uv run pytest tests/test_transport_error_contract.py -n 0 +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import Mock, AsyncMock +from collections.abc import Callable + +import httpx +import pytest + +from runloop_api_client import Runloop, APIError, AsyncRunloop, APIStatusError +from runloop_api_client.sdk.devbox import Devbox +from runloop_api_client.sdk.async_devbox import AsyncDevbox + + +def customer_shape(error: APIError) -> dict[str, object]: + """The supported diagnostic surface; callers need not inspect __cause__.""" + return { + "type": type(error).__name__, + "code": error.code, + "phase": error.phase, + "retryable": error.retryable, + "request_id": error.request_id, + "retry_after": error.retry_after, + "attempts": error.attempts, + "cause": type(error.cause).__name__ if error.cause else None, + } + + +INCIDENTS: list[tuple[str, Callable[[httpx.Request], httpx.HTTPError], dict[str, object]]] = [ + ( + "connect", + lambda request: httpx.ConnectTimeout("connect timed out", request=request), + { + "type": "APITimeoutError", + "code": "connection_timeout", + "phase": "connect", + "retryable": True, + "request_id": None, + "retry_after": None, + "attempts": 1, + "cause": "ConnectTimeout", + }, + ), + ( + "write", + lambda request: httpx.WriteError("write failed", request=request), + { + "type": "APIConnectionError", + "code": "request_write_failed", + "phase": "request_write", + "retryable": False, + "request_id": None, + "retry_after": None, + "attempts": 1, + "cause": "WriteError", + }, + ), + ( + "idle", + lambda request: httpx.RemoteProtocolError( + "", request=request + ), + { + "type": "APIConnectionError", + "code": "http2_idle_timeout", + "phase": "response_read", + "retryable": False, + "request_id": None, + "retry_after": None, + "attempts": 1, + "cause": "RemoteProtocolError", + }, + ), +] + + +@pytest.mark.parametrize(("_name", "failure", "expected"), INCIDENTS) +@pytest.mark.parametrize("operation", ["upload_file", "enable_tunnel"]) +def test_sync_incident_transport_shapes( + _name: str, + failure: Callable[[httpx.Request], httpx.HTTPError], + expected: dict[str, object], + operation: str, +) -> None: + def handler(request: httpx.Request) -> httpx.Response: + raise failure(request) + + with Runloop( + bearer_token="test", + base_url="https://example.test", + max_retries=0, + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) as client: + with pytest.raises(APIError) as caught: + if operation == "upload_file": + client.devboxes.upload_file("dbx", path="data.bin", file=b"data") + else: + client.devboxes.enable_tunnel("dbx") + + assert customer_shape(caught.value) == expected + assert caught.value.__cause__ is caught.value.cause + + +@pytest.mark.parametrize(("_name", "failure", "expected"), INCIDENTS) +@pytest.mark.parametrize("operation", ["upload_file", "enable_tunnel"]) +async def test_async_incident_transport_shapes( + _name: str, + failure: Callable[[httpx.Request], httpx.HTTPError], + expected: dict[str, object], + operation: str, +) -> None: + async def handler(request: httpx.Request) -> httpx.Response: + raise failure(request) + + async with AsyncRunloop( + bearer_token="test", + base_url="https://example.test", + max_retries=0, + http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) as client: + with pytest.raises(APIError) as caught: + if operation == "upload_file": + await client.devboxes.upload_file("dbx", path="data.bin", file=b"data") + else: + await client.devboxes.enable_tunnel("dbx") + + assert customer_shape(caught.value) == expected + assert caught.value.__cause__ is caught.value.cause + + +def structured_not_ready(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 503, + request=request, + headers={ + "X-Runloop-Error-Code": "tunnel_service_not_ready", + "X-Runloop-Request-Id": "req_header", + "Retry-After": "1.5", + "X-Should-Retry": "false", + }, + json={ + "error": "legacy_tunnel_error", + "message": "Tunnel service is still becoming ready.", + "retryable": True, + "phase": "tunnel_readiness", + "request_id": "req_body", + "details": {"port": 8080, "path": "/health"}, + }, + ) + + +def test_sync_structured_tunnel_not_ready_shape() -> None: + client = Runloop( + bearer_token="test", + base_url="https://example.test", + max_retries=0, + http_client=httpx.Client(transport=httpx.MockTransport(structured_not_ready)), + ) + with client, pytest.raises(APIError) as caught: + client.devboxes.enable_tunnel("dbx") + assert customer_shape(caught.value) == { + "type": "InternalServerError", + "code": "tunnel_service_not_ready", + "phase": "tunnel_readiness", + "retryable": True, + "request_id": "req_header", + "retry_after": 1.5, + "attempts": 1, + "cause": None, + } + + +async def test_async_structured_tunnel_not_ready_shape() -> None: + async def handler(request: httpx.Request) -> httpx.Response: + return structured_not_ready(request) + + client = AsyncRunloop( + bearer_token="test", + base_url="https://example.test", + max_retries=0, + http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + async with client: + with pytest.raises(APIError) as caught: + await client.devboxes.enable_tunnel("dbx") + assert caught.value.code == "tunnel_service_not_ready" + assert caught.value.request_id == "req_header" + assert caught.value.retry_after == 1.5 + + +def readiness_error(*, code: str = "tunnel_service_not_ready", retry_after: str = "2") -> APIStatusError: + request = httpx.Request("POST", "https://example.test/v1/devboxes/dbx/enable_tunnel") + response = httpx.Response( + 503, + request=request, + headers={"X-Runloop-Error-Code": code, "Retry-After": retry_after}, + ) + return APIStatusError("not ready", response=response, body={"error": code, "retryable": True}) + + +def test_high_level_tunnel_readiness_honors_retry_after_and_deadline() -> None: + clock_value = [0.0] + delays: list[float] = [] + + def sleep(delay: float) -> None: + delays.append(delay) + clock_value[0] += delay + + generated = Mock() + generated.with_options.return_value = generated + generated.devboxes.enable_tunnel.side_effect = [readiness_error(), readiness_error()] + + with pytest.raises(APIStatusError) as caught: + Devbox(generated, "dbx").net.wait_for_tunnel_ready( + 8080, + "/health", + timeout_seconds=2, + clock=lambda: clock_value[0], + sleep=sleep, + ) + + generated.with_options.assert_called_once_with(max_retries=0) + assert delays == [2.0] + assert caught.value.code == "tunnel_service_not_ready" + assert caught.value.attempts == 2 + assert "port 8080" in str(caught.value) + assert "'/health'" in str(caught.value) + + +async def test_async_high_level_tunnel_readiness_stops_on_terminal_error() -> None: + generated = Mock() + generated.with_options.return_value = generated + generated.devboxes = SimpleNamespace( + enable_tunnel=AsyncMock(side_effect=readiness_error(code="tunnel_unavailable")) + ) + sleep = AsyncMock() + + with pytest.raises(APIStatusError) as caught: + await AsyncDevbox(generated, "dbx").net.wait_for_tunnel_ready(3000, sleep=sleep) + + assert caught.value.code == "tunnel_unavailable" + assert caught.value.attempts == 1 + sleep.assert_not_awaited() From 3443be0cc66b6dafd67c3c04d0d930da9701d622 Mon Sep 17 00:00:00 2001 From: Reflex Date: Sun, 16 Aug 2026 21:25:00 +0000 Subject: [PATCH 2/4] fix(tunnel): poll established service readiness --- src/runloop_api_client/lib/error_contract.py | 5 + .../lib/tunnel_readiness.py | 154 ++++++++++++++---- src/runloop_api_client/sdk/async_devbox.py | 36 +++- src/runloop_api_client/sdk/devbox.py | 40 +++-- tests/test_transport_error_contract.py | 140 +++++++++++++--- 5 files changed, 299 insertions(+), 76 deletions(-) diff --git a/src/runloop_api_client/lib/error_contract.py b/src/runloop_api_client/lib/error_contract.py index 92473344d..5783e38ef 100644 --- a/src/runloop_api_client/lib/error_contract.py +++ b/src/runloop_api_client/lib/error_contract.py @@ -6,6 +6,8 @@ from __future__ import annotations +import time +import email.utils from typing import Mapping, cast from dataclasses import dataclass @@ -37,6 +39,9 @@ def parse_retry_after(headers: httpx.Headers, body: object = None) -> float | No seconds = _number(headers.get("retry-after")) if seconds is not None: return seconds + retry_date = email.utils.parsedate_tz(headers.get("retry-after")) + if retry_date is not None: + return max(float(email.utils.mktime_tz(retry_date) - time.time()), 0) if isinstance(body, Mapping): payload = cast(Mapping[str, object], body) details = payload.get("details") diff --git a/src/runloop_api_client/lib/tunnel_readiness.py b/src/runloop_api_client/lib/tunnel_readiness.py index e0857ee1d..eb7540e95 100644 --- a/src/runloop_api_client/lib/tunnel_readiness.py +++ b/src/runloop_api_client/lib/tunnel_readiness.py @@ -1,17 +1,73 @@ -"""Bounded retry helpers for tunnel service readiness.""" +"""Bounded polling of an established tunnel's service endpoint.""" from __future__ import annotations +import json import time -import inspect -from typing import TypeVar, Callable, Awaitable +from typing import Mapping, Callable, Awaitable, cast -from .._exceptions import APIStatusError +import httpx -T = TypeVar("T") +from .._exceptions import APIError, APIStatusError, APITimeoutError, APIConnectionError +from .error_contract import is_safe_transport_retry -def _timeout(error: APIStatusError, *, port: int, path: str, timeout_seconds: float, attempts: int) -> None: +def tunnel_url(*, api_host: str, tunnel_key: str, port: int, path: str = "/") -> str: + """Construct the established tunnel URL using the SDK's domain convention.""" + if not 1 <= port <= 65535: + raise ValueError("port must be between 1 and 65535") + if not path.startswith("/"): + raise ValueError("path must start with '/'") + if not tunnel_key or not all( + character.isascii() and (character.isalnum() or character in "-_") for character in tunnel_key + ): + raise ValueError("tunnel_key contains characters that are unsafe in a tunnel hostname") + base_domain = api_host[4:] if api_host.startswith("api.") else api_host + return f"https://{port}-{tunnel_key}.tunnel.{base_domain}{path}" + + +def tunnel_auth_headers(*, auth_mode: str, auth_token: str | None, request: httpx.Request) -> Mapping[str, str]: + """Return tunnel authentication without leaking the Runloop API bearer token.""" + if auth_mode != "authenticated": + return {} + if auth_token: + return {"X-Runloop-Tunnel-Authorization": f"Bearer {auth_token}"} + error = APIConnectionError( + message="Authenticated tunnel is missing its tunnel authorization token.", + request=request, + ) + error.code = "tunnel_authentication_required" + error.phase = "tunnel_readiness" + error.retryable = False + raise error + + +def _status_error(response: httpx.Response, attempts: int) -> APIStatusError: + body: object = response.text + try: + body = cast(object, json.loads(response.text)) + except (TypeError, ValueError): + pass + payload = cast(Mapping[str, object], body) if isinstance(body, dict) else None + body_message = payload.get("message") if payload is not None else None + message = ( + body_message + if isinstance(body_message, str) + else f"Tunnel readiness check failed with HTTP status {response.status_code}." + ) + error = APIStatusError(message, response=response, body=cast(object, body), attempts=attempts) + if error.phase == "api": + error.phase = "tunnel_readiness" + return error + + +def _connection_error(error: httpx.HTTPError, request: httpx.Request, attempts: int) -> APIConnectionError: + if isinstance(error, httpx.TimeoutException): + return APITimeoutError(request=request, cause=error, attempts=attempts) + return APIConnectionError(request=request, cause=error, attempts=attempts) + + +def _raise_timeout(error: APIError, *, port: int, path: str, timeout_seconds: float, attempts: int) -> None: message = f"Tunnel service was not ready for port {port} path {path!r} within {timeout_seconds:g} seconds." error.message = message error.args = (message,) @@ -19,61 +75,93 @@ def _timeout(error: APIStatusError, *, port: int, path: str, timeout_seconds: fl raise error +def _retry_delay(error: APIError) -> float: + return error.retry_after if error.retry_after is not None else 0.5 + + +def _is_transient_status(error: APIStatusError) -> bool: + if error.code == "tunnel_unavailable": + return False + return error.code == "tunnel_service_not_ready" and error.response.headers.get("x-should-retry") != "false" + + def wait_for_tunnel_service( - operation: Callable[[], T], + request: Callable[[float], httpx.Response], *, port: int, path: str = "/", timeout_seconds: float = 30.0, clock: Callable[[], float] = time.monotonic, sleep: Callable[[float], None] = time.sleep, -) -> T: - """Retry only ``tunnel_service_not_ready`` until a bounded deadline.""" +) -> None: + """Poll an established tunnel URL until it returns a successful response.""" if timeout_seconds <= 0: raise ValueError("timeout_seconds must be greater than zero") deadline = clock() + timeout_seconds attempts = 0 + failure: APIError | None = None while True: + remaining = deadline - clock() + if remaining <= 0: + if failure is None: + raise ValueError("tunnel readiness deadline expired before the first request") + _raise_timeout(failure, port=port, path=path, timeout_seconds=timeout_seconds, attempts=attempts) attempts += 1 try: - return operation() - except APIStatusError as error: - error.attempts = attempts - if error.code != "tunnel_service_not_ready": - raise - remaining = deadline - clock() - if remaining <= 0 or attempts >= 1000: - _timeout(error, port=port, path=path, timeout_seconds=timeout_seconds, attempts=attempts) - delay = error.retry_after if error.retry_after is not None else 0.5 - sleep(min(max(delay, 0), remaining)) + response = request(remaining) + if response.is_success: + return + failure = _status_error(response, attempts) + if not _is_transient_status(failure): + raise failure + except httpx.HTTPError as cause: + request_object = cause.request + failure = _connection_error(cause, request_object, attempts) + if not is_safe_transport_retry(cause): + raise failure from cause + + remaining = deadline - clock() + if remaining <= 0 or attempts >= 1000: + _raise_timeout(failure, port=port, path=path, timeout_seconds=timeout_seconds, attempts=attempts) + sleep(min(max(_retry_delay(failure), 0), remaining)) async def async_wait_for_tunnel_service( - operation: Callable[[], Awaitable[T]], + request: Callable[[float], Awaitable[httpx.Response]], *, port: int, path: str = "/", timeout_seconds: float = 30.0, clock: Callable[[], float] = time.monotonic, sleep: Callable[[float], Awaitable[None]], -) -> T: +) -> None: """Async counterpart to :func:`wait_for_tunnel_service`.""" if timeout_seconds <= 0: raise ValueError("timeout_seconds must be greater than zero") deadline = clock() + timeout_seconds attempts = 0 + failure: APIError | None = None while True: + remaining = deadline - clock() + if remaining <= 0: + if failure is None: + raise ValueError("tunnel readiness deadline expired before the first request") + _raise_timeout(failure, port=port, path=path, timeout_seconds=timeout_seconds, attempts=attempts) attempts += 1 try: - return await operation() - except APIStatusError as error: - error.attempts = attempts - if error.code != "tunnel_service_not_ready": - raise - remaining = deadline - clock() - if remaining <= 0 or attempts >= 1000: - _timeout(error, port=port, path=path, timeout_seconds=timeout_seconds, attempts=attempts) - delay = error.retry_after if error.retry_after is not None else 0.5 - result = sleep(min(max(delay, 0), remaining)) - if inspect.isawaitable(result): - await result + response = await request(remaining) + if response.is_success: + return + failure = _status_error(response, attempts) + if not _is_transient_status(failure): + raise failure + except httpx.HTTPError as cause: + request_object = cause.request + failure = _connection_error(cause, request_object, attempts) + if not is_safe_transport_retry(cause): + raise failure from cause + + remaining = deadline - clock() + if remaining <= 0 or attempts >= 1000: + _raise_timeout(failure, port=port, path=path, timeout_seconds=timeout_seconds, attempts=attempts) + await sleep(min(max(_retry_delay(failure), 0), remaining)) diff --git a/src/runloop_api_client/sdk/async_devbox.py b/src/runloop_api_client/sdk/async_devbox.py index 1146dde33..70196581a 100644 --- a/src/runloop_api_client/sdk/async_devbox.py +++ b/src/runloop_api_client/sdk/async_devbox.py @@ -8,6 +8,8 @@ from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence, Awaitable, cast from typing_extensions import Unpack, override +import httpx + from ..types import ( DevboxView, TunnelView, @@ -38,7 +40,7 @@ from ..lib.polling import PollingConfig from ..types.devboxes import ExecutionUpdateChunk from .async_execution import AsyncExecution, _AsyncStreamingGroup -from ..lib.tunnel_readiness import async_wait_for_tunnel_service +from ..lib.tunnel_readiness import tunnel_url, tunnel_auth_headers, async_wait_for_tunnel_service from .async_execution_result import AsyncExecutionResult from ..types.devbox_execute_async_params import DevboxNiceExecuteAsyncParams from ..types.devboxes.devbox_logs_list_view import DevboxLogsListView @@ -838,20 +840,36 @@ async def wait_for_tunnel_ready( path: str = "/", *, timeout_seconds: float = 30.0, + http_client: httpx.AsyncClient | None = None, clock: Callable[[], float] = time.monotonic, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, **params: Unpack[SDKDevboxEnableTunnelParams], ) -> TunnelView: - """Enable a tunnel, waiting through transient service readiness failures.""" + """Enable a tunnel and poll the requested service until it is ready.""" client = self._devbox._client.with_options(max_retries=0) - return await async_wait_for_tunnel_service( - lambda: client.devboxes.enable_tunnel(self._devbox.id, **params), - port=port, - path=path, - timeout_seconds=timeout_seconds, - clock=clock, - sleep=sleep, + enable_params: dict[str, Any] = dict(params) + enable_params.setdefault("timeout", timeout_seconds) + tunnel = await client.devboxes.enable_tunnel(self._devbox.id, **enable_params) + url = tunnel_url(api_host=client.base_url.host, tunnel_key=tunnel.tunnel_key, port=port, path=path) + headers = tunnel_auth_headers( + auth_mode=tunnel.auth_mode, + auth_token=tunnel.auth_token, + request=httpx.Request("GET", url), ) + probe_client = http_client or httpx.AsyncClient(follow_redirects=True) + try: + await async_wait_for_tunnel_service( + lambda remaining: probe_client.get(url, headers=headers, timeout=remaining, follow_redirects=True), + port=port, + path=path, + timeout_seconds=timeout_seconds, + clock=clock, + sleep=sleep, + ) + finally: + if http_client is None: + await probe_client.aclose() + return tunnel async def remove_tunnel( self, diff --git a/src/runloop_api_client/sdk/devbox.py b/src/runloop_api_client/sdk/devbox.py index f8569d820..9329429d7 100644 --- a/src/runloop_api_client/sdk/devbox.py +++ b/src/runloop_api_client/sdk/devbox.py @@ -8,6 +8,8 @@ from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence from typing_extensions import Unpack, override +import httpx + from ..types import ( DevboxView, TunnelView, @@ -40,7 +42,7 @@ from ..lib.polling import PollingConfig from ..types.devboxes import ExecutionUpdateChunk from .execution_result import ExecutionResult -from ..lib.tunnel_readiness import wait_for_tunnel_service +from ..lib.tunnel_readiness import tunnel_url, tunnel_auth_headers, wait_for_tunnel_service from ..types.devbox_execute_async_params import DevboxNiceExecuteAsyncParams from ..types.devboxes.devbox_logs_list_view import DevboxLogsListView from ..types.devbox_async_execution_detail_view import DevboxAsyncExecutionDetailView @@ -841,24 +843,40 @@ def wait_for_tunnel_ready( path: str = "/", *, timeout_seconds: float = 30.0, + http_client: httpx.Client | None = None, clock: Callable[[], float] = time.monotonic, sleep: Callable[[float], None] = time.sleep, **params: Unpack[SDKDevboxEnableTunnelParams], ) -> TunnelView: - """Enable a tunnel, waiting through transient service readiness failures. + """Enable a tunnel and poll the requested service until it is ready. - The generated client's own retries are disabled so this bounded helper - owns the deadline and preserves the final normalized error. + The readiness probe uses the tunnel-specific authorization token when + required and never forwards the Runloop API bearer token. """ client = self._devbox._client.with_options(max_retries=0) - return wait_for_tunnel_service( - lambda: client.devboxes.enable_tunnel(self._devbox.id, **params), - port=port, - path=path, - timeout_seconds=timeout_seconds, - clock=clock, - sleep=sleep, + enable_params: dict[str, Any] = dict(params) + enable_params.setdefault("timeout", timeout_seconds) + tunnel = client.devboxes.enable_tunnel(self._devbox.id, **enable_params) + url = tunnel_url(api_host=client.base_url.host, tunnel_key=tunnel.tunnel_key, port=port, path=path) + headers = tunnel_auth_headers( + auth_mode=tunnel.auth_mode, + auth_token=tunnel.auth_token, + request=httpx.Request("GET", url), ) + probe_client = http_client or httpx.Client(follow_redirects=True) + try: + wait_for_tunnel_service( + lambda remaining: probe_client.get(url, headers=headers, timeout=remaining, follow_redirects=True), + port=port, + path=path, + timeout_seconds=timeout_seconds, + clock=clock, + sleep=sleep, + ) + finally: + if http_client is None: + probe_client.close() + return tunnel def remove_tunnel( self, diff --git a/tests/test_transport_error_contract.py b/tests/test_transport_error_contract.py index 26a965b02..b361aa273 100644 --- a/tests/test_transport_error_contract.py +++ b/tests/test_transport_error_contract.py @@ -175,6 +175,18 @@ def test_sync_structured_tunnel_not_ready_shape() -> None: } +def test_retry_after_http_date_is_exposed_without_negative_delay() -> None: + request = httpx.Request("GET", "https://8080-key.tunnel.runloop.ai/health") + response = httpx.Response( + 503, + request=request, + headers={"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}, + json={"error": "tunnel_service_not_ready", "retryable": True}, + ) + error = APIStatusError("not ready", response=response, body=response.json()) + assert error.retry_after == 0 + + async def test_async_structured_tunnel_not_ready_shape() -> None: async def handler(request: httpx.Request) -> httpx.Response: return structured_not_ready(request) @@ -193,56 +205,138 @@ async def handler(request: httpx.Request) -> httpx.Response: assert caught.value.retry_after == 1.5 -def readiness_error(*, code: str = "tunnel_service_not_ready", retry_after: str = "2") -> APIStatusError: - request = httpx.Request("POST", "https://example.test/v1/devboxes/dbx/enable_tunnel") - response = httpx.Response( - 503, - request=request, - headers={"X-Runloop-Error-Code": code, "Retry-After": retry_after}, - ) - return APIStatusError("not ready", response=response, body={"error": code, "retryable": True}) - - -def test_high_level_tunnel_readiness_honors_retry_after_and_deadline() -> None: +def test_high_level_tunnel_readiness_polls_established_authenticated_url() -> None: clock_value = [0.0] delays: list[float] = [] + requests: list[httpx.Request] = [] def sleep(delay: float) -> None: delays.append(delay) clock_value[0] += delay + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if len(requests) == 1: + return httpx.Response( + 503, + request=request, + headers={ + "X-Runloop-Error-Code": "tunnel_service_not_ready", + "X-Runloop-Request-Id": "req_probe", + "Retry-After": "1.5", + }, + json={ + "error": "tunnel_service_not_ready", + "message": "Tunnel routing is not ready for this backend.", + "retryable": True, + "phase": "tunnel_readiness", + "request_id": "req_body", + "details": {"port": 8080, "path": "/health"}, + }, + ) + return httpx.Response(204, request=request) + generated = Mock() generated.with_options.return_value = generated - generated.devboxes.enable_tunnel.side_effect = [readiness_error(), readiness_error()] + generated.base_url = httpx.URL("https://api.runloop.ai") + tunnel = SimpleNamespace( + tunnel_key="tunnel-key", + auth_mode="authenticated", + auth_token="tunnel-secret", + ) + generated.devboxes.enable_tunnel.return_value = tunnel + probe_client = httpx.Client(transport=httpx.MockTransport(handler)) - with pytest.raises(APIStatusError) as caught: - Devbox(generated, "dbx").net.wait_for_tunnel_ready( + with probe_client: + result = Devbox(generated, "dbx").net.wait_for_tunnel_ready( 8080, "/health", - timeout_seconds=2, + timeout_seconds=3, + http_client=probe_client, clock=lambda: clock_value[0], sleep=sleep, ) generated.with_options.assert_called_once_with(max_retries=0) - assert delays == [2.0] - assert caught.value.code == "tunnel_service_not_ready" - assert caught.value.attempts == 2 - assert "port 8080" in str(caught.value) - assert "'/health'" in str(caught.value) + generated.devboxes.enable_tunnel.assert_called_once_with("dbx", timeout=3) + assert result is tunnel + assert delays == [1.5] + assert [str(request.url) for request in requests] == [ + "https://8080-tunnel-key.tunnel.runloop.ai/health", + "https://8080-tunnel-key.tunnel.runloop.ai/health", + ] + assert all(request.headers["X-Runloop-Tunnel-Authorization"] == "Bearer tunnel-secret" for request in requests) + assert all("authorization" not in request.headers for request in requests) async def test_async_high_level_tunnel_readiness_stops_on_terminal_error() -> None: generated = Mock() generated.with_options.return_value = generated + generated.base_url = httpx.URL("https://api.runloop.ai") generated.devboxes = SimpleNamespace( - enable_tunnel=AsyncMock(side_effect=readiness_error(code="tunnel_unavailable")) + enable_tunnel=AsyncMock(return_value=SimpleNamespace(tunnel_key="key", auth_mode="open", auth_token=None)) ) sleep = AsyncMock() - with pytest.raises(APIStatusError) as caught: - await AsyncDevbox(generated, "dbx").net.wait_for_tunnel_ready(3000, sleep=sleep) + async def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 503, + request=request, + headers={"X-Runloop-Error-Code": "tunnel_unavailable", "Retry-After": "0"}, + json={ + "error": "tunnel_unavailable", + "message": "Tunnel is unavailable.", + "retryable": False, + "phase": "tunnel_readiness", + }, + ) + + probe_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + async with probe_client: + with pytest.raises(APIStatusError) as caught: + await AsyncDevbox(generated, "dbx").net.wait_for_tunnel_ready( + 3000, + http_client=probe_client, + sleep=sleep, + ) assert caught.value.code == "tunnel_unavailable" assert caught.value.attempts == 1 + assert caught.value.retryable is False sleep.assert_not_awaited() + + +def test_tunnel_readiness_timeout_identifies_port_and_path() -> None: + clock_value = [0.0] + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 503, + request=request, + headers={"X-Runloop-Error-Code": "tunnel_service_not_ready", "Retry-After": "2"}, + json={"error": "tunnel_service_not_ready", "retryable": True}, + ) + + def sleep(delay: float) -> None: + clock_value[0] += delay + + generated = Mock() + generated.with_options.return_value = generated + generated.base_url = httpx.URL("https://api.runloop.ai") + generated.devboxes.enable_tunnel.return_value = SimpleNamespace(tunnel_key="key", auth_mode="open", auth_token=None) + probe_client = httpx.Client(transport=httpx.MockTransport(handler)) + + with probe_client, pytest.raises(APIStatusError) as caught: + Devbox(generated, "dbx").net.wait_for_tunnel_ready( + 9090, + "/ready", + timeout_seconds=2, + http_client=probe_client, + clock=lambda: clock_value[0], + sleep=sleep, + ) + + assert caught.value.code == "tunnel_service_not_ready" + assert caught.value.attempts == 1 + assert "port 9090" in str(caught.value) + assert "'/ready'" in str(caught.value) From 43fe1f1b7193baeb787a97a07da55b2cd59d7563 Mon Sep 17 00:00:00 2001 From: Reflex Date: Sun, 16 Aug 2026 21:31:28 +0000 Subject: [PATCH 3/4] fix(transport): address validated retry findings --- src/runloop_api_client/_base_client.py | 46 ++++- src/runloop_api_client/lib/error_contract.py | 8 +- .../lib/tunnel_readiness.py | 33 +++- src/runloop_api_client/sdk/async_devbox.py | 11 +- src/runloop_api_client/sdk/devbox.py | 6 +- tests/test_client.py | 4 +- tests/test_transport_error_contract.py | 163 +++++++++++++++++- 7 files changed, 254 insertions(+), 17 deletions(-) diff --git a/src/runloop_api_client/_base_client.py b/src/runloop_api_client/_base_client.py index b91507aec..46de1b499 100644 --- a/src/runloop_api_client/_base_client.py +++ b/src/runloop_api_client/_base_client.py @@ -1439,6 +1439,18 @@ def request( except httpx.HTTPStatusError as err: # thrown on 4xx and 5xx status code log.debug("Encountered httpx.HTTPStatusError", exc_info=True) + if remaining_retries > 0 and not err.response.is_closed: + try: + err.response.read() + except httpx.TimeoutException as read_error: + raise APITimeoutError( + request=request, cause=read_error, attempts=retries_taken + 1 + ) from read_error + except httpx.HTTPError as read_error: + raise APIConnectionError( + request=request, cause=read_error, attempts=retries_taken + 1 + ) from read_error + if remaining_retries > 0 and self._should_retry(err.response): err.response.close() self._sleep_for_retry( @@ -1452,7 +1464,16 @@ def request( # If the response is streamed then we need to explicitly read the response # to completion before attempting to access the response text. if not err.response.is_closed: - err.response.read() + try: + err.response.read() + except httpx.TimeoutException as read_error: + raise APITimeoutError( + request=request, cause=read_error, attempts=retries_taken + 1 + ) from read_error + except httpx.HTTPError as read_error: + raise APIConnectionError( + request=request, cause=read_error, attempts=retries_taken + 1 + ) from read_error log.debug("Re-raising status error") raise self._make_status_error_from_response(err.response, attempts=retries_taken + 1) from None @@ -2149,6 +2170,18 @@ async def request( except httpx.HTTPStatusError as err: # thrown on 4xx and 5xx status code log.debug("Encountered httpx.HTTPStatusError", exc_info=True) + if remaining_retries > 0 and not err.response.is_closed: + try: + await err.response.aread() + except httpx.TimeoutException as read_error: + raise APITimeoutError( + request=request, cause=read_error, attempts=retries_taken + 1 + ) from read_error + except httpx.HTTPError as read_error: + raise APIConnectionError( + request=request, cause=read_error, attempts=retries_taken + 1 + ) from read_error + if remaining_retries > 0 and self._should_retry(err.response): await err.response.aclose() await self._sleep_for_retry( @@ -2162,7 +2195,16 @@ async def request( # If the response is streamed then we need to explicitly read the response # to completion before attempting to access the response text. if not err.response.is_closed: - await err.response.aread() + try: + await err.response.aread() + except httpx.TimeoutException as read_error: + raise APITimeoutError( + request=request, cause=read_error, attempts=retries_taken + 1 + ) from read_error + except httpx.HTTPError as read_error: + raise APIConnectionError( + request=request, cause=read_error, attempts=retries_taken + 1 + ) from read_error log.debug("Re-raising status error") raise self._make_status_error_from_response(err.response, attempts=retries_taken + 1) from None diff --git a/src/runloop_api_client/lib/error_contract.py b/src/runloop_api_client/lib/error_contract.py index 5783e38ef..d81208a1d 100644 --- a/src/runloop_api_client/lib/error_contract.py +++ b/src/runloop_api_client/lib/error_contract.py @@ -39,7 +39,8 @@ def parse_retry_after(headers: httpx.Headers, body: object = None) -> float | No seconds = _number(headers.get("retry-after")) if seconds is not None: return seconds - retry_date = email.utils.parsedate_tz(headers.get("retry-after")) + retry_header = headers.get("retry-after") + retry_date = email.utils.parsedate_tz(retry_header) if isinstance(retry_header, str) else None if retry_date is not None: return max(float(email.utils.mktime_tz(retry_date) - time.time()), 0) if isinstance(body, Mapping): @@ -94,9 +95,6 @@ def transport_error_details(error: BaseException) -> ErrorDetails: def is_safe_transport_retry(error: BaseException) -> bool: """Only retry failures that prove the request body was not partially sent.""" - # Preserve the generated client's handling of non-HTTPX exceptions (for - # example a pre-send auth hook failure). HTTPX errors carry enough phase - # information for the stricter partial-write audit below. if not isinstance(error, httpx.HTTPError): - return True + return False return isinstance(error, (httpx.ConnectTimeout, httpx.ConnectError)) diff --git a/src/runloop_api_client/lib/tunnel_readiness.py b/src/runloop_api_client/lib/tunnel_readiness.py index eb7540e95..9015cd965 100644 --- a/src/runloop_api_client/lib/tunnel_readiness.py +++ b/src/runloop_api_client/lib/tunnel_readiness.py @@ -72,6 +72,8 @@ def _raise_timeout(error: APIError, *, port: int, path: str, timeout_seconds: fl error.message = message error.args = (message,) error.attempts = attempts + if error.cause is not None: + raise error from error.cause raise error @@ -82,7 +84,36 @@ def _retry_delay(error: APIError) -> float: def _is_transient_status(error: APIStatusError) -> bool: if error.code == "tunnel_unavailable": return False - return error.code == "tunnel_service_not_ready" and error.response.headers.get("x-should-retry") != "false" + should_retry = error.response.headers.get("x-should-retry") + if should_retry == "false": + return False + if should_retry == "true": + return True + return error.code == "tunnel_service_not_ready" + + +def send_tunnel_probe( + client: httpx.Client, + url: str, + headers: Mapping[str, str], + timeout: float, +) -> httpx.Response: + """Send one probe without client auth or cross-origin redirect forwarding.""" + request = client.build_request("GET", url, headers=headers, timeout=timeout) + request.headers.pop("authorization", None) + return client.send(request, auth=None, follow_redirects=False) + + +async def async_send_tunnel_probe( + client: httpx.AsyncClient, + url: str, + headers: Mapping[str, str], + timeout: float, +) -> httpx.Response: + """Async counterpart to :func:`send_tunnel_probe`.""" + request = client.build_request("GET", url, headers=headers, timeout=timeout) + request.headers.pop("authorization", None) + return await client.send(request, auth=None, follow_redirects=False) def wait_for_tunnel_service( diff --git a/src/runloop_api_client/sdk/async_devbox.py b/src/runloop_api_client/sdk/async_devbox.py index 70196581a..2c052b797 100644 --- a/src/runloop_api_client/sdk/async_devbox.py +++ b/src/runloop_api_client/sdk/async_devbox.py @@ -40,7 +40,12 @@ from ..lib.polling import PollingConfig from ..types.devboxes import ExecutionUpdateChunk from .async_execution import AsyncExecution, _AsyncStreamingGroup -from ..lib.tunnel_readiness import tunnel_url, tunnel_auth_headers, async_wait_for_tunnel_service +from ..lib.tunnel_readiness import ( + tunnel_url, + tunnel_auth_headers, + async_send_tunnel_probe, + async_wait_for_tunnel_service, +) from .async_execution_result import AsyncExecutionResult from ..types.devbox_execute_async_params import DevboxNiceExecuteAsyncParams from ..types.devboxes.devbox_logs_list_view import DevboxLogsListView @@ -856,10 +861,10 @@ async def wait_for_tunnel_ready( auth_token=tunnel.auth_token, request=httpx.Request("GET", url), ) - probe_client = http_client or httpx.AsyncClient(follow_redirects=True) + probe_client = http_client or httpx.AsyncClient() try: await async_wait_for_tunnel_service( - lambda remaining: probe_client.get(url, headers=headers, timeout=remaining, follow_redirects=True), + lambda remaining: async_send_tunnel_probe(probe_client, url, headers, remaining), port=port, path=path, timeout_seconds=timeout_seconds, diff --git a/src/runloop_api_client/sdk/devbox.py b/src/runloop_api_client/sdk/devbox.py index 9329429d7..8cff2b345 100644 --- a/src/runloop_api_client/sdk/devbox.py +++ b/src/runloop_api_client/sdk/devbox.py @@ -42,7 +42,7 @@ from ..lib.polling import PollingConfig from ..types.devboxes import ExecutionUpdateChunk from .execution_result import ExecutionResult -from ..lib.tunnel_readiness import tunnel_url, tunnel_auth_headers, wait_for_tunnel_service +from ..lib.tunnel_readiness import tunnel_url, send_tunnel_probe, tunnel_auth_headers, wait_for_tunnel_service from ..types.devbox_execute_async_params import DevboxNiceExecuteAsyncParams from ..types.devboxes.devbox_logs_list_view import DevboxLogsListView from ..types.devbox_async_execution_detail_view import DevboxAsyncExecutionDetailView @@ -863,10 +863,10 @@ def wait_for_tunnel_ready( auth_token=tunnel.auth_token, request=httpx.Request("GET", url), ) - probe_client = http_client or httpx.Client(follow_redirects=True) + probe_client = http_client or httpx.Client() try: wait_for_tunnel_service( - lambda remaining: probe_client.get(url, headers=headers, timeout=remaining, follow_redirects=True), + lambda remaining: send_tunnel_probe(probe_client, url, headers, remaining), port=port, path=path, timeout_seconds=timeout_seconds, diff --git a/tests/test_client.py b/tests/test_client.py index 7728bf5bb..00be551c9 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -970,7 +970,7 @@ def retry_handler(_request: httpx.Request) -> httpx.Response: if nb_retries < failures_before_success: nb_retries += 1 if failure_mode == "exception": - raise RuntimeError("oops") + raise httpx.ConnectError("oops", request=_request) return httpx.Response(500) return httpx.Response(200) @@ -1949,7 +1949,7 @@ def retry_handler(_request: httpx.Request) -> httpx.Response: if nb_retries < failures_before_success: nb_retries += 1 if failure_mode == "exception": - raise RuntimeError("oops") + raise httpx.ConnectError("oops", request=_request) return httpx.Response(500) return httpx.Response(200) diff --git a/tests/test_transport_error_contract.py b/tests/test_transport_error_contract.py index b361aa273..bc58e86d7 100644 --- a/tests/test_transport_error_contract.py +++ b/tests/test_transport_error_contract.py @@ -15,6 +15,7 @@ from runloop_api_client import Runloop, APIError, AsyncRunloop, APIStatusError from runloop_api_client.sdk.devbox import Devbox from runloop_api_client.sdk.async_devbox import AsyncDevbox +from runloop_api_client.lib.tunnel_readiness import wait_for_tunnel_service def customer_shape(error: APIError) -> dict[str, object]: @@ -245,7 +246,10 @@ def handler(request: httpx.Request) -> httpx.Response: auth_token="tunnel-secret", ) generated.devboxes.enable_tunnel.return_value = tunnel - probe_client = httpx.Client(transport=httpx.MockTransport(handler)) + probe_client = httpx.Client( + headers={"Authorization": "Bearer api-secret"}, + transport=httpx.MockTransport(handler), + ) with probe_client: result = Devbox(generated, "dbx").net.wait_for_tunnel_ready( @@ -340,3 +344,160 @@ def sleep(delay: float) -> None: assert caught.value.attempts == 1 assert "port 9090" in str(caught.value) assert "'/ready'" in str(caught.value) + + +def test_tunnel_readiness_does_not_follow_authenticated_redirects() -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(302, request=request, headers={"Location": "https://attacker.test/token"}) + + generated = Mock() + generated.with_options.return_value = generated + generated.base_url = httpx.URL("https://api.runloop.ai") + generated.devboxes.enable_tunnel.return_value = SimpleNamespace( + tunnel_key="key", auth_mode="authenticated", auth_token="tunnel-secret" + ) + probe_client = httpx.Client( + headers={"Authorization": "Bearer api-secret"}, + transport=httpx.MockTransport(handler), + ) + + with probe_client, pytest.raises(APIStatusError) as caught: + Devbox(generated, "dbx").net.wait_for_tunnel_ready(8080, http_client=probe_client) + + assert caught.value.code == "http_302" + assert len(requests) == 1 + assert requests[0].url.host == "8080-key.tunnel.runloop.ai" + assert "authorization" not in requests[0].headers + assert requests[0].headers["X-Runloop-Tunnel-Authorization"] == "Bearer tunnel-secret" + + +def test_tunnel_readiness_timeout_preserves_connect_cause() -> None: + clock_value = [0.0] + + def request(_remaining: float) -> httpx.Response: + http_request = httpx.Request("GET", "https://8080-key.tunnel.runloop.ai/health") + raise httpx.ConnectTimeout("connect timed out", request=http_request) + + def sleep(delay: float) -> None: + clock_value[0] += delay + + with pytest.raises(APIError) as caught: + wait_for_tunnel_service( + request, + port=8080, + path="/health", + timeout_seconds=1, + clock=lambda: clock_value[0], + sleep=sleep, + ) + + assert caught.value.code == "connection_timeout" + assert caught.value.attempts == 2 + assert isinstance(caught.value.cause, httpx.ConnectTimeout) + assert caught.value.__cause__ is caught.value.cause + + +def test_explicit_should_retry_applies_to_non_terminal_readiness_error() -> None: + attempts = 0 + + def request(_remaining: float) -> httpx.Response: + nonlocal attempts + attempts += 1 + http_request = httpx.Request("GET", "https://8080-key.tunnel.runloop.ai/health") + if attempts == 1: + return httpx.Response( + 503, + request=http_request, + headers={ + "X-Runloop-Error-Code": "tunnel_backend_connect_timeout", + "X-Should-Retry": "true", + "Retry-After": "0", + }, + json={"error": "tunnel_backend_connect_timeout", "retryable": True}, + ) + return httpx.Response(204, request=http_request) + + wait_for_tunnel_service(request, port=8080, sleep=lambda _delay: None) + assert attempts == 2 + + +def test_generic_exception_is_not_retried_after_ambiguous_receipt() -> None: + attempts = 0 + + def handler(_request: httpx.Request) -> httpx.Response: + nonlocal attempts + attempts += 1 + raise RuntimeError("response hook failed") + + client = Runloop( + bearer_token="test", + base_url="https://example.test", + max_retries=2, + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + with client, pytest.raises(APIError) as caught: + client.devboxes.enable_tunnel("dbx") + + assert attempts == 1 + assert caught.value.attempts == 1 + + +def ambiguous_transfer_response(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 503, + request=request, + json={ + "error": "tunnel_backend_connection_reset", + "message": "Tunnel response ended after partial delivery.", + "retryable": True, + "phase": "response_read", + }, + ) + + +def test_streamed_structured_ambiguous_error_is_not_retried() -> None: + attempts = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal attempts + attempts += 1 + return ambiguous_transfer_response(request) + + client = Runloop( + bearer_token="test", + base_url="https://example.test", + max_retries=1, + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + with client, pytest.raises(APIStatusError) as caught: + with client.devboxes.with_streaming_response.enable_tunnel("dbx"): + pass + + assert attempts == 1 + assert caught.value.code == "tunnel_backend_connection_reset" + + +async def test_async_streamed_structured_ambiguous_error_is_not_retried() -> None: + attempts = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal attempts + attempts += 1 + return ambiguous_transfer_response(request) + + client = AsyncRunloop( + bearer_token="test", + base_url="https://example.test", + max_retries=1, + http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + async with client: + with pytest.raises(APIStatusError) as caught: + async with client.devboxes.with_streaming_response.enable_tunnel("dbx"): + pass + + assert attempts == 1 + assert caught.value.code == "tunnel_backend_connection_reset" From 3d59e3d4dab9ca834238f9e37439e3ad30cf9e25 Mon Sep 17 00:00:00 2001 From: Reflex Date: Sun, 16 Aug 2026 22:56:11 +0000 Subject: [PATCH 4/4] fix(tunnel): restrict readiness status retries --- .../lib/tunnel_readiness.py | 9 +-- tests/test_transport_error_contract.py | 64 ++++++++++++++----- 2 files changed, 49 insertions(+), 24 deletions(-) diff --git a/src/runloop_api_client/lib/tunnel_readiness.py b/src/runloop_api_client/lib/tunnel_readiness.py index 9015cd965..4ca04a8f2 100644 --- a/src/runloop_api_client/lib/tunnel_readiness.py +++ b/src/runloop_api_client/lib/tunnel_readiness.py @@ -82,14 +82,7 @@ def _retry_delay(error: APIError) -> float: def _is_transient_status(error: APIStatusError) -> bool: - if error.code == "tunnel_unavailable": - return False - should_retry = error.response.headers.get("x-should-retry") - if should_retry == "false": - return False - if should_retry == "true": - return True - return error.code == "tunnel_service_not_ready" + return error.code == "tunnel_service_not_ready" and error.response.headers.get("x-should-retry") != "false" def send_tunnel_probe( diff --git a/tests/test_transport_error_contract.py b/tests/test_transport_error_contract.py index bc58e86d7..9af6b0f73 100644 --- a/tests/test_transport_error_contract.py +++ b/tests/test_transport_error_contract.py @@ -15,7 +15,7 @@ from runloop_api_client import Runloop, APIError, AsyncRunloop, APIStatusError from runloop_api_client.sdk.devbox import Devbox from runloop_api_client.sdk.async_devbox import AsyncDevbox -from runloop_api_client.lib.tunnel_readiness import wait_for_tunnel_service +from runloop_api_client.lib.tunnel_readiness import wait_for_tunnel_service, async_wait_for_tunnel_service def customer_shape(error: APIError) -> dict[str, object]: @@ -400,28 +400,60 @@ def sleep(delay: float) -> None: assert caught.value.__cause__ is caught.value.cause -def test_explicit_should_retry_applies_to_non_terminal_readiness_error() -> None: +def test_retry_headers_do_not_make_terminal_readiness_error_transient() -> None: attempts = 0 + sleep = Mock() def request(_remaining: float) -> httpx.Response: nonlocal attempts attempts += 1 http_request = httpx.Request("GET", "https://8080-key.tunnel.runloop.ai/health") - if attempts == 1: - return httpx.Response( - 503, - request=http_request, - headers={ - "X-Runloop-Error-Code": "tunnel_backend_connect_timeout", - "X-Should-Retry": "true", - "Retry-After": "0", - }, - json={"error": "tunnel_backend_connect_timeout", "retryable": True}, - ) - return httpx.Response(204, request=http_request) + return httpx.Response( + 503, + request=http_request, + headers={ + "X-Runloop-Error-Code": "tunnel_backend_connect_timeout", + "X-Should-Retry": "true", + "Retry-After": "0", + }, + json={"error": "tunnel_backend_connect_timeout", "retryable": True}, + ) + + with pytest.raises(APIStatusError) as caught: + wait_for_tunnel_service(request, port=8080, sleep=sleep) - wait_for_tunnel_service(request, port=8080, sleep=lambda _delay: None) - assert attempts == 2 + assert caught.value.code == "tunnel_backend_connect_timeout" + assert caught.value.attempts == 1 + assert attempts == 1 + sleep.assert_not_called() + + +async def test_async_retry_headers_do_not_make_terminal_readiness_error_transient() -> None: + attempts = 0 + sleep = AsyncMock() + + async def request(_remaining: float) -> httpx.Response: + nonlocal attempts + attempts += 1 + http_request = httpx.Request("GET", "https://8080-key.tunnel.runloop.ai/health") + return httpx.Response( + 503, + request=http_request, + headers={ + "X-Runloop-Error-Code": "tunnel_backend_connect_timeout", + "X-Should-Retry": "true", + "Retry-After": "0", + }, + json={"error": "tunnel_backend_connect_timeout", "retryable": True}, + ) + + with pytest.raises(APIStatusError) as caught: + await async_wait_for_tunnel_service(request, port=8080, sleep=sleep) + + assert caught.value.code == "tunnel_backend_connect_timeout" + assert caught.value.attempts == 1 + assert attempts == 1 + sleep.assert_not_awaited() def test_generic_exception_is_not_retried_after_ambiguous_receipt() -> None: