Failed gap-recovery leaves zombie Subscription that resurrects on next reconnect #315

Problem

When _handle_seq_gap calls resubscribe_one(client_id) and the server rejects the fresh subscribe (rate limit, transient auth blip, any KalshiSubscriptionError), the recovery path calls broadcast_error(...) and returns. broadcast_error only puts an error sentinel on the queue — it does NOT pop the Subscription from _subscriptions, and resubscribe_one has already cleared sub.server_sid. The same lingering-sub shape arises on the KalshiBackpressureError path inside _process_frame.

Three concrete consequences:

1. KalshiWebSocket.unsubscribe(client_id) → SubscriptionManager.unsubscribe short-circuits on if not sub or sub.server_sid is None: return. The entry is permanently stuck — the user cannot clean it up.
2. The next reconnect's resubscribe_all iterates self._subscriptions and re-establishes the dead sub on the server with a brand-new sid. The server starts feeding frames again.
3. The dispatcher routes those frames to sub.queue.put(...), which silently no-ops because the queue is already closed by the error sentinel. The user never sees the data, but server quota / bandwidth is being consumed.

Evidence

• kalshi/ws/client.py:606-624 — gap-recovery failure path returns without cleanup:

try:
    await self._sub_mgr.resubscribe_one(client_id)
except KalshiSubscriptionError as e:
    logger.error("Resubscribe after gap failed ...", exc_info=True)
    await self._sub_mgr.broadcast_error(
        client_id,
        KalshiSequenceGapError(...),
    )
    return

• kalshi/ws/channels.py:548-575 — broadcast_error only puts the error sentinel; does not pop _subscriptions or _sid_to_client:

async def broadcast_error(self, client_id: int, exc: BaseException) -> None:
    sub = self._subscriptions.get(client_id)
    if sub is None:
        return
    ...
    await sub.queue.put_error(exc)

• kalshi/ws/channels.py:323-327 — user-driven unsubscribe short-circuits when server_sid is None:

async def unsubscribe(self, client_id: int) -> None:
    sub = self._subscriptions.get(client_id)
    if not sub or sub.server_sid is None:
        return

• kalshi/ws/client.py:451-467 — the KalshiBackpressureError rollback path inside _process_frame also routes through broadcast_error without popping the subscription.

Suggested fix

Add a _discard(client_id) helper on SubscriptionManager that:

async def _discard(self, client_id: int) -> None:
    sub = self._subscriptions.pop(client_id, None)
    if sub is not None and sub.server_sid is not None:
        self._sid_to_client.pop(sub.server_sid, None)

Have broadcast_error callers in the gap-recovery and backpressure paths invoke _discard after the sentinel is delivered, so the next reconnect's resubscribe_all skips the dead entry. Add a regression test asserting len(mgr._subscriptions) == 0 after gap-recovery failure followed by a reconnect.

Source

Round-3 independent audit (reviewer: websocket).