[TexasCoding]

Closes #241
Closes #245
Closes #244

Part of #273 (post-v2.4 audit), Wave 1 money-risk batch.

[claude]

Code Review — PR #278

fix+perf(ws): seq rollback on pre-validation failure; drop per-frame shield; cache materialized orderbook

Summary: Three distinct fixes bundled together — #241 (seq rollback), #245 (cooperative pause), #244 (orderbook cache). The correlation makes sense: all three touch the recv-loop / _process_frame / orderbook hot path and would have conflicted as separate PRs.

---

#241 — Seq rollback on pre-validation failure

The fix is correct and important. Moving the orderbook pre-validation inside the existing try block and adding a catch-all except Exception rollback are both the right calls.

One question on the except Exception clause in _process_frame:

except Exception:
    if tracked and sid is not None and self._seq_tracker:
        self._seq_tracker.rollback(sid, prev_seq)
    raise

KalshiBackpressureError already has its own rollback above, so no double-rollback is possible. But KalshiSubscriptionError now falls into this clause too — is that the intended behaviour? If KalshiSubscriptionError should also trigger a seq rollback, a comment clarifying that would help future readers (the inline comment only mentions ValidationError / KeyError).

dispatch.py change (return → raise): straightforward and correct. The old silent swallow was the root of the desync.

---

#245 — Cooperative pause (drop per-frame shield)

The cooperative design is cleaner and eliminates the per-frame create_task / shield overhead. A few things to check:

Potential deadlock: recv loop dies while _pause_recv_loop is waiting

If the connection dies (e.g., ConnectionClosed takes the break path) after _pause_recv_loop has set _pause_pending=True but before the loop reaches the top-of-loop checkpoint, _pause_granted is never set and _pause_recv_loop blocks forever on await self._pause_granted.wait().

The _stop() fix sets _resume_signal (to unblock a parked loop), but that doesn't help _pause_recv_loop which is blocked on _pause_granted, not _resume_signal. Mitigations worth considering:

# Option A: timeout in _pause_recv_loop
try:
    await asyncio.wait_for(self._pause_granted.wait(), timeout=2.0)
except asyncio.TimeoutError:
    self._pause_pending = False  # loop is gone, abandon pause

# Option B: _stop also sets _pause_granted
async def _stop(self) -> None:
    self._running = False
    self._pause_granted.set()   # unblock any in-flight _pause_recv_loop
    self._resume_signal.set()
    ...

Option B is one line and directly reflects the invariant: if we're stopping, any pending pause should be considered "granted" (the loop won't come back anyway).

_stop() exception suppression is too broad:

with contextlib.suppress(asyncio.CancelledError, Exception):
    await asyncio.wait_for(self._recv_task, timeout=2.0)

Exception silently swallows RuntimeError, AttributeError, bugs, etc. during shutdown. Prefer:

with contextlib.suppress(asyncio.CancelledError, asyncio.TimeoutError):
    await asyncio.wait_for(self._recv_task, timeout=2.0)

Unexpected exceptions here are worth seeing, even at shutdown time.

asyncio.wait_for task overhead comment:

On a busy channel the timeout is cancelled before firing, so polling adds no overhead in the hot path.

In Python < 3.12, asyncio.wait_for wraps the coroutine in a Task unconditionally, even when the timeout never fires. The task allocation is still there on every iteration. The overhead is much less than the old per-frame shield approach, but the comment overstates it slightly. Worth a hedged reword or a # (pre-3.12: still one Task per poll, but far fewer than one per frame) note.

---

#244 — Materialized orderbook cache

Clean implementation. The invalidate() helper is a nice explicit signal.

_cached field should be init=False:

_cached: Orderbook | None = field(default=None, repr=False, compare=False)

Without init=False, _cached appears in _BookState.__init__'s signature. Any call site using positional arguments could silently pass an unexpected value. Since this is purely derived state, field(default=None, repr=False, compare=False, init=False) is the correct form.

_apply_snapshot_inplace implicit invalidation: The test test_snapshot_resets_cache passes, implying the implementation creates a fresh _BookState on snapshot. This is the right approach — no explicit invalidate() call needed. Worth adding a one-line note in _apply_snapshot_inplace's docstring so future maintainers don't wonder.

---

Tests

Overall test quality is high:

• test_seq_rollback.py: end-to-end fake-server approach is the right call. Using asyncio.wait_for + polling loops instead of fixed sleep is correct.
• test_orderbook_perf.py: smoke tests for identity stability are lightweight and meaningful.
• test_recv_loop_hardening.py: the added assert not pause_task.done() assertion makes the F-P-04 invariant explicit and machine-checkable.

Minor nit on test_seq_rollback.py: session._seq_tracker._on_gap is a private field on a private field. If SequenceTracker ever moves this callback to a different name, the test silently stops exercising anything. Consider exposing a for_testing_inject_gap_callback(fn) helper, or at minimum adding an assert session._seq_tracker._on_gap is capture guard after the sequence events to confirm the hook was still wired.

---

Summary

Area	Status
#241 seq rollback	✅ Correct. Minor: KalshiSubscriptionError rollback intent should be documented.
#245 cooperative pause	✅ Cleaner than shield. Deadlock on mid-pause connection drop needs a one-line fix (_stop sets _pause_granted). Broad Exception suppression in _stop should be narrowed.
#244 orderbook cache	✅ Works correctly. init=False missing from _cached field.
Tests	✅ Good coverage, correct polling approach.

