diff --git a/lib/web/websocket/stream/websocketstream.js b/lib/web/websocket/stream/websocketstream.js index 383fd0bf7aa..83853df9323 100644 --- a/lib/web/websocket/stream/websocketstream.js +++ b/lib/web/websocket/stream/websocketstream.js @@ -41,6 +41,12 @@ class WebSocketStream { // Each WebSocketStream object has an associated boolean handshake aborted , which is initially false. #handshakeAborted = false + // Disposable returned by addAbortListener during the opening handshake. Kept + // so the abort listener can be removed once the handshake concludes, otherwise + // the (potentially long-lived) signal retains a reference to this + // WebSocketStream and the listener leaks. + #handshakeAbortListener = null + /** @type {import('../websocket').Handler} */ #handler = { // https://whatpr.org/websockets/48/7b748d3...d5570f3.html#feedback-to-websocket-stream-from-the-protocol @@ -133,7 +139,7 @@ class WebSocketStream { } // 8.3. Add the following abort steps to signal : - addAbortListener(signal, () => { + this.#handshakeAbortListener = addAbortListener(signal, () => { // 8.3.1. If the WebSocket connection is not yet established : [WSP] if (!isEstablished(this.#handler.readyState)) { // 8.3.1.1. Fail the WebSocket connection . @@ -149,6 +155,10 @@ class WebSocketStream { // Set this 's handshake aborted to true. this.#handshakeAborted = true } + + // The abort steps have run; drop the listener so the signal no longer + // references this WebSocketStream. + this.#removeHandshakeAbortListener() }) } @@ -254,8 +264,18 @@ class WebSocketStream { return promise.promise } + #removeHandshakeAbortListener () { + if (this.#handshakeAbortListener !== null) { + this.#handshakeAbortListener[Symbol.dispose]() + this.#handshakeAbortListener = null + } + } + /** @type {import('../websocket').Handler['onConnectionEstablished']} */ #onConnectionEstablished (response, parsedExtensions) { + // The handshake succeeded; the abort listener is now a no-op, so remove it. + this.#removeHandshakeAbortListener() + this.#handler.socket = response.socket // Get options from dispatcher options @@ -354,6 +374,9 @@ class WebSocketStream { /** @type {import('../websocket').Handler['onSocketClose']} */ #onSocketClose () { + // The connection is gone; ensure the handshake abort listener is removed. + this.#removeHandshakeAbortListener() + const wasClean = this.#handler.closeState.has(sentCloseFrameState.SENT) && this.#handler.closeState.has(sentCloseFrameState.RECEIVED) diff --git a/test/websocket/stream/abort-listener-leak.js b/test/websocket/stream/abort-listener-leak.js new file mode 100644 index 00000000000..2a905e62393 --- /dev/null +++ b/test/websocket/stream/abort-listener-leak.js @@ -0,0 +1,56 @@ +'use strict' + +const { test } = require('node:test') +const { getEventListeners } = require('node:events') +const { WebSocketServer } = require('ws') +const { WebSocketStream } = require('../../..') + +// A WebSocketStream created with a signal must remove its abort listener once +// the opening handshake concludes, otherwise a long-lived signal keeps the +// closed WebSocketStream (and its listener) alive. See the abort steps in +// https://websockets.spec.whatwg.org/#dom-websocketstream-websocketstream +test('WebSocketStream removes its abort listener after a clean close', async (t) => { + const server = new WebSocketServer({ port: 0 }) + t.after(() => server.close()) + + const controller = new AbortController() + const wss = new WebSocketStream(`ws://localhost:${server.address().port}`, { + signal: controller.signal + }) + + const { writable } = await wss.opened + + // Listener must already be gone once the handshake has succeeded. + t.assert.strictEqual(getEventListeners(controller.signal, 'abort').length, 0) + + const writer = writable.getWriter() + await writer.close() + await Promise.allSettled([wss.closed]) + + t.assert.strictEqual(getEventListeners(controller.signal, 'abort').length, 0) +}) + +test('WebSocketStream removes its abort listener after an aborted handshake', async (t) => { + const sockets = new Set() + const server = new WebSocketServer({ port: 0 }) + server.on('connection', (ws) => { + sockets.add(ws) + ws.on('close', () => sockets.delete(ws)) + }) + + t.after(() => { + for (const ws of sockets) ws.terminate() + server.close() + }) + + const controller = new AbortController() + const wss = new WebSocketStream(`ws://localhost:${server.address().port}`, { + signal: controller.signal + }) + + controller.abort(new Error('abort before open')) + + await Promise.allSettled([wss.opened, wss.closed]) + + t.assert.strictEqual(getEventListeners(controller.signal, 'abort').length, 0) +})