Merge 3148fad1a1 into fa81d697fe

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

@@ -0,0 +1,176 @@
+import { expect, it } from 'vitest'
+import { serializeBatches } from '../../utils/serial-batches'
+import { createWatchers, filterChangedFiles, shutdownWatchMode } from './index'
+
+type WatchEvent = { type: 'create' | 'update' | 'delete'; path: string }
+type WatchCallback = (error: Error | null, events: WatchEvent[]) => 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('removes duplicate output and map events from a coalesced batch', () => {
+  expect(
+    filterChangedFiles(
+      ['output.css', 'source.html', 'output.css', 'output.css.map'],
+      'output.css',
+      'output.css.map',
+    ),
+  ).toEqual(['source.html'])
+})
+
+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)
+
+  await fake.callbacks[0](null, [{ type: 'delete', path: 'last-change' }])
+  await generation.cleanup()
+  await queue.close()
+
+  expect(calls).toEqual([['last-change']])
+})
+
+it('waits for an entered watcher callback before shutdown flushes changes', async () => {
+  let releaseLstat!: () => void
+  let lstatCanFinish = new Promise<void>((resolve) => (releaseLstat = resolve))
+  let calls: string[][] = []
+  let queue = serializeBatches<string>(async (files) => {
+    calls.push(files)
+  })
+  let fake = fakeWatcher()
+  let generation = await createWatchers(
+    ['/watch'],
+    async () => {},
+    queue,
+    fake.watcher,
+    async () => {
+      await lstatCanFinish
+      return { isFile: () => true, isSymbolicLink: () => false }
+    },
+  )
+
+  let callback = fake.callbacks[0](null, [{ type: 'update', path: 'delayed-change' }])
+  let cleanup = generation.cleanup()
+  await nextTask()
+  expect(calls).toEqual([])
+
+  releaseLstat()
+  await Promise.all([callback, cleanup])
+  await queue.close()
+  expect(calls).toEqual([['delayed-change']])
+})
+
+it('writes the newest change last when an earlier rebuild is slower', async () => {
+  // The reported bug: a rebuild for an older change finished *after* the rebuild
+  // for a newer one and overwrote it, leaving stale CSS on disk. The serial
+  // batches tests pin the scheduling on its own; this pins the outcome through
+  // the watcher wiring, which is the shape the bug was reported in.
+  let written: string[] = []
+  let rebuildCount = 0
+  let startFirstRebuild!: () => void
+  let firstRebuildStarted = new Promise<void>((resolve) => (startFirstRebuild = resolve))
+  let releaseSlowRebuild!: () => void
+  let slowRebuildCanFinish = new Promise<void>((resolve) => (releaseSlowRebuild = resolve))
+
+  let queue = serializeBatches<string>(async (files) => {
+    // Only the *first* rebuild is slow. Counting rebuilds rather than writes
+    // matters: the first rebuild is suspended below, so a write-count check
+    // would also suspend the second one and the test would pass unserialized.
+    if (rebuildCount++ === 0) {
+      startFirstRebuild()
+      await slowRebuildCanFinish
+    }
+    written.push(files.at(-1)!)
+  })
+  let fake = fakeWatcher()
+  await createWatchers(
+    ['/watch'],
+    async () => {},
+    queue,
+    fake.watcher,
+    async () => ({
+      isFile: () => true,
+      isSymbolicLink: () => false,
+    }),
+  )
+
+  await fake.callbacks[0](null, [{ type: 'update', path: 'older-change' }])
+  await firstRebuildStarted
+  await fake.callbacks[0](null, [{ type: 'update', path: 'newer-change' }])
+  await nextTask()
+
+  releaseSlowRebuild()
+  await queue.close()
+
+  expect(written).toEqual(['older-change', 'newer-change'])
+})
+
+it('waits for an in-flight rebuild before tearing down the watchers', async () => {
+  // A full rebuild runs inside the queue and swaps the watcher generation while
+  // it does. Tearing the watchers down first races that swap.
+  let order: string[] = []
+  let finishRebuild!: () => void
+  let rebuildDone = new Promise<void>((resolve) => (finishRebuild = resolve))
+  let queue = serializeBatches<string>(async () => {
+    order.push('rebuild:start')
+    await rebuildDone
+    order.push('rebuild:end')
+  })
+
+  void queue.push(['change'])
+  await nextTask()
+
+  let shutdown = shutdownWatchMode(
+    [
+      async () => {
+        order.push('cleanup')
+      },
+    ],
+    queue,
+  )
+  await nextTask()
+  expect(order).toEqual(['rebuild:start'])
+
+  finishRebuild()
+  await shutdown
+
+  expect(order).toEqual(['rebuild:start', 'rebuild:end', 'cleanup'])
+})
+
+it('processes what the watchers flush on the way out', async () => {
+  // The watchers flush what they collected as they are torn down. That has to
+  // land in a queue that is still open, or it is dropped and we exit as if all
+  // was well.
+  let processed: string[][] = []
+  let queue = serializeBatches<string>(async (files) => {
+    processed.push(files)
+  })
+
+  await shutdownWatchMode(
+    [
+      async () => {
+        void queue.push(['flushed-on-shutdown'])
+      },
+    ],
+    queue,
+  )
+
+  expect(processed).toEqual([['flushed-on-shutdown']])
+})

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,6 +327,13 @@ 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 setEventHandler!: (handler: (files: string[]) => Promise<void>) => void
+  let eventHandler = new Promise<(files: string[]) => Promise<void>>(
+    (resolve) => (setEventHandler = resolve),
+  )
 
   // Watch for changes
   if (args['--watch'] && pollInterval === false) {
@@ -333,12 +341,18 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
     // 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(
+      async (files) => (await 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.
-          if (files.length === 1 && files[0] === args['--output']) return
+          // Ignore our own writes so they don't trigger another rebuild.
+          files = filterChangedFiles(files, args['--output'], args['--map'])
+          if (files.length === 0) return
 
           using I = new Instrumentation()
           DEBUG && I.start('[@tailwindcss/cli] (watcher)')
@@ -388,16 +402,21 @@ 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
+            // Clear old watchers. Register the new generation *before* awaiting
+            // the old one, so shutdown never observes an empty cleanup list.
             DEBUG && I.start('Cleanup old watchers')
-            await Promise.all(cleanupWatchers.splice(0).map((cleanup) => cleanup()))
+            let previousCleanups = cleanupWatchers.splice(0)
+            cleanupWatchers.push(newWatchers.cleanup)
+            await Promise.all(previousCleanups.map((cleanup) => cleanup()))
             DEBUG && I.end('Cleanup old watchers')
 
-            cleanupWatchers.push(newCleanupFunction)
-
             // Re-compile the CSS
             DEBUG && I.start('Build CSS')
             compiledCss = compiler.build(candidates)
@@ -481,14 +500,17 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
           let end = process.hrtime.bigint()
           if (!args['--silent']) eprintln(`Done in ${formatDuration(end - start)}`)
         }
-      }),
+      },
+      eventBatches,
     )
