fs.watch: stop dispatching watcher events into a terminated Worker VM

src/runtime/node/node_fs_watcher.rs

@@ -94,6 +94,14 @@ impl FSWatcher {
         self.vm().event_loop()
     }
 
+    /// A worker's sticky `TerminationException` is not cleared by the emit
+    /// paths below, so a second listener call after it fires would re-enter
+    /// `executeCallImpl` under `scope.assertNoException()`.
+    #[inline]
+    fn can_call_into_js(&self) -> bool {
+        self.vm().script_execution_status() == jsc::ScriptExecutionStatus::Running
+    }
+
     /// `task` must point to a live heap-allocated `ConcurrentTask` node that
     /// the caller releases ownership of; the concurrent queue takes ownership
     /// and frees it on the JS thread after dispatch.
@@ -188,6 +196,9 @@ impl FSWatchTaskPosix {
         // this runs on JS Context Thread
 
         for i in 0..self.count as usize {
+            if !self.ctx().can_call_into_js() {
+                break;
+            }
             // SAFETY: entries [0..count) were written by `append`.
             let entry = unsafe { self.entries[i].assume_init_ref() };
             match &entry.event {
@@ -447,13 +458,15 @@ impl FSWatchTaskWindows {
         // match is sound (aliased shared borrows are fine; the old `*mut Self`
         // re-derive dance is no longer needed). `ParentRef` Derefs to `&T`.
         let ctx: &FSWatcher = &self.ctx.expect("FSWatchTask.ctx unset");
-        match &mut self.event {
-            Event::Rename(path) => Self::run_path::<{ EventType::Rename }>(ctx, path),
-            Event::Change(path) => Self::run_path::<{ EventType::Change }>(ctx, path),
-            Event::Error(err) => ctx.emit_error(err),
-            Event::NoFilename(event_type) => ctx.emit_null_filename(*event_type),
-            Event::Abort => ctx.emit_if_aborted(),
-            Event::Close => ctx.emit::<{ EventType::Close }>(b""),
+        if ctx.can_call_into_js() {
+            match &mut self.event {
+                Event::Rename(path) => Self::run_path::<{ EventType::Rename }>(ctx, path),
+                Event::Change(path) => Self::run_path::<{ EventType::Change }>(ctx, path),
+                Event::Error(err) => ctx.emit_error(err),
+                Event::NoFilename(event_type) => ctx.emit_null_filename(*event_type),
+                Event::Abort => ctx.emit_if_aborted(),
+                Event::Close => ctx.emit::<{ EventType::Close }>(b""),
+            }
         }
 
         ctx.unref_task();
@@ -991,7 +1004,9 @@ impl FSWatcher {
             self.detach();
 
             if let Some(js_this) = js_this {
-                if let Some(listener) = js::listener_get_cached(js_this) {
+                if self.can_call_into_js()
+                    && let Some(listener) = js::listener_get_cached(js_this)
+                {
                     // `closed` is already true so `refTask()` would return false without
                     // incrementing; bump the counter directly so the `unrefTask()` below is
                     // balanced and the count stays > 0 while the close event is emitted.

test/js/node/watch/fs.watch.test.ts

@@ -1311,6 +1311,71 @@ test.skipIf(!isWindows)(
   30000,
 );
 
+// A Worker owns an fs.watch(); the main thread queues a burst of inotify events
+// for the watched directory and then terminate()s the Worker while those events
+// are still being dispatched. The unfixed build's FSWatchTask::run() kept
+// calling the listener after the sticky TerminationException was set on the
+// worker VM, re-entering executeCallImpl() under scope.assertNoException() and
+// aborting. Must run in a subprocess: on an unfixed debug build the abort takes
+// down the whole runtime.
+test("terminating a worker while its fs.watch has queued events does not crash", async () => {
+  using dir = tempDir("fswatch-worker-terminate", { ".keep": "" });
+  const watched = String(dir);
+
+  const fixture = /* js */ `
+    const fs = require("node:fs");
+    const { Worker } = require("node:worker_threads");
+
+    const d = ${JSON.stringify(watched)};
+    const workerCode = \`
+      const fs = require("node:fs");
+      const { parentPort, workerData } = require("node:worker_threads");
+      const w = fs.watch(workerData.d, () => {});
+      w.on("error", () => {});
+      parentPort.postMessage("READY");
+    \`;
+
+    (async () => {
+      for (let i = 0; i < 8; i++) {
+        const wk = new Worker(workerCode, { eval: true, workerData: { d } });
+        await new Promise((resolve, reject) => {
+          wk.once("message", resolve);
+          wk.once("error", reject);
+        });
+        for (let j = 0; j < 50; j++) fs.writeFileSync(d + "/b" + j, "x");
+        await wk.terminate();
+      }
+      console.log("ok");
+    })().catch(err => {
+      console.error(String(err));
+      process.exit(1);
+    });
+  `;
+
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", fixture],
+    env: {
+      ...bunEnv,
+      // detect_leaks=0: ConcurrentTask nodes left in a terminated worker's
+      // undrained concurrent queue are a known pre-existing leak (see #32071);
+      // this test asserts no crash, not no leaks. symbolize=0 so a pre-fix
+      // ASAN abort exits promptly instead of spending seconds in
+      // llvm-symbolizer.
+      ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "symbolize=0", "detect_leaks=0"].filter(Boolean).join(":"),
+    },
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+  expect({ stdout: stdout.trim(), stderr, exitCode, signalCode: proc.signalCode }).toEqual({
+    stdout: "ok",
+    stderr: expect.not.stringContaining("ASSERTION"),
+    exitCode: 0,
+    signalCode: null,
+  });
+}, 30_000);
+
 // FSWatcher::init joins the user-supplied watch path with the process cwd into a
 // fixed pooled path buffer. The raw-path length validator only bounds the path
 // itself, so a relative path just under the platform path limit used to overflow