Fix overlapping CLI watch rebuilds

Author: michaelglass

integrations/cli/index.test.ts

@@ -409,6 +409,80 @@ describe.each([
     },
   )
 
+  test(
+    'watch mode keeps the newest CSS when a rebuild is already running',
+    {
+      fs: {
+        'package.json': json`{}`,
+        'pnpm-workspace.yaml': yaml` packages:
+            - project-a `,
+        'project-a/package.json': json`
+          {
+            "dependencies": {
+              "tailwindcss": "workspace:^",
+              "@tailwindcss/cli": "workspace:^"
+            }
+          }
+        `,
+        'project-a/index.html': html`<div class="from-a from-b"></div>`,
+        'project-a/src/index.css': css`@import 'tailwindcss/utilities';`,
+        'project-a/plugin-a.mjs': js`
+          import fs from 'node:fs/promises'
+
+          await fs.writeFile(new URL('./a-started', import.meta.url), 'started')
+          await new Promise((resolve) => setTimeout(resolve, 2_000))
+          await fs.appendFile(new URL('./build-order', import.meta.url), 'A')
+
+          export default function ({ addUtilities }) {
+            addUtilities({ '.from-a': { color: 'red' } })
+          }
+        `,
+        'project-a/plugin-b.mjs': js`
+          import fs from 'node:fs/promises'
+
+          await fs.appendFile(new URL('./build-order', import.meta.url), 'B')
+
+          export default function ({ addUtilities }) {
+            addUtilities({ '.from-b': { color: 'green' } })
+          }
+        `,
+      },
+    },
+    async ({ root, fs, spawn, expect }) => {
+      let process = await spawn(`${command} --input src/index.css --output dist/out.css --watch`, {
+        cwd: path.join(root, 'project-a'),
+      })
+      await process.onStderr((message) => message.includes('Done in'))
+
+      await fs.write(
+        'project-a/src/index.css',
+        css`
+          @import 'tailwindcss/utilities';
+          @plugin '../plugin-a.mjs';
+        `,
+      )
+      await retryAssertion(async () => {
+        expect(await fs.read('project-a/a-started')).toBe('started')
+      })
+
+      await fs.write(
+        'project-a/src/index.css',
+        css`
+          @import 'tailwindcss/utilities';
+          @plugin '../plugin-b.mjs';
+        `,
+      )
+
+      await process.onStderr((message) => message.includes('Done in'))
+      await process.onStderr((message) => message.includes('Done in'))
+
+      expect(await fs.read('project-a/build-order')).toBe('AB')
+      await fs.expectFileToContain('project-a/dist/out.css', [candidate`from-b`])
+      await fs.expectFileNotToContain('project-a/dist/out.css', [candidate`from-a`])
+    },
+    { skip: kind !== 'CLI' },
+  )
+
   test(
     "watch mode with unknown @source paths shouldn't crash on Windows",
     {

packages/@tailwindcss-cli/src/commands/build/index.test.ts

@@ -0,0 +1,68 @@
+import { expect, it } from 'vitest'
+import { serializeBatches } from '../../utils/serial-batches'
+import { createWatchers } from './index'
+
+type WatchCallback = (
+  error: Error | null,
+  events: { type: 'delete'; path: string }[],
+) => Promise<void>
+
+function fakeWatcher() {
+  let callbacks: WatchCallback[] = []
+  return {
+    callbacks,
+    watcher: {
+      async subscribe(_directory: string, callback: WatchCallback) {
+        callbacks.push(callback)
+        return { unsubscribe() {} }
+      },
+    },
+  }
+}
+
+function nextTask() {
+  return new Promise((resolve) => setTimeout(resolve, 0))
+}
+
+it('uses one rebuild queue across watcher generations', async () => {
+  let releaseFirst!: () => void
+  let firstCanFinish = new Promise<void>((resolve) => (releaseFirst = resolve))
+  let calls: string[][] = []
+  let queue = serializeBatches<string>(async (files) => {
+    calls.push(files)
+    if (calls.length === 1) await firstCanFinish
+  })
+  let first = fakeWatcher()
+  let second = fakeWatcher()
+
+  let oldGeneration = await createWatchers(['/old'], async () => {}, queue, first.watcher as any)
+  await first.callbacks[0](null, [{ type: 'delete', path: 'old-change' }])
+  await nextTask()
+  expect(calls).toEqual([['old-change']])
+
+  let newGeneration = await createWatchers(['/new'], async () => {}, queue, second.watcher as any)
+  await second.callbacks[0](null, [{ type: 'delete', path: 'new-change' }])
+  await nextTask()
+  expect(calls).toEqual([['old-change']])
+
+  releaseFirst()
+  await queue.close()
+  expect(calls).toEqual([['old-change'], ['new-change']])
+
+  await Promise.all([oldGeneration.cleanup(), newGeneration.cleanup()])
+})
+
+it('flushes a collected event when shutdown cancels its debounce timer', async () => {
+  let calls: string[][] = []
+  let queue = serializeBatches<string>(async (files) => {
+    calls.push(files)
+  })
+  let fake = fakeWatcher()
+  let generation = await createWatchers(['/watch'], async () => {}, queue, fake.watcher as any)
+
+  await fake.callbacks[0](null, [{ type: 'delete', path: 'last-change' }])
+  await generation.cleanup()
+  await queue.close()
+
+  expect(calls).toEqual([['last-change']])
+})

packages/@tailwindcss-cli/src/commands/build/index.ts

@@ -22,6 +22,7 @@ import {
   relative,
   wordWrap,
 } from '../../utils/renderer'
+import { serializeBatches, type SerialBatches } from '../../utils/serial-batches'
 import { drainStdin, outputFile } from './utils'
 
 const css = String.raw
@@ -326,15 +327,25 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
 
   let [compiler, scanner] = await handleError(() => createCompiler(input, I))
   let cleanupWatchers: (() => Promise<void>)[] = []
+  let finishInitialBuild!: () => void
+  let initialBuildFinished = new Promise<void>((resolve) => (finishInitialBuild = resolve))
+  let eventBatches: SerialBatches<string> | null = null
+  let eventHandler: ((files: string[]) => Promise<void>) | null = null
 
   // Watch for changes
   if (args['--watch'] && pollInterval === false) {
     // Ensure the file watcher can be loaded before setting up any watchers,
     // such that we can present a helpful error message if needed.
     await handleError(() => loadWatcher())
 
-    cleanupWatchers.push(
-      await createWatchers(await watchDirectories(scanner), async function handle(files) {
+    eventBatches = serializeBatches(
+      (files) => eventHandler!(files),
+      initialBuildFinished,
+      (error) => eprintln(formatError(error)),
+    )
+    let initialWatchers = await createWatchers(
+      await watchDirectories(scanner),
+      async function handle(files) {
         try {
           // If the only change happened to the output file, then we don't want to
           // trigger a rebuild because that will result in an infinite loop.
@@ -388,15 +399,19 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
 
             // Setup new watchers
             DEBUG && I.start('Setup new watchers')
-            let newCleanupFunction = await createWatchers(await watchDirectories(scanner), handle)
+            let newWatchers = await createWatchers(
+              await watchDirectories(scanner),
+              handle,
+              eventBatches!,
+            )
             DEBUG && I.end('Setup new watchers')
 
             // Clear old watchers
             DEBUG && I.start('Cleanup old watchers')
             await Promise.all(cleanupWatchers.splice(0).map((cleanup) => cleanup()))
             DEBUG && I.end('Cleanup old watchers')
 
-            cleanupWatchers.push(newCleanupFunction)
+            cleanupWatchers.push(newWatchers.cleanup)
 
             // Re-compile the CSS
             DEBUG && I.start('Build CSS')
@@ -481,17 +496,22 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
           let end = process.hrtime.bigint()
           if (!args['--silent']) eprintln(`Done in ${formatDuration(end - start)}`)
         }
-      }),
+      },
+      eventBatches,
     )
+    eventHandler = initialWatchers.callback
+    cleanupWatchers.push(initialWatchers.cleanup)
 
     // Abort the watcher if `stdin` is closed to avoid zombie processes. You can
     // disable this behavior with `--watch=always`.
     if (args['--watch'] !== 'always') {
       process.stdin.on('end', () => {
-        Promise.all(cleanupWatchers.map((fn) => fn())).then(
-          () => process.exit(0),
-          () => process.exit(1),
-        )
+        Promise.all(cleanupWatchers.map((fn) => fn()))
+          .then(() => eventBatches?.close())
+          .then(
+            () => process.exit(0),
+            () => process.exit(1),
+          )
       })
     }
 
@@ -515,6 +535,7 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
   }
 
   await write(output, map, args, I)
+  finishInitialBuild()
 
   let end = process.hrtime.bigint()
   if (!args['--silent']) eprintln(`Done in ${formatDuration(end - start)}`)
@@ -693,9 +714,13 @@ async function loadWatcher(): Promise<typeof import('@parcel/watcher')> {
   }
 }
 
-async function createWatchers(dirs: string[], cb: (files: string[]) => void) {
-  let watcher = await loadWatcher()
-
+export async function createWatchers(
+  dirs: string[],
+  cb: (files: string[]) => Promise<void>,
+  batches: SerialBatches<string>,
+  watcher?: Awaited<ReturnType<typeof loadWatcher>>,
+) {
+  watcher ??= await loadWatcher()
   // Remove any directories that are children of an already watched directory.
   // If we don't we may not get notified of certain filesystem events regardless
   // of whether or not they are for the directory that is duplicated.
@@ -740,8 +765,9 @@ async function createWatchers(dirs: string[], cb: (files: string[]) => void) {
 
     // Setup a new macrotask to handle the files in batch.
     debounceQueue.queueMacrotask(() => {
-      cb(Array.from(files))
+      let batch = Array.from(files)
       files.clear()
+      void batches.push(batch)
     })
   }
 
@@ -791,9 +817,17 @@ async function createWatchers(dirs: string[], cb: (files: string[]) => void) {
   }
 
   // Cleanup
-  return async () => {
-    await watchers.dispose()
-    await debounceQueue.dispose()
+  return {
+    callback: cb,
+    cleanup: async () => {
+      await watchers.dispose()
+      await debounceQueue.dispose()
+      if (files.size > 0) {
+        let batch = Array.from(files)
+        files.clear()
+        void batches.push(batch)
+      }
+    },
   }
 }