fix(websocket): apply WebSocketStream receive backpressure

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 .

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