Coalesce high-frequency socket updates and protect against slow consumers #96

Overview

Every market-core Trade event the indexer stores becomes a full TRADE_EXECUTED broadcast to every connected socket (io.emit in src/server.ts). That is fine at demo volume, but it scales badly in exactly the moments that matter. During a busy period the price of a single market can change many times per second; during indexer catch-up (after a restart, an RPC outage, or a first boot with no cursor) backfillEvents() pages through up to the whole RPC retention window, about 7 days on testnet, and emits every historical trade live, as if it had just happened. Clients render thousands of stale prices in a burst and every socket's outbound buffer grows.

socket.io has no per-socket backpressure: emit appends to the client's write buffer and the server keeps the data in memory until the transport drains it. A single slow mobile client or a stalled tab can therefore make the process accumulate unbounded memory, and with polling transport the buffered packets sit in the server until the next poll.

This issue introduces a throttling layer between the indexer and socket.io: per-market latest-wins MARKET_PRICE ticks emitted at a configurable rate, volatile delivery for non-essential events, suppression of live emits during historical catch-up, and slow-consumer detection that drops or disconnects sockets whose buffers exceed a threshold. The documented TRADE_EXECUTED semantics must stay available for clients that rely on every trade, and the change must be proven with a load test showing bounded memory.

Background & current behavior

• src/server.ts: startEventListener((event, payload) => io.emit(event, payload)); no rooms, no transport options, default maxHttpBufferSize, polling and websocket enabled.
• src/services/indexer.service.ts: backfillEvents(fromLedger?, deps) reads getHealth(), starts at cursor + 1 or oldestLedger, pages getEvents with PAGE_LIMIT = 200 until a short page, and for each newly stored event runs applySideEffects(), which for market-core Trade updates Market.currentYesProb = priceYesBps(reserve_yes, reserve_no) / 10000 and emits TRADE_EXECUTED with { onChainMarketId, marketId, marketQuestion, trader, direction, isBuy, collateralRaw, sharesRaw, feeRaw, priceYesBps, txHash, ledger, eventId }. REASONING_PUBLISHED is emitted for TracePublished. There is no notion of "live" vs "historical": the emitter cannot tell whether an event is seconds or days old.
• startEventListener polls every INDEXER_POLL_MS (default 5000) and reruns backfillEvents until caught up; there is no lag metric.
• docs/api.md "Realtime (socket.io)" documents TRADE_EXECUTED as "Emitted when the indexer stores a market-core trade event" and the frontend uses it for both a trade feed and price updates.
• Wave issue Feature/integration and fixes #4 adds per-market rooms and a 20-event replay on subscribe; this issue is compatible with rooms and must not depend on them.

Goals

• New MARKET_PRICE event: { onChainMarketId, marketId, priceYesBps, ledger, eventId, ts }, latest-wins per market, at most once per SOCKET_PRICE_TICK_MS per market.
• TRADE_EXECUTED still emitted for every live trade, but as volatile by default (configurable), so a socket that cannot take it now skips it instead of buffering.
• No live emits for historical events during catch-up; a single CATCHUP_COMPLETE summary instead.
• Slow-consumer detection with drop and disconnect thresholds.
• A src/lib/socket-throttle.ts module that is unit-testable without socket.io.
• Load test proving bounded memory with slow and stalled clients.

Non-goals

• Cross-instance fan-out (separate event-bus issue); the throttle runs where the emit happens.
• Changing REASONING_PUBLISHED semantics beyond volatile delivery being configurable.
• Price history or candles (separate issue).
• Socket authentication or connection caps (separate issues).

Proposed implementation

1. Liveness classification in src/services/indexer.service.ts: compute live = page.latestLedger - event.ledger <= SOCKET_LIVE_LEDGER_LAG (new env, default 12, about one minute of ledgers) and pass it through the emitter: change IndexerDeps.emit to (event: string, payload: unknown, meta: { live: boolean; ledger: number; marketKey: string | null }) => void. Update tests/indexer.test.ts (it builds emit = (e, p) => emitted.push([e, p])) to accept the extra argument. During a poll that starts more than SOCKET_LIVE_LEDGER_LAG behind health.latestLedger, mark the poll as catch-up; at its end emit one CATCHUP_COMPLETE { fromLedger, toLedger, events: n } so clients refetch REST state.
2. src/lib/socket-throttle.ts (new):

   export interface Sink { emit(event: string, payload: unknown, opts: { volatile: boolean }): void }
   export interface ThrottleOptions { priceTickMs: number; tradeVolatile: boolean; dropHistorical: boolean; now?: () => number; schedule?: (fn: () => void, ms: number) => unknown }
   export class SocketThrottle {
     constructor(sink: Sink, opts: ThrottleOptions);
     handle(event: string, payload: any, meta: { live: boolean; ledger: number; marketKey: string | null }): void;
     flush(): void;          // emit pending latest prices now (shutdown, tests)
     stop(): void;
     stats(): { coalesced: number; droppedHistorical: number; emitted: Record<string, number> };
   }

   Behaviour: for TRADE_EXECUTED, record pending[marketKey] = { priceYesBps, ledger, eventId } if newer (compare ledger, then eventId) and schedule a flush for that market at lastEmit + priceTickMs; emit the trade itself only if meta.live (or dropHistorical is false), with volatile per options. For REASONING_PUBLISHED, emit non-volatile if live. Never reorder price ticks backwards: drop a pending update older than the last emitted ledger for that market.
