v4.3.1: coalesce quote/depth forwards so NEW_BAR is not starved (#143)

[TexasCoding]

Fixes #143.

Problem

At CME RTH open, every SignalR quote_update / market_depth message scheduled its own run_coroutine_threadsafe task (_use_batching defaults false). That queue grew without bound and starved data-manager tick processing, so 1-minute NEW_BAR arrived tens of seconds late (or not at all) while a 1s heartbeat still looked healthy. Halt then cancelled a pile of pending _forward_event_async tasks.

Related #137 (get_session_data hang) was not the stall this time — on_bar never ran.

Fix

Latest-wins coalescing per (event_type, contract_id) for quote_update and market_depth:

• One in-flight drain task + one pending slot
• Extra messages replace the pending payload instead of enqueueing another loop task
• Applies even when enable_batching() is on (that path also used to schedule one coroutine per tick)

Never dropped: market_trade (volume / OHLC / NEW_BAR) and user-hub events (order_update, position_update, account_update, trade_execution).

Dropped counts are on suite.realtime.get_stats() as coalesced_quote_dropped and coalesced_depth_dropped.

Quotes are snapshots, so latest-wins is correct. Depth is per price level, so a flood may skip intermediate book updates until later rows or a reset arrive. That matches the live consumer workaround that unstarved NEW_BAR.

Tests

TestCoalescedMarketEventScheduling in tests/realtime/test_event_handling.py:

• Quote/depth floods deliver first + latest, not every intermediate payload
• Pending latest flushes after in-flight completes
• MNQ flood does not overwrite MES
• market_trade and user events are never dropped during a quote flood
• Batching handle_quote path is also bounded
• cleanup() clears coalesce state

Local verification: 3253 passed, 2 skipped (uv run pytest tests/ --ignore=tests/benchmarks/). mypy clean (118 files). Pre-commit ruff/mypy/bandit passed.

Release

PATCH 4.3.1. Changelog, README, docs/guide/realtime.md, User-Agent, and version strings updated.

[TexasCoding]

Follow-up after review: live SignalR calls handler(*[contract_id, data]), so the coalesce key now uses the string first argument. Packed list/dict payloads still key correctly. coalesced_*_dropped only increments when a pending payload is overwritten (the parked latest is still delivered). Drain generations prevent cleanup/cancel from clearing a newer inflight bit.