Skip to content

fix(transport): normalize lifecycle incident failures #830

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tode-rl wants to merge 4 commits into
base: main
Choose a base branch
from
Open

fix(transport): normalize lifecycle incident failures #830

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
44e6b62
fix(transport): normalize lifecycle incident failures
Aug 16, 2026
3443be0
fix(tunnel): poll established service readiness
Aug 16, 2026
43fe1f1
fix(transport): address validated retry findings
Aug 16, 2026
3d59e3d
fix(tunnel): restrict readiness status retries
Aug 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 88 additions & 15 deletions src/runloop_api_client/_base_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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',
Expand All @@ -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(
Expand All @@ -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

Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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',
Expand All @@ -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(
Expand All @@ -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

Expand Down
76 changes: 69 additions & 7 deletions src/runloop_api_client/_exceptions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@

import httpx

from .lib.error_contract import status_error_details, transport_error_details

__all__ = [
"BadRequestError",
"AuthenticationError",
Expand Down Expand Up @@ -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):
Expand All @@ -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):
Expand Down
Loading