From fc9d6f693a5f9a2aa586a783353315140d75b7d0 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Thu, 21 May 2026 06:21:16 -0500 Subject: [PATCH] fix(transport): retry idempotent verbs on httpx transport errors; add KalshiNetworkError Closes #240 --- docs/errors.md | 13 +++++- kalshi/__init__.py | 2 + kalshi/_base_client.py | 65 +++++++++++++++++++++-------- kalshi/errors.py | 10 +++++ tests/test_async_client.py | 84 ++++++++++++++++++++++++++++++++++++++ tests/test_client.py | 72 ++++++++++++++++++++++++++++++++ 6 files changed, 227 insertions(+), 19 deletions(-) diff --git a/docs/errors.md b/docs/errors.md index a1a851c5..13258342 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -28,6 +28,7 @@ KalshiError # base, .status_code: int | None ├── KalshiConflictError # 409 (e.g., duplicate client_order_id) ├── KalshiTimeoutError # request timed out; commit-status unknown on POST ├── KalshiPoolExhaustedError # local pool full; request never sent +├── KalshiNetworkError # TCP/TLS/DNS/protocol fault after retries ├── KalshiServerError # 5xx └── KalshiWebSocketError # base for WS errors ├── KalshiConnectionError # handshake / reconnect failure @@ -93,8 +94,14 @@ Non-HTTP failures are wrapped to a typed exception with the original as never reached the wire, so it's safe to retry regardless of HTTP method. Persistent pool exhaustion means you should raise `KalshiConfig.limits.max_connections`. -- **Connection failures** (DNS, TLS, RST) raise `KalshiError` immediately on - any verb, no retry. +- **Network failures** (DNS, TLS, TCP RST, HTTP/2 RST_STREAM, half-close) + raise `KalshiNetworkError`. On idempotent verbs (`GET`, `HEAD`, `OPTIONS`) + the transport retries first; on `POST` / `DELETE` / `PUT` only + `httpx.ConnectError` is retried (request never reached the wire — mirrors + `KalshiPoolExhaustedError`). All other transport faults on non-idempotent + verbs surface immediately so the caller can reconcile a possibly-committed + request via `client_order_id`. The original `httpx` exception is preserved + via `__cause__`. ## Catching everything from the SDK @@ -174,6 +181,8 @@ re-established at all. ::: kalshi.errors.KalshiPoolExhaustedError +::: kalshi.errors.KalshiNetworkError + ::: kalshi.errors.KalshiServerError ::: kalshi.errors.KalshiWebSocketError diff --git a/kalshi/__init__.py b/kalshi/__init__.py index 58ab4ae9..88e07c2e 100644 --- a/kalshi/__init__.py +++ b/kalshi/__init__.py @@ -11,6 +11,7 @@ KalshiConflictError, KalshiConnectionError, KalshiError, + KalshiNetworkError, KalshiNotFoundError, KalshiPoolExhaustedError, KalshiRateLimitError, @@ -264,6 +265,7 @@ "KalshiConflictError", "KalshiConnectionError", "KalshiError", + "KalshiNetworkError", "KalshiNotFoundError", "KalshiPoolExhaustedError", "KalshiRateLimitError", diff --git a/kalshi/_base_client.py b/kalshi/_base_client.py index b3b0101c..38b254a9 100644 --- a/kalshi/_base_client.py +++ b/kalshi/_base_client.py @@ -23,6 +23,7 @@ KalshiAuthError, KalshiConflictError, KalshiError, + KalshiNetworkError, KalshiNotFoundError, KalshiPoolExhaustedError, KalshiRateLimitError, @@ -259,9 +260,7 @@ def request( # include the full URL with query string and can leak # token-like values into uncaught-exception sinks. The # underlying exception is preserved via `__cause__`. - last_error = KalshiTimeoutError( - f"Request timed out: {method.upper()} {path}" - ) + last_error = KalshiTimeoutError(f"Request timed out: {method.upper()} {path}") if method.upper() in RETRYABLE_METHODS and attempt < self._config.max_retries: delay = _compute_backoff(attempt, self._config) if _would_exceed_budget(start, delay, total_timeout): @@ -270,6 +269,30 @@ def request( time.sleep(delay) continue raise last_error from e + except httpx.TransportError as e: + # #240: TCP/TLS/DNS/protocol faults. Idempotent verbs retry on + # any transport error; non-idempotent verbs only retry on + # ConnectError (mirrors PoolTimeout — request never reached + # the wire, so no partial-commit risk). ReadError/WriteError/ + # RemoteProtocolError on POST/DELETE may indicate the server + # accepted bytes already; surface them so the caller can + # reconcile via client_order_id. + last_error = KalshiNetworkError(f"Network error: {method.upper()} {path}") + pre_wire = isinstance(e, httpx.ConnectError) + idempotent = method.upper() in RETRYABLE_METHODS + if (idempotent or pre_wire) and attempt < self._config.max_retries: + delay = _compute_backoff(attempt, self._config) + if _would_exceed_budget(start, delay, total_timeout): + raise last_error from e + logger.warning( + "Network error on %s %s, retrying in %.1fs", + method, + path, + delay, + ) + time.sleep(delay) + continue + raise last_error from e except httpx.HTTPError as e: raise KalshiError(f"HTTP error: {method.upper()} {path}") from e @@ -293,10 +316,7 @@ def request( # Use Retry-After header if available for 429. # `is not None` — not truthy — so Retry-After: 0 ("retry immediately") is honored. - if ( - isinstance(error, KalshiRateLimitError) - and error.retry_after is not None - ): + if isinstance(error, KalshiRateLimitError) and error.retry_after is not None: delay = min(error.retry_after, self._config.retry_max_delay) else: delay = _compute_backoff(attempt, self._config) @@ -438,9 +458,7 @@ async def request( raise last_error from e except httpx.TimeoutException as e: # F-O-09: see sync transport above. - last_error = KalshiTimeoutError( - f"Request timed out: {method.upper()} {path}" - ) + last_error = KalshiTimeoutError(f"Request timed out: {method.upper()} {path}") if method.upper() in RETRYABLE_METHODS and attempt < self._config.max_retries: delay = _compute_backoff(attempt, self._config) if _would_exceed_budget(start, delay, total_timeout): @@ -449,12 +467,28 @@ async def request( await asyncio.sleep(delay) continue raise last_error from e + except httpx.TransportError as e: + # #240: see SyncTransport above for the retry rules. + last_error = KalshiNetworkError(f"Network error: {method.upper()} {path}") + pre_wire = isinstance(e, httpx.ConnectError) + idempotent = method.upper() in RETRYABLE_METHODS + if (idempotent or pre_wire) and attempt < self._config.max_retries: + delay = _compute_backoff(attempt, self._config) + if _would_exceed_budget(start, delay, total_timeout): + raise last_error from e + logger.warning( + "Network error on %s %s, retrying in %.1fs", + method, + path, + delay, + ) + await asyncio.sleep(delay) + continue + raise last_error from e except httpx.HTTPError as e: raise KalshiError(f"HTTP error: {method.upper()} {path}") from e - logger.debug( - "Async response: %s %s → %d", method.upper(), path, response.status_code - ) + logger.debug("Async response: %s %s → %d", method.upper(), path, response.status_code) if response.is_success: return response @@ -472,10 +506,7 @@ async def request( raise error # `is not None` so Retry-After: 0 ("retry immediately") is honored. - if ( - isinstance(error, KalshiRateLimitError) - and error.retry_after is not None - ): + if isinstance(error, KalshiRateLimitError) and error.retry_after is not None: delay = min(error.retry_after, self._config.retry_max_delay) else: delay = _compute_backoff(attempt, self._config) diff --git a/kalshi/errors.py b/kalshi/errors.py index cd4f8786..77c0a58e 100644 --- a/kalshi/errors.py +++ b/kalshi/errors.py @@ -82,6 +82,16 @@ class KalshiPoolExhaustedError(KalshiError): """ +class KalshiNetworkError(KalshiError): + """Transport-level network failure (TCP/TLS/DNS/protocol). + + Raised after the retry loop has exhausted attempts on a transient + ``httpx.TransportError`` — ``ConnectError``, ``ReadError``, + ``WriteError``, ``RemoteProtocolError``, etc. The original httpx + exception is chained via ``__cause__``. + """ + + class KalshiServerError(KalshiError): """Server-side error (5xx).""" diff --git a/tests/test_async_client.py b/tests/test_async_client.py index ec36af62..977a6f51 100644 --- a/tests/test_async_client.py +++ b/tests/test_async_client.py @@ -21,6 +21,7 @@ KalshiAuthError, KalshiConflictError, KalshiError, + KalshiNetworkError, KalshiPoolExhaustedError, KalshiServerError, KalshiTimeoutError, @@ -769,3 +770,86 @@ async def test_from_env_owns_auth( client = AsyncKalshiClient.from_env() assert client._auth_owned is True await client.close() + + +class TestAsyncTransportNetworkRetry: + """#240: async httpx.TransportError retry policy. Mirrors sync tests.""" + + @respx.mock + @pytest.mark.asyncio + async def test_get_retries_on_connect_error_then_succeeds( + self, transport: AsyncTransport, monkeypatch: pytest.MonkeyPatch + ) -> None: + async def fake_sleep(d: float) -> None: + pass + + monkeypatch.setattr("asyncio.sleep", fake_sleep) + route = respx.get("https://test.kalshi.com/trade-api/v2/markets").mock( + side_effect=[ + httpx.ConnectError("connection refused"), + httpx.ConnectError("connection refused"), + httpx.Response(200, json={"markets": []}), + ] + ) + resp = await transport.request("GET", "/markets") + assert resp.status_code == 200 + assert route.call_count == 3 + + @respx.mock + @pytest.mark.asyncio + async def test_get_exhausts_retries_raises_KalshiNetworkError( # noqa: N802 + self, transport: AsyncTransport, monkeypatch: pytest.MonkeyPatch + ) -> None: + async def fake_sleep(d: float) -> None: + pass + + monkeypatch.setattr("asyncio.sleep", fake_sleep) + route = respx.get("https://test.kalshi.com/trade-api/v2/markets").mock( + side_effect=httpx.ConnectError("connection refused") + ) + with pytest.raises(KalshiNetworkError, match="Network error") as exc_info: + await transport.request("GET", "/markets") + assert route.call_count == 3 + assert isinstance(exc_info.value.__cause__, httpx.ConnectError) + + @respx.mock + @pytest.mark.asyncio + async def test_post_retries_on_connect_error( + self, transport: AsyncTransport, monkeypatch: pytest.MonkeyPatch + ) -> None: + async def fake_sleep(d: float) -> None: + pass + + monkeypatch.setattr("asyncio.sleep", fake_sleep) + route = respx.post( + "https://test.kalshi.com/trade-api/v2/portfolio/orders" + ).mock( + side_effect=[ + httpx.ConnectError("connection refused"), + httpx.Response(200, json={"order_id": "abc"}), + ] + ) + resp = await transport.request( + "POST", "/portfolio/orders", json={"ticker": "T"} + ) + assert resp.status_code == 200 + assert route.call_count == 2 + + @respx.mock + @pytest.mark.asyncio + async def test_post_does_not_retry_on_read_error( + self, transport: AsyncTransport, monkeypatch: pytest.MonkeyPatch + ) -> None: + async def fake_sleep(d: float) -> None: + pass + + monkeypatch.setattr("asyncio.sleep", fake_sleep) + route = respx.post( + "https://test.kalshi.com/trade-api/v2/portfolio/orders" + ).mock(side_effect=httpx.ReadError("socket read failed")) + with pytest.raises(KalshiNetworkError, match="Network error") as exc_info: + await transport.request( + "POST", "/portfolio/orders", json={"ticker": "T"} + ) + assert route.call_count == 1 + assert isinstance(exc_info.value.__cause__, httpx.ReadError) diff --git a/tests/test_client.py b/tests/test_client.py index a9704b32..3aea4cc7 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -22,6 +22,7 @@ KalshiAuthError, KalshiConflictError, KalshiError, + KalshiNetworkError, KalshiNotFoundError, KalshiPoolExhaustedError, KalshiRateLimitError, @@ -1307,3 +1308,74 @@ def handler(request: httpx.Request) -> httpx.Response: # Non-auth headers still pass through. assert sent.headers["X-Trace"] == "ok" transport.close() + + +class TestTransportNetworkRetry: + """#240: httpx.TransportError retry policy. + + Idempotent verbs (GET/HEAD/OPTIONS) retry on any transport error; + POST/DELETE/PUT only retry on ConnectError (pre-wire). + """ + + @respx.mock + def test_get_retries_on_connect_error_then_succeeds( + self, transport: SyncTransport, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr("kalshi._base_client.time.sleep", lambda d: None) + route = respx.get("https://test.kalshi.com/trade-api/v2/markets").mock( + side_effect=[ + httpx.ConnectError("connection refused"), + httpx.ConnectError("connection refused"), + httpx.Response(200, json={"markets": []}), + ] + ) + resp = transport.request("GET", "/markets") + assert resp.status_code == 200 + assert route.call_count == 3 + + @respx.mock + def test_get_exhausts_retries_raises_KalshiNetworkError( # noqa: N802 + self, transport: SyncTransport, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr("kalshi._base_client.time.sleep", lambda d: None) + route = respx.get("https://test.kalshi.com/trade-api/v2/markets").mock( + side_effect=httpx.ConnectError("connection refused") + ) + with pytest.raises(KalshiNetworkError, match="Network error") as exc_info: + transport.request("GET", "/markets") + # max_retries=2 → 1 + 2 = 3 attempts. + assert route.call_count == 3 + # Original httpx exception chained via __cause__. + assert isinstance(exc_info.value.__cause__, httpx.ConnectError) + + @respx.mock + def test_post_retries_on_connect_error( + self, transport: SyncTransport, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr("kalshi._base_client.time.sleep", lambda d: None) + route = respx.post( + "https://test.kalshi.com/trade-api/v2/portfolio/orders" + ).mock( + side_effect=[ + httpx.ConnectError("connection refused"), + httpx.Response(200, json={"order_id": "abc"}), + ] + ) + resp = transport.request("POST", "/portfolio/orders", json={"ticker": "T"}) + assert resp.status_code == 200 + assert route.call_count == 2 + + @respx.mock + def test_post_does_not_retry_on_read_error( + self, transport: SyncTransport, monkeypatch: pytest.MonkeyPatch + ) -> None: + # ReadError on POST may indicate the server accepted a partial commit; + # surface immediately so the caller can reconcile via client_order_id. + monkeypatch.setattr("kalshi._base_client.time.sleep", lambda d: None) + route = respx.post( + "https://test.kalshi.com/trade-api/v2/portfolio/orders" + ).mock(side_effect=httpx.ReadError("socket read failed")) + with pytest.raises(KalshiNetworkError, match="Network error") as exc_info: + transport.request("POST", "/portfolio/orders", json={"ticker": "T"}) + assert route.call_count == 1 + assert isinstance(exc_info.value.__cause__, httpx.ReadError)