feat(ws): reliability bundle (#254, #257, #255, #256)

docs/errors.md

@@ -139,7 +139,13 @@ WebSocket failures are a separate sub-hierarchy under
 - **`KalshiBackpressureError`** — raised from `MessageQueue.put()` when the
   queue is full and the overflow strategy is `ERROR`. The receive loop treats
   this as **fatal**: it broadcasts sentinels to every active iterator and
-  exits. See [WebSocket → Backpressure](websockets.md#backpressure).
+  exits. See [WebSocket → Backpressure](websockets.md#backpressure). Carries:
+    - `channel: str | None` — the channel whose queue overflowed.
+    - `sid: int | None` — server-side subscription id (populated at the
+      ``broadcast_error`` site since the queue itself doesn't track sid).
+    - `client_id: int | None` — durable client-side id.
+    - `maxsize: int | None` — the configured queue ceiling at the time of
+      overflow.
 - **`KalshiSequenceGapError`** — exposed for callers wiring their own resync
   logic on top of the SDK's primitives. The built-in receive loop **does not
   raise** this — it recovers from gaps silently (drops the message, clears
@@ -193,4 +199,6 @@ re-established at all.
 
 ::: kalshi.errors.KalshiBackpressureError
 
+::: kalshi.errors.KalshiOrderbookUnavailableError
+
 ::: kalshi.errors.KalshiSubscriptionError

kalshi/__init__.py

@@ -13,6 +13,7 @@
     KalshiError,
     KalshiNetworkError,
     KalshiNotFoundError,
+    KalshiOrderbookUnavailableError,
     KalshiPoolExhaustedError,
     KalshiRateLimitError,
     KalshiSequenceGapError,
@@ -267,6 +268,7 @@
     "KalshiError",
     "KalshiNetworkError",
     "KalshiNotFoundError",
+    "KalshiOrderbookUnavailableError",
     "KalshiPoolExhaustedError",
     "KalshiRateLimitError",
     "KalshiSequenceGapError",

kalshi/errors.py

@@ -129,7 +129,49 @@ def __init__(
 
 
 class KalshiBackpressureError(KalshiWebSocketError):
-    """Message queue overflow with ERROR strategy."""
+    """Message queue overflow with ERROR strategy.
+
+    Carries structured context so consumers iterating multiple
+    subscriptions can route the failure by channel/sid/client_id
+    without parsing the error message string.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        channel: str | None = None,
+        sid: int | None = None,
+        client_id: int | None = None,
+        maxsize: int | None = None,
+    ) -> None:
+        self.channel = channel
+        self.sid = sid
+        self.client_id = client_id
+        self.maxsize = maxsize
+        super().__init__(message)
+
+
+class KalshiOrderbookUnavailableError(KalshiWebSocketError):
+    """Local orderbook is empty between teardown and resync snapshot.
+
+    Raised by :class:`KalshiWebSocket.orderbook` when the underlying
+    delta stream yields an item but the orderbook manager has no book
+    for the requested ticker. This can happen between a gap-triggered
+    ``remove_by_sid`` and the resync snapshot landing — the iterator
+    previously yielded an empty :class:`Orderbook`, which is
+    indistinguishable from a real market with zero liquidity and could
+    cause a strategy to place orders against an empty book.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        ticker: str | None = None,
+    ) -> None:
+        self.ticker = ticker
+        super().__init__(message)
 
 
 class KalshiSubscriptionError(KalshiWebSocketError):

kalshi/ws/backpressure.py

@@ -54,9 +54,18 @@ def __init__(
         self,
         maxsize: int = 1000,
         overflow: OverflowStrategy = OverflowStrategy.DROP_OLDEST,
+        *,
+        channel: str | None = None,
+        client_id: int | None = None,
     ) -> None:
         self._maxsize = maxsize
         self._overflow = overflow
+        # #256: identity metadata used to populate ``KalshiBackpressureError``
+        # fields at the ``put()`` raise site, so consumers iterating multiple
+        # queues can ``except KalshiBackpressureError as e: route(e.channel,
+        # e.client_id, ...)`` without parsing the message string.
+        self._channel = channel
+        self._client_id = client_id
         # `maxlen=maxsize+1` is a hard memory ceiling enforced by deque itself,
         # independent of `_size`. If the counter ever drifts (a put path that
         # forgets to increment, an exception between append and increment, etc.)
@@ -82,7 +91,10 @@ async def put(self, item: T) -> None:
                 raise KalshiBackpressureError(
                     f"Message queue full ({self._maxsize} items). "
                     "Consumer is too slow. Consider increasing maxsize or "
-                    "switching to DROP_OLDEST overflow strategy."
+                    "switching to DROP_OLDEST overflow strategy.",
+                    channel=self._channel,
+                    client_id=self._client_id,
+                    maxsize=self._maxsize,
                 )
 
         self._buffer.append(item)

kalshi/ws/channels.py

@@ -11,7 +11,11 @@
 
 from websockets.exceptions import ConnectionClosed
 
-from kalshi.errors import KalshiConnectionError, KalshiSubscriptionError
+from kalshi.errors import (
+    KalshiBackpressureError,
+    KalshiConnectionError,
+    KalshiSubscriptionError,
+)
 from kalshi.ws.backpressure import MessageQueue, OverflowStrategy
 from kalshi.ws.connection import ConnectionManager
 
@@ -266,12 +270,19 @@ async def subscribe(
         maxsize: int = 1000,
     ) -> Subscription:
         """Subscribe to a channel. Returns a Subscription with a durable client_id."""
+        client_id = self._next_client_id
+        self._next_client_id += 1
         if queue is None:
-            queue = MessageQueue(maxsize=maxsize, overflow=overflow)
+            # #256: propagate identity to the queue so a backpressure overflow
+            # raises with structured channel/client_id fields populated.
+            queue = MessageQueue(
+                maxsize=maxsize,
+                overflow=overflow,
+                channel=channel,
+                client_id=client_id,
+            )
 
         sub_params = params or {}
-        client_id = self._next_client_id
-        self._next_client_id += 1
         sub = Subscription(
             client_id=client_id, channel=channel, params=sub_params, queue=queue
         )
@@ -544,10 +555,23 @@ async def broadcast_error(
         ``StopAsyncIteration``. Subsequent puts on the queue become
         no-ops (the queue is now closed). Safe to call for unknown
         ``client_id`` — it becomes a no-op.
+
+        #256: for ``KalshiBackpressureError`` the raise site populates
+        ``channel``/``client_id``/``maxsize`` from the queue's
+        construction-time metadata. Sid is not known to the queue
+        (it's assigned at subscribe-ack and may change on resubscribe)
+        so enrich it here from the subscription's current
+        ``server_sid`` when missing.
         """
         sub = self._subscriptions.get(client_id)
         if sub is None:
             return
+        if (
+            isinstance(exc, KalshiBackpressureError)
+            and exc.sid is None
+            and sub.server_sid is not None
+        ):
+            exc.sid = sub.server_sid
         await sub.queue.put_error(exc)
 
     def take_stash(self) -> dict[int, collections.deque[str]]:

kalshi/ws/client.py

@@ -17,6 +17,7 @@
 from kalshi.errors import (
     KalshiBackpressureError,
     KalshiConnectionError,
+    KalshiOrderbookUnavailableError,
     KalshiSequenceGapError,
     KalshiSubscriptionError,
 )
@@ -341,6 +342,28 @@ async def _process_frame(self, raw: str) -> None:
         seq = data.get("seq")
         msg_type = data.get("type", "")
 
+        # #255: stale orderbook frames arriving after a teardown
+        # (resubscribe_one or server-initiated unsubscribe reaped the
+        # sid) must not mutate the local book. The dispatcher already
+        # drops unknown-sid frames at routing time, but the orderbook
+        # apply paths below ran BEFORE that check, so a stale snapshot
+        # could re-seed the manager under the old sid index and a stale
+        # delta could clobber the freshly-resynced book. Gate ALL
+        # downstream processing (seq tracking, validation, apply,
+        # dispatch) on the sid being currently mapped. Frames without a
+        # sid (control envelopes, untracked channels) fall through.
+        if (
+            msg_type in ("orderbook_snapshot", "orderbook_delta")
+            and isinstance(sid, int)
+            and self._sub_mgr is not None
+            and self._sub_mgr.get_subscription_by_sid(sid) is None
+        ):
+            logger.debug(
+                "Dropping stale %s for unmapped sid=%d (post-teardown race)",
+                msg_type, sid,
+            )
+            return
+
         prev_seq: int | None = None
         tracked = False
         if sid is not None and seq is not None and self._seq_tracker:
@@ -453,19 +476,22 @@ async def _handle_reconnect(self) -> None:
                 self._running = False
 
     async def _drain_resubscribe_stash(self) -> None:
-        """Replay frames captured during ``resubscribe_all`` through dispatch.
-
-        #176: ``SubscriptionManager._wait_for_response`` captures non-matching
-        data frames into a per-sid stash while a resubscribe is in flight,
-        because between ``_sid_to_client.clear()`` and the new sid mapping
-        the dispatcher has no route for those frames. After all subscribes
-        complete, this method drains the stash via ``_process_frame`` so
-        seq tracking and orderbook state stay consistent with the natural
-        arrival path.
-
-        Frames whose sid did not get re-mapped (subscription failed during
-        ``resubscribe_all``) are dropped with a debug log — there's no
-        consumer to deliver them to.
+        """Replay frames captured during a resubscribe through dispatch.
+
+        #176/#254: ``SubscriptionManager._wait_for_response`` captures
+        non-matching data frames into a per-sid stash while ``_stashing``
+        is set — which happens around both ``resubscribe_all`` (full
+        reconnect) and ``resubscribe_one`` (single-sub gap recovery).
+        Between ``_sid_to_client.clear()`` (or the per-sid teardown) and
+        the new sid mapping landing, the dispatcher has no route for
+        those frames. After the resubscribe completes, this method
+        drains the stash via ``_process_frame`` so seq tracking and
+        orderbook state stay consistent with the natural arrival path.
+
+        Frames whose sid did not get re-mapped (subscription failed
+        during ``resubscribe_all``, or the targeted sid was replaced
+        without re-mapping in ``resubscribe_one``) are dropped with a
+        debug log — there's no consumer to deliver them to.
         """
         if self._sub_mgr is None:
             return
@@ -557,6 +583,20 @@ async def _handle_seq_gap(self, gap: SequenceGap) -> None:
                     next_seq=gap.received,
                 ),
             )
+            return
+
+        # 3. #254: drain the resubscribe-window stash. ``resubscribe_one``
+        # toggles ``_stashing`` around its unsubscribe+subscribe pair, so
+        # any data frames the server emitted during the sid handoff are
+        # captured in ``_sub_mgr._stash``. Without this drain they sit
+        # there until the next full reconnect, and the post-resubscribe
+        # consumer can see a forward seq gap (delta with seq>1 arriving
+        # before the snapshot it depends on). The shared helper replays
+        # any sid that still has a mapping (covers the new sid and any
+        # OTHER live subscriptions whose frames happened to land during
+        # the resubscribe window) and drops frames for sids that no
+        # longer route to a subscription, with a debug log.
+        await self._drain_resubscribe_stash()
 
     # ------------------------------------------------------------------
     # Internal subscribe helper
@@ -912,8 +952,21 @@ async def __anext__(self) -> Orderbook:
         await self._stream.__anext__()
         book = self._mgr.get(self._ticker)
         if book is None:
-            # This shouldn't happen after snapshot, but handle gracefully
-            return Orderbook(ticker=self._ticker)
+            # #257: between a gap-triggered ``remove_by_sid`` and the
+            # resync snapshot landing in the manager, the underlying
+            # stream can yield a frame while ``get(ticker)`` is still
+            # None. Previously this iterator masked the gap by yielding
+            # a fresh empty :class:`Orderbook` — indistinguishable from
+            # a real market with zero liquidity. A strategy reading
+            # bid/ask off the empty book might place orders against
+            # nothing. Surface the absence as a typed error so callers
+            # can halt (or wait for the next snapshot) explicitly.
+            raise KalshiOrderbookUnavailableError(
+                f"Local orderbook for {self._ticker!r} is unavailable "
+                "(snapshot torn down, resync snapshot not yet applied). "
+                "Halt or wait for the next snapshot before reading bid/ask.",
+                ticker=self._ticker,
+            )
         return book
 
 

tests/ws/test_multi_ticker_gap.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import asyncio
+import collections
 
 import pytest
 from cryptography.hazmat.primitives.asymmetric import rsa
@@ -187,6 +188,9 @@ async def broadcast_error(
             ) -> None:
                 self.broadcast_called = True
 
+            def take_stash(self) -> dict[int, collections.deque[str]]:
+                return {}
+
         stub = _StubMgr()
         ws._sub_mgr = stub  # type: ignore[assignment]
 
@@ -231,6 +235,9 @@ async def broadcast_error(
             ) -> None:
                 self.broadcast_args = (client_id, exc)
 
+            def take_stash(self) -> dict[int, collections.deque[str]]:
+                return {}
+
         stub = _FailingMgr()
         ws._sub_mgr = stub  # type: ignore[assignment]