diff --git a/src/runloop_api_client/_base_client.py b/src/runloop_api_client/_base_client.py index da359063d..46de1b499 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', @@ -1408,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( @@ -1421,10 +1464,19 @@ 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) from None + raise self._make_status_error_from_response(err.response, attempts=retries_taken + 1) from None break @@ -2076,7 +2128,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 +2139,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 +2154,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', @@ -2118,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( @@ -2131,10 +2195,19 @@ 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) 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..d81208a1d --- /dev/null +++ b/src/runloop_api_client/lib/error_contract.py @@ -0,0 +1,100 @@ +"""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 + +import time +import email.utils +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 + 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): + 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.""" + if not isinstance(error, httpx.HTTPError): + 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 new file mode 100644 index 000000000..4ca04a8f2 --- /dev/null +++ b/src/runloop_api_client/lib/tunnel_readiness.py @@ -0,0 +1,191 @@ +"""Bounded polling of an established tunnel's service endpoint.""" + +from __future__ import annotations + +import json +import time +from typing import Mapping, Callable, Awaitable, cast + +import httpx + +from .._exceptions import APIError, APIStatusError, APITimeoutError, APIConnectionError +from .error_contract import is_safe_transport_retry + + +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,) + error.attempts = attempts + if error.cause is not None: + raise error from error.cause + 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: + return error.code == "tunnel_service_not_ready" and error.response.headers.get("x-should-retry") != "false" + + +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( + 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, +) -> 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: + 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( + 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]], +) -> 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: + 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 bed785a7d..2c052b797 100644 --- a/src/runloop_api_client/sdk/async_devbox.py +++ b/src/runloop_api_client/sdk/async_devbox.py @@ -2,11 +2,14 @@ from __future__ import annotations +import time import asyncio import logging from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence, Awaitable, cast from typing_extensions import Unpack, override +import httpx + from ..types import ( DevboxView, TunnelView, @@ -37,6 +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_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 @@ -830,6 +839,43 @@ async def enable_tunnel( **params, ) + async def wait_for_tunnel_ready( + self, + port: int, + 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 and poll the requested service until it is ready.""" + client = self._devbox._client.with_options(max_retries=0) + 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() + try: + await async_wait_for_tunnel_service( + lambda remaining: async_send_tunnel_probe(probe_client, url, headers, remaining), + 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, **params: Unpack[SDKDevboxRemoveTunnelParams], diff --git a/src/runloop_api_client/sdk/devbox.py b/src/runloop_api_client/sdk/devbox.py index c97ea2682..8cff2b345 100644 --- a/src/runloop_api_client/sdk/devbox.py +++ b/src/runloop_api_client/sdk/devbox.py @@ -2,11 +2,14 @@ from __future__ import annotations +import time import logging import threading from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence from typing_extensions import Unpack, override +import httpx + from ..types import ( DevboxView, TunnelView, @@ -39,6 +42,7 @@ from ..lib.polling import PollingConfig from ..types.devboxes import ExecutionUpdateChunk from .execution_result import ExecutionResult +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 @@ -833,6 +837,47 @@ def enable_tunnel( **params, ) + def wait_for_tunnel_ready( + self, + port: int, + 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 and poll the requested service until it is ready. + + 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) + 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() + try: + wait_for_tunnel_service( + lambda remaining: send_tunnel_probe(probe_client, url, headers, remaining), + 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, **params: Unpack[SDKDevboxRemoveTunnelParams], 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 new file mode 100644 index 000000000..9af6b0f73 --- /dev/null +++ b/tests/test_transport_error_contract.py @@ -0,0 +1,535 @@ +"""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 +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]: + """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, + } + + +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) + + 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 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.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( + headers={"Authorization": "Bearer api-secret"}, + transport=httpx.MockTransport(handler), + ) + + with probe_client: + result = Devbox(generated, "dbx").net.wait_for_tunnel_ready( + 8080, + "/health", + timeout_seconds=3, + http_client=probe_client, + clock=lambda: clock_value[0], + sleep=sleep, + ) + + generated.with_options.assert_called_once_with(max_retries=0) + 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(return_value=SimpleNamespace(tunnel_key="key", auth_mode="open", auth_token=None)) + ) + sleep = AsyncMock() + + 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) + + +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_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") + 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) + + 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: + 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"