Serialize filesystem watcher batch flush operations (#17602)

* Serialize filesystem watcher batch flush operations

- Prevent dropped events during rapid concurrent file changes
- Queue and drain follow-up batches to preserve event ordering
- Cancel pending batch work when watchers are torn down

* Prevent queued batch drain while debounce timer is armed

An armed timer means the debounce window is still open. Drain only after
the window closes to avoid splitting related filesystem events across
separate payloads.

* Remove redundant batch timer cleanup

Rely on cancelLocalBatchFlush to handle the batch timer
teardown, eliminating duplicate logic in the watcher
cleanup path.

src/main/ipc/filesystem-watcher-batch-control.ts

@@ -0,0 +1,37 @@
+import type { Event as WatcherEvent } from '@parcel/watcher'
+import type { WatchedRoot } from './filesystem-watcher-wsl'
+
+export type DebouncedBatch = {
+  events: WatcherEvent[]
+  overflowed: boolean
+  timer: ReturnType<typeof setTimeout> | null
+  firstEventAt: number
+  flushInFlight: boolean
+  flushQueued: boolean
+  cancelled: boolean
+}
+
+export function createDebouncedBatch(): DebouncedBatch {
+  return {
+    events: [],
+    overflowed: false,
+    timer: null,
+    firstEventAt: 0,
+    flushInFlight: false,
+    flushQueued: false,
+    cancelled: false
+  }
+}
+
+/** Cancel pending and queued flush work when a root is torn down. */
+export function cancelLocalBatchFlush(root: WatchedRoot): void {
+  root.batch.cancelled = true
+  root.batch.flushQueued = false
+  if (root.batch.timer) {
+    clearTimeout(root.batch.timer)
+    root.batch.timer = null
+  }
+  root.batch.events = []
+  root.batch.overflowed = false
+  root.batch.firstEventAt = 0
+}

src/main/ipc/filesystem-watcher-listener-lifecycle.ts

@@ -7,6 +7,7 @@ import {
   UNWATCHABLE_ROOT_CACHE_MAX,
   watcherLifecycleState
 } from './filesystem-watcher-lifecycle-state'
+import { cancelLocalBatchFlush } from './filesystem-watcher-batch-control'
 
 export function rememberUnwatchableRoot(rootKey: string): void {
   const { unwatchableRoots } = watcherLifecycleState
@@ -203,9 +204,7 @@ function cleanupLocalWatchersForSender(senderId: number): void {
         clearTimeout(pending)
         watcherLifecycleState.pendingTeardowns.delete(key)
       }
-      if (watchedRoot.batch.timer) {
-        clearTimeout(watchedRoot.batch.timer)
-      }
+      cancelLocalBatchFlush(watchedRoot)
       trackDetachedLocalUnsubscribe(key, watchedRoot)
       watcherLifecycleState.watchedRoots.delete(key)
     }

src/main/ipc/filesystem-watcher-local-events.test.ts

@@ -0,0 +1,171 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import type { Event as WatcherEvent } from '@parcel/watcher'
+import type { FsChangedPayload } from '../../shared/filesystem-entry-types'
+import { WATCH_BATCH_TRAILING_MS } from '../../shared/filesystem-watch-batch-window'
+
+const { statMock, subscribeMock } = vi.hoisted(() => ({
+  statMock: vi.fn(),
+  subscribeMock: vi.fn()
+}))
+
+vi.mock('fs/promises', () => ({ stat: statMock }))
+vi.mock('./parcel-watcher-process', () => ({ subscribeViaWatcherProcess: subscribeMock }))
+
+import { createLocalWatcher } from './filesystem-watcher-local-events'
+
+function deferred<T>(): { promise: Promise<T>; resolve: (value: T) => void } {
+  let resolve!: (value: T) => void
+  const promise = new Promise<T>((nextResolve) => {
+    resolve = nextResolve
+  })
+  return { promise, resolve }
+}
+
+async function flushMicrotasks(): Promise<void> {
+  for (let i = 0; i < 6; i++) {
+    await Promise.resolve()
+  }
+}
+
+type Sender = { isDestroyed: () => boolean; send: ReturnType<typeof vi.fn> }
+
+describe('local filesystem watcher flush serialization', () => {
+  let watcherCallback: ((error: Error | null, events: WatcherEvent[]) => void) | undefined
+  let sender: Sender
+
+  beforeEach(() => {
+    vi.useFakeTimers()
+    statMock.mockReset()
+    subscribeMock.mockReset()
+    watcherCallback = undefined
+    sender = { isDestroyed: () => false, send: vi.fn() }
+    subscribeMock.mockImplementation(async (_root: string, callback: typeof watcherCallback) => {
+      watcherCallback = callback
+      return { unsubscribe: vi.fn() }
+    })
+  })
+
+  it('serializes an inflight flush and drains one follow-up without overlap', async () => {
+    const firstStat = deferred<{ isDirectory: () => boolean }>()
+    const secondStat = deferred<{ isDirectory: () => boolean }>()
+    statMock.mockReturnValueOnce(firstStat.promise).mockReturnValueOnce(secondStat.promise)
+    const root = await createLocalWatcher('/repo', '/repo')
+    root.listeners.set(1, sender as never)
+    const firstPath = '/repo/first.ts'
+    const secondPath = '/repo/second.ts'
+
+    watcherCallback?.(null, [{ type: 'update', path: firstPath }])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS)
+    await flushMicrotasks()
+    expect(statMock).toHaveBeenCalledTimes(1)
+
+    watcherCallback?.(null, [{ type: 'update', path: secondPath }])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS)
+    await flushMicrotasks()
+    expect(statMock).toHaveBeenCalledTimes(1)
+    expect(sender.send).not.toHaveBeenCalled()
+
+    firstStat.resolve({ isDirectory: () => true })
+    await flushMicrotasks()
+    expect(statMock).toHaveBeenCalledTimes(2)
+    expect(sender.send).toHaveBeenCalledTimes(1)
+
+    secondStat.resolve({ isDirectory: () => false })
+    await flushMicrotasks()
+    expect(sender.send).toHaveBeenCalledTimes(2)
+    expect((sender.send.mock.calls[1][1] as FsChangedPayload).events).toEqual([
+      { kind: 'update', absolutePath: secondPath, isDirectory: false }
+    ])
+  })
+
+  it('coalesces a queued storm while preserving delete-before-create ordering', async () => {
+    const firstStat = deferred<{ isDirectory: () => boolean }>()
+    const createStat = deferred<{ isDirectory: () => boolean }>()
+    statMock.mockReturnValueOnce(firstStat.promise).mockReturnValueOnce(createStat.promise)
+    const root = await createLocalWatcher('/repo', '/repo')
+    root.listeners.set(1, sender as never)
+    const firstPath = '/repo/first.ts'
+    const transientPath = '/repo/transient.ts'
+    const replacedPath = '/repo/replaced.ts'
+
+    watcherCallback?.(null, [{ type: 'update', path: firstPath }])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS)
+    await flushMicrotasks()
+    watcherCallback?.(null, [
+      { type: 'create', path: transientPath },
+      { type: 'delete', path: transientPath },
+      { type: 'delete', path: replacedPath },
+      { type: 'create', path: replacedPath }
+    ])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS)
+    await flushMicrotasks()
+    expect(statMock).toHaveBeenCalledTimes(1)
+
+    firstStat.resolve({ isDirectory: () => true })
+    await flushMicrotasks()
+    expect(statMock).toHaveBeenCalledTimes(2)
+    createStat.resolve({ isDirectory: () => true })
+    await flushMicrotasks()
+    expect((sender.send.mock.calls[1][1] as FsChangedPayload).events).toEqual([
+      { kind: 'delete', absolutePath: replacedPath },
+      { kind: 'create', absolutePath: replacedPath, isDirectory: true }
+    ])
+  })
+
+  it('drops queued events when the last listener is removed', async () => {
+    const firstStat = deferred<{ isDirectory: () => boolean }>()
+    const root = await createLocalWatcher('/repo', '/repo')
+    root.listeners.set(1, sender as never)
+    statMock.mockReturnValueOnce(firstStat.promise)
+
+    watcherCallback?.(null, [{ type: 'update', path: '/repo/first.ts' }])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS)
+    await flushMicrotasks()
+    watcherCallback?.(null, [{ type: 'update', path: '/repo/queued.ts' }])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS)
+    await flushMicrotasks()
+    root.listeners.clear()
+
+    firstStat.resolve({ isDirectory: () => true })
+    await flushMicrotasks()
+    expect(statMock).toHaveBeenCalledTimes(1)
+    expect(sender.send).not.toHaveBeenCalled()
+  })
+
+  it('leaves an open debounce window to the armed timer instead of draining early', async () => {
+    const firstStat = deferred<{ isDirectory: () => boolean }>()
+    const secondStat = deferred<{ isDirectory: () => boolean }>()
+    statMock.mockReturnValueOnce(firstStat.promise).mockReturnValueOnce(secondStat.promise)
+    const root = await createLocalWatcher('/repo', '/repo')
+    root.listeners.set(1, sender as never)
+    const transientPath = '/repo/transient.ts'
+    const otherPath = '/repo/other.ts'
+
+    watcherCallback?.(null, [{ type: 'update', path: '/repo/first.ts' }])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS)
+    await flushMicrotasks()
+
+    // Queue an event mid-flush, then settle the flush before its debounce window closes.
+    watcherCallback?.(null, [{ type: 'create', path: transientPath }])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS - 50)
+    firstStat.resolve({ isDirectory: () => true })
+    await flushMicrotasks()
+    expect(sender.send).toHaveBeenCalledTimes(1)
+    expect(statMock).toHaveBeenCalledTimes(1)
+
+    // The still-open window coalesces the create away instead of emitting a transient one.
+    watcherCallback?.(null, [
+      { type: 'delete', path: transientPath },
+      { type: 'update', path: otherPath }
+    ])
+    vi.advanceTimersByTime(WATCH_BATCH_TRAILING_MS)
+    await flushMicrotasks()
+    secondStat.resolve({ isDirectory: () => false })
+    await flushMicrotasks()
+
+    expect(sender.send).toHaveBeenCalledTimes(2)
+    expect((sender.send.mock.calls[1][1] as FsChangedPayload).events).toEqual([
+      { kind: 'update', absolutePath: otherPath, isDirectory: false }
+    ])
+  })
+})

