diff --git a/src/roe/utils/transport.py b/src/roe/utils/transport.py index d18fce6..e7f3f12 100644 --- a/src/roe/utils/transport.py +++ b/src/roe/utils/transport.py @@ -3,9 +3,9 @@ Retries failed requests with exponential backoff (capped at ~10 seconds): - Transport errors: ``httpx.TransportError`` (disconnects, timeouts, etc.). -- HTTP statuses: ``5xx``, ``408``, ``429``. -- Applies to **all** HTTP methods, including POST (replay is handled by httpx / - rewindable buffers for JSON-encoded bodies). + POST and PATCH retry only errors raised before the request was sent, so a + timed-out run is never started twice. +- HTTP statuses: ``5xx``, ``408``, ``429``, for all HTTP methods. Multipart agent-run helpers opt out via the ``x-roe-skip-retry`` header so those POSTs are not retried (non-idempotent streamed bodies). @@ -26,6 +26,9 @@ logger = logging.getLogger(__name__) _BYPASS_HEADER = "x-roe-skip-retry" +_NON_IDEMPOTENT_METHODS = {"POST", "PATCH"} +# Raised before any bytes reach the server, so resending cannot duplicate work. +_UNSENT_ERRORS = (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout) def _should_retry_status(status_code: int) -> bool: @@ -47,7 +50,10 @@ def handle_request(self, request: httpx.Request) -> httpx.Response: try: response = super().handle_request(request) except httpx.TransportError as exc: - if attempt >= self.max_retries: + if attempt >= self.max_retries or ( + request.method in _NON_IDEMPOTENT_METHODS + and not isinstance(exc, _UNSENT_ERRORS) + ): raise wait_time = min(2**attempt, 10) logger.warning( diff --git a/tests/unit/test_retry_transport.py b/tests/unit/test_retry_transport.py new file mode 100644 index 0000000..a45b63e --- /dev/null +++ b/tests/unit/test_retry_transport.py @@ -0,0 +1,44 @@ +"""Verify RoeRetryTransport only replays requests that are safe to send twice.""" + +from __future__ import annotations + +import os +import sys + +import httpx +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "src")) + +from roe.utils import transport # noqa: E402 +from roe.utils.transport import RoeRetryTransport # noqa: E402 + + +@pytest.mark.parametrize( + ("method", "error", "expected_attempts"), + [ + # The request may already have reached the server: never resend a POST. + ("POST", httpx.ReadTimeout, 1), + ("PATCH", httpx.RemoteProtocolError, 1), + # The request never left the client, so resending is safe. + ("POST", httpx.ConnectError, 4), + ("POST", httpx.ConnectTimeout, 4), + # Idempotent methods keep retrying every transport error. + ("GET", httpx.ReadTimeout, 4), + ("DELETE", httpx.ReadTimeout, 4), + ], +) +def test_transport_error_retries(monkeypatch, method, error, expected_attempts): + attempts = [] + + def fail(self, request): + attempts.append(request.method) + raise error("boom", request=request) + + monkeypatch.setattr(httpx.HTTPTransport, "handle_request", fail) + monkeypatch.setattr(transport.time, "sleep", lambda _: None) + + request = httpx.Request(method, "https://api.roe-ai.com/v1/workflows/w/runs/") + with pytest.raises(error): + RoeRetryTransport(max_retries=3).handle_request(request) + assert len(attempts) == expected_attempts