fix(ws): stash + replay data frames during resubscribe window (#176)

[TexasCoding]

Closes #176.

The race

SubscriptionManager.resubscribe_all clears _sid_to_client at the start (#77 F-P-03 — prevents stale-sid mis-routing) and rebuilds it as each new sid lands in _wait_for_response. Between those steps, the server can send data frames on the just-assigned sid. The client's _wait_for_response reads them off the wire, sees a non-matching id, logs a debug line, and drops them. Under market-burst reconnects on ticker / trade / fill, the SDK could lose tens of messages per reconnect with no visible signal.

The drop site is _wait_for_response, not the dispatcher — the recv loop is blocked inside _handle_reconnect during resubscribe, so _wait_for_response is the only path reading the socket. The issue body's "reach the dispatcher" framing is a touch off, but the bug is real.

The fix

SubscriptionManager gains a per-sid bounded stash, toggled on for the duration of resubscribe_all:

def __init__(self, connection, *, stash_maxlen: int = 1000) -> None:
    ...
    self._stash: dict[int, collections.deque[str]] = {}
    self._stashing: bool = False
    self._stash_maxlen: int = stash_maxlen

_wait_for_response branches on the flag — outside resubscribe (_stashing=False) the existing debug-drop behavior is unchanged; inside resubscribe (_stashing=True) non-matching frames go to self._stash[sid]. Frames without a sid (control envelopes) are still dropped — there's nowhere to replay them to.

if self._stashing:
    self._maybe_stash(raw, data)
else:
    logger.debug("Discarding non-matching frame during subscribe: type=%s", ...)

After resubscribe_all returns (try/finally guarantees _stashing=False), KalshiWebSocket._handle_reconnect drains the stash through _process_frame so each frame flows through the normal pipeline:

await self._sub_mgr.resubscribe_all()
await self._drain_resubscribe_stash()
await self._connection.mark_streaming()

_drain_resubscribe_stash calls take_stash() (atomic return-and-clear), then for each (sid, raw_frames):

• If the sid did NOT get re-mapped (per-sub resubscribe failure, F-P-01 path): drop the frames with a debug log.
• Otherwise: await self._process_frame(raw) for each, in arrival order.

Going through _process_frame (not directly through _dispatcher.dispatch) keeps seq tracking and orderbook state consistent — the replayed frames update the seq watermark exactly once, so the next live frame doesn't trip a spurious gap.

Bound + degradation

Per-sid deque uses collections.deque(maxlen=self._stash_maxlen). On overflow:

• Deque silently evicts oldest (deque semantics).
• A single WARNING fires per fill event (not per frame) noting the sid + maxlen, so callers notice congestion without log spam.

Memory worst-case: stash_maxlen × len(active_subs) × avg_frame_size. With defaults (1000 per sid), this is ~1 MB per active subscription assuming ~1 KB frames — bounded and proportional to the user's existing subscribe footprint.

Coordination with #139 (seq-gap correctness)

The original issue specifically called out that "a stashed-then-replayed frame must still flow through the seq tracker correctly." The drain routes through _process_frame which calls seq_tracker.track(sid, seq, channel) exactly once per frame. Because seq_tracker.reset_all() is called at the start of _handle_reconnect, the first stashed frame for each sid establishes the baseline watermark. Subsequent live frames track normally against that baseline. No spurious 0 → N gap.

Verified by test_stash_drain_advances_seq_tracker_on_sequenced_channels (orderbook_delta channel; ticker isn't sequenced).

Tests

5 unit tests in tests/ws/test_channels.py::TestResubscribeStash cover the mechanics:

• test_maybe_stash_collects_by_sid — per-sid arrival-order deques
• test_maybe_stash_drops_frame_without_sid — control envelopes have nowhere to replay
• test_maybe_stash_maxlen_evicts_oldest — deque maxlen + single WARNING per fill
• test_take_stash_returns_and_clears — atomic return-and-clear
• test_stashing_flag_off_by_default — default state guards against accidental stash

3 integration tests in tests/ws/test_recv_loop_hardening.py::TestResubscribeStashIntegration cover the dispatcher integration:

• test_stash_drain_replays_through_dispatch — frame reaches iterator queue
• test_stash_drain_advances_seq_tracker_on_sequenced_channels — fix(ws): roll back seq on backpressure + reset all tickers on multi-ticker gap #139 coordination
• test_resubscribe_stash_skips_unmapped_sids — failed-resubscribe sids drop gracefully

The race-engineered end-to-end test (server sends pre-ack data during a reconnect) is intentionally not included: race timing is hardware-dependent and the value is the post-drain dispatch path, which the direct tests cover deterministically. Existing tests in test_recv_loop_hardening.py already exercise the close → reconnect → resubscribe pipeline (TestResubscribeIsolation, TestReconnectHoldsSubscribeLock).

Verification

• uv run pytest tests/ --ignore=tests/integration → 2125 passed, 0 warnings.
• uv run ruff check . → clean.
• uv run mypy kalshi/ → Success: no issues found in 76 source files.

Soft-breaking footprint

None. SubscriptionManager.__init__ gains a keyword-only stash_maxlen=1000 parameter; positional usage (SubscriptionManager(connection)) is unchanged. New behavior only fires during resubscribe_all, which is internal-only.

Backlog after this lands

Two open issues remaining: #179 (ROADMAP cleanup, pure docs) and whatever the next nightly catches. The run_forever() hardening trio (#175, #177) + this stash fix make the WS recv path correctness-complete against the F-P / F-R audit findings.

[claude]

Code Review — PR #187: stash + replay data frames during resubscribe window

Overview

Solid fix for a real correctness bug. The race is accurately diagnosed: between _sid_to_client.clear() and the new sid mapping landing, _wait_for_response was the sole reader of the socket and was silently dropping data frames. The architecture is clean — bounded per-sid deque, try/finally guaranteeing the flag is cleared, atomic take_stash(), drain through _process_frame to keep seq tracking consistent. The seq-gap coordination (#139) is well thought out.

---

Bug: _maybe_stash warning fires on every overflow frame, not once

kalshi/ws/channels.py, lines 163–170

elif len(bucket) == self._stash_maxlen:
    # About to evict oldest. Log once per fill, not per frame.
    logger.warning(...)
bucket.append(raw)

The condition len(bucket) == self._stash_maxlen is True on every append once the deque is full. With maxlen=3 and 5 frames, the warning fires at frames 4 and 5 (and every subsequent frame). The comment says "Log once per fill, not per frame" but the code doesn't enforce that.

The PR description also says "A single WARNING fires per fill event (not per frame)" — this claim isn't met.

The test passes because it uses any(...), not len(warnings) == 1.

Fix: track warned sids with a set cleared in take_stash():

# in __init__
self._stash_warned: set[int] = set()

# in _maybe_stash
elif len(bucket) == self._stash_maxlen and sid not in self._stash_warned:
    self._stash_warned.add(sid)
    logger.warning(...)

# in take_stash
def take_stash(self) -> dict[int, collections.deque[str]]:
    stash, self._stash = self._stash, {}
    self._stash_warned.clear()
    return stash

Under a prolonged resubscribe stall, the current code would spam WARNING for every incoming frame on every high-volume channel — which defeats the "not per frame" intent.

---

Minor observations

asyncio.get_event_loop() is deprecated (channels.py:111,113) — pre-existing, not introduced here, but asyncio.get_running_loop().time() is the correct call inside an async def. Low priority but worth a follow-up issue.

Test comment vs. assertion mismatch (test_channels.py:379): the comment says "WARNING fired (once per fill event, not per frame)" but any(...) only proves it fired at least once. After the warning-dedup fix above, this can be tightened to sum(1 for r in caplog.records if "is full" in r.message) == 1.

import logging inside test body (test_channels.py:370): minor style inconsistency; all other imports in the file are at module level.

---

Everything else looks good

• try/finally around _stashing = True is the right shape; no way to leave the flag set on error.
• take_stash() atomic swap is correct and safe under cooperative async.
• Drain through _process_frame (not direct dispatch) to advance seq watermark exactly once — this is the right call and the integration test verifies it.
• Per-sub failure isolation in resubscribe_all is preserved correctly with the indentation restructure.
• Stash state is harmless if resubscribe_all throws (WebSocket shuts down via _running = False; next reconnect won't happen on the same instance).
• stash_maxlen as a keyword-only __init__ parameter is a clean non-breaking extension.
• 8 new tests cover the mechanics and integration paths well; the deterministic approach (inject stash → drain → assert) is the right tradeoff over a race-dependent end-to-end test.

The warning-overfiring issue is the only one worth fixing before merge.

[TexasCoding]

Real bug. Fixed in 6eafeab.

The overfire — you're right: elif len(bucket) == self._stash_maxlen is True on every append once the deque is full, so the WARNING fires on every frame after the fill. The comment ("Log once per fill, not per frame") and PR description both claimed otherwise; only the test was lying via any(...) instead of len(...) == 1.

Fix — added _stash_warned: set[int] to SubscriptionManager.__init__. The warning condition is now len(bucket) == self._stash_maxlen and sid not in self._stash_warned; the sid gets added to the set on the first warning. take_stash() clears the set so the next resubscribe cycle gets fresh per-sid warnings:

elif len(bucket) == self._stash_maxlen and sid not in self._stash_warned:
    self._stash_warned.add(sid)
    logger.warning("Stash for sid %d is full (%d frames); …", sid, self._stash_maxlen)
bucket.append(raw)

# in take_stash()
stash, self._stash = self._stash, {}
self._stash_warned.clear()
return stash

Tightened test — renamed to test_maybe_stash_maxlen_evicts_oldest_with_one_warning_per_fill. Pushes 10 frames at maxlen=3, asserts exactly 1 WARNING (not any(...) >= 1). Then calls take_stash() and pushes 5 more, asserting the total grows from 1 → 2 (proving the next-cycle reset works). Pre-fix this would have produced 8 warnings in the first batch + 5 in the second = 13 total instead of 2.

Other notes:

• import logging — moved to module level in test_channels.py. Good catch.
• asyncio.get_event_loop() deprecation on lines 111/113 — agreed it's worth fixing but pre-existing and out of scope here. Worth a one-line follow-up issue or batching into the next polish PR. Not touching in this PR to keep the diff focused.

Verification: 2125 unit tests pass (test_maybe_stash_maxlen_evicts_oldest_with_one_warning_per_fill exercises both the per-frame-suppression invariant and the take_stash reset). ruff check . clean, mypy kalshi/ clean.

[claude]

Code Review — PR #187: stash + replay data frames during resubscribe window

Overview

Solid fix for a real correctness bug. The race is accurately diagnosed: between _sid_to_client.clear() and the new sid mapping landing, _wait_for_response was the sole reader of the socket and was silently dropping data frames. The architecture is clean — bounded per-sid deque, try/finally guaranteeing the flag is cleared, atomic take_stash(), drain through _process_frame to keep seq tracking consistent. The seq-gap coordination (#139) is well thought out.

The _stash_warned gate (added to fix a previous review finding) is correctly implemented: the overflow WARNING fires exactly once per (sid, resubscribe cycle) and take_stash() clears the suppression set atomically with the stash itself. The test test_maybe_stash_maxlen_evicts_oldest_with_one_warning_per_fill verifies both cycles correctly.

---

Issue 1 — Pre-existing: asyncio.get_event_loop() is deprecated (non-blocking)

kalshi/ws/channels.py, lines 118 and 120

deadline = asyncio.get_event_loop().time() + timeout
...
remaining = deadline - asyncio.get_event_loop().time()

asyncio.get_event_loop() is deprecated since Python 3.10 and emits a DeprecationWarning in 3.12 when called without a running loop. Inside a coroutine there is always a running loop, so the correct call is asyncio.get_running_loop().time(). The PR's docstring edit touches this function, so this is a good moment to clean it up even though it's pre-existing.

# before
deadline = asyncio.get_event_loop().time() + timeout
remaining = deadline - asyncio.get_event_loop().time()

# after
deadline = asyncio.get_running_loop().time() + timeout
remaining = deadline - asyncio.get_running_loop().time()

---

Issue 2 — Stale stash survives a reconnect failure (low risk, worth hardening)

kalshi/ws/channels.py — resubscribe_all finally block
kalshi/ws/client.py — _handle_reconnect

The flow is:

await self._sub_mgr.resubscribe_all()   # sets _stashing=False in finally
await self._drain_resubscribe_stash()   # calls take_stash() → clears _stash + _stash_warned

resubscribe_all catches all per-subscription exceptions internally, so it should not propagate. But if it does (connection drops mid-resubscribe, recv() raises, programming bug), the except Exception in _handle_reconnect fires, _drain_resubscribe_stash is never called, and _stash plus _stash_warned survive into the next reconnect cycle. On the next cycle the stale frames get drained alongside fresh ones — wrong delivery order — and _stash_warned suppresses overflow warnings for those sids.

Minimal fix: clear _stash and _stash_warned at the top of resubscribe_all (before setting _stashing = True) or in the finally block alongside _stashing = False.

# in resubscribe_all, at the start (or in finally alongside _stashing = False)
self._stash.clear()
self._stash_warned.clear()

take_stash() clears them too, so _drain_resubscribe_stash remains the happy-path reset. The clear() calls in resubscribe_all are purely defensive for the failure path.

---

Test coverage — looks good

• Unit test split (mechanics in test_channels.py, integration in test_recv_loop_hardening.py) is clean.
• test_stash_drain_advances_seq_tracker_on_sequenced_channels directly validates the fix(ws): roll back seq on backpressure + reset all tickers on multi-ticker gap #139 coordination concern — exactly the right test for this scenario.
• Decision not to write a race-engineered end-to-end test is well-reasoned; the dispatch path is fully exercised by the direct injection tests.
• The two-cycle assertion in test_maybe_stash_maxlen_evicts_oldest_with_one_warning_per_fill (checking len(warns) == 1 then len(warns2) == 2) correctly verifies that take_stash() resets the suppression gate.

---

Summary

Design	✅ Sound — bounded stash, atomic take, drain before mark_streaming is the right ordering
_stash_warned fix	✅ Correctly gates overflow warnings to once per (sid, cycle)
Seq-gap coordination	✅ Drain through _process_frame ensures seq watermark is set before live frames arrive
asyncio.get_event_loop()	⚠️ Pre-existing deprecation — easy fix, good time to do it since the function is touched
Stash leak on reconnect failure	⚠️ Low risk but worth one clear() call in the finally block as a defensive measure
Tests	✅ Good coverage of mechanics and integration

The core fix is correct and the PR description is thorough. Address issues 1 and 2 before merge — both are small changes.

[TexasCoding]

Both addressed in 9767e57. You were right that since this PR already touches _wait_for_response for the stashing branch, the deprecated get_event_loop() is in-scope.

1. asyncio.get_event_loop() → asyncio.get_running_loop() — both call sites in _wait_for_response swapped. Pre-existing API call but trivial to clean up while the function is open.

2. Defensive stash clear on resubscribe_all entry — added self._stash.clear() + self._stash_warned.clear() immediately after the _sid_to_client.clear() (which is the existing F-P-03 reset). take_stash() still clears on the happy path; this is the failure-mode guard for the case you described.

Your analysis is right that within the current instance, _handle_reconnect's except setting _running=False makes the next-reconnect-on-same-instance case effectively impossible. The defensive clear is for the invariant rather than an observed bug — if a future code path ever calls resubscribe_all on a manager carrying state from a failed prior cycle, stale frames won't replay alongside fresh ones.

Added a regression test (test_resubscribe_clears_stale_stash_at_start): injects stale state into _stash + _stash_warned, runs a clean resubscribe_all on an empty subscriptions set (the for-loop body is skipped, but the clear still runs), and asserts both are empty afterward. Tests the invariant directly without needing to engineer the failure mode.

Verification: 2126 unit tests pass. mypy kalshi/ clean, ruff check . clean.

[claude]

Code Review — PR #187: WS resubscribe-window frame stashing

Overall: This is a well-reasoned fix for a real race condition. The stash/replay pattern is clean, the memory bound is explicit, and the seq-tracker coordination with #139 is handled correctly. A few small issues below, none of them blockers.

---

What the PR does

SubscriptionManager._wait_for_response is the sole reader of the socket during resubscribe_all. Between _sid_to_client.clear() and the new sid mapping landing, the server can emit data frames on the freshly-assigned sid. Previously those were logged-and-dropped. This PR stashes them in a per-sid bounded deque, then drains through _process_frame (not directly through dispatch) after resubscribe completes, so seq tracking and orderbook state see the frames exactly once in arrival order.

---

Correctness

The core logic is correct:

• try/finally in resubscribe_all guarantees _stashing = False even on exception — no way to leave the flag stuck on.
• take_stash() is atomic (swap-and-return in one expression), so the caller owns the data immediately.
• Routing replay through _process_frame rather than directly into the dispatcher is the right call — seq watermarks advance naturally and orderbook state stays consistent.
• Stale-old-sid frames (emitted on the pre-clear sids) end up in the stash keyed by that old sid, then dropped in _drain_resubscribe_stash because get_subscription_by_sid won't find them after _sid_to_client.clear(). That's the correct outcome.
• The defensive self._stash.clear() / self._stash_warned.clear() at the start of resubscribe_all is a sound hardening measure for the case where a prior cycle raised before the drain ran.

---

Issues

1. Redundant import logging inside a test body (minor bug)

test_resubscribe_stash_skips_unmapped_sids has a local import logging at line ~634 of the diff, but logging is already imported at the top of tests/ws/test_recv_loop_hardening.py. The local import shadows the module-level one harmlessly, but it's dead code that should be removed.

# tests/ws/test_recv_loop_hardening.py — inside test_resubscribe_stash_skips_unmapped_sids
import logging   # ← redundant; already imported at file top

2. PR-number comment in production code (CLAUDE.md style violation)

channels.py line ~213 has:

# Defensive reset: if a prior resubscribe cycle raised before
# _drain_resubscribe_stash ... (#187 review)

Per CLAUDE.md: "Don't reference the current task, fix, or callers … those belong in the PR description and rot as the codebase evolves." The explanation is valuable — just drop the (#187 review) tag.

3. Incidental get_event_loop() → get_running_loop() fix (positive, but note it)

The diff silently migrates two calls in _wait_for_response:

- deadline = asyncio.get_event_loop().time() + timeout
+ deadline = asyncio.get_running_loop().time() + timeout

get_running_loop() is the correct form inside an already-running coroutine (raises RuntimeError if there is no running loop, which is the right failure mode; get_event_loop() was deprecated for this use in 3.10). This is a good fix — just worth calling out explicitly in the CHANGELOG entry so it doesn't get lost.

---

Test coverage

Good. The 8 tests cover:

Scenario	Test
Per-sid arrival-order deques	test_maybe_stash_collects_by_sid
Control envelopes without sid → drop	test_maybe_stash_drops_frame_without_sid
maxlen eviction + single-warning-per-cycle gate	test_maybe_stash_maxlen_evicts_oldest_with_one_warning_per_fill
Atomic take-and-clear	test_take_stash_returns_and_clears
Default flag state	test_stashing_flag_off_by_default
Stale stash cleared at start of new cycle	test_resubscribe_clears_stale_stash_at_start
Replay reaches iterator queue	test_stash_drain_replays_through_dispatch
Seq tracker advances on replay (#139 coord.)	test_stash_drain_advances_seq_tracker_on_sequenced_channels
Unmapped sid drop with debug log	test_resubscribe_stash_skips_unmapped_sids

The decision to test drain mechanics directly rather than writing a timing-dependent race test is correct — race tests are flaky by nature and this approach gives full deterministic coverage of the dispatch path.

One gap worth noting: there is no test verifying that a normal (non-resubscribe) subscribe still silently drops non-matching frames (i.e., that _stashing=False keeps the old discard path intact). It's covered implicitly by the large existing test suite passing, but an explicit assertion would make the contract clearer.

---

Performance / memory

Memory bound is stash_maxlen × active_subs × avg_frame_size as documented. With stash_maxlen=1000 and ~1 KB frames, worst case is ~1 MB per active subscription — proportional to the user's subscribe footprint. The maxlen parameter is exposed on __init__ as keyword-only, which is the right API shape. No concerns here.

---

Summary

Two small fixes before merge:

1. Remove the redundant import logging inside test_resubscribe_stash_skips_unmapped_sids.
2. Drop (#187 review) from the comment in channels.py (keep the text, just drop the PR reference).

The get_running_loop() migration is a welcome bonus; worth a one-liner in the CHANGELOG if it's not already there.

[TexasCoding]

Addressed in 6b4db52:

1. Redundant import logging in test_resubscribe_stash_skips_unmapped_sids — removed. The module-level import already covers it.

2. (#187 review) PR-ref in production comment — dropped the tag, kept the rationale. Note: the CLAUDE.md quote you cited ("Don't reference the current task, fix, or callers … those belong in the PR description and rot as the codebase evolves") doesn't appear in this repo's CLAUDE.md or AGENTS.md. But the underlying point is sound — PR numbers rot once merged, issue numbers (WS: data frames arriving during reconnect+resubscribe are dropped instead of stashed #176, Decision needed: required-but-optional drift policy (22 REST + 12 WS models, ~190 field violations) #172, WS reconnect/resubscribe: 5 race-condition bugs in the recv loop #77) are the durable references. Aligned the comment with what existing production code does (no PR refs; freely uses issue refs).

3. CHANGELOG note for get_running_loop() migration — added a one-line drive-by entry under the WS: data frames arriving during reconnect+resubscribe are dropped instead of stashed #176 section so it doesn't get lost.

Not actioned: the suggested test for "_stashing=False keeps discard path intact." The existing test_stashing_flag_off_by_default locks in the default state, and the structural invariant — _maybe_stash is only called inside the if self._stashing: branch in _wait_for_response — is visible at the call site. A standalone discard-path test would need to mock the underlying recv() to feed a non-matching frame to _wait_for_response and then assert the stash stays empty; that's a lot of test scaffolding for a contract that's already locked by the existing test + code structure. Happy to add it if the contract gets harder to verify later.

Verification: 2126 unit tests pass. mypy kalshi/ clean, ruff check . clean.

[claude]

Code Review — PR #187: fix(ws): stash + replay data frames during resubscribe window

Overview

This PR fixes a real, observable bug: data frames sent by the server on a freshly-assigned sid, during the window between _sid_to_client.clear() and the new sid mapping landing in _wait_for_response, were silently dropped. The fix is well-targeted — stash frames in _wait_for_response during resubscribe_all, then drain through _process_frame after resubscribe completes. The design is minimal, the stash is bounded, and the CHANGELOG/PR description are thorough.

---

What's Good

Correctness of the core fix

• _stashing is guarded with try/finally → guaranteed to be cleared on any exception path. ✓
• take_stash() uses the swap idiom (stash, self._stash = self._stash, {}) — atomic in single-threaded async context. ✓
• Drain routes through _process_frame, not _dispatcher.dispatch directly → seq tracker and orderbook state see the replayed frames exactly once. This is the right call and the fix(ws): roll back seq on backpressure + reset all tickers on multi-ticker gap #139 coordination test verifies it. ✓
• Defensive stash clear at the top of resubscribe_all handles the edge case where a prior cycle raised before _drain_resubscribe_stash ran. ✓
• asyncio.get_running_loop() drive-by fix is correct; get_event_loop() inside a coroutine is deprecated in 3.10+. ✓

Memory bound

• Per-sid deque(maxlen=1000) is proportional to the user's subscribe footprint and bounded regardless of server burst rate. ✓
• Single WARNING per (sid, resubscribe-cycle) via _stash_warned prevents per-frame log spam on high-volume channels. ✓

Test coverage
Eight tests covering all the meaningful paths:

• Per-sid arrival-order deques, no-sid control envelope drop, maxlen eviction + warning dedup, warning reset between cycles, atomic take_stash, default stashing=False, stale stash cleared at cycle start, replay through dispatch, seq tracker coordination (fix(ws): roll back seq on backpressure + reset all tickers on multi-ticker gap #139), unmapped-sid graceful drop.

---

Issues

1. Untyped sid from data.get("sid") — minor

In _maybe_stash, sid = data.get("sid") returns Any (since data is dict[str, Any]). The stash is typed as dict[int, ...]. If a malformed server frame sends a string sid or a float, the bucket insertion would silently produce a mixed-type stash key. Mypy passes because Any satisfies int at the call site. A one-line guard would close the gap:

if not isinstance(sid, int):
    logger.debug("Stash mode: dropping frame with non-int sid: type=%s sid=%r", data.get("type"), sid)
    return

Not a blocker — the Kalshi protocol always sends integer sids — but it's a cheap defensive check.

2. Self-referential #187 review fix in a test docstring — cosmetic

test_maybe_stash_maxlen_evicts_oldest_with_one_warning_per_fill contains:

"one WARNING fires per fill event (not per frame) (#187 review fix)."

PR numbers in test docstrings become stale quickly and aren't meaningful to future readers. The what and why are already captured in the assertion comments. Worth trimming before merge.

3. _drain_resubscribe_stash: per-frame exception handler silences unexpected errors — minor

except Exception:
    logger.warning("Failed to replay stashed frame for sid %d", sid, exc_info=True)

Catching Exception broadly here means a programming error (e.g., a TypeError or AttributeError in _process_frame) is logged as a WARNING and silently skipped rather than propagating. Given that this is the orchestration path (not the hot recv loop), a failed frame won't crash the WebSocket, but it also won't surface bugs in _process_frame during development. Consider narrowing to KalshiError or at least re-raising on KeyboardInterrupt/SystemExit (already excluded by Exception, so that's fine). Alternatively, keeping it broad is defensible here — a bad stashed frame shouldn't abort the entire reconnect. Noting it for awareness.

4. Theoretical: server sid collision across resubscribes — edge case

If the server reuses a numeric sid value across two subscriptions (unlikely — sids are typically monotonically increasing), frames stashed under that value could be associated with the wrong subscription after the _sid_to_client rebuild. _process_frame would route them by the new mapping, not the stash key's origin. In practice this can't happen with the Kalshi protocol, but a comment noting that stash keys are assumed to be unique-per-reconnect would pre-empt confusion.

---

Summary

The fix is correct, well-bounded, and thoroughly tested. Issues 2 and 4 are cosmetic/theoretical. Issue 1 is a cheap defensive improvement. Issue 3 is a trade-off judgment call that's reasonably decided in the current direction.

Verdict: approve with minor suggestions. None of these block merge; the core race fix is sound.

[TexasCoding]

Round 4 addressed in 2dfebc1.

1. isinstance(sid, int) defensive guard — added. data.get("sid") returning Any was a real soft spot; a malformed frame with a non-int sid would corrupt the typed dict[int, ...] stash. The guard now catches None, strings, floats, and anything else non-int. Added a 5-line comment noting:

• The Kalshi protocol's monotonic-integer-sid contract
• The within-cycle sid-uniqueness assumption (your point 4) — keys collide only if the server reused a value, which would be a protocol bug, not a stash bug

Renamed the test test_maybe_stash_drops_frame_without_sid → test_maybe_stash_drops_frame_with_non_int_sid and expanded to cover all three cases (no sid, string sid, float sid). All drop with empty stash.

2. (#187 review fix) / #187 review: in test docstrings — dropped from both tests. The what + why was already in the docstring body; the PR tag was redundant.

Not actioned:

• Broad except Exception in _drain_resubscribe_stash — you noted "reasonably decided in the current direction." Keeping it: a malformed replayed frame shouldn't crash the reconnect orchestration; logging + continuing is the right call.
• Theoretical sid collision — covered by the new comment in _maybe_stash. No code change since the protocol guarantees uniqueness.

Verification: 2126 unit tests pass. ruff check . clean, mypy kalshi/ clean.

This is the 4th review round. Each one has found something real (overfire bug → sentinel divergence → close() exception safety → simultaneous-completion → defensive stash clear → now the sid type guard); the implementation is meaningfully tighter than the original push. Happy to keep iterating if there are more substantive items, but otherwise this looks ready to merge.

[claude]

Code Review — PR #187: WS resubscribe-window frame stashing

Overview

This fixes a real, observable signal-loss bug: frames sent by the server on a freshly-assigned sid, before the subscribe ack lands in _wait_for_response, were silently dropped with a debug log. The fix is architecturally sound — stash in SubscriptionManager (where the frames are intercepted), drain in KalshiWebSocket._handle_reconnect (which owns the reconnect lifecycle). The _process_frame replay path is the correct choice; it keeps seq tracking and orderbook state consistent without special-casing.

---

Correctness ✅

The core logic is correct:

• try/finally: self._stashing = False guarantees the flag is always reset, even on exception.
• Defensive _stash.clear() / _stash_warned.clear() at the top of resubscribe_all correctly handles the case where a prior cycle raised before take_stash() ran.
• The overflow guard (len(bucket) == self._stash_maxlen and sid not in self._stash_warned) fires correctly on the first overflow event (before the deque auto-evicts), and is gated by _stash_warned to suppress per-frame spam. The reset in take_stash() gives each resubscribe cycle its own fresh warning budget.
• take_stash() atomically swaps self._stash with {} — fine for single-threaded async code; the term "atomic" in the docstring is technically correct in this context.
• asyncio.get_event_loop().time() → asyncio.get_running_loop().time() in _wait_for_response is the correct cleanup: inside an async def the running-loop API is unambiguous.

---

Issues

1. Comment verbosity violates CLAUDE.md conventions (minor, but consistent)

CLAUDE.md specifies: "Never write multi-paragraph docstrings or multi-line comment blocks — one short line max." Several new additions exceed this:

• The 11-line _stash comment block in __init__ (lines ~77–85)
• The 7-line _stash_warned comment block (lines ~90–95)
• The multi-paragraph docstrings on _maybe_stash, take_stash, and _drain_resubscribe_stash

The PR description and commit message carry this context better. Suggest trimming each to a single # #176: <one-line why> comment and a one-line docstring.

2. No validation on stash_maxlen (minor)

stash_maxlen=0 would silently disable stashing (a deque(maxlen=0) drops every append immediately) and stash_maxlen < 0 raises at deque construction with an unhelpful error. A simple if stash_maxlen < 1: raise ValueError(...) guards against accidental misconfiguration. Not critical since this is a keyword-only param with a sensible default.

3. _drain_resubscribe_stash exception handling swallows per-frame errors silently after the first

except Exception:
    logger.warning("Failed to replay stashed frame for sid %d", sid, exc_info=True)

Continuing on a per-frame failure is the right choice, but if _process_frame is consistently failing for a sid (e.g., a deserialization bug), every frame for that sid logs a WARNING. Consider whether this should increment a counter or emit a single summary after draining, rather than one warning per failed frame. Not a blocker, but worth being deliberate about.

---

Test Coverage ✅

Eight new tests are well-scoped:

• TestResubscribeStash (unit): mechanics of stash, overflow, take_stash atomicity, default _stashing=False, and defensive clear on cycle start.
• TestResubscribeStashIntegration (integration): replay reaches iterator queue, seq tracker advances correctly (fix(ws): roll back seq on backpressure + reset all tickers on multi-ticker gap #139 coordination), unmapped sids drop gracefully.

The decision to skip a race-engineered end-to-end test is called out and justified. The asyncio_mode = "auto" setting in pyproject.toml covers the async test in TestResubscribeStash without a per-class marker.

One gap: no test for stash_maxlen=0. If that's a supported value (silent no-op), it should be documented. If it's not supported, add a validation test.

---

Performance ✅

Stash is a dict[int, deque] — O(1) append, proportional to active subscriptions × maxlen × avg frame size. Only active during resubscribe_all. No concerns.

Security ✅

No new attack surface. Stash is internal, bounded, and cleared each cycle.

---

Summary

The fix is correct and well-tested. Two issues worth addressing before merge:

1. Trim comment verbosity to comply with CLAUDE.md style.
2. Add stash_maxlen > 0 validation to fail fast on misconfiguration.

The per-frame exception handling in _drain_resubscribe_stash is worth a second look but is not a blocker.