src/main/ipc/filesystem-watcher-local-events.ts

@@ -16,6 +16,7 @@ import {
   retainLocalWatcherPhysicalFailure,
   trackDetachedLocalUnsubscribe
 } from './filesystem-watcher-listener-lifecycle'
+import { createDebouncedBatch } from './filesystem-watcher-batch-control'
 
 // ── Event coalescing ─────────────────────────────────────────────────
 // Why: keep the last event per path in a flush window; delete→create emits both (delete cleans the subtree, create refreshes the parent), create→delete is dropped (§4.4).
@@ -94,50 +95,92 @@ function emitOverflowPayload(root: WatchedRoot): void {
 }
 
 async function flushBatch(root: WatchedRoot): Promise<void> {
+  if (root.batch.cancelled) {
+    return
+  }
+  if (root.batch.flushInFlight) {
+    root.batch.flushQueued = true
+    return
+  }
+
+  root.batch.flushInFlight = true
+  if (root.batch.timer) {
+    clearTimeout(root.batch.timer)
+    root.batch.timer = null
+  }
   const overflowed = root.batch.overflowed
   const rawEvents = root.batch.events.splice(0)
   root.batch.overflowed = false
-  root.batch.timer = null
   root.batch.firstEventAt = 0
 
-  if ((rawEvents.length === 0 && !overflowed) || root.listeners.size === 0) {
-    return
-  }
+  try {
+    if ((rawEvents.length === 0 && !overflowed) || root.listeners.size === 0) {
+      return
+    }
 
-  if (overflowed || rawEvents.length > MAX_BATCHED_WATCHER_EVENTS) {
-    // Why: deletion storms can be too large to coalesce/stat per path; one overflow asks the renderer for the same conservative refresh.
-    emitOverflowPayload(root)
-    return
-  }
-
-  const coalesced = coalesceEvents(rawEvents)
-
-  const events: FsChangeEvent[] = await Promise.all(
-    coalesced.map(async (evt) => {
-      // Why: a deleted path can't be stat'd; leave isDirectory undefined and let the renderer infer from dirCache.
-      const isDirectory = evt.type === 'delete' ? undefined : await tryStatIsDirectory(evt.path)
-
-      return {
-        kind: evt.type,
-        absolutePath: evt.path,
-        isDirectory
+    if (overflowed || rawEvents.length > MAX_BATCHED_WATCHER_EVENTS) {
+      // Why: deletion storms can be too large to coalesce/stat per path; one overflow asks the renderer for the same conservative refresh.
+      if (!root.batch.cancelled) {
+        emitOverflowPayload(root)
       }
-    })
-  )
+      return
+    }
 
-  const payload: FsChangedPayload = {
-    worktreePath: root.rootPath,
-    events
-  }
+    const coalesced = coalesceEvents(rawEvents)
 
-  for (const [, wc] of root.listeners) {
-    if (!wc.isDestroyed()) {
-      wc.send('fs:changed', payload)
+    const events: FsChangeEvent[] = await Promise.all(
+      coalesced.map(async (evt) => {
+        // Why: a deleted path can't be stat'd; leave isDirectory undefined and let the renderer infer from dirCache.
+        const isDirectory = evt.type === 'delete' ? undefined : await tryStatIsDirectory(evt.path)
+
+        return {
+          kind: evt.type,
+          absolutePath: evt.path,
+          isDirectory
+        }
+      })
+    )
+
+    if (root.batch.cancelled || root.listeners.size === 0) {
+      return
+    }
+
+    const payload: FsChangedPayload = {
+      worktreePath: root.rootPath,
+      events
+    }
+
+    for (const [, wc] of root.listeners) {
+      if (!wc.isDestroyed()) {
+        wc.send('fs:changed', payload)
+      }
+    }
+  } finally {
+    root.batch.flushInFlight = false
+    if (root.batch.flushQueued) {
+      root.batch.flushQueued = false
+      if (
+        !root.batch.cancelled &&
+        // Why: an armed timer still owns its debounce window; draining here would split related events across payloads.
+        !root.batch.timer &&
+        (root.batch.events.length > 0 || root.batch.overflowed)
+      ) {
+        // Drain the queued batch only after the current payload has settled,
+        // preserving watcher event ordering without dropping a storm tail.
+        void flushBatch(root)
+      }
     }
   }
 }
 
 export function scheduleLocalBatchFlush(root: WatchedRoot): void {
+  if (root.batch.cancelled) {
+    return
+  }
+  if (root.batch.flushInFlight) {
+    root.batch.flushQueued = true
+  }
+
   const now = Date.now()
 
   if (root.batch.firstEventAt === 0) {
@@ -148,6 +191,7 @@ export function scheduleLocalBatchFlush(root: WatchedRoot): void {
   if (now - root.batch.firstEventAt >= WATCH_BATCH_MAX_WAIT_MS) {
     if (root.batch.timer) {
       clearTimeout(root.batch.timer)
+      root.batch.timer = null
     }
     void flushBatch(root)
     return
@@ -157,7 +201,11 @@ export function scheduleLocalBatchFlush(root: WatchedRoot): void {
   if (root.batch.timer) {
     clearTimeout(root.batch.timer)
   }
-  root.batch.timer = setTimeout(() => void flushBatch(root), WATCH_BATCH_TRAILING_MS)
+  // Why: clear the handle as it fires so `batch.timer` means "a debounce window is still open", which gates the queued drain.
+  root.batch.timer = setTimeout(() => {
+    root.batch.timer = null
+    void flushBatch(root)
+  }, WATCH_BATCH_TRAILING_MS)
 }
 
 // ── Watcher creation ─────────────────────────────────────────────────
@@ -170,7 +218,7 @@ export async function createLocalWatcher(
   const root: WatchedRoot = {
     subscription: null!,
     listeners: new Map(),
-    batch: { events: [], overflowed: false, timer: null, firstEventAt: 0 },
+    batch: createDebouncedBatch(),
     rootPath
   }
 
@@ -211,6 +259,9 @@ export async function createLocalWatcher(
           return
         }
 
+        if (root.batch.cancelled) {
+          return
+        }
         queueWatcherEvents(root.batch, events)
         scheduleLocalBatchFlush(root)
       },

src/main/ipc/filesystem-watcher-local-install.ts

@@ -16,6 +16,7 @@ import {
   retainLocalWatcherPhysicalFailure,
   trackDetachedLocalUnsubscribe
 } from './filesystem-watcher-listener-lifecycle'
+import { cancelLocalBatchFlush } from './filesystem-watcher-batch-control'
 
 export async function installLocalWatcher(
   rootKey: string,
@@ -78,9 +79,7 @@ export async function installLocalWatcher(
     Array.from(cancelToken.listeners.entries()).filter(([, listener]) => !listener.isDestroyed())
   )
   if (cancelToken.cancelled || liveListeners.size === 0) {
-    if (root.batch.timer) {
-      clearTimeout(root.batch.timer)
-    }
+    cancelLocalBatchFlush(root)
     void trackDetachedLocalUnsubscribe(rootKey, root)
     return 'cancelled'
   }

src/main/ipc/filesystem-watcher-local-removal.ts

@@ -14,6 +14,7 @@ import {
   trackDetachedLocalUnsubscribe
 } from './filesystem-watcher-listener-lifecycle'
 import { subscribeLocalWatcher } from './filesystem-watcher-local-subscription'
+import { cancelLocalBatchFlush } from './filesystem-watcher-batch-control'
 
 export async function closeLocalWatcherForWorktreePath(
   worktreePath: string,
@@ -100,9 +101,7 @@ export async function closeLocalWatcherForWorktreePath(
   if (!root) {
     return
   }
-  if (root.batch.timer) {
-    clearTimeout(root.batch.timer)
-  }
+  cancelLocalBatchFlush(root)
   watcherLifecycleState.watchedRoots.delete(rootKey)
   // Why: the in-process Parcel fallback has no unsubscribe timeout of its own, so an unbounded await
   // here would hang delete forever and hold the removal gate. The promise stays tracked in

src/main/ipc/filesystem-watcher-local-subscription.ts

@@ -14,6 +14,7 @@ import {
   takeLocalCapacityRetryListeners,
   trackDetachedLocalUnsubscribe
 } from './filesystem-watcher-listener-lifecycle'
+import { cancelLocalBatchFlush } from './filesystem-watcher-batch-control'
 import { scheduleLocalCapacityRetry } from './filesystem-watcher-local-capacity'
 import { installLocalWatcher } from './filesystem-watcher-local-install'
 
@@ -199,7 +200,6 @@ export function unsubscribeLocalWatcher(worktreePath: string, senderId: number):
     if (root.batch.timer) {
       clearTimeout(root.batch.timer)
     }
-
     // Why: duplicate unwatch calls for a root would leak overwritten grace timers; keep just one.
     if (watcherLifecycleState.pendingTeardowns.has(rootKey)) {
       return
@@ -213,6 +213,7 @@ export function unsubscribeLocalWatcher(worktreePath: string, senderId: number):
         return
       }
       void trackDetachedLocalUnsubscribe(rootKey, currentRoot)
+      cancelLocalBatchFlush(currentRoot)
       watcherLifecycleState.watchedRoots.delete(rootKey)
     }, WATCHER_TEARDOWN_GRACE_MS)
 

src/main/ipc/filesystem-watcher-shutdown.ts

@@ -1,6 +1,7 @@
 import { disposeWatcherProcess } from './parcel-watcher-process'
 import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state'
 import { trackDetachedLocalUnsubscribe } from './filesystem-watcher-listener-lifecycle'
+import { cancelLocalBatchFlush } from './filesystem-watcher-batch-control'
 
 /** Tear down all watchers on app shutdown. */
 export async function closeAllWatchers(): Promise<void> {
@@ -57,9 +58,7 @@ export async function closeAllWatchers(): Promise<void> {
   }
 
   for (const [rootKey, root] of watcherLifecycleState.watchedRoots) {
-    if (root.batch.timer) {
-      clearTimeout(root.batch.timer)
-    }
+    cancelLocalBatchFlush(root)
     await trackDetachedLocalUnsubscribe(rootKey, root).catch(() => undefined)
   }
   watcherLifecycleState.watchedRoots.clear()

src/main/ipc/filesystem-watcher-wsl.ts

@@ -13,18 +13,12 @@ import { queueWatcherEvents } from './filesystem-watcher-event-batch'
 import { parseWslUncPath } from '../../shared/wsl-paths'
 import { createWslWatcherProcessExit, createWslWatcherStartup } from './wsl-watcher-process-exit'
 import { reserveWatcherChild, WatcherChildCapacityError } from './parcel-watcher-child-registry'
+import { createDebouncedBatch, type DebouncedBatch } from './filesystem-watcher-batch-control'
 
 export type WatcherSubscription = {
   unsubscribe(): Promise<void>
 }
 
-type DebouncedBatch = {
-  events: WatcherEvent[]
-  overflowed: boolean
-  timer: ReturnType<typeof setTimeout> | null
-  firstEventAt: number
-}
-
 export type WatchedRoot = {
   subscription: WatcherSubscription
   listeners: Map<number, WebContents>
@@ -171,7 +165,7 @@ export async function createWslWatcher(
   const root: WatchedRoot = {
     subscription: null!,
     listeners: new Map(),
-    batch: { events: [], overflowed: false, timer: null, firstEventAt: 0 },
+    batch: createDebouncedBatch(),
     rootPath: worktreePath
   }
 
@@ -199,6 +193,9 @@ export async function createWslWatcher(
   }
 
   function ingestFrame(frame: string): void {
+    if (root.batch.cancelled) {
+      return
+    }
     const nextSnapshot = parseSnapshotFrame(frame, distro)
     if (!prevSnapshot) {
       prevSnapshot = nextSnapshot

src/main/ipc/filesystem-watcher.test.ts

@@ -51,6 +51,7 @@ import {
 import { stat } from 'node:fs/promises'
 import { subscribe as subscribeParcelWatcher } from '@parcel/watcher'
 import { createWslWatcher } from './filesystem-watcher-wsl'
+import { createDebouncedBatch } from './filesystem-watcher-batch-control'
 import {
   MAX_PHYSICAL_WATCHER_CHILDREN,
   reserveWatcherChild,
@@ -135,7 +136,7 @@ describe('registerFilesystemWatcherHandlers', () => {
       return {
         subscription: { unsubscribe: vi.fn(async () => release()) },
         listeners: new Map(),
-        batch: { events: [], overflowed: false, timer: null, firstEventAt: 0 },
+        batch: createDebouncedBatch(),
         rootPath: worktreePath
       }
     })