Repository: websockets/ws (js)

Issue:
Dispatched events are not scheduled as tasks
### Is there an existing issue for this?

- [X] I've searched for any related issues and avoided creating a duplicate issue.

### Description

I am not sure this is a bug or rather a topic for discussion. I have noticed that there is a subtle difference between `ws` and the WebSockets standard. The standard says that when a message is received the agent has [to queue a new task](https://websockets.spec.whatwg.org/#feedback-from-the-protocol:~:text=When%20a%20WebSocket%20message%20has%20been%20received%20with%20type%20type%20and%20data%20data%2C%20the%20user%20agent%20must%20queue%20a%20task%20to%20follow%20these%20steps%3A%20%5BWSP%5D). This has an important side effect that the microtask queue will be processed after the task is over and before the next task is picked from the queue. This behaviour is important in the cases where the messages need to be processed in a specified order.

This can be demonstrated by the following server code:

```js
import { WebSocketServer } from 'ws';

const wss = new WebSocketServer({ port: 8080 });

wss.on('connection', function connection(ws) {
  ws.send('something1');
  ws.send('something2');
});
```

And the client code:

```js
import { WebSocket } from 'ws'; // in the browser, omit this line.

const ws = new WebSocket('ws://127.0.0.1:8080');

ws.addEventListener('message', function message(event) {
  console.log('received: %s', event.data);
  new Promise(resolve => {
    resolve()
  }).then(() => {
    console.log('microtask', event.data.toString())
  })
});
```

If you run this code in Node and in the browser, you should observe that the order of console messages is different. It's probably not feasible to change the behaviour of 'ws' as it could be a breaking change but I wonder if there could be some compatibility mode for this? I believe it's possible to workaround that using setTimeout or setImmediate but the downside is that the same code cannot be used across platforms. If I have overlooked an existing issue about this topic, I apologize. 

### ws version

8.13.0

### Node.js Version

v20.3.1

### System

  System:
    OS: macOS 13.5
    CPU: (10) arm64 Apple M1 Max
    Memory: 696.63 MB / 64.00 GB
    Shell: 5.9 - /bin/zsh

### Expected result

The order of console messages is the same in Node and in the browser and matches the behaviour defined by the standard:

```
received: something1
microtask something1
received: something2
microtask something2 
```

### Actual result

The order of console messages in Node is different (microtasks are processed after all messages):

```
received: something1
received: something2
microtask something1
microtask something2
```

### Attachments

_No response_

Pull request:
Emit at most one event per microtask
To improve compatibility with the WHATWG standard, emit at most one of `'message'`, `'ping'`, and `'pong'` events per tick.

Fixes #2159

Patch:
diff --git a/lib/receiver.js b/lib/receiver.js
index 96f572cb..b5e9a8bc 100644
--- a/lib/receiver.js
+++ b/lib/receiver.js
@@ -13,12 +13,15 @@ const { concat, toArrayBuffer, unmask } = require('./buffer-util');
 const { isValidStatusCode, isValidUTF8 } = require('./validation');
 
 const FastBuffer = Buffer[Symbol.species];
+const promise = Promise.resolve();
+
 const GET_INFO = 0;
 const GET_PAYLOAD_LENGTH_16 = 1;
 const GET_PAYLOAD_LENGTH_64 = 2;
 const GET_MASK = 3;
 const GET_DATA = 4;
 const INFLATING = 5;
+const WAIT_MICROTASK = 6;
 
 /**
  * HyBi Receiver implementation.
@@ -157,9 +160,23 @@ class Receiver extends Writable {
         case GET_DATA:
           err = this.getData(cb);
           break;
+        case INFLATING:
+          this._loop = false;
+          return;
         default:
-          // `INFLATING`
+          //
+          // `WAIT_MICROTASK`.
+          //
           this._loop = false;
+
+          //
+          // `queueMicrotask()` is not available in Node.js < 11 and is no
+          // better anyway.
+          //
+          promise.then(() => {
+            this._state = GET_INFO;
+            this.startLoop(cb);
+          });
           return;
       }
     } while (this._loop);
@@ -542,7 +559,7 @@ class Receiver extends Writable {
       }
     }
 
-    this._state = GET_INFO;
+    this._state = WAIT_MICROTASK;
   }
 
   /**
@@ -559,6 +576,8 @@ class Receiver extends Writable {
       if (data.length === 0) {
         this.emit('conclude', 1005, EMPTY_BUFFER);
         this.end();
+
+        this._state = GET_INFO;
       } else {
         const code = data.readUInt16BE(0);
 
@@ -590,14 +609,16 @@ class Receiver extends Writable {
 
         this.emit('conclude', code, buf);
         this.end();
+
+        this._state = GET_INFO;
       }
     } else if (this._opcode === 0x09) {
       this.emit('ping', data);
+      this._state = WAIT_MICROTASK;
     } else {
       this.emit('pong', data);
+      this._state = WAIT_MICROTASK;
     }
-
-    this._state = GET_INFO;
   }
 }
 


Test patch:
diff --git a/test/receiver.test.js b/test/receiver.test.js
index 4ae27946..a4e1bb5a 100644
--- a/test/receiver.test.js
+++ b/test/receiver.test.js
@@ -1083,4 +1083,96 @@ describe('Receiver', () => {
 
     receiver.write(Buffer.from([0x88, 0x03, 0x03, 0xe8, 0xf8]));
   });
+
+  it("waits a microtask after each 'message' event", (done) => {
+    const messages = [];
+    const receiver = new Receiver();
+
+    receiver.on('message', (data, isBinary) => {
+      assert.ok(!isBinary);
+
+      const message = data.toString();
+      messages.push(message);
+
+      // `queueMicrotask()` is not available in Node.js < 11.
+      Promise.resolve().then(() => {
+        messages.push(`microtask ${message}`);
+
+        if (messages.length === 6) {
+          assert.deepStrictEqual(messages, [
+            '1',
+            'microtask 1',
+            '2',
+            'microtask 2',
+            '3',
+            'microtask 3'
+          ]);
+
+          done();
+        }
+      });
+    });
+
+    receiver.write(Buffer.from('810131810132810133', 'hex'));
+  });
+
+  it("waits a microtask after each 'ping' event", (done) => {
+    const actual = [];
+    const receiver = new Receiver();
+
+    receiver.on('ping', (data) => {
+      const message = data.toString();
+      actual.push(message);
+
+      // `queueMicrotask()` is not available in Node.js < 11.
+      Promise.resolve().then(() => {
+        actual.push(`microtask ${message}`);
+
+        if (actual.length === 6) {
+          assert.deepStrictEqual(actual, [
+            '1',
+            'microtask 1',
+            '2',
+            'microtask 2',
+            '3',
+            'microtask 3'
+          ]);
+
+          done();
+        }
+      });
+    });
+
+    receiver.write(Buffer.from('890131890132890133', 'hex'));
+  });
+
+  it("waits a microtask after each 'pong' event", (done) => {
+    const actual = [];
+    const receiver = new Receiver();
+
+    receiver.on('pong', (data) => {
+      const message = data.toString();
+      actual.push(message);
+
+      // `queueMicrotask()` is not available in Node.js < 11.
+      Promise.resolve().then(() => {
+        actual.push(`microtask ${message}`);
+
+        if (actual.length === 6) {
+          assert.deepStrictEqual(actual, [
+            '1',
+            'microtask 1',
+            '2',
+            'microtask 2',
+            '3',
+            'microtask 3'
+          ]);
+
+          done();
+        }
+      });
+    });
+
+    receiver.write(Buffer.from('8A01318A01328A0133', 'hex'));
+  });
 });
diff --git a/test/websocket.test.js b/test/websocket.test.js
index 487b02ac..4d45fad4 100644
--- a/test/websocket.test.js
+++ b/test/websocket.test.js
@@ -4075,18 +4075,18 @@ describe('WebSocket', () => {
             const messages = [];
             const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
 
-            ws.on('open', () => {
-              ws._socket.on('end', () => {
-                assert.strictEqual(ws._receiver._state, 5);
-              });
-            });
-
             ws.on('message', (message, isBinary) => {
               assert.ok(!isBinary);
 
               if (messages.push(message.toString()) > 1) return;
 
-              ws.close(1000);
+              // `queueMicrotask()` is not available in Node.js < 11.
+              Promise.resolve().then(() => {
+                process.nextTick(() => {
+                  assert.strictEqual(ws._receiver._state, 5);
+                  ws.close(1000);
+                });
+              });
             });
 
             ws.on('close', (code, reason) => {
@@ -4331,9 +4331,12 @@ describe('WebSocket', () => {
 
               if (messages.push(message.toString()) > 1) return;
 
-              process.nextTick(() => {
-                assert.strictEqual(ws._receiver._state, 5);
-                ws.terminate();
+              // `queueMicrotask()` is not available in Node.js < 11.
+              Promise.resolve().then(() => {
+                process.nextTick(() => {
+                  assert.strictEqual(ws._receiver._state, 5);
+                  ws.terminate();
+                });
               });
             });