From 235a15b9cca3b5d38a915dfac665a11fcbe63d89 Mon Sep 17 00:00:00 2001 From: bbassie <17990055+bbassie@users.noreply.github.com> Date: Thu, 8 Oct 2026 11:15:27 +0000 Subject: [PATCH 1/2] Limit unacknowledged messages buffered for a connection Every message sent over a link is kept in the send buffer until the other side acknowledges it with a heartbeat, so it can be resent if the session resumes. Nothing forced a peer to acknowledge, so a client that kept heartbeating without a sequence number while subscribed to a busy event made the controller buffer messages until it ran out of memory. Store buffered messages serialised, track their total size, and close the connection once it goes over maxSendBufferSize. The controller sets this to 64 MiB for the connections it accepts. A client that far behind could not resume in a reasonable time anyway. Co-Authored-By: Claude Opus 5.5 --- packages/controller/src/WsServerConnector.ts | 6 ++- packages/lib/src/link/connectors.ts | 38 ++++++++++++++++--- test/lib/link/connector.js | 39 ++++++++++++++++++-- 3 files changed, 74 insertions(+), 9 deletions(-) diff --git a/packages/controller/src/WsServerConnector.ts b/packages/controller/src/WsServerConnector.ts index 439a2056..a66f1f61 100644 --- a/packages/controller/src/WsServerConnector.ts +++ b/packages/controller/src/WsServerConnector.ts @@ -13,6 +13,9 @@ import { logger } from "@clusterio/lib"; * @extends module:lib.WebSocketBaseConnector * @alias module:controller/src/WsServerConnector */ +// Clients that leave this much unacknowledged are dropped rather than buffered for. +const maxSendBufferSize = 64 * 1024 * 1024; + export default class WsServerConnector extends lib.WebSocketBaseConnector { _timeoutId: ReturnType | null = null; @@ -23,6 +26,7 @@ export default class WsServerConnector extends lib.WebSocketBaseConnector { public _heartbeatInterval: number, ) { super(new lib.Address(lib.Address.controller, 0), dst); + this.maxSendBufferSize = maxSendBufferSize; // The following states are used in the server connector // closed: Connection is closed @@ -110,7 +114,7 @@ export default class WsServerConnector extends lib.WebSocketBaseConnector { `Connector | resuming session with ${this.dst}, resending ${this._sendBuffer.length} buffered messages` ); for (let message of this._sendBuffer) { - this._socket.send(JSON.stringify(message)); + this._socket.send(message.text); } this.emit("resume"); } diff --git a/packages/lib/src/link/connectors.ts b/packages/lib/src/link/connectors.ts index d14b89e3..c9890915 100644 --- a/packages/lib/src/link/connectors.ts +++ b/packages/lib/src/link/connectors.ts @@ -171,7 +171,11 @@ export abstract class WebSocketBaseConnector< _heartbeatId: ReturnType | null = null; _heartbeatInterval: number | null = null; _lastReceivedSeq = undefined; - _sendBuffer: (libData.MessageRoutable)[] = []; + /** Serialised messages sent but not yet acknowledged by the other side */ + _sendBuffer: { seq: number, text: string }[] = []; + _sendBufferSize = 0; + /** Connection is closed if the unacknowledged messages exceed this many characters */ + maxSendBufferSize = Infinity; _reset() { this._state = "closed"; @@ -182,6 +186,7 @@ export abstract class WebSocketBaseConnector< this._heartbeatInterval = null; this._lastReceivedSeq = undefined; this._sendBuffer.length = 0; + this._sendBufferSize = 0; super._reset(); } @@ -194,6 +199,7 @@ export abstract class WebSocketBaseConnector< this._heartbeatInterval = null; this._lastReceivedSeq = undefined; this._sendBuffer.length = 0; + this._sendBufferSize = 0; super._invalidate(); } @@ -219,6 +225,7 @@ export abstract class WebSocketBaseConnector< for (let index = 0; index < this._sendBuffer.length; index++) { if (this._sendBuffer[index].seq <= seq) { dropCount += 1; + this._sendBufferSize -= this._sendBuffer[index].text.length; } else { break; } @@ -326,9 +333,30 @@ export abstract class WebSocketBaseConnector< throw new libErrors.SessionLost("No session"); } - this._sendBuffer.push(message); + const text = JSON.stringify(message); + this._sendBuffer.push({ seq: message.seq, text }); + this._sendBufferSize += text.length; + if (this._sendBufferSize > this.maxSendBufferSize) { + this._sendBufferOverflow(); + return; + } if (this._state === "connected") { - this._sendInternal(message); + this._socket!.send(text); + } + } + + _sendBufferOverflow() { + logger.warn( + `Connector | closing connection to ${this.dst} after ${this._sendBuffer.length} messages ` + + "were left unacknowledged" + ); + // Resuming is no longer possible, free the memory right away. + this._sendBuffer.length = 0; + this._sendBufferSize = 0; + if (!this._closing) { + this.close(ConnectionClosed.PolicyViolation, "Send buffer limit exceeded").catch(err => { + logger.error(`Connector | error closing connection:\n${err.stack}`); + }); } } @@ -708,7 +736,7 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector Date: Thu, 8 Oct 2026 15:18:02 +0000 Subject: [PATCH 2/2] Make the send buffer limit a config and start a new session on overflow Add controller.max_send_buffer_size in MiB, default 64 and 0 for no limit. Close with 1013 instead of 1008 on overflow: clients treat 1008 as fatal, so a host would shut down and the web UI would log out. With 1013 the client resumes, the controller answers with invalidate since the session has ended, and the client registers a new session. Co-Authored-By: Claude Opus 5.5 --- docs/configuration.md | 11 +++++++ packages/controller/src/WsServer.ts | 3 +- packages/controller/src/WsServerConnector.ts | 4 +-- packages/lib/src/config/definitions.ts | 9 ++++++ packages/lib/src/link/connectors.ts | 11 +++---- test/integration/link.js | 30 +++++++++++++++++++- test/lib/link/connector.js | 2 +- 7 files changed, 59 insertions(+), 11 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 818f0e48..5878e7a8 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -165,6 +165,17 @@ Must be greater than 0. Defaults to 60. +### controller.max_send_buffer_size + +Size in MiB of messages sent to a connection that it has not yet acknowledged before the controller ends its session. +The controller keeps these so they can be resent if the connection drops and resumes. +A client that stops acknowledging messages, or falls far behind, gets disconnected and starts a new session when it reconnects. +Changes apply to new sessions. +Set to 0 for no limit. + +Defaults to 64. + + ### controller.metrics_timeout Timeout in seconds before a call to gather metrics from a host times out. diff --git a/packages/controller/src/WsServer.ts b/packages/controller/src/WsServer.ts index efaf1061..db582ec1 100644 --- a/packages/controller/src/WsServer.ts +++ b/packages/controller/src/WsServer.ts @@ -226,7 +226,8 @@ ${err.stack}` let sessionToken = jwt.sign({ aud: this.sessionAud, sid: sessionId }, this.controller.authSecret); let sessionTimeout = this.controller.config.get("controller.session_timeout"); let heartbeatInterval = this.controller.config.get("controller.heartbeat_interval"); - let connector = new WsServerConnector(dst, sessionId, sessionTimeout, heartbeatInterval); + let maxSendBufferSize = this.controller.config.get("controller.max_send_buffer_size") * 1024 * 1024; + let connector = new WsServerConnector(dst, sessionId, sessionTimeout, heartbeatInterval, maxSendBufferSize); this.activeConnectors.set(sessionId, connector); connector.on("close", () => { this.activeConnectors.delete(sessionId); diff --git a/packages/controller/src/WsServerConnector.ts b/packages/controller/src/WsServerConnector.ts index a66f1f61..3b256d1a 100644 --- a/packages/controller/src/WsServerConnector.ts +++ b/packages/controller/src/WsServerConnector.ts @@ -13,9 +13,6 @@ import { logger } from "@clusterio/lib"; * @extends module:lib.WebSocketBaseConnector * @alias module:controller/src/WsServerConnector */ -// Clients that leave this much unacknowledged are dropped rather than buffered for. -const maxSendBufferSize = 64 * 1024 * 1024; - export default class WsServerConnector extends lib.WebSocketBaseConnector { _timeoutId: ReturnType | null = null; @@ -24,6 +21,7 @@ export default class WsServerConnector extends lib.WebSocketBaseConnector { private _sessionId: number, private _sessionTimeout: number, public _heartbeatInterval: number, + maxSendBufferSize = 0, ) { super(new lib.Address(lib.Address.controller, 0), dst); this.maxSendBufferSize = maxSendBufferSize; diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index 27042c44..83437105 100644 --- a/packages/lib/src/config/definitions.ts +++ b/packages/lib/src/config/definitions.ts @@ -25,6 +25,7 @@ export interface ControllerConfigFields { "controller.auth_secret": string | null; "controller.heartbeat_interval": number; "controller.session_timeout": number; + "controller.max_send_buffer_size": number; "controller.metrics_timeout": number; "controller.system_metrics_interval": number; "controller.proxy_stream_timeout": number; @@ -201,6 +202,14 @@ export class ControllerConfig extends classes.Config { initialValue: 60, validator: validators.greaterThanZero, }, + "controller.max_send_buffer_size": { + title: "Max Send Buffer Size", + description: + "MiB of unacknowledged messages to keep for a connection before ending its session, 0 for no limit.", + type: "number", + initialValue: 64, + validator: validators.greaterThanEqualZero, + }, "controller.metrics_timeout": { title: "Metrics Timeout", description: "Timeout in seconds for metrics gathering from hosts.", diff --git a/packages/lib/src/link/connectors.ts b/packages/lib/src/link/connectors.ts index c9890915..7bd62609 100644 --- a/packages/lib/src/link/connectors.ts +++ b/packages/lib/src/link/connectors.ts @@ -174,8 +174,8 @@ export abstract class WebSocketBaseConnector< /** Serialised messages sent but not yet acknowledged by the other side */ _sendBuffer: { seq: number, text: string }[] = []; _sendBufferSize = 0; - /** Connection is closed if the unacknowledged messages exceed this many characters */ - maxSendBufferSize = Infinity; + /** Session is ended if the unacknowledged messages exceed this many characters, 0 for no limit */ + maxSendBufferSize = 0; _reset() { this._state = "closed"; @@ -336,7 +336,7 @@ export abstract class WebSocketBaseConnector< const text = JSON.stringify(message); this._sendBuffer.push({ seq: message.seq, text }); this._sendBufferSize += text.length; - if (this._sendBufferSize > this.maxSendBufferSize) { + if (this.maxSendBufferSize && this._sendBufferSize > this.maxSendBufferSize) { this._sendBufferOverflow(); return; } @@ -350,11 +350,12 @@ export abstract class WebSocketBaseConnector< `Connector | closing connection to ${this.dst} after ${this._sendBuffer.length} messages ` + "were left unacknowledged" ); - // Resuming is no longer possible, free the memory right away. + // Resuming is no longer possible, free the memory right away. The + // other side sees a normal drop and gets a new session when it reconnects. this._sendBuffer.length = 0; this._sendBufferSize = 0; if (!this._closing) { - this.close(ConnectionClosed.PolicyViolation, "Send buffer limit exceeded").catch(err => { + this.close(ConnectionClosed.TryAgainLater, "Send buffer limit exceeded").catch(err => { logger.error(`Connector | error closing connection:\n${err.stack}`); }); } diff --git a/test/integration/link.js b/test/integration/link.js index 8bfdd211..01beb471 100644 --- a/test/integration/link.js +++ b/test/integration/link.js @@ -3,8 +3,9 @@ import events from "node:events"; import jwt from "jsonwebtoken"; import { - TestControlConnector, TestHostConnector, TestControl, url, controlToken, + TestControlConnector, TestHostConnector, TestControl, url, controlToken, getControl, } from "./index.js"; +import * as lib from "@clusterio/lib"; import { ConnectionClosed, ProtocolError, PolicyViolation, AuthenticationFailed } from "@clusterio/lib"; import { slowTest } from "../common.js"; @@ -92,6 +93,33 @@ describe("Integration of lib/link", function() { await events.once(controlConnector, "connect"); }); + it("should start a new session when the send buffer limit is exceeded", async function() { + await getControl().send(new lib.ControllerConfigSetFieldRequest("controller.max_send_buffer_size", "0.01")); + try { + await controlConnector.connect(); + } finally { + await getControl().send(new lib.ControllerConfigSetFieldRequest("controller.max_send_buffer_size", "64")); + } + + // Never acknowledge received messages + controlConnector._doHeartbeat = function() { + this._sendInternal(new lib.MessageHeartbeat()); + }; + let invalidated = false; + controlConnector.once("invalidate", () => { invalidated = true; }); + const connected = events.once(controlConnector, "invalidate").then( + () => events.once(controlConnector, "connect") + ); + for (let i = 0; i < 100; i++) { + if (invalidated) { + break; + } + await control.send(new lib.ControllerConfigGetRequest()).catch(() => {}); + } + assert(invalidated, "session was not invalidated"); + await connected; + }); + it("should properly close connector if close is called during reconnect wait", async function() { await controlConnector.connect(); controlConnector._socket.close(ConnectionClosed.Normal, "Test"); diff --git a/test/lib/link/connector.js b/test/lib/link/connector.js index e6f12316..a19f3696 100644 --- a/test/lib/link/connector.js +++ b/test/lib/link/connector.js @@ -65,7 +65,7 @@ describe("lib/link/connectors", function() { describe("send buffer limit", function() { afterEach(function() { - testConnector.maxSendBufferSize = Infinity; + testConnector.maxSendBufferSize = 0; testConnector._closing = false; }); it("should keep unacknowledged messages under the limit", function() {