fix(ext/node): worker_threads doesn't exit if there are message listeners

runtime/js/99_main.js

@@ -293,7 +293,7 @@ async function pollForMessages() {
     // In a Node.js worker, unref() the op promise to prevent it from
     // keeping the event loop alive. This avoids the need to explicitly
     // call self.close() or worker.terminate().
-    if (closeOnIdle) {
+    if (closeOnIdle && event.listenerCount(globalThis, "message") === 0) {
       core.unrefOpPromise(op);
     }
     const data = await op;

runtime/web_worker.rs

@@ -20,6 +20,7 @@ use deno_core::futures::channel::mpsc;
 use deno_core::futures::future::poll_fn;
 use deno_core::futures::stream::StreamExt;
 use deno_core::futures::task::AtomicWaker;
+use deno_core::futures::FutureExt;
 use deno_core::located_script_name;
 use deno_core::serde::Deserialize;
 use deno_core::serde::Serialize;
@@ -53,6 +54,8 @@ use deno_web::MessagePort;
 use log::debug;
 use std::cell::RefCell;
 use std::fmt;
+use std::future::Future;
+use std::pin::Pin;
 use std::rc::Rc;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::AtomicU32;
@@ -718,11 +721,20 @@ impl WebWorker {
   /// Loads, instantiates and executes specified JavaScript module.
   ///
   /// This module will have "import.meta.main" equal to true.
-  pub async fn execute_main_module(
+  ///
+  /// The returned future must be run concurrently with the event loop
+  /// after starting to poll for messages. Use `wait_for_main_module_evaluation`.
+  fn evaluate_main_module(
     &mut self,
     id: ModuleId,
+  ) -> Pin<Box<dyn Future<Output = Result<(), AnyError>>>> {
+    self.js_runtime.mod_evaluate(id).boxed_local()
+  }
+
+  async fn wait_for_main_module_evaluation(
+    &mut self,
+    mut receiver: Pin<Box<dyn Future<Output = Result<(), AnyError>>>>,
   ) -> Result<(), AnyError> {
-    let mut receiver = self.js_runtime.mod_evaluate(id);
     let poll_options = PollEventLoopOptions::default();
 
     tokio::select! {
@@ -851,8 +863,9 @@ pub fn run_web_worker(
       // script instead of module
       match worker.preload_main_module(&specifier).await {
         Ok(id) => {
+          let fut = worker.evaluate_main_module(id);
           worker.start_polling_for_messages();
-          worker.execute_main_module(id).await
+          worker.wait_for_main_module_evaluation(fut).await
         }
         Err(e) => Err(e),
       }

tests/integration/worker_tests.rs

@@ -119,6 +119,7 @@ itest!(worker_ids_are_sequential {
 });
 
 // Test for https://github.com/denoland/deno/issues/22629
+// Test for https://github.com/denoland/deno/issues/22934
 itest!(node_worker_auto_exits {
   args: "run --quiet --allow-read workers/node_worker_auto_exits.mjs",
   output: "workers/node_worker_auto_exits.mjs.out",

tests/testdata/workers/node_worker_auto_exits.mjs

@@ -1,9 +1,19 @@
-import { isMainThread, Worker } from "node:worker_threads";
+import { isMainThread, parentPort, Worker } from "node:worker_threads";
+
+function onMessageOneshot() {
+  console.log("Got message from main thread!");
+  parentPort.off("message", onMessageOneshot);
+}
 
 if (isMainThread) {
   // This re-loads the current file inside a Worker instance.
   const w = new Worker(import.meta.filename);
+
+  setTimeout(() => {
+    w.postMessage("Hello! I am from the main thread.");
+  }, 1000);
 } else {
   console.log("Inside Worker!");
   console.log(isMainThread); // Prints 'false'.
+  parentPort.on("message", onMessageOneshot);
 }

tests/testdata/workers/node_worker_auto_exits.mjs.out

@@ -1,2 +1,3 @@
 Inside Worker!
 false
+Got message from main thread!