Repository: websockets/ws (js)

Issue:
Messages are dispatched while microtask queue is not empty
### Is there an existing issue for this?

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

### Description

I have previously reported https://github.com/websockets/ws/issues/2159 which was fixed but it looks like the solution might not have been complete. 

With the latest version of ws the microtasks scheduled in the event callback are handled first but the microtasks created by those microtasks are not. This is different from the WebSocket client behavior in the browser.

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');
});
```

Client code:

```
import { WebSocket } from 'ws';

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())
      return Promise.resolve().then(() => {
        console.log('microtask-nested', event.data.toString());
      });
    })
});
```

### ws version

8.16.0

### Node.js Version

v20.10.0

### System

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

### Expected result

```
received: something1
microtask something1
microtask-nested something1
received: something2
microtask something2
microtask-nested something2
```

### Actual result

```
received: something1
microtask something1
received: something2
microtask-nested something1
microtask something2
microtask-nested something2
```


### Attachments

_No response_

Pull request description:
Emit at most one event per event loop iteration
Fixes #2216

Patch:
diff --git a/lib/receiver.js b/lib/receiver.js
index 9e87d811..4515e688 100644
--- a/lib/receiver.js
+++ b/lib/receiver.js
@@ -13,13 +13,6 @@ const { concat, toArrayBuffer, unmask } = require('./buffer-util');
 const { isValidStatusCode, isValidUTF8 } = require('./validation');
 
 const FastBuffer = Buffer[Symbol.species];
-const promise = Promise.resolve();
-
-//
-// `queueMicrotask()` is not available in Node.js < 11.
-//
-const queueTask =
-  typeof queueMicrotask === 'function' ? queueMicrotask : queueMicrotaskShim;
 
 const GET_INFO = 0;
 const GET_PAYLOAD_LENGTH_16 = 1;
@@ -567,17 +560,12 @@ class Receiver extends Writable {
         data = fragments;
       }
 
-      //
-      // If the state is `INFLATING`, it means that the frame data was
-      // decompressed asynchronously, so there is no need to defer the event
-      // as it will be emitted asynchronously anyway.
-      //
-      if (this._state === INFLATING || this._allowSynchronousEvents) {
+      if (this._allowSynchronousEvents) {
         this.emit('message', data, true);
         this._state = GET_INFO;
       } else {
         this._state = DEFER_EVENT;
-        queueTask(() => {
+        setImmediate(() => {
           this.emit('message', data, true);
           this._state = GET_INFO;
           this.startLoop(cb);
@@ -604,7 +592,7 @@ class Receiver extends Writable {
         this._state = GET_INFO;
       } else {
         this._state = DEFER_EVENT;
-        queueTask(() => {
+        setImmediate(() => {
           this.emit('message', buf, false);
           this._state = GET_INFO;
           this.startLoop(cb);
@@ -675,7 +663,7 @@ class Receiver extends Writable {
       this._state = GET_INFO;
     } else {
       this._state = DEFER_EVENT;
-      queueTask(() => {
+      setImmediate(() => {
         this.emit(this._opcode === 0x09 ? 'ping' : 'pong', data);
         this._state = GET_INFO;
         this.startLoop(cb);
@@ -711,32 +699,3 @@ class Receiver extends Writable {
 }
 
 module.exports = Receiver;
-
-/**
- * A shim for `queueMicrotask()`.
- *
- * @param {Function} cb Callback
- */
-function queueMicrotaskShim(cb) {
-  promise.then(cb).catch(throwErrorNextTick);
-}
-
-/**
- * Throws an error.
- *
- * @param {Error} err The error to throw
- * @private
- */
-function throwError(err) {
-  throw err;
-}
-
-/**
- * Throws an error in the next tick.
- *
- * @param {Error} err The error to throw
- * @private
- */
-function throwErrorNextTick(err) {
-  process.nextTick(throwError, err);
-}


Test patch:
diff --git a/test/receiver.test.js b/test/receiver.test.js
index a88f29b9..f3a0fa64 100644
--- a/test/receiver.test.js
+++ b/test/receiver.test.js
@@ -1085,17 +1085,21 @@ describe('Receiver', () => {
     receiver.write(Buffer.from([0x88, 0x03, 0x03, 0xe8, 0xf8]));
   });
 
-  it('emits at most one event per microtask', (done) => {
+  it('emits at most one event per event loop iteration', (done) => {
     const actual = [];
     const expected = [
       '1',
-      'microtask 1',
+      '- 1',
+      '-- 1',
       '2',
-      'microtask 2',
+      '- 2',
+      '-- 2',
       '3',
-      'microtask 3',
+      '- 3',
+      '-- 3',
       '4',
-      'microtask 4'
+      '- 4',
+      '-- 4'
     ];
 
     function listener(data) {
@@ -1104,12 +1108,16 @@ describe('Receiver', () => {
 
       // `queueMicrotask()` is not available in Node.js < 11.
       Promise.resolve().then(() => {
-        actual.push(`microtask ${message}`);
+        actual.push(`- ${message}`);
 
-        if (actual.length === 8) {
-          assert.deepStrictEqual(actual, expected);
-          done();
-        }
+        Promise.resolve().then(() => {
+          actual.push(`-- ${message}`);
+
+          if (actual.length === 12) {
+            assert.deepStrictEqual(actual, expected);
+            done();
+          }
+        });
       });
     }
 
diff --git a/test/websocket.test.js b/test/websocket.test.js
index 7cb17f0e..5570b1ca 100644
--- a/test/websocket.test.js
+++ b/test/websocket.test.js
@@ -4234,8 +4234,7 @@ describe('WebSocket', () => {
 
               if (messages.push(message.toString()) > 1) return;
 
-              // `queueMicrotask()` is not available in Node.js < 11.
-              Promise.resolve().then(() => {
+              setImmediate(() => {
                 process.nextTick(() => {
                   assert.strictEqual(ws._receiver._state, 5);
                   ws.close(1000);
@@ -4485,8 +4484,7 @@ describe('WebSocket', () => {
 
               if (messages.push(message.toString()) > 1) return;
 
-              // `queueMicrotask()` is not available in Node.js < 11.
-              Promise.resolve().then(() => {
+              setImmediate(() => {
                 process.nextTick(() => {
                   assert.strictEqual(ws._receiver._state, 5);
                   ws.terminate();