fix(watcher): isolate stale native watch generations

Keep retained watcher handles, callbacks, and shims from pausing, draining, closing, purging, or injecting into a live native-watch generation. Acknowledge obsolete aggregates so delivery cannot stall and cover long-lived raw listeners, stale handles, and generation replacement.

Author: matthewdavis-oai

packages/rspack/src/NativeWatchFileSystem.ts

@@ -169,9 +169,11 @@ export default class NativeWatchFileSystem implements WatchFileSystem {
     // Fresh shim per cycle (see field comment). Events are emitted to both the
     // long-lived `#events` (the `on`/`once` API) and this cycle's shim (the
     // `.watcher` surface).
-    const watcher = new NativeWatcherShim((kind, path) =>
-      this.#inner?.triggerEvent(kind, path),
-    );
+    const watcher = new NativeWatcherShim((kind, path) => {
+      if (this.#watcher === watcher) {
+        nativeWatcher.triggerEvent(kind, path);
+      }
+    });
     this.#watcher = watcher;
     let lastDrainedGeneration: number | undefined;
 
@@ -181,6 +183,12 @@ export default class NativeWatchFileSystem implements WatchFileSystem {
       this.formatWatchDependencies(missing),
       BigInt(startTime),
       (err: Error | null, result) => {
+        if (this.#watcher !== watcher) {
+          if (!err) {
+            nativeWatcher.acknowledgePendingEvents(result.generation);
+          }
+          return;
+        }
         if (err) {
           callback(err, new Map(), new Map(), new Set(), new Set());
           return;
@@ -232,9 +240,10 @@ export default class NativeWatchFileSystem implements WatchFileSystem {
 
     return {
       close: () => {
-        if (this.#watcher === watcher) {
-          this.#watcher = undefined;
+        if (this.#watcher !== watcher) {
+          return;
         }
+        this.#watcher = undefined;
         if (this.#inner === nativeWatcher) {
           this.#inner = undefined;
           this.#isFirstWatch = true;
@@ -245,10 +254,20 @@ export default class NativeWatchFileSystem implements WatchFileSystem {
       },
 
       pause: () => {
-        nativeWatcher.pause();
+        if (this.#watcher === watcher) {
+          nativeWatcher.pause();
+        }
       },
 
       getInfo: () => {
+        if (this.#watcher !== watcher) {
+          return {
+            changes: new Set(),
+            removals: new Set(),
+            fileTimeInfoEntries: new Map(),
+            contextTimeInfoEntries: new Map(),
+          };
+        }
         const { changedFiles, removedFiles, generation } =
           nativeWatcher.takePendingEvents();
         lastDrainedGeneration = generation;

tests/rspack-test/NativeWatcherGeneration.test.js

@@ -3,6 +3,9 @@ const { rspack } = require("@rspack/core");
 class FakeNativeWatcher {
 	acknowledgements = [];
 	pauses = 0;
+	closes = 0;
+	drains = 0;
+	triggered = [];
 	pendingDrain = {
 		changedFiles: [],
 		removedFiles: [],
@@ -15,6 +18,7 @@ class FakeNativeWatcher {
 	}
 
 	takePendingEvents() {
+		this.drains++;
 		return this.pendingDrain;
 	}
 
@@ -26,7 +30,12 @@ class FakeNativeWatcher {
 		this.pauses++;
 	}
 
+	triggerEvent(kind, path) {
+		this.triggered.push([kind, path]);
+	}
+
 	close() {
+		this.closes++;
 		return Promise.resolve();
 	}
 }
@@ -154,4 +163,118 @@ describe("NativeWatchFileSystem aggregate generations", () => {
 		nativeWatcher.onRaw({ kind: "change", path: "/after-close" });
 		expect(fresh).toEqual(["/fresh"]);
 	});
+
+	it("forwards concrete child events to long-lived watch-file-system listeners", () => {
+		const events = [];
+		const { nativeWatcher, watchFileSystem } = createWatcherHarness();
+		watchFileSystem.on("change", (file, mtime) =>
+			events.push(["change", file, mtime])
+		);
+		watchFileSystem.on("remove", file => events.push(["remove", file]));
+		const staleRaw = nativeWatcher.onRaw;
+
+		const currentWatcher = watchFileSystem.watch(
+			dependencies(),
+			dependencies(),
+			dependencies(),
+			Date.now(),
+			{},
+			() => {},
+			() => {}
+		);
+
+		staleRaw({ kind: "change", path: "/src/stale.js" });
+		staleRaw({ kind: "remove", path: "/src/stale.js" });
+		nativeWatcher.onRaw({ kind: "change", path: "/src/created.js" });
+		nativeWatcher.onRaw({ kind: "remove", path: "/src/created.js" });
+		expect(events).toEqual([
+			["change", "/src/created.js", expect.any(Number)],
+			["remove", "/src/created.js"]
+		]);
+
+		currentWatcher.close();
+		nativeWatcher.onRaw({ kind: "change", path: "/src/after-close.js" });
+		nativeWatcher.onRaw({ kind: "remove", path: "/src/after-close.js" });
+		expect(events).toHaveLength(2);
+	});
+
+	it("prevents stale watcher handles and callbacks from affecting the live generation", () => {
+		const events = [];
+		const {
+			nativeWatcher,
+			purged,
+			callbackChanges,
+			watcher: staleWatcher,
+			watchFileSystem
+		} = createWatcherHarness();
+		watchFileSystem.on("aggregated", (changes, removals) =>
+			events.push([changes, removals])
+		);
+		const staleAggregate = nativeWatcher.onAggregate;
+		const staleShim = watchFileSystem.watcher;
+
+		const currentChanges = [];
+		const currentWatcher = watchFileSystem.watch(
+			dependencies(),
+			dependencies(),
+			dependencies(),
+			Date.now(),
+			{},
+			(_error, _fileTimes, _contextTimes, changes) =>
+				currentChanges.push(changes),
+			() => {}
+		);
+
+		nativeWatcher.pendingDrain = {
+			changedFiles: ["/src/live.js"],
+			removedFiles: [],
+			generation: 2
+		};
+		staleWatcher.pause();
+		expect(staleWatcher.getInfo()).toEqual({
+			changes: new Set(),
+			removals: new Set(),
+			fileTimeInfoEntries: new Map(),
+			contextTimeInfoEntries: new Map()
+		});
+		staleWatcher.close();
+		staleShim._onChange("/src/stale.js");
+		staleShim._onRemove("/src/stale.js");
+		staleAggregate(null, {
+			changedFiles: ["/src/stale.js"],
+			removedFiles: [],
+			generation: 1
+		});
+
+		expect(nativeWatcher.pauses).toBe(0);
+		expect(nativeWatcher.drains).toBe(0);
+		expect(nativeWatcher.closes).toBe(0);
+		expect(nativeWatcher.triggered).toEqual([]);
+		expect(nativeWatcher.acknowledgements).toEqual([1]);
+		expect(purged).toEqual([]);
+		expect(callbackChanges).toEqual([]);
+		expect(currentChanges).toEqual([]);
+		expect(events).toEqual([]);
+
+		nativeWatcher.onAggregate(null, {
+			changedFiles: ["/src/live.js"],
+			removedFiles: [],
+			generation: 2
+		});
+		expect(nativeWatcher.pauses).toBe(1);
+		expect(nativeWatcher.acknowledgements).toEqual([1, 2]);
+		expect(purged).toEqual(["/src/live.js"]);
+		expect(callbackChanges).toEqual([]);
+		expect(currentChanges).toEqual([new Set(["/src/live.js"])]);
+		expect(events).toEqual([[new Set(["/src/live.js"]), new Set()]]);
+		watchFileSystem.watcher._onChange("/src/live.js");
+		watchFileSystem.watcher._onRemove("/src/live.js");
+		expect(nativeWatcher.triggered).toEqual([
+			["change", "/src/live.js"],
+			["remove", "/src/live.js"]
+		]);
+
+		currentWatcher.close();
+		expect(nativeWatcher.closes).toBe(1);
+	});
 });