fix(websocket): remove WebSocketStream handshake abort listener after the handshake

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)

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)
+})