diff --git a/README.md b/README.md index 8b6d4e06..00dcc378 100644 --- a/README.md +++ b/README.md @@ -146,3 +146,58 @@ clearly named getters and setters for ease of use. See `src/ToolSocketMessage.js f/frameCount: (number of binary buffers that will be sent following this message) } ``` + +## Connection Info API + +Server-side diagnostics for realtime connection quality. Fully dormant until enabled: +the info module is only loaded on first use and adds zero per-message overhead — +measurement rides on TCP byte counters, WebSocket protocol ping/pong frames, and the +existing keepalive, all sampled by a single 1 Hz ticker. + +### Per connection (server side) +```javascript +webSocketServer.on('connection', (socket) => { + socket.info(true, (report) => { + // Every 5 seconds: + // report.data.networkLatency - protocol-level ping RTT (pure network path) + // report.data.appLatency - application-level RTT (network + client pressure) + // report.data.transport - exact wire bytes: average/peak per second, totals + // report.data.networkQuality - score, rating, flow (realtime/buffered/stalled), + // trend, per-dimension sub-scores, issue flags + // report.data.history - issue counts/episodes since info was enabled + // report.data.probe - latest on-demand throughput probe result + }, {name: 'alice'}); // optional connection name, included as report.data.name + + socket.info(true, undefined, {probe: true}); // one-shot up/downstream capacity probe + socket.info(false); // full teardown, back to dormant +}); +``` +When the connection closes, one final report is pushed (`flow: 'ended'`, including +the close code — an abnormal close without a handshake is the typical signature of +proxies or zero-trust gateways cutting the socket). + +### Across all connections (server side) +```javascript +webSocketServer.stagedProbe((result) => { + // result.individual - every client probed alone, sequentially (per-client limits) + // result.stages - 2, 4, 8, ... clients probed simultaneously + // result.networkLimit - all-at-once totals (shared-medium capacity) + // result.sharedBottleneck - ratio of individual sum to network limit; detected: true + // means clients limit each other (e.g. shared WiFi) +}, {sizeBytes: 256 * 1024}); +``` + +### Remote (client side) +```javascript +clientSocket.info(true, (bundle) => { + // Every 5 seconds, pushed by the server: + // bundle.reports - info reports of ALL server connections + // bundle.recentlyClosed - final reports of recently closed connections + // bundle.stagedProbe - latest staged probe result +}); +clientSocket.info(true, undefined, {probe: true}); // trigger a server-wide staged probe +clientSocket.info(false); // stop the stream +``` +The stream also ends automatically when the subscribing connection closes. There is +no built-in authorization: gate this at the application level if info reports should +not be visible to every client. diff --git a/src/IncomingToolSocket.js b/src/IncomingToolSocket.js index f394906e..587a4212 100644 --- a/src/IncomingToolSocket.js +++ b/src/IncomingToolSocket.js @@ -1,4 +1,5 @@ const ToolSocket = require("./ToolSocket"); +const { generateUniqueId } = require("./utilities.js"); class IncomingToolSocket extends ToolSocket { /** @@ -12,9 +13,92 @@ class IncomingToolSocket extends ToolSocket { this.networkId = 'toolbox'; // Or 'io'? this.origin = server.origin; this.server = server; + + /** + * Lazy-initialized info handler (see info()). Stays null while info mode is off, + * in which case the info module is never loaded and no info code runs at all. + * @type {?Object} + */ + this.infoHandler = null; + + /** + * Stable identifier for this connection, included in every info report as + * data.id — lets UIs and the staged probe address a specific connection + * (named or not) for its whole lifetime. + * @type {string} + */ + this.infoId = generateUniqueId(8); + + /** + * Connection name announced by the remote end via the meta route info/name + * (client-side infoName() API). Lives on the socket rather than the handler + * so it survives info enable/disable cycles; stamped into the handler + * whenever info is (re-)enabled. + * @type {?string} + */ + this.announcedInfoName = null; + this.configureSocket(); } + /** + * Enables or disables info updates about this server-side WebSocket connection. + * This API is server side only and intentionally not available on client sockets. + * + * When enabled is false (the default), the entire info subsystem is dormant: + * the info module is not loaded, no listeners are registered, and the send/receive + * hot paths carry zero extra processing overhead. + * + * When enabled is true, the info module is lazily loaded on first use and begins + * delivering info reports to the provided callback every 5 seconds. Calling + * info(true, cb) again simply replaces the callback. Calling info(false) (or + * info()) tears the info subsystem down completely, returning the socket to its + * dormant state. + * + * @param {boolean} [enabled=false] - Whether info updates should be active + * @param {?function} [infoCallback] - Called with info report objects while enabled. + * Omit (undefined) to keep the current callback, + * e.g. when only triggering a probe. + * Report content is defined in ToolSocketInfo.js. + * @param {?Object} [options] - Additional actions: + * @param {string} [options.name] - Assigns a name to this connection (e.g. the + * user name the server identified it with); + * included in every report as data.name + * @param {boolean} [options.probe] - If true, runs a one-shot throughput probe + * (max upstream/downstream measurement). The + * result is included in every report's + * data.probe until the next probe replaces it. + * Intended to be triggered by a UI button. + * @param {number} [options.probeSizeBytes] - Probe payload size per direction + * (default 256 KB, capped at 4 MB) + */ + info(enabled = false, infoCallback, options) { + if (enabled) { + if (!this.infoHandler) { + // Lazy require: this module is only ever loaded once info mode is activated + const ToolSocketInfo = require('./ToolSocketInfo.js'); + this.infoHandler = new ToolSocketInfo(this); + } + if (infoCallback !== undefined) { + this.infoHandler.setCallback(infoCallback); + } + this.infoHandler.start(); + if (options && typeof options.name === 'string') { + this.infoHandler.setName(options.name); + } else if (this.announcedInfoName) { + // name announced by the remote end (infoName()); re-applied on every + // enable, so it survives the auto-enable/disable subscriber cycles + this.infoHandler.setName(this.announcedInfoName); + } + if (options && options.probe) { + this.infoHandler.startProbe(options.probeSizeBytes); + } + } else if (this.infoHandler) { + this.infoHandler.stop(); + this.infoHandler = null; + } + } + /** * Requests the source to create another ToolSocket connection for parallel data transfer. * @return {Promise} - The parallel socket we just created. diff --git a/src/ToolSocket.js b/src/ToolSocket.js index 3ccda268..3cd64d5e 100644 --- a/src/ToolSocket.js +++ b/src/ToolSocket.js @@ -3,7 +3,7 @@ const ToolSocketMessage = require('./ToolSocketMessage.js'); const ToolSocketResponse = require('./ToolSocketResponse.js'); const MessageBundle = require('./MessageBundle.js'); -const { generateUniqueId, addSearchParams, isBrowser, WebSocketWrapper } = require('./utilities.js'); +const { generateUniqueId, addSearchParams, isBrowser, WebSocketWrapper, makeProbePayload } = require('./utilities.js'); const { VALID_METHODS, MAX_MESSAGE_SIZE } = require('./constants.js'); const { URL_SCHEMA, MESSAGE_BUNDLE_SCHEMA } = require('./schemas.js'); @@ -27,6 +27,12 @@ class ToolSocket { this.eventCallbacks = {}; // For events this.responseCallbacks = {}; // For handling direct responses to sent messages + + // Client-side remote info subscription state (see info()) + /** @type {?function} */ + this.remoteInfoCallback = null; + this.remoteInfoSubscribed = false; + this.remoteInfoReattachArmed = false; /** @type {?BinaryBuffer} */ this.binaryBuffer = null; @@ -128,6 +134,22 @@ class ToolSocket { this.eventCallbacks[eventType].forEach(callback => callback(...args)); } + /** + * Removes a previously added event listener + * @param {string} eventType - The event type the listener was added for + * @param {function} callback - The exact callback that was passed to addEventListener + */ + removeEventListener(eventType, callback) { + if (!this.eventCallbacks[eventType]) { + return; + } + this.eventCallbacks[eventType] = this.eventCallbacks[eventType].filter(cb => cb !== callback); + if (this.eventCallbacks[eventType].length === 0) { + // Restore the "no listeners" fast path in triggerEvent + delete this.eventCallbacks[eventType]; + } + } + /** * Clears all event listeners */ @@ -158,11 +180,61 @@ class ToolSocket { } }); - this.addEventListener('meta', (route, body, _response, _binaryData, _messageBundle) => { + this.addEventListener('meta', (route, body, response, _binaryData, _messageBundle) => { if (route === 'requestParallel') { this.triggerEvent('requestParallel', body); // body = id } else if (route === 'confirmParallel') { this.triggerEvent('confirmParallel', body); // body = id + } else if (route === 'probe/down') { + // Throughput probe (see ToolSocketInfo.js): a large payload just + // arrived; a tiny acknowledgement lets the sender compute the + // downstream rate. Only runs when a probe is explicitly requested. + if (response) { + response.send('ok'); + } + } else if (route === 'probe/up') { + // Throughput probe: the sender asks for `body` bytes of + // incompressible data to measure the upstream rate + if (response) { + response.send('ok', makeProbePayload(body)); + } + } else if (route === 'info/report') { + // A server-info bundle pushed by the other side for a subscription + // created via the client-side info(true, callback) API + if (this.remoteInfoCallback) { + this.remoteInfoCallback(body); + } + } else if (route === 'info/subscribe') { + // Only meaningful on server-side sockets (this.server is set there) + if (this.server && this.server.subscribeServerInfo) { + this.server.subscribeServerInfo(this); + } + } else if (route === 'info/unsubscribe') { + if (this.server && this.server.unsubscribeServerInfo) { + this.server.unsubscribeServerInfo(this); + } + } else if (route === 'info/probe') { + // Client-requested staged throughput probe across all connections; + // the result is sent back as the response and also appears in the + // stagedProbe field of subsequent info/report bundles + if (this.server && this.server.stagedProbe) { + this.server.stagedProbe((result) => { + if (response) { + response.send(result); + } + }, body || {}); + } + } else if (route === 'info/name') { + // The remote end names its own connection (e.g. the avatar or user + // it represents), sent via the client-side infoName() API. Stored on + // the socket — not the info handler — so it survives the info + // enable/disable cycles that come with subscribers joining/leaving. + const name = (body && typeof body.name === 'string' && body.name.length > 0) + ? body.name.slice(0, 256) : null; + this.announcedInfoName = name; + if (this.infoHandler && this.infoHandler.setName) { + this.infoHandler.setName(name); + } } else { console.warn(`Received unknown meta route: "${route}"`); } @@ -626,6 +698,105 @@ class ToolSocket { this.sendMethod('meta', route, body, callback, binaryData); } + /** + * Client-side info API: asks the connected server to stream its info reports for + * ALL of its connections to this client via the given callback, every 5 seconds, + * until info(false) is called or this connection closes. Rides on ToolSocket's + * meta transport (routes info/subscribe, info/unsubscribe, info/report, + * info/probe, info/name) — the server only responds if it supports the info API. The + * subscription automatically re-arms after a reconnect. Note: there is no + * built-in authorization; gate access at the application level if needed. + * (On server-side IncomingToolSockets this method is overridden by the local + * per-connection info API.) + * @param {boolean} [enabled=false] - Start (true) or stop (false) the stream + * @param {?function} [infoCallback] - Receives {type: 'serverInfo', timestamp, + * connections, reports: [per-connection info report objects], recentlyClosed: + * [final reports of recently closed connections], stagedProbe}. Omit + * (undefined) to keep the current callback. + * @param {?Object} [options] + * @param {boolean} [options.probe] - Ask the server to run a staged throughput + * probe across its connections (results appear in stagedProbe and in each + * probed connection's data.probe) + * @param {number} [options.probeSizeBytes] - Payload per direction per probe + * @param {string[]} [options.probeNames] - Probe only the connections carrying + * one of these names (assigned via infoName()); omit to probe all + * @param {string[]} [options.probeIds] - Probe only the connections with one of + * these ids (data.id in the server's info reports); addresses any + * connection, named or not + * @param {boolean} [options.probeRamp] - Pass false to skip the growing 2, 4, + * 8... intermediate stages: the probe then measures each connection alone + * and all of them at once, nothing in between + */ + info(enabled = false, infoCallback, options) { + if (enabled) { + if (infoCallback !== undefined) { + this.remoteInfoCallback = infoCallback || null; + } + if (!this.remoteInfoSubscribed) { + this.remoteInfoSubscribed = true; + this.meta('info/subscribe', null); + if (!this.remoteInfoReattachArmed) { + // Re-subscribe automatically when the connection re-opens + this.remoteInfoReattachArmed = true; + this.addEventListener('open', () => { + if (this.remoteInfoSubscribed) { + this.meta('info/subscribe', null); + } + }); + } + } + if (options && options.probe) { + const probeBody = {}; + if (options.probeSizeBytes) probeBody.sizeBytes = options.probeSizeBytes; + if (Array.isArray(options.probeNames) && options.probeNames.length > 0) { + probeBody.names = options.probeNames.slice(0, 64); + } + if (Array.isArray(options.probeIds) && options.probeIds.length > 0) { + probeBody.ids = options.probeIds.slice(0, 128); + } + if (options.probeRamp === false) { + probeBody.ramp = false; + } + this.meta('info/probe', Object.keys(probeBody).length ? probeBody : null); + } + } else if (this.remoteInfoSubscribed) { + this.remoteInfoSubscribed = false; + this.remoteInfoCallback = null; + this.meta('info/unsubscribe', null); + } + } + + /** + * Client-side info API: names this connection on the connected server (e.g. the + * avatar or user id this client represents). The server keeps the name on the + * connection and stamps it into every info report as data.name whenever info is + * active — independent of whether this client ever subscribes. One tiny meta + * message per call; automatically re-sent after a reconnect. Call it once when + * the client knows who it is. + * @param {?string} name - up to 256 chars; null or '' clears the name + */ + infoName(name) { + this.remoteInfoName = (typeof name === 'string' && name.length > 0) + ? name.slice(0, 256) : null; + this.meta('info/name', {name: this.remoteInfoName}); + if (!this.remoteInfoNameReattachArmed) { + // Re-introduce ourselves when the connection re-opens + this.remoteInfoNameReattachArmed = true; + this.addEventListener('open', () => { + if (this.remoteInfoName) { + this.meta('info/name', {name: this.remoteInfoName}); + } + }); + } + // parallel sockets belong to this connection: keep their names in sync so + // diagnostics group them under this name even when naming happens late + if (this.parallelSockets) { + for (const parallel of this.parallelSockets) { + parallel.infoName(this.remoteInfoName ? this.remoteInfoName + ' · data' : null); + } + } + } + /** * Adds aliases for backwards compatibility */ @@ -646,7 +817,16 @@ class ToolSocket { * @returns {ToolSocket} - A new ToolSocket created to the same endpoint as the original. */ static makeParallelSocket(toolsocket) { - return new ToolSocket(toolsocket.url, toolsocket.networkId, 'parallel'); + const parallel = new ToolSocket(toolsocket.url, toolsocket.networkId, 'parallel'); + // a parallel socket belongs to its source connection: track it and inherit + // the announced name (suffixed) so diagnostics group it under its parent — + // infoName() keeps the children in sync if the parent is named later + if (!toolsocket.parallelSockets) toolsocket.parallelSockets = []; + toolsocket.parallelSockets.push(parallel); + if (toolsocket.remoteInfoName) { + parallel.infoName(toolsocket.remoteInfoName + ' · data'); + } + return parallel; } } diff --git a/src/ToolSocketInfo.js b/src/ToolSocketInfo.js new file mode 100644 index 00000000..43813525 --- /dev/null +++ b/src/ToolSocketInfo.js @@ -0,0 +1,990 @@ +/** + * ToolSocketInfo — live info reporting for a server-side ToolSocket connection. + * + * This module is ONLY loaded when IncomingToolSocket.info(true, callback) is called + * for the first time — the info API is server side only; client sockets do not expose + * it. While info mode is off, none of this code is loaded or executed and the + * ToolSocket hot paths carry zero extra work: observation happens exclusively through + * ToolSocket's event system, whose triggerEvent() early-returns with no listeners. + * + * Reporting model: data is collected continuously while enabled. A single 1 Hz ticker + * rolls per-second buckets, and every 5th tick a report is pushed to the callback: + * { + * type: 'info' + * timestamp: number — Date.now() at the moment the report is pushed + * data: { + * name: ?string — the connection's name, if one was assigned via + * info(true, cb, {name: '...'}), e.g. a user name + * networkLatency: { — round trips of WebSocket protocol-level ping/pong frames + * currentMs: number — round trip time of the most recent ping/pong + * averageMs: number — average over samples since the last report + * minMs: number — fastest sample since the last report + * maxMs: number — slowest sample since the last report + * samples: number — how many pings were measured since the last report + * } | null — null if no ping completed since the last report + * appLatency: { — round trips of application-level ToolSocket ping messages + * (same fields as networkLatency) + * } | null + * transport: { + * averageBytesPerSecond: number — total traffic / elapsed time since last report + * peakBytesPerSecond: number — busiest single 1s bucket since the last report + * sentBytes: number — outgoing bytes since the last report + * receivedBytes: number — incoming bytes since the last report + * } + * probe: { — latest throughput probe result; null until the first + * probe, then persists unchanged until the next one. + * Triggered on demand via info(true, undefined, {probe: true}) + * status: 'running' | 'ok' | 'failed' + * timestamp: number — when the probe finished (or started, while running) + * sizeBytes: number — payload size used per direction + * rttMsAtProbe: number — network RTT baseline subtracted from timings + * downstreamBytesPerSecond: ?number — measured server -> client rate + * upstreamBytesPerSecond: ?number — measured client -> server rate + * concurrentSentBytes: number — approx. app traffic sent during the probe + * concurrentReceivedBytes: number — approx. app traffic received during it + * contended: boolean — true if significant app traffic shared the connection, + * meaning the rates are what the probe could grab while + * competing; probe rate + concurrent rate approximates + * the total capacity of the path + * latencyUnderLoadMs: ?number — worst network RTT observed while the probe + * was transferring (rising far above idle RTT = the + * path buffers under load, i.e. bufferbloat) + * reason: string — only when failed, e.g. 'not-connected', + * 'timeout-or-unsupported-client' + * } + * history: { — issue accumulation over the whole info-active period + * (never reset per window; ends up in the final report) + * since: number — when info was enabled (tracking period start) + * durationSeconds: number — how long info has been active + * reports: number — how many report windows were observed + * issues: { — per issue id seen at least once: + * count: number — report windows in which the issue appeared + * episodes: number — distinct occurrences (consecutive windows with the + * same issue count as ONE episode), i.e. how OFTEN + * firstAt: number — timestamp of the first appearance + * lastAt: number — timestamp of the most recent appearance + * } + * } + * networkQuality: { — plain-language interpretation of realtime connection quality + * score: number — 0 (unusable) to 100 (perfect realtime behavior) + * rating: string — 'excellent' | 'good' | 'degraded' | 'poor' + * flow: string — 'realtime' (data moves live) | 'buffered' (held by the + * network and pumped in bursts) | 'stalled' (nothing + * arriving) | 'ended' (connection closed; final report) + * trend: string — 'improving' | 'stable' | 'degrading' vs recent reports + * scores: { — per-dimension indicators, each 0-100 (null = no data): + * continuity: is data arriving every second (the realtime dimension) + * latency: is the round trip fast enough for realtime use + * stability: is the round trip consistent (jitter) + * delivery: is outbound data draining (send backpressure) + * } + * issues: string[] — flags, present only when detected: + * 'data-arriving-in-bursts-not-realtime', + * 'incoming-data-stalled', 'outgoing-data-queuing-locally', + * 'high-latency', 'latency-unstable', + * 'client-under-pressure', 'connection-cut-abnormally' + * details: { — the underlying low-level numbers, for experts + * jitterMs, silentSeconds, longestSilenceSeconds, maxBufferedBytes, + * connectionAgeSeconds, and on the final report closeCode + endedCleanly + * } + * } + * } + * } + * + * How network quality is derived: because a protocol ping goes out every second, a + * healthy realtime link ALWAYS has inbound bytes every second — so seconds with zero + * inbound bytes ('silent seconds') mean the path is not live: a few of them with data + * still arriving overall is the store-and-forward pattern of buffering middleboxes + * (flow 'buffered'); a streak of them is a stall. Outgoing pressure is read from the + * socket's bufferedAmount (data queued locally because the path isn't draining). + * Jitter is the spread (max - min) of the window's protocol ping round trips. On + * close, the close code is captured and a final report is pushed: codes 1000/1001 + * (and 1005, a close frame without a status code) are a clean end; anything else, + * especially 1006 (no close frame at all), means the connection was cut - the + * typical signature of proxies and zero-trust gateways killing the socket. + * + * Throughput probe: an on-demand (never automatic) measurement of the maximum data + * rate each direction sustains. Downstream: a payload of incompressible bytes is sent + * to the client via meta route 'probe/down'; the client acknowledges with a tiny + * response, so the elapsed time minus the RTT baseline is the payload's transfer + * time. Upstream: meta route 'probe/up' asks the client to respond with the same + * amount of incompressible data. Both responders are built into ToolSocket's default + * meta routes and are completely passive until a probe request arrives. + * + * The probe is IN-BAND: it shares the one TCP stream with live application traffic, + * competing with it (and briefly delaying it — a probe frame head-of-line-blocks + * messages queued behind it). This is deliberate: it measures the capacity available + * to THIS connection through THIS network path. To keep the numbers honest, the app + * traffic that flowed during the probe is reported alongside (concurrent* fields) and + * 'contended' flags results that competed with significant traffic. Report windows + * overlapping a probe are marked (details.probeTrafficInWindow), their delivery + * dimension is withheld (the probe itself causes send-buffer pressure), and their + * score is excluded from the trend history, so a probe never triggers false alarms + * about the network. Latency and stability remain as measured — RTT under probe load + * is genuine information about how the path behaves when saturated. + * + * Scoring: each dimension maps to 0-100 through the piecewise-linear anchor tables + * below (continuity is simply the fraction of seconds with inbound data). The overall + * score is 50% the worst dimension + 50% a weighted average with continuity weighted + * highest — realtime systems fail on their weakest dimension, so one bad dimension + * must drag the overall down. An abnormal connection cut caps the final report's + * score at 30. 'trend' compares the score against the previous three reports. + * + * How latency is measured — two complementary signals: + * + * networkLatency: on each 1 Hz tick a WebSocket protocol-level PING control frame + * (RFC 6455) is sent via the ws library with a timestamp embedded in its payload; the + * peer's networking layer echoes it back in the PONG frame, and the timestamp is read + * straight out of it (stateless, no pending map). Control frames are ~10 bytes and are + * answered below the application — no JSON parsing, no routing, no user code — so this + * approximates pure network round trip time. In browsers the reply comes from the + * network stack, largely independent of main-thread load; note that Node.js peers + * answer PINGs on their event loop, so a fully blocked Node peer delays these too. + * + * appLatency: ToolSocket's built-in keepalive sends an application-level 'ping' + * message (route 'action/ping') every ~5s with a response callback, which assigns it + * a message id. We listen to the 'send' event to record the send time of each ping id + * and to the 'res' event to match the response id back to it. This round trip includes + * the peer's full message pipeline (event loop, JSON parse, schema validation, route + * dispatch), so appLatency minus networkLatency approximates client-side processing + * pressure. + * + * How transport is measured (zero per-message overhead): Node.js already maintains + * byte counters on every TCP socket (net.Socket bytesRead / bytesWritten) — they are + * counted by Node core whether or not anyone reads them. The 1 Hz ticker samples the + * counters of the WebSocket's underlying socket and diffs them against the previous + * sample. No message events are hooked at all, so the send/receive paths carry zero + * added work even while info mode is ENABLED. The numbers are exact wire bytes, + * including WebSocket frame headers, client-side masking, and protocol-level control + * frames; keepalive ping/pong traffic is included. If the underlying socket is ever + * replaced, the sampler re-baselines automatically and that second reads as zero. + */ + +const { makeProbePayload } = require('./utilities.js'); + +const BUCKET_INTERVAL_MS = 1000; +const BUCKETS_PER_REPORT = 5; // push a report to the callback every 5 seconds +// Drop pending pings that never got a response (e.g. connection dropped mid-flight) +const PENDING_PING_TIMEOUT_MS = 30000; +// Marks our protocol-level PING payloads so we only interpret our own PONGs +const PROTOCOL_PING_PREFIX = 'tsinfo:'; +// Throughput probe defaults: payload per direction and overall timeout +const DEFAULT_PROBE_SIZE_BYTES = 256 * 1024; +const PROBE_TIMEOUT_MS = 10000; + +// Piecewise-linear anchor tables: [measurement, score] pairs mapping a raw value to +// a 0-100 dimension score. Values between anchors are linearly interpolated. +// Average round trip in ms -> latency score +const LATENCY_SCORE_ANCHORS = [[0, 100], [50, 95], [150, 80], [300, 55], [600, 30], [1200, 10], [2000, 0]]; +// Round trip spread (max - min) in ms -> stability score +const JITTER_SCORE_ANCHORS = [[0, 100], [10, 95], [30, 85], [75, 65], [150, 40], [400, 15], [1000, 0]]; +// Peak bytes stuck in the local send buffer -> delivery score (before persistence penalty) +const BUFFERED_SCORE_ANCHORS = [[0, 100], [16384, 85], [131072, 60], [1048576, 30], [8388608, 0]]; +// How many recent scores 'trend' compares against, and the change it must exceed +const TREND_HISTORY_LENGTH = 3; +const TREND_THRESHOLD = 8; + +/** + * Maps a measurement to a 0-100 score by linear interpolation over an anchor table + * @param {number[][]} anchors - [value, score] pairs, ascending by value + * @param {number} value + * @returns {number} + */ +function interpolateScore(anchors, value) { + if (value <= anchors[0][0]) { + return anchors[0][1]; + } + for (let i = 1; i < anchors.length; i++) { + if (value <= anchors[i][0]) { + const [x0, y0] = anchors[i - 1]; + const [x1, y1] = anchors[i]; + return Math.round(y0 + (y1 - y0) * ((value - x0) / (x1 - x0))); + } + } + return anchors[anchors.length - 1][1]; +} + +const now = (typeof performance !== 'undefined' && performance.now) + ? () => performance.now() + : () => Date.now(); + +class ToolSocketInfo { + /** + * @param {Object} toolsocket - The server-side ToolSocket instance to observe + */ + constructor(toolsocket) { + this.toolsocket = toolsocket; + /** @type {?function} */ + this.callback = null; + /** @type {?Object} the most recent report, kept for server-side collection */ + this.latestReport = null; + this.active = false; + /** + * eventType -> handler map of every listener this instance registered, + * so stop() can remove exactly what start() added and nothing else. + * @type {Object} + */ + this.listeners = {}; + /** @type {?ReturnType} */ + this.tickInterval = null; + this.tickCount = 0; + this.windowStartMs = 0; + + // --- app latency state (ToolSocket-level ping messages) --- + /** @type {Object} ping message id -> send time */ + this.pendingAppPings = {}; + /** @type {number[]} completed round trip times since the last report */ + this.appLatencySamples = []; + /** @type {?number} most recent completed round trip time */ + this.lastAppLatencyMs = null; + + // --- network latency state (WebSocket protocol-level PING/PONG frames) --- + /** @type {?Object} the ws WebSocket our 'pong' listener is attached to */ + this.pingedSocket = null; + /** @type {?function} */ + this.pongHandler = null; + /** @type {number[]} completed round trip times since the last report */ + this.networkLatencySamples = []; + /** @type {?number} most recent completed round trip time */ + this.lastNetworkLatencyMs = null; + + // --- transport state --- + // Baseline sample of the underlying net.Socket's built-in byte counters + /** @type {?Object} the net.Socket the baseline belongs to */ + this.countedSocket = null; + this.lastBytesRead = 0; + this.lastBytesWritten = 0; + // Window accumulators: rolled up from counter deltas on each tick + this.windowBytesSent = 0; + this.windowBytesReceived = 0; + this.peakBytesPerSecond = 0; + + // --- network quality state --- + this.connectionStartMs = 0; + // Seconds in this window with zero inbound bytes (not live if > 0) + this.silentSeconds = 0; + // Running streak of consecutive silent seconds (spans window boundaries) + this.currentSilenceStreak = 0; + this.longestSilenceSeconds = 0; + // Outgoing backpressure observed this window + this.maxBufferedBytes = 0; + this.bufferedTicks = 0; + /** @type {?{closeCode: ?number, endedCleanly: boolean}} set once on close */ + this.closeInfo = null; + /** @type {number[]} recent overall scores, for the trend indicator */ + this.scoreHistory = []; + + // --- connection identity --- + /** @type {?string} name assigned by the server, e.g. a user name */ + this.connectionName = null; + + // --- issue history: accumulates for the whole info-active period --- + this.historyStart = 0; + this.reportCount = 0; + /** @type {Object} */ + this.issueHistory = {}; + /** @type {Set} issues present in the previous report window */ + this.previousIssues = new Set(); + + // --- throughput probe state --- + /** @type {?Object} latest probe result; persists until the next probe */ + this.probeResult = null; + this.probeRunning = false; + // True if a probe transferred during the current report window + this.probeActiveInWindow = false; + // TCP counter snapshot at probe start, for measuring concurrent app traffic + /** @type {?{read: number, written: number}} */ + this.probeCounterStart = null; + } + + /** + * Sets (or replaces) the callback that receives info reports + * @param {?function} callback + */ + setCallback(callback) { + this.callback = typeof callback === 'function' ? callback : null; + } + + /** + * Assigns (or replaces) this connection's name, included in every report + * @param {?string} name + */ + setName(name) { + this.connectionName = (typeof name === 'string' && name.length > 0) ? name : null; + } + + /** + * Attaches listeners and starts the ticker. Idempotent. + */ + start() { + if (this.active) { + return; + } + this.active = true; + + // --- app latency collection --------------------------------------- + // Record the send time of every outgoing ping that expects a response + this._listen('send', (messageBundle) => { + const message = messageBundle && messageBundle.message; + if (message && message.method === 'ping' && message.id) { + this.pendingAppPings[message.id] = now(); + } + }); + // Match incoming responses back to their ping by message id + this._listen('res', (_route, _body, _response, _binaryData, messageBundle) => { + const message = messageBundle && messageBundle.message; + if (!message || !message.id) { + return; + } + const sentAt = this.pendingAppPings[message.id]; + if (sentAt === undefined) { + return; + } + delete this.pendingAppPings[message.id]; + const roundTripMs = Math.round((now() - sentAt) * 10) / 10; + this.lastAppLatencyMs = roundTripMs; + this.appLatencySamples.push(roundTripMs); + }); + + // --- network latency collection ------------------------------------ + this._ensureProtocolPingHooks(); + + // --- connection end capture ----------------------------------------- + // A clean close is code 1000/1001; anything else (especially 1006, closed + // without a close frame) means the connection was cut, e.g. by a proxy + this._listen('close', (event) => { + if (this.closeInfo) { + return; + } + const code = (event && typeof event.code === 'number') ? event.code : null; + this.closeInfo = { + closeCode: code, + // 1000 normal, 1001 going away, 1005 close frame without a status + // code (e.g. a plain client.close()) - all orderly close handshakes + endedCleanly: code === 1000 || code === 1001 || code === 1005, + }; + this._report(); // push a final report for this connection immediately + clearInterval(this.tickInterval); + this.tickInterval = null; + }); + + this.connectionStartMs = Date.now(); + this.historyStart = Date.now(); + + // --- transport collection: baseline the TCP counters --------------- + this._sampleSocketCounters(); // establishes the baseline, returns zero deltas + + // --- ticker: buckets at 1 Hz, report every 5th tick --------------- + this.tickCount = 0; + this.windowStartMs = Date.now(); + this.tickInterval = setInterval(() => this._tick(), BUCKET_INTERVAL_MS); + // Don't let the ticker keep a Node.js process alive on its own + if (this.tickInterval.unref) { + this.tickInterval.unref(); + } + } + + /** + * Detaches every listener, stops the ticker, and clears all collected state. + * After stop(), this instance holds no hooks into the ToolSocket. + */ + stop() { + if (!this.active) { + return; + } + clearInterval(this.tickInterval); + this.tickInterval = null; + for (const [eventType, handler] of Object.entries(this.listeners)) { + this.toolsocket.removeEventListener(eventType, handler); + } + this.listeners = {}; + this._detachProtocolPingHooks(); + this.pendingAppPings = {}; + this.appLatencySamples = []; + this.lastAppLatencyMs = null; + this.networkLatencySamples = []; + this.lastNetworkLatencyMs = null; + this.connectionStartMs = 0; + this.silentSeconds = 0; + this.currentSilenceStreak = 0; + this.longestSilenceSeconds = 0; + this.maxBufferedBytes = 0; + this.bufferedTicks = 0; + this.closeInfo = null; + this.countedSocket = null; + this.lastBytesRead = 0; + this.lastBytesWritten = 0; + this.windowBytesSent = 0; + this.windowBytesReceived = 0; + this.peakBytesPerSecond = 0; + this.callback = null; + this.active = false; + } + + /** + * Runs a one-shot throughput probe measuring the maximum sustained data rate in + * both directions. The result is stored in every report's data.probe until the + * next probe replaces it. No-op while a probe is already running. + * @param {number} [sizeBytes] - Payload size per direction (default 256 KB) + * @param {?function} [onDone] - Called with the finished result + * @returns {boolean} - Whether the probe was started + */ + startProbe(sizeBytes, onDone) { + if (!this.active || this.probeRunning) { + return false; + } + const size = (typeof sizeBytes === 'number' && sizeBytes > 0) + ? Math.floor(sizeBytes) : DEFAULT_PROBE_SIZE_BYTES; + const rttMs = this.lastNetworkLatencyMs || 0; + this.probeRunning = true; + this.probeActiveInWindow = true; + // The 'running' placeholder is live in reports; the pong handler also writes + // latencyUnderLoadMs into it while the transfer is in flight + this.probeResult = { + status: 'running', + timestamp: Date.now(), + sizeBytes: size, + rttMsAtProbe: rttMs, + downstreamBytesPerSecond: null, + upstreamBytesPerSecond: null, + concurrentSentBytes: 0, + concurrentReceivedBytes: 0, + contended: false, + latencyUnderLoadMs: null, + }; + runProbe(this.toolsocket, size, rttMs).then((result) => { + result.latencyUnderLoadMs = this.probeResult + ? this.probeResult.latencyUnderLoadMs : null; + this.probeRunning = false; + if (this.active) { + this.probeResult = result; + } + if (typeof onDone === 'function') { + onDone(result); + } + }); + return true; + } + + /** + * 1 Hz: rolls the current second's counters into the window and tracks the peak. + * Every BUCKETS_PER_REPORT ticks, pushes a report. + */ + _tick() { + const {deltaRead, deltaWritten, rebaselined} = this._sampleSocketCounters(); + const secondTotal = deltaRead + deltaWritten; + if (secondTotal > this.peakBytesPerSecond) { + this.peakBytesPerSecond = secondTotal; + } + this.windowBytesSent += deltaWritten; + this.windowBytesReceived += deltaRead; + + // Flow probing: our own 1 Hz protocol ping guarantees inbound bytes every + // second on a healthy realtime link, so a silent second means "not live" + if (!rebaselined && !this.closeInfo) { + if (deltaRead === 0) { + this.silentSeconds++; + this.currentSilenceStreak++; + if (this.currentSilenceStreak > this.longestSilenceSeconds) { + this.longestSilenceSeconds = this.currentSilenceStreak; + } + } else { + this.currentSilenceStreak = 0; + } + } + + // Outgoing backpressure: bytes stuck in the local send buffer because the + // network path is not draining them. NB-enhanced sockets deliberately keep + // the ws buffer below their low-water mark and hold real send pressure in + // the NB scheduler queue instead, so include that queue when present. + const websocket = this.toolsocket.socket; + let buffered = (websocket && typeof websocket.bufferedAmount === 'number') + ? websocket.bufferedAmount : 0; + if (typeof this.toolsocket.getBackpressure === 'function') { + const nbStats = this.toolsocket.getBackpressure(); + if (nbStats && typeof nbStats.queuedBytes === 'number') { + buffered += nbStats.queuedBytes; + } + } + if (buffered > 0) { + this.bufferedTicks++; + if (buffered > this.maxBufferedBytes) { + this.maxBufferedBytes = buffered; + } + } + + if (this.probeRunning) { + this.probeActiveInWindow = true; + } + + // Network latency: one protocol-level ping per tick (re-hooking if the + // underlying socket was replaced) + this._ensureProtocolPingHooks(); + this._sendProtocolPing(); + + this.tickCount++; + if (this.tickCount >= BUCKETS_PER_REPORT) { + this._report(); + } + } + + /** + * Assembles and pushes one info report, then resets the collection window + */ + _report() { + this._prunePendingPings(); + + const networkLatency = this._summarizeLatency( + this.networkLatencySamples, this.lastNetworkLatencyMs); + const appLatency = this._summarizeLatency( + this.appLatencySamples, this.lastAppLatencyMs); + + // Use real elapsed time, not the nominal window length: under heavy event + // loop load, timers fire late and the nominal value would overstate rates + const elapsedSeconds = Math.max((Date.now() - this.windowStartMs) / 1000, 0.001); + const transport = { + averageBytesPerSecond: Math.round( + (this.windowBytesSent + this.windowBytesReceived) / elapsedSeconds), + peakBytesPerSecond: this.peakBytesPerSecond, + sentBytes: this.windowBytesSent, + receivedBytes: this.windowBytesReceived, + }; + + const networkQuality = this._deriveNetworkQuality(networkLatency, appLatency); + + // Accumulate the issue history: count = windows with the issue, episodes = + // distinct occurrences (issue absent in the previous window = new episode) + this.reportCount++; + const reportTime = Date.now(); + const currentIssues = new Set(networkQuality.issues); + for (const id of currentIssues) { + let entry = this.issueHistory[id]; + if (!entry) { + entry = {count: 0, episodes: 0, firstAt: reportTime, lastAt: reportTime}; + this.issueHistory[id] = entry; + } + entry.count++; + if (!this.previousIssues.has(id)) { + entry.episodes++; + } + entry.lastAt = reportTime; + } + this.previousIssues = currentIssues; + + const history = { + since: this.historyStart, + durationSeconds: Math.round((reportTime - this.historyStart) / 1000), + reports: this.reportCount, + // copied so report consumers cannot mutate the accumulator + issues: Object.fromEntries(Object.entries(this.issueHistory) + .map(([id, entry]) => [id, {...entry}])), + }; + + // Reset the collection window + this.networkLatencySamples = []; + this.appLatencySamples = []; + this.windowBytesSent = 0; + this.windowBytesReceived = 0; + this.peakBytesPerSecond = 0; + this.silentSeconds = 0; + this.longestSilenceSeconds = 0; + this.maxBufferedBytes = 0; + this.bufferedTicks = 0; + this.probeActiveInWindow = false; + this.tickCount = 0; + this.windowStartMs = Date.now(); + + const report = { + type: 'info', + timestamp: Date.now(), + data: { + id: this.toolsocket.infoId || null, + name: this.connectionName, + networkLatency: networkLatency, + appLatency: appLatency, + transport: transport, + networkQuality: networkQuality, + probe: this.probeResult ? {...this.probeResult} : null, + history: history, + }, + }; + this.latestReport = report; + if (this.callback) { + this.callback(report); + } + } + + /** + * Samples the underlying net.Socket's built-in TCP byte counters and returns + * the change since the previous sample. Re-baselines (returning zero deltas) + * on the first call and whenever the underlying socket has been replaced. + * @returns {{deltaRead: number, deltaWritten: number}} + */ + _sampleSocketCounters() { + const websocket = this.toolsocket.socket; + const raw = websocket && websocket._socket; + if (!raw || typeof raw.bytesRead !== 'number') { + // No usable underlying socket (e.g. not connected yet) + this.countedSocket = null; + return {deltaRead: 0, deltaWritten: 0, rebaselined: true}; + } + if (raw !== this.countedSocket) { + // First sample, or the socket was replaced: establish a new baseline + this.countedSocket = raw; + this.lastBytesRead = raw.bytesRead; + this.lastBytesWritten = raw.bytesWritten; + return {deltaRead: 0, deltaWritten: 0, rebaselined: true}; + } + const deltaRead = raw.bytesRead - this.lastBytesRead; + const deltaWritten = raw.bytesWritten - this.lastBytesWritten; + this.lastBytesRead = raw.bytesRead; + this.lastBytesWritten = raw.bytesWritten; + return {deltaRead, deltaWritten, rebaselined: false}; + } + + /** + * Attaches the protocol-level 'pong' listener to the current underlying ws + * WebSocket. No-op if already attached to it; re-attaches if it was replaced. + * The 'pong' event is part of ws's Node.js EventEmitter API, so this quietly + * does nothing on sockets that don't support it. + */ + _ensureProtocolPingHooks() { + const websocket = this.toolsocket.socket; + if (websocket === this.pingedSocket) { + return; + } + this._detachProtocolPingHooks(); + if (!websocket || typeof websocket.on !== 'function') { + return; + } + this.pongHandler = (data) => { + const text = data.toString(); + if (!text.startsWith(PROTOCOL_PING_PREFIX)) { + return; // a pong for someone else's ping + } + const sentAt = parseFloat(text.slice(PROTOCOL_PING_PREFIX.length)); + if (!isFinite(sentAt)) { + return; + } + const roundTripMs = Math.round((now() - sentAt) * 10) / 10; + this.lastNetworkLatencyMs = roundTripMs; + this.networkLatencySamples.push(roundTripMs); + if (this.probeRunning && this.probeResult + && (this.probeResult.latencyUnderLoadMs === null + || roundTripMs > this.probeResult.latencyUnderLoadMs)) { + this.probeResult.latencyUnderLoadMs = roundTripMs; + } + }; + websocket.on('pong', this.pongHandler); + this.pingedSocket = websocket; + } + + /** + * Detaches the protocol-level 'pong' listener, if attached + */ + _detachProtocolPingHooks() { + if (this.pingedSocket && this.pongHandler + && typeof this.pingedSocket.off === 'function') { + this.pingedSocket.off('pong', this.pongHandler); + } + this.pingedSocket = null; + this.pongHandler = null; + } + + /** + * Sends one WebSocket protocol-level PING frame with the current time embedded + * in its payload, so the echoed PONG carries its own send time (stateless RTT). + */ + _sendProtocolPing() { + const websocket = this.toolsocket.socket; + if (!websocket || typeof websocket.ping !== 'function' || websocket.readyState !== 1) { + return; + } + try { + websocket.ping(PROTOCOL_PING_PREFIX + now()); + } catch (_e) { + // socket raced into a closing state; skip this sample + } + } + + /** + * Turns this window's low-level signals into a plain-language quality summary + * that a non-expert can act on. See the module comment for the reasoning. + * @param {?Object} networkLatency - this window's protocol ping stats + * @param {?Object} appLatency - this window's ToolSocket ping stats + * @returns {Object} + */ + _deriveNetworkQuality(networkLatency, appLatency) { + const issues = []; + const measuredSeconds = Math.max(this.tickCount, 1); + + // --- continuity: the realtime dimension. Our own 1 Hz protocol ping + // guarantees inbound bytes every second on a live path, so continuity is + // simply the fraction of measured seconds that actually carried data. + const liveSeconds = Math.max(measuredSeconds - this.silentSeconds, 0); + const continuity = Math.round(100 * liveSeconds / measuredSeconds); + + let flow = 'realtime'; + if (this.closeInfo) { + flow = 'ended'; + } else if (this.longestSilenceSeconds >= 3) { + flow = 'stalled'; + issues.push('incoming-data-stalled'); + } else if (this.silentSeconds >= 1) { + flow = 'buffered'; + issues.push('data-arriving-in-bursts-not-realtime'); + } + + // --- latency and stability, from this window's protocol ping round trips + let jitterMs = null; + let latencyScore = null; + let stabilityScore = null; + if (networkLatency) { + jitterMs = Math.round((networkLatency.maxMs - networkLatency.minMs) * 10) / 10; + latencyScore = interpolateScore(LATENCY_SCORE_ANCHORS, networkLatency.averageMs); + stabilityScore = interpolateScore(JITTER_SCORE_ANCHORS, jitterMs); + if (networkLatency.averageMs > 150) { + issues.push('high-latency'); + } + if (jitterMs > 100 || (jitterMs > 20 && jitterMs > networkLatency.averageMs * 2)) { + issues.push('latency-unstable'); + } + } + + // --- delivery: is our outbound data draining, or queuing locally? + // Score from the worst queue depth seen, reduced further the more of the + // window the queue existed for (persistent backpressure is worse than a blip). + // Withheld entirely for windows in which a probe transferred: the probe's own + // burst causes send-buffer pressure, and we must not raise alarms about it. + let delivery = null; + if (!this.probeActiveInWindow) { + const persistencePenalty = Math.round(30 * this.bufferedTicks / measuredSeconds); + delivery = Math.max(0, + interpolateScore(BUFFERED_SCORE_ANCHORS, this.maxBufferedBytes) - persistencePenalty); + if (this.bufferedTicks >= 1 && (this.maxBufferedBytes > 16 * 1024 || this.bufferedTicks >= 3)) { + issues.push('outgoing-data-queuing-locally'); + } + } + + // Client pressure: informational, not a network dimension + if (networkLatency && appLatency + && appLatency.averageMs - networkLatency.averageMs > 100) { + issues.push('client-under-pressure'); + } + + // --- overall: 50% the worst dimension + 50% a weighted average, so a single + // failing dimension drags the overall down the way it drags realtime down + const weighted = [ + [continuity, 0.4], + [latencyScore, 0.25], + [stabilityScore, 0.2], + [delivery, 0.15], + ].filter(([value]) => value !== null); + const weightSum = weighted.reduce((sum, [, weight]) => sum + weight, 0); + const weightedAverage = weighted.reduce( + (sum, [value, weight]) => sum + value * weight, 0) / weightSum; + const worst = Math.min(...weighted.map(([value]) => value)); + let score = Math.round(0.5 * worst + 0.5 * weightedAverage); + + if (this.closeInfo && !this.closeInfo.endedCleanly) { + // Cut without a close handshake: the signature of proxies and zero-trust + // gateways killing the socket. Cap the final report's score accordingly. + score = Math.min(score, 30); + issues.push('connection-cut-abnormally'); + } + score = Math.max(0, Math.min(100, score)); + + let rating; + if (score >= 90) { + rating = 'excellent'; + } else if (score >= 70) { + rating = 'good'; + } else if (score >= 40) { + rating = 'degraded'; + } else { + rating = 'poor'; + } + + // --- trend: compare against the recent scores. Windows with probe traffic + // are compared but never recorded, so self-inflicted load can't shape the trend + let trend = 'stable'; + if (this.scoreHistory.length > 0) { + const previousAverage = this.scoreHistory.reduce((a, b) => a + b, 0) + / this.scoreHistory.length; + if (score >= previousAverage + TREND_THRESHOLD) { + trend = 'improving'; + } else if (score <= previousAverage - TREND_THRESHOLD) { + trend = 'degrading'; + } + } + if (!this.probeActiveInWindow) { + this.scoreHistory.push(score); + if (this.scoreHistory.length > TREND_HISTORY_LENGTH) { + this.scoreHistory.shift(); + } + } + + const quality = { + score: score, + rating: rating, + flow: flow, + trend: trend, + scores: { + continuity: continuity, + latency: latencyScore, + stability: stabilityScore, + delivery: delivery, + }, + issues: issues, + details: { + jitterMs: jitterMs, + silentSeconds: this.silentSeconds, + longestSilenceSeconds: this.longestSilenceSeconds, + maxBufferedBytes: this.maxBufferedBytes, + connectionAgeSeconds: Math.round((Date.now() - this.connectionStartMs) / 1000), + probeTrafficInWindow: this.probeActiveInWindow, + }, + }; + if (this.closeInfo) { + quality.details.closeCode = this.closeInfo.closeCode; + quality.details.endedCleanly = this.closeInfo.endedCleanly; + } + return quality; + } + + /** + * Summarizes a window of round trip samples into the report's latency shape + * @param {number[]} samples + * @param {?number} currentMs - the most recent round trip measured + * @returns {?Object} stats, or null if there were no samples this window + */ + _summarizeLatency(samples, currentMs) { + if (samples.length === 0) { + return null; + } + const sum = samples.reduce((a, b) => a + b, 0); + return { + currentMs: currentMs, + averageMs: Math.round((sum / samples.length) * 10) / 10, + minMs: Math.min(...samples), + maxMs: Math.max(...samples), + samples: samples.length, + }; + } + + /** + * Drops pending ping entries that never received a response + */ + _prunePendingPings() { + const cutoff = now() - PENDING_PING_TIMEOUT_MS; + for (const [id, sentAt] of Object.entries(this.pendingAppPings)) { + if (sentAt < cutoff) { + delete this.pendingAppPings[id]; + } + } + } + + /** + * Registers a listener on the ToolSocket and records it for later removal + * @param {string} eventType + * @param {function} handler + */ + _listen(eventType, handler) { + this.listeners[eventType] = handler; + this.toolsocket.addEventListener(eventType, handler); + } +} + +/** + * Runs one throughput probe on any connected ToolSocket: sends sizeBytes of + * incompressible data downstream (meta 'probe/down', tiny ack back), then requests + * the same amount upstream (meta 'probe/up'). Standalone core used both by + * ToolSocketInfo.startProbe and by ToolSocketServer.stagedProbe. Never rejects. + * @param {Object} toolsocket - Any ToolSocket (info does not need to be enabled) + * @param {number} [sizeBytes] - Payload per direction (default 256 KB) + * @param {number} [rttMs] - RTT baseline subtracted from transfer timings + * @returns {Promise} - The probe result object + */ +function runProbe(toolsocket, sizeBytes, rttMs = 0) { + return new Promise((resolve) => { + const size = (typeof sizeBytes === 'number' && sizeBytes > 0) + ? Math.floor(sizeBytes) : DEFAULT_PROBE_SIZE_BYTES; + const result = { + status: 'running', + timestamp: Date.now(), + sizeBytes: size, + rttMsAtProbe: rttMs, + downstreamBytesPerSecond: null, + upstreamBytesPerSecond: null, + concurrentSentBytes: 0, + concurrentReceivedBytes: 0, + contended: false, + latencyUnderLoadMs: null, + }; + if (!toolsocket.connected) { + result.status = 'failed'; + result.reason = 'not-connected'; + resolve(result); + return; + } + let done = false; + const finish = () => { + if (!done) { + done = true; + result.timestamp = Date.now(); + resolve(result); + } + }; + const timeout = setTimeout(() => { + result.status = 'failed'; + result.reason = 'timeout-or-unsupported-client'; + finish(); + }, PROBE_TIMEOUT_MS); + if (timeout.unref) { + timeout.unref(); + } + + const raw = toolsocket.socket && toolsocket.socket._socket; + const counterStart = (raw && typeof raw.bytesRead === 'number') + ? {read: raw.bytesRead, written: raw.bytesWritten} : null; + const toRate = (bytes, elapsedMs) => + Math.round(bytes / Math.max(elapsedMs - rttMs, 0.5) * 1000); + + // Phase 1 - downstream: send a large incompressible payload, get a tiny ack + const downStart = now(); + toolsocket.meta('probe/down', null, () => { + if (done) { + return; + } + result.downstreamBytesPerSecond = toRate(size, now() - downStart); + + // Phase 2 - upstream: ask the peer for the same amount back + const upStart = now(); + toolsocket.meta('probe/up', size, (_body, binaryData) => { + if (done) { + return; + } + const received = (binaryData && binaryData.byteLength) || size; + result.upstreamBytesPerSecond = toRate(received, now() - upStart); + // Concurrent app traffic during the probe (approximate; envelope + // and frame overhead slightly overstate it) + if (counterStart && raw && typeof raw.bytesRead === 'number') { + result.concurrentSentBytes = Math.max(0, + raw.bytesWritten - counterStart.written - size); + result.concurrentReceivedBytes = Math.max(0, + raw.bytesRead - counterStart.read - received); + result.contended = (result.concurrentSentBytes + + result.concurrentReceivedBytes) > 0.1 * (size + received); + } + result.status = 'ok'; + clearTimeout(timeout); + finish(); + }); + }, makeProbePayload(size)); + }); +} + +ToolSocketInfo.runProbe = runProbe; +ToolSocketInfo.DEFAULT_PROBE_SIZE_BYTES = DEFAULT_PROBE_SIZE_BYTES; + +module.exports = ToolSocketInfo; diff --git a/src/ToolSocketInfo.test.js b/src/ToolSocketInfo.test.js new file mode 100644 index 00000000..3c851d9d --- /dev/null +++ b/src/ToolSocketInfo.test.js @@ -0,0 +1,272 @@ +/* global jest, describe, test, expect, afterEach */ + +/** + * Integration tests for the connection info API: per-connection info reports + * (ToolSocketInfo), throughput probes, the server-wide staged probe, and the + * client-side remote info subscription. Uses real sockets on an ephemeral port. + */ +const path = require('path'); +const ToolSocket = require('./index.js'); + +jest.setTimeout(45000); + +const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +// Poll until the condition holds — generous ceiling so timing-sensitive +// expectations survive slow CI runners and parallel-suite contention. +async function until(condition, timeoutMs = 25000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (condition()) return true; + await wait(150); + } + return condition(); +} + +async function startServer() { + const server = new ToolSocket.Server({port: 0}); + await new Promise((resolve) => server.on('listening', resolve)); + return {server, port: server.server.address().port}; +} + +function connectClient(port) { + const client = new ToolSocket(`ws://localhost:${port}`, 'testnet', 'web'); + return new Promise((resolve) => client.on('open', () => resolve(client))); +} + +describe('connection info API', () => { + let server = null; + let clients = []; + + afterEach(async () => { + for (const client of clients) { + try { + client.close(); + } catch (_e) { /* already closed */ } + } + clients = []; + if (server) { + try { + server.close(); + } catch (_e) { /* already closed */ } + server = null; + } + await wait(100); + }); + + test('is dormant until enabled, reports every 5s, and tears down cleanly', async () => { + const infoModulePath = path.resolve(__dirname, 'ToolSocketInfo.js'); + const started = await startServer(); + server = started.server; + const connectionPromise = new Promise((resolve) => server.on('connection', resolve)); + clients.push(await connectClient(started.port)); + const serverSocket = await connectionPromise; + + // Dormant: module not even loaded, no handler, client has no local info API + expect(require.cache[infoModulePath]).toBeUndefined(); + expect(serverSocket.infoHandler).toBeNull(); + const baselineListeners = Object.values(serverSocket.eventCallbacks).flat().length; + + const reports = []; + serverSocket.info(true, (report) => reports.push(report), {name: 'jest-user'}); + expect(require.cache[infoModulePath]).toBeDefined(); + + await until(() => reports.length >= 1); + expect(reports.length).toBeGreaterThanOrEqual(1); + const data = reports[0].data; + expect(reports[0].type).toBe('info'); + expect(data.name).toBe('jest-user'); + expect(data.networkLatency).not.toBeNull(); + expect(data.networkLatency.samples).toBeGreaterThanOrEqual(1); + expect(data.transport.receivedBytes).toBeGreaterThan(0); + expect(data.networkQuality.flow).toBe('realtime'); + expect(data.networkQuality.score).toBeGreaterThanOrEqual(0); + expect(data.networkQuality.score).toBeLessThanOrEqual(100); + expect(data.networkQuality.scores).toHaveProperty('continuity'); + expect(data.history.reports).toBe(1); + expect(data.probe).toBeNull(); + + // Teardown: handler gone, listener count restored, no further reports + serverSocket.info(false); + expect(serverSocket.infoHandler).toBeNull(); + expect(Object.values(serverSocket.eventCallbacks).flat().length).toBe(baselineListeners); + const countAfterDisable = reports.length; + await wait(5600); + expect(reports.length).toBe(countAfterDisable); + }); + + test('a graceful client close produces a clean final report', async () => { + const started = await startServer(); + server = started.server; + const connectionPromise = new Promise((resolve) => server.on('connection', resolve)); + const client = await connectClient(started.port); + const serverSocket = await connectionPromise; + + const reports = []; + serverSocket.info(true, (report) => reports.push(report)); + await wait(300); + client.close(); // orderly close handshake (close frame without status code) + await until(() => reports.length > 0 && reports[reports.length - 1].data.networkQuality.flow === 'ended'); + + const finalReport = reports[reports.length - 1]; + expect(finalReport.data.networkQuality.flow).toBe('ended'); + expect(finalReport.data.networkQuality.details.endedCleanly).toBe(true); + expect(finalReport.data.networkQuality.issues) + .not.toContain('connection-cut-abnormally'); + }); + + test('runProbe measures both directions on any connection', async () => { + const started = await startServer(); + server = started.server; + const connectionPromise = new Promise((resolve) => server.on('connection', resolve)); + clients.push(await connectClient(started.port)); + const serverSocket = await connectionPromise; + + const ToolSocketInfo = require('./ToolSocketInfo.js'); + const result = await ToolSocketInfo.runProbe(serverSocket, 32 * 1024); + expect(result.status).toBe('ok'); + expect(result.sizeBytes).toBe(32 * 1024); + expect(result.downstreamBytesPerSecond).toBeGreaterThan(0); + expect(result.upstreamBytesPerSecond).toBeGreaterThan(0); + }); + + test('stagedProbe separates individual limits from the network limit', async () => { + const started = await startServer(); + server = started.server; + const connections = []; + server.on('connection', (socket) => connections.push(socket)); + clients.push(await connectClient(started.port)); + clients.push(await connectClient(started.port)); + clients.push(await connectClient(started.port)); + await wait(200); + + const result = await new Promise((resolve) => + server.stagedProbe(resolve, {sizeBytes: 32 * 1024})); + expect(result.status).toBe('ok'); + expect(result.connections).toBe(3); + expect(result.individual.perConnection).toHaveLength(3); + expect(result.individual.totalDownstreamBytesPerSecond).toBeGreaterThan(0); + expect(result.stages[result.stages.length - 1].concurrent).toBe(3); + expect(result.networkLimit.downstreamBytesPerSecond).toBeGreaterThan(0); + expect(typeof result.sharedBottleneck.detected).toBe('boolean'); + expect(server.lastStagedProbeResult).toBe(result); + }); + + test('stagedProbe options.names probes only the named connections', async () => { + const started = await startServer(); + server = started.server; + const wifiClient = await connectClient(started.port); + clients.push(wifiClient); + clients.push(await connectClient(started.port)); // unnamed bystander + wifiClient.infoName('wifi-A'); + await wait(300); + + const result = await new Promise((resolve) => + server.stagedProbe(resolve, {sizeBytes: 32 * 1024, ramp: false, names: ['wifi-A']})); + expect(result.status).toBe('ok'); + expect(result.connections).toBe(1); + expect(result.individual.perConnection).toHaveLength(1); + expect(result.individual.perConnection[0].name).toBe('wifi-A'); + expect(typeof result.individual.perConnection[0].id).toBe('string'); + + // ids address any connection, named or not + const wifiId = result.individual.perConnection[0].id; + const byId = await new Promise((resolve) => + server.stagedProbe(resolve, {sizeBytes: 32 * 1024, ramp: false, ids: [wifiId]})); + expect(byId.status).toBe('ok'); + expect(byId.connections).toBe(1); + expect(byId.individual.perConnection[0].id).toBe(wifiId); + + // a filter that matches nothing fails cleanly instead of probing everyone + const miss = await new Promise((resolve) => + server.stagedProbe(resolve, {names: ['no-such-name']})); + expect(miss.status).toBe('failed'); + expect(miss.reason).toBe('no-connections'); + }); + + test('client-side info() streams server bundles and stops on request', async () => { + const started = await startServer(); + server = started.server; + const subscriber = await connectClient(started.port); + clients.push(subscriber); + clients.push(await connectClient(started.port)); + + const bundles = []; + subscriber.info(true, (bundle) => bundles.push(bundle)); + await until(() => bundles.length >= 1 && bundles[bundles.length - 1].connections === 2); + + expect(bundles.length).toBeGreaterThanOrEqual(1); + const bundle = bundles[bundles.length - 1]; + expect(bundle.type).toBe('serverInfo'); + expect(bundle.connections).toBe(2); + expect(Array.isArray(bundle.reports)).toBe(true); + expect(Array.isArray(bundle.recentlyClosed)).toBe(true); + + subscriber.info(false); + await wait(300); + const countAfterStop = bundles.length; + await wait(5600); + expect(bundles.length).toBe(countAfterStop); + expect(server.infoBroadcastInterval).toBeNull(); + // Auto-enabled per-connection info was turned off again + expect(server.sockets.every((socket) => socket.infoHandler === null)).toBe(true); + }); + + test('client-side infoName() names the connection and the name survives enable/disable cycles', async () => { + const started = await startServer(); + server = started.server; + const connectionPromise = new Promise((resolve) => server.on('connection', resolve)); + const namedClient = await connectClient(started.port); + clients.push(namedClient); + const namedServerSocket = await connectionPromise; + + // Announcing a name is not an info activation: the socket stays dormant + namedClient.infoName('avatar-alice'); + await wait(300); + expect(namedServerSocket.announcedInfoName).toBe('avatar-alice'); + expect(namedServerSocket.infoHandler).toBeNull(); + + // A subscriber auto-enables info on all connections: reports carry the name + const subscriber = await connectClient(started.port); + clients.push(subscriber); + const bundles = []; + subscriber.info(true, (bundle) => bundles.push(bundle)); + await until(() => bundles.some((b) => b.reports.some((r) => r.data && r.data.name === 'avatar-alice'))); + const names = bundles.flatMap((b) => b.reports.map((r) => r.data && r.data.name)); + expect(names).toContain('avatar-alice'); + + // Last subscriber leaves: info returns to dormant but the name is retained... + subscriber.info(false); + await wait(300); + expect(namedServerSocket.infoHandler).toBeNull(); + expect(namedServerSocket.announcedInfoName).toBe('avatar-alice'); + + // ...so a fresh subscription still sees the named connection + const bundlesAgain = []; + subscriber.info(true, (bundle) => bundlesAgain.push(bundle)); + await until(() => bundlesAgain.some((b) => b.reports.some((r) => r.data && r.data.name === 'avatar-alice'))); + const namesAgain = bundlesAgain.flatMap((b) => b.reports.map((r) => r.data && r.data.name)); + expect(namesAgain).toContain('avatar-alice'); + subscriber.info(false); + + // Clearing the name + namedClient.infoName(null); + await wait(300); + expect(namedServerSocket.announcedInfoName).toBeNull(); + }); + + test('removeEventListener removes exactly the given listener', () => { + const socket = new ToolSocket(); + const calls = []; + const listenerA = () => calls.push('a'); + const listenerB = () => calls.push('b'); + socket.addEventListener('custom', listenerA); + socket.addEventListener('custom', listenerB); + socket.triggerEvent('custom'); + socket.removeEventListener('custom', listenerA); + socket.triggerEvent('custom'); + expect(calls).toEqual(['a', 'b', 'b']); + socket.removeEventListener('custom', listenerB); + expect(socket.eventCallbacks.custom).toBeUndefined(); + }); +}); diff --git a/src/ToolSocketServer.js b/src/ToolSocketServer.js index ea9e9e6d..b5a968f6 100644 --- a/src/ToolSocketServer.js +++ b/src/ToolSocketServer.js @@ -3,6 +3,9 @@ const { URL_SCHEMA, MESSAGE_BUNDLE_SCHEMA } = require('./schemas.js'); const IncomingToolSocket = require('./IncomingToolSocket'); const {generateUniqueId} = require("./utilities"); +// How many final reports of closed connections are kept for info subscribers +const MAX_CLOSED_INFO_REPORTS = 25; + /** * A server for ToolSocket */ @@ -23,6 +26,21 @@ class ToolSocketServer { this.pendingParallelRequests = new Map(); + // Staged throughput probe state (see stagedProbe()) + this.stagedProbeRunning = false; + /** @type {?Object} latest staged probe result */ + this.lastStagedProbeResult = null; + + // Remote info subscriptions (see subscribeServerInfo()) + /** @type {Set} */ + this.infoSubscribers = new Set(); + /** @type {?ReturnType} */ + this.infoBroadcastInterval = null; + /** @type {Set} connections whose info THIS feature enabled */ + this.infoAutoEnabled = new Set(); + /** @type {Object[]} final info reports of recently closed connections */ + this.infoClosedReports = []; + this.server.on('listening', (...args) => { this.triggerEvent('listening', ...args); }); @@ -42,6 +60,20 @@ class ToolSocketServer { socket.on('close', () => { this.sockets.splice(this.sockets.indexOf(toolSocket), 1); + this.infoAutoEnabled.delete(toolSocket); + // Keep the connection's final report (its info handler pushes it on + // 'close' before this listener runs) for remote info subscribers - + // closed connections are the main evidence of network-level cuts + if (toolSocket.infoHandler && toolSocket.infoHandler.latestReport) { + this.infoClosedReports.push(toolSocket.infoHandler.latestReport); + if (this.infoClosedReports.length > MAX_CLOSED_INFO_REPORTS) { + this.infoClosedReports.shift(); + } + } + // A closing subscriber ends its own subscription + if (this.infoSubscribers.has(toolSocket)) { + this.unsubscribeServerInfo(toolSocket); + } }); }); @@ -145,8 +177,260 @@ class ToolSocketServer { } close() { + if (this.infoBroadcastInterval) { + clearInterval(this.infoBroadcastInterval); + this.infoBroadcastInterval = null; + } + this.infoSubscribers.clear(); + this.infoAutoEnabled.clear(); this.server.close(); } + + /** + * Starts streaming this server's info reports for ALL connections to the given + * subscriber socket via meta 'info/report' bundles, every 5 seconds, until + * unsubscribeServerInfo() or the subscriber's connection closes. Normally invoked + * through the client-side info(true, callback) API rather than directly. Enables + * info on every connection that doesn't have it yet (and turns exactly those off + * again when the last subscriber leaves, unless a local callback was attached to + * them in the meantime). + * @param {IncomingToolSocket} subscriber + */ + subscribeServerInfo(subscriber) { + this.infoSubscribers.add(subscriber); + if (!this.infoBroadcastInterval) { + this.infoBroadcastInterval = setInterval(() => this._broadcastServerInfo(), 5000); + if (this.infoBroadcastInterval.unref) { + this.infoBroadcastInterval.unref(); + } + this._broadcastServerInfo(); // arm info on all connections right away + } + } + + /** + * Ends a subscriber's info stream. When the last subscriber leaves, the + * broadcast stops and auto-enabled connection info is turned off again. + * @param {IncomingToolSocket} subscriber + */ + unsubscribeServerInfo(subscriber) { + this.infoSubscribers.delete(subscriber); + if (this.infoSubscribers.size > 0 || !this.infoBroadcastInterval) { + return; + } + clearInterval(this.infoBroadcastInterval); + this.infoBroadcastInterval = null; + for (const socket of this.infoAutoEnabled) { + // Leave info running if someone attached a local callback meanwhile + if (socket.infoHandler && !socket.infoHandler.callback) { + socket.info(false); + } + } + this.infoAutoEnabled.clear(); + } + + /** + * Collects every connection's latest info report and pushes one bundle to each + * subscriber. Also enables info on connections that joined after subscription. + */ + _broadcastServerInfo() { + const reports = []; + for (const socket of this.sockets) { + if (!socket.infoHandler && socket.connected && socket.info) { + socket.info(true); // no callback: reports land in latestReport only + this.infoAutoEnabled.add(socket); + } + if (socket.infoHandler && socket.infoHandler.latestReport) { + reports.push(socket.infoHandler.latestReport); + } + } + const bundle = { + type: 'serverInfo', + timestamp: Date.now(), + connections: this.sockets.length, + reports: reports, + recentlyClosed: this.infoClosedReports.slice(), + stagedProbe: this.lastStagedProbeResult, + }; + for (const subscriber of this.infoSubscribers) { + if (subscriber.connected) { + subscriber.meta('info/report', bundle); + } + } + } + + /** + * Runs a staged throughput probe across all connected sockets to distinguish the + * per-client limit from the shared network limit (e.g. ten clients on one WiFi). + * + * Stage 'individual': every connection is probed alone, one after another — each + * client's own capacity, free of probe-vs-probe contention. + * Growing stages: 2, 4, 8, ... connections probed simultaneously, ending with all + * at once. If the summed rate stops growing while more clients join in, the + * shared medium is saturated — that plateau is the network limit. + * + * sharedBottleneck.downstreamRatio = (sum of individual capacities) / (all-at-once + * total). ~1 means clients barely limit each other; above ~1.5 the shared network + * is the bottleneck (flagged as detected: true). + * + * Deliberately saturates the network while running; intended for manual trigger. + * One staged probe at a time per server. Connections with info enabled get their + * per-connection probe results updated along the way. + * + * @param {function} callback - Receives the full result object when finished + * @param {?Object} [options] + * @param {number} [options.sizeBytes] - Payload per direction per probe (default 256 KB) + * @param {boolean} [options.ramp] - Include the growing 2, 4, 8... stages between + * 'individual' and all-at-once (default true) + * @param {string[]} [options.names] - Probe only the connections carrying one of + * these names (assigned via infoName() or + * info options.name) — e.g. just the clients + * on one Wi-Fi. Omit to probe all. + * @param {string[]} [options.ids] - Probe only the connections with one of these + * ids (data.id in info reports) — addresses any + * connection, named or not. Combines with + * options.names as a union. + */ + stagedProbe(callback, options = {}) { + if (typeof callback !== 'function') { + return; + } + if (this.stagedProbeRunning) { + callback({status: 'failed', reason: 'staged-probe-already-running'}); + return; + } + // Lazy require: staged probing shares the dormant info module + const ToolSocketInfo = require('./ToolSocketInfo.js'); + let connections = this.sockets.filter(socket => socket.connected); + const wantedNames = (Array.isArray(options.names) && options.names.length > 0) ? new Set(options.names) : null; + const wantedIds = (Array.isArray(options.ids) && options.ids.length > 0) ? new Set(options.ids) : null; + if (wantedNames || wantedIds) { + connections = connections.filter((socket) => + (wantedIds && wantedIds.has(socket.infoId)) || + (wantedNames && wantedNames.has( + (socket.infoHandler && socket.infoHandler.connectionName) || socket.announcedInfoName))); + } + if (connections.length === 0) { + callback({status: 'failed', reason: 'no-connections'}); + return; + } + this.stagedProbeRunning = true; + const sizeBytes = (typeof options.sizeBytes === 'number' && options.sizeBytes > 0) + ? Math.floor(options.sizeBytes) : ToolSocketInfo.DEFAULT_PROBE_SIZE_BYTES; + const ramp = options.ramp !== false; + + // Probe one connection: through its info handler when enabled (keeps its + // per-connection reports and quality windows consistent), bare otherwise + const probeOne = (connection) => new Promise((resolve) => { + if (connection.infoHandler && connection.infoHandler.active) { + if (!connection.infoHandler.startProbe(sizeBytes, resolve)) { + resolve({status: 'failed', reason: 'probe-already-running'}); + } + } else { + ToolSocketInfo.runProbe(connection, sizeBytes, 0).then(resolve); + } + }); + const describe = (connection, result) => ({ + id: connection.infoId || null, + // the handler's name when info is active, else the remotely announced one + // (infoName()) — probes can run without any info subscriber + name: (connection.infoHandler && connection.infoHandler.connectionName) + || connection.announcedInfoName || null, + status: result.status, + downstreamBytesPerSecond: result.downstreamBytesPerSecond, + upstreamBytesPerSecond: result.upstreamBytesPerSecond, + contended: result.contended || false, + }); + const total = (list, key) => list.reduce( + (accumulated, entry) => accumulated + + ((entry.status === 'ok' && entry[key]) ? entry[key] : 0), 0); + + const run = async () => { + const startedAt = Date.now(); + + // Stage 'individual': sequential, one connection at a time + const individual = []; + for (const connection of connections) { + const result = await probeOne(connection); + individual.push(describe(connection, result)); + } + const individualTotalDown = total(individual, 'downstreamBytesPerSecond'); + const individualTotalUp = total(individual, 'upstreamBytesPerSecond'); + + // Growing stages: 2, 4, 8, ... simultaneous probes, always ending with all + const groupSizes = []; + if (ramp) { + for (let k = 2; k < connections.length; k *= 2) { + groupSizes.push(k); + } + } + if (connections.length > 1) { + groupSizes.push(connections.length); + } + const stages = []; + for (const concurrent of groupSizes) { + const group = connections.slice(0, concurrent); + const results = await Promise.all(group.map(probeOne)); + const perConnection = results.map( + (result, index) => describe(group[index], result)); + stages.push({ + concurrent: concurrent, + totalDownstreamBytesPerSecond: total(perConnection, 'downstreamBytesPerSecond'), + totalUpstreamBytesPerSecond: total(perConnection, 'upstreamBytesPerSecond'), + perConnection: perConnection, + }); + } + + // Network limit: the all-at-once totals (with a single connection, its + // individual capacity IS the network limit) + const allStage = stages.length > 0 ? stages[stages.length - 1] : { + totalDownstreamBytesPerSecond: individualTotalDown, + totalUpstreamBytesPerSecond: individualTotalUp, + }; + const ratio = (individualSum, allSum) => allSum > 0 + ? Math.round((individualSum / allSum) * 100) / 100 : null; + const downstreamRatio = ratio(individualTotalDown, allStage.totalDownstreamBytesPerSecond); + const upstreamRatio = ratio(individualTotalUp, allStage.totalUpstreamBytesPerSecond); + + return { + status: 'ok', + startedAt: startedAt, + finishedAt: Date.now(), + sizeBytes: sizeBytes, + connections: connections.length, + individual: { + perConnection: individual, + totalDownstreamBytesPerSecond: individualTotalDown, + totalUpstreamBytesPerSecond: individualTotalUp, + }, + stages: stages, + networkLimit: { + downstreamBytesPerSecond: allStage.totalDownstreamBytesPerSecond, + upstreamBytesPerSecond: allStage.totalUpstreamBytesPerSecond, + }, + sharedBottleneck: { + downstreamRatio: downstreamRatio, + upstreamRatio: upstreamRatio, + detected: (downstreamRatio !== null && downstreamRatio > 1.5) + || (upstreamRatio !== null && upstreamRatio > 1.5), + }, + }; + }; + run().then((result) => { + this.lastStagedProbeResult = result; + this.stagedProbeRunning = false; + try { + callback(result); + } catch (error) { + console.warn('stagedProbe callback threw', error); + } + }, (error) => { + console.warn('stagedProbe failed', error); + this.stagedProbeRunning = false; + try { + callback({status: 'failed', reason: 'internal-error'}); + } catch (_e) { /* app callback error */ } + }); + } } module.exports = ToolSocketServer; diff --git a/src/utilities.js b/src/utilities.js index 70b2287e..12cfcead 100644 --- a/src/utilities.js +++ b/src/utilities.js @@ -58,6 +58,31 @@ function addSearchParams(url, newParams) { return newUrl; } +// Largest payload a throughput probe may request (see ToolSocketInfo.js) +const MAX_PROBE_PAYLOAD_BYTES = 4 * 1024 * 1024; + +/** + * Builds an incompressible payload for throughput probes. Random-ish content is + * required so that compression anywhere on the path cannot fake higher rates. + * Uses a fast xorshift generator; only runs when a probe is explicitly requested. + * @param {number} requestedBytes + * @returns {Uint8Array} + */ +function makeProbePayload(requestedBytes) { + const size = (typeof requestedBytes === 'number' && requestedBytes > 0) + ? Math.min(Math.floor(requestedBytes), MAX_PROBE_PAYLOAD_BYTES) + : 1024; + const payload = new Uint8Array(size); + let state = (Date.now() & 0x7fffffff) | 1; + for (let i = 0; i < size; i++) { + state ^= state << 13; + state ^= state >>> 17; + state ^= state << 5; + payload[i] = state & 0xff; + } + return payload; +} + const isBrowser = typeof window !== 'undefined'; /** @type {WebSocket} */ const WebSocketWrapper = isBrowser ? WebSocket : require('ws'); @@ -68,5 +93,6 @@ module.exports = { generateUniqueId, addSearchParams, isBrowser, - WebSocketWrapper + WebSocketWrapper, + makeProbePayload };