Skip to content

WS: data frames arriving during reconnect+resubscribe are dropped instead of stashed #176

Description

@TexasCoding

From #106 F-R-13. Severity: medium correctness (silent message loss during reconnect bursts).

Behavior

kalshi/ws/client.py:301-319 (reconnect path) and kalshi/ws/channels.py:207-220 (resubscribe_all) handle reconnect by:

  1. Acquiring _subscribe_lock.
  2. Clearing _sid_to_client (channels.py:220, F-P-03 — prevents stale-sid mis-routing).
  3. Sending resubscribe commands and reading their ok responses.
  4. Rebuilding _sid_to_client with new SIDs.
  5. Releasing the lock.

Between steps 2 and 4, data frames sent by the server immediately after assigning the new SID (before the resubscribe response is fully processed) reach the dispatcher with a SID that hasn't been wired yet. The dispatcher has no route and the frame is logged and dropped.

Rare under normal reconnects, but observable under high-volume channels (trade, fill, ticker) when reconnect happens during a market burst.

Fix

Stash data frames arriving during the resubscribe window into a per-SID staging buffer. When the new _sid_to_client mapping lands, drain the staging buffer through the normal dispatch path. Bound the buffer to avoid memory growth if resubscribe stalls.

Sketch

class Dispatcher:
    _stash: dict[int, deque[ParsedFrame]] = field(default_factory=dict)
    _stashing: bool = False

    def dispatch(self, frame: ParsedFrame) -> None:
        sid = frame.sid
        if self._stashing and sid not in self._sid_to_client:
            self._stash.setdefault(sid, deque(maxlen=1000)).append(frame)
            return
        # ... normal route

ConnectionManager flips _stashing on at reconnect-start, off after resubscribe_all returns. On flip-off, drain _stash[sid] for each newly-mapped SID through dispatch(), then clear _stash.

Acceptance

  • Integration test in tests/ws/test_recv_loop_hardening.py (or new file) that simulates a reconnect while a high-rate stream is active and asserts the consumer iterator sees no message gap.
  • The stash is bounded; if resubscribe takes longer than the stash maxlen permits, excess frames are dropped with a logged WARNING (graceful degradation, not unbounded growth).
  • Coordinate with the existing seq-gap correctness (fix(ws): roll back seq on backpressure + reset all tickers on multi-ticker gap #139) — a stashed-then-replayed frame must still flow through the seq tracker correctly.

Metadata

Metadata

Assignees

No one assigned

    Labels

    polishCode-quality and DX improvements; non-functionalwsWebSocket-related

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions