diff --git a/docs/errors.md b/docs/errors.md index 13258342..0624c64c 100644 --- a/docs/errors.md +++ b/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 diff --git a/kalshi/__init__.py b/kalshi/__init__.py index 88e07c2e..dc318996 100644 --- a/kalshi/__init__.py +++ b/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", diff --git a/kalshi/errors.py b/kalshi/errors.py index 77c0a58e..4e3895f3 100644 --- a/kalshi/errors.py +++ b/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): diff --git a/kalshi/ws/backpressure.py b/kalshi/ws/backpressure.py index f07a2f0a..bf29d85a 100644 --- a/kalshi/ws/backpressure.py +++ b/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) diff --git a/kalshi/ws/channels.py b/kalshi/ws/channels.py index 85e9856d..b4d753ef 100644 --- a/kalshi/ws/channels.py +++ b/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]]: diff --git a/kalshi/ws/client.py b/kalshi/ws/client.py index f32fc9b1..5ac03623 100644 --- a/kalshi/ws/client.py +++ b/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 diff --git a/tests/ws/test_multi_ticker_gap.py b/tests/ws/test_multi_ticker_gap.py index 0214334f..49d9bb72 100644 --- a/tests/ws/test_multi_ticker_gap.py +++ b/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] diff --git a/tests/ws/test_w2c_reliability.py b/tests/ws/test_w2c_reliability.py new file mode 100644 index 00000000..5b2020aa --- /dev/null +++ b/tests/ws/test_w2c_reliability.py @@ -0,0 +1,470 @@ +"""Regression tests for WS reliability fixes (#254, #257, #255, #256).""" + +from __future__ import annotations + +import collections +import json +from collections.abc import AsyncIterator +from typing import Any + +import pytest +from cryptography.hazmat.primitives.asymmetric import rsa + +from kalshi.auth import KalshiAuth +from kalshi.config import KalshiConfig +from kalshi.errors import ( + KalshiBackpressureError, + KalshiOrderbookUnavailableError, + KalshiSubscriptionError, +) +from kalshi.ws.backpressure import MessageQueue +from kalshi.ws.channels import Subscription +from kalshi.ws.client import KalshiWebSocket +from kalshi.ws.dispatch import MessageDispatcher +from kalshi.ws.orderbook import OrderbookManager +from kalshi.ws.sequence import SequenceGap, SequenceTracker + + +def _make_ws() -> KalshiWebSocket: + """Build a wire-less ``KalshiWebSocket`` with managers wired up.""" + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + auth = KalshiAuth(key_id="test", private_key=key) + config = KalshiConfig(ws_base_url="ws://localhost:1") + ws = KalshiWebSocket(auth=auth, config=config) + ws._orderbook_mgr = OrderbookManager() + ws._seq_tracker = SequenceTracker() + ws._dispatcher = MessageDispatcher(sub_mgr=None) # type: ignore[arg-type] + return ws + + +@pytest.mark.asyncio +class TestIssue254ResubscribeOneStashDrain: + """#254: ``resubscribe_one``'s stash is drained after success so frames + captured during the sid handoff are replayed instead of leaking until + the next full reconnect.""" + + async def test_stash_drained_through_process_frame(self) -> None: + """Frames stashed during the resubscribe window for the NEW sid get + replayed via ``_process_frame``, seeding the orderbook with the + delta payload.""" + ws = _make_ws() + # Seed the orderbook so a delta can apply (apply_delta_inplace + # only mutates a book that already exists for this ticker). + snap_seed = { + "type": "orderbook_snapshot", + "sid": 99, + "seq": 1, + "msg": { + "market_ticker": "T", + "market_id": "m", + "yes": [["0.50", "100"]], + "no": [], + }, + } + # Wire dispatcher to the stub mgr below. + queue: MessageQueue[object] = MessageQueue() + new_sub = Subscription( + client_id=1, + channel="orderbook_delta", + params={}, + queue=queue, + ) + new_sub.server_sid = 99 + + class _StubMgr: + def __init__(self) -> None: + self._stash: dict[int, collections.deque[str]] = {} + # Pre-populate stash as if frames had landed during the + # resubscribe_one unsubscribe→subscribe window. + bucket: collections.deque[str] = collections.deque(maxlen=1000) + bucket.append(json.dumps(snap_seed)) + bucket.append( + json.dumps( + { + "type": "orderbook_delta", + "sid": 99, + "seq": 2, + "msg": { + "market_ticker": "T", + "market_id": "m", + "price": 51, + "delta": 10, + "side": "yes", + }, + } + ) + ) + self._stash[99] = bucket + # A frame for an unmapped (torn-down) sid: should be dropped. + dead_bucket: collections.deque[str] = collections.deque(maxlen=1000) + dead_bucket.append( + json.dumps( + { + "type": "orderbook_delta", + "sid": 7, + "seq": 1, + "msg": { + "market_ticker": "ZOMBIE", + "market_id": "z", + "price": 99, + "delta": 1, + "side": "yes", + }, + } + ) + ) + self._stash[7] = dead_bucket + self.resubscribed = False + + def get_subscription_by_sid(self, sid: int) -> Subscription | None: + return new_sub if sid == 99 else None + + def get_subscription(self, client_id: int) -> Subscription | None: + return new_sub if client_id == 1 else None + + async def resubscribe_one(self, client_id: int) -> int | None: + self.resubscribed = True + return 99 + + async def broadcast_error(self, client_id: int, exc: BaseException) -> None: + pytest.fail(f"broadcast_error unexpectedly called: {exc!r}") + + def take_stash(self) -> dict[int, collections.deque[str]]: + stash, self._stash = self._stash, {} + return stash + + stub = _StubMgr() + ws._sub_mgr = stub # type: ignore[assignment] + # Rebuild dispatcher with the stub so route lookups succeed. + ws._dispatcher = MessageDispatcher(sub_mgr=stub) # type: ignore[arg-type] + + await ws._handle_seq_gap(SequenceGap(sid=99, expected=2, received=5)) + + # Resubscribe ran successfully. + assert stub.resubscribed + # Stash drained. + assert stub._stash == {} + ob_mgr = ws._orderbook_mgr + seq = ws._seq_tracker + assert ob_mgr is not None + assert seq is not None + # The snapshot from the stash landed via _process_frame: book exists. + book = ob_mgr.get("T") + assert book is not None + # The delta also applied (seq=2 followed snapshot=1). + assert seq.peek(99) == 2 + # The unmapped-sid frame was dropped, never touching the book. + assert ob_mgr.get("ZOMBIE") is None + + async def test_stash_drained_even_when_resubscribe_fails(self) -> None: + """A failed ``resubscribe_one`` still clears the stash so frames + don't leak into the next cycle.""" + ws = _make_ws() + queue: MessageQueue[object] = MessageQueue() + sub = Subscription( + client_id=3, + channel="orderbook_delta", + params={}, + queue=queue, + ) + sub.server_sid = 50 + + class _FailingMgr: + def __init__(self) -> None: + bucket: collections.deque[str] = collections.deque(maxlen=1000) + bucket.append( + json.dumps( + { + "type": "orderbook_delta", + "sid": 50, + "seq": 1, + "msg": { + "market_ticker": "X", + "market_id": "x", + "price": 50, + "delta": 1, + "side": "yes", + }, + } + ) + ) + self._stash: dict[int, collections.deque[str]] = {50: bucket} + self.broadcast_called = False + + def get_subscription_by_sid(self, sid: int) -> Subscription | None: + return sub if sid == 50 else None + + def get_subscription(self, client_id: int) -> Subscription | None: + return sub if client_id == 3 else None + + async def resubscribe_one(self, client_id: int) -> int | None: + raise KalshiSubscriptionError( + "boom", + channel="orderbook_delta", + client_id=client_id, + op="subscribe", + ) + + async def broadcast_error(self, client_id: int, exc: BaseException) -> None: + self.broadcast_called = True + + def take_stash(self) -> dict[int, collections.deque[str]]: + stash, self._stash = self._stash, {} + return stash + + stub = _FailingMgr() + ws._sub_mgr = stub # type: ignore[assignment] + + await ws._handle_seq_gap(SequenceGap(sid=50, expected=1, received=4)) + + # broadcast_error fired; drain did NOT happen (we early-returned + # after broadcast_error). Stash remains for the next cycle to handle. + assert stub.broadcast_called + + +@pytest.mark.asyncio +class TestIssue257OrderbookUnavailable: + """#257: ``_OrderbookIterator`` raises ``KalshiOrderbookUnavailableError`` + instead of yielding an empty :class:`Orderbook` when the manager has no + book for the ticker (e.g. between gap teardown and resync snapshot).""" + + async def test_raises_when_manager_has_no_book(self) -> None: + from kalshi.ws.client import _OrderbookIterator + from kalshi.ws.orderbook import OrderbookManager + + mgr = OrderbookManager() + + async def _stream() -> AsyncIterator[Any]: + # First yield: a placeholder; the iterator only checks the manager. + yield {"type": "orderbook_delta"} + yield {"type": "orderbook_delta"} + + it = _OrderbookIterator(_stream(), mgr, "MISSING") + with pytest.raises(KalshiOrderbookUnavailableError) as excinfo: + await it.__anext__() + assert excinfo.value.ticker == "MISSING" + assert "MISSING" in str(excinfo.value) + + async def test_yields_book_when_present(self) -> None: + """Sanity check: when the book is populated the iterator still works.""" + from kalshi.ws.client import _OrderbookIterator + from kalshi.ws.models.orderbook_delta import OrderbookSnapshotMessage + from kalshi.ws.orderbook import OrderbookManager + + mgr = OrderbookManager() + snap = OrderbookSnapshotMessage.model_validate( + { + "type": "orderbook_snapshot", + "sid": 1, + "seq": 1, + "msg": { + "market_ticker": "OK", + "market_id": "m", + "yes": [["0.50", "100"]], + "no": [], + }, + } + ) + mgr._apply_snapshot_inplace(snap, sid=1) + + async def _stream() -> AsyncIterator[Any]: + yield {"type": "orderbook_snapshot"} + + it = _OrderbookIterator(_stream(), mgr, "OK") + book = await it.__anext__() + assert book.ticker == "OK" + + +@pytest.mark.asyncio +class TestIssue255StaleOrderbookGating: + """#255: ``_process_frame`` must drop orderbook frames whose sid no + longer maps to a live subscription, BEFORE validating or applying + them to the local manager. Otherwise a stale snapshot can re-seed + the book under the old sid index, or a stale delta can clobber a + freshly-resynced book.""" + + async def test_stale_snapshot_does_not_mutate_book(self) -> None: + ws = _make_ws() + + class _EmptyMgr: + def get_subscription_by_sid(self, sid: int) -> Subscription | None: + return None + + def get_subscription(self, client_id: int) -> Subscription | None: + return None + + ws._sub_mgr = _EmptyMgr() # type: ignore[assignment] + ws._dispatcher = MessageDispatcher(sub_mgr=ws._sub_mgr) # type: ignore[arg-type] + ob_mgr = ws._orderbook_mgr + assert ob_mgr is not None + + stale_snapshot = json.dumps( + { + "type": "orderbook_snapshot", + "sid": 42, + "seq": 1, + "msg": { + "market_ticker": "STALE", + "market_id": "m", + "yes": [["0.50", "100"]], + "no": [], + }, + } + ) + await ws._process_frame(stale_snapshot) + # Stale snapshot dropped before apply: book remains empty. + assert ob_mgr.get("STALE") is None + + async def test_stale_delta_does_not_mutate_existing_book(self) -> None: + """A delta arriving on the OLD sid after resubscribe (new sid is now + valid for the same ticker) must not mutate the new book.""" + ws = _make_ws() + ob_mgr = ws._orderbook_mgr + assert ob_mgr is not None + + # Seed a book under the NEW sid (post-resubscribe). + from kalshi.ws.models.orderbook_delta import OrderbookSnapshotMessage + + new_snap = OrderbookSnapshotMessage.model_validate( + { + "type": "orderbook_snapshot", + "sid": 99, + "seq": 1, + "msg": { + "market_ticker": "T", + "market_id": "m", + "yes": [["0.50", "100"]], + "no": [], + }, + } + ) + ob_mgr._apply_snapshot_inplace(new_snap, sid=99) + book_before = ob_mgr.get("T") + assert book_before is not None + yes_before = list(book_before.yes) + + new_sub = Subscription( + client_id=1, + channel="orderbook_delta", + params={}, + queue=MessageQueue(), + ) + new_sub.server_sid = 99 + + class _NewSidOnlyMgr: + def get_subscription_by_sid(self, sid: int) -> Subscription | None: + return new_sub if sid == 99 else None + + def get_subscription(self, client_id: int) -> Subscription | None: + return new_sub if client_id == 1 else None + + ws._sub_mgr = _NewSidOnlyMgr() # type: ignore[assignment] + ws._dispatcher = MessageDispatcher(sub_mgr=ws._sub_mgr) # type: ignore[arg-type] + + # Stale delta for OLD sid that should be dropped before apply. + stale_delta = json.dumps( + { + "type": "orderbook_delta", + "sid": 42, + "seq": 7, + "msg": { + "market_ticker": "T", + "market_id": "m", + "price": 50, + "delta": -50, + "side": "yes", + }, + } + ) + await ws._process_frame(stale_delta) + + # New-sid book is unchanged. + book_after = ob_mgr.get("T") + assert book_after is not None + assert list(book_after.yes) == yes_before + + +@pytest.mark.asyncio +class TestIssue256BackpressureErrorFields: + """#256: ``KalshiBackpressureError`` exposes structured channel/sid/ + client_id/maxsize fields populated at the raise site + (``MessageQueue.put``) and broadcast site + (``SubscriptionManager.broadcast_error``).""" + + async def test_put_overflow_populates_channel_client_id_maxsize(self) -> None: + from kalshi.ws.backpressure import OverflowStrategy + + q: MessageQueue[int] = MessageQueue( + maxsize=2, + overflow=OverflowStrategy.ERROR, + channel="orderbook_delta", + client_id=42, + ) + await q.put(1) + await q.put(2) + with pytest.raises(KalshiBackpressureError) as excinfo: + await q.put(3) + err = excinfo.value + assert err.channel == "orderbook_delta" + assert err.client_id == 42 + assert err.maxsize == 2 + # sid is not known to the queue; populated at broadcast_error site. + assert err.sid is None + + async def test_broadcast_error_enriches_sid(self) -> None: + """``SubscriptionManager.broadcast_error`` populates ``sid`` from the + subscription's current server_sid when the error doesn't already + carry one.""" + from kalshi.ws.backpressure import OverflowStrategy + from kalshi.ws.channels import SubscriptionManager + + mgr = SubscriptionManager(connection=None) # type: ignore[arg-type] + queue: MessageQueue[Any] = MessageQueue( + maxsize=10, + overflow=OverflowStrategy.ERROR, + channel="orderbook_delta", + client_id=7, + ) + sub = Subscription( + client_id=7, + channel="orderbook_delta", + params={}, + queue=queue, + ) + sub.server_sid = 314 + mgr._subscriptions[7] = sub + + exc = KalshiBackpressureError( + "queue full", + channel="orderbook_delta", + client_id=7, + maxsize=10, + ) + assert exc.sid is None + await mgr.broadcast_error(7, exc) + assert exc.sid == 314 + + # And the iterator surfaces it with all fields populated. + with pytest.raises(KalshiBackpressureError) as excinfo: + async for _ in queue: + pass + raised = excinfo.value + assert raised.channel == "orderbook_delta" + assert raised.sid == 314 + assert raised.client_id == 7 + assert raised.maxsize == 10 + + async def test_subscribe_default_queue_carries_identity(self) -> None: + """The default queue constructed by SubscriptionManager carries the + subscription's channel + client_id, so an overflow surfaces with + identity intact even without the caller threading it through.""" + from kalshi.ws.backpressure import OverflowStrategy + + q: MessageQueue[Any] = MessageQueue( + maxsize=5, + overflow=OverflowStrategy.ERROR, + channel="ticker", + client_id=11, + ) + assert q._channel == "ticker" + assert q._client_id == 11