fix: buffer live trades/candles until the async subscribe snapshot flushes (#97)

[emrebulutlar]

Mechanism

MarketDataWebSocket handles subscribe like this: the channel joins the market broadcast ChannelGroup synchronously (subscribeToMarket), then sendInitialState writes the BOOK snapshot synchronously but fetches recent-trades and candle-history asynchronously (recentTradesJsonAsync / buildCandleHistoryJsonAsync, a DB round trip up to ~1.5s, completing on a read-pool thread).

Because the group join happens first, live TRADES_BATCH / CANDLE_UPDATE frames start flowing immediately. Any that arrive during the async window were delivered before the older "initial" trades snapshot. A client that renders each TRADES_BATCH as the current list briefly showed a live trade, then dropped it when the stale initial snapshot landed on top: the tape appeared to rewind. (The initial recent-trades snapshot is itself a TRADES_BATCH, so it overwrites the newer live one.) It gets likelier under connect/resync storms contending for the 2-thread read pool.

Design

Per-channel buffering of the live frame types whose initial state is async, held until the initial snapshot is flushed (joining after the async completes would instead lose frames; buffering is the lossless option).

• On subscribe, a per-channel InitState (channel attribute) is armed before the group join, so no live frame can slip out ahead of it. sendInitialState re-arms on the refresh/resync path (channelWritabilityChanged) too, which has the same race.
• While initializing, broadcastMarketData routes each frame through deliver(...): a live TRADES_BATCH / CANDLE_UPDATE destined to an initializing channel is buffered in arrival order instead of written. Book frames are never buffered: the book snapshot is synchronous and BOOK_DELTAs are version-chained on top of it (already safe today), so the change stays scoped to the async frame types.
• The two async fetches are composed with allOf(...).whenComplete(...); on settle we schedule a flush on the channel event loop: write the initial snapshot (whichever fetch succeeded), then the buffered live frames in arrival order, then stop buffering.
• Failure/timeout still flushes and clears. A failed fetch writes no snapshot (preserving the pre-fix behavior of writing nothing on failure) but the buffer is drained and the flag cleared, so a channel never wedges permanently buffering.
• The buffer is bounded (MAX_INIT_BUFFER_FRAMES = 512): past the cap the oldest frame is dropped (the stream is conflatable) and a single warning is logged, so a stuck DB read cannot balloon the heap.
• Thread-safety: broadcasts land on the state-manager thread while the flush runs on the read-pool -> event loop. The initializing flag and the drain are held under one lock, and the flush is a single event-loop task, so a broadcast that loses the race sees initializing == false and writes through after the drained frames. No frame is lost or reordered.

The change is scoped to the subscribe/initial-state flow in MarketDataWebSocket.java; GatewayStateManager is untouched.

Tests

New MarketDataWebSocketInitOrderingTest (EmbeddedChannel, driving the real MarketDataHandler with a state manager whose async fetches complete on command):

• liveFramesAreBufferedThenDeliveredAfterInitialSnapshot - trades/candle frames broadcast during the async window are held (not written early) and delivered after the initial snapshot, preserving arrival order.
• asyncFailureStillFlushesAndUnblocksChannel - both fetches fail: no snapshot written, buffered frame still flushed, and a subsequent live frame is delivered directly (channel unblocked).
• otherMarketsAndChannelsAreUnaffected - buffering is per-channel/per-market; market-1 and market-2 live frames never cross channels.
• bufferIsBoundedDroppingOldest - past the cap the oldest frames are dropped and survivors keep arrival order.

mvn -pl match-gateway test: 266 run, 0 failures, 4 skipped (the skipped 4 are the TimescaleDB integration tests, self-skipped via Assume when MARKET_PG_TEST_URL is unset). No live stack required.

Closes #97

🤖 Generated with Claude Code