Count send buffer size in UTF-8 bytes (#1116)

The send buffer limit in controller.max_send_buffer_size is set in MiB, but
buffered messages were measured with String.length, which counts UTF-16 code
units. Messages with non-ASCII text could go over the configured size in bytes
before the session was ended.

Measure each serialised message with Buffer.byteLength when it is buffered,
keep that size with the entry, and subtract the same size when it is
acknowledged. The web UI build already provides Buffer through the buffer
polyfill.

Fixes #1108

Co-authored-by: ColumbusLabs <287001685+ColumbusLabs@users.noreply.github.com>

packages/lib/src/link/connectors.ts

@@ -171,10 +171,11 @@ export abstract class WebSocketBaseConnector<
 	_heartbeatId: ReturnType<typeof setInterval> | null = null;
 	_heartbeatInterval: number | null = null;
 	_lastReceivedSeq = undefined;
-	/** Serialised messages sent but not yet acknowledged by the other side */
-	_sendBuffer: { seq: number, text: string }[] = [];
+	/** Serialised messages sent but not yet acknowledged by the other side, with their UTF-8 size in bytes */
+	_sendBuffer: { seq: number, text: string, size: number }[] = [];
+	/** Total UTF-8 size in bytes of the messages in the send buffer */
 	_sendBufferSize = 0;
-	/** Session is ended if the unacknowledged messages exceed this many characters, 0 for no limit */
+	/** Session is ended if the unacknowledged messages exceed this many bytes, 0 for no limit */
 	maxSendBufferSize = 0;
 
 	_reset() {
@@ -225,7 +226,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;
+				this._sendBufferSize -= this._sendBuffer[index].size;
 			} else {
 				break;
 			}
@@ -334,8 +335,9 @@ export abstract class WebSocketBaseConnector<
 		}
 
 		const text = JSON.stringify(message);
-		this._sendBuffer.push({ seq: message.seq, text });
-		this._sendBufferSize += text.length;
+		const size = Buffer.byteLength(text, "utf8");
+		this._sendBuffer.push({ seq: message.seq, text, size });
+		this._sendBufferSize += size;
 		if (this.maxSendBufferSize && this._sendBufferSize > this.maxSendBufferSize) {
 			this._sendBufferOverflow();
 			return;

test/lib/link/connector.js

@@ -13,9 +13,9 @@ describe("lib/link/connectors", function() {
 		describe("._dropSendBufferSeq()", function() {
 			beforeEach(function() {
 				testConnector._sendBuffer = [
-					{ seq: 1, text: "a" },
-					{ seq: 2, text: "bb" },
-					{ seq: 4, text: "cccc" },
+					{ seq: 1, text: "a", size: 1 },
+					{ seq: 2, text: "bb", size: 2 },
+					{ seq: 4, text: "cccc", size: 4 },
 				];
 				testConnector._sendBufferSize = 7;
 			});
@@ -64,10 +64,83 @@ describe("lib/link/connectors", function() {
 		});
 
 		describe("send buffer limit", function() {
+			const addr = new lib.Address(lib.Address.control, 1);
+			function textEvent(seq, text) {
+				return new lib.MessageEvent(seq, addr, addr, "TestEvent", { text });
+			}
+			function utf8Size(message) {
+				return Buffer.byteLength(JSON.stringify(message), "utf8");
+			}
+			function clearBuffer(connector) {
+				connector._state = "connected";
+				connector._sendBuffer = [];
+				connector._sendBufferSize = 0;
+				connector._socket.closeCalled = false;
+				connector._socket.sentMessages = [];
+			}
+
 			afterEach(function() {
 				testConnector.maxSendBufferSize = 0;
 				testConnector._closing = false;
 			});
+			it("should count buffered messages in UTF-8 bytes", function() {
+				// Extra bytes over the UTF-16 length for ASCII, 2 and 3 byte BMP characters and a surrogate pair
+				for (const [text, extraBytes] of [["ascii", 0], ["\u00e9", 1], ["\u20ac", 2], ["\u{1f600}", 2]]) {
+					clearBuffer(testConnector);
+					const event = textEvent(1, text);
+					testConnector.send(event);
+					const expected = JSON.stringify(event).length + extraBytes;
+					assert.equal(utf8Size(event), expected, text);
+					assert.equal(testConnector._sendBufferSize, expected, text);
+				}
+			});
+			it("should allow the limit to be reached exactly and close one byte over it", function() {
+				const event = textEvent(1, "\u20ac\u20ac\u20ac\u20ac");
+				clearBuffer(testConnector);
+				testConnector.maxSendBufferSize = utf8Size(event);
+				testConnector.send(event);
+				assert.equal(testConnector._sendBuffer.length, 1);
+				assert(!testConnector._socket.closeCalled, "Close was called on the socket");
+
+				clearBuffer(testConnector);
+				testConnector.maxSendBufferSize = utf8Size(event) - 1;
+				assert(JSON.stringify(event).length <= testConnector.maxSendBufferSize);
+				testConnector.send(event);
+				assert.equal(testConnector._sendBuffer.length, 0);
+				assert.equal(testConnector._sendBufferSize, 0);
+				assert(testConnector._socket.closeCalled, "Close was not called on the socket");
+				assert.equal(testConnector._socket.sentMessages.length, 0);
+			});
+			it("should subtract the bytes of acknowledged messages", function() {
+				const messages = [textEvent(1, "\u00e9"), textEvent(2, "\u20ac"), textEvent(3, "\u{1f600}")];
+				clearBuffer(testConnector);
+				for (const event of messages) {
+					testConnector.send(event);
+				}
+				assert.equal(testConnector._sendBufferSize, messages.reduce((sum, event) => sum + utf8Size(event), 0));
+				testConnector._dropSendBufferSeq(2);
+				assert.equal(testConnector._sendBuffer.length, 1);
+				assert.equal(testConnector._sendBufferSize, utf8Size(messages[2]));
+				testConnector._dropSendBufferSeq(3);
+				assert.equal(testConnector._sendBuffer.length, 0);
+				assert.equal(testConnector._sendBufferSize, 0);
+			});
+			it("should clear the byte count when the session is invalidated or reset", function() {
+				const connector = new lib.WebSocketClientConnector("url", 1);
+				connector._socket = new mock.MockSocket();
+				clearBuffer(connector);
+				connector.send(textEvent(1, "\u20ac"));
+				connector._invalidate();
+				assert.equal(connector._sendBuffer.length, 0);
+				assert.equal(connector._sendBufferSize, 0);
+
+				clearBuffer(connector);
+				connector.send(textEvent(1, "\u20ac"));
+				connector._socket = null;
+				connector._reset();
+				assert.equal(connector._sendBuffer.length, 0);
+				assert.equal(connector._sendBufferSize, 0);
+			});
 			it("should keep unacknowledged messages under the limit", function() {
 				testConnector._state = "connected";
 				testConnector._sendBuffer = [];