worker: use special message as MessagePort close command

When a `MessagePort` connected to another `MessagePort` closes, the
latter `MessagePort` will be closed as well. Until now, this is done
by testing whether the ports are still entangled after processing
messages. This leaves open a race condition window in which messages
sent just before the closure can be lost when timing is unfortunate.
(A description of the timing is in the test file.)

This can be addressed by using a special message instead, which is
the last message received by a `MessagePort`. This way, all previously
sent messages are processed first.

Fixes: https://github.com/nodejs/node/issues/22762
PR-URL: https://github.com/nodejs/node/pull/27705
Reviewed-By: Rich Trott <rtrott@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>

1 files changed, 38 insertions, 0 deletions

diff --git a/test/parallel/test-worker-message-port-message-before-close.js b/test/parallel/test-worker-message-port-message-before-close.js
new file mode 100644
index 0000000000..ecaad9c876
--- /dev/null
+++ b/test/parallel/test-worker-message-port-message-before-close.js
@@ -0,0 +1,38 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const { once } = require('events');
+const { Worker, MessageChannel } = require('worker_threads');
+
+// This is a regression test for the race condition underlying
+// https://github.com/nodejs/node/issues/22762.
+// It ensures that all messages send before a MessagePort#close() call are
+// received. Previously, what could happen was a race condition like this:
+// - Thread 1 sends message A
+// - Thread 2 begins receiving/emitting message A
+// - Thread 1 sends message B
+// - Thread 1 closes its side of the channel
+// - Thread 2 finishes receiving/emitting message A
+// - Thread 2 sees that the port should be closed
+// - Thread 2 closes the port, discarding message B in the process.
+
+async function test() {
+  const worker = new Worker(`
+  require('worker_threads').parentPort.on('message', ({ port }) => {
+    port.postMessage('firstMessage');
+    port.postMessage('lastMessage');
+    port.close();
+  });
+  `, { eval: true });
+
+  for (let i = 0; i < 10000; i++) {
+    const { port1, port2 } = new MessageChannel();
+    worker.postMessage({ port: port2 }, [ port2 ]);
+    await once(port1, 'message');  // 'complexObject'
+    assert.deepStrictEqual(await once(port1, 'message'), ['lastMessage']);
+  }
+
+  worker.terminate();
+}
+
+test().then(common.mustCall());