From 50938b2dbd117e4bab7cedfc058d6bf1571666b7 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Mon, 31 Aug 2026 18:28:26 -0700 Subject: [PATCH] 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. --- .../ipc/filesystem-watcher-batch-control.ts | 37 ++++ .../filesystem-watcher-listener-lifecycle.ts | 5 +- .../filesystem-watcher-local-events.test.ts | 171 ++++++++++++++++++ .../ipc/filesystem-watcher-local-events.ts | 115 ++++++++---- .../ipc/filesystem-watcher-local-install.ts | 5 +- .../ipc/filesystem-watcher-local-removal.ts | 5 +- .../filesystem-watcher-local-subscription.ts | 3 +- src/main/ipc/filesystem-watcher-shutdown.ts | 5 +- src/main/ipc/filesystem-watcher-wsl.ts | 13 +- src/main/ipc/filesystem-watcher.test.ts | 3 +- 10 files changed, 308 insertions(+), 54 deletions(-) create mode 100644 src/main/ipc/filesystem-watcher-batch-control.ts create mode 100644 src/main/ipc/filesystem-watcher-local-events.test.ts diff --git a/src/main/ipc/filesystem-watcher-batch-control.ts b/src/main/ipc/filesystem-watcher-batch-control.ts new file mode 100644 index 00000000000..06a45fe59d7 --- /dev/null +++ b/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 | 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 +} diff --git a/src/main/ipc/filesystem-watcher-listener-lifecycle.ts b/src/main/ipc/filesystem-watcher-listener-lifecycle.ts index ce0cfaa80fc..713305069e3 100644 --- a/src/main/ipc/filesystem-watcher-listener-lifecycle.ts +++ b/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) } diff --git a/src/main/ipc/filesystem-watcher-local-events.test.ts b/src/main/ipc/filesystem-watcher-local-events.test.ts new file mode 100644 index 00000000000..15a6fd0e956 --- /dev/null +++ b/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(): { promise: Promise; resolve: (value: T) => void } { + let resolve!: (value: T) => void + const promise = new Promise((nextResolve) => { + resolve = nextResolve + }) + return { promise, resolve } +} + +async function flushMicrotasks(): Promise { + for (let i = 0; i < 6; i++) { + await Promise.resolve() + } +} + +type Sender = { isDestroyed: () => boolean; send: ReturnType } + +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 } + ]) + }) +}) diff --git a/src/main/ipc/filesystem-watcher-local-events.ts b/src/main/ipc/filesystem-watcher-local-events.ts index 9618c6e74a7..7d5b383176f 100644 --- a/src/main/ipc/filesystem-watcher-local-events.ts +++ b/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 { + 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) }, diff --git a/src/main/ipc/filesystem-watcher-local-install.ts b/src/main/ipc/filesystem-watcher-local-install.ts index 089c553297b..27cfc6dafcf 100644 --- a/src/main/ipc/filesystem-watcher-local-install.ts +++ b/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' } diff --git a/src/main/ipc/filesystem-watcher-local-removal.ts b/src/main/ipc/filesystem-watcher-local-removal.ts index b0b92a9c118..a3d4b6ff558 100644 --- a/src/main/ipc/filesystem-watcher-local-removal.ts +++ b/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 diff --git a/src/main/ipc/filesystem-watcher-local-subscription.ts b/src/main/ipc/filesystem-watcher-local-subscription.ts index 6d27f58f233..933d02bdb37 100644 --- a/src/main/ipc/filesystem-watcher-local-subscription.ts +++ b/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) diff --git a/src/main/ipc/filesystem-watcher-shutdown.ts b/src/main/ipc/filesystem-watcher-shutdown.ts index 96018b57512..9f9cd2de353 100644 --- a/src/main/ipc/filesystem-watcher-shutdown.ts +++ b/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 { @@ -57,9 +58,7 @@ export async function closeAllWatchers(): Promise { } 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() diff --git a/src/main/ipc/filesystem-watcher-wsl.ts b/src/main/ipc/filesystem-watcher-wsl.ts index f2599f41d61..86f10c2ee68 100644 --- a/src/main/ipc/filesystem-watcher-wsl.ts +++ b/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 } -type DebouncedBatch = { - events: WatcherEvent[] - overflowed: boolean - timer: ReturnType | null - firstEventAt: number -} - export type WatchedRoot = { subscription: WatcherSubscription listeners: Map @@ -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 diff --git a/src/main/ipc/filesystem-watcher.test.ts b/src/main/ipc/filesystem-watcher.test.ts index 6fff613e0d9..2d9181d1155 100644 --- a/src/main/ipc/filesystem-watcher.test.ts +++ b/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 } })