worker: fix interaction of terminate() with messaging port

When a Worker is terminated, its own handle and the public
`MessagePort` are `.ref()`’ed, so that all relevant events,
including the `'exit'` events, end up being received.

However, this is problematic if messages end up being queued
from the Worker between the beginning of the `.terminate()` call
and its completion, and there are no `'message'` event handlers
present at that time. In that situation, currently the messages
would not end up being processed, and since the MessagePort
is still `.ref()`’ed, it would keep the event loop alive
indefinitely.

To fix this:

- Make sure that all messages end up being received by
  `drainMessagePort()`, including cases in which the port had
  been stopped (i.e. there are no `'message'` listeners) and
  cases in which we exceed the limit for messages being processed
  in one batch.
- Unref the Worker’s internal ports manually after the Worker
  has exited.

Either of these solutions should be solving this on its own,
but I think it makes sense to make sure that both of them
happen during cleanup.

PR-URL: https://github.com/nodejs/node/pull/37319
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

1 files changed, 7 insertions, 2 deletions

diff --git a/src/node_messaging.h b/src/node_messaging.h
index ad065659772..2e63b22e4ce 100644
--- a/src/node_messaging.h
+++ b/src/node_messaging.h
@@ -285,11 +285,16 @@ class MessagePort : public HandleWrap {
   SET_SELF_SIZE(MessagePort)
 
  private:
+  enum class MessageProcessingMode {
+    kNormalOperation,
+    kForceReadMessages
+  };
+
   void OnClose() override;
-  void OnMessage();
+  void OnMessage(MessageProcessingMode mode);
   void TriggerAsync();
   v8::MaybeLocal<v8::Value> ReceiveMessage(v8::Local<v8::Context> context,
-                                           bool only_if_receiving);
+                                           MessageProcessingMode mode);
 
   std::unique_ptr<MessagePortData> data_ = nullptr;
   bool receiving_messages_ = false;