The deadlock risk in _pause_recv_loop and the _stop exception suppression are the two items I'd fix before merging. Everything else is polish/minor.

[claude]

Code Review — PR #278: seq rollback + cooperative pause + orderbook cache

Summary: Three focused fixes addressing real correctness and performance problems in the WebSocket layer. The code is well-reasoned and the intent is clear throughout. A few issues worth addressing before merge.

---

#241 — Seq Rollback on Pre-validation Failure

Correct and well-structured. The two-part fix (re-raise in dispatch.py, rollback in _process_frame's new except Exception handler) cleanly closes the gap-detection hole.

One subtle observation: the except Exception block in _process_frame also covers KalshiSubscriptionError. It rolls back the seq watermark and re-raises, then the recv loop's except (KalshiBackpressureError, KalshiSubscriptionError) handler catches it. That path is correct — the loop is exiting anyway — but it's a non-obvious interaction worth a brief comment, since it's easy for a future reader to wonder whether KalshiSubscriptionError should be excluded from this catch.

Partial-mutation / rollback asymmetry: if _apply_snapshot_inplace or _apply_delta_inplace succeeds but dispatch() then raises, the orderbook has been mutated but the consumer never saw the message. The seq rollback surfaces this as a gap that drives resubscribe+snapshot recovery, which resets the orderbook — so the end state is correct. The comment in _process_frame explains the seq side but doesn't mention the orderbook mutation side. A one-liner like "orderbook mutation is not rolled back; gap-triggered resubscribe resets it" would close the reasoning for future readers.

---

#245 — Drop Per-Frame Shield; Cooperative Pause

The cooperative pause design is cleaner than the previous shield approach. Replacing a per-frame Task allocation + contextvars copy with a simple boolean flag and two Events is a meaningful hot-path improvement, and the F-P-04 invariant is correctly preserved at the top-of-loop checkpoint.

Issue — _pause_recv_loop double-pause deadlock (no guard):

If _pause_recv_loop is called when _pause_pending is already True (recv loop currently parked on _resume_signal), the method will:

1. Clear _pause_granted
2. Clear _resume_signal (which the parked loop is waiting on)
3. Set _pause_pending = True (no change)
4. Block on _pause_granted.wait() (which the loop can no longer set, because it's no longer at the checkpoint)

Result: deadlock. This is prevented in practice by _subscribe_lock, but there's no assertion or guard to catch a programming error that bypasses the lock. Adding assert not self._pause_pending at the top of _pause_recv_loop would make the invariant explicit and catch bugs early.

Issue — missing test for _stop with paused loop:

The _stop method's self._resume_signal.set() call (lines 46–57) is the only path that wakes a loop parked on _resume_signal during shutdown. This is a critical recovery path with no test coverage. A test along the lines of: pause the loop, call _stop, assert the task exits cleanly would close the gap.

Minor — _stop comment vs. actual exit path:

The comment on the wait_for call says "Closing the connection unblocks any in-flight recv()", but _stop doesn't close the connection — the actual exit path for a non-paused loop is the 50ms _RECV_POLL_S timeout, after which while self._running evaluates False. The comment is misleading; it should reference the polling exit path instead. (Connection close likely happens in the context manager's __aexit__, which is a different concern.)

---

#244 — Cached Materialized Orderbook

Clean and correct. The invalidation coverage is complete: snapshots create a new _BookState (cache starts None); both mutation paths in _apply_delta_inplace call invalidate(); the no-op delta path (delta <= 0, no existing qty) correctly skips invalidate(). Immutability of handed-out Orderbook objects is preserved at the value level.

Minor — _cached field should use init=False:

_cached: Orderbook | None = field(default=None, repr=False, compare=False)

Without init=False, _cached appears as a keyword parameter in _BookState.__init__. Since it's a private implementation detail that callers should never pass, init=False makes the intent explicit and prevents accidental misuse:

_cached: Orderbook | None = field(default=None, init=False, repr=False, compare=False)

---

Test Coverage

Strengths:

• test_seq_rollback.py is excellent: end-to-end via fake_ws, exercises _recv_loop + _process_frame, covers both orderbook_delta and order_group_updates channels, deterministic polling instead of fixed sleeps.
• test_orderbook_perf.py properly asserts object identity for cache hits and identity bust across mutations.
• Updated test_recv_loop_hardening.py correctly validates the new cooperative pause semantics, including the new in-flight assertion before allow_dispatch_finish.set().

Gaps:

• No test for _stop while the recv loop is paused (see above).
• No test for the _ensure_recv_loop early-return path (if self._pause_pending: ... return) — this branch resumes a parked loop; an integration test verifying that the loop actually resumes and processes frames post-subscribe would close it.

---

Summary

Area	Status
#241 seq rollback correctness	✅ Correct
#245 cooperative pause correctness	✅ Correct; minor guard/test gaps
#244 orderbook cache correctness	✅ Correct; init=False suggested
Test coverage	Mostly solid; 2 missing paths
Comments / docstrings	Clear; 2 minor clarifications suggested

The core logic is sound. The _pause_recv_loop double-pause guard and the _stop-with-paused-loop test are the most worth addressing before merge; the rest are polish.