Bound dashboard websocket backpressure
All checks were successful
deploy / deploy (push) Successful in 57s
All checks were successful
deploy / deploy (push) Successful in 57s
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.
This commit is contained in:
parent
2f455f0140
commit
38f58139e1
4 changed files with 58 additions and 1 deletions
|
|
@ -9,12 +9,14 @@ import { createConsumer } from '../bus/kafka/consumer.mjs';
|
||||||
import { parseEventMessage } from '../core/event-envelope.mjs';
|
import { parseEventMessage } from '../core/event-envelope.mjs';
|
||||||
import { buildMakerCompetitivenessSummary } from '../core/maker-competitiveness.mjs';
|
import { buildMakerCompetitivenessSummary } from '../core/maker-competitiveness.mjs';
|
||||||
import {
|
import {
|
||||||
|
DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES,
|
||||||
applyDashboardLiveEvent,
|
applyDashboardLiveEvent,
|
||||||
buildDashboardBootstrap,
|
buildDashboardBootstrap,
|
||||||
buildDashboardControlErrorResponse,
|
buildDashboardControlErrorResponse,
|
||||||
buildLiveQuoteLifecycleRows,
|
buildLiveQuoteLifecycleRows,
|
||||||
buildLiveStatusBar,
|
buildLiveStatusBar,
|
||||||
createDashboardLiveState,
|
createDashboardLiveState,
|
||||||
|
isDashboardWebSocketBackpressured,
|
||||||
listDashboardServices,
|
listDashboardServices,
|
||||||
resolveDashboardControl,
|
resolveDashboardControl,
|
||||||
resolveDashboardControlTimeoutMs,
|
resolveDashboardControlTimeoutMs,
|
||||||
|
|
@ -769,7 +771,25 @@ function broadcast(payload) {
|
||||||
const encoded = JSON.stringify(payload);
|
const encoded = JSON.stringify(payload);
|
||||||
for (const socket of webSockets) {
|
for (const socket of webSockets) {
|
||||||
if (socket.readyState !== 1) continue;
|
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?.();
|
||||||
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -8,6 +8,7 @@ import { inferServiceFreshnessTimestamp as inferRuntimeFreshnessTimestamp } from
|
||||||
|
|
||||||
export const DASHBOARD_LIVE_QUOTE_LIMIT = 10;
|
export const DASHBOARD_LIVE_QUOTE_LIMIT = 10;
|
||||||
export const DASHBOARD_LIVE_LIFECYCLE_LIMIT = 20;
|
export const DASHBOARD_LIVE_LIFECYCLE_LIMIT = 20;
|
||||||
|
export const DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES = 1_000_000;
|
||||||
|
|
||||||
const DECIMAL_SCALE = 18;
|
const DECIMAL_SCALE = 18;
|
||||||
const DECIMAL_FACTOR = 10n ** BigInt(DECIMAL_SCALE);
|
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 } = {}) {
|
export function resolveDashboardControlTimeoutMs({ control, config } = {}) {
|
||||||
const baseTimeoutMs = Number(config?.operatorDashboardUpstreamTimeoutMs || 3_000);
|
const baseTimeoutMs = Number(config?.operatorDashboardUpstreamTimeoutMs || 3_000);
|
||||||
if (control?.service !== 'trade-executor') return baseTimeoutMs;
|
if (control?.service !== 'trade-executor') return baseTimeoutMs;
|
||||||
|
|
|
||||||
|
|
@ -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', () => {
|
test('operator dashboard bootstrap does not synchronously refresh intent request outcomes', () => {
|
||||||
assert.match(source, /loadRecentIntentRequests\(pool, \{[\s\S]*refreshOutcomes: false/);
|
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\?\.\(\)/);
|
||||||
|
});
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@ import test from 'node:test';
|
||||||
import assert from 'node:assert/strict';
|
import assert from 'node:assert/strict';
|
||||||
|
|
||||||
import {
|
import {
|
||||||
|
DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES,
|
||||||
applyDashboardLiveEvent,
|
applyDashboardLiveEvent,
|
||||||
buildDashboardBootstrap,
|
buildDashboardBootstrap,
|
||||||
buildDashboardControlErrorResponse,
|
buildDashboardControlErrorResponse,
|
||||||
|
|
@ -9,6 +10,7 @@ import {
|
||||||
buildProfitabilitySummary,
|
buildProfitabilitySummary,
|
||||||
createDashboardLiveState,
|
createDashboardLiveState,
|
||||||
deriveQuoteLifecycleRows,
|
deriveQuoteLifecycleRows,
|
||||||
|
isDashboardWebSocketBackpressured,
|
||||||
resolveDashboardControl,
|
resolveDashboardControl,
|
||||||
resolveDashboardControlTimeoutMs,
|
resolveDashboardControlTimeoutMs,
|
||||||
} from '../src/core/operator-dashboard.mjs';
|
} 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);
|
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', () => {
|
test('control routing only resolves the allowlisted safe dashboard actions', () => {
|
||||||
const refresh = resolveDashboardControl({
|
const refresh = resolveDashboardControl({
|
||||||
service: 'liquidity-manager',
|
service: 'liquidity-manager',
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue