fix: wake event loop when V8 posts foreground tasks from background threads

Author: fraidev

Cargo.lock

@@ -99,7 +99,7 @@ deno_task_shell = "=0.29.0"
 deno_terminal = "=0.2.3"
 deno_unsync = { version = "0.4.4", default-features = false }
 deno_whoami = "0.1.0"
-v8 = { version = "146.3.0", default-features = false }
+v8 = { version = "146.4.0", default-features = false }
 
 denokv_proto = "0.13.0"
 denokv_remote = "0.13.0"

Cargo.toml

@@ -336,6 +336,8 @@ struct JSDuplexResource {
   readable: Arc<Mutex<tokio::sync::mpsc::Receiver<Bytes>>>,
   writable: tokio::sync::mpsc::Sender<Bytes>,
   read_buffer: Arc<Mutex<VecDeque<Bytes>>>,
+  closed: AtomicBool,
+  close_notify: tokio::sync::Notify,
 }
 
 impl JSDuplexResource {
@@ -347,6 +349,8 @@ impl JSDuplexResource {
       readable: Arc::new(Mutex::new(readable)),
       writable,
       read_buffer: Arc::new(Mutex::new(VecDeque::new())),
+      closed: AtomicBool::new(false),
+      close_notify: tokio::sync::Notify::new(),
     }
   }
 
@@ -355,6 +359,10 @@ impl JSDuplexResource {
     self: Rc<Self>,
     data: &mut [u8],
   ) -> Result<usize, std::io::Error> {
+    if self.closed.load(Ordering::Relaxed) {
+      return Ok(0);
+    }
+
     // First check if we have buffered data from previous partial read
     if let Ok(mut buffer) = self.read_buffer.lock()
       && let Some(buffered_data) = buffer.pop_front()
@@ -370,13 +378,19 @@ impl JSDuplexResource {
       return Ok(len);
     }
 
-    // No buffered data, receive new data from channel
+    // No buffered data, receive new data from channel.
+    // We use select! so that close() can wake us up via close_notify
+    // even though we hold the readable mutex across the await (the
+    // close() method uses try_lock to avoid deadlock).
     let bytes = {
       let mut receiver = self
         .readable
         .lock()
         .map_err(|_| Error::other("Failed to acquire lock"))?;
-      receiver.recv().await
+      tokio::select! {
+        result = receiver.recv() => result,
+        _ = self.close_notify.notified() => None,
+      }
     };
 
     match bytes {
@@ -394,7 +408,7 @@ impl JSDuplexResource {
         Ok(len)
       }
       None => {
-        // Channel closed
+        // Channel closed or resource closing
         Ok(0)
       }
     }
@@ -423,6 +437,28 @@ impl Resource for JSDuplexResource {
   fn name(&self) -> Cow<'_, str> {
     "JSDuplexResource".into()
   }
+
+  fn close(self: Rc<Self>) {
+    // Signal that this resource is closing.  The read() method checks
+    // this flag and the close_notify to break out of pending recv().
+    //
+    // Without this cleanup, a circular Rc dependency between
+    // JSDuplexResource and JSStreamTlsResource prevents either from
+    // being dropped, keeping the event loop alive indefinitely.
+    self.closed.store(true, Ordering::Relaxed);
+
+    // Wake up any pending read via Notify.  We use notify_one() which
+    // stores a permit if no one is currently waiting, so the next
+    // notified().await will complete immediately.
+    self.close_notify.notify_one();
+
+    // Also try to close the receiver directly.  We use try_lock()
+    // because read() holds the mutex across an await point; using
+    // lock() here would deadlock.
+    if let Ok(mut rx) = self.readable.try_lock() {
+      rx.close();
+    }
+  }
 }
 
 #[derive(FromV8)]

ext/node/ops/tls.rs

