fix(watch): coalesce in-flight lazy and file generations

Author: matthewdavis-oai

packages/rspack/src/MultiCompiler.ts

@@ -431,7 +431,12 @@ export class MultiCompiler {
       if (node.state === 'done') {
         node.state = 'pending';
       } else if (node.state === 'running') {
-        node.state = 'running-outdated';
+        // Watching will coalesce its own in-flight invalidation before
+        // delivering `done`; keep the node runnable so that generation can
+        // finish and unblock its already-invalidated dependents.
+        if (!node.compiler.watching?.invalid) {
+          node.state = 'running-outdated';
+        }
       }
       for (const child of node.children) {
         nodeInvalidFromParent(child);

packages/rspack/src/Watching.ts

@@ -153,6 +153,12 @@ export class Watching {
       },
       (fileName, changeTime) => {
         this.#recordInvalidation('normal');
+        if (this.running) {
+          // The aggregate callback can arrive after an in-flight compilation
+          // finishes. Suppress that stale generation and drain the paused
+          // watcher before starting the coalesced rebuild.
+          this.invalid = true;
+        }
         if (!this.#invalidReported) {
           this.#invalidReported = true;
           this.compiler.hooks.invalid.call(fileName, changeTime);
@@ -294,6 +300,7 @@ export class Watching {
     if (this.suspended || (this.isBlocked() && (this.blocked = true))) {
       return;
     }
+    this.blocked = false;
 
     if (this.running) {
       this.invalid = true;
@@ -339,14 +346,7 @@ export class Watching {
       this.compiler.fileTimestamps = fileTimeInfoEntries;
       this.compiler.contextTimestamps = contextTimeInfoEntries;
     } else if (this.pausedWatcher) {
-      const { changes, removals, fileTimeInfoEntries, contextTimeInfoEntries } =
-        this.pausedWatcher.getInfo();
-      if (changes.size > 0 || removals.size > 0) {
-        this.#recordInvalidation('normal');
-      }
-      this.#mergeWithCollected(changes, removals);
-      this.compiler.fileTimestamps = fileTimeInfoEntries;
-      this.compiler.contextTimestamps = contextTimeInfoEntries;
+      this.#drainPausedWatcher();
     }
 
     this.compiler.__internal__watchInvalidationKind =
@@ -451,6 +451,21 @@ export class Watching {
 
     stats = new Stats(compilation);
 
+    const watcherStartTime = Date.now();
+    if (
+      !this.invalid &&
+      this.pausedWatcher?.hasPendingEvents?.() !== false &&
+      this.#drainPausedWatcher()
+    ) {
+      // Watchpack continues collecting while paused but does not deliver an
+      // invalid callback. Coalesce those changes before publishing the stale
+      // generation and advance the baseline so they are not replayed.
+      this.lastWatcherStartTime = watcherStartTime;
+      this.invalid = true;
+      this.#notifyInvalid();
+      this.onInvalid();
+    }
+
     if (
       this.invalid &&
       !this.suspended &&
@@ -513,6 +528,21 @@ export class Watching {
     });
   }
 
+  #drainPausedWatcher() {
+    if (!this.pausedWatcher) return false;
+
+    const { changes, removals, fileTimeInfoEntries, contextTimeInfoEntries } =
+      this.pausedWatcher.getInfo();
+    const hasChanges = changes.size > 0 || removals.size > 0;
+    if (hasChanges) {
+      this.#recordInvalidation('normal');
+    }
+    this.#mergeWithCollected(changes, removals);
+    this.compiler.fileTimestamps = fileTimeInfoEntries;
+    this.compiler.contextTimestamps = contextTimeInfoEntries;
+    return hasChanges;
+  }
+
   #mergeWithCollected(
     changedFiles?: ReadonlySet<string>,
     removedFiles?: ReadonlySet<string>,

packages/rspack/src/node/NodeWatchFileSystem.ts

@@ -189,9 +189,24 @@ export default class NodeWatchFileSystem implements WatchFileSystem {
         "Watcher.getContextTimeInfoEntries is deprecated in favor of Watcher.getInfo since that's more performant.",
         'DEP_WEBPACK_WATCHER_CONTEXT_TIME_INFO_ENTRIES',
       ),
+      hasPendingEvents: () => {
+        const watcher = this.watcher;
+        return Boolean(
+          watcher &&
+          (watcher.aggregatedChanges.size > 0 ||
+            watcher.aggregatedRemovals.size > 0),
+        );
+      },
       getInfo: () => {
-        const removals = this.watcher?.aggregatedRemovals ?? new Set();
-        const changes = this.watcher?.aggregatedChanges ?? new Set();
+        const watcher = this.watcher;
+        const { changes, removals } = watcher
+          ? watcher.paused
+            ? watcher.getAggregated()
+            : {
+                changes: watcher.aggregatedChanges,
+                removals: watcher.aggregatedRemovals,
+              }
+          : { changes: new Set<string>(), removals: new Set<string>() };
         if (this.inputFileSystem?.purge) {
           const fs = this.inputFileSystem;
           if (removals) {

packages/rspack/src/util/fs.ts

@@ -19,6 +19,7 @@ export interface Watcher {
   getAggregatedRemovals?(): Set<string>; // get current aggregated removals that have not yet send to callback
   getFileTimeInfoEntries?(): Map<string, FileSystemInfoEntry | 'ignore'>; // get info about files
   getContextTimeInfoEntries?(): Map<string, FileSystemInfoEntry | 'ignore'>; // get info about directories
+  hasPendingEvents?(): boolean; // cheaply checks for changes collected while paused
   getInfo(): WatcherInfo; // get info about timestamps and changes
 }
 

tests/rspack-test/LazyCompilationArtifactChain.test.js

@@ -31,13 +31,14 @@ async function withTimeout(promise, description) {
 }
 
 describe("lazy MultiCompiler artifact provenance", () => {
-	it("keeps dependent and coalesced file-backed generations normal", async () => {
+	async function runArtifactChain(_watcherName, nativeWatcher) {
 		const root = await fs.mkdtemp(path.join(os.tmpdir(), "rspack-lazy-chain-"));
 		const sourceFile = path.join(root, "source.js");
 		const coalescedEntry = path.join(root, "coalesced-source.js");
 		const stampFile = name => path.join(root, name, "stamp.txt");
 		const generations = Object.fromEntries(names.map(name => [name, 0]));
 		const events = [];
+		const invalidations = [];
 		let delayedSource;
 		let resolveFileInvalidation;
 		let server;
@@ -56,6 +57,14 @@ describe("lazy MultiCompiler artifact provenance", () => {
 				coalescedEntry,
 				'export const coalesced = "coalesced";\n'
 			);
+			// Keep Watchpack's initial scan from treating fresh fixture files as edits.
+			const beforeWatch = new Date(Date.now() - 3000);
+			await Promise.all(
+				[
+					...names.map(name => path.join(root, `${name}.js`)),
+					coalescedEntry
+				].map(file => fs.utimes(file, beforeWatch, beforeWatch))
+			);
 
 			const config = name => ({
 				context: root,
@@ -65,7 +74,7 @@ describe("lazy MultiCompiler artifact provenance", () => {
 					name === "source"
 						? { main: "./source.js", coalesced: "./coalesced-source.js" }
 						: { main: `./${name}.js` },
-				experiments: { nativeWatcher: true },
+				experiments: { nativeWatcher },
 				lazyCompilation: { entries: true, imports: false },
 				mode: "development",
 				name,
@@ -78,6 +87,7 @@ describe("lazy MultiCompiler artifact provenance", () => {
 					{
 						apply(compiler) {
 							compiler.hooks.invalid.tap("lazy-artifact-chain", file => {
+								invalidations.push({ name, file });
 								if (name === "source" && file === sourceFile) {
 									resolveFileInvalidation?.();
 								}
@@ -278,19 +288,32 @@ describe("lazy MultiCompiler artifact provenance", () => {
 			const fileInvalidated = new Promise(resolve => {
 				resolveFileInvalidation = resolve;
 			});
-			delayedSource = { started: signalSourceStarted, release: sourceRelease };
+			delayedSource = {
+				started: signalSourceStarted,
+				release: sourceRelease
+			};
 			const beforeCoalesced = events.length;
 			const beforeGenerations = { ...generations };
 			assert.equal(await activate("source", JSON.parse(coalescedId)), 200);
 			await withTimeout(sourceStarted, "the coalesced source build to start");
+			const beforeFileInvalidations = invalidations.length;
 			await fs.writeFile(
 				sourceFile,
 				'export const source = "source";\nexport const edited = "file-edit";\n'
 			);
-			await withTimeout(fileInvalidated, "the coalesced source file event");
+			if (nativeWatcher) {
+				await withTimeout(fileInvalidated, "the coalesced source file event");
+			}
 			releaseSource();
 			assertBuild(await nextBuild(), ["source", "dependent", "tail"], "normal");
 			const coalesced = events.slice(beforeCoalesced);
+			assert.equal(
+				invalidations
+					.slice(beforeFileInvalidations)
+					.filter(event => event.name === "source").length,
+				1,
+				"a coalesced file edit must notify the source compiler exactly once"
+			);
 			assert.equal(
 				coalesced.every(event => event.kind === "normal"),
 				true,
@@ -319,6 +342,9 @@ describe("lazy MultiCompiler artifact provenance", () => {
 			assert.equal(generations.tail, beforeGenerations.tail + 1);
 			assert.equal(generations.unrelated, beforeGenerations.unrelated);
 			await assertTail();
+			const deliveredGenerations = { ...generations };
+			await new Promise(resolve => setTimeout(resolve, 80));
+			assert.deepEqual(generations, deliveredGenerations);
 		} finally {
 			if (watching) {
 				await new Promise((resolve, reject) =>
@@ -328,5 +354,13 @@ describe("lazy MultiCompiler artifact provenance", () => {
 			if (server) await new Promise(resolve => server.close(resolve));
 			await fs.rm(root, { force: true, recursive: true });
 		}
-	});
+	}
+
+	it.each([
+		["native", true],
+		["watchpack", false]
+	])(
+		"keeps dependent and coalesced file-backed generations normal with %s watching",
+		runArtifactChain
+	);
 });

tests/rspack-test/NodeWatcherGetInfo.test.js

@@ -0,0 +1,52 @@
+const fs = require("node:fs/promises");
+const os = require("node:os");
+const path = require("node:path");
+const { rspack } = require("@rspack/core");
+
+describe("NodeWatchFileSystem getInfo", () => {
+	it("keeps an active aggregate callback and consumes paused changes once", async () => {
+		const root = await fs.mkdtemp(path.join(os.tmpdir(), "rspack-watchpack-"));
+		const entry = path.join(root, "entry.js");
+		await fs.writeFile(entry, "export default 1;\n");
+		const beforeWatch = new Date(Date.now() - 3000);
+		await fs.utimes(entry, beforeWatch, beforeWatch);
+
+		const compiler = rspack({
+			context: root,
+			entry,
+			experiments: { nativeWatcher: false }
+		});
+		const watchFileSystem = compiler.watchFileSystem;
+		const callbackChanges = [];
+		const watcher = watchFileSystem.watch(
+			[entry],
+			[],
+			[],
+			Date.now(),
+			{ aggregateTimeout: 1000 },
+			(_error, _fileTimes, _contextTimes, changes) =>
+				callbackChanges.push(changes),
+			() => {}
+		);
+
+		try {
+			const watchpack = watchFileSystem.watcher;
+			watchpack._onChange(entry, Date.now(), entry, "change");
+			expect(watcher.hasPendingEvents()).toBe(true);
+			expect(watcher.getInfo().changes).toEqual(new Set([entry]));
+			expect(watchpack.aggregateTimer).toBeDefined();
+			watchpack._onTimeout();
+			expect(callbackChanges).toEqual([new Set([entry])]);
+
+			watchpack._onChange(entry, Date.now(), entry, "change");
+			expect(watcher.hasPendingEvents()).toBe(true);
+			expect(watcher.getInfo().changes).toEqual(new Set([entry]));
+			expect(watcher.getInfo().changes).toEqual(new Set());
+			expect(watcher.hasPendingEvents()).toBe(false);
+		} finally {
+			watcher.close();
+			await new Promise(resolve => compiler.close(resolve));
+			await fs.rm(root, { force: true, recursive: true });
+		}
+	});
+});

tests/rspack-test/multiCompilerCases/invalidate.js

@@ -80,7 +80,9 @@ module.exports = [
 					  a done,
 					]
 				`);
-                  expect(lazyCompilationCycles).toEqual([undefined, "lazy"]);
+                  // `a` consumes `b`'s lazy artifact, so its dependent
+                  // generation must remain normal and emit fresh output.
+                  expect(lazyCompilationCycles).toEqual([undefined, "normal"]);
                   events.length = 0;
                   expect(state).toBe(1);
                   state = 2;
@@ -110,8 +112,8 @@ module.exports = [
 			`);
               expect(lazyCompilationCycles).toEqual([
                 undefined,
-                "lazy",
                 "normal",
+                "normal"
               ]);
               state = 3;
               compiler.close(error => (error ? reject(error) : resolve()));