Fix overlapping CLI watch rebuilds

Author: michaelglass

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,7 +714,11 @@ async function loadWatcher(): Promise<typeof import('@parcel/watcher')> {
   }
 }
 
-async function createWatchers(dirs: string[], cb: (files: string[]) => void) {
+async function createWatchers(
+  dirs: string[],
+  cb: (files: string[]) => Promise<void>,
+  batches: SerialBatches<string>,
+) {
   let watcher = await loadWatcher()
 
   // Remove any directories that are children of an already watched directory.
@@ -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,12 @@ 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()
+    },
   }
 }
 

packages/@tailwindcss-cli/src/utils/serial-batches.test.ts

@@ -0,0 +1,128 @@
+import { expect, it } from 'vitest'
+import { serializeBatches } from './serial-batches'
+
+it('serializes callbacks and coalesces batches received while one is running', async () => {
+  let releaseFirst!: () => void
+  let firstCanFinish = new Promise<void>((resolve) => (releaseFirst = resolve))
+  let batches: string[][] = []
+  let active = 0
+  let maxActive = 0
+
+  let batchesQueue = serializeBatches<string>(async (batch) => {
+    batches.push(batch)
+    active++
+    maxActive = Math.max(maxActive, active)
+
+    if (batches.length === 1) {
+      await firstCanFinish
+    }
+
+    active--
+  })
+
+  let first = batchesQueue.push(['a'])
+  await Promise.resolve()
+
+  let second = batchesQueue.push(['b'])
+  let third = batchesQueue.push(['c'])
+
+  expect(batches).toEqual([['a']])
+  expect(maxActive).toBe(1)
+
+  releaseFirst()
+  await Promise.all([first, second, third])
+
+  expect(batches).toEqual([['a'], ['b', 'c']])
+  expect(maxActive).toBe(1)
+})
+
+it('shares serialization across watcher generations', async () => {
+  let releaseFirst!: () => void
+  let firstCanFinish = new Promise<void>((resolve) => (releaseFirst = resolve))
+  let calls: string[][] = []
+  let active = 0
+  let maxActive = 0
+  let batchesQueue = serializeBatches<string>(async (batch) => {
+    calls.push(batch)
+    active++
+    maxActive = Math.max(maxActive, active)
+    if (calls.length === 1) await firstCanFinish
+    active--
+  })
+
+  // Both watcher generations publish into the command-lifetime queue.
+  let oldGeneration = (files: string[]) => batchesQueue.push(files)
+  let newGeneration = (files: string[]) => batchesQueue.push(files)
+  let first = oldGeneration(['old'])
+  await Promise.resolve()
+  let second = newGeneration(['new'])
+
+  expect(calls).toEqual([['old']])
+  releaseFirst()
+  await Promise.all([first, second])
+
+  expect(calls).toEqual([['old'], ['new']])
+  expect(maxActive).toBe(1)
+})
+
+it('drains accepted batches and ignores new work after close', async () => {
+  let release!: () => void
+  let canFinish = new Promise<void>((resolve) => (release = resolve))
+  let calls: string[][] = []
+  let batchesQueue = serializeBatches<string>(async (batch) => {
+    calls.push(batch)
+    await canFinish
+  })
+
+  void batchesQueue.push(['accepted'])
+  await Promise.resolve()
+  let closing = batchesQueue.close()
+  await batchesQueue.push(['late'])
+  release()
+  await closing
+
+  expect(calls).toEqual([['accepted']])
+})
+
+it('holds early watcher events until the initial build is complete', async () => {
+  let finishInitialBuild!: () => void
+  let initialBuild = new Promise<void>((resolve) => (finishInitialBuild = resolve))
+  let calls: string[][] = []
+  let batchesQueue = serializeBatches<string>(async (batch) => calls.push(batch), initialBuild)
+
+  let earlyEvent = batchesQueue.push(['changed-during-initial-build'])
+  await Promise.resolve()
+  expect(calls).toEqual([])
+
+  finishInitialBuild()
+  await earlyEvent
+  expect(calls).toEqual([['changed-during-initial-build']])
+})
+
+it('reports callback failures and continues draining accepted batches', async () => {
+  let releaseFirst!: () => void
+  let firstCanFail = new Promise<void>((resolve) => (releaseFirst = resolve))
+  let calls: string[][] = []
+  let errors: unknown[] = []
+  let batchesQueue = serializeBatches<string>(
+    async (batch) => {
+      calls.push(batch)
+      if (calls.length === 1) {
+        await firstCanFail
+        throw new Error('rebuild failed')
+      }
+    },
+    Promise.resolve(),
+    (error) => errors.push(error),
+  )
+
+  let first = batchesQueue.push(['first'])
+  await Promise.resolve()
+  let second = batchesQueue.push(['accepted-during-first'])
+  releaseFirst()
+  await Promise.all([first, second])
+
+  expect(calls).toEqual([['first'], ['accepted-during-first']])
+  expect(errors).toHaveLength(1)
+  expect(errors[0]).toEqual(new Error('rebuild failed'))
+})

packages/@tailwindcss-cli/src/utils/serial-batches.ts

@@ -0,0 +1,46 @@
+export interface SerialBatches<T> {
+  push(batch: T[]): Promise<void>
+  close(): Promise<void>
+}
+
+export function serializeBatches<T>(
+  callback: (batch: T[]) => Promise<void>,
+  startAfter: Promise<void> = Promise.resolve(),
+  onError: (error: unknown) => void = console.error,
+): SerialBatches<T> {
+  let pending: T[] = []
+  let inFlight: Promise<void> | null = null
+  let closed = false
+
+  function push(batch: T[]): Promise<void> {
+    if (closed) return Promise.resolve()
+    pending.push(...batch)
+
+    if (inFlight === null) {
+      inFlight = (async () => {
+        await startAfter
+        while (pending.length > 0) {
+          let next = pending
+          pending = []
+          try {
+            await callback(next)
+          } catch (error) {
+            onError(error)
+          }
+        }
+      })().finally(() => {
+        inFlight = null
+      })
+    }
+
+    return inFlight
+  }
+
+  return {
+    push,
+    async close() {
+      closed = true
+      await inFlight
+    },
+  }
+}