fix(watcher): serialize watch and close on the native watcher

crates/node_binding/napi-binding.d.ts

@@ -470,8 +470,8 @@ export declare class NativeWatcher {
    * # Safety
    *
    * This function is unsafe because it uses `&mut self` to call the watcher asynchronously.
-   * It's important to ensure that the watcher is not used in any other places before this function is finished.
-   * You must ensure that the watcher not call watch, close or pause in the same time, otherwise it may lead to undefined behavior.
+   * `watch` and `close` are serialized internally, but calling `pause` while this function is
+   * in flight is still undefined behavior.
    */
   close(): Promise<void>
   pause(): void

crates/rspack_binding_api/Cargo.toml

@@ -117,7 +117,7 @@ serde                                  = { workspace = true }
 serde_json                             = { workspace = true }
 simd-json                              = { workspace = true }
 swc_core                               = { workspace = true, default-features = false, features = ["ecma_transforms_react"] }
-tokio                                  = { workspace = true, features = ["rt", "rt-multi-thread", "macros", "test-util", "tracing", "parking_lot"] }
+tokio                                  = { workspace = true, features = ["rt", "rt-multi-thread", "macros", "sync", "test-util", "tracing", "parking_lot"] }
 ustr                                   = { workspace = true }
 
 [package.metadata.cargo-shear]

crates/rspack_binding_api/src/native_watcher.rs

@@ -2,6 +2,7 @@ use std::{
   boxed::Box,
   panic::AssertUnwindSafe,
   path::{Path, PathBuf},
+  sync::Arc,
   time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
@@ -59,6 +60,8 @@ pub struct NativeWatchUndelayedEvent {
 pub struct NativeWatcher {
   watcher: FsWatcher,
   closed: bool,
+  watch_task: Option<tokio::task::JoinHandle<()>>,
+  watcher_lock: Arc<tokio::sync::Mutex<()>>,
 }
 
 fn timestamp_to_system_time(millis: u64) -> SystemTime {
@@ -81,6 +84,8 @@ impl NativeWatcher {
     Self {
       watcher,
       closed: false,
+      watch_task: None,
+      watcher_lock: Arc::new(tokio::sync::Mutex::new(())),
     }
   }
 
@@ -110,8 +115,15 @@ impl NativeWatcher {
 
     let start_time = start_time.get_u64().1;
 
+    if let Some(prev) = self.watch_task.take() {
+      prev.abort();
+    }
+
+    let watcher_lock = Arc::clone(&self.watcher_lock);
+    let mut watch_task = None;
     reference.share_with(env, |native_watcher| {
-      rspack_napi::runtime::spawn(async move {
+      watch_task = Some(rspack_napi::runtime::spawn_handle(async move {
+        let _guard = watcher_lock.lock_owned().await;
         native_watcher
           .watcher
           .watch(
@@ -123,9 +135,10 @@ impl NativeWatcher {
             Box::new(js_event_handler_undelayed),
           )
           .await
-      });
+      }));
       Ok(())
     })?;
+    self.watch_task = watch_task;
 
     Ok(())
   }
@@ -148,8 +161,8 @@ impl NativeWatcher {
   /// # Safety
   ///
   /// This function is unsafe because it uses `&mut self` to call the watcher asynchronously.
-  /// It's important to ensure that the watcher is not used in any other places before this function is finished.
-  /// You must ensure that the watcher not call watch, close or pause in the same time, otherwise it may lead to undefined behavior.
+  /// `watch` and `close` are serialized internally, but calling `pause` while this function is
+  /// in flight is still undefined behavior.
   pub unsafe fn close<'env>(
     &mut self,
     env: &'env Env,
@@ -160,6 +173,10 @@ impl NativeWatcher {
     let shared_reference = reference.share_with(Env::from_raw(env.raw()), |native_watcher| {
       rspack_napi::runtime::spawn(async move {
         let result = AssertUnwindSafe(async {
+          if let Some(task) = native_watcher.watch_task.take() {
+            task.abort();
+          }
+          let _guard = Arc::clone(&native_watcher.watcher_lock).lock_owned().await;
           native_watcher
             .watcher
             .close()

crates/rspack_napi/src/runtime.rs

@@ -38,6 +38,14 @@ where
   std::mem::drop(spawn_inner(future));
 }
 
+pub fn spawn_handle<F>(future: F) -> tokio::task::JoinHandle<F::Output>
+where
+  F: Future + Send + 'static,
+  F::Output: Send + 'static,
+{
+  spawn_inner(future)
+}
+
 pub fn block_on<F: Future>(future: F) -> F::Output {
   with_runtime(|runtime| runtime.block_on(future))
 }

tests/rspack-test/NativeWatcherLifecycle.test.js

@@ -0,0 +1,59 @@
+const fs = require("node:fs");
+const os = require("node:os");
+const path = require("node:path");
+const binding = require("@rspack/binding");
+
+const ITERATIONS = 30;
+const WATCHES_PER_ITERATION = 5;
+const FILE_COUNT = 400;
+
+let dir;
+let files;
+let dirs;
+
+const startWatch = watcher => {
+	watcher.watch(
+		[files, []],
+		[dirs, []],
+		[[], []],
+		BigInt(Date.now()),
+		() => {},
+		() => {}
+	);
+};
+
+describe("NativeWatcher lifecycle", () => {
+	beforeAll(() => {
+		dir = fs.mkdtempSync(path.join(os.tmpdir(), "rspack-native-watcher-"));
+		files = [];
+		dirs = [dir];
+		for (let i = 0; i < FILE_COUNT; i++) {
+			const sub = path.join(dir, `sub${i % 20}`);
+			fs.mkdirSync(sub, { recursive: true });
+			const file = path.join(sub, `f${i}.js`);
+			fs.writeFileSync(file, `module.exports = ${i};`);
+			files.push(file);
+			if (!dirs.includes(sub)) {
+				dirs.push(sub);
+			}
+		}
+	});
+
+	afterAll(() => {
+		fs.rmSync(dir, { recursive: true, force: true });
+	});
+
+	it("does not crash when watch and close race on the same watcher", async () => {
+		for (let i = 0; i < ITERATIONS; i++) {
+			const watcher = new binding.NativeWatcher({ aggregateTimeout: 1 });
+
+			for (let j = 0; j < WATCHES_PER_ITERATION; j++) {
+				startWatch(watcher);
+			}
+
+			await watcher.close();
+
+			expect(() => startWatch(watcher)).toThrow(/has been closed/);
+		}
+	});
+});

tests/rspack-test/package.json

@@ -18,6 +18,7 @@
     "@module-federation/runtime-tools": "2.7.0",
     "@prefresh/core": "1.5.10",
     "@prefresh/utils": "1.2.1",
+    "@rspack/binding": "workspace:*",
     "@rspack/binding-testing": "workspace:*",
     "@rspack/cli": "workspace:*",
     "@rspack/core": "workspace:*",