feat(transport): widen retry codes; typed timeouts; bounded errors; base_url validation; close ownership

kalshi/_base_client.py

@@ -19,41 +19,67 @@
 from kalshi.config import KalshiConfig
 from kalshi.errors import (
     KalshiAuthError,
+    KalshiConflictError,
     KalshiError,
     KalshiNotFoundError,
+    KalshiPoolExhaustedError,
     KalshiRateLimitError,
     KalshiServerError,
+    KalshiTimeoutError,
     KalshiValidationError,
 )
 
 logger = logging.getLogger("kalshi")
 
-RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
+RETRYABLE_STATUS_CODES = {408, 425, 429, 500, 502, 503, 504, 520, 521, 522, 523, 524}
 # DELETE excluded: cancel/batch_cancel are not safely idempotent
 RETRYABLE_METHODS = {"GET", "HEAD", "OPTIONS"}
 
+# _map_error body-size guards: refuse to parse and never let one error
+# message dominate a log line / Sentry payload.
+MAX_ERROR_BODY_BYTES = 16 * 1024
+MAX_ERROR_MESSAGE_CHARS = 1024
+
 
 def _map_error(response: httpx.Response) -> KalshiError:
     """Map an HTTP error response to the appropriate SDK exception."""
     status = response.status_code
+
+    # #203: cap body buffering. If the server advertises an oversized body,
+    # short-circuit before httpx materialises it into memory.
+    content_length = response.headers.get("content-length")
+    if content_length:
+        try:
+            if int(content_length) > MAX_ERROR_BODY_BYTES:
+                return KalshiError(
+                    message=f"HTTP {status} (body {content_length} bytes, suppressed)",
+                    status_code=status,
+                )
+        except ValueError:
+            pass  # malformed Content-Length — fall through to normal parse
+
     try:
         body = response.json()
     except Exception:
         body = {}
 
-    message = body.get("message") or body.get("error") or response.text or f"HTTP {status}"
+    raw_message = body.get("message") or body.get("error") or response.text or f"HTTP {status}"
+    # Truncate to bound log-line and exception-message size.
+    message = str(raw_message)[:MAX_ERROR_MESSAGE_CHARS]
 
-    if status == 400:
+    if status in (400, 422):
         details = body.get("details") or body.get("errors")
         return KalshiValidationError(
-            message=str(message),
+            message=message,
             status_code=status,
             details=details if isinstance(details, dict) else None,
         )
     if status in (401, 403):
-        return KalshiAuthError(message=str(message), status_code=status)
+        return KalshiAuthError(message=message, status_code=status)
     if status == 404:
-        return KalshiNotFoundError(message=str(message), status_code=status)
+        return KalshiNotFoundError(message=message, status_code=status)
+    if status == 409:
+        return KalshiConflictError(message=message, status_code=status)
     if status == 429:
         retry_after = response.headers.get("Retry-After")
         retry_after_val: float | None = None
@@ -66,12 +92,12 @@ def _map_error(response: httpx.Response) -> KalshiError:
             except ValueError:
                 retry_after_val = None  # HTTP-date format, fall back to computed backoff
         return KalshiRateLimitError(
-            message=str(message), status_code=status, retry_after=retry_after_val
+            message=message, status_code=status, retry_after=retry_after_val
         )
     if status >= 500:
-        return KalshiServerError(message=str(message), status_code=status)
+        return KalshiServerError(message=message, status_code=status)
 
-    return KalshiError(message=str(message), status_code=status)
+    return KalshiError(message=message, status_code=status)
 
 
 def _compute_backoff(attempt: int, config: KalshiConfig) -> float:
@@ -87,6 +113,13 @@ def _compute_backoff(attempt: int, config: KalshiConfig) -> float:
     return float(random.uniform(0, capped))
 
 
+def _would_exceed_budget(start: float, delay: float, total_timeout: float | None) -> bool:
+    """#193: True iff sleeping ``delay`` now would push past ``total_timeout``."""
+    if total_timeout is None:
+        return False
+    return (time.monotonic() - start) + delay > total_timeout
+
+
 class SyncTransport:
     """Synchronous HTTP transport using httpx.Client."""
 