+    setEventHandler(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(
+        shutdownWatchMode(cleanupWatchers, eventBatches).then(
           () => process.exit(0),
           () => process.exit(1),
         )
@@ -515,6 +537,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)}`)
@@ -682,6 +705,22 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
 // Load `@parcel/watcher` lazily so a missing or broken native binding only
 // affects `--watch` (without `--poll`), instead of crashing one-off builds and
 // polling mode as well.
+/// Shut watch mode down in an order that cannot drop collected files.
+///
+/// A full rebuild runs inside the queue and swaps the watcher generation while it
+/// does, so draining the queue is what waits for a rebuild to finish — including
+/// one that has not yet replaced its watchers. Only then is it safe to tear the
+/// watchers down, because the files they flush on the way out have to land in a
+/// queue that is still open. Closing first drops them and exits with stale CSS.
+export async function shutdownWatchMode(
+  cleanups: (() => Promise<void>)[],
+  batches: { drain(): Promise<void>; close(): Promise<void> } | null,
+) {
+  await batches?.drain()
+  await Promise.all(cleanups.map((cleanup) => cleanup()))
+  await batches?.close()
+}
+
 async function loadWatcher(): Promise<typeof import('@parcel/watcher')> {
   try {
     return (await import('@parcel/watcher')).default
@@ -693,9 +732,29 @@ async function loadWatcher(): Promise<typeof import('@parcel/watcher')> {
   }
 }
 
-async function createWatchers(dirs: string[], cb: (files: string[]) => void) {
-  let watcher = await loadWatcher()
+type WatchEvent = { type: 'create' | 'update' | 'delete'; path: string }
+type WatcherBackend = {
+  subscribe(
+    directory: string,
+    callback: (error: Error | null, events: WatchEvent[]) => Promise<void>,
+  ): Promise<{ unsubscribe(): void | Promise<void> }>
+}
 
+export async function createWatchers(
+  dirs: string[],
+  cb: (files: string[]) => Promise<void>,
+  batches: SerialBatches<string>,
+  watcher?: WatcherBackend,
+  lstat: (path: string) => Promise<Pick<Stats, 'isFile' | 'isSymbolicLink'>> = fs.lstat,
+) {
+  if (!watcher) {
+    let nativeWatcher = await loadWatcher()
+    watcher = {
+      subscribe(directory, callback) {
+        return nativeWatcher.subscribe(directory, callback)
+      },
+    }
+  }
   // 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.
@@ -730,6 +789,7 @@ async function createWatchers(dirs: string[], cb: (files: string[]) => void) {
 
   // Keep track of the debounce queue to avoid multiple rebuilds.
   let debounceQueue = new Disposables()
+  let activeCallbacks = new Set<Promise<void>>()
 
   // A changed file can be watched by multiple watchers, but we only want to
   // handle the file once. We debounce the handle function with the collected
@@ -740,50 +800,60 @@ 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)
     })
   }
 
   // Setup a watcher for every directory.
   for (let dir of dirs) {
-    let { unsubscribe } = await watcher.subscribe(dir, async (err, events) => {
-      // Whenever an error occurs we want to let the user know about it but we
-      // want to keep watching for changes.
-      if (err) {
-        console.error(err)
-        return
-      }
+    let { unsubscribe } = await watcher.subscribe(dir, (err, events) => {
+      let callback = (async () => {
+        // Whenever an error occurs we want to let the user know about it but we
+        // want to keep watching for changes.
+        if (err) {
+          console.error(err)
+          return
+        }
 
-      await Promise.all(
-        events.map(async (event) => {
-          // When a file is deleted, a rebuild should be triggered such that we
-          // can figure out whether this file must trigger a fresh build or not.
-          //
-          // If it must trigger a fresh build, then we will temporarily end up
-          // in a broken state, but an error will be shown to the user. Once the
-          // user resolves the issue, the CLI will recover.
-          if (event.type === 'delete') {
+        await Promise.all(
+          events.map(async (event) => {
+            // When a file is deleted, a rebuild should be triggered such that we
+            // can figure out whether this file must trigger a fresh build or not.
+            //
+            // If it must trigger a fresh build, then we will temporarily end up
+            // in a broken state, but an error will be shown to the user. Once the
+            // user resolves the issue, the CLI will recover.
+            if (event.type === 'delete') {
+              files.add(event.path)
+              return
+            }
+
+            // Ignore directory changes. We only care about file changes
+            let stats: Stats | null = null
+            try {
+              stats = (await lstat(event.path)) as Stats
+            } catch {}
+            if (!stats?.isFile() && !stats?.isSymbolicLink()) {
+              return
+            }
+
+            // Track the changed file.
             files.add(event.path)
-            return
-          }
+          }),
+        )
 
-          // Ignore directory changes. We only care about file changes
-          let stats: Stats | null = null
-          try {
-            stats = await fs.lstat(event.path)
-          } catch {}
-          if (!stats?.isFile() && !stats?.isSymbolicLink()) {
-            return
-          }
+        // Handle the tracked files at some point in the future.
+        await enqueueCallback()
+      })()
 
-          // Track the changed file.
-          files.add(event.path)
-        }),
+      activeCallbacks.add(callback)
+      void callback.then(
+        () => activeCallbacks.delete(callback),
+        () => activeCallbacks.delete(callback),
       )
-
-      // Handle the tracked files at some point in the future.
-      await enqueueCallback()
+      return callback
     })
 
     // Ensure we cleanup the watcher when we're done.
@@ -791,12 +861,29 @@ 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 Promise.all(activeCallbacks)
+      await debounceQueue.dispose()
+      if (files.size > 0) {
+        let batch = Array.from(files)
+        files.clear()
+        void batches.push(batch)
+      }
+    },
   }
 }
 
+export function filterChangedFiles(
+  files: string[],
+  output: string | null,
+  map: boolean | string,
+): string[] {
+  return files.filter((file) => file !== output && file !== map)
+}
+
 function getRebuildStrategy(
   files: string[],
   fullRebuildPaths: string[],

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

@@ -0,0 +1,188 @@
+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('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'))
+})
+
+it('drains work queued by a callback promise reaction', async () => {
+  let finishFirst!: () => void
+  let firstCallback = new Promise<void>((resolve) => (finishFirst = resolve))
+  let calls: string[][] = []
+  let queue = serializeBatches<string>((batch) => {
+    calls.push(batch)
+    return calls.length === 1 ? firstCallback : Promise.resolve()
+  })
+
+  let first = queue.push(['first'])
+  let reaction = firstCallback.then(() => queue.push(['queued-by-reaction']))
+  finishFirst()
+  await Promise.all([first, reaction])
+
+  expect(calls).toEqual([['first'], ['queued-by-reaction']])
+})
+
+it('deduplicates repeated items across pending batches', async () => {
+  let release!: () => void
+  let blocked = new Promise<void>((resolve) => (release = resolve))
+  let calls: string[][] = []
+  let queue = serializeBatches<string>(async (batch) => {
+    calls.push(batch)
+    if (calls.length === 1) await blocked
+  })
+
+  void queue.push(['first'])
+  await Promise.resolve()
+  void queue.push(['same', 'same'])
+  void queue.push(['same'])
+  release()
+  await queue.close()
+
+  expect(calls).toEqual([['first'], ['same']])
+})
+
+it('reports a rejected initial barrier once and settles', async () => {
+  let errors: unknown[] = []
+  let calls: string[][] = []
+  let queue = serializeBatches<string>(
+    async (batch) => {
+      calls.push(batch)
+    },
+    Promise.reject(new Error('initial build failed')),
+    (error) => errors.push(error),
+  )
+
+  await queue.push(['change'])
+  await queue.close()
+
+  expect(calls).toEqual([])
+  expect(errors).toEqual([new Error('initial build failed')])
+})
+
+it('drains in-flight work but stays open for more', async () => {
+  let release!: () => void
+  let canFinish = new Promise<void>((resolve) => (release = resolve))
+  let calls: string[][] = []
+  let queue = serializeBatches<string>(async (batch) => {
+    calls.push(batch)
+    if (calls.length === 1) await canFinish
+  })
+
+  void queue.push(['first'])
+  await Promise.resolve()
+
+  let drained = queue.drain()
+  release()
+  await drained
+  expect(calls).toEqual([['first']])
+
+  // Draining must not close the queue — a shutdown drains before the watchers
+  // have flushed what they collected, and that work still has to be accepted.
+  await queue.push(['after-drain'])
+  expect(calls).toEqual([['first'], ['after-drain']])
+})
+
+it('drain returns immediately when nothing is in flight', async () => {
+  let calls: string[][] = []
+  let queue = serializeBatches<string>(async (batch) => {
+    calls.push(batch)
+  })
+
+  await queue.drain()
+  expect(calls).toEqual([])
+})

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

@@ -0,0 +1,76 @@
+export interface SerialBatches<T> {
+  push(batch: T[]): Promise<void>
+  /// Wait for in-flight work to finish, without closing. Callers that need the
+  /// queue to still accept work afterwards — a shutdown that has yet to flush
+  /// what the watchers collected — drain first and close last.
+  drain(): 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 = new Set<T>()
+  let inFlight: Promise<void> | null = null
+  let closed = false
+  let startErrorReported = false
+
+  function report(error: unknown) {
+    try {
+      onError(error)
+    } catch {}
+  }
+
+  function startDrain(): Promise<void> {
+    inFlight = (async () => {
+      try {
+        await startAfter
+      } catch (error) {
+        if (!startErrorReported) {
+          startErrorReported = true
+          report(error)
+        }
+        pending.clear()
+        return
+      }
+      while (pending.size > 0) {
+        let next = Array.from(pending)
+        pending.clear()
+        try {
+          await callback(next)
+        } catch (error) {
+          report(error)
+        }
+      }
+    })().finally(() => {
+      inFlight = null
+      if (pending.size > 0) return startDrain()
+    })
+
+    return inFlight
+  }
+
+  function push(batch: T[]): Promise<void> {
+    if (closed) return Promise.resolve()
+    for (let item of batch) pending.add(item)
+
+    return inFlight ?? startDrain()
+  }
+
+  return {
+    push,
+    async drain() {
+      // `inFlight` is re-chained by finalization when work arrived mid-drain, so
+      // loop until it settles rather than awaiting whichever promise is current.
+      while (inFlight) {
+        await inFlight
+      }
+    },
+    async close() {
+      closed = true
+      await inFlight
+    },
+  }
+}