fix(ws): dispatcher correctness — fan-out, server-unsub cleanup, error envelope surface

CHANGELOG.md

@@ -21,6 +21,15 @@ All notable changes to kalshi-sdk will be documented in this file.
   guard (`KalshiError` on a repeating cursor) provides the runaway protection
   it always did. Callers who want a cap pass `max_pages=N` explicitly. No
   user-visible change for anyone iterating <1000 pages (#98).
+- **WS callbacks no longer suppress queue delivery (#80).** Previously,
+  registering a callback for a channel silently disabled the iterator queue
+  for that channel — a user who held both an `@on()` callback AND an
+  `async for msg in subscription` consumer would never see the iterator
+  fire. Now messages fan out to both. A WARNING is logged at
+  `register_callback` time if an active subscription already exists, so
+  users relying on the suppress-side-effect are alerted. Callback-only
+  users now accumulate up to the queue's `maxsize=1000`; existing
+  `DROP_OLDEST` backpressure prevents unbounded growth.
 
 ### Fixed
 

kalshi/ws/dispatch.py

@@ -9,7 +9,7 @@
 from pydantic import BaseModel
 
 from kalshi.ws.channels import SubscriptionManager
-from kalshi.ws.models.base import ErrorMessage
+from kalshi.ws.models.base import ErrorMessage, ErrorPayload
 from kalshi.ws.models.communications import CommunicationsMessage
 from kalshi.ws.models.fill import FillMessage
 from kalshi.ws.models.market_lifecycle import MarketLifecycleMessage
@@ -20,6 +20,7 @@
 from kalshi.ws.models.ticker import TickerMessage
 from kalshi.ws.models.trade import TradeMessage
 from kalshi.ws.models.user_orders import UserOrdersMessage
+from kalshi.ws.sequence import SequenceTracker
 
 logger = logging.getLogger("kalshi.ws")
 
@@ -50,21 +51,122 @@ def __init__(
         self,
         sub_mgr: SubscriptionManager,
         on_error: Callable[[ErrorMessage], Awaitable[None]] | None = None,
+        seq_tracker: SequenceTracker | None = None,
     ) -> None:
         self._sub_mgr = sub_mgr
         self._on_error = on_error
+        self._seq_tracker = seq_tracker
         self._callbacks: dict[str, Callable[[Any], Awaitable[None]]] = {}
 
     def register_callback(
         self, channel: str, callback: Callable[[Any], Awaitable[None]]
     ) -> None:
-        """Register a callback for a specific channel type."""
+        """Register a callback for a specific channel type.
+
+        Messages on this channel are fanned out to BOTH the callback and any
+        active subscription queue for the same channel. If an active
+        subscription queue exists, a warning is logged so users know the
+        callback was registered after the fact.
+        """
+        if any(
+            sub.channel == channel
+            for sub in self._sub_mgr.active_subscriptions.values()
+        ):
+            logger.warning(
+                "register_callback(%r): an active subscription exists on this "
+                "channel; messages will be delivered to BOTH the callback and "
+                "the existing iterator queue.",
+                channel,
+            )
         self._callbacks[channel] = callback
 
     def unregister_callback(self, channel: str) -> None:
         """Remove a callback for a channel type."""
         self._callbacks.pop(channel, None)
 
+    async def _handle_server_unsubscribe(self, data: dict[str, Any]) -> None:
+        """Reap state for a server-initiated unsubscribe envelope (#81).
+
+        Client-initiated unsubscribes consume the ack inside
+        SubscriptionManager._wait_for_response, so any unsubscribed frame
+        that reaches the dispatcher is server-initiated (admin action,
+        session expiry) or a late ack after teardown. In either case the
+        sid mappings, seq tracker state, and any held iterator must be
+        cleaned up so they don't leak across reconnects.
+        """
+        # The envelope can carry sid at the top level or nested under msg.
+        sid = data.get("sid")
+        if sid is None:
+            msg_body = data.get("msg")
+            if isinstance(msg_body, dict):
+                sid = msg_body.get("sid")
+        if sid is None:
+            logger.debug("server unsubscribed envelope without sid: %s", data)
+            return
+
+        client_id = self._sub_mgr._sid_to_client.pop(sid, None)
+        if self._seq_tracker is not None:
+            self._seq_tracker.reset(sid)
+
+        if client_id is None:
+            logger.debug(
+                "server unsubscribed for unknown sid %d (already reaped)", sid
+            )
+            return
+
+        sub = self._sub_mgr._subscriptions.pop(client_id, None)
+        if sub is not None:
+            # Wake any held iterator so async-for exits cleanly.
+            await sub.queue.put_sentinel()
+            logger.info(
+                "Server-initiated unsubscribe: channel=%s sid=%d client_id=%d",
+                sub.channel, sid, client_id,
+            )
+
+    async def _surface_channel_error(
+        self, msg_type: str, data: dict[str, Any]
+    ) -> None:
+        """Route a channel-level error envelope to on_error (#82).
+
+        AsyncAPI allows errors to ride on a typed message (e.g. a payload
+        with an `error` field, or an unrecognized type that still carries
+        error semantics) rather than the top-level type="error" envelope.
+        Previously these fell through to a debug log and were silently
+        dropped; surface them via on_error if registered, or log at
+        WARNING with the envelope contents so the user has a signal.
+        """
+        if self._on_error is not None:
+            # Best-effort: synthesize an ErrorMessage. If the payload
+            # doesn't match the strict schema, fall back to model_construct
+            # so the handler still fires with the raw payload visible.
+            try:
+                synthesized = {
+                    "id": data.get("id", 0),
+                    "type": "error",
+                    "msg": data.get("error")
+                    if isinstance(data.get("error"), dict)
+                    else data.get("msg"),
+                }
+                error = ErrorMessage.model_validate(synthesized)
+            except Exception:
+                logger.error(
+                    "Channel-level error envelope (type=%s) failed strict "
+                    "ErrorMessage validation; firing on_error with raw payload: %s",
+                    msg_type, data,
+                )
+                error = ErrorMessage.model_construct(
+                    id=data.get("id", 0),
+                    type="error",
+                    msg=ErrorPayload.model_construct(code="unknown", msg=str(data)),
+                )
+            await self._on_error(error)
+        else:
+            logger.warning(
+                "Channel-level error envelope (type=%s) with no on_error "
+                "handler registered: %s",
+                msg_type, data,
+            )
+
     async def dispatch(self, raw: str) -> None:
         """Parse a raw JSON frame and route it."""
         try:
@@ -80,6 +182,16 @@ async def dispatch(self, raw: str) -> None:
             if msg_type == "error" and self._on_error is not None:
                 error = ErrorMessage.model_validate(data)
                 await self._on_error(error)
+            elif msg_type == "unsubscribed":
+                await self._handle_server_unsubscribe(data)
+            return
+
+        # Channel-level error semantics on a non-"error" envelope (#82):
+        # `data.get("error") is not None` (NOT `"error" in data`) so a
+        # legitimate `"error": null` field on a typed envelope isn't
+        # falsely routed as an error.
+        if data.get("error") is not None:
+            await self._surface_channel_error(msg_type, data)
             return
 
         # Parse into typed model
@@ -105,9 +217,9 @@ async def dispatch(self, raw: str) -> None:
             logger.debug("Message for unknown sid %d (may be stale)", sid)
             return
 
-        # Check for callback first
+        # Fan out: deliver to the callback (if any) AND the subscription queue.
+        # Previously the queue was skipped when a callback was registered,
+        # which silently stranded any iterator on the same channel (#80).
         if sub.channel in self._callbacks:
             await self._callbacks[sub.channel](parsed)
-        else:
-            # Route to queue
-            await sub.queue.put(parsed)
+        await sub.queue.put(parsed)