polish(ws,auth): MessageQueue maxlen + types consolidation + RSA-PSS sign offload (#173 + #174 + #178)

[TexasCoding]

Closes #173, #174, #178.

Three independent WS / auth polish items batched into one PR.

#173 — MessageQueue defense-in-depth

kalshi/ws/backpressure.py — the underlying collections.deque now carries maxlen=maxsize+1 as a hard memory ceiling enforced by deque itself, independent of the manual _size counter.

If the counter ever drifts (a put path that forgets to increment, an exception between append and increment) the buffer cannot grow without bound — maxlen evicts at the C level. +1 is the safety margin so the manual DROP_OLDEST / RAISE overflow logic still gets to observe the boundary one slot before deque silently evicts; put_sentinel consumes the +1.

New regression test in tests/ws/test_backpressure.py injects counter drift (q._size = 0 mid-stream) and asserts the cap holds for the next 1000 puts.

#174 — _to_decimal_* consolidation

kalshi/types.py — _to_decimal_dollars and _to_decimal_fp were byte-identical apart from docstrings. Collapsed into a single private _coerce_decimal helper shared by both DollarDecimal and FixedPointCount. The public Annotated aliases stay distinct (different BeforeValidator instances would have been fine for Pydantic identity, but sharing the same validator is fine too — Pydantic distinguishes by the Annotated tuple, not the inner function identity, and audit confirmed no test introspects the validator function).

#178 — async RSA-PSS sign offload

KalshiAuth.sign_request_async() routes the ~1-10 ms RSA-PSS sign through a dedicated ThreadPoolExecutor(max_workers=2), lazy-initialised on first use. Async REST (AsyncTransport.request) and async WS connect (ConnectionManager._build_auth_headers) now use the async sign path; the sync sign_request API is unchanged.

Community feedback (issue #178 comment)

The commenter raised three substantive points. Triaging:

1. Shared default executor collision with getaddrinfo — adopted. The original acceptance criterion called asyncio.to_thread() which dispatches to loop.run_in_executor(None, ...), sharing the default pool with loop.getaddrinfo and other to_thread() work. A reconnect storm where DNS dominates (5-50 ms) would push the sign behind it and reintroduce the loop stall as await-latency. This PR uses a dedicated ThreadPoolExecutor(max_workers=2) for sign-only work, isolating crypto from any unrelated executor traffic at trivial cost (2 threads, lazy-init).

2. asyncio.sleep(0) is not a wall-clock ticker — adopted. The original acceptance methodology was wrong: asyncio.sleep(0) is special-cased to a single call_soon yield (tasks.__sleep0) and does NOT measure wall-clock blocking. The microbench in scripts/bench_sign_offload.py uses real asyncio.sleep(target_interval) ticks and measures loop.time() deltas.

3. DNS dominates over sign on reconnect — acknowledged. Valid for the WS reconnect path; doesn't change the offload value for the batch-REST-burst path (no DNS, just many signs). The dedicated executor matters more than the offload itself for one of the two cited scenarios; both ship in this PR.

Microbench results

scripts/bench_sign_offload.py, 2048-bit key, 200 signs vs. 50x ticker @ 5 ms:

    inline (on loop): p50=  0.67 ms  p95=  2.22 ms  p99=  2.95 ms
offloaded (executor): p50=  0.12 ms  p95=  0.68 ms  p99=  0.68 ms

Inline gaps track the sign cost (1-3 ms per sign on 2048-bit). Offloaded gaps stay at scheduler jitter.

Behavior tests

tests/test_auth.py::TestSignRequestAsync — 4 tests:

• test_async_signature_verifies_for_same_message — RSA-PSS is randomized (PSS salt), so sync vs. async signatures are byte-different but both verify against the same public key + canonical message.
• test_lazy_executor_creation — executor lazy-initialised on first call, reused on subsequent calls, cleared by close().
• test_close_is_idempotent — close() safe to call before any sign and twice in a row.
• test_concurrent_signs_do_not_stall_event_loop — in-suite version of the microbench; bound is conservative (max gap < 30 ms) so CI runners don't flake.

Lifecycle

KalshiClient.close() / AsyncKalshiClient.close() now shut down the sign executor too. The executor is daemon-style — long-lived clients that don't call close() get cleaned up via interpreter-shutdown anyway (atexit-joined).

Verification

• uv run pytest tests/ --ignore=tests/integration → 2112 passed.
• uv run pytest tests/ws/test_backpressure.py tests/test_auth.py tests/test_types.py tests/test_models.py -v → all green including the 4 new async sign tests + the counter-drift regression test.
• uv run ruff check . → clean.
• uv run mypy kalshi/ → Success: no issues found in 76 source files.

[claude]

Code Review — PR #182: WS/auth polish batch (#173 + #174 + #178)

Three well-scoped, well-documented changes. Each is independently justified and passes the project's full test suite. Overall the PR is in good shape; a few things worth examining before merge.

---

#173 — MessageQueue defense-in-depth

What it does: Adds maxlen=maxsize+1 to the underlying deque so the buffer has a C-level hard ceiling independent of _size.

Assessment: Correct and useful. The +1 is needed because put_sentinel() appends without going through the size-check/overflow path, so the hard ceiling must accommodate maxsize real items + 1 sentinel.

One subtle point in the regression test (test_deque_maxlen_caps_memory_even_if_counter_drifts): The test injects drift by zeroing _size, but put()'s overflow check is len(self._buffer) >= self._maxsize — it uses the actual buffer length, not _size. So the overflow logic still fires correctly throughout the 1000-put loop; it isn't disabled by the counter drift. The maxlen ceiling isn't what's keeping the buffer at 5 here — the overflow check already handles it.

This doesn't invalidate the defense-in-depth value (a future bug that either switches the check to _size, or adds a raw-append bypass, would be caught by maxlen), but the test comment overstates what's being tested: "the only thing protecting memory is the deque's maxlen" isn't accurate for the current code. The test assertion (len(q._buffer) <= maxsize + 1) is still correct; just the comment needs softening.

Suggested comment fix:

# Even with _size zeroed, the overflow check (`len(self._buffer) >= self._maxsize`)
# still fires. The deque maxlen is the defense-in-depth backstop for any future
# code path that bypasses that check (e.g., raw append, check changed to use _size).

---

#174 — _to_decimal_* consolidation

Assessment: Clean and correct. The two functions were byte-identical; merging them into _coerce_decimal is the right call.

Minor: The comment block that replaces the old _to_decimal_fp function body:

# FixedPointCount: count/volume fields (e.g., ``count_fp``, ``volume_fp``).
# Uses the same decimal coercion as :data:`DollarDecimal`; see _coerce_decimal.

…is slightly redundant since FixedPointCount already has a full docstring on the Annotated alias below it. Not a blocker — just noting it could be removed.

No other concerns here.

---

#178 — Async RSA-PSS sign offload

Assessment: Solid design. The dedicated executor, lazy-init with double-checked locking, and lifecycle tie-in to close() are all correct. The benchmark methodology is sound (real loop.time() deltas, not the asyncio.sleep(0) scheduler-yield trick).

Issue 1 — close() + concurrent sign_request_async() race (minor):

_get_sign_executor() reads _sign_executor outside the lock on the fast path:

if self._sign_executor is None:   # ← lock-free check
    with self._sign_executor_lock:
        ...
return self._sign_executor        # ← lock-free return

If close() runs concurrently (shutdown(wait=False) + self._sign_executor = None), a racing sign_request_async() could receive a reference to a shut-down executor. run_in_executor() on a shut-down pool raises RuntimeError: cannot schedule new futures after shutdown. For typical lifecycle usage (close on teardown, not mid-flight), this is fine, but worth adding a brief docstring note to _get_sign_executor or close() that they aren't safe to call concurrently.

Issue 2 — shutdown(wait=False) + potential thread leak if auth reused after close() (low risk):

close() calls shutdown(wait=False) and sets _sign_executor = None. If sign_request_async() is called again afterward (e.g., the same KalshiAuth object is shared), _get_sign_executor() creates a new executor while the old one's threads may still be running a sign. This is documented implicitly ("interpreter-shutdown cleanup"), but it means calling close() on a live client that continues signing leaks the old pool's threads until the sign finishes. Again, low risk for the intended lifecycle, but worth a note.

Issue 3 — test_concurrent_signs_do_not_stall_event_loop threshold:

The 30 ms ceiling is generous intentionally, and the comment says it's a "regression bound, not a precision benchmark." That's the right framing. One potential flake source: on a heavily loaded CI runner, even scheduler jitter for 40 ticks at 10 ms each can exceed 30 ms. Consider whether it's worth marking this test @pytest.mark.slow or adding a skip/xfail on known-slow CI environments. Not a blocker, but something to watch.

Positive note on _build_auth_headers: The transition from sync → async is handled correctly at all call sites (connect(), reconnect(), and the test). The await is not missing anywhere.

Positive note on close() wiring: Both KalshiClient.close() and AsyncKalshiClient.close() call self._auth.close() with a None guard. The sync client doesn't use the executor, but wiring it there anyway means if a user switches transports or if auth is shared, teardown stays clean. Good defensive call.

---

General

• All three changes are independently mergeable and none touch shared state in unexpected ways.
• Ruff and mypy pass; the 2112-test suite is green.
• CHANGELOG is clear and complete.
• The benchmark script is a useful addition for future regression testing.

Verdict: Two minor comment/doc issues and one lifecycle edge case worth documenting. No correctness bugs. Ready to merge after addressing (or explicitly accepting) the points above.