Drain the queue on shutdown instead of tracking watcher swaps

Tracking the swap promise was the wrong shape. It could always be read at a
moment when the in-flight rebuild had not assigned its swap yet, so shutdown
saw the old one settle, closed the queue, and dropped whatever the later
generation flushed. Each fix narrowed that window without closing it.

A full rebuild runs inside the queue, so the queue already knows when one is in
flight — including a rebuild that has not yet replaced its watchers. Add
`drain()`, wait for it before tearing the watchers down, and close only after
their flush has landed.

That removes the swap bookkeeping entirely.

Both orderings are pinned: cleanup must not run while a rebuild is in flight,
and what the watchers flush on the way out must still be processed. Against
no-drain the cleanup runs early; against closing first the flush is ignored.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Lcj4iQ3fBxMwAu2rf4zLbC

Author: Michael Glass

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

@@ -122,63 +122,55 @@ it('writes the newest change last when an earlier rebuild is slower', async () =
   expect(written).toEqual(['older-change', 'newer-change'])
 })
 
-it('does not close the queue while a watcher swap is still flushing', async () => {
-  // A rebuild swaps the watcher generation, and the old generation flushes what
-  // it collected as it is torn down. If shutdown closes the queue first, those
-  // files land in a closed queue and the process exits with stale CSS.
-  let closed = false
-  let flushed: string[] = []
-  let finishSwap!: () => void
-  let swap = new Promise<void>((resolve) => (finishSwap = resolve)).then(() => {
-    flushed.push('collected-during-swap')
+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')
   })
 
-  let shutdown = shutdownWatchMode(() => swap, [], {
-    async close() {
-      closed = true
-    },
-  })
+  void queue.push(['change'])
   await nextTask()
-  expect(closed).toBe(false)
 
-  finishSwap()
+  let shutdown = shutdownWatchMode(
+    [
+      async () => {
+        order.push('cleanup')
+      },
+    ],
+    queue,
+  )
+  await nextTask()
+  expect(order).toEqual(['rebuild:start'])
+
+  finishRebuild()
   await shutdown
 
-  expect(flushed).toEqual(['collected-during-swap'])
-  expect(closed).toBe(true)
+  expect(order).toEqual(['rebuild:start', 'rebuild:end', 'cleanup'])
 })
 
-it('waits for a watcher swap that starts after shutdown begins', async () => {
-  // A rebuild already in flight can swap watchers while we are shutting down.
-  // Reading the swap once captures whichever was current when stdin closed, and
-  // closes the queue while the later one is still flushing.
-  let closed = false
-  let flushed: string[] = []
-  let finishFirstSwap!: () => void
-  let firstSwap = new Promise<void>((resolve) => (finishFirstSwap = resolve)).then(() => {
-    flushed.push('first-swap')
-  })
-  let finishSecondSwap!: () => void
-  let secondSwap = new Promise<void>((resolve) => (finishSecondSwap = resolve)).then(() => {
-    flushed.push('second-swap')
-  })
-
-  let currentSwap: Promise<unknown> = firstSwap
-  let shutdown = shutdownWatchMode(() => currentSwap, [], {
-    async close() {
-      closed = true
-    },
+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)
   })
 
-  // The in-flight rebuild starts its own swap while shutdown is already waiting.
-  currentSwap = secondSwap
-  finishFirstSwap()
-  await nextTask()
-  expect(closed).toBe(false)
-
-  finishSecondSwap()
-  await shutdown
+  await shutdownWatchMode(
+    [
+      async () => {
+        void queue.push(['flushed-on-shutdown'])
+      },
+    ],
+    queue,
+  )
 
-  expect(flushed).toEqual(['first-swap', 'second-swap'])
-  expect(closed).toBe(true)
+  expect(processed).toEqual([['flushed-on-shutdown']])
 })

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

@@ -327,10 +327,6 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
 
   let [compiler, scanner] = await handleError(() => createCompiler(input, I))
   let cleanupWatchers: (() => Promise<void>)[] = []
-  // A rebuild swaps the watcher generation. Shutdown must not close the queue
-  // while that swap is in flight, or the files the old generation flushes are
-  // pushed into a closed queue and silently dropped.
-  let watcherSwap: Promise<unknown> = Promise.resolve()
   let finishInitialBuild!: () => void
   let initialBuildFinished = new Promise<void>((resolve) => (finishInitialBuild = resolve))
   let eventBatches: SerialBatches<string> | null = null
@@ -418,8 +414,7 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
             DEBUG && I.start('Cleanup old watchers')
             let previousCleanups = cleanupWatchers.splice(0)
             cleanupWatchers.push(newWatchers.cleanup)
-            watcherSwap = Promise.all(previousCleanups.map((cleanup) => cleanup()))
-            await watcherSwap
+            await Promise.all(previousCleanups.map((cleanup) => cleanup()))
             DEBUG && I.end('Cleanup old watchers')
 
             // Re-compile the CSS
@@ -515,7 +510,7 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
     // disable this behavior with `--watch=always`.
     if (args['--watch'] !== 'always') {
       process.stdin.on('end', () => {
-        shutdownWatchMode(() => watcherSwap, cleanupWatchers, eventBatches).then(
+        shutdownWatchMode(cleanupWatchers, eventBatches).then(
           () => process.exit(0),
           () => process.exit(1),
         )
@@ -712,25 +707,16 @@ export async function handle(args: Result<ReturnType<typeof options>>) {
 // polling mode as well.
 /// Shut watch mode down in an order that cannot drop collected files.
 ///
-/// A rebuild swaps the watcher generation, and the old generation flushes what it
-/// collected as it is torn down. Closing the queue before that flush lands means the
-/// files are pushed into a closed queue and ignored, so the process exits successfully
-/// with stale CSS. Wait for an in-flight swap first, then the current generation, and
-/// only then close.
+/// 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(
-  pendingSwap: () => Promise<unknown>,
   cleanups: (() => Promise<void>)[],
-  batches: { close(): Promise<void> } | null,
+  batches: { drain(): Promise<void>; close(): Promise<void> } | null,
 ) {
-  // A rebuild already in flight can start its own swap while we are shutting
-  // down, so read the current one each time round rather than capturing it
-  // once. Capturing it once waits for the swap that happened to be current when
-  // stdin closed and closes the queue while a later one is still flushing.
-  let awaited: Promise<unknown> | undefined
-  while (awaited !== pendingSwap()) {
-    awaited = pendingSwap()
-    await awaited
-  }
+  await batches?.drain()
   await Promise.all(cleanups.map((cleanup) => cleanup()))
   await batches?.close()
 }

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

@@ -1,5 +1,9 @@
 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>
 }
 
@@ -57,6 +61,13 @@ export function serializeBatches<T>(
 
   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