Skip to content

polish(transport): async close cancellation-safe + _delete_with_body params + cycle detector + method.upper hoist #372

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

Merged
TexasCoding merged 1 commit into from
May 22, 2026
Merged

polish(transport): async close cancellation-safe + _delete_with_body params + cycle detector + method.upper hoist #372

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
63 changes: 36 additions & 27 deletions kalshi/_base_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -297,23 +300,23 @@ 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).
merged_headers = {**base_headers, **auth_headers}

logger.debug(
"Request: %s %s (attempt %d/%d)",
method.upper(),
method_upper,
path,
attempt + 1,
self._config.max_retries + 1,
)

try:
response = self._client.request(
method=method.upper(),
method=method_upper,
url=path,
params=params,
json=json,
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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):
Expand All @@ -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
Expand All @@ -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
)

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -537,15 +542,15 @@ async def request(

logger.debug(
"Async request: %s %s (attempt %d/%d)",
method.upper(),
method_upper,
path,
attempt + 1,
self._config.max_retries + 1,
)

try:
response = await self._client.request(
method=method.upper(),
method=method_upper,
url=path,
params=params,
json=json,
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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):
Expand All @@ -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
Expand All @@ -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
)

Expand All @@ -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,
Expand All @@ -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
31 changes: 25 additions & 6 deletions kalshi/resources/_base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)."""
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand All @@ -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

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand All @@ -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
59 changes: 59 additions & 0 deletions tests/test_async_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)."""
Expand Down
Loading
Loading