3. Wiring in src/server.ts:

   const throttle = new SocketThrottle({ emit: (e, p, { volatile }) => (volatile ? io.volatile.emit(e, p) : io.emit(e, p)) }, fromConfig());
   stopEventListener = startEventListener((e, p, meta) => throttle.handle(e, p, meta));

   Call throttle.flush(); throttle.stop() in shutdown before io.close().
4. Slow-consumer detection: new attachSlowConsumerGuard(io, { warnBytes, dropBytes, checkMs }) in src/lib/socket-throttle.ts. Every checkMs iterate io.of('/').sockets and measure buffered bytes: for the websocket transport, socket.conn.transport.socket?.bufferedAmount (the underlying ws socket); for polling, socket.conn.writeBuffer.length packets times an estimated size. Above warnBytes mark the socket degraded (only MARKET_PRICE ticks, which are latest-wins, are sent to it; trades are skipped); above dropBytes emit nothing further and socket.disconnect(true) with a log line containing only socket.id and the byte count (sampled).
   Because global io.emit cannot skip individual sockets, implement degraded handling by having degraded sockets leave a realtime:full room that all sockets join on connection, and emit trades to that room instead of globally (io.to('realtime:full').volatile.emit(...)), while price ticks still go to everyone.
5. Engine.IO options in src/server.ts: set pingInterval/pingTimeout explicitly (defaults documented), perMessageDeflate: false (CPU cost on bursty small messages), and maxHttpBufferSize to a small inbound limit (clients send nothing today). Connection caps and origin hardening are left to the socket hardening issue.
6. Config (src/config/index.ts, .env.example): SOCKET_PRICE_TICK_MS (default 1000), SOCKET_TRADE_VOLATILE (default true), SOCKET_LIVE_LEDGER_LAG (default 12), SOCKET_SLOW_WARN_BYTES (default 262144), SOCKET_SLOW_DROP_BYTES (default 2097152), SOCKET_SLOW_CHECK_MS (default 2000).
7. Docs: update "Realtime (socket.io)" in docs/api.md: MARKET_PRICE payload and rate, TRADE_EXECUTED is best-effort (volatile) and live-only, CATCHUP_COMPLETE, and that clients needing every trade should use the REST/indexed history (or the SSE stream if it exists) rather than the socket.
8. Load test (tests/socket-throttle.test.ts for unit behaviour plus a new script scripts/socket-load.mjs for the soak): spin up the server in-process with a fake event source that emits 50 trades/s across 20 markets for 60 s, connect 200 fast clients and 20 clients that never read (paused underlying socket), and record process.memoryUsage().heapUsed and RSS every second. The soak asserts peak heap growth stays below a configured budget and that stalled clients are disconnected.

Edge cases & pitfalls

• Clients that use TRADE_EXECUTED.priceYesBps to render prices will see fewer updates if they miss volatile trades; that is why MARKET_PRICE must be emitted reliably (non-volatile) and documented as the price source.
• A burst of trades on one market followed by silence must still emit the final price (the scheduled flush must fire even if no further events arrive).
• Timers per market must be cleaned up for markets that stop trading, or the throttle leaks.
• bufferedAmount is not available on every transport; fall back gracefully and never throw from the guard loop.
• Catch-up detection must not suppress events forever if latestLedger from RPC is stale; cap catch-up mode per poll.

Testing plan

tests/socket-throttle.test.ts (fake clock via injected now/schedule, recording sink):

• bursts of trades on one market produce one MARKET_PRICE per tick with the latest price
• final price after a burst is emitted when the tick elapses
• out-of-order older ledger never overwrites a newer pending price
• historical events during catch-up are not emitted live and CATCHUP_COMPLETE is emitted once
• TRADE_EXECUTED is volatile by default and non-volatile when configured
• idle markets release their timers
• slow-consumer guard degrades then disconnects sockets above thresholds (fake socket objects with bufferedAmount)
• integration with socket.io-client: fast client receives MARKET_PRICE while a stalled client is disconnected

Soak: node scripts/socket-load.mjs --duration 60 documented in the test plan; run manually and in a nightly job if available. Commands: npm test, npm run build.

Acceptance criteria

• src/lib/socket-throttle.ts implements coalescing, volatile delivery, catch-up suppression and the slow-consumer guard with injected clocks.
• The indexer passes live, ledger and marketKey metadata to the emitter, and existing indexer tests are updated and pass.
• MARKET_PRICE is emitted at most once per tick per market with the latest price, including the trailing update.
• Historical events are not emitted live during catch-up; CATCHUP_COMPLETE is emitted once per catch-up.
• Sockets above the warn threshold stop receiving trades; above the drop threshold they are disconnected.
• New env vars are validated in src/config/index.ts and documented in .env.example.
• The soak script demonstrates bounded heap growth with stalled clients; results are recorded in the PR.
• docs/api.md documents MARKET_PRICE, CATCHUP_COMPLETE and the changed delivery guarantees of TRADE_EXECUTED.
• npm run build, npm test and CI pass.

Resources

• src/server.ts, src/services/indexer.service.ts, src/generated/fpmm.ts (priceYesBps), tests/indexer.test.ts
• src/config/index.ts, .env.example, docs/api.md, docs/wave-issues/04-socket-market-rooms.md
• socket.io volatile events: https://socket.io/docs/v4/emitting-events/#volatile-events
• socket.io server options (pingInterval, maxHttpBufferSize, perMessageDeflate): https://socket.io/docs/v4/server-options/
• ws bufferedAmount: https://github.com/websockets/ws/blob/master/doc/ws.md#websocketbufferedamount
• Stellar RPC getEvents: https://developers.stellar.org/docs/data/apis/rpc/api-reference/methods/getEvents

Generated by Claude Code