From 4a982818ab3073c2f3645c262d0591769aff7e44 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 15 Jul 2026 10:21:35 +0000 Subject: [PATCH 1/2] fix(websocket): apply WebSocketStream receive backpressure Signed-off-by: Matteo Collina --- lib/web/websocket/stream/websocketstream.js | 31 +++++++++++++++++++-- test/web-platform-tests/expectation.json | 3 +- 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/lib/web/websocket/stream/websocketstream.js b/lib/web/websocket/stream/websocketstream.js index 383fd0bf7aa..8f135ba3517 100644 --- a/lib/web/websocket/stream/websocketstream.js +++ b/lib/web/websocket/stream/websocketstream.js @@ -47,7 +47,7 @@ class WebSocketStream { onConnectionEstablished: (response, extensions) => this.#onConnectionEstablished(response, extensions), onMessage: (opcode, data) => this.#onMessage(opcode, data), onParserError: (err) => failWebsocketConnection(this.#handler, null, err.message), - onParserDrain: () => this.#handler.socket.resume(), + onParserDrain: () => this.#resumeSocketIfNeeded(), onSocketData: (chunk) => { if (!this.#parser.write(chunk)) { this.#handler.socket.pause() @@ -117,7 +117,7 @@ class WebSocketStream { this.#closedPromise = Promise.withResolvers() // 7. Apply backpressure to the WebSocket. - // TODO + // This is done when the socket becomes available. // 8. If options [" signal "] exists , if (options.signal != null) { @@ -195,6 +195,22 @@ class WebSocketStream { // 3. Close the WebSocket with this , code , and reason . closeWebSocketConnection(this.#handler, code, reason, true) + this.#resumeSocketIfNeeded() + } + + #resumeSocketIfNeeded () { + const socket = this.#handler.socket + + if (socket == null) { + return + } + + if ( + this.#handler.readyState === states.CLOSING || + (this.#readableStreamController?.desiredSize > 0 && !this.#parser.writableNeedDrain) + ) { + socket.resume() + } } #write (chunk) { @@ -257,6 +273,7 @@ class WebSocketStream { /** @type {import('../websocket').Handler['onConnectionEstablished']} */ #onConnectionEstablished (response, parsedExtensions) { this.#handler.socket = response.socket + this.#handler.socket.pause() // Get options from dispatcher options const maxFragments = this.#handler.controller.dispatcher?.webSocketOptions?.maxFragments @@ -291,6 +308,7 @@ class WebSocketStream { start: (controller) => { this.#readableStreamController = controller }, + pull: () => this.#resumeSocketIfNeeded(), cancel: (reason) => this.#cancel(reason) }) @@ -301,7 +319,10 @@ class WebSocketStream { // 13. Set up writable with writeAlgorithm , closeAlgorithm , and abortAlgorithm . const writable = new WritableStream({ write: (chunk) => this.#write(chunk), - close: () => closeWebSocketConnection(this.#handler, null, null), + close: () => { + closeWebSocketConnection(this.#handler, null, null) + this.#resumeSocketIfNeeded() + }, abort: (reason) => this.#closeUsingReason(reason) }) @@ -350,6 +371,9 @@ class WebSocketStream { this.#readableStreamController.enqueue(chunk) // 4. Apply backpressure to the WebSocket. + if (this.#readableStreamController.desiredSize <= 0) { + this.#handler.socket.pause() + } } /** @type {import('../websocket').Handler['onSocketClose']} */ @@ -441,6 +465,7 @@ class WebSocketStream { // 4. Close the WebSocket with stream , code , and reasonString . If this throws an exception, // discard code and reasonString and close the WebSocket with stream . closeWebSocketConnection(this.#handler, code, reasonString) + this.#resumeSocketIfNeeded() } // To cancel a WebSocketStream stream given reason , close using reason giving stream and reason . diff --git a/test/web-platform-tests/expectation.json b/test/web-platform-tests/expectation.json index 001ed3ffe9e..5fe462501ff 100644 --- a/test/web-platform-tests/expectation.json +++ b/test/web-platform-tests/expectation.json @@ -39163,8 +39163,7 @@ "cases": [ { "name": "backpressure should be applied to received messages", - "success": false, - "message": "assert_greater_than_equal: data send should have taken at least 2 seconds expected a number greater than or equal to 1.8 but got 0.0826730728149414" + "success": true } ] }, From d74a456a9cd18a4e3dda260b656829e126ad1de4 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Fri, 17 Jul 2026 10:48:48 +0000 Subject: [PATCH 2/2] refactor(websocket): read upgraded sockets on demand Signed-off-by: Matteo Collina --- lib/web/websocket/connection.js | 2 +- lib/web/websocket/stream/websocketstream.js | 40 ++++++++++----------- lib/web/websocket/websocket.js | 27 +++++++++----- 3 files changed, 39 insertions(+), 30 deletions(-) diff --git a/lib/web/websocket/connection.js b/lib/web/websocket/connection.js index 4ecc8a195fc..d6f22bc2c6b 100644 --- a/lib/web/websocket/connection.js +++ b/lib/web/websocket/connection.js @@ -206,7 +206,7 @@ function establishWebSocketConnection (url, protocols, client, handler, options) } } - response.socket.on('data', handler.onSocketData) + response.socket.on('readable', handler.onSocketReadable) response.socket.on('close', handler.onSocketClose) response.socket.on('error', handler.onSocketError) diff --git a/lib/web/websocket/stream/websocketstream.js b/lib/web/websocket/stream/websocketstream.js index 8f135ba3517..5063a5b2f25 100644 --- a/lib/web/websocket/stream/websocketstream.js +++ b/lib/web/websocket/stream/websocketstream.js @@ -47,12 +47,8 @@ class WebSocketStream { onConnectionEstablished: (response, extensions) => this.#onConnectionEstablished(response, extensions), onMessage: (opcode, data) => this.#onMessage(opcode, data), onParserError: (err) => failWebsocketConnection(this.#handler, null, err.message), - onParserDrain: () => this.#resumeSocketIfNeeded(), - onSocketData: (chunk) => { - if (!this.#parser.write(chunk)) { - this.#handler.socket.pause() - } - }, + onParserDrain: () => this.#readFromSocket(), + onSocketReadable: () => this.#readFromSocket(), onSocketError: (err) => { this.#handler.readyState = states.CLOSING @@ -117,7 +113,7 @@ class WebSocketStream { this.#closedPromise = Promise.withResolvers() // 7. Apply backpressure to the WebSocket. - // This is done when the socket becomes available. + // Bytes are not read until the readable stream has demand. // 8. If options [" signal "] exists , if (options.signal != null) { @@ -195,21 +191,28 @@ class WebSocketStream { // 3. Close the WebSocket with this , code , and reason . closeWebSocketConnection(this.#handler, code, reason, true) - this.#resumeSocketIfNeeded() + this.#readFromSocket() } - #resumeSocketIfNeeded () { + #readFromSocket () { const socket = this.#handler.socket - if (socket == null) { + if (socket == null || this.#parser == null || this.#parser.writableNeedDrain) { return } - if ( + while ( this.#handler.readyState === states.CLOSING || - (this.#readableStreamController?.desiredSize > 0 && !this.#parser.writableNeedDrain) + ( + this.#handler.readyState === states.OPEN && + this.#readableStreamController?.desiredSize > 0 + ) ) { - socket.resume() + const chunk = socket.read() + + if (chunk === null || !this.#parser.write(chunk)) { + return + } } } @@ -273,7 +276,6 @@ class WebSocketStream { /** @type {import('../websocket').Handler['onConnectionEstablished']} */ #onConnectionEstablished (response, parsedExtensions) { this.#handler.socket = response.socket - this.#handler.socket.pause() // Get options from dispatcher options const maxFragments = this.#handler.controller.dispatcher?.webSocketOptions?.maxFragments @@ -308,7 +310,7 @@ class WebSocketStream { start: (controller) => { this.#readableStreamController = controller }, - pull: () => this.#resumeSocketIfNeeded(), + pull: () => this.#readFromSocket(), cancel: (reason) => this.#cancel(reason) }) @@ -321,7 +323,7 @@ class WebSocketStream { write: (chunk) => this.#write(chunk), close: () => { closeWebSocketConnection(this.#handler, null, null) - this.#resumeSocketIfNeeded() + this.#readFromSocket() }, abort: (reason) => this.#closeUsingReason(reason) }) @@ -371,9 +373,7 @@ class WebSocketStream { this.#readableStreamController.enqueue(chunk) // 4. Apply backpressure to the WebSocket. - if (this.#readableStreamController.desiredSize <= 0) { - this.#handler.socket.pause() - } + // The next socket read is gated on the controller's desired size. } /** @type {import('../websocket').Handler['onSocketClose']} */ @@ -465,7 +465,7 @@ class WebSocketStream { // 4. Close the WebSocket with stream , code , and reasonString . If this throws an exception, // discard code and reasonString and close the WebSocket with stream . closeWebSocketConnection(this.#handler, code, reasonString) - this.#resumeSocketIfNeeded() + this.#readFromSocket() } // To cancel a WebSocketStream stream given reason , close using reason giving stream and reason . diff --git a/lib/web/websocket/websocket.js b/lib/web/websocket/websocket.js index e473a1bc491..dce2e1b8c6e 100644 --- a/lib/web/websocket/websocket.js +++ b/lib/web/websocket/websocket.js @@ -43,7 +43,7 @@ function getSocketAddress (socket) { * @property {(opcode: number, data: Buffer) => void} onMessage * @property {(error: Error) => void} onParserError * @property {() => void} onParserDrain - * @property {(chunk: Buffer) => void} onSocketData + * @property {() => void} onSocketReadable * @property {(err: Error) => void} onSocketError * @property {() => void} onSocketClose * @property {(body: Buffer) => void} onPing @@ -77,12 +77,8 @@ class WebSocket extends EventTarget { onConnectionEstablished: (response, extensions) => this.#onConnectionEstablished(response, extensions), onMessage: (opcode, data) => this.#onMessage(opcode, data), onParserError: (err) => failWebsocketConnection(this.#handler, null, err.message), - onParserDrain: () => this.#onParserDrain(), - onSocketData: (chunk) => { - if (!this.#parser.write(chunk)) { - this.#handler.socket.pause() - } - }, + onParserDrain: () => this.#readFromSocket(), + onSocketReadable: () => this.#readFromSocket(), onSocketError: (err) => { this.#handler.readyState = states.CLOSING @@ -521,6 +517,8 @@ class WebSocket extends EventTarget { } }) } + + this.#readFromSocket() } #onMessage (type, data) { @@ -564,8 +562,19 @@ class WebSocket extends EventTarget { }) } - #onParserDrain () { - this.#handler.socket.resume() + #readFromSocket () { + const socket = this.#handler.socket + + if (socket == null || this.#parser == null || this.#parser.writableNeedDrain) { + return + } + + let chunk + while ((chunk = socket.read()) !== null) { + if (!this.#parser.write(chunk)) { + return + } + } } /**