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

runtime/js/99_main.js

@@ -279,7 +279,10 @@ function postMessage(message, transferOrOptions = {}) {
 
 let isClosing = false;
 let globalDispatchEvent;
-let closeOnIdle;
+
+function hasMessageEventListener() {
+  return event.listenerCount(globalThis, "message") > 0;
+}
 
 async function pollForMessages() {
   if (!globalDispatchEvent) {
@@ -289,14 +292,7 @@ async function pollForMessages() {
     );
   }
   while (!isClosing) {
-    const op = op_worker_recv_message();
-    // 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) {
-      core.unrefOpPromise(op);
-    }
-    const data = await op;
+    const data = await op_worker_recv_message();
     if (data === null) break;
     const v = messagePort.deserializeJsMessageData(data);
     const message = v[0];
@@ -813,7 +809,6 @@ function bootstrapWorkerRuntime(
     7: shouldDisableDeprecatedApiWarning,
     8: shouldUseVerboseDeprecatedApiWarning,
     9: _future,
-    10: closeOnIdle_,
   } = runtimeOptions;
 
   deprecatedApiWarningDisabled = shouldDisableDeprecatedApiWarning;
@@ -875,8 +870,8 @@ function bootstrapWorkerRuntime(
 
   location.setLocationHref(location_);
 
-  closeOnIdle = closeOnIdle_;
   globalThis.pollForMessages = pollForMessages;
+  globalThis.hasMessageEventListener = hasMessageEventListener;
 
   // TODO(bartlomieju): deprecate --unstable
   if (unstableFlag) {

runtime/web_worker.rs

@@ -339,6 +339,7 @@ pub struct WebWorker {
   pub worker_type: WebWorkerType,
   pub main_module: ModuleSpecifier,
   poll_for_messages_fn: Option<v8::Global<v8::Value>>,
+  has_message_event_listener_fn: Option<v8::Global<v8::Value>>,
   bootstrap_fn_global: Option<v8::Global<v8::Function>>,
   // Consumed when `bootstrap_fn` is called
   maybe_worker_metadata: Option<JsMessageData>,
@@ -609,6 +610,7 @@ impl WebWorker {
         worker_type: options.worker_type,
         main_module,
         poll_for_messages_fn: None,
+        has_message_event_listener_fn: None,
         bootstrap_fn_global: Some(bootstrap_fn_global),
         close_on_idle: options.close_on_idle,
         maybe_worker_metadata: options.maybe_worker_metadata,
@@ -662,6 +664,21 @@ impl WebWorker {
       .execute_script(located_script_name!(), script)
       .expect("Failed to execute worker bootstrap script");
     self.poll_for_messages_fn = Some(poll_for_messages_fn);
+    // TODO(bartlomieju): this could be done using V8 API, without calling `execute_script`.
+    // Save a reference to function that will start polling for messages
+    // from a worker host; it will be called after the user code is loaded.
+    let script = ascii_str!(
+      r#"
+    const hasMessageEventListener = globalThis.hasMessageEventListener;
+    delete globalThis.hasMessageEventListener;
+    hasMessageEventListener
+    "#
+    );
+    let has_message_event_listener_fn = self
+      .js_runtime
+      .execute_script(located_script_name!(), script)
+      .expect("Failed to execute worker bootstrap script");
+    self.has_message_event_listener_fn = Some(has_message_event_listener_fn);
   }
 
   /// See [JsRuntime::execute_script](deno_core::JsRuntime::execute_script)
@@ -781,7 +798,13 @@ impl WebWorker {
           Poll::Ready(Ok(()))
         }
       }
-      Poll::Pending => Poll::Pending,
+      Poll::Pending => {
+        if self.close_on_idle && !self.has_message_event_listener() {
+          Poll::Ready(Ok(()))
+        } else {
+          Poll::Pending
+        }
+      }
     }
   }
 
@@ -803,6 +826,22 @@ impl WebWorker {
     // This call may return `None` if worker is terminated.
     fn_.call(scope, undefined.into(), &[]);
   }
+
+  fn has_message_event_listener(&mut self) -> bool {
+    let has_message_event_listener_fn =
+      self.has_message_event_listener_fn.as_ref().unwrap();
+    let scope = &mut self.js_runtime.handle_scope();
+    let has_message_event_listener =
+      v8::Local::<v8::Value>::new(scope, has_message_event_listener_fn);
+    let fn_ =
+      v8::Local::<v8::Function>::try_from(has_message_event_listener).unwrap();
+    let undefined = v8::undefined(scope);
+    // This call may return `None` if worker is terminated.
+    match fn_.call(scope, undefined.into(), &[]) {
+      Some(result) => result.is_true(),
+      None => false,
+    }
+  }
 }
 
 fn print_worker_error(

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.");
+  }, 500);
 } 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!