@@ -129,6 +162,9 @@ def request(
         # Sign with path-only (not full URL). Kalshi expects: /trade-api/v2/endpoint
         sign_path = self._base_path + path
         last_error: KalshiError | None = None
+        # #193: wall-clock budget across the whole request including retries.
+        start = time.monotonic()
+        total_timeout = self._config.total_timeout
 
         for attempt in range(self._config.max_retries + 1):
             auth_headers = self._auth.sign_request(method.upper(), sign_path) if self._auth else {}
@@ -149,14 +185,38 @@ def request(
                     json=json,
                     headers=auth_headers,
                 )
+            except httpx.PoolTimeout as e:
+                # #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"Raise KalshiConfig.limits.max_connections."
+                )
+                if attempt < self._config.max_retries:
+                    delay = _compute_backoff(attempt, self._config)
+                    if _would_exceed_budget(start, delay, total_timeout):
+                        raise last_error from None
+                    logger.warning(
+                        "Pool exhausted on %s %s, retrying in %.1fs",
+                        method,
+                        path,
+                        delay,
+                    )
+                    time.sleep(delay)
+                    continue
+                raise last_error from e
             except httpx.TimeoutException as e:
                 # F-O-09: don't interpolate httpx exception strings — they
                 # 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 = KalshiError(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):
+                        raise last_error from None
                     logger.warning("Timeout on %s %s, retrying in %.1fs", method, path, delay)
                     time.sleep(delay)
                     continue
