From a77d9a41899f9ab2d60e55cc83118106a4f81d1c Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sun, 17 May 2026 12:45:41 -0500 Subject: [PATCH 1/3] fix(ws): roll back seq watermark on backpressure failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When MessageQueue.put() raises KalshiBackpressureError on an ERROR-strategy queue, the seq tracker had already advanced _last_seq for that sid. The dropped message's seq was then treated as already-seen, so a post-reconnect or post-resync stream would silently desync without ever firing a gap. Fix: capture peek(sid) before track(), wrap dispatch in a try/except, rollback the watermark if the dispatch raises BackpressureError, then re-raise so the recv loop's existing #83 handler broadcasts sentinels and tears the loop down (invariant preserved). Orderbook state is left as-is — the loop teardown via sentinel means the local book is orphaned, never desynced-and-still-consumed. Adds SequenceTracker.peek(sid) and rollback(sid, prev). Regression: maxsize=1 ERROR queue, snapshot fills, delta overflows -> tracker.peek(sid) must remain at the snapshot's seq, not advance to the dropped delta. Closes #78 Co-Authored-By: Claude Opus 4.7 (1M context) --- kalshi/ws/client.py | 28 ++++++- kalshi/ws/sequence.py | 24 ++++++ tests/ws/test_seq_correctness.py | 126 +++++++++++++++++++++++++++++++ 3 files changed, 176 insertions(+), 2 deletions(-) create mode 100644 tests/ws/test_seq_correctness.py diff --git a/kalshi/ws/client.py b/kalshi/ws/client.py index b499888..754669f 100644 --- a/kalshi/ws/client.py +++ b/kalshi/ws/client.py @@ -229,13 +229,25 @@ async def _recv_loop(self) -> None: raise async def _process_frame(self, raw: str) -> None: - """Parse, track seq, update orderbook, and dispatch a single frame.""" + """Parse, track seq, update orderbook, and dispatch a single frame. + + #78: track + orderbook-apply are provisional. If + :meth:`MessageDispatcher.dispatch` raises (e.g. + :class:`KalshiBackpressureError` on an ERROR-strategy queue), the + seq watermark is rolled back so the dropped message stays visible + as a future gap rather than being silently treated as already-seen. + Orderbook state isn't rolled back — the recv loop tears down via + sentinel broadcast on this error, so the local book becomes + orphaned but never desynced-and-still-consumed. + """ assert self._dispatcher is not None data = json.loads(raw) sid = data.get("sid") seq = data.get("seq") msg_type = data.get("type", "") + prev_seq: int | None = None + tracked = False if sid is not None and seq is not None and self._seq_tracker: channel = "" sub = ( @@ -245,9 +257,11 @@ async def _process_frame(self, raw: str) -> None: ) if sub: channel = sub.channel + prev_seq = self._seq_tracker.peek(sid) ok = await self._seq_tracker.track( sid, seq, msg_type if msg_type else channel ) + tracked = True if not ok: # Gap detected — skip dispatching this message. # The gap handler will trigger resync. @@ -261,7 +275,17 @@ async def _process_frame(self, raw: str) -> None: delta = OrderbookDeltaMessage.model_validate(data) self._orderbook_mgr.apply_delta(delta) - await self._dispatcher.dispatch(raw) + try: + await self._dispatcher.dispatch(raw) + except KalshiBackpressureError: + # #78: dispatch failed -> the consumer never saw this message. + # Roll the seq watermark back so the dropped seq is treated as + # a future gap (not silently as already-seen). Then re-raise so + # the recv loop's existing handler broadcasts sentinels and + # tears the loop down. + if tracked and sid is not None and self._seq_tracker: + self._seq_tracker.rollback(sid, prev_seq) + raise async def _handle_reconnect(self) -> None: """Reconnect after ConnectionClosed and re-establish subscriptions. diff --git a/kalshi/ws/sequence.py b/kalshi/ws/sequence.py index c97f453..c6f5e9f 100644 --- a/kalshi/ws/sequence.py +++ b/kalshi/ws/sequence.py @@ -74,6 +74,30 @@ async def track(self, sid: int, seq: int | None, channel: str) -> bool: return False + def peek(self, sid: int) -> int | None: + """Return the current last-seen seq for ``sid``, or None if untracked. + + Used by :class:`KalshiWebSocket._process_frame` to capture the + pre-track watermark so it can be restored via :meth:`rollback` if the + downstream dispatch fails (issue #78: ERROR-overflow backpressure must + not silently advance the seq watermark past a dropped message). + """ + return self._last_seq.get(sid) + + def rollback(self, sid: int, prev: int | None) -> None: + """Restore the last-seen seq for ``sid`` to ``prev``. + + If ``prev`` is None, the entry is removed entirely (the message was + the first one seen for this sid and never landed). This is the + compensation for a failed downstream dispatch — pair every successful + :meth:`track` whose dispatch may raise with a captured :meth:`peek` + and call this on failure. + """ + if prev is None: + self._last_seq.pop(sid, None) + else: + self._last_seq[sid] = prev + def reset(self, sid: int) -> None: """Reset tracking for a subscription (after resync/resubscribe).""" self._last_seq.pop(sid, None) diff --git a/tests/ws/test_seq_correctness.py b/tests/ws/test_seq_correctness.py new file mode 100644 index 0000000..e4cacc1 --- /dev/null +++ b/tests/ws/test_seq_correctness.py @@ -0,0 +1,126 @@ +"""Regression tests for WS seq-tracker correctness (#78). + +#78: ERROR-overflow backpressure must not silently advance the seq +watermark past a dropped message. + +Determinism: scenarios drive the seq tracker via the public API so we +don't rely on asyncio.sleep to race the recv loop. For the end-to-end +backpressure test, we await ``session._recv_task`` — the recv loop +completes once dispatch raises BackpressureError, no sleep required. +""" +from __future__ import annotations + +import asyncio + +import pytest + +from kalshi.config import KalshiConfig +from kalshi.ws.client import KalshiWebSocket +from kalshi.ws.sequence import SequenceTracker + + +@pytest.mark.asyncio +class TestSequenceTrackerRollback: + """#78: peek + rollback let callers undo a track() that downstream rejected.""" + + async def test_peek_returns_none_for_unknown_sid(self) -> None: + tracker = SequenceTracker() + assert tracker.peek(42) is None + + async def test_peek_returns_last_seq(self) -> None: + tracker = SequenceTracker() + await tracker.track(1, 5, "orderbook_delta") + assert tracker.peek(1) == 5 + + async def test_rollback_restores_prior_seq(self) -> None: + tracker = SequenceTracker() + await tracker.track(1, 5, "orderbook_delta") + prev = tracker.peek(1) + await tracker.track(1, 6, "orderbook_delta") + assert tracker.peek(1) == 6 + tracker.rollback(1, prev) + assert tracker.peek(1) == 5 + + async def test_rollback_to_none_removes_entry(self) -> None: + tracker = SequenceTracker() + # First message ever; prev is None. + prev = tracker.peek(1) + await tracker.track(1, 1, "orderbook_delta") + assert tracker.peek(1) == 1 + tracker.rollback(1, prev) + assert tracker.peek(1) is None + # And the next track is treated as the first message again. + assert await tracker.track(1, 1, "orderbook_delta") is True + + +@pytest.mark.asyncio +class TestBackpressureDoesNotAdvanceSeq: + """#78 end-to-end: ERROR-overflow on a full queue must not advance _last_seq.""" + + async def test_dropped_delta_keeps_seq_watermark_unchanged( + self, + fake_ws, # type: ignore[no-untyped-def] + test_auth, # type: ignore[no-untyped-def] + ) -> None: + """ERROR-strategy queue at maxsize -> send N+1 messages -> tracker's + last_seq reflects only successfully-landed messages. + + Setup: subscribe with maxsize=1, send snapshot (seq=1) — lands in + queue. Send delta (seq=2) — backpressure raises. The recv loop + exits via sentinel broadcast. Tracker's last_seq for the sid must + still be 1, not 2. + """ + config = KalshiConfig(ws_base_url=fake_ws.url, timeout=5.0) + ws = KalshiWebSocket(auth=test_auth, config=config) + async with ws.connect() as session: + stream = await session.subscribe_orderbook_delta( + tickers=["T1"], maxsize=1, + ) + + assert session._sub_mgr is not None + assert session._seq_tracker is not None + sid = next(iter(session._sub_mgr.active_subscriptions.values())).server_sid + assert sid is not None + + # Snapshot fills the queue (maxsize=1). + await fake_ws.send_to_all({ + "type": "orderbook_snapshot", "sid": sid, "seq": 1, + "msg": { + "market_ticker": "T1", "market_id": "x", + "yes": [["0.50", "100"]], "no": [], + }, + }) + # Yield so the recv loop processes the snapshot before we + # push the overflow delta. + await asyncio.sleep(0.1) + + assert session._seq_tracker.peek(sid) == 1 + + # This delta overflows -> BackpressureError -> recv loop tears + # down via sentinel broadcast. + await fake_ws.send_to_all({ + "type": "orderbook_delta", "sid": sid, "seq": 2, + "msg": { + "market_ticker": "T1", "market_id": "x", + "price": "0.51", "delta": "10", "side": "yes", + }, + }) + + recv_task = session._recv_task + assert recv_task is not None + await asyncio.wait_for(recv_task, timeout=2.0) + + # #78: the seq watermark MUST still be 1, not 2. If it were 2, + # a post-reconnect message with seq=2 would be treated as a + # duplicate (silent desync); a message with seq=3 would not + # be detected as a gap. + assert session._seq_tracker.peek(sid) == 1, ( + "#78 regression: ERROR-overflow advanced last_seq past a " + "dropped message; orderbook is now silently desynced." + ) + + # Drain the iterator so the session can shut down cleanly. + collected: list[object] = [] + async for msg in stream: + collected.append(msg) + assert len(collected) == 1 # only the snapshot landed From 55d5f35381f405a4d7b9697caeae48e30a7d6d5e Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sun, 17 May 2026 12:46:14 -0500 Subject: [PATCH 2/3] fix(ws): clear every ticker on multi-ticker orderbook gap A single orderbook_delta subscription with tickers=["A","B","C"] returns one sid covering all books. On a sequence gap, _handle_seq_gap was only clearing tickers[0] from the orderbook manager, leaving the rest with stale state that diverges from server truth. The gap envelope does not identify which ticker missed an update, so the safe option is to clear every ticker in the affected subscription. Iterate the full market_tickers list. Single-ticker subs and subscribe-to-all subs (no market_tickers param) continue to work unchanged. Closes #79 Co-Authored-By: Claude Opus 4.7 (1M context) --- kalshi/ws/client.py | 17 ++- tests/ws/test_multi_ticker_gap.py | 168 ++++++++++++++++++++++++++++++ 2 files changed, 181 insertions(+), 4 deletions(-) create mode 100644 tests/ws/test_multi_ticker_gap.py diff --git a/kalshi/ws/client.py b/kalshi/ws/client.py index 754669f..c98e02e 100644 --- a/kalshi/ws/client.py +++ b/kalshi/ws/client.py @@ -319,7 +319,14 @@ async def _handle_reconnect(self) -> None: self._running = False async def _handle_seq_gap(self, gap: SequenceGap) -> None: - """Handle a sequence gap by logging and triggering resync.""" + """Handle a sequence gap by logging and triggering resync. + + #79: a single ``orderbook_delta`` subscription can cover multiple + tickers under one sid. The gap envelope does not identify which + ticker missed an update, so the safe option is to clear EVERY + ticker for the affected subscription. Previously only ``tickers[0]`` + was cleared, leaving the other books diverged from server truth. + """ logger.warning( "Sequence gap on sid %d: expected %d, got %d. Triggering resync.", gap.sid, gap.expected, gap.received, @@ -327,10 +334,12 @@ async def _handle_seq_gap(self, gap: SequenceGap) -> None: if self._sub_mgr: sub = self._sub_mgr.get_subscription_by_sid(gap.sid) if sub and sub.channel == "orderbook_delta": - # Clear orderbook state for this ticker and reset sequence tracking + # Clear orderbook state for ALL tickers in the sub and reset + # sequence tracking so the next snapshot rebootstraps cleanly. tickers = sub.params.get("market_tickers", []) - if tickers and self._orderbook_mgr: - self._orderbook_mgr.remove(tickers[0]) + if self._orderbook_mgr: + for ticker in tickers: + self._orderbook_mgr.remove(ticker) if self._seq_tracker: self._seq_tracker.reset(gap.sid) diff --git a/tests/ws/test_multi_ticker_gap.py b/tests/ws/test_multi_ticker_gap.py new file mode 100644 index 0000000..a0a8fe4 --- /dev/null +++ b/tests/ws/test_multi_ticker_gap.py @@ -0,0 +1,168 @@ +"""Regression tests for multi-ticker orderbook_delta gap clear (#79). + +A single ``orderbook_delta`` subscription can cover multiple tickers +under one sid. The gap envelope does not identify which ticker missed +an update, so the safe option is to clear EVERY ticker for the affected +subscription. Previously only ``tickers[0]`` was cleared, leaving the +other books diverged from server truth. + +Determinism: scenarios drive the gap handler directly via the public +API (``session._handle_seq_gap``) so we don't race the recv loop forcing +a real gap on the wire. +""" +from __future__ import annotations + +import asyncio + +import pytest +from cryptography.hazmat.primitives.asymmetric import rsa + +from kalshi.auth import KalshiAuth +from kalshi.config import KalshiConfig +from kalshi.ws.backpressure import MessageQueue +from kalshi.ws.channels import Subscription +from kalshi.ws.client import KalshiWebSocket +from kalshi.ws.orderbook import OrderbookManager +from kalshi.ws.sequence import SequenceGap, SequenceTracker + + +@pytest.mark.asyncio +class TestMultiTickerGapClear: + """#79: gap on a multi-ticker orderbook_delta sub clears EVERY ticker.""" + + async def test_gap_clears_all_tickers_in_sub( + self, + fake_ws, # type: ignore[no-untyped-def] + test_auth, # type: ignore[no-untyped-def] + ) -> None: + """Subscribe to tickers=["A","B","C"], seed all three books via + snapshots, then fire a gap on the shared sid -> all three books + must be cleared (previously only "A" was cleared). + """ + config = KalshiConfig(ws_base_url=fake_ws.url, timeout=5.0) + ws = KalshiWebSocket(auth=test_auth, config=config) + async with ws.connect() as session: + await session.subscribe_orderbook_delta( + tickers=["A", "B", "C"], + ) + + assert session._sub_mgr is not None + assert session._orderbook_mgr is not None + assert session._seq_tracker is not None + sid = next(iter(session._sub_mgr.active_subscriptions.values())).server_sid + assert sid is not None + + # Seed all three books via individual snapshots on the same sid. + for i, ticker in enumerate(["A", "B", "C"], start=1): + await fake_ws.send_to_all({ + "type": "orderbook_snapshot", "sid": sid, "seq": i, + "msg": { + "market_ticker": ticker, "market_id": "x", + "yes": [["0.50", "100"]], "no": [], + }, + }) + + # Wait for all three to land in the manager via an async-event + # poll (no fixed sleep timing). + async def all_seeded() -> bool: + return all( + session._orderbook_mgr.get(t) is not None # type: ignore[union-attr] + for t in ("A", "B", "C") + ) + + async with asyncio.timeout(2.0): + while not await all_seeded(): + await asyncio.sleep(0.01) + + # Directly invoke the gap handler — equivalent to the seq + # tracker firing on_gap on its own. Doing it directly removes + # any flakiness around forcing a real gap through the wire. + await session._handle_seq_gap( + SequenceGap(sid=sid, expected=4, received=10) + ) + + # #79: every ticker in the sub must be cleared. + assert session._orderbook_mgr.get("A") is None, ( + "#79: A was cleared (already worked pre-fix)" + ) + assert session._orderbook_mgr.get("B") is None, ( + "#79 regression: B's book was not cleared; only tickers[0] " + "was cleared, leaving B diverged from server truth." + ) + assert session._orderbook_mgr.get("C") is None, ( + "#79 regression: C's book was not cleared; only tickers[0] " + "was cleared, leaving C diverged from server truth." + ) + + # Seq tracker is also reset for the sid. + assert session._seq_tracker.peek(sid) is None + + async def test_single_ticker_sub_still_cleared( + self, + fake_ws, # type: ignore[no-untyped-def] + test_auth, # type: ignore[no-untyped-def] + ) -> None: + """Sanity check: single-ticker case still clears the (only) ticker.""" + config = KalshiConfig(ws_base_url=fake_ws.url, timeout=5.0) + ws = KalshiWebSocket(auth=test_auth, config=config) + async with ws.connect() as session: + await session.subscribe_orderbook_delta(tickers=["SOLO"]) + + assert session._sub_mgr is not None + assert session._orderbook_mgr is not None + sid = next(iter(session._sub_mgr.active_subscriptions.values())).server_sid + assert sid is not None + + await fake_ws.send_to_all({ + "type": "orderbook_snapshot", "sid": sid, "seq": 1, + "msg": { + "market_ticker": "SOLO", "market_id": "x", + "yes": [["0.50", "100"]], "no": [], + }, + }) + + async with asyncio.timeout(2.0): + while session._orderbook_mgr.get("SOLO") is None: + await asyncio.sleep(0.01) + + await session._handle_seq_gap( + SequenceGap(sid=sid, expected=2, received=5) + ) + assert session._orderbook_mgr.get("SOLO") is None + + +@pytest.mark.asyncio +class TestGapHandlerWithoutTickers: + """#79 edge case: an orderbook_delta sub with no market_tickers param + (subscribe-to-all) shouldn't crash the gap handler. + """ + + async def test_no_tickers_param_does_not_raise(self) -> None: + # We just need a KalshiWebSocket instance with the managers wired + # up directly — no real connection. + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + auth = KalshiAuth(key_id="test", private_key=key) + config = KalshiConfig(ws_base_url="ws://localhost:1") + ws = KalshiWebSocket(auth=auth, config=config) + + ws._orderbook_mgr = OrderbookManager() + ws._seq_tracker = SequenceTracker() + + # Hand-build a SubscriptionManager-shaped object with one sub. + class _StubMgr: + def __init__(self) -> None: + queue: MessageQueue[object] = MessageQueue() + self._sub = Subscription( + client_id=1, channel="orderbook_delta", + params={}, # no market_tickers + queue=queue, + ) + self._sub.server_sid = 42 + + def get_subscription_by_sid(self, sid: int) -> Subscription | None: + return self._sub if sid == 42 else None + + ws._sub_mgr = _StubMgr() # type: ignore[assignment] + + # Should not raise — no tickers to iterate, just reset seq. + await ws._handle_seq_gap(SequenceGap(sid=42, expected=1, received=5)) From 1b27d7cfa5dcda28aabc5988e5368c062b51bb0e Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sun, 17 May 2026 12:56:46 -0500 Subject: [PATCH 3/3] review(#139): pin broken transitive + address bot nits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🔴 CI failure: ast-serialize 0.5.0 ships cp314-only wheels mypy 2.1.0 pulls in ast-serialize transitively. ast-serialize 0.5.0 was published between Wave 3 landing and Wave 4 PR open; it dropped the cp39-abi3 wheels and only ships cp314, breaking CI's fresh resolution on every Python 3.12/3.13 runner (uv.lock is gitignored so each CI run re-resolves). Drift-check + test (3.13) both failed at `uv sync`. Fix: add `ast-serialize<0.5` to the dev dependency-group with a comment explaining the situation and the trigger to remove it. Fresh local re-resolve picks ast-serialize 0.4.0 cleanly with the existing mypy 2.1.0. 🟦 Bot review nits (per pass on #139): - `tracked = True` was set before the `if not ok: return` early-exit branch; the value was only meaningful inside the `except KalshiBackpressureError` block, where the early return makes it unreachable. Moved the assignment after the early return so the read order matches the intent. - Replaced `await asyncio.sleep(0.1)` in test_dropped_delta_keeps_seq_watermark_unchanged with a poll loop on `peek(sid) == 1` bounded by asyncio.wait_for(2.0). Matches the pattern already used in test_multi_ticker_gap.py. - Stripped #78 / #79 references from production-code docstrings and inline comments per CLAUDE.md §3 (issue numbers belong in PR/commit context, not source). The invariants themselves are still documented — just without the "#78:" prefix. Test-file docstrings retain the cross-references since that's reasonable in tests. Verify: tests/ws/test_seq_correctness.py + test_multi_ticker_gap.py 8 passed. ruff + mypy --strict clean. Fresh uv lock resolves ast-serialize 0.4.0. Co-Authored-By: Claude Opus 4.7 (1M context) --- kalshi/ws/client.py | 39 ++++++++++++++++---------------- kalshi/ws/sequence.py | 8 +++---- pyproject.toml | 4 ++++ tests/ws/test_seq_correctness.py | 9 +++++--- 4 files changed, 34 insertions(+), 26 deletions(-) diff --git a/kalshi/ws/client.py b/kalshi/ws/client.py index c98e02e..4fe3e2c 100644 --- a/kalshi/ws/client.py +++ b/kalshi/ws/client.py @@ -231,14 +231,14 @@ async def _recv_loop(self) -> None: async def _process_frame(self, raw: str) -> None: """Parse, track seq, update orderbook, and dispatch a single frame. - #78: track + orderbook-apply are provisional. If - :meth:`MessageDispatcher.dispatch` raises (e.g. - :class:`KalshiBackpressureError` on an ERROR-strategy queue), the - seq watermark is rolled back so the dropped message stays visible - as a future gap rather than being silently treated as already-seen. - Orderbook state isn't rolled back — the recv loop tears down via - sentinel broadcast on this error, so the local book becomes - orphaned but never desynced-and-still-consumed. + Seq tracking + orderbook-apply are provisional with respect to the + downstream dispatch. If dispatch raises ``KalshiBackpressureError`` + (ERROR-strategy queue overflow), the seq watermark is rolled back + so the dropped message stays visible as a future gap rather than + being silently treated as already-seen. Orderbook state isn't + rolled back — the recv loop tears down via sentinel broadcast on + this error, so the local book becomes orphaned but is never + desynced-and-still-consumed. """ assert self._dispatcher is not None data = json.loads(raw) @@ -261,11 +261,12 @@ async def _process_frame(self, raw: str) -> None: ok = await self._seq_tracker.track( sid, seq, msg_type if msg_type else channel ) - tracked = True if not ok: # Gap detected — skip dispatching this message. # The gap handler will trigger resync. return + # Only meaningful once we know we'll dispatch (and might roll back). + tracked = True # Check for orderbook messages if msg_type == "orderbook_snapshot" and self._orderbook_mgr: @@ -278,11 +279,11 @@ async def _process_frame(self, raw: str) -> None: try: await self._dispatcher.dispatch(raw) except KalshiBackpressureError: - # #78: dispatch failed -> the consumer never saw this message. - # Roll the seq watermark back so the dropped seq is treated as - # a future gap (not silently as already-seen). Then re-raise so - # the recv loop's existing handler broadcasts sentinels and - # tears the loop down. + # Dispatch failed -> the consumer never saw this message. Roll + # the seq watermark back so the dropped seq is treated as a + # future gap (not silently as already-seen). Re-raise so the + # recv loop's existing handler broadcasts sentinels and tears + # the loop down. if tracked and sid is not None and self._seq_tracker: self._seq_tracker.rollback(sid, prev_seq) raise @@ -321,11 +322,11 @@ async def _handle_reconnect(self) -> None: async def _handle_seq_gap(self, gap: SequenceGap) -> None: """Handle a sequence gap by logging and triggering resync. - #79: a single ``orderbook_delta`` subscription can cover multiple - tickers under one sid. The gap envelope does not identify which - ticker missed an update, so the safe option is to clear EVERY - ticker for the affected subscription. Previously only ``tickers[0]`` - was cleared, leaving the other books diverged from server truth. + A single ``orderbook_delta`` subscription can cover multiple tickers + under one sid. The gap envelope doesn't identify which ticker + missed an update, so we clear EVERY ticker on the affected + subscription — clearing only ``tickers[0]`` would leave the other + books silently diverged from server truth. """ logger.warning( "Sequence gap on sid %d: expected %d, got %d. Triggering resync.", diff --git a/kalshi/ws/sequence.py b/kalshi/ws/sequence.py index c6f5e9f..863b297 100644 --- a/kalshi/ws/sequence.py +++ b/kalshi/ws/sequence.py @@ -77,10 +77,10 @@ async def track(self, sid: int, seq: int | None, channel: str) -> bool: def peek(self, sid: int) -> int | None: """Return the current last-seen seq for ``sid``, or None if untracked. - Used by :class:`KalshiWebSocket._process_frame` to capture the - pre-track watermark so it can be restored via :meth:`rollback` if the - downstream dispatch fails (issue #78: ERROR-overflow backpressure must - not silently advance the seq watermark past a dropped message). + Capture this before calling :meth:`track` so the watermark can be + restored via :meth:`rollback` if downstream dispatch fails — an + already-advanced watermark would silently treat the dropped message + as already-seen on the next gap check. """ return self._last_seq.get(sid) diff --git a/pyproject.toml b/pyproject.toml index 44aedc9..ec5e2a1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,6 +81,10 @@ dev = [ "respx>=0.21,<1", "ruff>=0.8,<1", "mypy>=1.13,<3", + # Transitive of mypy 2.1.x. Pin <0.5 because 0.5.0 only ships cp314 + # wheels (cp39-abi3 dropped), breaking CI on Python 3.12 + 3.13 runners. + # Drop once upstream republishes broader wheels. + "ast-serialize<0.5", "datamodel-code-generator>=0.26", "pyyaml>=6", "python-dotenv>=1", diff --git a/tests/ws/test_seq_correctness.py b/tests/ws/test_seq_correctness.py index e4cacc1..f26922d 100644 --- a/tests/ws/test_seq_correctness.py +++ b/tests/ws/test_seq_correctness.py @@ -90,9 +90,12 @@ async def test_dropped_delta_keeps_seq_watermark_unchanged( "yes": [["0.50", "100"]], "no": [], }, }) - # Yield so the recv loop processes the snapshot before we - # push the overflow delta. - await asyncio.sleep(0.1) + # Deterministic wait: poll the seq tracker until it observes the + # snapshot, bounded by wait_for. Avoids CI flakes under load. + async def snapshot_landed() -> None: + while session._seq_tracker.peek(sid) != 1: + await asyncio.sleep(0.01) + await asyncio.wait_for(snapshot_landed(), timeout=2.0) assert session._seq_tracker.peek(sid) == 1