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, 7 insertions, 9 deletions

diff --git a/src/node_messaging.h b/src/node_messaging.h
index 0a729c1410..08a6798e3c 100644
--- a/src/node_messaging.h
+++ b/src/node_messaging.h
@@ -17,6 +17,9 @@ class MessagePort;
 // Represents a single communication message.
 class Message : public MemoryRetainer {
  public:
+  // Create a Message with a specific underlying payload, in the format of the
+  // V8 ValueSerializer API. If `payload` is empty, this message indicates
+  // that the receiving message port should close itself.
   explicit Message(MallocedBuffer<char>&& payload = MallocedBuffer<char>());
 
   Message(Message&& other) = default;
@@ -24,6 +27,10 @@ class Message : public MemoryRetainer {
   Message& operator=(const Message&) = delete;
   Message(const Message&) = delete;
 
+  // Whether this is a message indicating that the port is to be closed.
+  // This is the last message to be received by a MessagePort.
+  bool IsCloseMessage() const;
+
   // Deserialize the contained JS value. May only be called once, and only
   // after Serialize() has been called (e.g. by another thread).
   v8::MaybeLocal<v8::Value> Deserialize(Environment* env,
@@ -89,10 +96,6 @@ class MessagePortData : public MemoryRetainer {
   // This may be called from any thread.
   void AddToIncomingQueue(Message&& message);
 
-  // Returns true if and only this MessagePort is currently not entangled
-  // with another message port.
-  bool IsSiblingClosed() const;
-
   // Turns `a` and `b` into siblings, i.e. connects the sending side of one
   // to the receiving side of the other. This is not thread-safe.
   static void Entangle(MessagePortData* a, MessagePortData* b);
@@ -109,10 +112,6 @@ class MessagePortData : public MemoryRetainer {
   SET_SELF_SIZE(MessagePortData)
 
  private:
-  // After disentangling this message port, the owner handle (if any)
-  // is asynchronously triggered, so that it can close down naturally.
-  void PingOwnerAfterDisentanglement();
-
   // This mutex protects all fields below it, with the exception of
   // sibling_.
   mutable Mutex mutex_;
@@ -178,7 +177,6 @@ class MessagePort : public HandleWrap {
   // messages.
   std::unique_ptr<MessagePortData> Detach();
 
-  bool IsSiblingClosed() const;
   void Close(
       v8::Local<v8::Value> close_callback = v8::Local<v8::Value>()) override;