_handle_orphan_subscribed swallows send failures and bypasses unsubscribe correlation — orphan sids can leak #354

Problem

_handle_orphan_subscribed releases a server-side sid whose subscribe ack arrived after the awaiting task was cancelled. It does so by minting a fresh msg_id directly off the subscription manager and firing an unsubscribe cmd, with no correlation: the eventual server unsubscribed envelope falls into the general _handle_server_unsubscribe, which unconditionally pops _sid_to_client[sid], resets the seq tracker, and calls _orderbook_mgr.remove_by_sid(sid).

Two concrete failure modes:

1. The bare except Exception: clause swallows any send failure with only a WARNING log. The comment promises "any send failure is swallowed (the connection is likely already gone, in which case the server reaps the sid on socket close anyway)", but that assumption only holds for ConnectionClosed. A transient send-side error (e.g. JSON dump failure on a custom ws_json_dumps) leaves a live sid orphaned for the lifetime of the connection.
2. The bypass uses _sub_mgr._get_msg_id(), advancing the manager's id counter from outside any of its locks — encapsulation-breaking, and fragile if the manager ever moves to multi-task id allocation.

Evidence

• kalshi/ws/dispatch.py:152-183 — orphan handler

try:
    msg_id = self._sub_mgr._get_msg_id()
    cmd = {
        "id": msg_id,
        "cmd": "unsubscribe",
        "params": {"sids": [sid]},
    }
    await self._sub_mgr._connection.send(cmd)
except Exception:
    logger.warning(
        "Failed to send orphan-unsubscribe for sid=%d",
        sid,
    )

• kalshi/ws/dispatch.py:99-117 — _handle_server_unsubscribe tears down state unconditionally on the matching unsubscribed envelope.

Suggested fix

Narrow the swallow: re-raise on non-ConnectionClosed exceptions so the caller (or its logging) sees the failure, and only suppress on socket-already-closed conditions where the server-side reap is genuinely guaranteed. Long-term, route the orphan unsubscribe through SubscriptionManager._wait_for_response so the ack is correlated and _handle_server_unsubscribe no longer needs to reason about orphan-driven vs server-driven tear-down.

Source

Round-3 independent audit (reviewer: websocket).