diff --git a/src/main/ipc/filesystem-watcher-handlers.ts b/src/main/ipc/filesystem-watcher-handlers.ts index 39f9ef0c184..923f7596329 100644 --- a/src/main/ipc/filesystem-watcher-handlers.ts +++ b/src/main/ipc/filesystem-watcher-handlers.ts @@ -5,6 +5,7 @@ import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' import { getRemoteWatcherKey } from './filesystem-watcher-paths' import { cancelInFlightRemoteInstallIfUnowned, + forgetCancelledInstallWaiter, forgetDesiredRemoteWatcher, registerWatcherSenderCleanup, releaseRemoteWatchListener @@ -80,6 +81,11 @@ export function registerFilesystemWatcherHandlers(): void { // Why: the caller stopped watching on purpose — drop the intent or a later provider // registration would resurrect a watch nobody asked for. forgetDesiredRemoteWatcher(key, _event.sender.id) + forgetCancelledInstallWaiter( + watcherLifecycleState.cancelledRemoteInstallWaiters, + key, + _event.sender.id + ) const suspended = watcherLifecycleState.suspendedRemoteWatcherListeners.get(key) suspended?.listeners.delete(_event.sender.id) if (suspended?.listeners.size === 0) { diff --git a/src/main/ipc/filesystem-watcher-lifecycle-model.test.ts b/src/main/ipc/filesystem-watcher-lifecycle-model.test.ts new file mode 100644 index 00000000000..41c1af11dca --- /dev/null +++ b/src/main/ipc/filesystem-watcher-lifecycle-model.test.ts @@ -0,0 +1,396 @@ +// Randomized model test for local watcher listener bookkeeping across watch churn and worktree removals. +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createWatcherSender, fireWatcherSenderEvent } from './filesystem-watcher-test-sender' + +const { handleMock, statMock, subscribeMock } = vi.hoisted(() => ({ + handleMock: vi.fn(), + statMock: vi.fn(), + subscribeMock: vi.fn() +})) + +vi.mock('electron', () => ({ ipcMain: { handle: handleMock } })) +vi.mock('node:fs/promises', () => ({ stat: statMock })) +vi.mock('./parcel-watcher-process', () => ({ + subscribeViaWatcherProcess: subscribeMock, + disposeWatcherProcess: vi.fn() +})) +vi.mock('./filesystem-watcher-wsl', () => ({ createWslWatcher: vi.fn() })) +vi.mock('../providers/ssh-filesystem-dispatch', () => ({ + getSshFilesystemProvider: vi.fn(), + onSshFilesystemProviderRegistered: () => () => {} +})) + +import { closeAllWatchers } from './filesystem-watcher-shutdown' +import { registerFilesystemWatcherHandlers } from './filesystem-watcher-handlers' +import { desktopWorktreeWatcherRemoval } from './filesystem-watcher' +import { setWorktreeWatcherRemoval } from './worktree-watcher-removal' +import { createRuntimeFileWatcherRemoval } from '../runtime/runtime-file-watcher-removal' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { getLocalWatcherRoot } from './filesystem-watcher-paths' +import { + resetWatcherChildRegistryForTest, + WatcherChildCapacityError +} from './parcel-watcher-child-registry' + +type Handler = (event: unknown, args: unknown) => unknown +type Sender = ReturnType & { destroyed: boolean } +type SubscribeOutcome = 'ok' | 'capacity' +// Why 'wedged': a stuck watcher child ignores abort, so removals must fall back to their drain deadline. +type SubscribeDelay = 'none' | 'abortable' | 'wedged' +// Why 'rejected': a watch refused by an active removal may still come back from that removal's restore. +type Intent = 'yes' | 'no' | 'rejected' +type Removal = { + path: string + finish: ((removed: boolean) => Promise) | null +} + +const ROOTS = ['/repo', '/other'] +const OUTCOMES: SubscribeOutcome[] = ['ok', 'ok', 'ok', 'capacity'] +const DELAYS: SubscribeDelay[] = ['none', 'none', 'abortable', 'abortable', 'abortable', 'wedged'] +// Why: CI keeps this quick; raise ORCA_WATCH_MODEL_RUNS locally for deeper searches. +const RUNS = Number(process.env.ORCA_WATCH_MODEL_RUNS ?? 1000) +const MIN_STEPS = 20 +const MAX_STEPS = 60 +const GRACE_PLUS_MS = 31_000 + +function mulberry32(seed: number): () => number { + let a = seed >>> 0 + return () => { + a = (a + 0x6d2b79f5) >>> 0 + let t = a + t = Math.imul(t ^ (t >>> 15), t | 1) + t ^= t + Math.imul(t ^ (t >>> 7), t | 61) + return ((t ^ (t >>> 14)) >>> 0) / 4294967296 + } +} + +function rootKey(worktreePath: string): string { + return getLocalWatcherRoot(worktreePath).key +} + +async function settle(): Promise { + for (let i = 0; i < 40; i += 1) { + await Promise.resolve() + } + await vi.advanceTimersByTimeAsync(0) +} + +describe('local watcher lifecycle (randomized model)', () => { + const handlers: Record = {} + const deferred: (() => void)[] = [] + let liveSubscriptions = 0 + let nextOutcome: () => { outcome: SubscribeOutcome; delay: SubscribeDelay } + let trace: string[] = [] + const coordinator = createRuntimeFileWatcherRemoval({ + closeFileExplorerWatchersForPath: async () => {}, + restoreFileExplorerWatchersAfterFailedRemoval: async () => {}, + forgetFileExplorerWatchersAfterRemoval: () => {} + }) + + async function resolveAllDeferred(): Promise { + while (deferred.length > 0) { + deferred.shift()!() + await settle() + } + } + + // Like the watcher process, a pending subscribe rejects as soon as its install is aborted. + function waitForRelease(signal: AbortSignal | undefined, delay: SubscribeDelay): Promise { + return new Promise((resolve, reject) => { + const done = (): void => { + signal?.removeEventListener('abort', abort) + resolve() + } + const abort = (): void => { + deferred.splice(deferred.indexOf(done), 1) + reject(new Error('aborted')) + } + deferred.push(done) + if (delay === 'abortable') { + signal?.addEventListener('abort', abort, { once: true }) + } + }) + } + + beforeEach(async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] }) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + vi.spyOn(console, 'error').mockImplementation(() => {}) + resetWatcherChildRegistryForTest() + setWorktreeWatcherRemoval(desktopWorktreeWatcherRemoval) + handleMock.mockImplementation((channel: string, handler: Handler) => { + handlers[channel] = handler + }) + statMock.mockResolvedValue({ isDirectory: () => true }) + subscribeMock.mockImplementation( + async (dir: string, _cb: unknown, _opts: unknown, hooks?: { signal?: AbortSignal }) => { + const { outcome, delay } = nextOutcome() + trace.push(` subscribe ${dir}: ${delay} ${outcome}`) + if (delay !== 'none') { + await waitForRelease(hooks?.signal, delay) + } + if (outcome === 'capacity') { + throw new WatcherChildCapacityError() + } + liveSubscriptions += 1 + let unsubscribed = false + return { + unsubscribe: vi.fn(async () => { + if (unsubscribed) { + throw new Error('double unsubscribe') + } + unsubscribed = true + liveSubscriptions -= 1 + }) + } + } + ) + deferred.length = 0 + liveSubscriptions = 0 + registerFilesystemWatcherHandlers() + await closeAllWatchers() + }) + + afterEach(async () => { + await resolveAllDeferred() + await closeAllWatchers() + setWorktreeWatcherRemoval(null) + resetWatcherChildRegistryForTest() + vi.useRealTimers() + vi.restoreAllMocks() + }) + + async function runModel(seed: number, steps: number): Promise { + const rand = mulberry32(seed) + const pick = (items: readonly T[]): T => items[Math.floor(rand() * items.length)] + nextOutcome = () => ({ outcome: pick(OUTCOMES), delay: pick(DELAYS) }) + deferred.length = 0 + liveSubscriptions = 0 + let nextSenderId = 1 + const makeSender = (): Sender => { + const sender = Object.assign(createWatcherSender(nextSenderId++), { + destroyed: false + }) + sender.isDestroyed = () => sender.destroyed + return sender + } + const senders: Sender[] = [makeSender(), makeSender(), makeSender()] + // The renderer's last request per sender+root; missing = never asked. + const wanted = new Map() + const lastWatchOp = new Map() + const want = (senderId: number, key: string): string => `${senderId}|${key}` + let nextOp = 0 + trace = [] + const inFlight: Promise[] = [] + const removals: Removal[] = [] + const failed = (message: string): string => `${message}\n${trace.join('\n')}` + const liveSenders = (): Sender[] => senders.filter((candidate) => !candidate.destroyed) + + // Removals go through the real coordinator so the install fence and gate release order apply. + const startRemoval = (path: string): void => { + const removal: Removal = { path, finish: null } + removals.push(removal) + inFlight.push( + coordinator.acquire(path).then( + (handle) => { + removal.finish = handle.finish + }, + () => { + removals.splice(removals.indexOf(removal), 1) + } + ) + ) + } + + const finishRemoval = (removal: Removal, removed: boolean): void => { + removals.splice(removals.indexOf(removal), 1) + trace.push(`finish removal ${removal.path} (${removed ? 'succeeds' : 'fails, restore'})`) + // Why not awaited: the restore re-subscribes, which may itself be a deferred install. + inFlight.push(removal.finish!(removed)) + if (!removed) { + return + } + // The renderer drops the deleted workspace's watches. + const key = rootKey(removal.path) + for (const owner of liveSenders()) { + if (wanted.has(want(owner.id, key))) { + wanted.set(want(owner.id, key), 'no') + handlers['fs:unwatchWorktree']({ sender: owner }, { worktreePath: removal.path }) + } + } + } + + const drain = async (): Promise => { + while (deferred.length > 0 || inFlight.length > 0 || removals.length > 0) { + await resolveAllDeferred() + await Promise.allSettled(inFlight.splice(0)) + await settle() + for (const removal of removals.filter((candidate) => candidate.finish)) { + finishRemoval(removal, rand() < 0.5) + } + await settle() + } + } + + const checkListeners = (phase: string): string | null => { + const sources = [ + ['listens on', watcherLifecycleState.watchedRoots], + ['waits on capacity for', watcherLifecycleState.pendingLocalCapacityRetries] + ] as const + for (const [label, byKey] of sources) { + for (const [key, { listeners }] of byKey) { + for (const senderId of listeners.keys()) { + const intent = wanted.get(want(senderId, key)) + if (intent !== 'yes' && intent !== 'rejected') { + return `${phase}: unwatched sender ${senderId} still ${label} ${key}` + } + } + } + } + return null + } + + for (let step = 0; step < steps; step += 1) { + const roll = rand() + const sender = pick(liveSenders()) + const worktreePath = pick(ROOTS) + const key = rootKey(worktreePath) + const entry = want(sender.id, key) + if (roll < 0.36) { + trace.push(`watch s${sender.id} ${worktreePath}`) + const op = nextOp++ + wanted.set(entry, 'yes') + lastWatchOp.set(entry, op) + inFlight.push( + Promise.resolve(handlers['fs:watchWorktree']({ sender }, { worktreePath })).catch(() => { + if (lastWatchOp.get(entry) === op && wanted.get(entry) === 'yes') { + trace.push(` watch s${sender.id} ${worktreePath} rejected`) + wanted.set(entry, 'rejected') + } + }) + ) + } else if (roll < 0.58) { + trace.push(`unwatch s${sender.id} ${worktreePath}`) + wanted.set(entry, 'no') + lastWatchOp.delete(entry) + handlers['fs:unwatchWorktree']({ sender }, { worktreePath }) + } else if (roll < 0.72 && deferred.length > 0) { + trace.push('resolve-subscribe') + deferred.splice(Math.floor(rand() * deferred.length), 1)[0]() + } else if (roll < 0.77) { + trace.push('grace-elapses') + await vi.advanceTimersByTimeAsync(GRACE_PLUS_MS) + } else if (roll < 0.84) { + trace.push(`start removal ${worktreePath}`) + startRemoval(worktreePath) + } else if (roll < 0.91 && removals.some((removal) => removal.finish)) { + finishRemoval(pick(removals.filter((removal) => removal.finish)), rand() < 0.5) + } else if (roll < 0.97) { + // A reload keeps the WebContents id but starts a new page lifetime with no watches. + const closes = rand() < 0.5 + trace.push(`window s${sender.id} ${closes ? 'closes' : 'reloads'}`) + sender.destroyed = closes + fireWatcherSenderEvent(sender, closes ? 'destroyed' : 'did-navigate') + for (const wantedEntry of wanted.keys()) { + if (wantedEntry.startsWith(`${sender.id}|`)) { + wanted.set(wantedEntry, 'no') + lastWatchOp.delete(wantedEntry) + } + } + if (closes) { + senders.push(makeSender()) + } + } + await settle() + const stepFailure = checkListeners(`step ${step}`) + if (stepFailure) { + return failed(stepFailure) + } + } + + trace.push('drain') + await drain() + const quiescent = checkListeners('quiescent') + if (quiescent) { + return failed(quiescent) + } + for (const key of watcherLifecycleState.watchedRoots.keys()) { + if (watcherLifecycleState.inFlightLocalInstalls.has(key)) { + return failed(`root ${key} is installed and still installing`) + } + } + if (liveSubscriptions !== watcherLifecycleState.watchedRoots.size) { + return failed( + `${liveSubscriptions} native subscriptions for ${watcherLifecycleState.watchedRoots.size} roots` + ) + } + for (const [entry, intent] of wanted) { + const [senderIdText, key] = entry.split('|') + const senderId = Number(senderIdText) + const listening = watcherLifecycleState.watchedRoots.get(key)?.listeners.has(senderId) + const capacityWaiting = watcherLifecycleState.pendingLocalCapacityRetries + .get(key) + ?.listeners.has(senderId) + if (intent === 'yes' && !listening && !capacityWaiting) { + return failed(`wanted ${entry} is neither delivered nor waiting`) + } + } + + // Unwatching everything and waiting out the grace leaves no roots or native handles. + for (const sender of liveSenders()) { + for (const worktreePath of ROOTS) { + wanted.set(want(sender.id, rootKey(worktreePath)), 'no') + handlers['fs:unwatchWorktree']({ sender }, { worktreePath }) + } + } + await vi.advanceTimersByTimeAsync(GRACE_PLUS_MS) + await settle() + const leaked = { + watchedRoots: watcherLifecycleState.watchedRoots.size, + pendingTeardowns: watcherLifecycleState.pendingTeardowns.size, + inFlightInstalls: watcherLifecycleState.inFlightLocalInstalls.size, + pendingInstalls: watcherLifecycleState.pendingLocalInstallPromises.size, + capacityRetries: watcherLifecycleState.pendingLocalCapacityRetries.size, + suspended: watcherLifecycleState.suspendedLocalWatcherListeners.size, + liveSubscriptions + } + if (Object.values(leaked).some((count) => count !== 0)) { + return failed(`leaked after full unwatch: ${JSON.stringify(leaked)}`) + } + return null + } + + it('drops a re-watch that was unwatched while waiting out a cancelled install', async () => { + nextOutcome = () => ({ outcome: 'ok', delay: 'abortable' }) + const sender = createWatcherSender(1) + const args = { worktreePath: '/repo' } + const first = handlers['fs:watchWorktree']({ sender }, args) + await settle() + handlers['fs:unwatchWorktree']({ sender }, args) + const second = handlers['fs:watchWorktree']({ sender }, args) + await settle() + handlers['fs:unwatchWorktree']({ sender }, args) + await resolveAllDeferred() + await Promise.all([first, second]) + await settle() + + expect(watcherLifecycleState.watchedRoots.get(rootKey('/repo'))?.listeners.has(1)).not.toBe( + true + ) + await vi.advanceTimersByTimeAsync(GRACE_PLUS_MS) + expect(liveSubscriptions).toBe(0) + }) + + it('never leaks listeners, roots or native subscriptions across random watch histories', async () => { + for (let run = 0; run < RUNS; run += 1) { + const seed = 7_000_000 + run + const failure = await runModel(seed, MIN_STEPS + (seed % (MAX_STEPS - MIN_STEPS + 1))) + if (failure) { + expect(`seed ${seed}: ${failure}`).toBeNull() + } + await resolveAllDeferred() + await closeAllWatchers() + await settle() + } + }, 600_000) +}) diff --git a/src/main/ipc/filesystem-watcher-lifecycle-state.ts b/src/main/ipc/filesystem-watcher-lifecycle-state.ts index feef5d40373..823d5f86700 100644 --- a/src/main/ipc/filesystem-watcher-lifecycle-state.ts +++ b/src/main/ipc/filesystem-watcher-lifecycle-state.ts @@ -61,6 +61,10 @@ export const REMOTE_WATCH_RESYNC_COALESCE_MS = 5_000 export const REMOTE_WATCH_DORMANT_RETRY_MS = 60_000 export const REMOTE_WATCH_DORMANT_RETRY_MAX_MS = 30 * 60_000 +// Why: a sender waiting out a cancelled install is in no listener map; unwatch and removal still need to reach it. +export type CancelledInstallWaiter = { token: symbol; sender: WebContents } +export type CancelledInstallWaiters = Map> + export const watcherLifecycleState = { watchedRoots: new Map(), unwatchableRoots: new Set(), @@ -86,6 +90,7 @@ export const watcherLifecycleState = { inFlightLocalInstalls: new Map(), pendingLocalInstallPromises: new Map>(), pendingLocalCapacityRetries: new Map(), + cancelledLocalInstallWaiters: new Map>(), // Remote watcher state // Key: `${connectionId}:${worktreePath}`, Value: shared remote watch state. remoteWatchers: new Map(), @@ -119,6 +124,7 @@ export const watcherLifecycleState = { inFlightRemoteInstalls: new Map(), // Why: dedupe concurrent installRemoteWatcher calls per key so overlapping watches share one watcher instead of clobbering per-key state. pendingRemoteInstallPromises: new Map>(), + cancelledRemoteInstallWaiters: new Map>(), // Why: block installs beginning after closeAllWatchers (joiner recursion / retry tick bypass the abort loop); a new fs:watchWorktree clears it. remoteWatchersClosed: false, // Why: closeAllWatchers bumps this so a joiner that awaited across shutdown+reopen is refused (the latch alone can't tell it from a fresh call). diff --git a/src/main/ipc/filesystem-watcher-listener-lifecycle.ts b/src/main/ipc/filesystem-watcher-listener-lifecycle.ts index c2dd5f3dc77..438e177a0ef 100644 --- a/src/main/ipc/filesystem-watcher-listener-lifecycle.ts +++ b/src/main/ipc/filesystem-watcher-listener-lifecycle.ts @@ -2,6 +2,8 @@ import type { WebContents } from 'electron' import { isWatcherProcessFailure } from './parcel-watcher-process-failure' import type { WatchedRoot } from './filesystem-watcher-wsl' import { + type CancelledInstallWaiter, + type CancelledInstallWaiters, type LocalWatcherInstallToken, type RemoteWatcherInstallToken, UNWATCHABLE_ROOT_CACHE_MAX, @@ -173,6 +175,44 @@ export function releaseRemoteWatchListener(key: string, senderId: number): void watcherLifecycleState.remoteWatchers.delete(key) } +export function addCancelledInstallWaiter( + waitersByKey: CancelledInstallWaiters, + key: string, + sender: WebContents +): symbol { + const token = Symbol('cancelled-install-wait') + const waiters = waitersByKey.get(key) ?? new Map() + waiters.set(sender.id, { token, sender }) + waitersByKey.set(key, waiters) + return token +} + +/** Removes the waiter and reports whether it was still `token`'s, i.e. not withdrawn meanwhile. */ +export function takeCancelledInstallWaiter( + waitersByKey: CancelledInstallWaiters, + key: string, + senderId: number, + token: symbol +): boolean { + if (waitersByKey.get(key)?.get(senderId)?.token !== token) { + return false + } + forgetCancelledInstallWaiter(waitersByKey, key, senderId) + return true +} + +export function forgetCancelledInstallWaiter( + waitersByKey: CancelledInstallWaiters, + key: string, + senderId: number +): void { + const waiters = waitersByKey.get(key) + waiters?.delete(senderId) + if (waiters?.size === 0) { + waitersByKey.delete(key) + } +} + export function registerWatcherSenderCleanup(sender: WebContents): AbortSignal { return captureWatcherSenderLifetime(sender, () => { cleanupLocalWatchersForSender(sender.id) @@ -181,6 +221,9 @@ export function registerWatcherSenderCleanup(sender: WebContents): AbortSignal { } function cleanupLocalWatchersForSender(senderId: number): void { + for (const key of Array.from(watcherLifecycleState.cancelledLocalInstallWaiters.keys())) { + forgetCancelledInstallWaiter(watcherLifecycleState.cancelledLocalInstallWaiters, key, senderId) + } for (const [rootKey, suspended] of watcherLifecycleState.suspendedLocalWatcherListeners) { suspended.listeners.delete(senderId) if (suspended.listeners.size === 0) { @@ -208,6 +251,9 @@ function cleanupLocalWatchersForSender(senderId: number): void { } function cleanupRemoteWatchersForSender(senderId: number): void { + for (const key of Array.from(watcherLifecycleState.cancelledRemoteInstallWaiters.keys())) { + forgetCancelledInstallWaiter(watcherLifecycleState.cancelledRemoteInstallWaiters, key, senderId) + } for (const key of Array.from(watcherLifecycleState.desiredRemoteWatchers.keys())) { forgetDesiredRemoteWatcher(key, senderId) } diff --git a/src/main/ipc/filesystem-watcher-local-removal.ts b/src/main/ipc/filesystem-watcher-local-removal.ts index 78f015f7b94..252a1523f26 100644 --- a/src/main/ipc/filesystem-watcher-local-removal.ts +++ b/src/main/ipc/filesystem-watcher-local-removal.ts @@ -27,10 +27,12 @@ export async function closeLocalWatcherForWorktreePath( worktreePath, listeners: new Map() } + const waiters = watcherLifecycleState.cancelledLocalInstallWaiters.get(rootKey) for (const source of [ watcherLifecycleState.pendingLocalCapacityRetries.get(rootKey)?.listeners, watcherLifecycleState.inFlightLocalInstalls.get(rootKey)?.listeners, - watcherLifecycleState.watchedRoots.get(rootKey)?.listeners + watcherLifecycleState.watchedRoots.get(rootKey)?.listeners, + new Map(Array.from(waiters?.values() ?? [], ({ sender }) => [sender.id, sender])) ]) { for (const [senderId, sender] of source ?? []) { if (!sender.isDestroyed()) { @@ -38,6 +40,8 @@ export async function closeLocalWatcherForWorktreePath( } } } + // Why: a waiter's retry would install after this close; the restore re-subscribes it instead. + watcherLifecycleState.cancelledLocalInstallWaiters.delete(rootKey) if (suspended.listeners.size > 0) { watcherLifecycleState.suspendedLocalWatcherListeners.set(rootKey, suspended) } @@ -125,14 +129,24 @@ export async function restoreLocalWatcherAfterFailedRemoval(worktreePath: string if (!suspended) { return } - watcherLifecycleState.suspendedLocalWatcherListeners.delete(rootKey) - const failures: { sender: WebContents; signal: AbortSignal; error: unknown }[] = [] + const failures: { + sender: WebContents + signal: AbortSignal + error: unknown + }[] = [] const owners = Array.from(suspended.listeners.values(), (sender) => ({ sender, signal: registerWatcherSenderCleanup(sender) })) + const suspendedByKey = watcherLifecycleState.suspendedLocalWatcherListeners for (const { sender, signal } of owners) { - if (!isCurrentWatcherSender(sender, signal)) { + // Why: a forgotten or replaced snapshot belongs to a newer removal; this restore must not consume it. + if (suspendedByKey.get(rootKey) !== suspended) { + break + } + // Why: owners stay suspended until their turn, so an unwatch during an earlier owner's re-subscribe still withdraws them. + // Why lifetime first: a reloaded window keeps its id, and its new entry is not this restore's to take. + if (!isCurrentWatcherSender(sender, signal) || !suspended.listeners.delete(sender.id)) { continue } try { @@ -151,14 +165,22 @@ export async function restoreLocalWatcherAfterFailedRemoval(worktreePath: string failures.push({ sender, signal, error }) } } + // Why: an overlapping retried removal may have re-suspended owners into this same entry. + if (suspendedByKey.get(rootKey) === suspended && suspended.listeners.size === 0) { + suspendedByKey.delete(rootKey) + } const liveFailures = failures.filter(({ sender, signal }) => isCurrentWatcherSender(sender, signal) ) if (liveFailures.length > 0) { - watcherLifecycleState.suspendedLocalWatcherListeners.set(rootKey, { + const current = suspendedByKey.get(rootKey) ?? { worktreePath: suspended.worktreePath, - listeners: new Map(liveFailures.map(({ sender }) => [sender.id, sender])) - }) + listeners: new Map() + } + for (const { sender } of liveFailures) { + current.listeners.set(sender.id, sender) + } + suspendedByKey.set(rootKey, current) throw liveFailures[0].error } } diff --git a/src/main/ipc/filesystem-watcher-local-subscription.ts b/src/main/ipc/filesystem-watcher-local-subscription.ts index c667210078a..dcd4b838c20 100644 --- a/src/main/ipc/filesystem-watcher-local-subscription.ts +++ b/src/main/ipc/filesystem-watcher-local-subscription.ts @@ -7,11 +7,14 @@ import { } from './filesystem-watcher-lifecycle-state' import { getLocalWatcherRoot } from './filesystem-watcher-paths' import { + addCancelledInstallWaiter, addInFlightLocalInstallListener, addLocalWatchListener, clearLocalCapacityRetry, + forgetCancelledInstallWaiter, rememberUnwatchableRoot, registerWatcherSenderCleanup, + takeCancelledInstallWaiter, takeLocalCapacityRetryListeners, trackDetachedLocalUnsubscribe } from './filesystem-watcher-listener-lifecycle' @@ -22,6 +25,8 @@ import { isCurrentWatcherSender } from './filesystem-watcher-sender-lifetime' // ── Subscribe / Unsubscribe ────────────────────────────────────────── +const cancelledInstallWaiters = watcherLifecycleState.cancelledLocalInstallWaiters + export async function subscribeLocalWatcher( worktreePath: string, sender: WebContents, @@ -99,11 +104,17 @@ async function subscribeWhileRemovalAllowed( addInFlightLocalInstallListener(inFlight, listener) } } + const waitToken = canJoinInstall + ? null + : addCancelledInstallWaiter(cancelledInstallWaiters, rootKey, sender) const result = await pendingInstall + const stillWanted = + waitToken === null || + takeCancelledInstallWaiter(cancelledInstallWaiters, rootKey, sender.id, waitToken) const liveCapacityListeners = capacityRetryListeners.filter((listener) => isCurrentWatcherSender(listener, retrySignals.get(listener.id)!) ) - const senderIsCurrent = isCurrentWatcherSender(sender, senderSignal) + const senderIsCurrent = stillWanted && isCurrentWatcherSender(sender, senderSignal) if ( result === 'cancelled' && !canJoinInstall && @@ -188,6 +199,7 @@ async function subscribeWhileRemovalAllowed( export function unsubscribeLocalWatcher(worktreePath: string, senderId: number): void { const { key: rootKey } = getLocalWatcherRoot(worktreePath) + forgetCancelledInstallWaiter(cancelledInstallWaiters, rootKey, senderId) const suspended = watcherLifecycleState.suspendedLocalWatcherListeners.get(rootKey) suspended?.listeners.delete(senderId) if (suspended?.listeners.size === 0) { diff --git a/src/main/ipc/filesystem-watcher-remote-cancellation.test.ts b/src/main/ipc/filesystem-watcher-remote-cancellation.test.ts index b0d6e3b62a9..8639fdc2fb0 100644 --- a/src/main/ipc/filesystem-watcher-remote-cancellation.test.ts +++ b/src/main/ipc/filesystem-watcher-remote-cancellation.test.ts @@ -20,6 +20,8 @@ vi.mock('../providers/ssh-filesystem-dispatch', () => ({ import { closeAllWatchers } from './filesystem-watcher-shutdown' import { registerFilesystemWatcherHandlers } from './filesystem-watcher-handlers' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { getRemoteWatcherKey } from './filesystem-watcher-paths' type HandlerMap = Record unknown> @@ -122,6 +124,41 @@ describe('remote filesystem watcher cancellation', () => { expect(secondUnwatch).toHaveBeenCalledTimes(1) }) + it('drops a re-watch that was unwatched while waiting out a cancelled SSH install', async () => { + let resolveFirst: ((unwatch: () => void) => void) | undefined + let firstSignal: AbortSignal | undefined + const lateUnwatch = vi.fn() + const watchMock = vi + .fn() + .mockImplementationOnce( + (_rootPath, _callback, options?: { signal?: AbortSignal }) => + new Promise<() => void>((resolve) => { + firstSignal = options?.signal + resolveFirst = resolve + }) + ) + .mockImplementation(() => Promise.resolve(vi.fn())) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const args = { worktreePath: '/home/me/repo', connectionId: 'conn-1' } + const sender = createWatcherSender(1) + const first = Promise.resolve(handlers['fs:watchWorktree']({ sender }, args)) + await Promise.resolve() + handlers['fs:unwatchWorktree']({ sender }, args) + await vi.waitFor(() => expect(firstSignal?.aborted).toBe(true)) + const second = Promise.resolve(handlers['fs:watchWorktree']({ sender }, args)) + await Promise.resolve() + handlers['fs:unwatchWorktree']({ sender }, args) + + resolveFirst?.(lateUnwatch) + await Promise.all([first, second]) + + expect(watchMock).toHaveBeenCalledTimes(1) + expect(lateUnwatch).toHaveBeenCalledTimes(1) + expect( + watcherLifecycleState.remoteWatchers.get(getRemoteWatcherKey('conn-1', args.worktreePath)) + ).toBeUndefined() + }) + it('aborts pending SSH setup on sender destruction and watcher shutdown', async () => { const installs = new Map< string, diff --git a/src/main/ipc/filesystem-watcher-remote-controller.ts b/src/main/ipc/filesystem-watcher-remote-controller.ts index 8a4f4cc2e91..b51b74ba251 100644 --- a/src/main/ipc/filesystem-watcher-remote-controller.ts +++ b/src/main/ipc/filesystem-watcher-remote-controller.ts @@ -85,6 +85,7 @@ const handleRemoteWatcherTerminalError: RemoteWatcherTerminalErrorHandler = ( } watcherLifecycleState.remoteWatchers.delete(key) state.batch.close() + // Why: a suspended snapshot means a removal or its restore owns these listeners; a retry would race it. if ( watcherLifecycleState.remoteWatchersClosed || watcherLifecycleState.suspendedRemoteWatcherListeners.has(key) diff --git a/src/main/ipc/filesystem-watcher-remote-install.ts b/src/main/ipc/filesystem-watcher-remote-install.ts index b1a6bab61fa..3ee37739aad 100644 --- a/src/main/ipc/filesystem-watcher-remote-install.ts +++ b/src/main/ipc/filesystem-watcher-remote-install.ts @@ -17,11 +17,15 @@ import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' import { getRemoteWatcherKey } from './filesystem-watcher-paths' import { isCurrentWatcherSender } from './filesystem-watcher-sender-lifetime' import { + addCancelledInstallWaiter, addInFlightRemoteInstallListener, addRemoteWatchListener, - registerWatcherSenderCleanup + registerWatcherSenderCleanup, + takeCancelledInstallWaiter } from './filesystem-watcher-listener-lifecycle' +const cancelledRemoteInstallWaiters = watcherLifecycleState.cancelledRemoteInstallWaiters + export type RemoteWatcherTerminalErrorHandler = ( key: string, connectionId: string, @@ -89,8 +93,14 @@ async function installRemoteWatcherWhileRemovalAllowed( // Why: a new watcher joining before provider.watch() resolves should revive the install instead of inheriting the stale cancellation. addInFlightRemoteInstallListener(inFlight, sender) } + const waitToken = canJoinInstall + ? null + : addCancelledInstallWaiter(cancelledRemoteInstallWaiters, key, sender) const result = await pendingInstall - if (!isCurrentWatcherSender(sender, senderSignal)) { + const stillWanted = + waitToken === null || + takeCancelledInstallWaiter(cancelledRemoteInstallWaiters, key, sender.id, waitToken) + if (!stillWanted || !isCurrentWatcherSender(sender, senderSignal)) { return 'cancelled' } if ( diff --git a/src/main/ipc/filesystem-watcher-remote-removal.ts b/src/main/ipc/filesystem-watcher-remote-removal.ts index a6c7f2a6636..94da50724b3 100644 --- a/src/main/ipc/filesystem-watcher-remote-removal.ts +++ b/src/main/ipc/filesystem-watcher-remote-removal.ts @@ -2,7 +2,10 @@ import { isCurrentWatcherSender } from './filesystem-watcher-sender-lifetime' import type { WebContents } from 'electron' import type { FsChangedPayload } from '../../shared/filesystem-entry-types' import { getSshFilesystemProvider } from '../providers/ssh-filesystem-dispatch' -import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { + watcherLifecycleState, + type RemoteWatcherInstallResult +} from './filesystem-watcher-lifecycle-state' import { getRemoteWatcherKey } from './filesystem-watcher-paths' import { clearDormantRemoteWatcher, @@ -25,10 +28,12 @@ export async function closeRemoteWatcherForWorktreePath( worktreePath, listeners: new Map() } + const waiters = watcherLifecycleState.cancelledRemoteInstallWaiters.get(key) for (const source of [ watcherLifecycleState.pendingRemoteWatcherRetryListeners.get(key)?.listeners, watcherLifecycleState.inFlightRemoteInstalls.get(key)?.listeners, - watcherLifecycleState.remoteWatchers.get(key)?.listeners + watcherLifecycleState.remoteWatchers.get(key)?.listeners, + new Map(Array.from(waiters?.values() ?? [], ({ sender }) => [sender.id, sender])) ]) { for (const [senderId, sender] of source ?? []) { if (!sender.isDestroyed()) { @@ -36,6 +41,8 @@ export async function closeRemoteWatcherForWorktreePath( } } } + // Why: a waiter's retry would install after this close; the restore re-subscribes it instead. + watcherLifecycleState.cancelledRemoteInstallWaiters.delete(key) if (suspended.listeners.size > 0) { watcherLifecycleState.suspendedRemoteWatcherListeners.set(key, suspended) } @@ -72,16 +79,35 @@ export async function restoreRemoteWatcherAfterFailedRemoval( if (!suspended) { return } - watcherLifecycleState.suspendedRemoteWatcherListeners.delete(key) + const suspendedByKey = watcherLifecycleState.suspendedRemoteWatcherListeners + const failures: { + sender: WebContents + signal: AbortSignal + error: unknown + }[] = [] const owners = Array.from(suspended.listeners.values(), (sender) => ({ sender, signal: registerWatcherSenderCleanup(sender) })) for (const { sender, signal } of owners) { - if (!isCurrentWatcherSender(sender, signal)) { + // Why: a forgotten or replaced snapshot belongs to a newer removal; this restore must not consume it. + if (suspendedByKey.get(key) !== suspended) { + break + } + // Why: owners stay suspended until their turn, so an unwatch during an earlier owner's re-subscribe still withdraws them. + // Why lifetime first: a reloaded window keeps its id, and its new entry is not this restore's to take. + if (!isCurrentWatcherSender(sender, signal) || !suspended.listeners.delete(sender.id)) { + continue + } + let result: RemoteWatcherInstallResult + try { + result = await installRemoteWatcher(sender, connectionId, worktreePath, undefined, signal) + } catch (error) { + if (isCurrentWatcherSender(sender, signal)) { + failures.push({ sender, signal, error }) + } continue } - const result = await installRemoteWatcher(sender, connectionId, worktreePath, undefined, signal) if (!isCurrentWatcherSender(sender, signal)) { continue } @@ -95,6 +121,25 @@ export async function restoreRemoteWatcherAfterFailedRemoval( events: [{ kind: 'overflow', absolutePath: worktreePath }] } satisfies FsChangedPayload) } + // Why: an overlapping retried removal may have re-suspended owners into this same entry. + if (suspendedByKey.get(key) === suspended && suspended.listeners.size === 0) { + suspendedByKey.delete(key) + } + const liveFailures = failures.filter(({ sender, signal }) => + isCurrentWatcherSender(sender, signal) + ) + if (liveFailures.length > 0) { + const current = suspendedByKey.get(key) ?? { + connectionId, + worktreePath, + listeners: new Map() + } + for (const { sender } of liveFailures) { + current.listeners.set(sender.id, sender) + } + suspendedByKey.set(key, current) + throw liveFailures[0].error + } } export function forgetRemoteWatcherRemovalSnapshot( diff --git a/src/main/ipc/filesystem-watcher-removal-restore.test.ts b/src/main/ipc/filesystem-watcher-removal-restore.test.ts new file mode 100644 index 00000000000..c7f046ad4b6 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-removal-restore.test.ts @@ -0,0 +1,253 @@ +// Restores after a failed worktree removal must not resurrect unwatched windows or lose a newer removal's owners. +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createWatcherSender, fireWatcherSenderEvent } from './filesystem-watcher-test-sender' + +const { handleMock, statMock, subscribeMock, getSshFilesystemProviderMock } = vi.hoisted(() => ({ + handleMock: vi.fn(), + statMock: vi.fn(), + subscribeMock: vi.fn(), + getSshFilesystemProviderMock: vi.fn() +})) + +vi.mock('electron', () => ({ ipcMain: { handle: handleMock } })) +vi.mock('node:fs/promises', () => ({ stat: statMock })) +vi.mock('./parcel-watcher-process', () => ({ + subscribeViaWatcherProcess: subscribeMock, + disposeWatcherProcess: vi.fn() +})) +vi.mock('./filesystem-watcher-wsl', () => ({ createWslWatcher: vi.fn() })) +vi.mock('../providers/ssh-filesystem-dispatch', () => ({ + getSshFilesystemProvider: getSshFilesystemProviderMock, + onSshFilesystemProviderRegistered: () => () => {} +})) + +import { closeAllWatchers } from './filesystem-watcher-shutdown' +import { registerFilesystemWatcherHandlers } from './filesystem-watcher-handlers' +import { desktopWorktreeWatcherRemoval } from './filesystem-watcher' +import { setWorktreeWatcherRemoval } from './worktree-watcher-removal' +import { createRuntimeFileWatcherRemoval } from '../runtime/runtime-file-watcher-removal' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { getLocalWatcherRoot, getRemoteWatcherKey } from './filesystem-watcher-paths' +import { resetWatcherChildRegistryForTest } from './parcel-watcher-child-registry' + +type Handler = (event: unknown, args: unknown) => unknown + +const CONNECTION = 'conn-1' +const LOCAL: { worktreePath: string; connectionId?: string } = { worktreePath: '/repo' } +const REMOTE = { worktreePath: '/home/me/repo', connectionId: CONNECTION } + +async function settle(): Promise { + for (let i = 0; i < 40; i += 1) { + await Promise.resolve() + } +} + +describe('watcher restore after a failed worktree removal', () => { + const handlers: Record = {} + const deferred: (() => void)[] = [] + let deferSubscribe = false + // A wedged watcher child ignores abort until it finally answers. + let wedged = false + const coordinator = createRuntimeFileWatcherRemoval({ + closeFileExplorerWatchersForPath: async () => {}, + restoreFileExplorerWatchersAfterFailedRemoval: async () => {}, + forgetFileExplorerWatchersAfterRemoval: () => {} + }) + + async function resolveAllDeferred(): Promise { + while (deferred.length > 0) { + deferred.shift()!() + await settle() + } + } + + function waitIfDeferred(signal: AbortSignal | undefined): Promise { + if (!deferSubscribe) { + return Promise.resolve() + } + return new Promise((resolve, reject) => { + const done = (): void => { + signal?.removeEventListener('abort', abort) + resolve() + } + const abort = (): void => { + deferred.splice(deferred.indexOf(done), 1) + reject(new Error('aborted')) + } + deferred.push(done) + if (!wedged) { + signal?.addEventListener('abort', abort, { once: true }) + } + }) + } + + beforeEach(async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + vi.spyOn(console, 'error').mockImplementation(() => {}) + resetWatcherChildRegistryForTest() + setWorktreeWatcherRemoval(desktopWorktreeWatcherRemoval) + deferSubscribe = false + wedged = false + deferred.length = 0 + handleMock.mockImplementation((channel: string, handler: Handler) => { + handlers[channel] = handler + }) + statMock.mockResolvedValue({ isDirectory: () => true }) + subscribeMock.mockImplementation( + async (_dir: string, _cb: unknown, _opts: unknown, hooks?: { signal?: AbortSignal }) => { + await waitIfDeferred(hooks?.signal) + return { unsubscribe: vi.fn(async () => {}) } + } + ) + getSshFilesystemProviderMock.mockReturnValue({ + watch: vi.fn(async (_root: string, _cb: unknown, options?: { signal?: AbortSignal }) => { + await waitIfDeferred(options?.signal) + return () => {} + }) + }) + registerFilesystemWatcherHandlers() + await closeAllWatchers() + }) + + afterEach(async () => { + await resolveAllDeferred() + await closeAllWatchers() + setWorktreeWatcherRemoval(null) + resetWatcherChildRegistryForTest() + vi.restoreAllMocks() + }) + + const localListeners = (): number[] => [ + ...(watcherLifecycleState.watchedRoots + .get(getLocalWatcherRoot('/repo').key) + ?.listeners.keys() ?? []) + ] + const remoteListeners = (): number[] => [ + ...(watcherLifecycleState.remoteWatchers + .get(getRemoteWatcherKey(CONNECTION, REMOTE.worktreePath)) + ?.listeners.keys() ?? []) + ] + + it.each([ + ['local', LOCAL, localListeners], + ['SSH', REMOTE, remoteListeners] + ])( + 'drops a %s window that unwatches while an earlier window is being restored', + async (_label, args, listeners) => { + const first = createWatcherSender(1) + const second = createWatcherSender(2) + await handlers['fs:watchWorktree']({ sender: first }, args) + await handlers['fs:watchWorktree']({ sender: second }, args) + const removal = await coordinator.acquire(args.worktreePath, args.connectionId) + + deferSubscribe = true + const restoring = removal.finish(false) + await settle() + handlers['fs:unwatchWorktree']({ sender: second }, args) + await resolveAllDeferred() + await restoring + + expect(listeners()).toEqual([1]) + expect(watcherLifecycleState.suspendedLocalWatcherListeners.size).toBe(0) + expect(watcherLifecycleState.suspendedRemoteWatcherListeners.size).toBe(0) + } + ) + + it.each([ + ['local', LOCAL, localListeners], + ['SSH', REMOTE, remoteListeners] + ])( + 'keeps a retried %s removal’s owners when it overlaps the previous restore', + async (_label, args, listeners) => { + const first = createWatcherSender(1) + const second = createWatcherSender(2) + await handlers['fs:watchWorktree']({ sender: first }, args) + await handlers['fs:watchWorktree']({ sender: second }, args) + const firstRemoval = await coordinator.acquire(args.worktreePath, args.connectionId) + + // The first restore is mid-way through re-subscribing window 1 when the retry starts. + deferSubscribe = true + const firstRestore = firstRemoval.finish(false) + await settle() + const secondAcquire = coordinator.acquire(args.worktreePath, args.connectionId) + await settle() + await resolveAllDeferred() + const secondRemoval = await secondAcquire + await firstRestore + + deferSubscribe = false + await secondRemoval.finish(false) + await settle() + + expect(listeners().sort()).toEqual([1, 2]) + expect(watcherLifecycleState.suspendedLocalWatcherListeners.size).toBe(0) + expect(watcherLifecycleState.suspendedRemoteWatcherListeners.size).toBe(0) + } + ) + + it.each([ + ['local', LOCAL, localListeners], + ['SSH', REMOTE, remoteListeners] + ])( + 'keeps a %s window that reloads (same id) while overlapping restores run', + async (_label, args, listeners) => { + const first = createWatcherSender(1) + const second = createWatcherSender(2) + // Window 3 keeps the first restore's snapshot non-empty after window 2's old page is cleaned up. + const third = createWatcherSender(3) + for (const sender of [first, second, third]) { + await handlers['fs:watchWorktree']({ sender }, args) + } + const firstRemoval = await coordinator.acquire(args.worktreePath, args.connectionId) + + deferSubscribe = true + const firstRestore = firstRemoval.finish(false) + await settle() + // The first restore captured window 2's old lifetime; the reloaded page watches again. + fireWatcherSenderEvent(second, 'did-navigate') + const rewatch = Promise.resolve(handlers['fs:watchWorktree']({ sender: second }, args)) + await settle() + const secondAcquire = coordinator.acquire(args.worktreePath, args.connectionId) + await settle() + await resolveAllDeferred() + const secondRemoval = await secondAcquire + await firstRestore + await rewatch.catch(() => {}) + + deferSubscribe = false + await secondRemoval.finish(false) + await settle() + + expect(listeners().sort()).toEqual([1, 2, 3]) + } + ) + + it.each([ + ['local', LOCAL, localListeners], + ['SSH', REMOTE, remoteListeners] + ])( + 'restores a %s window that was waiting out a cancelled install when removal began', + async (_label, args, listeners) => { + const sender = createWatcherSender(1) + deferSubscribe = true + wedged = true + const first = Promise.resolve(handlers['fs:watchWorktree']({ sender }, args)) + await settle() + handlers['fs:unwatchWorktree']({ sender }, args) + await settle() + // The re-watch can't join the aborted install, so it waits for it in no listener map. + deferSubscribe = false + const second = Promise.resolve(handlers['fs:watchWorktree']({ sender }, args)) + await settle() + const acquiring = coordinator.acquire(args.worktreePath, args.connectionId) + await settle() + await resolveAllDeferred() + const removal = await acquiring + await Promise.allSettled([first, second]) + await removal.finish(false) + await settle() + + expect(listeners()).toEqual([1]) + } + ) +}) diff --git a/src/main/ipc/filesystem-watcher-shutdown.ts b/src/main/ipc/filesystem-watcher-shutdown.ts index 31a57ecab10..c956e0bcd0a 100644 --- a/src/main/ipc/filesystem-watcher-shutdown.ts +++ b/src/main/ipc/filesystem-watcher-shutdown.ts @@ -15,6 +15,8 @@ export async function closeAllWatchers(): Promise { watcherLifecycleState.unwatchableRoots.clear() watcherLifecycleState.suspendedLocalWatcherListeners.clear() watcherLifecycleState.suspendedRemoteWatcherListeners.clear() + watcherLifecycleState.cancelledLocalInstallWaiters.clear() + watcherLifecycleState.cancelledRemoteInstallWaiters.clear() for (const retry of watcherLifecycleState.pendingLocalCapacityRetries.values()) { retry.cancelWait() } diff --git a/src/main/ipc/filesystem-watcher-test-sender.ts b/src/main/ipc/filesystem-watcher-test-sender.ts index 4ff5a5bbcc1..a309d080087 100644 --- a/src/main/ipc/filesystem-watcher-test-sender.ts +++ b/src/main/ipc/filesystem-watcher-test-sender.ts @@ -16,3 +16,14 @@ export function createWatcherSender( } { return { id, isDestroyed: () => false, send, ...senderEvents() } } + +/** Fires the live (registered, not removed, not yet fired) `once` handlers for `event`, as Electron would. */ +export function fireWatcherSenderEvent(sender: SenderEvents, event: string): void { + const removed = new Set(sender.removeListener.mock.calls.map(([, callback]) => callback)) + for (const [name, callback] of sender.once.mock.calls) { + if (name === event && !removed.has(callback)) { + sender.removeListener(name, callback) + callback() + } + } +}