diff --git a/src/main/ipc/parcel-watcher-process-entry.test.ts b/src/main/ipc/parcel-watcher-process-entry.test.ts index 6ebdb5385dd..b84187d1594 100644 --- a/src/main/ipc/parcel-watcher-process-entry.test.ts +++ b/src/main/ipc/parcel-watcher-process-entry.test.ts @@ -135,6 +135,82 @@ describe('parcel watcher process canary', () => { ) }) + it.each(['error', 'delayed-close', 'throw'] as const)( + 'advances the real child lifecycle queue after shallow %s teardown', + async (schedule) => { + detectShallowWatchDeliveryMock.mockResolvedValue(true) + const handles: (EventEmitter & { close: () => void })[] = [] + watchMock.mockImplementation(() => { + const watcher = new EventEmitter() as EventEmitter & { + close: () => void + } + watcher.close = () => { + watcher.emit('close') + } + handles.push(watcher) + return watcher + }) + const sendMock = vi.fn() + process.send = sendMock + await import('./parcel-watcher-process-entry') + await vi.advanceTimersByTimeAsync(0) + for (const id of [1, 2]) { + process.emit('message', { + op: 'subscribe', + id, + dir: `/folder-${id}`, + opts: { mode: 'shallow', include: ['HEAD'] } + }) + await vi.advanceTimersByTimeAsync(0) + } + handles[0].close = () => { + if (schedule === 'throw') { + throw new Error('handle still active') + } + } + if (schedule === 'error') { + handles[0].emit('error', new Error('terminal')) + } + process.emit('message', { op: 'unsubscribe', id: 1 }) + process.emit('message', { + op: 'subscribe', + id: 3, + dir: '/folder-3', + opts: { mode: 'shallow', include: ['HEAD'] } + }) + await vi.advanceTimersByTimeAsync(0) + if (schedule === 'delayed-close') { + expect(sendMock).not.toHaveBeenCalledWith({ + op: 'subscribe-started', + id: 3 + }) + handles[0].emit('close') + await vi.advanceTimersByTimeAsync(0) + } + expect(sendMock).toHaveBeenCalledWith({ op: 'subscribed', id: 3 }) + if (schedule === 'throw') { + expect(sendMock).toHaveBeenCalledWith({ + op: 'unsubscribe-failed', + id: 1, + message: 'handle still active' + }) + expect(sendMock).not.toHaveBeenCalledWith({ + op: 'unsubscribed', + id: 1 + }) + handles[0].emit('close') + } else { + expect(sendMock).toHaveBeenCalledWith({ op: 'unsubscribed', id: 1 }) + } + for (const id of [1, 2, 3]) { + process.emit('message', { op: 'unsubscribe', id }) + } + await vi.advanceTimersByTimeAsync(0) + expect(sendMock).toHaveBeenCalledWith({ op: 'unsubscribed', id: 2 }) + expect(sendMock).toHaveBeenCalledWith({ op: 'unsubscribed', id: 3 }) + } + ) + it('does not restart while a native subscription is still crawling', async () => { subscribeMock .mockResolvedValueOnce({ unsubscribe: vi.fn() }) diff --git a/src/main/ipc/parcel-watcher-shallow-subscription.ts b/src/main/ipc/parcel-watcher-shallow-subscription.ts index 13319281535..c9473d4fd92 100644 --- a/src/main/ipc/parcel-watcher-shallow-subscription.ts +++ b/src/main/ipc/parcel-watcher-shallow-subscription.ts @@ -1,7 +1,8 @@ -import { statSync, watch, type FSWatcher } from 'node:fs' +import { statSync, watch } from 'node:fs' import { stat } from 'node:fs/promises' import { join } from 'node:path' import type { Event as ParcelWatcherEvent } from '@parcel/watcher' +import { createShallowWatcherBinding, type ShallowWatcherBinding } from './shallow-watcher-binding' export type ShallowWatcherSubscription = { unsubscribe: () => Promise @@ -14,17 +15,6 @@ export type ShallowWatcherSubscription = { // bounded cadence. Two stats per interval is the whole steady-state cost. const REBIND_CHECK_INTERVAL_MS = 30_000 -function closeFileSystemWatcher(watcher: FSWatcher): Promise { - const { promise, resolve } = Promise.withResolvers() - watcher.once('close', resolve) - try { - watcher.close() - } catch { - resolve() - } - return promise -} - export function startShallowWatcher( rootPath: string, relativePaths: readonly string[], @@ -44,8 +34,9 @@ export function startShallowWatcher( pathsByDirectory.set(parent, fileNames) } - const watchers = new Map() - const boundIdentities = new Map() + const watchers = new Map() + const ownedBindings = new Set() + let unsubscribePromise: Promise | undefined let disposed = false let reportedError = false @@ -73,13 +64,15 @@ export function startShallowWatcher( return } watchers.delete(parent) - boundIdentities.delete(parent) - void closeFileSystemWatcher(existing) + void existing.close().catch(reportError) } const directoryPath = join(rootPath, parent) + // Read before watch so a replacement in the gap is detected by the next sweep. + const identity = directoryIdentitySync(parent) try { + let binding: ShallowWatcherBinding const watcher = watch(directoryPath, { persistent: false }, (eventType, fileName) => { - if (disposed) { + if (disposed || watchers.get(parent) !== binding) { return } const name = fileName?.toString() @@ -100,8 +93,19 @@ export function startShallowWatcher( emitUpdates(parent, [name]) } }) - watcher.on('error', reportError) - watchers.set(parent, watcher) + binding = createShallowWatcherBinding( + watcher, + identity, + () => { + ownedBindings.delete(binding) + if (watchers.get(parent) === binding) { + watchers.delete(parent) + } + }, + reportError + ) + ownedBindings.add(binding) + watchers.set(parent, binding) } catch (error) { // Nested metadata directories may not exist until Git creates them. if (parent === '') { @@ -129,18 +133,18 @@ export function startShallowWatcher( } const refreshBinding = async (parent: string, fileNames: Set): Promise => { + const existing = watchers.get(parent) const identity = await directoryIdentity(parent) - if (disposed || identity === null) { + if (disposed || identity === null || watchers.get(parent) !== existing) { return } - const bound = boundIdentities.get(parent) + const bound = existing?.identity if (bound === identity) { return } // Either the directory appeared after we started, or it was replaced while // watched. Both leave the old binding deaf, so rebind and resync. watchDirectory(parent, fileNames, true) - boundIdentities.set(parent, identity) if (bound !== undefined) { emitUpdates(parent, fileNames) } @@ -157,22 +161,19 @@ export function startShallowWatcher( rebindTimer.unref?.() for (const [parent, fileNames] of pathsByDirectory) { - // Why: read identity BEFORE binding. If the directory is replaced in the gap, - // the recorded identity is stale and the first sweep rebinds — the harmless - // direction. Reading after would pin the dead inode's watcher to the new - // identity, and the sweep would then never rebind it. - const identityBeforeBind = directoryIdentitySync(parent) watchDirectory(parent, fileNames) - if (watchers.has(parent) && identityBeforeBind !== null) { - boundIdentities.set(parent, identityBeforeBind) - } } return { - unsubscribe: async () => { - disposed = true - clearInterval(rebindTimer) - await Promise.all([...watchers.values()].map(closeFileSystemWatcher)) + unsubscribe: () => { + if (!unsubscribePromise) { + disposed = true + clearInterval(rebindTimer) + unsubscribePromise = Promise.all([...ownedBindings].map((binding) => binding.close())).then( + () => undefined + ) + } + return unsubscribePromise } } } diff --git a/src/main/ipc/shallow-watcher-binding-lifecycle.test.ts b/src/main/ipc/shallow-watcher-binding-lifecycle.test.ts new file mode 100644 index 00000000000..d5c1b241194 --- /dev/null +++ b/src/main/ipc/shallow-watcher-binding-lifecycle.test.ts @@ -0,0 +1,202 @@ +import { EventEmitter } from 'node:events' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { watchMock, statMock, statSyncMock } = vi.hoisted(() => ({ + watchMock: vi.fn(), + statMock: vi.fn(), + statSyncMock: vi.fn() +})) +vi.mock('node:fs', () => ({ watch: watchMock, statSync: statSyncMock })) +vi.mock('node:fs/promises', () => ({ stat: statMock })) +import { startShallowWatcher } from './parcel-watcher-shallow-subscription' + +class Watcher extends EventEmitter { + close = vi.fn(() => { + this.emit('close') + }) + constructor(readonly change: (event: string, name: string) => void) { + super() + } +} +const root = join('folder', 'metadata') +const identity = (ino = 1) => ({ dev: 1, ino, isDirectory: () => true }) +let created: Watcher[] +const subscribe = (onError = vi.fn(), onEvents = vi.fn()) => + startShallowWatcher(root, ['HEAD', 'logs/HEAD'], onEvents, onError) + +beforeEach(() => { + vi.useFakeTimers() + created = [] + statMock.mockReset().mockResolvedValue(identity()) + statSyncMock.mockReset().mockReturnValue(identity()) + watchMock.mockReset().mockImplementation((_path, _options, change) => { + const watcher = new Watcher(change) + created.push(watcher) + return watcher + }) +}) +afterEach(() => { + vi.clearAllTimers() + vi.useRealTimers() +}) + +describe('shallow watcher binding ownership', () => { + it('reuses unsubscribe completion during and after delayed normal close', async () => { + const subscription = subscribe() + created[0].close.mockImplementation(() => {}) + const first = subscription.unsubscribe() + const settled = vi.fn() + void first.then(settled) + expect(subscription.unsubscribe()).toBe(first) + await vi.advanceTimersByTimeAsync(0) + expect(settled).not.toHaveBeenCalled() + created[0].emit('close') + await first + expect(subscription.unsubscribe()).toBe(first) + await subscription.unsubscribe() + expect(created[0].close).toHaveBeenCalledTimes(1) + expect(settled).toHaveBeenCalledTimes(1) + }) + + it.each(['close', 'error'] as const)( + 'remembers prior terminal %s without awaiting another close', + async (event) => { + const onError = vi.fn() + const subscription = subscribe(onError) + created[0].close.mockImplementation(() => {}) + created[0].emit(event, new Error('native terminal')) + await subscription.unsubscribe() + await subscription.unsubscribe() + expect(created[0].close).not.toHaveBeenCalled() + expect(onError).toHaveBeenCalledTimes(event === 'error' ? 1 : 0) + } + ) + + it('settles an in-flight close on terminal error without a close event', async () => { + const subscription = subscribe() + created[0].close.mockImplementation(() => {}) + const completion = subscription.unsubscribe() + created[0].emit('error', new Error('native terminal')) + await completion + created[0].emit('close') + expect(subscription.unsubscribe()).toBe(completion) + }) + + it('rejects throwing close without terminal evidence and reuses that failure', async () => { + const subscription = subscribe() + const failure = new Error('still live') + created[0].close.mockImplementation(() => { + throw failure + }) + const completion = subscription.unsubscribe() + await expect(completion).rejects.toBe(failure) + expect(subscription.unsubscribe()).toBe(completion) + await expect(subscription.unsubscribe()).rejects.toBe(failure) + expect(created[0].close).toHaveBeenCalledTimes(1) + expect(created[1].close).toHaveBeenCalledTimes(1) + created[0].emit('close') + }) + + it.each(['close', 'error'] as const)( + 'accepts terminal %s established before close throws', + async (event) => { + const subscription = subscribe() + created[0].close.mockImplementation(() => { + created[0].emit(event, new Error('terminal')) + throw new Error('after terminal') + }) + await subscription.unsubscribe() + } + ) + + it.each(['close', 'error'] as const)( + 'isolates late old %s and updates from replacement identity', + async (event) => { + const onEvents = vi.fn() + const subscription = subscribe(vi.fn(), onEvents) + const old = created[1] + old.close.mockImplementation(() => {}) + created[0].change('rename', 'logs') + const replacement = created[2] + old.emit(event, new Error('old terminal')) + onEvents.mockClear() + old.change('change', 'HEAD') + expect(onEvents).not.toHaveBeenCalled() + replacement.change('change', 'HEAD') + expect(onEvents).toHaveBeenCalledWith([{ type: 'update', path: join(root, 'logs', 'HEAD') }]) + await vi.advanceTimersByTimeAsync(30_000) + expect(created).toHaveLength(3) + await subscription.unsubscribe() + expect(replacement.close).toHaveBeenCalledTimes(1) + } + ) + + it('awaits the retired generation as well as its replacement', async () => { + const subscription = subscribe() + const old = created[1] + old.close.mockImplementation(() => {}) + created[0].change('rename', 'logs') + const settled = vi.fn() + const completion = subscription.unsubscribe() + void completion.then(settled) + await vi.advanceTimersByTimeAsync(0) + expect(settled).not.toHaveBeenCalled() + expect(created[2].close).toHaveBeenCalledTimes(1) + old.emit('close') + await completion + expect(old.close).toHaveBeenCalledTimes(1) + }) + + it('preserves failed retired teardown for unsubscribe and host recovery', async () => { + const onError = vi.fn() + const subscription = subscribe(onError) + const failure = new Error('retired handle still live') + created[1].close.mockImplementation(() => { + throw failure + }) + created[0].change('rename', 'logs') + await vi.advanceTimersByTimeAsync(0) + expect(onError).toHaveBeenCalledWith(failure) + await expect(subscription.unsubscribe()).rejects.toBe(failure) + expect(created[2].close).toHaveBeenCalledTimes(1) + created[1].emit('close') + }) + + it('rebinds after terminal error even when the directory inode is unchanged', async () => { + const subscription = subscribe() + created[1].emit('error', new Error('terminal')) + await vi.advanceTimersByTimeAsync(30_000) + expect(created).toHaveLength(3) + await subscription.unsubscribe() + expect(created[2].close).toHaveBeenCalledTimes(1) + }) + + it('retries failed binding creation instead of recording an unwatched identity', async () => { + const subscription = subscribe() + statMock.mockResolvedValue(identity(2)) + statSyncMock.mockReturnValue(identity(2)) + watchMock.mockImplementationOnce(() => { + throw new Error('missing') + }) + await vi.advanceTimersByTimeAsync(30_000) + const attempts = watchMock.mock.calls.length + await vi.advanceTimersByTimeAsync(30_000) + expect(watchMock).toHaveBeenCalledTimes(attempts + 1) + await subscription.unsubscribe() + }) + + it('does not let an old async identity probe rebind a replacement', async () => { + const subscription = subscribe() + const pending = Promise.withResolvers>() + statMock.mockImplementation((path) => + path === join(root, 'logs') ? pending.promise : Promise.resolve(identity()) + ) + await vi.advanceTimersByTimeAsync(30_000) + created[0].change('rename', 'logs') + pending.resolve(identity(2)) + await vi.advanceTimersByTimeAsync(0) + expect(created).toHaveLength(3) + await subscription.unsubscribe() + }) +}) diff --git a/src/main/ipc/shallow-watcher-binding.ts b/src/main/ipc/shallow-watcher-binding.ts new file mode 100644 index 00000000000..907339c6163 --- /dev/null +++ b/src/main/ipc/shallow-watcher-binding.ts @@ -0,0 +1,48 @@ +import type { FSWatcher } from 'node:fs' + +export type ShallowWatcherBinding = { + identity: string | null + close: () => Promise +} + +export function createShallowWatcherBinding( + watcher: FSWatcher, + identity: string | null, + onTerminal: () => void, + onError: (error: unknown) => void +): ShallowWatcherBinding { + const { promise, resolve, reject } = Promise.withResolvers() + let terminal = false + let closing = false + const finish = (): void => { + if (terminal) { + return + } + terminal = true + resolve() + onTerminal() + } + watcher.once('close', finish) + watcher.on('error', (error) => { + // Node closes its native handle before emitting error, without emitting close. + finish() + onError(error) + }) + return { + identity, + close: () => { + if (!terminal && !closing) { + closing = true + try { + watcher.close() + } catch (error) { + // A throw alone does not prove the native handle was released. + if (!terminal) { + reject(error) + } + } + } + return promise + } + } +}