fix(websocket): apply WebSocketStream receive backpressure

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)
 

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.#handler.socket.resume(),
-    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.
-    // TODO
+    // Bytes are not read until the readable stream has demand.
 
     // 8.  If options [" signal "] exists ,
     if (options.signal != null) {
@@ -195,6 +191,29 @@ class WebSocketStream {
 
     // 3. Close the WebSocket with this , code , and reason .
     closeWebSocketConnection(this.#handler, code, reason, true)
+    this.#readFromSocket()
+  }
+
+  #readFromSocket () {
+    const socket = this.#handler.socket
+
+    if (socket == null || this.#parser == null || this.#parser.writableNeedDrain) {
+      return
+    }
+
+    while (
+      this.#handler.readyState === states.CLOSING ||
+      (
+        this.#handler.readyState === states.OPEN &&
+        this.#readableStreamController?.desiredSize > 0
+      )
+    ) {
+      const chunk = socket.read()
+
+      if (chunk === null || !this.#parser.write(chunk)) {
+        return
+      }
+    }
   }
 
   #write (chunk) {
@@ -291,6 +310,7 @@ class WebSocketStream {
       start: (controller) => {
         this.#readableStreamController = controller
       },
+      pull: () => this.#readFromSocket(),
       cancel: (reason) => this.#cancel(reason)
     })
 
@@ -301,7 +321,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.#readFromSocket()
+      },
       abort: (reason) => this.#closeUsingReason(reason)
     })
 
@@ -350,6 +373,7 @@ class WebSocketStream {
     this.#readableStreamController.enqueue(chunk)
 
     // 4. Apply backpressure to the WebSocket.
+    // The next socket read is gated on the controller's desired size.
   }
 
   /** @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.#readFromSocket()
   }
 
   //  To cancel a WebSocketStream stream given reason , close using reason giving stream and reason .

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
+      }
+    }
   }
 
   /**

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
             }
           ]
         },