From 38f58139e1faecec252bc31854c9b56507c0d7d9 Mon Sep 17 00:00:00 2001 From: philipp Date: Wed, 10 Jun 2026 23:07:34 +0200 Subject: [PATCH] Bound dashboard websocket backpressure Proof: Operator dashboard now disconnects websocket clients whose send buffers exceed a bounded threshold, with tests covering the backpressure predicate and production broadcast wiring. Assumptions: The dashboard OOM restarts are consistent with buffered websocket updates from slow clients under high quote-update volume; disconnecting lagging dashboard clients is safe because they can reconnect and reload canonical state. Still fake: This does not prove every historical dashboard restart was caused by websocket backpressure; it prevents the identified unbounded buffering path without changing trading behavior. --- src/apps/operator-dashboard.mjs | 22 ++++++++++++++++++++- src/core/operator-dashboard.mjs | 12 +++++++++++ test/operator-dashboard-app-static.test.mjs | 7 +++++++ test/operator-dashboard.test.mjs | 18 +++++++++++++++++ 4 files changed, 58 insertions(+), 1 deletion(-) diff --git a/src/apps/operator-dashboard.mjs b/src/apps/operator-dashboard.mjs index 371ac01..1eab31b 100644 --- a/src/apps/operator-dashboard.mjs +++ b/src/apps/operator-dashboard.mjs @@ -9,12 +9,14 @@ import { createConsumer } from '../bus/kafka/consumer.mjs'; import { parseEventMessage } from '../core/event-envelope.mjs'; import { buildMakerCompetitivenessSummary } from '../core/maker-competitiveness.mjs'; import { + DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES, applyDashboardLiveEvent, buildDashboardBootstrap, buildDashboardControlErrorResponse, buildLiveQuoteLifecycleRows, buildLiveStatusBar, createDashboardLiveState, + isDashboardWebSocketBackpressured, listDashboardServices, resolveDashboardControl, resolveDashboardControlTimeoutMs, @@ -769,7 +771,25 @@ function broadcast(payload) { const encoded = JSON.stringify(payload); for (const socket of webSockets) { if (socket.readyState !== 1) continue; - socket.send(encoded); + if (isDashboardWebSocketBackpressured(socket)) { + logger.warn('dashboard_websocket_backpressure_disconnect', { + details: { + buffered_amount: Number(socket.bufferedAmount || 0), + max_buffered_bytes: DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES, + }, + }); + webSockets.delete(socket); + dashboardRuntimeState.websocket_clients = webSockets.size; + socket.terminate?.(); + continue; + } + socket.send(encoded, (error) => { + if (!error) return; + dashboardRuntimeState.last_live_event_error = serializeError(error); + webSockets.delete(socket); + dashboardRuntimeState.websocket_clients = webSockets.size; + socket.terminate?.(); + }); } } diff --git a/src/core/operator-dashboard.mjs b/src/core/operator-dashboard.mjs index 36dc57f..6d6d0d8 100644 --- a/src/core/operator-dashboard.mjs +++ b/src/core/operator-dashboard.mjs @@ -8,6 +8,7 @@ import { inferServiceFreshnessTimestamp as inferRuntimeFreshnessTimestamp } from export const DASHBOARD_LIVE_QUOTE_LIMIT = 10; export const DASHBOARD_LIVE_LIFECYCLE_LIMIT = 20; +export const DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES = 1_000_000; const DECIMAL_SCALE = 18; const DECIMAL_FACTOR = 10n ** BigInt(DECIMAL_SCALE); @@ -314,6 +315,17 @@ export function buildDashboardControlErrorResponse(error, { control = null } = { }; } +export function isDashboardWebSocketBackpressured(socket, { + maxBufferedBytes = DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES, +} = {}) { + const bufferedAmount = Number(socket?.bufferedAmount || 0); + const maxBytes = Number(maxBufferedBytes); + return Number.isFinite(bufferedAmount) + && Number.isFinite(maxBytes) + && maxBytes >= 0 + && bufferedAmount > maxBytes; +} + export function resolveDashboardControlTimeoutMs({ control, config } = {}) { const baseTimeoutMs = Number(config?.operatorDashboardUpstreamTimeoutMs || 3_000); if (control?.service !== 'trade-executor') return baseTimeoutMs; diff --git a/test/operator-dashboard-app-static.test.mjs b/test/operator-dashboard-app-static.test.mjs index eb79b94..787fe81 100644 --- a/test/operator-dashboard-app-static.test.mjs +++ b/test/operator-dashboard-app-static.test.mjs @@ -40,3 +40,10 @@ test('operator dashboard API auth failures are JSON for frontend fetches', () => test('operator dashboard bootstrap does not synchronously refresh intent request outcomes', () => { assert.match(source, /loadRecentIntentRequests\(pool, \{[\s\S]*refreshOutcomes: false/); }); + +test('operator dashboard disconnects backpressured websocket clients before buffering unbounded updates', () => { + assert.match(source, /isDashboardWebSocketBackpressured/); + assert.match(source, /dashboard_websocket_backpressure_disconnect/); + assert.match(source, /buffered_amount: Number\(socket\.bufferedAmount \|\| 0\)/); + assert.match(source, /socket\.terminate\?\.\(\)/); +}); diff --git a/test/operator-dashboard.test.mjs b/test/operator-dashboard.test.mjs index 2e6c4f0..f9868a8 100644 --- a/test/operator-dashboard.test.mjs +++ b/test/operator-dashboard.test.mjs @@ -2,6 +2,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { + DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES, applyDashboardLiveEvent, buildDashboardBootstrap, buildDashboardControlErrorResponse, @@ -9,6 +10,7 @@ import { buildProfitabilitySummary, createDashboardLiveState, deriveQuoteLifecycleRows, + isDashboardWebSocketBackpressured, resolveDashboardControl, resolveDashboardControlTimeoutMs, } from '../src/core/operator-dashboard.mjs'; @@ -193,6 +195,22 @@ test('dashboard control errors become structured responses instead of uncaught f assert.match(response.payload.reason, /timeout/i); }); +test('dashboard websocket backpressure guard drops lagging clients before OOM', () => { + assert.equal( + isDashboardWebSocketBackpressured({ + bufferedAmount: DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES, + }), + false, + ); + assert.equal( + isDashboardWebSocketBackpressured({ + bufferedAmount: DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES + 1, + }), + true, + ); + assert.equal(isDashboardWebSocketBackpressured({}), false); +}); + test('control routing only resolves the allowlisted safe dashboard actions', () => { const refresh = resolveDashboardControl({ service: 'liquidity-manager',