@@ -303,6 +303,8 @@ export class TLSSocket extends net.Socket {
 
 class JSStreamSocket {
   #rid;
+  #channelRid;
+  #closed = false;
 
   constructor(stream) {
     this.stream = stream;
@@ -311,7 +313,8 @@ class JSStreamSocket {
   init(options) {
     op_node_tls_start(options, tlsStreamRids);
     this.#rid = tlsStreamRids[0];
-    const channelRid = tlsStreamRids[1];
+    this.#channelRid = tlsStreamRids[1];
+    const channelRid = this.#channelRid;
 
     this.stream.on("data", (data) => {
       core.write(channelRid, data);
@@ -321,7 +324,9 @@ class JSStreamSocket {
     (async () => {
       while (true) {
         try {
-          const nread = await core.read(channelRid, buf);
+          const readPromise = core.read(channelRid, buf);
+          core.unrefOpPromise(readPromise);
+          const nread = await readPromise;
           this.stream.write(buf.slice(0, nread));
         } catch {
           break;
@@ -330,17 +335,41 @@ class JSStreamSocket {
     })();
 
     this.stream.on("close", () => {
-      core.close(this.#rid);
-      core.close(channelRid);
+      this.close();
     });
   }
 
+  // Called by stream_wrap's _onClose() via kStreamBaseField.close(),
+  // or by event listeners when the transport/DuplexPair is destroyed.
+  close() {
+    if (this.#closed) return;
+    this.#closed = true;
+    if (this.#rid !== undefined) {
+      try {
+        core.close(this.#rid);
+      } catch {
+        // already closed
+      }
+      this.#rid = undefined;
+    }
+    if (this.#channelRid !== undefined) {
+      try {
+        core.close(this.#channelRid);
+      } catch {
+        // already closed
+      }
+      this.#channelRid = undefined;
+    }
+  }
+
   handshake() {
     return op_node_tls_handshake(this.#rid);
   }
 
   read(buf) {
-    return core.read(this.#rid, buf);
+    const promise = core.read(this.#rid, buf);
+    core.unrefOpPromise(promise);
+    return promise;
   }
 
   write(data) {

ext/node/polyfills/_tls_wrap.js

@@ -386,7 +386,8 @@ export class LibuvStreamWrap extends HandleWrap {
 
       if (
         ObjectPrototypeIsPrototypeOf(Deno.errors.Interrupted.prototype, e) ||
-        ObjectPrototypeIsPrototypeOf(Deno.errors.BadResource.prototype, e)
+        ObjectPrototypeIsPrototypeOf(Deno.errors.BadResource.prototype, e) ||
+        ObjectPrototypeIsPrototypeOf(Deno.errors.UnexpectedEof.prototype, e)
       ) {
         nread = MapPrototypeGet(codeMap, "EOF")!;
       } else if (

ext/node/polyfills/internal_binding/stream_wrap.ts

@@ -180,6 +180,10 @@ impl InnerIsolateState {
     self.main_realm.0.context_state.pending_ops.shutdown();
     let inspector = self.state.inspector.take();
 
+    // Unregister isolate waker before dropping the isolate
+    let isolate_ptr = unsafe { self.v8_isolate.as_raw_isolate_ptr() };
+    setup::unregister_isolate_waker(setup::isolate_ptr_to_key(isolate_ptr));
+
     let state_ptr = self.v8_isolate.get_data(STATE_DATA_OFFSET);
     // SAFETY: We are sure that it's a valid pointer for whole lifetime of
     // the runtime.
@@ -450,6 +454,8 @@ pub struct JsRuntimeState {
   pub(crate) function_templates: Rc<RefCell<FunctionTemplateData>>,
   pub(crate) callsite_prototype: RefCell<Option<v8::Global<v8::Object>>>,
   waker: Arc<AtomicWaker>,
+  /// Guards the 100ms safety-net timer so at most one is active at a time.
+  safety_net_active: Arc<std::sync::atomic::AtomicBool>,
   /// Accessed through [`JsRuntimeState::with_inspector`].
   inspector: RefCell<Option<Rc<JsRuntimeInspector>>>,
   has_inspector: Cell<bool>,
@@ -776,7 +782,8 @@ impl JsRuntime {
       eval_context_code_cache_ready_cb: RefCell::new(
         eval_context_set_code_cache_cb,
       ),
-      waker,
+      waker: waker.clone(),
+      safety_net_active: Arc::new(std::sync::atomic::AtomicBool::new(false)),
       // Some fields are initialized later after isolate is created
       inspector: None.into(),
       has_inspector: false.into(),
@@ -852,6 +859,14 @@ impl JsRuntime {
     );
 
     let isolate_ptr = unsafe { isolate.as_raw_isolate_ptr() };
+
+    // Register this isolate's waker so the custom platform can wake
+    // the event loop when V8 posts foreground tasks from background threads.
+    setup::register_isolate_waker(
+      setup::isolate_ptr_to_key(isolate_ptr),
+      waker.clone(),
+    );
+
     // ...isolate is fully set up, we can forward its pointer to the ops to finish
     // their' setup...
     for op_ctx in op_ctxs.iter_mut() {
@@ -2268,6 +2283,27 @@ impl JsRuntime {
       scope.perform_microtask_checkpoint();
     }
 
+    // Safety net: if V8 has pending background tasks (e.g. module compilation),
+    // schedule a delayed wake to pump the message loop in case the platform
+    // callback was missed due to a race condition. Uses an OS thread (not
+    // tokio) to avoid depending on the async runtime being cooperative.
+    // The AtomicBool guard ensures at most one safety-net timer is active.
+    if pending_state.has_pending_background_tasks
+      && !self
+        .inner
+        .state
+        .safety_net_active
+        .swap(true, std::sync::atomic::Ordering::Relaxed)
+    {
+      let waker = cx.waker().clone();
+      let flag = self.inner.state.safety_net_active.clone();
+      std::thread::spawn(move || {
+        std::thread::sleep(std::time::Duration::from_millis(100));
+        flag.store(false, std::sync::atomic::Ordering::Relaxed);
+        waker.wake_by_ref();
+      });
+    }
+
     // Re-wake logic for next iteration
     #[allow(clippy::suspicious_else_formatting, clippy::if_same_then_else)]
     {