From f93e92646cb3c3791e8aec584974f6a41da8a246 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Mon, 13 Jul 2026 19:12:30 -0700 Subject: [PATCH] P2 watcher lifecycle bounds (#8640) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix watcher lifecycle cancellation bounds * fix(review): guard remote watcher installs against post-shutdown resurrection closeAllWatchers aborted the in-flight install *tokens* it could see, but a same-key joiner awaiting a 'cancelled' resolution (and a fired retry tick) calls installRemoteWatcher directly and, on the fresh-generation recursion, builds a brand-new non-aborted AbortController and calls provider.watch() after teardown — leaking an SSH watcher into the just-cleared remoteWatchers map. Latch the subsystem shut in closeAllWatchers and refuse installs while latched; a genuine new fs:watchWorktree clears it. Adds a regression test (fails without the latch) plus a test for the same-tick handoff-revival guard that had no coverage. Also extract the duplicated isolated-quarantine-vs-fuse branch shared by retireSlot and releaseFailedRoot into quarantineOrFuse (behavior-preserving). * Add lifecycle generation guard to refuse stale remote-watcher joiners - A boolean latch alone can't distinguish a pre-shutdown joiner from a fresh call once a genuine new watch reopens the subsystem, letting a stale joiner recurse and register a post-shutdown provider.watch() - Each installRemoteWatcher call now captures a generation counter that closeAllWatchers bumps, so a waiter that resumes after a later shutdown+reopen is refused instead of resurrecting - Adds a regression test covering the shutdown-then-reopen race --- ...system-watcher-remote-cancellation.test.ts | 299 ++++++++++++++++++ src/main/ipc/filesystem-watcher.test.ts | 8 +- src/main/ipc/filesystem-watcher.ts | 121 +++++-- .../ipc/runtime-watcher-process-pool.test.ts | 55 ++++ src/main/ipc/runtime-watcher-process-pool.ts | 44 ++- 5 files changed, 483 insertions(+), 44 deletions(-) create mode 100644 src/main/ipc/filesystem-watcher-remote-cancellation.test.ts diff --git a/src/main/ipc/filesystem-watcher-remote-cancellation.test.ts b/src/main/ipc/filesystem-watcher-remote-cancellation.test.ts new file mode 100644 index 000000000..3e22cb6b8 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-cancellation.test.ts @@ -0,0 +1,299 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { handleMock, getSshFilesystemProviderMock } = vi.hoisted(() => ({ + handleMock: vi.fn(), + getSshFilesystemProviderMock: vi.fn() +})) + +vi.mock('electron', () => ({ + ipcMain: { handle: handleMock } +})) + +vi.mock('fs/promises', () => ({ stat: vi.fn() })) +vi.mock('@parcel/watcher', () => ({ subscribe: vi.fn() })) +vi.mock('./filesystem-watcher-wsl', () => ({ createWslWatcher: vi.fn() })) +vi.mock('../providers/ssh-filesystem-dispatch', () => ({ + getSshFilesystemProvider: getSshFilesystemProviderMock +})) + +import { closeAllWatchers, registerFilesystemWatcherHandlers } from './filesystem-watcher' + +type HandlerMap = Record Promise | unknown> + +describe('remote filesystem watcher cancellation', () => { + const handlers: HandlerMap = {} + + beforeEach(async () => { + handleMock.mockReset() + getSshFilesystemProviderMock.mockReset() + for (const key of Object.keys(handlers)) { + delete handlers[key] + } + handleMock.mockImplementation((channel, handler) => { + handlers[channel] = handler + }) + registerFilesystemWatcherHandlers() + await closeAllWatchers() + }) + + it('aborts pending SSH setup after the last same-root listener leaves and cleans late success', async () => { + let installSignal: AbortSignal | undefined + let resolveInstall: ((unwatch: () => void) => void) | undefined + const lateUnwatch = vi.fn() + const watchMock = vi.fn( + (_rootPath, _callback, options?: { signal?: AbortSignal }) => + new Promise<() => void>((resolve) => { + installSignal = options?.signal + resolveInstall = resolve + }) + ) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const args = { worktreePath: '/home/me/repo', connectionId: 'conn-1' } + const senderOne = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 } + const senderTwo = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 } + const first = handlers['fs:watchWorktree']({ sender: senderOne }, args) as Promise + const second = handlers['fs:watchWorktree']({ sender: senderTwo }, args) as Promise + + await Promise.resolve() + try { + expect(watchMock).toHaveBeenCalledTimes(1) + expect(installSignal?.aborted).toBe(false) + + handlers['fs:unwatchWorktree']({ sender: { id: 1 } }, args) + await Promise.resolve() + expect(installSignal?.aborted).toBe(false) + + handlers['fs:unwatchWorktree']({ sender: { id: 2 } }, args) + await Promise.resolve() + expect(installSignal?.aborted).toBe(true) + } finally { + resolveInstall?.(lateUnwatch) + await Promise.all([first, second]) + } + expect(lateUnwatch).toHaveBeenCalledTimes(1) + }) + + it('starts a fresh same-root generation when a listener arrives after physical abort', async () => { + let firstSignal: AbortSignal | undefined + let secondCallback: ((events: unknown[]) => void) | undefined + const secondUnwatch = vi.fn() + const watchMock = vi + .fn() + .mockImplementationOnce( + (_rootPath, _callback, options?: { signal?: AbortSignal }) => + new Promise<() => void>((_resolve, reject) => { + firstSignal = options?.signal + options?.signal?.addEventListener( + 'abort', + () => { + const error = new Error('cancelled') + error.name = 'AbortError' + reject(error) + }, + { once: true } + ) + }) + ) + .mockImplementationOnce((_rootPath, callback) => { + secondCallback = callback + return Promise.resolve(secondUnwatch) + }) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const args = { worktreePath: '/home/me/repo', connectionId: 'conn-1' } + const firstSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 } + const secondSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 } + const first = handlers['fs:watchWorktree']({ sender: firstSender }, args) as Promise + + await Promise.resolve() + handlers['fs:unwatchWorktree']({ sender: { id: 1 } }, args) + await vi.waitFor(() => expect(firstSignal?.aborted).toBe(true)) + const second = handlers['fs:watchWorktree']({ sender: secondSender }, args) as Promise + + await Promise.all([first, second]) + expect(watchMock).toHaveBeenCalledTimes(2) + secondCallback?.([{ kind: 'update', absolutePath: '/home/me/repo/file.ts' }]) + expect(secondSender.send).toHaveBeenCalledTimes(1) + + handlers['fs:unwatchWorktree']({ sender: { id: 2 } }, args) + expect(secondUnwatch).toHaveBeenCalledTimes(1) + }) + + it('aborts pending SSH setup on sender destruction and watcher shutdown', async () => { + const installs = new Map< + string, + { signal: AbortSignal | undefined; resolve: (unwatch: () => void) => void } + >() + const watchMock = vi.fn( + (rootPath: string, _callback, options?: { signal?: AbortSignal }) => + new Promise<() => void>((resolve) => { + installs.set(rootPath, { signal: options?.signal, resolve }) + }) + ) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const destroyedCallbacks: (() => void)[] = [] + const destroyedSender = { + isDestroyed: () => false, + send: vi.fn(), + once: vi.fn((event: string, callback: () => void) => { + if (event === 'destroyed') { + destroyedCallbacks.push(callback) + } + }), + id: 1 + } + const destroyedArgs = { worktreePath: '/destroyed', connectionId: 'conn-1' } + const destroyedWatch = handlers['fs:watchWorktree']( + { sender: destroyedSender }, + destroyedArgs + ) as Promise + + await Promise.resolve() + destroyedCallbacks[0]() + await Promise.resolve() + expect(installs.get('/destroyed')?.signal?.aborted).toBe(true) + installs.get('/destroyed')?.resolve(vi.fn()) + await destroyedWatch + + const shutdownSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 } + const shutdownArgs = { worktreePath: '/shutdown', connectionId: 'conn-1' } + const shutdownWatch = handlers['fs:watchWorktree']( + { sender: shutdownSender }, + shutdownArgs + ) as Promise + + await Promise.resolve() + await closeAllWatchers() + expect(installs.get('/shutdown')?.signal?.aborted).toBe(true) + installs.get('/shutdown')?.resolve(vi.fn()) + await shutdownWatch + }) + + it('keeps the shared install alive when a replacement sender joins before the deferred abort fires', async () => { + let installSignal: AbortSignal | undefined + let resolveInstall: ((unwatch: () => void) => void) | undefined + const watchMock = vi.fn( + (_rootPath, _callback, options?: { signal?: AbortSignal }) => + new Promise<() => void>((resolve) => { + installSignal = options?.signal + resolveInstall = resolve + }) + ) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const args = { worktreePath: '/home/me/repo', connectionId: 'conn-1' } + const senderOne = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 } + const senderTwo = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 } + const first = handlers['fs:watchWorktree']({ sender: senderOne }, args) as Promise + + await Promise.resolve() + expect(watchMock).toHaveBeenCalledTimes(1) + + // The last listener leaves and a replacement joins in the SAME tick — before + // the queued abort microtask runs. The shared install must survive. + handlers['fs:unwatchWorktree']({ sender: { id: 1 } }, args) + const second = handlers['fs:watchWorktree']({ sender: senderTwo }, args) as Promise + + await Promise.resolve() + await Promise.resolve() + expect(installSignal?.aborted).toBe(false) + expect(watchMock).toHaveBeenCalledTimes(1) + + resolveInstall?.(vi.fn()) + await Promise.all([first, second]) + }) + + it('refuses a post-shutdown joiner recursion instead of resurrecting the install', async () => { + let firstSignal: AbortSignal | undefined + let resolveFirst: ((unwatch: () => void) => void) | undefined + const lateUnwatch = vi.fn() + const watchMock = vi.fn( + (_rootPath, _callback, options?: { signal?: AbortSignal }) => + new Promise<() => void>((resolve) => { + firstSignal = options?.signal + resolveFirst = resolve + }) + ) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const args = { worktreePath: '/home/me/repo', connectionId: 'conn-1' } + const firstSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 } + const secondSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 } + const first = handlers['fs:watchWorktree']({ sender: firstSender }, args) as Promise + + await Promise.resolve() + expect(watchMock).toHaveBeenCalledTimes(1) + + // Last listener leaves -> deferred abort fires while the install is still pending. + handlers['fs:unwatchWorktree']({ sender: { id: 1 } }, args) + await vi.waitFor(() => expect(firstSignal?.aborted).toBe(true)) + + // Joiner arrives after physical abort (canJoinInstall === false); it awaits the + // 'cancelled' resolution and would recurse into a fresh install. + const second = handlers['fs:watchWorktree']({ sender: secondSender }, args) as Promise + await Promise.resolve() + + // Shutdown latches the subsystem before the joiner's recursion runs. + await closeAllWatchers() + + // Late success of the aborted generation must be unwatched, not registered. + resolveFirst?.(lateUnwatch) + await Promise.all([first, second]) + + // The recursion is refused post-shutdown: provider.watch() is never called again. + expect(watchMock).toHaveBeenCalledTimes(1) + expect(lateUnwatch).toHaveBeenCalledTimes(1) + }) + + it('refuses a pre-shutdown joiner recursion even after a new watch reopens the subsystem', async () => { + const installs = new Map< + string, + { signal: AbortSignal | undefined; resolve: (unwatch: () => void) => void } + >() + const watchMock = vi.fn( + (rootPath: string, _callback, options?: { signal?: AbortSignal }) => + new Promise<() => void>((resolve) => { + installs.set(rootPath, { signal: options?.signal, resolve }) + }) + ) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const args = { worktreePath: '/home/me/repo', connectionId: 'conn-1' } + const reopenArgs = { worktreePath: '/home/me/other', connectionId: 'conn-1' } + const lateUnwatch = vi.fn() + const firstSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 } + const joinerSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 } + const reopenSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 3 } + const first = handlers['fs:watchWorktree']({ sender: firstSender }, args) as Promise + + await Promise.resolve() + expect(watchMock).toHaveBeenCalledTimes(1) + + // Last listener leaves -> deferred abort fires while the install is still pending. + handlers['fs:unwatchWorktree']({ sender: { id: 1 } }, args) + await vi.waitFor(() => expect(installs.get('/home/me/repo')?.signal?.aborted).toBe(true)) + + // Joiner arrives after physical abort (canJoinInstall === false); it captures the + // current lifecycle generation and awaits the 'cancelled' resolution. + const joiner = handlers['fs:watchWorktree']({ sender: joinerSender }, args) as Promise + await Promise.resolve() + + // Shutdown bumps the generation, then a genuine new watch reopens the subsystem + // (clearing the boolean latch) before the joiner resumes. + await closeAllWatchers() + const reopen = handlers['fs:watchWorktree']( + { sender: reopenSender }, + reopenArgs + ) as Promise + await Promise.resolve() + expect(watchMock).toHaveBeenCalledTimes(2) + + // Now let the aborted install resolve; the joiner recurses on the stale generation. + installs.get('/home/me/repo')?.resolve(lateUnwatch) + installs.get('/home/me/other')?.resolve(vi.fn()) + await Promise.all([first, joiner, reopen]) + + // The joiner's recursion is refused despite the reopen: no third provider.watch(). + expect(watchMock).toHaveBeenCalledTimes(2) + expect(watchMock.mock.calls.filter(([rootPath]) => rootPath === '/home/me/repo')).toHaveLength( + 1 + ) + expect(lateUnwatch).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/main/ipc/filesystem-watcher.test.ts b/src/main/ipc/filesystem-watcher.test.ts index 0b3b07ca0..f776e0d49 100644 --- a/src/main/ipc/filesystem-watcher.test.ts +++ b/src/main/ipc/filesystem-watcher.test.ts @@ -124,7 +124,9 @@ describe('registerFilesystemWatcherHandlers', () => { getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) await vi.advanceTimersByTimeAsync(1_000) - expect(watchMock).toHaveBeenCalledWith('/home/me/repo', expect.any(Function)) + expect(watchMock).toHaveBeenCalledWith('/home/me/repo', expect.any(Function), { + signal: expect.any(AbortSignal) + }) const onEvents = watchMock.mock.calls[0][1] onEvents([{ path: '/home/me/repo/file.txt', type: 'update' }]) expect(sendMock).toHaveBeenCalledWith('fs:changed', { @@ -161,7 +163,9 @@ describe('registerFilesystemWatcherHandlers', () => { await vi.advanceTimersByTimeAsync(1_000) - expect(retryWatchMock).toHaveBeenCalledWith('/home/me/repo', expect.any(Function)) + expect(retryWatchMock).toHaveBeenCalledWith('/home/me/repo', expect.any(Function), { + signal: expect.any(AbortSignal) + }) handlers['fs:unwatchWorktree']( { sender: { id: 1 } }, { worktreePath: '/home/me/repo', connectionId: 'conn-1' } diff --git a/src/main/ipc/filesystem-watcher.ts b/src/main/ipc/filesystem-watcher.ts index 983fc9ce9..665b54377 100644 --- a/src/main/ipc/filesystem-watcher.ts +++ b/src/main/ipc/filesystem-watcher.ts @@ -652,22 +652,31 @@ type RemoteWatcherState = { type RemoteWatcherInstallToken = { cancelled: boolean listeners: Map + abortController: AbortController + abortScheduled: boolean } // Key: `${connectionId}:${worktreePath}`, Value: shared remote watch state. const remoteWatchers = new Map() const loggedUnavailableRemoteWatchers = new Set() const pendingRemoteWatcherRetries = new Map>() -// Why: track in-flight `provider.watch()` calls so an unwatch/shutdown that -// arrives while a watch is still resolving can mark the install cancelled. -// Without this, the awaited unwatch handle would be installed after the -// renderer thinks the watch is gone, leaking a native watcher. +// Why: track in-flight `provider.watch()` calls so last-listener cleanup can +// abort relay setup, while late success is still unwatched instead of leaked. const inFlightRemoteInstalls = new Map() // Why: dedupe concurrent installRemoteWatcher calls for the same key so // overlapping fs:watchWorktree IPCs share one native watcher and one listener // map, instead of each call independently invoking provider.watch() and // overwriting the per-key state on resolution. const pendingRemoteInstallPromises = new Map>() +// Why: block installs that begin AFTER closeAllWatchers — an in-flight joiner +// recursion or a fired retry tick calls installRemoteWatcher directly, bypassing +// the token-abort loop. A genuine new fs:watchWorktree clears the latch. +let remoteWatchersClosed = false +// Why: the boolean latch alone can't tell a pre-shutdown waiter apart from a +// fresh call once a genuine new watch reopens the subsystem. Each call captures +// the generation at entry; closeAllWatchers bumps it, so a joiner that awaited +// across a shutdown+reopen recurses on a stale generation and is refused. +let remoteWatcherLifecycleGeneration = 0 const REMOTE_WATCH_RETRY_MS = 1_000 const REMOTE_WATCH_RETRY_TIMEOUT_MS = 60_000 @@ -675,7 +684,7 @@ function addInFlightRemoteInstallListener( token: RemoteWatcherInstallToken, sender: WebContents ): void { - if (sender.isDestroyed()) { + if (sender.isDestroyed() || token.abortController.signal.aborted) { return } token.listeners.set(sender.id, sender) @@ -683,12 +692,26 @@ function addInFlightRemoteInstallListener( registerSenderCleanup(sender) } +function cancelInFlightRemoteInstallIfUnowned(token: RemoteWatcherInstallToken): void { + token.cancelled = token.listeners.size === 0 + if (!token.cancelled || token.abortScheduled || token.abortController.signal.aborted) { + return + } + token.abortScheduled = true + // Why: a replacement sender can synchronously revive the shared install + // during a renderer handoff; otherwise stop the relay crawl next microtask. + queueMicrotask(() => { + token.abortScheduled = false + if (token.cancelled && token.listeners.size === 0) { + token.abortController.abort() + } + }) +} + function cleanupInFlightRemoteInstallsForSender(senderId: number): void { for (const token of inFlightRemoteInstalls.values()) { token.listeners.delete(senderId) - if (token.listeners.size === 0) { - token.cancelled = true - } + cancelInFlightRemoteInstallIfUnowned(token) } } @@ -726,8 +749,16 @@ type RemoteWatcherInstallResult = 'installed' | 'unavailable' | 'cancelled' async function installRemoteWatcher( sender: WebContents, connectionId: string, - worktreePath: string + worktreePath: string, + generation = remoteWatcherLifecycleGeneration ): Promise { + // Why: refuse installs racing in after teardown (joiner recursion, fired retry + // tick) so provider.watch() is never called and registered post-shutdown. The + // generation guard also refuses a waiter that captured an earlier lifecycle, + // even after a new watch reopened the subsystem. + if (remoteWatchersClosed || generation !== remoteWatcherLifecycleGeneration) { + return 'cancelled' + } const provider = getSshFilesystemProvider(connectionId) if (!provider || sender.isDestroyed()) { return 'unavailable' @@ -747,7 +778,8 @@ async function installRemoteWatcher( const pendingInstall = pendingRemoteInstallPromises.get(key) if (pendingInstall) { const inFlight = inFlightRemoteInstalls.get(key) - if (inFlight) { + const canJoinInstall = inFlight && !inFlight.abortController.signal.aborted + if (canJoinInstall) { // Why: a new watcher can join after all previous pending listeners // unwatched but before provider.watch() resolves; revive that install // instead of inheriting the stale cancellation. @@ -762,9 +794,27 @@ async function installRemoteWatcher( ) { addRemoteWatchListener(key, sender) } + if ( + result === 'cancelled' && + !canJoinInstall && + !sender.isDestroyed() && + generation === remoteWatcherLifecycleGeneration + ) { + // Why: AbortSignal cannot be revived. A listener arriving after physical + // cancellation waits out that generation, then owns a fresh install. + if (pendingRemoteInstallPromises.get(key) === pendingInstall) { + pendingRemoteInstallPromises.delete(key) + } + return installRemoteWatcher(sender, connectionId, worktreePath, generation) + } return result } - const cancelToken: RemoteWatcherInstallToken = { cancelled: false, listeners: new Map() } + const cancelToken: RemoteWatcherInstallToken = { + cancelled: false, + listeners: new Map(), + abortController: new AbortController(), + abortScheduled: false + } inFlightRemoteInstalls.set(key, cancelToken) addInFlightRemoteInstallListener(cancelToken, sender) const installPromise = doInstallRemoteWatcher(provider, key, worktreePath, cancelToken) @@ -786,22 +836,29 @@ async function doInstallRemoteWatcher( ): Promise { let unwatch: () => void try { - unwatch = await provider.watch(worktreePath, (events) => { - const state = remoteWatchers.get(key) - if (!state) { - return - } - for (const listener of state.listeners.values()) { - if (listener.isDestroyed()) { - continue + unwatch = await provider.watch( + worktreePath, + (events) => { + const state = remoteWatchers.get(key) + if (!state) { + return } - listener.send('fs:changed', { - worktreePath, - events - } satisfies FsChangedPayload) - } - }) + for (const listener of state.listeners.values()) { + if (listener.isDestroyed()) { + continue + } + listener.send('fs:changed', { + worktreePath, + events + } satisfies FsChangedPayload) + } + }, + { signal: cancelToken.abortController.signal } + ) } catch (err) { + if (cancelToken.cancelled || cancelToken.abortController.signal.aborted) { + return 'cancelled' + } console.warn(`[filesystem-watcher] SSH watcher unavailable for ${key}:`, err) return 'unavailable' } finally { @@ -884,6 +941,9 @@ export function registerFilesystemWatcherHandlers(): void { 'fs:watchWorktree', async (event, args: { worktreePath: string; connectionId?: string }): Promise => { if (args.connectionId) { + // Why: a real new watch reopens the subsystem after closeAllWatchers + // latched it shut (also how tests reset between cases). + remoteWatchersClosed = false const key = `${args.connectionId}:${args.worktreePath}` const result = await installRemoteWatcher( event.sender, @@ -923,7 +983,7 @@ export function registerFilesystemWatcherHandlers(): void { const inFlight = inFlightRemoteInstalls.get(key) if (inFlight) { inFlight.listeners.delete(_event.sender.id) - inFlight.cancelled = inFlight.listeners.size === 0 + cancelInFlightRemoteInstallIfUnowned(inFlight) } loggedUnavailableRemoteWatchers.delete(key) releaseRemoteWatchListener(key, _event?.sender?.id ?? 0) @@ -951,10 +1011,19 @@ export async function closeAllWatchers(): Promise { } pendingRemoteWatcherRetries.clear() loggedUnavailableRemoteWatchers.clear() + // Why: latch the subsystem shut and drop the dedup map so a late install that + // begins after teardown is refused instead of registering post-shutdown. Bump + // the generation so a waiter that resumes after a later reopen still recurses + // on a stale lifecycle and is refused. + remoteWatchersClosed = true + remoteWatcherLifecycleGeneration += 1 + pendingRemoteInstallPromises.clear() // Why: cancel any in-flight provider.watch() calls so their resolved // unwatch handles are discarded instead of being installed after shutdown. for (const token of inFlightRemoteInstalls.values()) { + token.listeners.clear() token.cancelled = true + token.abortController.abort() } for (const token of inFlightLocalInstalls.values()) { token.listeners.clear() diff --git a/src/main/ipc/runtime-watcher-process-pool.test.ts b/src/main/ipc/runtime-watcher-process-pool.test.ts index e43112e4e..5cd624589 100644 --- a/src/main/ipc/runtime-watcher-process-pool.test.ts +++ b/src/main/ipc/runtime-watcher-process-pool.test.ts @@ -162,6 +162,30 @@ describe('RuntimeWatcherProcessPool', () => { expect(supervisors[1].subscriptions.map(({ dir }) => dir)).toEqual(['/slow']) }) + it('allows only one quarantine generation when isolated setup also times out', async () => { + const timeout = new WatcherProcessFailure( + 'file watcher subscription timed out', + 'subscription', + 'subscribe_timeout' + ) + pool = new RuntimeWatcherProcessPool({ + maxSharedSupervisors: 1, + createSupervisor: () => { + const supervisor = new FakeSupervisor() + supervisor.subscribeError = timeout + supervisors.push(supervisor) + return supervisor + } + }) + + await expect(pool.subscribe('/slow', vi.fn(), {}, {})).rejects.toBe(timeout) + await expect(pool.subscribe('/slow', vi.fn(), {}, {})).rejects.toBe(timeout) + await expect(pool.subscribe('/slow', vi.fn(), {}, {})).rejects.toMatchObject({ + code: 'supervisor_crash_fuse' + }) + expect(supervisors).toHaveLength(2) + }) + it('moves a live root into quarantine when crash resubscription times out', async () => { const timeout = new WatcherProcessFailure( 'file watcher resubscription timed out', @@ -179,6 +203,37 @@ describe('RuntimeWatcherProcessPool', () => { expect(supervisors[1].subscriptions.map(({ dir }) => dir)).toEqual(['/slow-recovery']) }) + it('does not replace an isolated live root after its resubscription times out', async () => { + pool = new RuntimeWatcherProcessPool({ + maxSharedSupervisors: 1, + createSupervisor: () => { + const supervisor = new FakeSupervisor() + supervisors.push(supervisor) + return supervisor + } + }) + const fused = new WatcherProcessFailure( + 'file watcher process crashed repeatedly', + 'supervisor', + 'supervisor_crash_fuse' + ) + const timeout = new WatcherProcessFailure( + 'file watcher resubscription timed out', + 'subscription', + 'subscribe_timeout' + ) + await pool.subscribe('/slow-recovery', vi.fn(), {}, {}) + supervisors[0].subscriptions[0].hooks.onTerminalError?.(fused) + await pool.subscribe('/slow-recovery', vi.fn(), {}, {}) + + supervisors[1].subscriptions[0].hooks.onTerminalError?.(timeout) + + await expect(pool.subscribe('/slow-recovery', vi.fn(), {}, {})).rejects.toMatchObject({ + code: 'supervisor_crash_fuse' + }) + expect(supervisors).toHaveLength(2) + }) + it('keeps healthy shard assignments after a root-specific failure', async () => { pool = new RuntimeWatcherProcessPool({ maxSharedSupervisors: 1, diff --git a/src/main/ipc/runtime-watcher-process-pool.ts b/src/main/ipc/runtime-watcher-process-pool.ts index b5a2bf45d..b4e46cb3d 100644 --- a/src/main/ipc/runtime-watcher-process-pool.ts +++ b/src/main/ipc/runtime-watcher-process-pool.ts @@ -77,10 +77,7 @@ export class RuntimeWatcherProcessPool { if (isWatcherProcessFailure(error) && error.scope === 'supervisor') { this.retireSlot(slot) } else { - releaseAssignment() - if (isWatcherProcessFailure(error) && error.code === 'subscribe_timeout') { - this.isolatedRoots.add(dir) - } + this.releaseFailedRoot(assignment, dir, error, releaseAssignment) } hooks.onTerminalError?.(error) } @@ -94,10 +91,7 @@ export class RuntimeWatcherProcessPool { if (isWatcherProcessFailure(error) && error.scope === 'supervisor') { this.retireSlot(slot) } else { - releaseAssignment() - if (isWatcherProcessFailure(error) && error.code === 'subscribe_timeout') { - this.isolatedRoots.add(dir) - } + this.releaseFailedRoot(assignment, dir, error, releaseAssignment) } throw error } @@ -207,14 +201,7 @@ export class RuntimeWatcherProcessPool { if (this.assignments.get(root)?.slot === slot) { this.assignments.delete(root) } - if (slot.isolated) { - // Why: one bounded quarantine attempt is the recovery budget for a - // watch lifetime; repeated fused replacements would recreate churn. - this.isolatedRoots.delete(root) - this.failedQuarantineRoots.add(root) - } else { - this.isolatedRoots.add(root) - } + this.quarantineOrFuse(root, slot.isolated) } slot.roots.clear() // Why: failAllSubscriptions is still iterating callbacks; defer disposal @@ -240,6 +227,31 @@ export class RuntimeWatcherProcessPool { } } + private releaseFailedRoot( + assignment: RuntimeWatcherPoolAssignment, + dir: string, + error: unknown, + releaseAssignment: () => void + ): void { + releaseAssignment() + if (!isWatcherProcessFailure(error) || error.code !== 'subscribe_timeout') { + return + } + this.quarantineOrFuse(dir, assignment.slot.isolated) + } + + // Why: one bounded quarantine attempt is the recovery budget per watch + // lifetime; an already-isolated root that fails again is fused, not re-isolated, + // so it cannot spawn another child generation. + private quarantineOrFuse(dir: string, isolated: boolean): void { + if (isolated) { + this.isolatedRoots.delete(dir) + this.failedQuarantineRoots.add(dir) + return + } + this.isolatedRoots.add(dir) + } + private disposeSlot(slot: RuntimeWatcherPoolSlot): void { if (slot.disposed) { return