From 5c71a6cc6bfa386d9ee079ff99d03efa1116e8d3 Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Sun, 18 Oct 2026 23:44:01 -0700 Subject: [PATCH] fix(watcher): stop unwatched windows from being re-subscribed to local watchers (#30412) * fix(watcher): stop unwatched windows from being re-subscribed to local watchers Two pre-existing races left a window subscribed after it unwatched: - a re-watch waiting out a cancelled install was not tracked, so its unwatch could not withdraw it and the retry subscribed it anyway - restoring after a failed worktree removal dropped the suspended owners up front, so an owner that unwatched mid-restore was still re-subscribed Adds a randomized lifecycle model test plus regressions for both sequences. * fix(watcher): keep a retried removal's owners when it overlaps the previous restore The restore now leaves its snapshot in place while it works, so a retried removal can add owners to the same entry. Stop restoring once the snapshot is forgotten or replaced, delete it only when empty, and merge failed owners into the current entry instead of overwriting it. Apply the same per-owner restore to SSH watchers, which dropped the saved list before re-subscribing. The lifecycle model now drives removals through the real coordinator and lets them overlap restores; adds local and SSH restore regressions. * fix(watcher): drop SSH re-watches unwatched while waiting out a cancelled install Same leak as the local fix: a watch that could not join an aborted SSH install waited untracked, so its unwatch had nothing to remove and the retry installed it anyway. Record those waiters, clear them on unwatch, and retry only if the sender is still recorded. * fix(watcher): keep reloaded windows and cancelled-install waiters across removals - An older restore checked a window's lifetime only after taking its entry, so a window reloaded under the same id lost its place in a newer removal's snapshot. Check the lifetime first (local and SSH). - Removal now suspends senders waiting out a cancelled install and drops their retry; previously the retry installed mid-removal and the close tore it down, losing the window. Waiters keep their WebContents for this. - Waiter maps live in watcher state: cleared on shutdown and per sender on cleanup. - Model: windows can reload (same id, new lifetime) and subscribes can be wedged (ignore abort). --- src/main/ipc/filesystem-watcher-handlers.ts | 6 + ...filesystem-watcher-lifecycle-model.test.ts | 396 ++++++++++++++++++ .../ipc/filesystem-watcher-lifecycle-state.ts | 6 + .../filesystem-watcher-listener-lifecycle.ts | 46 ++ .../ipc/filesystem-watcher-local-removal.ts | 36 +- .../filesystem-watcher-local-subscription.ts | 14 +- ...system-watcher-remote-cancellation.test.ts | 37 ++ .../filesystem-watcher-remote-controller.ts | 1 + .../ipc/filesystem-watcher-remote-install.ts | 14 +- .../ipc/filesystem-watcher-remote-removal.ts | 55 ++- ...filesystem-watcher-removal-restore.test.ts | 253 +++++++++++ src/main/ipc/filesystem-watcher-shutdown.ts | 2 + .../ipc/filesystem-watcher-test-sender.ts | 11 + 13 files changed, 862 insertions(+), 15 deletions(-) create mode 100644 src/main/ipc/filesystem-watcher-lifecycle-model.test.ts create mode 100644 src/main/ipc/filesystem-watcher-removal-restore.test.ts 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() + } + } +}