diff --git a/kalshi/_base_client.py b/kalshi/_base_client.py index 1afc91d..62ddb36 100644 --- a/kalshi/_base_client.py +++ b/kalshi/_base_client.py @@ -276,6 +276,9 @@ def request( # is preserved. if path != "/" and path.endswith("/"): path = path.rstrip("/") or "/" + # #342: normalize verb once. Used by auth-sign, the httpx call, log + # statements, and method-class checks across every retry attempt. + method_upper = method.upper() # Sign with path-only (not full URL). Kalshi expects: /trade-api/v2/endpoint sign_path = self._base_path + path last_error: KalshiError | None = None @@ -297,7 +300,7 @@ def request( ) for attempt in range(self._config.max_retries + 1): - auth_headers = self._auth.sign_request(method.upper(), sign_path) if self._auth else {} + auth_headers = self._auth.sign_request(method_upper, sign_path) if self._auth else {} # auth_headers keys are case-canonical (always uppercase # KALSHI-ACCESS-*) and base_headers is already case-deduped by # _ci_merge above, so a plain dict-unpack is safe here (#298). @@ -305,7 +308,7 @@ def request( logger.debug( "Request: %s %s (attempt %d/%d)", - method.upper(), + method_upper, path, attempt + 1, self._config.max_retries + 1, @@ -313,7 +316,7 @@ def request( try: response = self._client.request( - method=method.upper(), + method=method_upper, url=path, params=params, json=json, @@ -324,7 +327,7 @@ def request( # #204: pool exhaustion never reached the wire — safe to retry # even for POST/DELETE. last_error = KalshiPoolExhaustedError( - f"Connection pool exhausted on {method.upper()} {path}. " + f"Connection pool exhausted on {method_upper} {path}. " f"Raise KalshiConfig.limits.max_connections." ) if attempt < self._config.max_retries: @@ -345,8 +348,8 @@ 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}") - if method.upper() in RETRYABLE_METHODS and attempt < self._config.max_retries: + 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): raise last_error from None @@ -362,9 +365,9 @@ def request( # 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}") + last_error = KalshiNetworkError(f"Network error: {method_upper} {path}") pre_wire = isinstance(e, httpx.ConnectError) - idempotent = method.upper() in RETRYABLE_METHODS + 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): @@ -379,9 +382,9 @@ def request( continue raise last_error from e except httpx.HTTPError as e: - raise KalshiError(f"HTTP error: {method.upper()} {path}") from e + raise KalshiError(f"HTTP error: {method_upper} {path}") from e - logger.debug("Response: %s %s → %d", method.upper(), path, response.status_code) + logger.debug("Response: %s %s → %d", method_upper, path, response.status_code) if response.is_success: return response @@ -392,7 +395,7 @@ def request( # Only retry safe methods on transient errors should_retry = ( response.status_code in RETRYABLE_STATUS_CODES - and method.upper() in RETRYABLE_METHODS + and method_upper in RETRYABLE_METHODS and attempt < self._config.max_retries ) @@ -422,7 +425,7 @@ def request( logger.warning( "%s %s returned %d, retrying in %.1fs (attempt %d/%d)", - method.upper(), + method_upper, path, response.status_code, delay, @@ -506,6 +509,8 @@ async def request( # P1.6: canonicalize trailing slash BEFORE both signing and httpx call. if path != "/" and path.endswith("/"): path = path.rstrip("/") or "/" + # #342: normalize verb once; see SyncTransport.request. + method_upper = method.upper() # Sign with path-only (not full URL). Kalshi expects: /trade-api/v2/endpoint sign_path = self._base_path + path last_error: KalshiError | None = None @@ -527,7 +532,7 @@ async def request( for attempt in range(self._config.max_retries + 1): if self._auth: - auth_headers = await self._auth.sign_request_async(method.upper(), sign_path) + auth_headers = await self._auth.sign_request_async(method_upper, sign_path) else: auth_headers = {} # auth_headers keys are case-canonical (always uppercase @@ -537,7 +542,7 @@ async def request( logger.debug( "Async request: %s %s (attempt %d/%d)", - method.upper(), + method_upper, path, attempt + 1, self._config.max_retries + 1, @@ -545,7 +550,7 @@ async def request( try: response = await self._client.request( - method=method.upper(), + method=method_upper, url=path, params=params, json=json, @@ -556,7 +561,7 @@ async def request( # #204: pool exhaustion never reached the wire — safe to retry # even for POST/DELETE. last_error = KalshiPoolExhaustedError( - f"Connection pool exhausted on {method.upper()} {path}. " + f"Connection pool exhausted on {method_upper} {path}. " f"Raise KalshiConfig.limits.max_connections." ) if attempt < self._config.max_retries: @@ -574,8 +579,8 @@ 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}") - if method.upper() in RETRYABLE_METHODS and attempt < self._config.max_retries: + 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): raise last_error from None @@ -585,9 +590,9 @@ async def request( 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}") + last_error = KalshiNetworkError(f"Network error: {method_upper} {path}") pre_wire = isinstance(e, httpx.ConnectError) - idempotent = method.upper() in RETRYABLE_METHODS + 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): @@ -602,9 +607,9 @@ async def request( continue raise last_error from e except httpx.HTTPError as e: - raise KalshiError(f"HTTP error: {method.upper()} {path}") from 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 @@ -614,7 +619,7 @@ async def request( should_retry = ( response.status_code in RETRYABLE_STATUS_CODES - and method.upper() in RETRYABLE_METHODS + and method_upper in RETRYABLE_METHODS and attempt < self._config.max_retries ) @@ -637,7 +642,7 @@ async def request( logger.warning( "%s %s returned %d, retrying in %.1fs (attempt %d/%d)", - method.upper(), + method_upper, path, response.status_code, delay, @@ -653,10 +658,14 @@ async def request( async def close(self) -> None: """Close the underlying httpx async client. Idempotent (#301). - Concurrent close() is safe under asyncio's cooperative scheduling — - no race between the ``_closed`` check and the flip. + #333: ``aclose()`` is awaited BEFORE ``_closed`` flips. If the + surrounding task is cancelled at the ``await`` point (or aclose + raises for any other reason) the flag stays ``False`` so a + follow-up ``close()`` on the same instance actually drains the + pool instead of returning a silent no-op. ``httpx.AsyncClient. + aclose`` is itself idempotent, so the retried call is safe. """ if self._closed: return - self._closed = True await self._client.aclose() + self._closed = True diff --git a/kalshi/resources/_base.py b/kalshi/resources/_base.py index f4a0936..95d1bf5 100644 --- a/kalshi/resources/_base.py +++ b/kalshi/resources/_base.py @@ -13,6 +13,12 @@ from kalshi.errors import AuthRequiredError, KalshiError from kalshi.models.common import Page +# #352: pagination cycle detector. The seen-set is capped so a legitimately +# long paginate (e.g. a fills export) can't blow memory; in the unlikely +# event the cap is exhausted the guard still catches the adjacent-repeat +# (A→A) shape via the legacy last_cursor check. +_CURSOR_SEEN_CAP = 1024 + def _enforce_response_body_cap(response: httpx.Response) -> None: """Raise ``KalshiError`` when a 2xx body exceeds ``MAX_RESPONSE_BODY_BYTES`` (#323).""" @@ -281,11 +287,13 @@ def _delete_with_body( path: str, *, json: dict[str, Any], + params: dict[str, Any] | None = None, extra_headers: dict[str, str] | None = None, ) -> dict[str, Any] | None: response = self._transport.request( "DELETE", path, + params=params, json=json, headers={"Content-Type": "application/json"}, extra_headers=extra_headers, @@ -363,12 +371,16 @@ def _list_all( cursor-repeat guard below provides the safety net against infinite loops. - Raises ``KalshiError`` if the previous page's cursor repeats — - the realistic server-pagination-bug shape (last page replayed - forever). Only the most recent cursor is retained, so memory is - O(1) regardless of page count. + Raises ``KalshiError`` if a cursor the server has already issued + recurs — covers both the trivial "last page replayed forever" + shape and the LB-rotation / cache-drift ``A→B→A→B→…`` cycles + (#352). The seen-set is bounded by ``_CURSOR_SEEN_CAP`` so a + legitimately very-long paginate can't grow unbounded; once the + cap is hit the guard degrades to the adjacent-repeat check. + Memory is therefore O(min(pages, cap)). """ current_params = dict(params) if params else {} + seen_cursors: set[str] = set() last_cursor: str | None = None pages_fetched = 0 while max_pages is None or pages_fetched < max_pages: @@ -384,12 +396,14 @@ def _list_all( pages_fetched += 1 if not page.cursor: break - if page.cursor == last_cursor: + if page.cursor in seen_cursors or page.cursor == last_cursor: raise KalshiError( f"Cursor loop detected on {path}: server returned " f"cursor {page.cursor!r} which was already used. " f"This indicates a server-side pagination bug." ) + if len(seen_cursors) < _CURSOR_SEEN_CAP: + seen_cursors.add(page.cursor) last_cursor = page.cursor current_params["cursor"] = page.cursor @@ -523,11 +537,13 @@ async def _delete_with_body( path: str, *, json: dict[str, Any], + params: dict[str, Any] | None = None, extra_headers: dict[str, str] | None = None, ) -> dict[str, Any] | None: response = await self._transport.request( "DELETE", path, + params=params, json=json, headers={"Content-Type": "application/json"}, extra_headers=extra_headers, @@ -592,6 +608,7 @@ async def _list_all( Raises ``KalshiError`` on repeated cursor; see sync docstring. """ current_params = dict(params) if params else {} + seen_cursors: set[str] = set() last_cursor: str | None = None pages_fetched = 0 while max_pages is None or pages_fetched < max_pages: @@ -608,11 +625,13 @@ async def _list_all( pages_fetched += 1 if not page.cursor: break - if page.cursor == last_cursor: + if page.cursor in seen_cursors or page.cursor == last_cursor: raise KalshiError( f"Cursor loop detected on {path}: server returned " f"cursor {page.cursor!r} which was already used. " f"This indicates a server-side pagination bug." ) + if len(seen_cursors) < _CURSOR_SEEN_CAP: + seen_cursors.add(page.cursor) last_cursor = page.cursor current_params["cursor"] = page.cursor diff --git a/tests/test_async_client.py b/tests/test_async_client.py index 655570e..3690ac5 100644 --- a/tests/test_async_client.py +++ b/tests/test_async_client.py @@ -370,6 +370,65 @@ async def test_close(self, test_auth: KalshiAuth, config: KalshiConfig) -> None: transport = AsyncTransport(test_auth, config) await transport.close() # should not raise + @pytest.mark.asyncio + async def test_issue_333_async_close_cancellation_safe( + self, test_auth: KalshiAuth, config: KalshiConfig + ) -> None: + """#333: cancellation through ``aclose()`` must leave ``_closed=False`` + so a follow-up ``close()`` actually drains the pool. The pre-fix + order set the flag first; a CancelledError between the flip and + the (unfinished) ``aclose`` left the pool orphaned. + """ + import asyncio + + transport = AsyncTransport(test_auth, config) + original_aclose = transport._client.aclose + cancelled_once = {"v": False} + + async def cancelling_aclose() -> None: + # Simulate the parent task being cancelled while we're inside + # the await point. After the first invocation we restore the + # real aclose so the second close() can drain the pool. + if not cancelled_once["v"]: + cancelled_once["v"] = True + transport._client.aclose = original_aclose # type: ignore[method-assign] + raise asyncio.CancelledError() + await original_aclose() + + transport._client.aclose = cancelling_aclose # type: ignore[method-assign] + + with pytest.raises(asyncio.CancelledError): + await transport.close() + + # Pre-fix: _closed was already True here, the follow-up close() + # short-circuited, and the httpx pool stayed open until GC. + assert transport._closed is False + await transport.close() + assert transport._closed is True + assert transport._client.is_closed is True + + @pytest.mark.asyncio + async def test_issue_342_method_upper_hoisted( + self, test_auth: KalshiAuth, config: KalshiConfig + ) -> None: + """#342: lowercase verb is normalized once at the top of ``request()`` + and reused. The wire request and auth signature must match the + uppercase form, and the call must complete without raising the + ``KeyError: 'get'``-style failure a partial revert would cause. + """ + with respx.mock(assert_all_called=True) as mock: + route = mock.get("https://test.kalshi.com/trade-api/v2/markets").mock( + return_value=httpx.Response(200, json={"markets": []}) + ) + transport = AsyncTransport(test_auth, config) + try: + resp = await transport.request("get", "/markets") + finally: + await transport.close() + assert resp.status_code == 200 + # Wire request used the normalized verb. + assert route.calls.last.request.method == "GET" + class TestAsyncTransportUnauthenticated: """Tests for AsyncTransport with auth=None (unauthenticated mode).""" diff --git a/tests/test_base_helpers.py b/tests/test_base_helpers.py index f9e1900..1edc821 100644 --- a/tests/test_base_helpers.py +++ b/tests/test_base_helpers.py @@ -183,27 +183,23 @@ def test_repeated_cursor_raises(self, test_auth: KalshiAuth, test_config: Kalshi assert route.call_count == 2 @respx.mock - def test_list_all_cursor_loop_detection_uses_last_cursor_only_o1( + def test_issue_352_pagination_cycle_detector_non_adjacent( self, test_auth: KalshiAuth, test_config: KalshiConfig ) -> None: - """P4.1: cursor guard is O(1) — only the previous page's cursor is - compared. A non-adjacent revisit (A → B → A) must NOT trip the guard; - the realistic server-pagination-bug shape is the *immediate* replay - captured by ``test_repeated_cursor_raises``. The old set-based - guard would have raised here and consumed unbounded memory on a - well-behaved server that legitimately reused cursor tokens across - non-adjacent pages.""" + """#352: a non-adjacent A → B → A cycle (LB drift / cache rotation + between two pod-local cursors) must trip the guard. Old O(1) + last-cursor-only check would have looped forever; the set-based + guard catches the revisit on the third request.""" responses = [ httpx.Response(200, json={"items": [{"id": "1"}], "cursor": "A"}), httpx.Response(200, json={"items": [{"id": "2"}], "cursor": "B"}), httpx.Response(200, json={"items": [{"id": "3"}], "cursor": "A"}), - httpx.Response(200, json={"items": [{"id": "4"}], "cursor": ""}), ] respx.get("https://test.kalshi.com/trade-api/v2/things").mock(side_effect=responses) resource = SyncResource(SyncTransport(test_auth, test_config)) - collected = list(resource._list_all("/things", _Item, "items")) - assert [item.id for item in collected] == ["1", "2", "3", "4"] + with pytest.raises(KalshiError, match=r"[Cc]ursor loop.*'A'"): + list(resource._list_all("/things", _Item, "items")) @respx.mock def test_list_all_cursor_loop_detection_catches_adjacent_replay( @@ -255,6 +251,24 @@ async def test_repeated_cursor_raises( assert route.call_count == 2 + @respx.mock + @pytest.mark.asyncio + async def test_issue_352_pagination_cycle_detector_non_adjacent( + self, test_auth: KalshiAuth, test_config: KalshiConfig + ) -> None: + """Async mirror of the sync non-adjacent cycle regression (#352).""" + responses = [ + httpx.Response(200, json={"items": [{"id": "1"}], "cursor": "A"}), + httpx.Response(200, json={"items": [{"id": "2"}], "cursor": "B"}), + httpx.Response(200, json={"items": [{"id": "3"}], "cursor": "A"}), + ] + respx.get("https://test.kalshi.com/trade-api/v2/things").mock(side_effect=responses) + resource = AsyncResource(AsyncTransport(test_auth, test_config)) + + with pytest.raises(KalshiError, match=r"[Cc]ursor loop.*'A'"): + async for _ in resource._list_all("/things", _Item, "items"): + pass + def _fresh_cursor_side_effect() -> Any: """Return a respx side_effect that yields a fresh unique cursor per call. @@ -582,3 +596,46 @@ async def test_issue_323_success_body_capped_async( resource = AsyncResource(AsyncTransport(test_auth, test_config)) with pytest.raises(KalshiError, match=r"max_response_bytes"): await resource._get("/things") + + +class TestDeleteWithBodyAcceptsParams: + """#340: ``_delete_with_body`` accepts a ``params`` kwarg for symmetry + with its bytes-fast-path sibling ``_delete_with_body_json``. Without + this, callers that need a query param on a DELETE-with-body endpoint + have to switch helpers just to plumb one string through.""" + + @respx.mock + def test_issue_340_delete_with_body_accepts_params( + self, test_auth: KalshiAuth, test_config: KalshiConfig + ) -> None: + route = respx.delete("https://test.kalshi.com/trade-api/v2/things").mock( + return_value=httpx.Response(204) + ) + resource = SyncResource(SyncTransport(test_auth, test_config)) + + result = resource._delete_with_body( + "/things", json={"ids": ["a"]}, params={"subaccount": "main"} + ) + + assert result is None + assert route.call_count == 1 + # Query param survived through to the wire. + assert dict(route.calls.last.request.url.params) == {"subaccount": "main"} + + @respx.mock + @pytest.mark.asyncio + async def test_issue_340_delete_with_body_accepts_params_async( + self, test_auth: KalshiAuth, test_config: KalshiConfig + ) -> None: + route = respx.delete("https://test.kalshi.com/trade-api/v2/things").mock( + return_value=httpx.Response(204) + ) + resource = AsyncResource(AsyncTransport(test_auth, test_config)) + + result = await resource._delete_with_body( + "/things", json={"ids": ["a"]}, params={"subaccount": "main"} + ) + + assert result is None + assert route.call_count == 1 + assert dict(route.calls.last.request.url.params) == {"subaccount": "main"}