@@ -192,6 +252,9 @@ def request(
             else:
                 delay = _compute_backoff(attempt, self._config)
 
+            if _would_exceed_budget(start, delay, total_timeout):
+                raise error from None
+
             logger.warning(
                 "%s %s returned %d, retrying in %.1fs (attempt %d/%d)",
                 method.upper(),
@@ -256,6 +319,9 @@ async def request(
         # Sign with path-only (not full URL). Kalshi expects: /trade-api/v2/endpoint
         sign_path = self._base_path + path
         last_error: KalshiError | None = None
+        # #193: wall-clock budget across the whole request including retries.
+        start = time.monotonic()
+        total_timeout = self._config.total_timeout
 
         for attempt in range(self._config.max_retries + 1):
             if self._auth:
@@ -279,11 +345,35 @@ async def request(
                     json=json,
                     headers=auth_headers,
                 )
+            except httpx.PoolTimeout as e:
+                # #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"Raise KalshiConfig.limits.max_connections."
+                )
+                if attempt < self._config.max_retries:
+                    delay = _compute_backoff(attempt, self._config)
+                    if _would_exceed_budget(start, delay, total_timeout):
+                        raise last_error from None
+                    logger.warning(
+                        "Pool exhausted on %s %s, retrying in %.1fs",
+                        method,
+                        path,
+                        delay,
+                    )
+                    await asyncio.sleep(delay)
+                    continue
+                raise last_error from e
             except httpx.TimeoutException as e:
                 # F-O-09: see sync transport above.
-                last_error = KalshiError(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):
+                        raise last_error from None
                     logger.warning("Timeout on %s %s, retrying in %.1fs", method, path, delay)
                     await asyncio.sleep(delay)
                     continue
@@ -319,6 +409,9 @@ async def request(
             else:
                 delay = _compute_backoff(attempt, self._config)
 
+            if _would_exceed_budget(start, delay, total_timeout):
+                raise error from None
+
             logger.warning(
                 "%s %s returned %d, retrying in %.1fs (attempt %d/%d)",
                 method.upper(),

kalshi/async_client.py

@@ -65,6 +65,9 @@ def __init__(
         # Reject empty strings that look like misconfigured credentials
         if key_id is not None and not key_id.strip():
             raise ValueError("key_id must not be empty. Omit it for unauthenticated access.")
+        # #210: only shut down auth on close() when WE built it. If the
+        # caller passed in their own KalshiAuth, they keep ownership.
+        self._auth_owned: bool = auth is None
         self._auth: KalshiAuth | None
         if auth is not None:
             self._auth = auth
@@ -166,12 +169,15 @@ def from_env(cls, **kwargs: object) -> AsyncKalshiClient:
         auth = KalshiAuth.try_from_env()
         demo = os.environ.get("KALSHI_DEMO", "").lower() == "true"
         base_url = os.environ.get("KALSHI_API_BASE_URL")
-        return cls(auth=auth, demo=demo, base_url=base_url, **kwargs)  # type: ignore[arg-type]
+        client = cls(auth=auth, demo=demo, base_url=base_url, **kwargs)  # type: ignore[arg-type]
+        # from_env constructs auth locally — caller never owned it.
+        client._auth_owned = auth is not None
+        return client
 
     async def close(self) -> None:
         """Close the underlying async HTTP connection pool."""
         await self._transport.close()
-        if self._auth is not None:
+        if self._auth is not None and self._auth_owned:
             self._auth.close()
 
     async def __aenter__(self) -> AsyncKalshiClient:

kalshi/client.py

@@ -67,6 +67,9 @@ def __init__(
         # Reject empty strings that look like misconfigured credentials
         if key_id is not None and not key_id.strip():
             raise ValueError("key_id must not be empty. Omit it for unauthenticated access.")
+        # #210: only shut down auth on close() when WE built it. If the
+        # caller passed in their own KalshiAuth, they keep ownership.
+        self._auth_owned: bool = auth is None
         self._auth: KalshiAuth | None
         if auth is not None:
             self._auth = auth
@@ -135,12 +138,15 @@ def from_env(cls, **kwargs: object) -> KalshiClient:
         auth = KalshiAuth.try_from_env()
         demo = os.environ.get("KALSHI_DEMO", "").lower() == "true"
         base_url = os.environ.get("KALSHI_API_BASE_URL")
-        return cls(auth=auth, demo=demo, base_url=base_url, **kwargs)  # type: ignore[arg-type]
+        client = cls(auth=auth, demo=demo, base_url=base_url, **kwargs)  # type: ignore[arg-type]
+        # from_env constructs auth locally — caller never owned it.
+        client._auth_owned = auth is not None
+        return client
 
     def close(self) -> None:
         """Close the underlying HTTP connection pool."""
         self._transport.close()
-        if self._auth is not None:
+        if self._auth is not None and self._auth_owned:
             self._auth.close()
 
     def __enter__(self) -> KalshiClient:

kalshi/config.py

@@ -36,6 +36,10 @@ class KalshiConfig:
         max_retries: Max retry attempts for transient errors. Defaults to 3.
         retry_base_delay: Base delay in seconds for exponential backoff. Defaults to 0.5.
         retry_max_delay: Maximum delay in seconds for backoff. Defaults to 30.
+        total_timeout: Hard cap on cumulative time spent inside a single
+            request including retries, in seconds. ``None`` disables (legacy
+            behaviour: retry until ``max_retries`` exhausted regardless of
+            wall-clock).
         http2: Enable HTTP/2 for REST requests. Off by default for compat.
             Requires the ``h2`` package (install ``httpx[http2]`` or ``h2``).
         limits: Custom ``httpx.Limits`` for connection pool tuning. ``None``
@@ -47,6 +51,7 @@ class KalshiConfig:
     max_retries: int = DEFAULT_MAX_RETRIES
     retry_base_delay: float = 0.5
     retry_max_delay: float = 30.0
+    total_timeout: float | None = None
     extra_headers: dict[str, str] = field(default_factory=dict)
     ws_base_url: str = PRODUCTION_WS_URL  # trailing slash is stripped automatically
     ws_max_retries: int = DEFAULT_WS_MAX_RETRIES
@@ -61,6 +66,14 @@ def __post_init__(self) -> None:
             object.__setattr__(self, "ws_base_url", self.ws_base_url.rstrip("/"))
         KalshiConfig._validate_url(self.base_url, "base_url", secure="https", plaintext="http")
         KalshiConfig._validate_url(self.ws_base_url, "ws_base_url", secure="wss", plaintext="ws")
+        # #202: prevent silently calling /trade-api or /v1 by enforcing the
+        # v2 path component. Trailing slashes are already stripped above.
+        base_path = urlparse(self.base_url).path
+        if base_path != "/trade-api/v2":
+            raise ValueError(
+                f"KalshiConfig.base_url must include the /trade-api/v2 path "
+                f"component, got base_url={self.base_url!r}"
+            )
 
     @staticmethod
     def _validate_url(url: str, field_name: str, *, secure: str, plaintext: str) -> None: