diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index ba972a6f80c..c6ab3abefa6 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -9,7 +9,6 @@ inline src/main/browser/browser-manager.ts inline src/main/codex-accounts/runtime-home-service.ts inline src/main/codex-accounts/service.ts inline src/main/index.ts -inline src/main/ipc/filesystem-watcher.ts inline src/main/ipc/filesystem.ts inline src/main/ipc/worktree-remote.ts inline src/main/rate-limits/service.ts diff --git a/src/main/ipc/filesystem-watcher-handlers.ts b/src/main/ipc/filesystem-watcher-handlers.ts new file mode 100644 index 00000000000..7a2b6230abd --- /dev/null +++ b/src/main/ipc/filesystem-watcher-handlers.ts @@ -0,0 +1,97 @@ +import { ipcMain } from 'electron' +import { onSshFilesystemProviderRegistered } from '../providers/ssh-filesystem-dispatch' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { getRemoteWatcherKey } from './filesystem-watcher-paths' +import { + cancelInFlightRemoteInstallIfUnowned, + forgetDesiredRemoteWatcher, + releaseRemoteWatchListener +} from './filesystem-watcher-listener-lifecycle' +import { + installRemoteWatcher, + reinstallRemoteWatchersForConnection, + scheduleRemoteWatcherRetry +} from './filesystem-watcher-remote-controller' +import { rememberDesiredRemoteWatcher } from './filesystem-watcher-remote-desired' +import { + subscribeLocalWatcher, + unsubscribeLocalWatcher +} from './filesystem-watcher-local-subscription' + +export function registerFilesystemWatcherHandlers(): void { + // Why: re-registration replaces the handler set, so drop the previous subscription instead of + // stacking a second re-arm on every provider registration. + watcherLifecycleState.unsubscribeFromProviderRegistrations?.() + watcherLifecycleState.unsubscribeFromProviderRegistrations = onSshFilesystemProviderRegistered( + reinstallRemoteWatchersForConnection + ) + + ipcMain.handle( + 'fs:watchWorktree', + async (event, args: { worktreePath: string; connectionId?: string }): Promise => { + if (args.connectionId) { + // Why: a real new watch reopens the subsystem after closeAllWatchers latched it shut (also resets tests between cases). + watcherLifecycleState.remoteWatchersClosed = false + const key = getRemoteWatcherKey(args.connectionId, args.worktreePath) + // Why: record intent before the install so a provider registering mid-flight (or long after + // this attempt gives up) can still re-arm this listener. + rememberDesiredRemoteWatcher(args.connectionId, args.worktreePath, event.sender) + const result = await installRemoteWatcher( + event.sender, + args.connectionId, + args.worktreePath + ) + if (result === 'unavailable') { + if (!watcherLifecycleState.loggedUnavailableRemoteWatchers.has(key)) { + watcherLifecycleState.loggedUnavailableRemoteWatchers.add(key) + console.warn( + `[filesystem-watcher] SSH filesystem provider unavailable; retrying watch for ${args.worktreePath} on connection ${args.connectionId}` + ) + } + scheduleRemoteWatcherRetry(event.sender, args.connectionId, args.worktreePath) + return + } + return + } + // Why: reopen the local subsystem for tests and post-shutdown reattachment; stale callers keep the prior generation. + watcherLifecycleState.localWatchersClosed = false + await subscribeLocalWatcher(args.worktreePath, event.sender) + } + ) + + ipcMain.handle( + 'fs:unwatchWorktree', + (_event, args: { worktreePath: string; connectionId?: string }): void => { + if (args.connectionId) { + const key = getRemoteWatcherKey(args.connectionId, args.worktreePath) + // 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) + const suspended = watcherLifecycleState.suspendedRemoteWatcherListeners.get(key) + suspended?.listeners.delete(_event.sender.id) + if (suspended?.listeners.size === 0) { + watcherLifecycleState.suspendedRemoteWatcherListeners.delete(key) + } + const retry = watcherLifecycleState.pendingRemoteWatcherRetryListeners.get(key) + retry?.listeners.delete(_event.sender.id) + const retryTimer = watcherLifecycleState.pendingRemoteWatcherRetries.get(key) + if (retryTimer && retry?.listeners.size === 0) { + clearTimeout(retryTimer) + watcherLifecycleState.pendingRemoteWatcherRetries.delete(key) + watcherLifecycleState.pendingRemoteWatcherRetryListeners.delete(key) + } + // Why: a retry-tick provider.watch() may still be in flight; mark cancelled so its resolved unwatch handle is discarded. + const inFlight = watcherLifecycleState.inFlightRemoteInstalls.get(key) + if (inFlight) { + inFlight.listeners.delete(_event.sender.id) + cancelInFlightRemoteInstallIfUnowned(inFlight) + } + watcherLifecycleState.loggedUnavailableRemoteWatchers.delete(key) + releaseRemoteWatchListener(key, _event?.sender?.id ?? 0) + return + } + const senderId = _event.sender.id + unsubscribeLocalWatcher(args.worktreePath, senderId) + } + ) +} diff --git a/src/main/ipc/filesystem-watcher-lifecycle-state.ts b/src/main/ipc/filesystem-watcher-lifecycle-state.ts new file mode 100644 index 00000000000..548d958f90c --- /dev/null +++ b/src/main/ipc/filesystem-watcher-lifecycle-state.ts @@ -0,0 +1,124 @@ +import type { WebContents } from 'electron' +import type { WatchedRoot } from './filesystem-watcher-wsl' +import type { RemoteWatcherEventBatch } from './remote-watcher-event-batch' + +// ── Per-root watcher state ─────────────────────────────────────────── +// WatchedRoot/WatcherSubscription live in filesystem-watcher-wsl.ts so native and WSL watchers share one shape. + +// ── Module state ───────────────────────────────────────────────────── + +export type LocalWatcherInstallToken = { + cancelled: boolean + listeners: Map + abortController: AbortController +} + +export type LocalWatcherInstallResult = 'installed' | 'unavailable' | 'capacity' | 'cancelled' + +export type LocalWatcherCapacityRetry = { + listeners: Map + cancelWait: () => void +} + +export type RemoteWatcherInstallToken = { + cancelled: boolean + listeners: Map + abortController: AbortController + abortScheduled: boolean + terminalError?: Error +} + +export type RemoteWatcherState = { + unwatch: () => void + listeners: Map + installToken: RemoteWatcherInstallToken + /** Held so every teardown can drop the trailing flush timer instead of letting it fire into a dead watch. */ + batch: RemoteWatcherEventBatch +} + +export type RemoteWatcherInstallResult = 'installed' | 'unavailable' | 'cancelled' + +export type RemoteWatcherResyncState = { + lastSentAt: number + listeners: Map + timer?: ReturnType + worktreePath: string +} + +// Why: cache roots that failed watcher creation (e.g. WSL UNC paths) so we don't retry every worktree switch and spam the console with errors. +export const UNWATCHABLE_ROOT_CACHE_MAX = 256 +// Why: recreating @parcel/watcher on Windows is expensive (ReadDirectoryChangesW + AV, 500ms+); a 30s grace lets rapid switches reuse the watcher. +export const WATCHER_TEARDOWN_GRACE_MS = 30_000 +export const REMOTE_WATCH_RETRY_MS = 1_000 +export const REMOTE_WATCH_RETRY_TIMEOUT_MS = 60_000 +// Why: preserve the first and latest resync while bounding full-tree SSH refreshes during flaps. +export const REMOTE_WATCH_RESYNC_COALESCE_MS = 5_000 +// Why: doubling from a minute to a half-hour ceiling costs a permanently broken remote ~7 fs.watch +// calls in the first hour and 2/hour after, which a flapping link can absorb. +export const REMOTE_WATCH_DORMANT_RETRY_MS = 60_000 +export const REMOTE_WATCH_DORMANT_RETRY_MAX_MS = 30 * 60_000 + +export const watcherLifecycleState = { + watchedRoots: new Map(), + unwatchableRoots: new Set(), + // Why: key cleanup by sender WebContents (not per root) to avoid MaxListeners warnings when a workspace has many worktrees open. + senderCleanupRegistered: new Set(), + pendingTeardowns: new Map>(), + // Why: @parcel/watcher unsubscribe does native async work that sender-destroy can start before shutdown, so will-quit must still await it. + pendingLocalUnsubscribes: new Set>(), + pendingLocalUnsubscribesByRoot: new Map>>(), + suspendedLocalWatcherListeners: new Map< + string, + { worktreePath: string; listeners: Map } + >(), + // Why: an install cancelled by shutdown can't be revived by a waiter that resumes after a later call reopens the subsystem. + localWatchersClosed: false, + localWatcherLifecycleGeneration: 0, + failedLocalUnsubscribes: new Map(), + // Why: a drain that timed out no longer gates the delete — Git removal already proceeded past it. Its + // late failure must not fail-close a *later* close of the same root, which would leave that path + // undeletable until the watcher process physically exits. + abandonedLocalUnsubscribes: new WeakSet>(), + // Why: watcher creation is async; concurrent watch requests for the same root must share one install or later resolves orphan listeners. + inFlightLocalInstalls: new Map(), + pendingLocalInstallPromises: new Map>(), + pendingLocalCapacityRetries: new Map(), + // Remote watcher state + // Key: `${connectionId}:${worktreePath}`, Value: shared remote watch state. + remoteWatchers: new Map(), + suspendedRemoteWatcherListeners: new Map< + string, + { connectionId: string; worktreePath: string; listeners: Map } + >(), + // Why: the renderer subscribes once per target and never re-issues, so the intent to watch has to + // outlive any single connection — an install that failed or died with a dropped transport is + // re-armed from here when a provider appears. Without it a reconnect (or a connect slower than the + // retry window) leaves the watch dead until the app restarts. + desiredRemoteWatchers: new Map< + string, + { connectionId: string; worktreePath: string; listeners: Map } + >(), + // Why: provider registration only fires on reconnect, so a watch that dies while the SSH link stays + // healthy (remote OOM, inotify/fd exhaustion, relay watcher killed) has no re-arm trigger at all + // once the fast window gives up. Backoff keeps the recovery attempt without the 1s storm. + dormantRemoteWatchers: new Map< + string, + { delayMs: number; timer: ReturnType } + >(), + loggedUnavailableRemoteWatchers: new Set(), + pendingRemoteWatcherRetries: new Map>(), + pendingRemoteWatcherRetryListeners: new Map< + string, + { listeners: Map; startedAt: number; resyncOnInstall: boolean } + >(), + remoteWatcherResyncStates: new Map(), + // Why: last-listener cleanup aborts relay setup; late success is unwatched rather than installed after the renderer stopped watching. + 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>(), + // 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). + remoteWatcherLifecycleGeneration: 0, + unsubscribeFromProviderRegistrations: null as (() => void) | null +} diff --git a/src/main/ipc/filesystem-watcher-listener-lifecycle.ts b/src/main/ipc/filesystem-watcher-listener-lifecycle.ts new file mode 100644 index 00000000000..ce0cfaa80fc --- /dev/null +++ b/src/main/ipc/filesystem-watcher-listener-lifecycle.ts @@ -0,0 +1,272 @@ +import type { WebContents } from 'electron' +import { isWatcherProcessFailure } from './parcel-watcher-process-failure' +import type { WatchedRoot } from './filesystem-watcher-wsl' +import { + type LocalWatcherInstallToken, + type RemoteWatcherInstallToken, + UNWATCHABLE_ROOT_CACHE_MAX, + watcherLifecycleState +} from './filesystem-watcher-lifecycle-state' + +export function rememberUnwatchableRoot(rootKey: string): void { + const { unwatchableRoots } = watcherLifecycleState + // Why: deleted worktrees churn through unique paths; cap the set so retry suppression stays useful without retaining every failed path. + unwatchableRoots.delete(rootKey) + unwatchableRoots.add(rootKey) + while (unwatchableRoots.size > UNWATCHABLE_ROOT_CACHE_MAX) { + const oldest = unwatchableRoots.keys().next().value + if (oldest === undefined) { + break + } + unwatchableRoots.delete(oldest) + } +} + +export function retainLocalWatcherPhysicalFailure(rootKey: string, error: unknown): void { + if (!isWatcherProcessFailure(error) || !error.physicalExit) { + return + } + watcherLifecycleState.failedLocalUnsubscribes.set(rootKey, error) + void error.physicalExit.then(() => { + if (watcherLifecycleState.failedLocalUnsubscribes.get(rootKey) === error) { + watcherLifecycleState.failedLocalUnsubscribes.delete(rootKey) + } + }) +} + +export function trackDetachedLocalUnsubscribe(rootKey: string, root: WatchedRoot): Promise { + const rootUnsubscribes = + watcherLifecycleState.pendingLocalUnsubscribesByRoot.get(rootKey) ?? new Set>() + watcherLifecycleState.pendingLocalUnsubscribesByRoot.set(rootKey, rootUnsubscribes) + const unsubscribePromise = Promise.resolve() + .then(() => root.subscription.unsubscribe()) + .finally(() => { + watcherLifecycleState.pendingLocalUnsubscribes.delete(unsubscribePromise) + rootUnsubscribes.delete(unsubscribePromise) + if (rootUnsubscribes.size === 0) { + watcherLifecycleState.pendingLocalUnsubscribesByRoot.delete(rootKey) + } + }) + watcherLifecycleState.pendingLocalUnsubscribes.add(unsubscribePromise) + rootUnsubscribes.add(unsubscribePromise) + // Why: swallow here to avoid unhandled rejections, but keep the original promise rejected so later destructive cleanup can fail closed. + void unsubscribePromise.catch((error: unknown) => { + if (!watcherLifecycleState.abandonedLocalUnsubscribes.has(unsubscribePromise)) { + retainLocalWatcherPhysicalFailure(rootKey, error) + } + console.error(`[filesystem-watcher] unsubscribe error for ${rootKey}:`, error) + }) + return unsubscribePromise +} + +/** Mark unsubscribes whose drain timed out so their late failures stay out of failedLocalUnsubscribes. */ +export function abandonLocalUnsubscribes(promises: Iterable>): void { + for (const promise of promises) { + watcherLifecycleState.abandonedLocalUnsubscribes.add(promise) + } +} + +export function addInFlightLocalInstallListener( + token: LocalWatcherInstallToken, + sender: WebContents +): void { + if (sender.isDestroyed() || token.abortController.signal.aborted) { + return + } + token.listeners.set(sender.id, sender) + token.cancelled = false + registerWatcherSenderCleanup(sender) +} + +function cleanupInFlightLocalInstallsForSender(senderId: number): void { + for (const token of watcherLifecycleState.inFlightLocalInstalls.values()) { + token.listeners.delete(senderId) + if (token.listeners.size === 0) { + token.cancelled = true + // Why: abort so a pending native/forked subscription stops early (matches closeLocalWatcherForWorktreePath / closeAllWatchers). + token.abortController.abort() + } + } + for (const [rootKey, retry] of watcherLifecycleState.pendingLocalCapacityRetries) { + retry.listeners.delete(senderId) + if (retry.listeners.size === 0) { + retry.cancelWait() + watcherLifecycleState.pendingLocalCapacityRetries.delete(rootKey) + } + } +} + +export function takeLocalCapacityRetryListeners(rootKey: string): WebContents[] { + const retry = watcherLifecycleState.pendingLocalCapacityRetries.get(rootKey) + if (!retry) { + return [] + } + retry.cancelWait() + watcherLifecycleState.pendingLocalCapacityRetries.delete(rootKey) + return [...retry.listeners.values()].filter((listener) => !listener.isDestroyed()) +} + +export function clearLocalCapacityRetry(rootKey: string): void { + const retry = watcherLifecycleState.pendingLocalCapacityRetries.get(rootKey) + retry?.cancelWait() + watcherLifecycleState.pendingLocalCapacityRetries.delete(rootKey) +} + +export function addLocalWatchListener(rootKey: string, sender: WebContents): void { + const root = watcherLifecycleState.watchedRoots.get(rootKey) + if (!root || sender.isDestroyed()) { + return + } + root.listeners.set(sender.id, sender) + registerWatcherSenderCleanup(sender) +} + +export function cancelInFlightRemoteInstallIfUnowned(token: RemoteWatcherInstallToken): void { + token.cancelled = token.listeners.size === 0 + if (!token.cancelled || token.abortScheduled || token.abortController.signal.aborted) { + return + } + token.abortScheduled = true + // Why: a replacement sender can synchronously revive the shared install during a renderer handoff; otherwise stop the relay crawl next microtask. + queueMicrotask(() => { + token.abortScheduled = false + if (token.cancelled && token.listeners.size === 0) { + token.abortController.abort() + } + }) +} + +export function addInFlightRemoteInstallListener( + token: RemoteWatcherInstallToken, + sender: WebContents +): void { + if (sender.isDestroyed() || token.abortController.signal.aborted) { + return + } + token.listeners.set(sender.id, sender) + token.cancelled = false + registerWatcherSenderCleanup(sender) +} + +export function addRemoteWatchListener(key: string, sender: WebContents): void { + const state = watcherLifecycleState.remoteWatchers.get(key) + if (!state) { + return + } + state.listeners.set(sender.id, sender) + registerWatcherSenderCleanup(sender) +} + +export function releaseRemoteWatchListener(key: string, senderId: number): void { + const state = watcherLifecycleState.remoteWatchers.get(key) + if (!state) { + return + } + state.listeners.delete(senderId) + if (state.listeners.size > 0) { + return + } + state.unwatch() + state.batch.close() + watcherLifecycleState.remoteWatchers.delete(key) +} + +export function registerWatcherSenderCleanup(sender: WebContents): void { + if (watcherLifecycleState.senderCleanupRegistered.has(sender.id)) { + return + } + watcherLifecycleState.senderCleanupRegistered.add(sender.id) + sender.once('destroyed', () => { + watcherLifecycleState.senderCleanupRegistered.delete(sender.id) + cleanupLocalWatchersForSender(sender.id) + cleanupRemoteWatchersForSender(sender.id) + }) +} + +function cleanupLocalWatchersForSender(senderId: number): void { + for (const [rootKey, suspended] of watcherLifecycleState.suspendedLocalWatcherListeners) { + suspended.listeners.delete(senderId) + if (suspended.listeners.size === 0) { + watcherLifecycleState.suspendedLocalWatcherListeners.delete(rootKey) + } + } + cleanupInFlightLocalInstallsForSender(senderId) + for (const [key, watchedRoot] of watcherLifecycleState.watchedRoots) { + if (!watchedRoot.listeners.has(senderId)) { + continue + } + watchedRoot.listeners.delete(senderId) + if (watchedRoot.listeners.size === 0) { + // Cancel any pending grace-period teardown for this root. + const pending = watcherLifecycleState.pendingTeardowns.get(key) + if (pending) { + clearTimeout(pending) + watcherLifecycleState.pendingTeardowns.delete(key) + } + if (watchedRoot.batch.timer) { + clearTimeout(watchedRoot.batch.timer) + } + trackDetachedLocalUnsubscribe(key, watchedRoot) + watcherLifecycleState.watchedRoots.delete(key) + } + } +} + +function cleanupRemoteWatchersForSender(senderId: number): void { + for (const key of Array.from(watcherLifecycleState.desiredRemoteWatchers.keys())) { + forgetDesiredRemoteWatcher(key, senderId) + } + for (const [key, suspended] of watcherLifecycleState.suspendedRemoteWatcherListeners) { + suspended.listeners.delete(senderId) + if (suspended.listeners.size === 0) { + watcherLifecycleState.suspendedRemoteWatcherListeners.delete(key) + } + } + for (const token of watcherLifecycleState.inFlightRemoteInstalls.values()) { + token.listeners.delete(senderId) + cancelInFlightRemoteInstallIfUnowned(token) + } + for (const [key, retry] of watcherLifecycleState.pendingRemoteWatcherRetryListeners) { + retry.listeners.delete(senderId) + if (retry.listeners.size === 0) { + const timer = watcherLifecycleState.pendingRemoteWatcherRetries.get(key) + if (timer) { + clearTimeout(timer) + watcherLifecycleState.pendingRemoteWatcherRetries.delete(key) + } + watcherLifecycleState.pendingRemoteWatcherRetryListeners.delete(key) + } + } + for (const key of Array.from(watcherLifecycleState.remoteWatchers.keys())) { + releaseRemoteWatchListener(key, senderId) + } +} + +export function clearRemoteWatcherResync(key: string): void { + const resync = watcherLifecycleState.remoteWatcherResyncStates.get(key) + if (resync?.timer) { + clearTimeout(resync.timer) + } + watcherLifecycleState.remoteWatcherResyncStates.delete(key) +} + +export function clearDormantRemoteWatcher(key: string): void { + const dormant = watcherLifecycleState.dormantRemoteWatchers.get(key) + if (dormant) { + clearTimeout(dormant.timer) + watcherLifecycleState.dormantRemoteWatchers.delete(key) + } +} + +export function forgetDesiredRemoteWatcher(key: string, senderId: number): void { + const desired = watcherLifecycleState.desiredRemoteWatchers.get(key) + if (!desired) { + return + } + desired.listeners.delete(senderId) + if (desired.listeners.size === 0) { + watcherLifecycleState.desiredRemoteWatchers.delete(key) + clearRemoteWatcherResync(key) + clearDormantRemoteWatcher(key) + } +} diff --git a/src/main/ipc/filesystem-watcher-local-capacity.ts b/src/main/ipc/filesystem-watcher-local-capacity.ts new file mode 100644 index 00000000000..a66f7822b14 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-local-capacity.ts @@ -0,0 +1,59 @@ +import type { WebContents } from 'electron' +import { onWatcherChildCapacityAvailable } from './parcel-watcher-child-registry' +import type { LocalWatcherCapacityRetry } from './filesystem-watcher-lifecycle-state' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { + clearLocalCapacityRetry, + registerWatcherSenderCleanup +} from './filesystem-watcher-listener-lifecycle' +import { isWatcherRemovalInProgressError } from './watcher-removal-gate' + +export type SubscribeLocalWatcher = (worktreePath: string, sender: WebContents) => Promise + +export function scheduleLocalCapacityRetry( + rootKey: string, + worktreePath: string, + listeners: Map, + subscribe: SubscribeLocalWatcher +): void { + const existing = watcherLifecycleState.pendingLocalCapacityRetries.get(rootKey) + if (existing) { + for (const listener of listeners.values()) { + if (!listener.isDestroyed()) { + existing.listeners.set(listener.id, listener) + } + } + return + } + + let retry!: LocalWatcherCapacityRetry + const cancelWait = onWatcherChildCapacityAvailable(async () => { + if (watcherLifecycleState.pendingLocalCapacityRetries.get(rootKey) !== retry) { + return + } + watcherLifecycleState.pendingLocalCapacityRetries.delete(rootKey) + await Promise.all( + [...retry.listeners.values()].map(async (listener) => { + if (listener.isDestroyed()) { + return + } + await subscribe(worktreePath, listener).catch((error: unknown) => { + if (!isWatcherRemovalInProgressError(error)) { + console.error(`[filesystem-watcher] capacity retry failed for ${rootKey}:`, error) + } + }) + }) + ) + }) + retry = { listeners: new Map(), cancelWait } + watcherLifecycleState.pendingLocalCapacityRetries.set(rootKey, retry) + for (const listener of listeners.values()) { + if (!listener.isDestroyed()) { + retry.listeners.set(listener.id, listener) + registerWatcherSenderCleanup(listener) + } + } + if (retry.listeners.size === 0) { + clearLocalCapacityRetry(rootKey) + } +} diff --git a/src/main/ipc/filesystem-watcher-local-events.ts b/src/main/ipc/filesystem-watcher-local-events.ts new file mode 100644 index 00000000000..9618c6e74a7 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-local-events.ts @@ -0,0 +1,239 @@ +import { stat } from 'node:fs/promises' +import type { Event as WatcherEvent } from '@parcel/watcher' +import type { FsChangeEvent, FsChangedPayload } from '../../shared/filesystem-entry-types' +import { + WATCH_BATCH_MAX_WAIT_MS, + WATCH_BATCH_TRAILING_MS +} from '../../shared/filesystem-watch-batch-window' +import { MAX_BATCHED_WATCHER_EVENTS, queueWatcherEvents } from './filesystem-watcher-event-batch' +// Why: suppress high-churn dirs at the watcher level (separate from the File Explorer display filter, which only hides rows). +import { WATCHER_IGNORE_DIRS, buildParcelWatcherIgnoreOptions } from './filesystem-watcher-ignore' +import type { WatchedRoot } from './filesystem-watcher-wsl' +import { subscribeViaWatcherProcess } from './parcel-watcher-process' +import { normalizeWatcherEventPath } from './filesystem-watcher-paths' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { + retainLocalWatcherPhysicalFailure, + trackDetachedLocalUnsubscribe +} from './filesystem-watcher-listener-lifecycle' + +// ── Event coalescing ───────────────────────────────────────────────── +// Why: keep the last event per path in a flush window; delete→create emits both (delete cleans the subtree, create refreshes the parent), create→delete is dropped (§4.4). +function coalesceEvents( + raw: WatcherEvent[] +): { type: 'create' | 'update' | 'delete'; path: string }[] { + const lastByPath = new Map() + const deleteBeforeCreate = new Set() + + for (let i = 0; i < raw.length; i++) { + const evt = raw[i] + const p = normalizeWatcherEventPath(evt.path) + const prev = lastByPath.get(p) + + if (prev) { + // delete followed by create → emit both + if (prev.type === 'delete' && evt.type === 'create') { + deleteBeforeCreate.add(p) + } + // create followed by delete → net no-op, remove both + if (prev.type === 'create' && evt.type === 'delete') { + lastByPath.delete(p) + deleteBeforeCreate.delete(p) + continue + } + } + + lastByPath.set(p, { type: evt.type, index: i }) + + // Why: drop the stale delete when a later event supersedes delete→create, else output has a spurious delete for a file that still exists (§4.4). + if (evt.type !== 'create' && deleteBeforeCreate.has(p)) { + deleteBeforeCreate.delete(p) + } + } + + const result: { type: 'create' | 'update' | 'delete'; path: string }[] = [] + + // Emit delete events first for paths that have delete→create + for (const p of deleteBeforeCreate) { + result.push({ type: 'delete', path: p }) + } + + // Emit the last event for each path + for (const [p, entry] of lastByPath) { + result.push({ type: entry.type, path: p }) + } + + return result +} + +// ── Stat helper for isDirectory ────────────────────────────────────── + +async function tryStatIsDirectory(filePath: string): Promise { + try { + const s = await stat(filePath) + return s.isDirectory() + } catch { + // Why: stat failure (EPERM, vanished file) → undefined; renderer treats it as a file event, the safe default (§4.4). + return undefined + } +} + +// ── Flush and emit ─────────────────────────────────────────────────── + +function emitOverflowPayload(root: WatchedRoot): void { + const { rootPath } = root + const payload: FsChangedPayload = { + worktreePath: rootPath, + events: [{ kind: 'overflow', absolutePath: rootPath }] + } + for (const [, wc] of root.listeners) { + if (!wc.isDestroyed()) { + wc.send('fs:changed', payload) + } + } +} + +async function flushBatch(root: WatchedRoot): Promise { + const overflowed = root.batch.overflowed + const rawEvents = root.batch.events.splice(0) + root.batch.overflowed = false + root.batch.timer = null + root.batch.firstEventAt = 0 + + if ((rawEvents.length === 0 && !overflowed) || root.listeners.size === 0) { + return + } + + if (overflowed || rawEvents.length > MAX_BATCHED_WATCHER_EVENTS) { + // Why: deletion storms can be too large to coalesce/stat per path; one overflow asks the renderer for the same conservative refresh. + emitOverflowPayload(root) + return + } + + const coalesced = coalesceEvents(rawEvents) + + const events: FsChangeEvent[] = await Promise.all( + coalesced.map(async (evt) => { + // Why: a deleted path can't be stat'd; leave isDirectory undefined and let the renderer infer from dirCache. + const isDirectory = evt.type === 'delete' ? undefined : await tryStatIsDirectory(evt.path) + + return { + kind: evt.type, + absolutePath: evt.path, + isDirectory + } + }) + ) + + const payload: FsChangedPayload = { + worktreePath: root.rootPath, + events + } + + for (const [, wc] of root.listeners) { + if (!wc.isDestroyed()) { + wc.send('fs:changed', payload) + } + } +} + +export function scheduleLocalBatchFlush(root: WatchedRoot): void { + const now = Date.now() + + if (root.batch.firstEventAt === 0) { + root.batch.firstEventAt = now + } + + // If we've exceeded the max wait, flush immediately + if (now - root.batch.firstEventAt >= WATCH_BATCH_MAX_WAIT_MS) { + if (root.batch.timer) { + clearTimeout(root.batch.timer) + } + void flushBatch(root) + return + } + + // Trailing-edge debounce: reset timer on each new event + if (root.batch.timer) { + clearTimeout(root.batch.timer) + } + root.batch.timer = setTimeout(() => void flushBatch(root), WATCH_BATCH_TRAILING_MS) +} + +// ── Watcher creation ───────────────────────────────────────────────── + +export async function createLocalWatcher( + rootKey: string, + rootPath: string, + signal?: AbortSignal +): Promise { + const root: WatchedRoot = { + subscription: null!, + listeners: new Map(), + batch: { events: [], overflowed: false, timer: null, firstEventAt: 0 }, + rootPath + } + + try { + // Why: if the error callback cleaned up before subscribe() resolved, its returned subscription is orphaned and leaks a native handle. + let errorCleanedUp = false + + const watcherOptions = { + ...buildParcelWatcherIgnoreOptions(WATCHER_IGNORE_DIRS), + // Why: Parcel probes Watchman first, which prints a shell-level "watchman not recognized" error on Windows; pin the backend to suppress it. + ...(process.platform === 'win32' ? { backend: 'windows' as const } : {}) + } + + const markWatcherInterrupted = (): void => { + root.batch.overflowed = true + scheduleLocalBatchFlush(root) + } + + // Why: fork the watcher process (issue #7547 — watcher.node teardown races crash the host); onInterruption marks overflow to refresh past the gap. + root.subscription = await subscribeViaWatcherProcess( + rootPath, + (err, events) => { + if (err) { + // Why: treat watcher errors as overflow so the renderer conservatively refreshes rather than trusting possibly-invalid caches (§7.2, §7.3). + console.error(`[filesystem-watcher] error for ${rootKey}:`, err) + emitOverflowPayload(root) + // Why: after an error the native subscription may be invalid (deleted root); tear down the dead watcher so it doesn't dangle (§7.3). + if (root.batch.timer) { + clearTimeout(root.batch.timer) + } + // Why: error callback can fire before subscribe() assigns root.subscription; guard against null so cleanup doesn't crash. + if (root.subscription) { + retainLocalWatcherPhysicalFailure(rootKey, err) + void trackDetachedLocalUnsubscribe(rootKey, root) + } + errorCleanedUp = true + watcherLifecycleState.watchedRoots.delete(rootKey) + return + } + + queueWatcherEvents(root.batch, events) + scheduleLocalBatchFlush(root) + }, + watcherOptions, + { + delivery: { maxEventsPerBatch: MAX_BATCHED_WATCHER_EVENTS }, + // A child restart or bounded-queue overflow loses path precision; both need the same conservative renderer refresh. + onInterruption: markWatcherInterrupted, + onOverflow: markWatcherInterrupted, + signal + } + ) + + // Why: error callback already cleaned up watchedRoots before subscribe() resolved; unsubscribe this orphaned subscription so it doesn't leak. + if (errorCleanedUp) { + void trackDetachedLocalUnsubscribe(rootKey, root) + throw new Error(`Watcher for ${rootKey} errored during subscribe`) + } + } catch (err) { + // Why: watcher backend can throw synchronously on a deleted root/permission error; log rather than crash the main process (§7.3). + console.error(`[filesystem-watcher] failed to subscribe ${rootKey}:`, err) + throw err + } + + return root +} diff --git a/src/main/ipc/filesystem-watcher-local-install.ts b/src/main/ipc/filesystem-watcher-local-install.ts new file mode 100644 index 00000000000..089c553297b --- /dev/null +++ b/src/main/ipc/filesystem-watcher-local-install.ts @@ -0,0 +1,94 @@ +import { stat } from 'node:fs/promises' +import { isWslPath } from '../wsl' +import { WATCHER_IGNORE_DIRS } from './filesystem-watcher-ignore' +import { createWslWatcher, type WatchedRoot } from './filesystem-watcher-wsl' +import { WatcherChildCapacityError } from './parcel-watcher-child-registry' +import { isWatcherProcessFailure } from './parcel-watcher-process-failure' +import type { + LocalWatcherInstallResult, + LocalWatcherInstallToken +} from './filesystem-watcher-lifecycle-state' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { createLocalWatcher, scheduleLocalBatchFlush } from './filesystem-watcher-local-events' +import { + registerWatcherSenderCleanup, + rememberUnwatchableRoot, + retainLocalWatcherPhysicalFailure, + trackDetachedLocalUnsubscribe +} from './filesystem-watcher-listener-lifecycle' + +export async function installLocalWatcher( + rootKey: string, + rootPath: string, + worktreePath: string, + cancelToken: LocalWatcherInstallToken, + scheduleCapacityRetry: (listeners: Map) => void +): Promise { + let root: WatchedRoot + try { + const s = await stat(rootPath) + if (!s.isDirectory()) { + console.warn(`[filesystem-watcher] not a directory: ${rootKey}`) + rememberUnwatchableRoot(rootKey) + return 'unavailable' + } + } catch { + console.warn(`[filesystem-watcher] cannot stat root: ${rootKey}`) + rememberUnwatchableRoot(rootKey) + return 'unavailable' + } + + try { + // Why: WSL paths use one snapshot subprocess inside the distro so `wsl --shutdown` can kill it; native Windows uses @parcel/watcher. + root = isWslPath(worktreePath) + ? await createWslWatcher( + rootKey, + worktreePath, + { + ignoreDirs: WATCHER_IGNORE_DIRS, + scheduleBatchFlush: scheduleLocalBatchFlush, + watchedRoots: watcherLifecycleState.watchedRoots + }, + cancelToken.abortController.signal + ) + : await createLocalWatcher(rootKey, rootPath, cancelToken.abortController.signal) + } catch (error) { + // Why: setup can fail after its child misses the exit deadline; retain that owner even when the renderer-facing error is swallowed. + retainLocalWatcherPhysicalFailure(rootKey, error) + if (cancelToken.cancelled) { + if (isWatcherProcessFailure(error) && error.code === 'process_unavailable') { + throw error + } + return 'cancelled' + } + // Why: capacity is transient — allow retry once another child exits instead of caching this root as permanently failed. + if (error instanceof WatcherChildCapacityError) { + scheduleCapacityRetry(cancelToken.listeners) + return 'capacity' + } + rememberUnwatchableRoot(rootKey) + return 'unavailable' + } finally { + if (watcherLifecycleState.inFlightLocalInstalls.get(rootKey) === cancelToken) { + watcherLifecycleState.inFlightLocalInstalls.delete(rootKey) + } + } + + const liveListeners = new Map( + Array.from(cancelToken.listeners.entries()).filter(([, listener]) => !listener.isDestroyed()) + ) + if (cancelToken.cancelled || liveListeners.size === 0) { + if (root.batch.timer) { + clearTimeout(root.batch.timer) + } + void trackDetachedLocalUnsubscribe(rootKey, root) + return 'cancelled' + } + + root.listeners = liveListeners + watcherLifecycleState.watchedRoots.set(rootKey, root) + for (const listener of liveListeners.values()) { + registerWatcherSenderCleanup(listener) + } + return 'installed' +} diff --git a/src/main/ipc/filesystem-watcher-local-removal.ts b/src/main/ipc/filesystem-watcher-local-removal.ts new file mode 100644 index 00000000000..b0b92a9c118 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-local-removal.ts @@ -0,0 +1,156 @@ +import type { WebContents } from 'electron' +import type { FsChangedPayload } from '../../shared/filesystem-entry-types' +import { + createWatcherRemovalDeadline, + drainBeforeWatcherRemoval, + WATCHER_REMOVAL_FINAL_DRAIN_RESERVE_MS, + type WatcherRemovalDeadline +} from './watcher-removal-drain' +import { getLocalWatcherRoot } from './filesystem-watcher-paths' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { + abandonLocalUnsubscribes, + clearLocalCapacityRetry, + trackDetachedLocalUnsubscribe +} from './filesystem-watcher-listener-lifecycle' +import { subscribeLocalWatcher } from './filesystem-watcher-local-subscription' + +export async function closeLocalWatcherForWorktreePath( + worktreePath: string, + deadline: WatcherRemovalDeadline = createWatcherRemovalDeadline() +): Promise { + const { key: rootKey } = getLocalWatcherRoot(worktreePath) + const suspended = watcherLifecycleState.suspendedLocalWatcherListeners.get(rootKey) ?? { + worktreePath, + listeners: new Map() + } + for (const source of [ + watcherLifecycleState.pendingLocalCapacityRetries.get(rootKey)?.listeners, + watcherLifecycleState.inFlightLocalInstalls.get(rootKey)?.listeners, + watcherLifecycleState.watchedRoots.get(rootKey)?.listeners + ]) { + for (const [senderId, sender] of source ?? []) { + if (!sender.isDestroyed()) { + suspended.listeners.set(senderId, sender) + } + } + } + if (suspended.listeners.size > 0) { + watcherLifecycleState.suspendedLocalWatcherListeners.set(rootKey, suspended) + } + clearLocalCapacityRetry(rootKey) + const pendingTeardown = watcherLifecycleState.pendingTeardowns.get(rootKey) + if (pendingTeardown) { + clearTimeout(pendingTeardown) + watcherLifecycleState.pendingTeardowns.delete(rootKey) + } + + const inFlight = watcherLifecycleState.inFlightLocalInstalls.get(rootKey) + if (inFlight) { + // Why: Windows locks watched directories; deletion must cancel an in-flight subscription before Git removes the tree. + inFlight.listeners.clear() + inFlight.cancelled = true + inFlight.abortController.abort() + } + // Why: abort alone is not enough if the native subscribe never settles; bound so delete cannot hang the app. + const pendingInstall = watcherLifecycleState.pendingLocalInstallPromises.get(rootKey) + const installDrain = await drainBeforeWatcherRemoval( + pendingInstall, + deadline, + `local watcher install for ${rootKey}`, + { reserveMs: WATCHER_REMOVAL_FINAL_DRAIN_RESERVE_MS } + ) + if (installDrain === 'timeout') { + // Why: an abandoned install never runs its own cleanup, so leaving these entries would make every + // later watch of this root queue behind the same wedged promise. Identity-checked so a late settle + // can't evict a newer install. + if (watcherLifecycleState.pendingLocalInstallPromises.get(rootKey) === pendingInstall) { + watcherLifecycleState.pendingLocalInstallPromises.delete(rootKey) + } + if (inFlight && watcherLifecycleState.inFlightLocalInstalls.get(rootKey) === inFlight) { + watcherLifecycleState.inFlightLocalInstalls.delete(rootKey) + } + } + const pendingUnsubscribes = watcherLifecycleState.pendingLocalUnsubscribesByRoot.get(rootKey) + if (pendingUnsubscribes) { + const draining = Array.from(pendingUnsubscribes) + const unsubscribeDrain = await drainBeforeWatcherRemoval( + // Why the per-promise catch: an already-abandoned unsubscribe belongs to a delete that finished + // without it; re-raising its rejection here would fail a later close on stale news. + Promise.all( + draining.map((unsubscribe) => + watcherLifecycleState.abandonedLocalUnsubscribes.has(unsubscribe) + ? unsubscribe.catch(() => undefined) + : unsubscribe + ) + ), + deadline, + `local watcher unsubscribe for ${rootKey}`, + { reserveMs: WATCHER_REMOVAL_FINAL_DRAIN_RESERVE_MS } + ) + if (unsubscribeDrain === 'timeout') { + abandonLocalUnsubscribes(draining) + } + } + if (watcherLifecycleState.failedLocalUnsubscribes.has(rootKey)) { + throw watcherLifecycleState.failedLocalUnsubscribes.get(rootKey) + } + + const root = watcherLifecycleState.watchedRoots.get(rootKey) + if (!root) { + return + } + if (root.batch.timer) { + clearTimeout(root.batch.timer) + } + watcherLifecycleState.watchedRoots.delete(rootKey) + // Why: the in-process Parcel fallback has no unsubscribe timeout of its own, so an unbounded await + // here would hang delete forever and hold the removal gate. The promise stays tracked in + // pendingLocalUnsubscribesByRoot, so a later close still observes its failure. + const finalUnsubscribe = trackDetachedLocalUnsubscribe(rootKey, root) + const finalDrain = await drainBeforeWatcherRemoval( + finalUnsubscribe, + deadline, + `local watcher unsubscribe for ${rootKey}` + ) + if (finalDrain === 'timeout') { + abandonLocalUnsubscribes([finalUnsubscribe]) + } +} + +export async function restoreLocalWatcherAfterFailedRemoval(worktreePath: string): Promise { + const { key: rootKey } = getLocalWatcherRoot(worktreePath) + const suspended = watcherLifecycleState.suspendedLocalWatcherListeners.get(rootKey) + if (!suspended) { + return + } + watcherLifecycleState.suspendedLocalWatcherListeners.delete(rootKey) + const failures: unknown[] = [] + const failedListeners = new Map() + for (const sender of suspended.listeners.values()) { + if (sender.isDestroyed()) { + continue + } + try { + await subscribeLocalWatcher(suspended.worktreePath, sender) + sender.send('fs:changed', { + worktreePath: suspended.worktreePath, + events: [{ kind: 'overflow', absolutePath: suspended.worktreePath }] + } satisfies FsChangedPayload) + } catch (error) { + failures.push(error) + failedListeners.set(sender.id, sender) + } + } + if (failures.length > 0) { + watcherLifecycleState.suspendedLocalWatcherListeners.set(rootKey, { + worktreePath: suspended.worktreePath, + listeners: failedListeners + }) + throw failures[0] + } +} + +export function forgetLocalWatcherRemovalSnapshot(worktreePath: string): void { + watcherLifecycleState.suspendedLocalWatcherListeners.delete(getLocalWatcherRoot(worktreePath).key) +} diff --git a/src/main/ipc/filesystem-watcher-local-subscription.ts b/src/main/ipc/filesystem-watcher-local-subscription.ts new file mode 100644 index 00000000000..6d27f58f233 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-local-subscription.ts @@ -0,0 +1,221 @@ +import type { WebContents } from 'electron' +import { beginWatcherInstall } from './watcher-removal-gate' +import type { LocalWatcherInstallToken } from './filesystem-watcher-lifecycle-state' +import { + WATCHER_TEARDOWN_GRACE_MS, + watcherLifecycleState +} from './filesystem-watcher-lifecycle-state' +import { getLocalWatcherRoot } from './filesystem-watcher-paths' +import { + addInFlightLocalInstallListener, + addLocalWatchListener, + clearLocalCapacityRetry, + rememberUnwatchableRoot, + takeLocalCapacityRetryListeners, + trackDetachedLocalUnsubscribe +} from './filesystem-watcher-listener-lifecycle' +import { scheduleLocalCapacityRetry } from './filesystem-watcher-local-capacity' +import { installLocalWatcher } from './filesystem-watcher-local-install' + +// ── Subscribe / Unsubscribe ────────────────────────────────────────── + +export async function subscribeLocalWatcher( + worktreePath: string, + sender: WebContents, + generation = watcherLifecycleState.localWatcherLifecycleGeneration +): Promise { + if ( + watcherLifecycleState.localWatchersClosed || + generation !== watcherLifecycleState.localWatcherLifecycleGeneration + ) { + return + } + const finishInstall = beginWatcherInstall(worktreePath) + try { + await subscribeWhileRemovalAllowed(worktreePath, sender, generation) + } finally { + finishInstall() + } +} + +async function subscribeWhileRemovalAllowed( + worktreePath: string, + sender: WebContents, + generation: number +): Promise { + if ( + watcherLifecycleState.localWatchersClosed || + generation !== watcherLifecycleState.localWatcherLifecycleGeneration + ) { + return + } + const { key: rootKey, path: rootPath } = getLocalWatcherRoot(worktreePath) + if (sender.isDestroyed()) { + return + } + + // Don't retry roots that already failed — avoids repeated error spam. + if (watcherLifecycleState.unwatchableRoots.has(rootKey)) { + rememberUnwatchableRoot(rootKey) + return + } + + const root = watcherLifecycleState.watchedRoots.get(rootKey) + + // Cancel any pending grace-period teardown — a new listener arrived. + const pendingTeardown = watcherLifecycleState.pendingTeardowns.get(rootKey) + if (pendingTeardown) { + clearTimeout(pendingTeardown) + watcherLifecycleState.pendingTeardowns.delete(rootKey) + } + const capacityRetryListeners = takeLocalCapacityRetryListeners(rootKey) + + if (root) { + for (const listener of capacityRetryListeners) { + addLocalWatchListener(rootKey, listener) + } + addLocalWatchListener(rootKey, sender) + return + } + + const pendingInstall = watcherLifecycleState.pendingLocalInstallPromises.get(rootKey) + if (pendingInstall) { + const inFlight = watcherLifecycleState.inFlightLocalInstalls.get(rootKey) + const canJoinInstall = inFlight && !inFlight.abortController.signal.aborted + if (canJoinInstall) { + // Why: an unwatch may cancel an install while another renderer awaits the same root; a new live listener keeps it alive. + addInFlightLocalInstallListener(inFlight, sender) + for (const listener of capacityRetryListeners) { + addInFlightLocalInstallListener(inFlight, listener) + } + } + const result = await pendingInstall + if ( + result === 'cancelled' && + !canJoinInstall && + !watcherLifecycleState.localWatchersClosed && + generation === watcherLifecycleState.localWatcherLifecycleGeneration + ) { + // Why: AbortSignal can't be revived; listeners arriving after cancellation wait out that generation, then own a fresh install. + if (watcherLifecycleState.pendingLocalInstallPromises.get(rootKey) === pendingInstall) { + watcherLifecycleState.pendingLocalInstallPromises.delete(rootKey) + } + const retryListeners = new Map( + capacityRetryListeners.map((listener) => [listener.id, listener]) + ) + retryListeners.set(sender.id, sender) + for (const listener of retryListeners.values()) { + if (!listener.isDestroyed()) { + await subscribeWhileRemovalAllowed(worktreePath, listener, generation) + } + } + return + } + if (!inFlight) { + if (result === 'installed') { + for (const listener of capacityRetryListeners) { + addLocalWatchListener(rootKey, listener) + } + } else if (result === 'capacity') { + const retryListeners = new Map( + capacityRetryListeners.map((listener) => [listener.id, listener]) + ) + retryListeners.set(sender.id, sender) + scheduleLocalCapacityRetry(rootKey, worktreePath, retryListeners, subscribeLocalWatcher) + } + } + if ( + result === 'installed' && + watcherLifecycleState.watchedRoots.has(rootKey) && + !sender.isDestroyed() && + (!inFlight || inFlight.listeners.has(sender.id)) + ) { + addLocalWatchListener(rootKey, sender) + } + return + } + + const cancelToken: LocalWatcherInstallToken = { + cancelled: false, + listeners: new Map(), + abortController: new AbortController() + } + watcherLifecycleState.inFlightLocalInstalls.set(rootKey, cancelToken) + for (const listener of capacityRetryListeners) { + addInFlightLocalInstallListener(cancelToken, listener) + } + addInFlightLocalInstallListener(cancelToken, sender) + const installPromise = installLocalWatcher( + rootKey, + rootPath, + worktreePath, + cancelToken, + (listeners) => + scheduleLocalCapacityRetry(rootKey, worktreePath, listeners, subscribeLocalWatcher) + ) + watcherLifecycleState.pendingLocalInstallPromises.set(rootKey, installPromise) + try { + await installPromise + } finally { + if (watcherLifecycleState.pendingLocalInstallPromises.get(rootKey) === installPromise) { + watcherLifecycleState.pendingLocalInstallPromises.delete(rootKey) + } + } +} + +export function unsubscribeLocalWatcher(worktreePath: string, senderId: number): void { + const { key: rootKey } = getLocalWatcherRoot(worktreePath) + const suspended = watcherLifecycleState.suspendedLocalWatcherListeners.get(rootKey) + suspended?.listeners.delete(senderId) + if (suspended?.listeners.size === 0) { + watcherLifecycleState.suspendedLocalWatcherListeners.delete(rootKey) + } + const capacityRetry = watcherLifecycleState.pendingLocalCapacityRetries.get(rootKey) + if (capacityRetry) { + capacityRetry.listeners.delete(senderId) + if (capacityRetry.listeners.size === 0) { + clearLocalCapacityRetry(rootKey) + } + } + const inFlight = watcherLifecycleState.inFlightLocalInstalls.get(rootKey) + if (inFlight) { + inFlight.listeners.delete(senderId) + inFlight.cancelled = inFlight.listeners.size === 0 + // Why: last normal disconnect must abort the pending native/forked install (same early-cancel as closeLocalWatcherForWorktreePath). + if (inFlight.cancelled) { + inFlight.abortController.abort() + } + } + + const root = watcherLifecycleState.watchedRoots.get(rootKey) + if (!root) { + return + } + + root.listeners.delete(senderId) + + // Defer teardown when the last subscriber leaves so rapid worktree switches reuse the native watcher. + if (root.listeners.size === 0) { + if (root.batch.timer) { + clearTimeout(root.batch.timer) + } + + // Why: duplicate unwatch calls for a root would leak overwritten grace timers; keep just one. + if (watcherLifecycleState.pendingTeardowns.has(rootKey)) { + return + } + + const teardownTimer = setTimeout(() => { + watcherLifecycleState.pendingTeardowns.delete(rootKey) + // Re-check: a new listener may have arrived during the grace period. + const currentRoot = watcherLifecycleState.watchedRoots.get(rootKey) + if (!currentRoot || currentRoot.listeners.size > 0) { + return + } + void trackDetachedLocalUnsubscribe(rootKey, currentRoot) + watcherLifecycleState.watchedRoots.delete(rootKey) + }, WATCHER_TEARDOWN_GRACE_MS) + + watcherLifecycleState.pendingTeardowns.set(rootKey, teardownTimer) + } +} diff --git a/src/main/ipc/filesystem-watcher-paths.ts b/src/main/ipc/filesystem-watcher-paths.ts new file mode 100644 index 00000000000..44cb2cc00e6 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-paths.ts @@ -0,0 +1,39 @@ +import * as path from 'node:path' +import { + isWindowsAbsolutePathLike, + normalizeRuntimePathForComparison +} from '../../shared/cross-platform-path' + +// ── Path normalization ─────────────────────────────────────────────── + +function normalizeWatcherRootPath(rootPath: string): string { + let resolved = isWindowsAbsolutePathLike(rootPath) + ? path.win32.resolve(rootPath) + : path.resolve(rootPath) + // Why: Windows watcher events may use lowercase drive letters vs stored uppercase; normalize so renderer casing stays consistent (§4.4). + if (/^[a-zA-Z]:/.test(resolved)) { + resolved = resolved.charAt(0).toUpperCase() + resolved.slice(1) + } + return resolved +} + +export function getLocalWatcherRoot(rootPath: string): { key: string; path: string } { + const normalizedPath = normalizeWatcherRootPath(rootPath) + return { + // Why: Windows drive/UNC paths are case-insensitive; cleanup must match the owner even when Git returns a different spelling. + key: normalizeRuntimePathForComparison(normalizedPath), + path: normalizedPath + } +} + +export function normalizeWatcherEventPath(eventPath: string): string { + let resolved = path.resolve(eventPath) + if (/^[a-zA-Z]:/.test(resolved)) { + resolved = resolved.charAt(0).toUpperCase() + resolved.slice(1) + } + return resolved +} + +export function getRemoteWatcherKey(connectionId: string, worktreePath: string): string { + return JSON.stringify([connectionId, normalizeRuntimePathForComparison(worktreePath)]) +} diff --git a/src/main/ipc/filesystem-watcher-remote-controller.ts b/src/main/ipc/filesystem-watcher-remote-controller.ts new file mode 100644 index 00000000000..75bc4aa75d5 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-controller.ts @@ -0,0 +1,95 @@ +import type { WebContents } from 'electron' +import type { RemoteWatcherInstallToken } from './filesystem-watcher-lifecycle-state' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { + installRemoteWatcherCore, + type RemoteWatcherTerminalErrorHandler +} from './filesystem-watcher-remote-install' +import { requestRemoteWatcherResync } from './filesystem-watcher-remote-resync' +import { scheduleRemoteWatcherRetryCore } from './filesystem-watcher-remote-retry' +import { scheduleDormantRemoteWatcherRearmCore } from './filesystem-watcher-remote-dormant' +import { reinstallRemoteWatchersForConnectionCore } from './filesystem-watcher-remote-provider-rearm' + +export function installRemoteWatcher( + sender: WebContents, + connectionId: string, + worktreePath: string, + generation = watcherLifecycleState.remoteWatcherLifecycleGeneration +) { + return installRemoteWatcherCore( + sender, + connectionId, + worktreePath, + handleRemoteWatcherTerminalError, + generation + ) +} + +export function scheduleRemoteWatcherRetry( + sender: WebContents, + connectionId: string, + worktreePath: string, + startedAt = Date.now(), + resyncOnInstall = false +): void { + scheduleRemoteWatcherRetryCore( + sender, + connectionId, + worktreePath, + { + install: installRemoteWatcher, + requestResync: requestRemoteWatcherResync, + scheduleDormant: scheduleDormantRemoteWatcherRearm + }, + startedAt, + resyncOnInstall + ) +} + +export function scheduleDormantRemoteWatcherRearm( + connectionId: string, + worktreePath: string, + delayMs?: number +): void { + scheduleDormantRemoteWatcherRearmCore( + connectionId, + worktreePath, + { install: installRemoteWatcher, requestResync: requestRemoteWatcherResync }, + delayMs + ) +} + +export function reinstallRemoteWatchersForConnection(connectionId: string): void { + reinstallRemoteWatchersForConnectionCore(connectionId, { + install: installRemoteWatcher, + requestResync: requestRemoteWatcherResync, + scheduleRetry: scheduleRemoteWatcherRetry + }) +} + +const handleRemoteWatcherTerminalError: RemoteWatcherTerminalErrorHandler = ( + key: string, + connectionId: string, + worktreePath: string, + installToken: RemoteWatcherInstallToken, + error: Error +): void => { + installToken.terminalError = error + const state = watcherLifecycleState.remoteWatchers.get(key) + if (!state || state.installToken !== installToken) { + return + } + watcherLifecycleState.remoteWatchers.delete(key) + state.batch.close() + if ( + watcherLifecycleState.remoteWatchersClosed || + watcherLifecycleState.suspendedRemoteWatcherListeners.has(key) + ) { + return + } + console.warn(`[filesystem-watcher] SSH watcher terminated for ${key}:`, error) + const startedAt = Date.now() + for (const listener of state.listeners.values()) { + scheduleRemoteWatcherRetry(listener, connectionId, worktreePath, startedAt, true) + } +} diff --git a/src/main/ipc/filesystem-watcher-remote-desired.ts b/src/main/ipc/filesystem-watcher-remote-desired.ts new file mode 100644 index 00000000000..858e8e22027 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-desired.ts @@ -0,0 +1,29 @@ +import type { WebContents } from 'electron' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { getRemoteWatcherKey } from './filesystem-watcher-paths' +import { registerWatcherSenderCleanup } from './filesystem-watcher-listener-lifecycle' + +export function rememberDesiredRemoteWatcher( + connectionId: string, + worktreePath: string, + sender: WebContents +): void { + if (sender.isDestroyed()) { + return + } + const key = getRemoteWatcherKey(connectionId, worktreePath) + const desired = watcherLifecycleState.desiredRemoteWatchers.get(key) ?? { + connectionId, + worktreePath, + listeners: new Map() + } + desired.listeners.set(sender.id, sender) + watcherLifecycleState.desiredRemoteWatchers.set(key, desired) + registerWatcherSenderCleanup(sender) +} + +export function isCurrentDesiredRemoteWatcher(key: string, listener: WebContents): boolean { + return ( + watcherLifecycleState.desiredRemoteWatchers.get(key)?.listeners.get(listener.id) === listener + ) +} diff --git a/src/main/ipc/filesystem-watcher-remote-dormant.ts b/src/main/ipc/filesystem-watcher-remote-dormant.ts new file mode 100644 index 00000000000..741753b42e5 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-dormant.ts @@ -0,0 +1,113 @@ +import { getSshFilesystemProvider } from '../providers/ssh-filesystem-dispatch' +import { isWatcherRemovalInProgressError } from './watcher-removal-gate' +import type { + InstallRemoteWatcher, + RequestRemoteWatcherResync +} from './filesystem-watcher-remote-retry' +import { + REMOTE_WATCH_DORMANT_RETRY_MAX_MS, + REMOTE_WATCH_DORMANT_RETRY_MS, + watcherLifecycleState +} from './filesystem-watcher-lifecycle-state' +import { getRemoteWatcherKey } from './filesystem-watcher-paths' + +type ScheduleDormantDependencies = { + install: InstallRemoteWatcher + requestResync: RequestRemoteWatcherResync +} + +export function scheduleDormantRemoteWatcherRearmCore( + connectionId: string, + worktreePath: string, + dependencies: ScheduleDormantDependencies, + delayMs = REMOTE_WATCH_DORMANT_RETRY_MS +): void { + const key = getRemoteWatcherKey(connectionId, worktreePath) + if ( + watcherLifecycleState.remoteWatchersClosed || + !watcherLifecycleState.desiredRemoteWatchers.has(key) || + watcherLifecycleState.dormantRemoteWatchers.has(key) + ) { + return + } + const timer = setTimeout(() => { + watcherLifecycleState.dormantRemoteWatchers.delete(key) + void rearmDormantRemoteWatcher(key, connectionId, worktreePath, delayMs, dependencies) + }, delayMs) + // Why: a half-hour timer shouldn't be what keeps the process alive at quit. + timer.unref?.() + watcherLifecycleState.dormantRemoteWatchers.set(key, { delayMs, timer }) +} + +async function rearmDormantRemoteWatcher( + key: string, + connectionId: string, + worktreePath: string, + delayMs: number, + dependencies: ScheduleDormantDependencies +): Promise { + const desired = watcherLifecycleState.desiredRemoteWatchers.get(key) + if (watcherLifecycleState.remoteWatchersClosed || !desired) { + return + } + for (const [senderId, sender] of Array.from(desired.listeners)) { + if (sender.isDestroyed()) { + desired.listeners.delete(senderId) + } + } + if (desired.listeners.size === 0) { + watcherLifecycleState.desiredRemoteWatchers.delete(key) + return + } + // Why: a live watch or an in-flight fast retry already owns this key; installing again would + // clobber the entry the running watch reads its listeners from. + if ( + watcherLifecycleState.remoteWatchers.has(key) || + watcherLifecycleState.pendingRemoteWatcherRetries.has(key) + ) { + return + } + // Why: no provider means the connection itself is down, and its registration re-arms for free — + // polling would only add wire traffic to a link that is already being rebuilt. + if (!getSshFilesystemProvider(connectionId)) { + return + } + + const listeners = Array.from(desired.listeners.values()) + let results: Awaited>[] + try { + results = await Promise.all( + listeners.map((listener) => dependencies.install(listener, connectionId, worktreePath)) + ) + } catch (error) { + if (isWatcherRemovalInProgressError(error)) { + // Why: removal owns the key now and either forgets the intent or restores the watch itself. + return + } + scheduleDormantRemoteWatcherRearmCore( + connectionId, + worktreePath, + dependencies, + nextDormantDelayMs(delayMs) + ) + return + } + dependencies.requestResync( + key, + worktreePath, + listeners.filter((_, index) => results[index] === 'installed') + ) + // Why: 'cancelled' means shutdown or the last listener left, so only 'unavailable' stays dormant. + if (results.some((result) => result === 'unavailable')) { + scheduleDormantRemoteWatcherRearmCore( + connectionId, + worktreePath, + dependencies, + nextDormantDelayMs(delayMs) + ) + } +} + +function nextDormantDelayMs(delayMs: number): number { + return Math.min(delayMs * 2, REMOTE_WATCH_DORMANT_RETRY_MAX_MS) +} diff --git a/src/main/ipc/filesystem-watcher-remote-install.ts b/src/main/ipc/filesystem-watcher-remote-install.ts new file mode 100644 index 00000000000..c069c8dc16b --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-install.ts @@ -0,0 +1,239 @@ +import type { WebContents } from 'electron' +import type { FsChangedPayload } from '../../shared/filesystem-entry-types' +import { + WATCH_BATCH_MAX_WAIT_MS, + WATCH_BATCH_TRAILING_MS +} from '../../shared/filesystem-watch-batch-window' +import { getSshFilesystemProvider } from '../providers/ssh-filesystem-dispatch' +import { MAX_BATCHED_WATCHER_EVENTS } from './filesystem-watcher-event-batch' +import { createRemoteWatcherEventBatch } from './remote-watcher-event-batch' +import { beginWatcherInstall } from './watcher-removal-gate' +import type { + RemoteWatcherInstallResult, + RemoteWatcherInstallToken +} from './filesystem-watcher-lifecycle-state' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { getRemoteWatcherKey } from './filesystem-watcher-paths' +import { + addInFlightRemoteInstallListener, + addRemoteWatchListener, + registerWatcherSenderCleanup +} from './filesystem-watcher-listener-lifecycle' + +export type RemoteWatcherTerminalErrorHandler = ( + key: string, + connectionId: string, + worktreePath: string, + installToken: RemoteWatcherInstallToken, + error: Error +) => void + +export async function installRemoteWatcherCore( + sender: WebContents, + connectionId: string, + worktreePath: string, + onTerminalError: RemoteWatcherTerminalErrorHandler, + generation = watcherLifecycleState.remoteWatcherLifecycleGeneration +): Promise { + // Why: refuse installs racing in after teardown (or a waiter from an earlier lifecycle) so provider.watch() isn't called post-shutdown. + if ( + watcherLifecycleState.remoteWatchersClosed || + generation !== watcherLifecycleState.remoteWatcherLifecycleGeneration + ) { + return 'cancelled' + } + const finishInstall = beginWatcherInstall(worktreePath, connectionId) + try { + return await installRemoteWatcherWhileRemovalAllowed( + sender, + connectionId, + worktreePath, + onTerminalError, + generation + ) + } finally { + finishInstall() + } +} + +async function installRemoteWatcherWhileRemovalAllowed( + sender: WebContents, + connectionId: string, + worktreePath: string, + onTerminalError: RemoteWatcherTerminalErrorHandler, + generation: number +): Promise { + const provider = getSshFilesystemProvider(connectionId) + if (!provider || sender.isDestroyed()) { + return 'unavailable' + } + + const key = getRemoteWatcherKey(connectionId, worktreePath) + const existing = watcherLifecycleState.remoteWatchers.get(key) + if (existing) { + addRemoteWatchListener(key, sender) + return 'installed' + } + // Why: concurrent same-key watches must share the first provider.watch(); separate watchers would clobber per-key state and drop the unwatch handle. + const pendingInstall = watcherLifecycleState.pendingRemoteInstallPromises.get(key) + if (pendingInstall) { + const inFlight = watcherLifecycleState.inFlightRemoteInstalls.get(key) + const canJoinInstall = inFlight && !inFlight.abortController.signal.aborted + if (canJoinInstall) { + // Why: a new watcher joining before provider.watch() resolves should revive the install instead of inheriting the stale cancellation. + addInFlightRemoteInstallListener(inFlight, sender) + } + const result = await pendingInstall + if ( + result === 'installed' && + watcherLifecycleState.remoteWatchers.has(key) && + !sender.isDestroyed() && + (!inFlight || inFlight.listeners.has(sender.id)) + ) { + addRemoteWatchListener(key, sender) + } + if ( + result === 'cancelled' && + !canJoinInstall && + !sender.isDestroyed() && + generation === watcherLifecycleState.remoteWatcherLifecycleGeneration + ) { + // Why: AbortSignal can't be revived; a listener arriving after cancellation waits out that generation, then owns a fresh install. + if (watcherLifecycleState.pendingRemoteInstallPromises.get(key) === pendingInstall) { + watcherLifecycleState.pendingRemoteInstallPromises.delete(key) + } + return installRemoteWatcherCore( + sender, + connectionId, + worktreePath, + onTerminalError, + generation + ) + } + return result + } + const cancelToken: RemoteWatcherInstallToken = { + cancelled: false, + listeners: new Map(), + abortController: new AbortController(), + abortScheduled: false + } + watcherLifecycleState.inFlightRemoteInstalls.set(key, cancelToken) + addInFlightRemoteInstallListener(cancelToken, sender) + const installPromise = doInstallRemoteWatcher( + provider, + key, + connectionId, + worktreePath, + cancelToken, + onTerminalError + ) + watcherLifecycleState.pendingRemoteInstallPromises.set(key, installPromise) + try { + return await installPromise + } finally { + if (watcherLifecycleState.pendingRemoteInstallPromises.get(key) === installPromise) { + watcherLifecycleState.pendingRemoteInstallPromises.delete(key) + } + } +} + +async function doInstallRemoteWatcher( + provider: NonNullable>, + key: string, + connectionId: string, + worktreePath: string, + cancelToken: RemoteWatcherInstallToken, + onTerminalError: RemoteWatcherTerminalErrorHandler +): Promise { + let unwatch: () => void + // Why: the abandon paths below close the batch directly; once it is stored on the remoteWatchers + // entry, every later teardown closes it through that handle. deliver's state/installToken guard + // still covers a flush that races a close. + const batch = createRemoteWatcherEventBatch({ + rootPath: worktreePath, + trailingMs: WATCH_BATCH_TRAILING_MS, + maxWaitMs: WATCH_BATCH_MAX_WAIT_MS, + maxEvents: MAX_BATCHED_WATCHER_EVENTS, + deliver: (events) => { + const state = watcherLifecycleState.remoteWatchers.get(key) + // Why: buffering defers delivery past install, so a bare non-null check would let a stale flush + // land on a different install generation. + if (!state || state.installToken !== cancelToken) { + return + } + for (const listener of state.listeners.values()) { + if (listener.isDestroyed()) { + continue + } + try { + listener.send('fs:changed', { + worktreePath, + events + } satisfies FsChangedPayload) + } catch (err) { + // Why: batching moved this send out of the SSH mux's notification try/catch into a bare + // timer, so a frame disposed mid-window would escape as a fatal main-process exception. + console.warn( + `[filesystem-watch] failed to deliver remote fs:changed for ${worktreePath}: ${ + err instanceof Error ? err.message : String(err) + }` + ) + } + } + } + }) + try { + unwatch = await provider.watch( + worktreePath, + (events) => { + batch.push(events) + }, + { + signal: cancelToken.abortController.signal, + onTerminalError: (error) => + onTerminalError(key, connectionId, worktreePath, cancelToken, error) + } + ) + } catch (err) { + // Why: the provider can fire its callback before rejecting (an aborted reinstall), so this + // abandon path owes the same close as the two below — nothing else holds the batch. + batch.close() + if (cancelToken.cancelled || cancelToken.abortController.signal.aborted) { + return 'cancelled' + } + console.warn(`[filesystem-watcher] SSH watcher unavailable for ${key}:`, err) + return 'unavailable' + } finally { + if (watcherLifecycleState.inFlightRemoteInstalls.get(key) === cancelToken) { + watcherLifecycleState.inFlightRemoteInstalls.delete(key) + } + } + const liveListeners = new Map( + Array.from(cancelToken.listeners.entries()).filter(([, listener]) => !listener.isDestroyed()) + ) + if (cancelToken.cancelled || liveListeners.size === 0) { + batch.close() + try { + unwatch() + } catch (err) { + console.error(`[filesystem-watcher] remote unwatch (post-cancel) error for ${key}:`, err) + } + return 'cancelled' + } + if (cancelToken.terminalError) { + batch.close() + return 'unavailable' + } + watcherLifecycleState.remoteWatchers.set(key, { + unwatch, + listeners: liveListeners, + installToken: cancelToken, + batch + }) + for (const listener of liveListeners.values()) { + registerWatcherSenderCleanup(listener) + } + watcherLifecycleState.loggedUnavailableRemoteWatchers.delete(key) + return 'installed' +} diff --git a/src/main/ipc/filesystem-watcher-remote-provider-rearm.ts b/src/main/ipc/filesystem-watcher-remote-provider-rearm.ts new file mode 100644 index 00000000000..6b0ea28ef9b --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-provider-rearm.ts @@ -0,0 +1,120 @@ +import type { WebContents } from 'electron' +import { isWatcherRemovalInProgressError } from './watcher-removal-gate' +import type { + InstallRemoteWatcher, + RequestRemoteWatcherResync +} from './filesystem-watcher-remote-retry' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { clearDormantRemoteWatcher } from './filesystem-watcher-listener-lifecycle' + +type ScheduleRemoteWatcherRetry = ( + sender: WebContents, + connectionId: string, + worktreePath: string, + startedAt?: number, + resyncOnInstall?: boolean +) => void + +/** + * Rebuild remote watches for a connection whose filesystem provider was just (re)registered. + * + * Why: the relay's watch registrations die with the transport they were made on, and the previous + * provider's unwatch handle is scoped to that dead transport. Reinstalling is the only way the + * subscription comes back, and consumers get an overflow so they resync whatever changed while the + * watch was down. + */ +export function reinstallRemoteWatchersForConnectionCore( + connectionId: string, + dependencies: { + install: InstallRemoteWatcher + requestResync: RequestRemoteWatcherResync + scheduleRetry: ScheduleRemoteWatcherRetry + } +): void { + if (watcherLifecycleState.remoteWatchersClosed) { + return + } + for (const [key, desired] of Array.from(watcherLifecycleState.desiredRemoteWatchers)) { + if (desired.connectionId !== connectionId) { + continue + } + for (const [senderId, sender] of Array.from(desired.listeners)) { + if (sender.isDestroyed()) { + desired.listeners.delete(senderId) + } + } + if (desired.listeners.size === 0) { + watcherLifecycleState.desiredRemoteWatchers.delete(key) + continue + } + + // Why: drop the entry the dead transport left behind first — installRemoteWatcher treats an + // existing entry as already-installed and would hand back a watcher that can never fire again. + const stale = watcherLifecycleState.remoteWatchers.get(key) + if (stale) { + watcherLifecycleState.remoteWatchers.delete(key) + stale.batch.close() + try { + stale.unwatch() + } catch { + // Why: the handle belongs to the replaced transport; failing to close it is expected. + } + } + const retryTimer = watcherLifecycleState.pendingRemoteWatcherRetries.get(key) + if (retryTimer) { + clearTimeout(retryTimer) + watcherLifecycleState.pendingRemoteWatcherRetries.delete(key) + watcherLifecycleState.pendingRemoteWatcherRetryListeners.delete(key) + } + // Why: a pending watch belongs to the replaced transport; joiners must retry on the new provider. + const inFlight = watcherLifecycleState.inFlightRemoteInstalls.get(key) + if (inFlight) { + inFlight.listeners.clear() + inFlight.cancelled = true + inFlight.abortController.abort() + } + // Why: this reinstall supersedes the pending backoff; leaving it armed double-installs the key. + clearDormantRemoteWatcher(key) + watcherLifecycleState.loggedUnavailableRemoteWatchers.delete(key) + + const listeners = Array.from(desired.listeners.values()) + void Promise.all( + listeners.map((listener) => + dependencies.install(listener, desired.connectionId, desired.worktreePath) + ) + ) + .then((results) => { + // Why: events between the transport dropping and this reinstall are gone for good. + dependencies.requestResync( + key, + desired.worktreePath, + listeners.filter((_, index) => results[index] === 'installed') + ) + if (results.some((result) => result === 'unavailable')) { + for (const listener of listeners) { + dependencies.scheduleRetry( + listener, + desired.connectionId, + desired.worktreePath, + Date.now(), + true + ) + } + } + }) + .catch((error: unknown) => { + if (isWatcherRemovalInProgressError(error)) { + return + } + for (const listener of listeners) { + dependencies.scheduleRetry( + listener, + desired.connectionId, + desired.worktreePath, + Date.now(), + true + ) + } + }) + } +} diff --git a/src/main/ipc/filesystem-watcher-remote-removal.ts b/src/main/ipc/filesystem-watcher-remote-removal.ts new file mode 100644 index 00000000000..61424dc1db9 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-removal.ts @@ -0,0 +1,105 @@ +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 { getRemoteWatcherKey } from './filesystem-watcher-paths' +import { + clearDormantRemoteWatcher, + clearRemoteWatcherResync +} from './filesystem-watcher-listener-lifecycle' +import { + installRemoteWatcher, + scheduleRemoteWatcherRetry +} from './filesystem-watcher-remote-controller' + +export async function closeRemoteWatcherForWorktreePath( + connectionId: string, + worktreePath: string +): Promise { + const key = getRemoteWatcherKey(connectionId, worktreePath) + const suspended = watcherLifecycleState.suspendedRemoteWatcherListeners.get(key) ?? { + connectionId, + worktreePath, + listeners: new Map() + } + for (const source of [ + watcherLifecycleState.pendingRemoteWatcherRetryListeners.get(key)?.listeners, + watcherLifecycleState.inFlightRemoteInstalls.get(key)?.listeners, + watcherLifecycleState.remoteWatchers.get(key)?.listeners + ]) { + for (const [senderId, sender] of source ?? []) { + if (!sender.isDestroyed()) { + suspended.listeners.set(senderId, sender) + } + } + } + if (suspended.listeners.size > 0) { + watcherLifecycleState.suspendedRemoteWatcherListeners.set(key, suspended) + } + clearRemoteWatcherResync(key) + const retryTimer = watcherLifecycleState.pendingRemoteWatcherRetries.get(key) + if (retryTimer) { + clearTimeout(retryTimer) + watcherLifecycleState.pendingRemoteWatcherRetries.delete(key) + watcherLifecycleState.pendingRemoteWatcherRetryListeners.delete(key) + } + // Why: removal is deliberate — a backoff firing mid-removal would re-watch the path being deleted. + clearDormantRemoteWatcher(key) + const inFlight = watcherLifecycleState.inFlightRemoteInstalls.get(key) + if (inFlight) { + inFlight.listeners.clear() + inFlight.cancelled = true + } + const state = watcherLifecycleState.remoteWatchers.get(key) + const provider = getSshFilesystemProvider(connectionId) + await (provider?.closeWatch + ? provider.closeWatch(worktreePath) + : Promise.resolve(state?.unwatch())) + state?.batch.close() + watcherLifecycleState.remoteWatchers.delete(key) + watcherLifecycleState.loggedUnavailableRemoteWatchers.delete(key) +} + +export async function restoreRemoteWatcherAfterFailedRemoval( + connectionId: string, + worktreePath: string +): Promise { + const key = getRemoteWatcherKey(connectionId, worktreePath) + const suspended = watcherLifecycleState.suspendedRemoteWatcherListeners.get(key) + if (!suspended) { + return + } + watcherLifecycleState.suspendedRemoteWatcherListeners.delete(key) + for (const sender of suspended.listeners.values()) { + if (sender.isDestroyed()) { + continue + } + const result = await installRemoteWatcher(sender, connectionId, worktreePath) + if (result === 'unavailable') { + scheduleRemoteWatcherRetry(sender, connectionId, worktreePath) + } + sender.send('fs:changed', { + worktreePath, + events: [{ kind: 'overflow', absolutePath: worktreePath }] + } satisfies FsChangedPayload) + } +} + +export function forgetRemoteWatcherRemovalSnapshot( + connectionId: string, + worktreePath: string +): void { + const key = getRemoteWatcherKey(connectionId, worktreePath) + watcherLifecycleState.suspendedRemoteWatcherListeners.delete(key) + clearRemoteWatcherResync(key) + const retryTimer = watcherLifecycleState.pendingRemoteWatcherRetries.get(key) + if (retryTimer) { + clearTimeout(retryTimer) + watcherLifecycleState.pendingRemoteWatcherRetries.delete(key) + } + watcherLifecycleState.pendingRemoteWatcherRetryListeners.delete(key) + // Why: the worktree is gone — keeping the intent lets a reconnect landing before the renderer's + // unwatch re-watch a deleted path (60s of retries against the host, then a bogus overflow). + watcherLifecycleState.desiredRemoteWatchers.delete(key) + clearDormantRemoteWatcher(key) +} diff --git a/src/main/ipc/filesystem-watcher-remote-resync.ts b/src/main/ipc/filesystem-watcher-remote-resync.ts new file mode 100644 index 00000000000..918b16e6795 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-resync.ts @@ -0,0 +1,77 @@ +import type { WebContents } from 'electron' +import type { FsChangedPayload } from '../../shared/filesystem-entry-types' +import { + REMOTE_WATCH_RESYNC_COALESCE_MS, + watcherLifecycleState +} from './filesystem-watcher-lifecycle-state' +import { isCurrentDesiredRemoteWatcher } from './filesystem-watcher-remote-desired' + +function flushRemoteWatcherResync(key: string): void { + const state = watcherLifecycleState.remoteWatcherResyncStates.get(key) + if (!state) { + return + } + state.timer = undefined + if ( + watcherLifecycleState.remoteWatchersClosed || + watcherLifecycleState.suspendedRemoteWatcherListeners.has(key) || + !watcherLifecycleState.remoteWatchers.has(key) + ) { + if (!watcherLifecycleState.desiredRemoteWatchers.has(key)) { + watcherLifecycleState.remoteWatcherResyncStates.delete(key) + } + return + } + let sent = false + for (const listener of state.listeners.values()) { + if (listener.isDestroyed() || !isCurrentDesiredRemoteWatcher(key, listener)) { + continue + } + try { + listener.send('fs:changed', { + worktreePath: state.worktreePath, + events: [{ kind: 'overflow', absolutePath: state.worktreePath }] + } satisfies FsChangedPayload) + sent = true + } catch (error) { + console.warn(`[filesystem-watcher] failed to send SSH watcher resync for ${key}:`, error) + } + } + state.listeners.clear() + if (sent) { + state.lastSentAt = Date.now() + } else { + watcherLifecycleState.remoteWatcherResyncStates.delete(key) + } +} + +export function requestRemoteWatcherResync( + key: string, + worktreePath: string, + listeners: Iterable +): void { + const state = watcherLifecycleState.remoteWatcherResyncStates.get(key) ?? { + lastSentAt: Number.NEGATIVE_INFINITY, + listeners: new Map(), + worktreePath + } + state.worktreePath = worktreePath + for (const listener of listeners) { + if (!listener.isDestroyed() && isCurrentDesiredRemoteWatcher(key, listener)) { + state.listeners.set(listener.id, listener) + } + } + if (state.listeners.size === 0) { + return + } + watcherLifecycleState.remoteWatcherResyncStates.set(key, state) + const delayMs = Math.max(0, state.lastSentAt + REMOTE_WATCH_RESYNC_COALESCE_MS - Date.now()) + if (delayMs === 0) { + flushRemoteWatcherResync(key) + return + } + if (!state.timer) { + state.timer = setTimeout(() => flushRemoteWatcherResync(key), delayMs) + state.timer.unref?.() + } +} diff --git a/src/main/ipc/filesystem-watcher-remote-retry.ts b/src/main/ipc/filesystem-watcher-remote-retry.ts new file mode 100644 index 00000000000..0151ecf4419 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-retry.ts @@ -0,0 +1,134 @@ +import type { WebContents } from 'electron' +import type { FsChangedPayload } from '../../shared/filesystem-entry-types' +import { isWatcherRemovalInProgressError } from './watcher-removal-gate' +import type { RemoteWatcherInstallResult } from './filesystem-watcher-lifecycle-state' +import { + REMOTE_WATCH_RETRY_MS, + REMOTE_WATCH_RETRY_TIMEOUT_MS, + watcherLifecycleState +} from './filesystem-watcher-lifecycle-state' +import { getRemoteWatcherKey } from './filesystem-watcher-paths' +import { clearRemoteWatcherResync } from './filesystem-watcher-listener-lifecycle' +import { isCurrentDesiredRemoteWatcher } from './filesystem-watcher-remote-desired' + +export type InstallRemoteWatcher = ( + sender: WebContents, + connectionId: string, + worktreePath: string +) => Promise + +export type RequestRemoteWatcherResync = ( + key: string, + worktreePath: string, + listeners: Iterable +) => void + +type ScheduleDormantRemoteWatcher = ( + connectionId: string, + worktreePath: string, + delayMs?: number +) => void + +export function scheduleRemoteWatcherRetryCore( + sender: WebContents, + connectionId: string, + worktreePath: string, + dependencies: { + install: InstallRemoteWatcher + requestResync: RequestRemoteWatcherResync + scheduleDormant: ScheduleDormantRemoteWatcher + }, + startedAt = Date.now(), + // Why: a retry that replaces a watch which was already live owes the renderer an overflow once it + // lands — the events lost while it was down are otherwise never signalled. + resyncOnInstall = false +): void { + const key = getRemoteWatcherKey(connectionId, worktreePath) + const existingRetry = watcherLifecycleState.pendingRemoteWatcherRetryListeners.get(key) + if (existingRetry) { + if (!sender.isDestroyed()) { + existingRetry.listeners.set(sender.id, sender) + } + existingRetry.resyncOnInstall ||= resyncOnInstall + return + } + + const retry = { + listeners: new Map(sender.isDestroyed() ? [] : [[sender.id, sender]]), + startedAt, + resyncOnInstall + } + watcherLifecycleState.pendingRemoteWatcherRetryListeners.set(key, retry) + + if (Date.now() - startedAt >= REMOTE_WATCH_RETRY_TIMEOUT_MS || sender.isDestroyed()) { + watcherLifecycleState.pendingRemoteWatcherRetries.delete(key) + watcherLifecycleState.pendingRemoteWatcherRetryListeners.delete(key) + watcherLifecycleState.loggedUnavailableRemoteWatchers.delete(key) + clearRemoteWatcherResync(key) + // Why: handler already resolved so the renderer thinks the watch is live; emit overflow to force a manual refresh instead of waiting forever. + for (const listener of retry.listeners.values()) { + if (listener.isDestroyed() || !isCurrentDesiredRemoteWatcher(key, listener)) { + continue + } + console.warn( + `[filesystem-watcher] giving up SSH watch retry for ${worktreePath} on connection ${connectionId} after ${REMOTE_WATCH_RETRY_TIMEOUT_MS}ms` + ) + listener.send('fs:changed', { + worktreePath, + events: [{ kind: 'overflow', absolutePath: worktreePath }] + } satisfies FsChangedPayload) + } + // Why: overflow only refreshes once — without this the watch stays dead until the app restarts. + dependencies.scheduleDormant(connectionId, worktreePath) + return + } + + const retryTimer = setTimeout(() => { + watcherLifecycleState.pendingRemoteWatcherRetries.delete(key) + watcherLifecycleState.pendingRemoteWatcherRetryListeners.delete(key) + const listeners = Array.from(retry.listeners.values()).filter( + (listener) => !listener.isDestroyed() && isCurrentDesiredRemoteWatcher(key, listener) + ) + void Promise.all( + listeners.map((listener) => dependencies.install(listener, connectionId, worktreePath)) + ) + .then((results) => { + if (retry.resyncOnInstall) { + dependencies.requestResync( + key, + worktreePath, + listeners.filter((_, index) => results[index] === 'installed') + ) + } + // Why: don't re-arm on 'cancelled' (renderer stopped watching) — it would fire a stale overflow when the 60s window expires. + if (results.some((result) => result === 'unavailable')) { + for (const listener of listeners) { + scheduleRemoteWatcherRetryCore( + listener, + connectionId, + worktreePath, + dependencies, + retry.startedAt, + retry.resyncOnInstall + ) + } + } + }) + .catch((error: unknown) => { + if (isWatcherRemovalInProgressError(error)) { + return + } + for (const listener of listeners) { + scheduleRemoteWatcherRetryCore( + listener, + connectionId, + worktreePath, + dependencies, + retry.startedAt, + retry.resyncOnInstall + ) + } + }) + }, REMOTE_WATCH_RETRY_MS) + watcherLifecycleState.pendingRemoteWatcherRetries.set(key, retryTimer) +} diff --git a/src/main/ipc/filesystem-watcher-shutdown.ts b/src/main/ipc/filesystem-watcher-shutdown.ts new file mode 100644 index 00000000000..96018b57512 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-shutdown.ts @@ -0,0 +1,81 @@ +import { disposeWatcherProcess } from './parcel-watcher-process' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { trackDetachedLocalUnsubscribe } from './filesystem-watcher-listener-lifecycle' + +/** Tear down all watchers on app shutdown. */ +export async function closeAllWatchers(): Promise { + // Why: drop the intent with the rest of the state, but keep the provider-registration + // subscription — a new fs:watchWorktree reopens the subsystem and still needs the re-arm hook. + watcherLifecycleState.desiredRemoteWatchers.clear() + watcherLifecycleState.senderCleanupRegistered.clear() + watcherLifecycleState.unwatchableRoots.clear() + watcherLifecycleState.suspendedLocalWatcherListeners.clear() + watcherLifecycleState.suspendedRemoteWatcherListeners.clear() + for (const retry of watcherLifecycleState.pendingLocalCapacityRetries.values()) { + retry.cancelWait() + } + watcherLifecycleState.pendingLocalCapacityRetries.clear() + + // Cancel any pending grace-period teardowns — we're tearing down everything. + for (const timer of watcherLifecycleState.pendingTeardowns.values()) { + clearTimeout(timer) + } + watcherLifecycleState.pendingTeardowns.clear() + + for (const timer of watcherLifecycleState.pendingRemoteWatcherRetries.values()) { + clearTimeout(timer) + } + watcherLifecycleState.pendingRemoteWatcherRetries.clear() + watcherLifecycleState.pendingRemoteWatcherRetryListeners.clear() + for (const state of watcherLifecycleState.remoteWatcherResyncStates.values()) { + if (state.timer) { + clearTimeout(state.timer) + } + } + watcherLifecycleState.remoteWatcherResyncStates.clear() + for (const dormant of watcherLifecycleState.dormantRemoteWatchers.values()) { + clearTimeout(dormant.timer) + } + watcherLifecycleState.dormantRemoteWatchers.clear() + watcherLifecycleState.loggedUnavailableRemoteWatchers.clear() + // Why: latch both subsystems shut so late installs can't register; generation bumps reject older-lifecycle waiters. + watcherLifecycleState.remoteWatchersClosed = true + watcherLifecycleState.remoteWatcherLifecycleGeneration += 1 + watcherLifecycleState.localWatchersClosed = true + watcherLifecycleState.localWatcherLifecycleGeneration += 1 + watcherLifecycleState.pendingRemoteInstallPromises.clear() + // Why: cancel in-flight provider.watch() calls so their resolved unwatch handles aren't installed post-shutdown. + for (const token of watcherLifecycleState.inFlightRemoteInstalls.values()) { + token.listeners.clear() + token.cancelled = true + token.abortController.abort() + } + for (const token of watcherLifecycleState.inFlightLocalInstalls.values()) { + token.listeners.clear() + token.cancelled = true + token.abortController.abort() + } + + for (const [rootKey, root] of watcherLifecycleState.watchedRoots) { + if (root.batch.timer) { + clearTimeout(root.batch.timer) + } + await trackDetachedLocalUnsubscribe(rootKey, root).catch(() => undefined) + } + watcherLifecycleState.watchedRoots.clear() + await Promise.allSettled(Array.from(watcherLifecycleState.pendingLocalUnsubscribes)) + watcherLifecycleState.failedLocalUnsubscribes.clear() + // Why: kill the forked watcher process instead of watcher.node's crash-prone async teardown; process death frees native handles. + disposeWatcherProcess() + + // Why: remote watchers are separate from local @parcel/watcher subs; unwatch here or the relay keeps polling FS after shutdown. + for (const [key, state] of watcherLifecycleState.remoteWatchers) { + state.batch.close() + try { + state.unwatch() + } catch (err) { + console.error(`[filesystem-watcher] remote unwatch error for ${key}:`, err) + } + } + watcherLifecycleState.remoteWatchers.clear() +} diff --git a/src/main/ipc/filesystem-watcher.ts b/src/main/ipc/filesystem-watcher.ts index 71830efc7c2..9680227cbc1 100644 --- a/src/main/ipc/filesystem-watcher.ts +++ b/src/main/ipc/filesystem-watcher.ts @@ -1,1966 +1,26 @@ -/* eslint-disable max-lines -- Why: one module keeps native (@parcel/watcher), WSL-snapshot, and SSH watcher lifecycle invariants and shared debounce/coalesce helpers auditable in one file. */ -import { ipcMain, type WebContents } from 'electron' -import * as path from 'node:path' -import { stat } from 'node:fs/promises' -import type { Event as WatcherEvent } from '@parcel/watcher' -import type { FsChangeEvent, FsChangedPayload } from '../../shared/filesystem-entry-types' -import { - isWindowsAbsolutePathLike, - normalizeRuntimePathForComparison -} from '../../shared/cross-platform-path' -import { isWslPath } from '../wsl' -import { createWslWatcher } from './filesystem-watcher-wsl' -import type { WatchedRoot } from './filesystem-watcher-wsl' -import { - getSshFilesystemProvider, - onSshFilesystemProviderRegistered -} from '../providers/ssh-filesystem-dispatch' -import { MAX_BATCHED_WATCHER_EVENTS, queueWatcherEvents } from './filesystem-watcher-event-batch' -import { - createRemoteWatcherEventBatch, - type RemoteWatcherEventBatch -} from './remote-watcher-event-batch' -import { - WATCH_BATCH_MAX_WAIT_MS, - WATCH_BATCH_TRAILING_MS -} from '../../shared/filesystem-watch-batch-window' -import { disposeWatcherProcess, subscribeViaWatcherProcess } from './parcel-watcher-process' -import { isWatcherProcessFailure } from './parcel-watcher-process-failure' -import { - onWatcherChildCapacityAvailable, - WatcherChildCapacityError -} from './parcel-watcher-child-registry' -import { beginWatcherInstall, isWatcherRemovalInProgressError } from './watcher-removal-gate' -import { - createWatcherRemovalDeadline, - drainBeforeWatcherRemoval, - WATCHER_REMOVAL_FINAL_DRAIN_RESERVE_MS, - type WatcherRemovalDeadline -} from './watcher-removal-drain' -// Why: suppress high-churn dirs at the watcher level (separate from the File Explorer display filter, which only hides rows). -import { WATCHER_IGNORE_DIRS, buildParcelWatcherIgnoreOptions } from './filesystem-watcher-ignore' import type { WorktreeWatcherRemoval } from './worktree-watcher-removal' - -// ── Per-root watcher state ─────────────────────────────────────────── -// WatchedRoot/WatcherSubscription live in filesystem-watcher-wsl.ts so native and WSL watchers share one shape. - -// ── Module state ───────────────────────────────────────────────────── - -const watchedRoots = new Map() - -// Why: cache roots that failed watcher creation (e.g. WSL UNC paths) so we don't retry every worktree switch and spam the console with errors. -const UNWATCHABLE_ROOT_CACHE_MAX = 256 -const unwatchableRoots = new Set() - -function rememberUnwatchableRoot(rootKey: string): void { - // Why: deleted worktrees churn through unique paths; cap the set so retry suppression stays useful without retaining every failed path. - unwatchableRoots.delete(rootKey) - unwatchableRoots.add(rootKey) - while (unwatchableRoots.size > UNWATCHABLE_ROOT_CACHE_MAX) { - const oldest = unwatchableRoots.keys().next().value - if (oldest === undefined) { - break - } - unwatchableRoots.delete(oldest) - } -} - -// Why: key cleanup by sender WebContents (not per root) to avoid MaxListeners warnings when a workspace has many worktrees open. -const senderCleanupRegistered = new Set() - -// Why: recreating @parcel/watcher on Windows is expensive (ReadDirectoryChangesW + AV, 500ms+); a 30s grace lets rapid switches reuse the watcher. -const WATCHER_TEARDOWN_GRACE_MS = 30_000 -const pendingTeardowns = new Map>() -// Why: @parcel/watcher unsubscribe does native async work that sender-destroy can start before shutdown, so will-quit must still await it. -const pendingLocalUnsubscribes = new Set>() -const pendingLocalUnsubscribesByRoot = new Map>>() -const suspendedLocalWatcherListeners = new Map< - string, - { worktreePath: string; listeners: Map } ->() -// Why: an install cancelled by shutdown can't be revived by a waiter that resumes after a later call reopens the subsystem. -let localWatchersClosed = false -let localWatcherLifecycleGeneration = 0 -const failedLocalUnsubscribes = new Map() -// Why: a drain that timed out no longer gates the delete — Git removal already proceeded past it. Its -// late failure must not fail-close a *later* close of the same root, which would leave that path -// undeletable until the watcher process physically exits. -const abandonedLocalUnsubscribes = new WeakSet>() -type LocalWatcherInstallToken = { - cancelled: boolean - listeners: Map - abortController: AbortController -} -type LocalWatcherInstallResult = 'installed' | 'unavailable' | 'capacity' | 'cancelled' -type LocalWatcherCapacityRetry = { - listeners: Map - cancelWait: () => void -} -// Why: watcher creation is async; concurrent watch requests for the same root must share one install or later resolves orphan listeners. -const inFlightLocalInstalls = new Map() -const pendingLocalInstallPromises = new Map>() -const pendingLocalCapacityRetries = new Map() - -function addInFlightLocalInstallListener( - token: LocalWatcherInstallToken, - sender: WebContents -): void { - if (sender.isDestroyed() || token.abortController.signal.aborted) { - return - } - token.listeners.set(sender.id, sender) - token.cancelled = false - registerSenderCleanup(sender) -} - -function cleanupInFlightLocalInstallsForSender(senderId: number): void { - for (const token of inFlightLocalInstalls.values()) { - token.listeners.delete(senderId) - if (token.listeners.size === 0) { - token.cancelled = true - // Why: abort so a pending native/forked subscription stops early (matches closeLocalWatcherForWorktreePath / closeAllWatchers). - token.abortController.abort() - } - } - for (const [rootKey, retry] of pendingLocalCapacityRetries) { - retry.listeners.delete(senderId) - if (retry.listeners.size === 0) { - retry.cancelWait() - pendingLocalCapacityRetries.delete(rootKey) - } - } -} - -function takeLocalCapacityRetryListeners(rootKey: string): WebContents[] { - const retry = pendingLocalCapacityRetries.get(rootKey) - if (!retry) { - return [] - } - retry.cancelWait() - pendingLocalCapacityRetries.delete(rootKey) - return [...retry.listeners.values()].filter((listener) => !listener.isDestroyed()) -} - -function clearLocalCapacityRetry(rootKey: string): void { - const retry = pendingLocalCapacityRetries.get(rootKey) - retry?.cancelWait() - pendingLocalCapacityRetries.delete(rootKey) -} - -function scheduleLocalCapacityRetry( - rootKey: string, - worktreePath: string, - listeners: Map -): void { - const existing = pendingLocalCapacityRetries.get(rootKey) - if (existing) { - for (const listener of listeners.values()) { - if (!listener.isDestroyed()) { - existing.listeners.set(listener.id, listener) - } - } - return - } - - let retry!: LocalWatcherCapacityRetry - const cancelWait = onWatcherChildCapacityAvailable(async () => { - if (pendingLocalCapacityRetries.get(rootKey) !== retry) { - return - } - pendingLocalCapacityRetries.delete(rootKey) - await Promise.all( - [...retry.listeners.values()].map(async (listener) => { - if (listener.isDestroyed()) { - return - } - await subscribe(worktreePath, listener).catch((error: unknown) => { - if (!isWatcherRemovalInProgressError(error)) { - console.error(`[filesystem-watcher] capacity retry failed for ${rootKey}:`, error) - } - }) - }) - ) - }) - retry = { listeners: new Map(), cancelWait } - pendingLocalCapacityRetries.set(rootKey, retry) - for (const listener of listeners.values()) { - if (!listener.isDestroyed()) { - retry.listeners.set(listener.id, listener) - registerSenderCleanup(listener) - } - } - if (retry.listeners.size === 0) { - clearLocalCapacityRetry(rootKey) - } -} - -// ── Path normalization ─────────────────────────────────────────────── - -function normalizeRootPath(rootPath: string): string { - let resolved = isWindowsAbsolutePathLike(rootPath) - ? path.win32.resolve(rootPath) - : path.resolve(rootPath) - // Why: Windows watcher events may use lowercase drive letters vs stored uppercase; normalize so renderer casing stays consistent (§4.4). - if (/^[a-zA-Z]:/.test(resolved)) { - resolved = resolved.charAt(0).toUpperCase() + resolved.slice(1) - } - return resolved -} - -function localWatcherRoot(rootPath: string): { key: string; path: string } { - const normalizedPath = normalizeRootPath(rootPath) - return { - // Why: Windows drive/UNC paths are case-insensitive; cleanup must match the owner even when Git returns a different spelling. - key: normalizeRuntimePathForComparison(normalizedPath), - path: normalizedPath - } -} - -function normalizeEventPath(eventPath: string): string { - let resolved = path.resolve(eventPath) - if (/^[a-zA-Z]:/.test(resolved)) { - resolved = resolved.charAt(0).toUpperCase() + resolved.slice(1) - } - return resolved -} - -// ── Event coalescing ───────────────────────────────────────────────── -// Why: keep the last event per path in a flush window; delete→create emits both (delete cleans the subtree, create refreshes the parent), create→delete is dropped (§4.4). - -function coalesceEvents( - raw: WatcherEvent[] -): { type: 'create' | 'update' | 'delete'; path: string }[] { - const lastByPath = new Map() - const deleteBeforeCreate = new Set() - - for (let i = 0; i < raw.length; i++) { - const evt = raw[i] - const p = normalizeEventPath(evt.path) - const prev = lastByPath.get(p) - - if (prev) { - // delete followed by create → emit both - if (prev.type === 'delete' && evt.type === 'create') { - deleteBeforeCreate.add(p) - } - // create followed by delete → net no-op, remove both - if (prev.type === 'create' && evt.type === 'delete') { - lastByPath.delete(p) - deleteBeforeCreate.delete(p) - continue - } - } - - lastByPath.set(p, { type: evt.type, index: i }) - - // Why: drop the stale delete when a later event supersedes delete→create, else output has a spurious delete for a file that still exists (§4.4). - if (evt.type !== 'create' && deleteBeforeCreate.has(p)) { - deleteBeforeCreate.delete(p) - } - } - - const result: { type: 'create' | 'update' | 'delete'; path: string }[] = [] - - // Emit delete events first for paths that have delete→create - for (const p of deleteBeforeCreate) { - result.push({ type: 'delete', path: p }) - } - - // Emit the last event for each path - for (const [p, entry] of lastByPath) { - result.push({ type: entry.type, path: p }) - } - - return result -} - -// ── Stat helper for isDirectory ────────────────────────────────────── - -async function tryStatIsDirectory(filePath: string): Promise { - try { - const s = await stat(filePath) - return s.isDirectory() - } catch { - // Why: stat failure (EPERM, vanished file) → undefined; renderer treats it as a file event, the safe default (§4.4). - return undefined - } -} - -// ── Flush and emit ─────────────────────────────────────────────────── - -function emitOverflowPayload(root: WatchedRoot): void { - const { rootPath } = root - const payload: FsChangedPayload = { - worktreePath: rootPath, - events: [{ kind: 'overflow', absolutePath: rootPath }] - } - for (const [, wc] of root.listeners) { - if (!wc.isDestroyed()) { - wc.send('fs:changed', payload) - } - } -} - -async function flushBatch(root: WatchedRoot): Promise { - const overflowed = root.batch.overflowed - const rawEvents = root.batch.events.splice(0) - root.batch.overflowed = false - root.batch.timer = null - root.batch.firstEventAt = 0 - - if ((rawEvents.length === 0 && !overflowed) || root.listeners.size === 0) { - return - } - - if (overflowed || rawEvents.length > MAX_BATCHED_WATCHER_EVENTS) { - // Why: deletion storms can be too large to coalesce/stat per path; one overflow asks the renderer for the same conservative refresh. - emitOverflowPayload(root) - return - } - - const coalesced = coalesceEvents(rawEvents) - - const events: FsChangeEvent[] = await Promise.all( - coalesced.map(async (evt) => { - // Why: a deleted path can't be stat'd; leave isDirectory undefined and let the renderer infer from dirCache. - const isDirectory = evt.type === 'delete' ? undefined : await tryStatIsDirectory(evt.path) - - return { - kind: evt.type, - absolutePath: evt.path, - isDirectory - } - }) - ) - - const payload: FsChangedPayload = { - worktreePath: root.rootPath, - events - } - - for (const [, wc] of root.listeners) { - if (!wc.isDestroyed()) { - wc.send('fs:changed', payload) - } - } -} - -function scheduleBatchFlush(root: WatchedRoot): void { - const now = Date.now() - - if (root.batch.firstEventAt === 0) { - root.batch.firstEventAt = now - } - - // If we've exceeded the max wait, flush immediately - if (now - root.batch.firstEventAt >= WATCH_BATCH_MAX_WAIT_MS) { - if (root.batch.timer) { - clearTimeout(root.batch.timer) - } - void flushBatch(root) - return - } - - // Trailing-edge debounce: reset timer on each new event - if (root.batch.timer) { - clearTimeout(root.batch.timer) - } - root.batch.timer = setTimeout(() => void flushBatch(root), WATCH_BATCH_TRAILING_MS) -} - -// ── Watcher creation ───────────────────────────────────────────────── - -async function createWatcher( - rootKey: string, - rootPath: string, - signal?: AbortSignal -): Promise { - const root: WatchedRoot = { - subscription: null!, - listeners: new Map(), - batch: { events: [], overflowed: false, timer: null, firstEventAt: 0 }, - rootPath - } - - try { - // Why: if the error callback cleaned up before subscribe() resolved, its returned subscription is orphaned and leaks a native handle. - let errorCleanedUp = false - - const watcherOptions = { - ...buildParcelWatcherIgnoreOptions(WATCHER_IGNORE_DIRS), - // Why: Parcel probes Watchman first, which prints a shell-level "watchman not recognized" error on Windows; pin the backend to suppress it. - ...(process.platform === 'win32' ? { backend: 'windows' as const } : {}) - } - - const markWatcherInterrupted = (): void => { - root.batch.overflowed = true - scheduleBatchFlush(root) - } - - // Why: fork the watcher process (issue #7547 — watcher.node teardown races crash the host); onInterruption marks overflow to refresh past the gap. - root.subscription = await subscribeViaWatcherProcess( - rootPath, - (err, events) => { - if (err) { - // Why: treat watcher errors as overflow so the renderer conservatively refreshes rather than trusting possibly-invalid caches (§7.2, §7.3). - console.error(`[filesystem-watcher] error for ${rootKey}:`, err) - emitOverflowPayload(root) - // Why: after an error the native subscription may be invalid (deleted root); tear down the dead watcher so it doesn't dangle (§7.3). - if (root.batch.timer) { - clearTimeout(root.batch.timer) - } - // Why: error callback can fire before subscribe() assigns root.subscription; guard against null so cleanup doesn't crash. - if (root.subscription) { - retainLocalWatcherPhysicalFailure(rootKey, err) - void trackLocalUnsubscribe(rootKey, root) - } - errorCleanedUp = true - watchedRoots.delete(rootKey) - return - } - - queueWatcherEvents(root.batch, events) - scheduleBatchFlush(root) - }, - watcherOptions, - { - delivery: { maxEventsPerBatch: MAX_BATCHED_WATCHER_EVENTS }, - // A child restart or bounded-queue overflow loses path precision; both need the same conservative renderer refresh. - onInterruption: markWatcherInterrupted, - onOverflow: markWatcherInterrupted, - signal - } - ) - - // Why: error callback already cleaned up watchedRoots before subscribe() resolved; unsubscribe this orphaned subscription so it doesn't leak. - if (errorCleanedUp) { - void trackLocalUnsubscribe(rootKey, root) - throw new Error(`Watcher for ${rootKey} errored during subscribe`) - } - } catch (err) { - // Why: watcher backend can throw synchronously on a deleted root/permission error; log rather than crash the main process (§7.3). - console.error(`[filesystem-watcher] failed to subscribe ${rootKey}:`, err) - throw err - } - - return root -} - -// ── Subscribe / Unsubscribe ────────────────────────────────────────── - -function cleanupLocalWatchersForSender(senderId: number): void { - for (const [rootKey, suspended] of suspendedLocalWatcherListeners) { - suspended.listeners.delete(senderId) - if (suspended.listeners.size === 0) { - suspendedLocalWatcherListeners.delete(rootKey) - } - } - cleanupInFlightLocalInstallsForSender(senderId) - for (const [key, watchedRoot] of watchedRoots) { - if (watchedRoot.listeners.has(senderId)) { - watchedRoot.listeners.delete(senderId) - if (watchedRoot.listeners.size === 0) { - // Cancel any pending grace-period teardown for this root. - const pending = pendingTeardowns.get(key) - if (pending) { - clearTimeout(pending) - pendingTeardowns.delete(key) - } - if (watchedRoot.batch.timer) { - clearTimeout(watchedRoot.batch.timer) - } - trackLocalUnsubscribe(key, watchedRoot) - watchedRoots.delete(key) - } - } - } -} - -function trackLocalUnsubscribe(rootKey: string, root: WatchedRoot): Promise { - const rootUnsubscribes = pendingLocalUnsubscribesByRoot.get(rootKey) ?? new Set>() - pendingLocalUnsubscribesByRoot.set(rootKey, rootUnsubscribes) - const unsubscribePromise = Promise.resolve() - .then(() => root.subscription.unsubscribe()) - .finally(() => { - pendingLocalUnsubscribes.delete(unsubscribePromise) - rootUnsubscribes.delete(unsubscribePromise) - if (rootUnsubscribes.size === 0) { - pendingLocalUnsubscribesByRoot.delete(rootKey) - } - }) - pendingLocalUnsubscribes.add(unsubscribePromise) - rootUnsubscribes.add(unsubscribePromise) - // Why: swallow here to avoid unhandled rejections, but keep the original promise rejected so later destructive cleanup can fail closed. - void unsubscribePromise.catch((error: unknown) => { - if (!abandonedLocalUnsubscribes.has(unsubscribePromise)) { - retainLocalWatcherPhysicalFailure(rootKey, error) - } - console.error(`[filesystem-watcher] unsubscribe error for ${rootKey}:`, error) - }) - return unsubscribePromise -} - -/** Mark unsubscribes whose drain timed out so their late failures stay out of failedLocalUnsubscribes. */ -function abandonLocalUnsubscribes(promises: Iterable>): void { - for (const promise of promises) { - abandonedLocalUnsubscribes.add(promise) - } -} - -function retainLocalWatcherPhysicalFailure(rootKey: string, error: unknown): void { - if (!isWatcherProcessFailure(error) || !error.physicalExit) { - return - } - failedLocalUnsubscribes.set(rootKey, error) - void error.physicalExit.then(() => { - if (failedLocalUnsubscribes.get(rootKey) === error) { - failedLocalUnsubscribes.delete(rootKey) - } - }) -} - -function registerSenderCleanup(sender: WebContents): void { - if (senderCleanupRegistered.has(sender.id)) { - return - } - senderCleanupRegistered.add(sender.id) - sender.once('destroyed', () => { - senderCleanupRegistered.delete(sender.id) - cleanupLocalWatchersForSender(sender.id) - cleanupRemoteWatchersForSender(sender.id) - }) -} - -function addLocalWatchListener(rootKey: string, sender: WebContents): void { - const root = watchedRoots.get(rootKey) - if (!root || sender.isDestroyed()) { - return - } - root.listeners.set(sender.id, sender) - registerSenderCleanup(sender) -} - -async function subscribe( - worktreePath: string, - sender: WebContents, - generation = localWatcherLifecycleGeneration -): Promise { - if (localWatchersClosed || generation !== localWatcherLifecycleGeneration) { - return - } - const finishInstall = beginWatcherInstall(worktreePath) - try { - await subscribeWhileRemovalAllowed(worktreePath, sender, generation) - } finally { - finishInstall() - } -} - -async function subscribeWhileRemovalAllowed( - worktreePath: string, - sender: WebContents, - generation: number -): Promise { - if (localWatchersClosed || generation !== localWatcherLifecycleGeneration) { - return - } - const { key: rootKey, path: rootPath } = localWatcherRoot(worktreePath) - if (sender.isDestroyed()) { - return - } - - // Don't retry roots that already failed — avoids repeated error spam. - if (unwatchableRoots.has(rootKey)) { - rememberUnwatchableRoot(rootKey) - return - } - - let root = watchedRoots.get(rootKey) - - // Cancel any pending grace-period teardown — a new listener arrived. - const pendingTeardown = pendingTeardowns.get(rootKey) - if (pendingTeardown) { - clearTimeout(pendingTeardown) - pendingTeardowns.delete(rootKey) - } - const capacityRetryListeners = takeLocalCapacityRetryListeners(rootKey) - - if (root) { - for (const listener of capacityRetryListeners) { - addLocalWatchListener(rootKey, listener) - } - addLocalWatchListener(rootKey, sender) - return - } - - const pendingInstall = pendingLocalInstallPromises.get(rootKey) - if (pendingInstall) { - const inFlight = inFlightLocalInstalls.get(rootKey) - const canJoinInstall = inFlight && !inFlight.abortController.signal.aborted - if (canJoinInstall) { - // Why: an unwatch may cancel an install while another renderer awaits the same root; a new live listener keeps it alive. - addInFlightLocalInstallListener(inFlight, sender) - for (const listener of capacityRetryListeners) { - addInFlightLocalInstallListener(inFlight, listener) - } - } - const result = await pendingInstall - if ( - result === 'cancelled' && - !canJoinInstall && - !localWatchersClosed && - generation === localWatcherLifecycleGeneration - ) { - // Why: AbortSignal can't be revived; listeners arriving after cancellation wait out that generation, then own a fresh install. - if (pendingLocalInstallPromises.get(rootKey) === pendingInstall) { - pendingLocalInstallPromises.delete(rootKey) - } - const retryListeners = new Map( - capacityRetryListeners.map((listener) => [listener.id, listener]) - ) - retryListeners.set(sender.id, sender) - for (const listener of retryListeners.values()) { - if (!listener.isDestroyed()) { - await subscribeWhileRemovalAllowed(worktreePath, listener, generation) - } - } - return - } - if (!inFlight) { - if (result === 'installed') { - for (const listener of capacityRetryListeners) { - addLocalWatchListener(rootKey, listener) - } - } else if (result === 'capacity') { - const retryListeners = new Map( - capacityRetryListeners.map((listener) => [listener.id, listener]) - ) - retryListeners.set(sender.id, sender) - scheduleLocalCapacityRetry(rootKey, worktreePath, retryListeners) - } - } - if ( - result === 'installed' && - watchedRoots.has(rootKey) && - !sender.isDestroyed() && - (!inFlight || inFlight.listeners.has(sender.id)) - ) { - addLocalWatchListener(rootKey, sender) - } - return - } - - const cancelToken: LocalWatcherInstallToken = { - cancelled: false, - listeners: new Map(), - abortController: new AbortController() - } - inFlightLocalInstalls.set(rootKey, cancelToken) - for (const listener of capacityRetryListeners) { - addInFlightLocalInstallListener(cancelToken, listener) - } - addInFlightLocalInstallListener(cancelToken, sender) - const installPromise = doInstallLocalWatcher(rootKey, rootPath, worktreePath, cancelToken) - pendingLocalInstallPromises.set(rootKey, installPromise) - try { - await installPromise - } finally { - if (pendingLocalInstallPromises.get(rootKey) === installPromise) { - pendingLocalInstallPromises.delete(rootKey) - } - } -} - -async function doInstallLocalWatcher( - rootKey: string, - rootPath: string, - worktreePath: string, - cancelToken: LocalWatcherInstallToken -): Promise { - let root: WatchedRoot - try { - const s = await stat(rootPath) - if (!s.isDirectory()) { - console.warn(`[filesystem-watcher] not a directory: ${rootKey}`) - rememberUnwatchableRoot(rootKey) - return 'unavailable' - } - } catch { - console.warn(`[filesystem-watcher] cannot stat root: ${rootKey}`) - rememberUnwatchableRoot(rootKey) - return 'unavailable' - } - - try { - // Why: WSL paths use one snapshot subprocess inside the distro so `wsl --shutdown` can kill it; native Windows uses @parcel/watcher. - root = isWslPath(worktreePath) - ? await createWslWatcher( - rootKey, - worktreePath, - { - ignoreDirs: WATCHER_IGNORE_DIRS, - scheduleBatchFlush, - watchedRoots - }, - cancelToken.abortController.signal - ) - : await createWatcher(rootKey, rootPath, cancelToken.abortController.signal) - } catch (error) { - // Why: setup can fail after its child misses the exit deadline; retain that owner even when the renderer-facing error is swallowed. - retainLocalWatcherPhysicalFailure(rootKey, error) - if (cancelToken.cancelled) { - if (isWatcherProcessFailure(error) && error.code === 'process_unavailable') { - throw error - } - return 'cancelled' - } - // Why: capacity is transient — allow retry once another child exits instead of caching this root as permanently failed. - if (error instanceof WatcherChildCapacityError) { - scheduleLocalCapacityRetry(rootKey, worktreePath, cancelToken.listeners) - return 'capacity' - } - rememberUnwatchableRoot(rootKey) - return 'unavailable' - } finally { - if (inFlightLocalInstalls.get(rootKey) === cancelToken) { - inFlightLocalInstalls.delete(rootKey) - } - } - - const liveListeners = new Map( - Array.from(cancelToken.listeners.entries()).filter(([, listener]) => !listener.isDestroyed()) - ) - if (cancelToken.cancelled || liveListeners.size === 0) { - if (root.batch.timer) { - clearTimeout(root.batch.timer) - } - void trackLocalUnsubscribe(rootKey, root) - return 'cancelled' - } - - root.listeners = liveListeners - watchedRoots.set(rootKey, root) - for (const listener of liveListeners.values()) { - registerSenderCleanup(listener) - } - return 'installed' -} - -function unsubscribe(worktreePath: string, senderId: number): void { - const { key: rootKey } = localWatcherRoot(worktreePath) - const suspended = suspendedLocalWatcherListeners.get(rootKey) - suspended?.listeners.delete(senderId) - if (suspended?.listeners.size === 0) { - suspendedLocalWatcherListeners.delete(rootKey) - } - const capacityRetry = pendingLocalCapacityRetries.get(rootKey) - if (capacityRetry) { - capacityRetry.listeners.delete(senderId) - if (capacityRetry.listeners.size === 0) { - clearLocalCapacityRetry(rootKey) - } - } - const inFlight = inFlightLocalInstalls.get(rootKey) - if (inFlight) { - inFlight.listeners.delete(senderId) - inFlight.cancelled = inFlight.listeners.size === 0 - // Why: last normal disconnect must abort the pending native/forked install (same early-cancel as closeLocalWatcherForWorktreePath). - if (inFlight.cancelled) { - inFlight.abortController.abort() - } - } - - const root = watchedRoots.get(rootKey) - if (!root) { - return - } - - root.listeners.delete(senderId) - - // Defer teardown when the last subscriber leaves so rapid worktree switches reuse the native watcher. - if (root.listeners.size === 0) { - if (root.batch.timer) { - clearTimeout(root.batch.timer) - } - - // Why: duplicate unwatch calls for a root would leak overwritten grace timers; keep just one. - if (pendingTeardowns.has(rootKey)) { - return - } - - const teardownTimer = setTimeout(() => { - pendingTeardowns.delete(rootKey) - // Re-check: a new listener may have arrived during the grace period. - const currentRoot = watchedRoots.get(rootKey) - if (!currentRoot || currentRoot.listeners.size > 0) { - return - } - void trackLocalUnsubscribe(rootKey, currentRoot) - watchedRoots.delete(rootKey) - }, WATCHER_TEARDOWN_GRACE_MS) - - pendingTeardowns.set(rootKey, teardownTimer) - } -} - -export async function closeLocalWatcherForWorktreePath( - worktreePath: string, - deadline: WatcherRemovalDeadline = createWatcherRemovalDeadline() -): Promise { - const { key: rootKey } = localWatcherRoot(worktreePath) - const suspended = suspendedLocalWatcherListeners.get(rootKey) ?? { - worktreePath, - listeners: new Map() - } - for (const source of [ - pendingLocalCapacityRetries.get(rootKey)?.listeners, - inFlightLocalInstalls.get(rootKey)?.listeners, - watchedRoots.get(rootKey)?.listeners - ]) { - for (const [senderId, sender] of source ?? []) { - if (!sender.isDestroyed()) { - suspended.listeners.set(senderId, sender) - } - } - } - if (suspended.listeners.size > 0) { - suspendedLocalWatcherListeners.set(rootKey, suspended) - } - clearLocalCapacityRetry(rootKey) - const pendingTeardown = pendingTeardowns.get(rootKey) - if (pendingTeardown) { - clearTimeout(pendingTeardown) - pendingTeardowns.delete(rootKey) - } - - const inFlight = inFlightLocalInstalls.get(rootKey) - if (inFlight) { - // Why: Windows locks watched directories; deletion must cancel an in-flight subscription before Git removes the tree. - inFlight.listeners.clear() - inFlight.cancelled = true - inFlight.abortController.abort() - } - // Why: abort alone is not enough if the native subscribe never settles; bound so delete cannot hang the app. - const pendingInstall = pendingLocalInstallPromises.get(rootKey) - const installDrain = await drainBeforeWatcherRemoval( - pendingInstall, - deadline, - `local watcher install for ${rootKey}`, - { reserveMs: WATCHER_REMOVAL_FINAL_DRAIN_RESERVE_MS } - ) - if (installDrain === 'timeout') { - // Why: an abandoned install never runs its own cleanup, so leaving these entries would make every - // later watch of this root queue behind the same wedged promise. Identity-checked so a late settle - // can't evict a newer install. - if (pendingLocalInstallPromises.get(rootKey) === pendingInstall) { - pendingLocalInstallPromises.delete(rootKey) - } - if (inFlight && inFlightLocalInstalls.get(rootKey) === inFlight) { - inFlightLocalInstalls.delete(rootKey) - } - } - const pendingUnsubscribes = pendingLocalUnsubscribesByRoot.get(rootKey) - if (pendingUnsubscribes) { - const draining = Array.from(pendingUnsubscribes) - const unsubscribeDrain = await drainBeforeWatcherRemoval( - // Why the per-promise catch: an already-abandoned unsubscribe belongs to a delete that finished - // without it; re-raising its rejection here would fail a later close on stale news. - Promise.all( - draining.map((unsubscribe) => - abandonedLocalUnsubscribes.has(unsubscribe) - ? unsubscribe.catch(() => undefined) - : unsubscribe - ) - ), - deadline, - `local watcher unsubscribe for ${rootKey}`, - { reserveMs: WATCHER_REMOVAL_FINAL_DRAIN_RESERVE_MS } - ) - if (unsubscribeDrain === 'timeout') { - abandonLocalUnsubscribes(draining) - } - } - if (failedLocalUnsubscribes.has(rootKey)) { - throw failedLocalUnsubscribes.get(rootKey) - } - - const root = watchedRoots.get(rootKey) - if (!root) { - return - } - if (root.batch.timer) { - clearTimeout(root.batch.timer) - } - watchedRoots.delete(rootKey) - // Why: the in-process Parcel fallback has no unsubscribe timeout of its own, so an unbounded await - // here would hang delete forever and hold the removal gate. The promise stays tracked in - // pendingLocalUnsubscribesByRoot, so a later close still observes its failure. - const finalUnsubscribe = trackLocalUnsubscribe(rootKey, root) - const finalDrain = await drainBeforeWatcherRemoval( - finalUnsubscribe, - deadline, - `local watcher unsubscribe for ${rootKey}` - ) - if (finalDrain === 'timeout') { - abandonLocalUnsubscribes([finalUnsubscribe]) - } -} - -export async function restoreLocalWatcherAfterFailedRemoval(worktreePath: string): Promise { - const { key: rootKey } = localWatcherRoot(worktreePath) - const suspended = suspendedLocalWatcherListeners.get(rootKey) - if (!suspended) { - return - } - suspendedLocalWatcherListeners.delete(rootKey) - const failures: unknown[] = [] - const failedListeners = new Map() - for (const sender of suspended.listeners.values()) { - if (sender.isDestroyed()) { - continue - } - try { - await subscribe(suspended.worktreePath, sender) - sender.send('fs:changed', { - worktreePath: suspended.worktreePath, - events: [{ kind: 'overflow', absolutePath: suspended.worktreePath }] - } satisfies FsChangedPayload) - } catch (error) { - failures.push(error) - failedListeners.set(sender.id, sender) - } - } - if (failures.length > 0) { - suspendedLocalWatcherListeners.set(rootKey, { - worktreePath: suspended.worktreePath, - listeners: failedListeners - }) - throw failures[0] - } -} - -export function forgetLocalWatcherRemovalSnapshot(worktreePath: string): void { - suspendedLocalWatcherListeners.delete(localWatcherRoot(worktreePath).key) -} - -// Remote watcher state -type RemoteWatcherState = { - unwatch: () => void - listeners: Map - installToken: RemoteWatcherInstallToken - /** Held so every teardown can drop the trailing flush timer instead of letting it fire into a dead watch. */ - batch: RemoteWatcherEventBatch -} - -type RemoteWatcherInstallToken = { - cancelled: boolean - listeners: Map - abortController: AbortController - abortScheduled: boolean - terminalError?: Error -} - -// Key: `${connectionId}:${worktreePath}`, Value: shared remote watch state. -const remoteWatchers = new Map() -const suspendedRemoteWatcherListeners = new Map< - string, - { connectionId: string; worktreePath: string; listeners: Map } ->() -// Why: the renderer subscribes once per target and never re-issues, so the intent to watch has to -// outlive any single connection — an install that failed or died with a dropped transport is -// re-armed from here when a provider appears. Without it a reconnect (or a connect slower than the -// retry window) leaves the watch dead until the app restarts. -const desiredRemoteWatchers = new Map< - string, - { connectionId: string; worktreePath: string; listeners: Map } ->() -// Why: provider registration only fires on reconnect, so a watch that dies while the SSH link stays -// healthy (remote OOM, inotify/fd exhaustion, relay watcher killed) has no re-arm trigger at all -// once the fast window gives up. Backoff keeps the recovery attempt without the 1s storm. -const dormantRemoteWatchers = new Map< - string, - { delayMs: number; timer: ReturnType } ->() -const loggedUnavailableRemoteWatchers = new Set() -const pendingRemoteWatcherRetries = new Map>() -const pendingRemoteWatcherRetryListeners = new Map< - string, - { listeners: Map; startedAt: number; resyncOnInstall: boolean } ->() -type RemoteWatcherResyncState = { - lastSentAt: number - listeners: Map - timer?: ReturnType - worktreePath: string -} -const remoteWatcherResyncStates = new Map() -// Why: last-listener cleanup aborts relay setup; late success is unwatched rather than installed after the renderer stopped watching. -const inFlightRemoteInstalls = new Map() -// Why: dedupe concurrent installRemoteWatcher calls per key so overlapping watches share one watcher instead of clobbering per-key state. -const pendingRemoteInstallPromises = new Map>() -// Why: block installs beginning after closeAllWatchers (joiner recursion / retry tick bypass the abort loop); a new fs:watchWorktree clears it. -let 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). -let remoteWatcherLifecycleGeneration = 0 -let unsubscribeFromProviderRegistrations: (() => void) | null = null -const REMOTE_WATCH_RETRY_MS = 1_000 -const REMOTE_WATCH_RETRY_TIMEOUT_MS = 60_000 -// Why: preserve the first and latest resync while bounding full-tree SSH refreshes during flaps. -const REMOTE_WATCH_RESYNC_COALESCE_MS = 5_000 -// Why: doubling from a minute to a half-hour ceiling costs a permanently broken remote ~7 fs.watch -// calls in the first hour and 2/hour after, which a flapping link can absorb. -const REMOTE_WATCH_DORMANT_RETRY_MS = 60_000 -const REMOTE_WATCH_DORMANT_RETRY_MAX_MS = 30 * 60_000 - -export async function closeRemoteWatcherForWorktreePath( - connectionId: string, - worktreePath: string -): Promise { - const key = remoteWatcherKey(connectionId, worktreePath) - const suspended = suspendedRemoteWatcherListeners.get(key) ?? { - connectionId, - worktreePath, - listeners: new Map() - } - for (const source of [ - pendingRemoteWatcherRetryListeners.get(key)?.listeners, - inFlightRemoteInstalls.get(key)?.listeners, - remoteWatchers.get(key)?.listeners - ]) { - for (const [senderId, sender] of source ?? []) { - if (!sender.isDestroyed()) { - suspended.listeners.set(senderId, sender) - } - } - } - if (suspended.listeners.size > 0) { - suspendedRemoteWatcherListeners.set(key, suspended) - } - clearRemoteWatcherResync(key) - const retryTimer = pendingRemoteWatcherRetries.get(key) - if (retryTimer) { - clearTimeout(retryTimer) - pendingRemoteWatcherRetries.delete(key) - pendingRemoteWatcherRetryListeners.delete(key) - } - // Why: removal is deliberate — a backoff firing mid-removal would re-watch the path being deleted. - clearDormantRemoteWatcher(key) - const inFlight = inFlightRemoteInstalls.get(key) - if (inFlight) { - inFlight.listeners.clear() - inFlight.cancelled = true - } - const state = remoteWatchers.get(key) - const provider = getSshFilesystemProvider(connectionId) - await (provider?.closeWatch - ? provider.closeWatch(worktreePath) - : Promise.resolve(state?.unwatch())) - state?.batch.close() - remoteWatchers.delete(key) - loggedUnavailableRemoteWatchers.delete(key) -} - -export async function restoreRemoteWatcherAfterFailedRemoval( - connectionId: string, - worktreePath: string -): Promise { - const key = remoteWatcherKey(connectionId, worktreePath) - const suspended = suspendedRemoteWatcherListeners.get(key) - if (!suspended) { - return - } - suspendedRemoteWatcherListeners.delete(key) - for (const sender of suspended.listeners.values()) { - if (sender.isDestroyed()) { - continue - } - const result = await installRemoteWatcher(sender, connectionId, worktreePath) - if (result === 'unavailable') { - scheduleRemoteWatcherRetry(sender, connectionId, worktreePath) - } - sender.send('fs:changed', { - worktreePath, - events: [{ kind: 'overflow', absolutePath: worktreePath }] - } satisfies FsChangedPayload) - } -} - -export function forgetRemoteWatcherRemovalSnapshot( - connectionId: string, - worktreePath: string -): void { - const key = remoteWatcherKey(connectionId, worktreePath) - suspendedRemoteWatcherListeners.delete(key) - clearRemoteWatcherResync(key) - const retryTimer = pendingRemoteWatcherRetries.get(key) - if (retryTimer) { - clearTimeout(retryTimer) - pendingRemoteWatcherRetries.delete(key) - } - pendingRemoteWatcherRetryListeners.delete(key) - // Why: the worktree is gone — keeping the intent lets a reconnect landing before the renderer's - // unwatch re-watch a deleted path (60s of retries against the host, then a bogus overflow). - desiredRemoteWatchers.delete(key) - clearDormantRemoteWatcher(key) -} - -function addInFlightRemoteInstallListener( - token: RemoteWatcherInstallToken, - sender: WebContents -): void { - if (sender.isDestroyed() || token.abortController.signal.aborted) { - return - } - token.listeners.set(sender.id, sender) - token.cancelled = false - registerSenderCleanup(sender) -} - -function cancelInFlightRemoteInstallIfUnowned(token: RemoteWatcherInstallToken): void { - token.cancelled = token.listeners.size === 0 - if (!token.cancelled || token.abortScheduled || token.abortController.signal.aborted) { - return - } - token.abortScheduled = true - // Why: a replacement sender can synchronously revive the shared install during a renderer handoff; otherwise stop the relay crawl next microtask. - queueMicrotask(() => { - token.abortScheduled = false - if (token.cancelled && token.listeners.size === 0) { - token.abortController.abort() - } - }) -} - -function cleanupInFlightRemoteInstallsForSender(senderId: number): void { - for (const token of inFlightRemoteInstalls.values()) { - token.listeners.delete(senderId) - cancelInFlightRemoteInstallIfUnowned(token) - } - for (const [key, retry] of pendingRemoteWatcherRetryListeners) { - retry.listeners.delete(senderId) - if (retry.listeners.size === 0) { - const timer = pendingRemoteWatcherRetries.get(key) - if (timer) { - clearTimeout(timer) - pendingRemoteWatcherRetries.delete(key) - } - pendingRemoteWatcherRetryListeners.delete(key) - } - } -} - -function addRemoteWatchListener(key: string, sender: WebContents): void { - const state = remoteWatchers.get(key) - if (!state) { - return - } - state.listeners.set(sender.id, sender) - registerSenderCleanup(sender) -} - -function releaseRemoteWatchListener(key: string, senderId: number): void { - const state = remoteWatchers.get(key) - if (!state) { - return - } - state.listeners.delete(senderId) - if (state.listeners.size > 0) { - return - } - state.unwatch() - state.batch.close() - remoteWatchers.delete(key) -} - -function cleanupRemoteWatchersForSender(senderId: number): void { - for (const key of Array.from(desiredRemoteWatchers.keys())) { - forgetDesiredRemoteWatcher(key, senderId) - } - for (const [key, suspended] of suspendedRemoteWatcherListeners) { - suspended.listeners.delete(senderId) - if (suspended.listeners.size === 0) { - suspendedRemoteWatcherListeners.delete(key) - } - } - cleanupInFlightRemoteInstallsForSender(senderId) - for (const key of Array.from(remoteWatchers.keys())) { - releaseRemoteWatchListener(key, senderId) - } -} - -type RemoteWatcherInstallResult = 'installed' | 'unavailable' | 'cancelled' - -async function installRemoteWatcher( - sender: WebContents, - connectionId: string, - worktreePath: string, - generation = remoteWatcherLifecycleGeneration -): Promise { - // Why: refuse installs racing in after teardown (or a waiter from an earlier lifecycle) so provider.watch() isn't called post-shutdown. - if (remoteWatchersClosed || generation !== remoteWatcherLifecycleGeneration) { - return 'cancelled' - } - const finishInstall = beginWatcherInstall(worktreePath, connectionId) - try { - return await installRemoteWatcherWhileRemovalAllowed( - sender, - connectionId, - worktreePath, - generation - ) - } finally { - finishInstall() - } -} - -async function installRemoteWatcherWhileRemovalAllowed( - sender: WebContents, - connectionId: string, - worktreePath: string, - generation: number -): Promise { - const provider = getSshFilesystemProvider(connectionId) - if (!provider || sender.isDestroyed()) { - return 'unavailable' - } - - const key = remoteWatcherKey(connectionId, worktreePath) - const existing = remoteWatchers.get(key) - if (existing) { - addRemoteWatchListener(key, sender) - return 'installed' - } - // Why: concurrent same-key watches must share the first provider.watch(); separate watchers would clobber per-key state and drop the unwatch handle. - const pendingInstall = pendingRemoteInstallPromises.get(key) - if (pendingInstall) { - const inFlight = inFlightRemoteInstalls.get(key) - const canJoinInstall = inFlight && !inFlight.abortController.signal.aborted - if (canJoinInstall) { - // Why: a new watcher joining before provider.watch() resolves should revive the install instead of inheriting the stale cancellation. - addInFlightRemoteInstallListener(inFlight, sender) - } - const result = await pendingInstall - if ( - result === 'installed' && - remoteWatchers.has(key) && - !sender.isDestroyed() && - (!inFlight || inFlight.listeners.has(sender.id)) - ) { - addRemoteWatchListener(key, sender) - } - if ( - result === 'cancelled' && - !canJoinInstall && - !sender.isDestroyed() && - generation === remoteWatcherLifecycleGeneration - ) { - // Why: AbortSignal can't be revived; a listener arriving after cancellation waits out that generation, then owns a fresh install. - if (pendingRemoteInstallPromises.get(key) === pendingInstall) { - pendingRemoteInstallPromises.delete(key) - } - return installRemoteWatcher(sender, connectionId, worktreePath, generation) - } - return result - } - const cancelToken: RemoteWatcherInstallToken = { - cancelled: false, - listeners: new Map(), - abortController: new AbortController(), - abortScheduled: false - } - inFlightRemoteInstalls.set(key, cancelToken) - addInFlightRemoteInstallListener(cancelToken, sender) - const installPromise = doInstallRemoteWatcher( - provider, - key, - connectionId, - worktreePath, - cancelToken - ) - pendingRemoteInstallPromises.set(key, installPromise) - try { - return await installPromise - } finally { - if (pendingRemoteInstallPromises.get(key) === installPromise) { - pendingRemoteInstallPromises.delete(key) - } - } -} - -async function doInstallRemoteWatcher( - provider: NonNullable>, - key: string, - connectionId: string, - worktreePath: string, - cancelToken: RemoteWatcherInstallToken -): Promise { - let unwatch: () => void - // Why: the abandon paths below close the batch directly; once it is stored on the remoteWatchers - // entry, every later teardown closes it through that handle. deliver's state/installToken guard - // still covers a flush that races a close. - const batch = createRemoteWatcherEventBatch({ - rootPath: worktreePath, - trailingMs: WATCH_BATCH_TRAILING_MS, - maxWaitMs: WATCH_BATCH_MAX_WAIT_MS, - maxEvents: MAX_BATCHED_WATCHER_EVENTS, - deliver: (events) => { - const state = remoteWatchers.get(key) - // Why: buffering defers delivery past install, so a bare non-null check would let a stale flush - // land on a different install generation. - if (!state || state.installToken !== cancelToken) { - return - } - for (const listener of state.listeners.values()) { - if (listener.isDestroyed()) { - continue - } - try { - listener.send('fs:changed', { - worktreePath, - events - } satisfies FsChangedPayload) - } catch (err) { - // Why: batching moved this send out of the SSH mux's notification try/catch into a bare - // timer, so a frame disposed mid-window would escape as a fatal main-process exception. - console.warn( - `[filesystem-watch] failed to deliver remote fs:changed for ${worktreePath}: ${ - err instanceof Error ? err.message : String(err) - }` - ) - } - } - } - }) - try { - unwatch = await provider.watch( - worktreePath, - (events) => { - batch.push(events) - }, - { - signal: cancelToken.abortController.signal, - onTerminalError: (error) => - handleRemoteWatcherTerminalError(key, connectionId, worktreePath, cancelToken, error) - } - ) - } catch (err) { - // Why: the provider can fire its callback before rejecting (an aborted reinstall), so this - // abandon path owes the same close as the two below — nothing else holds the batch. - batch.close() - if (cancelToken.cancelled || cancelToken.abortController.signal.aborted) { - return 'cancelled' - } - console.warn(`[filesystem-watcher] SSH watcher unavailable for ${key}:`, err) - return 'unavailable' - } finally { - if (inFlightRemoteInstalls.get(key) === cancelToken) { - inFlightRemoteInstalls.delete(key) - } - } - const liveListeners = new Map( - Array.from(cancelToken.listeners.entries()).filter(([, listener]) => !listener.isDestroyed()) - ) - if (cancelToken.cancelled || liveListeners.size === 0) { - batch.close() - try { - unwatch() - } catch (err) { - console.error(`[filesystem-watcher] remote unwatch (post-cancel) error for ${key}:`, err) - } - return 'cancelled' - } - if (cancelToken.terminalError) { - batch.close() - return 'unavailable' - } - remoteWatchers.set(key, { unwatch, listeners: liveListeners, installToken: cancelToken, batch }) - for (const listener of liveListeners.values()) { - registerSenderCleanup(listener) - } - loggedUnavailableRemoteWatchers.delete(key) - return 'installed' -} - -function handleRemoteWatcherTerminalError( - key: string, - connectionId: string, - worktreePath: string, - installToken: RemoteWatcherInstallToken, - error: Error -): void { - installToken.terminalError = error - const state = remoteWatchers.get(key) - if (!state || state.installToken !== installToken) { - return - } - remoteWatchers.delete(key) - state.batch.close() - if (remoteWatchersClosed || suspendedRemoteWatcherListeners.has(key)) { - return - } - console.warn(`[filesystem-watcher] SSH watcher terminated for ${key}:`, error) - const startedAt = Date.now() - for (const listener of state.listeners.values()) { - scheduleRemoteWatcherRetry(listener, connectionId, worktreePath, startedAt, true) - } -} - -function isCurrentDesiredRemoteWatcher(key: string, listener: WebContents): boolean { - return desiredRemoteWatchers.get(key)?.listeners.get(listener.id) === listener -} - -function clearRemoteWatcherResync(key: string): void { - const state = remoteWatcherResyncStates.get(key) - if (state?.timer) { - clearTimeout(state.timer) - } - remoteWatcherResyncStates.delete(key) -} - -function flushRemoteWatcherResync(key: string): void { - const state = remoteWatcherResyncStates.get(key) - if (!state) { - return - } - state.timer = undefined - if ( - remoteWatchersClosed || - suspendedRemoteWatcherListeners.has(key) || - !remoteWatchers.has(key) - ) { - if (!desiredRemoteWatchers.has(key)) { - remoteWatcherResyncStates.delete(key) - } - return - } - let sent = false - for (const listener of state.listeners.values()) { - if (listener.isDestroyed() || !isCurrentDesiredRemoteWatcher(key, listener)) { - continue - } - try { - listener.send('fs:changed', { - worktreePath: state.worktreePath, - events: [{ kind: 'overflow', absolutePath: state.worktreePath }] - } satisfies FsChangedPayload) - sent = true - } catch (error) { - console.warn(`[filesystem-watcher] failed to send SSH watcher resync for ${key}:`, error) - } - } - state.listeners.clear() - if (sent) { - state.lastSentAt = Date.now() - } else { - remoteWatcherResyncStates.delete(key) - } -} - -function requestRemoteWatcherResync( - key: string, - worktreePath: string, - listeners: Iterable -): void { - const state = remoteWatcherResyncStates.get(key) ?? { - lastSentAt: Number.NEGATIVE_INFINITY, - listeners: new Map(), - worktreePath - } - state.worktreePath = worktreePath - for (const listener of listeners) { - if (!listener.isDestroyed() && isCurrentDesiredRemoteWatcher(key, listener)) { - state.listeners.set(listener.id, listener) - } - } - if (state.listeners.size === 0) { - return - } - remoteWatcherResyncStates.set(key, state) - const delayMs = Math.max(0, state.lastSentAt + REMOTE_WATCH_RESYNC_COALESCE_MS - Date.now()) - if (delayMs === 0) { - flushRemoteWatcherResync(key) - return - } - if (!state.timer) { - state.timer = setTimeout(() => flushRemoteWatcherResync(key), delayMs) - state.timer.unref?.() - } -} - -function scheduleRemoteWatcherRetry( - sender: WebContents, - connectionId: string, - worktreePath: string, - startedAt = Date.now(), - // Why: a retry that replaces a watch which was already live owes the renderer an overflow once it - // lands — the events lost while it was down are otherwise never signalled. - resyncOnInstall = false -): void { - const key = remoteWatcherKey(connectionId, worktreePath) - const existingRetry = pendingRemoteWatcherRetryListeners.get(key) - if (existingRetry) { - if (!sender.isDestroyed()) { - existingRetry.listeners.set(sender.id, sender) - } - existingRetry.resyncOnInstall ||= resyncOnInstall - return - } - - const retry = { - listeners: new Map(sender.isDestroyed() ? [] : [[sender.id, sender]]), - startedAt, - resyncOnInstall - } - pendingRemoteWatcherRetryListeners.set(key, retry) - - if (Date.now() - startedAt >= REMOTE_WATCH_RETRY_TIMEOUT_MS || sender.isDestroyed()) { - pendingRemoteWatcherRetries.delete(key) - pendingRemoteWatcherRetryListeners.delete(key) - loggedUnavailableRemoteWatchers.delete(key) - clearRemoteWatcherResync(key) - // Why: handler already resolved so the renderer thinks the watch is live; emit overflow to force a manual refresh instead of waiting forever. - for (const listener of retry.listeners.values()) { - if (listener.isDestroyed() || !isCurrentDesiredRemoteWatcher(key, listener)) { - continue - } - console.warn( - `[filesystem-watcher] giving up SSH watch retry for ${worktreePath} on connection ${connectionId} after ${REMOTE_WATCH_RETRY_TIMEOUT_MS}ms` - ) - listener.send('fs:changed', { - worktreePath, - events: [{ kind: 'overflow', absolutePath: worktreePath }] - } satisfies FsChangedPayload) - } - // Why: overflow only refreshes once — without this the watch stays dead until the app restarts. - scheduleDormantRemoteWatcherRearm(connectionId, worktreePath) - return - } - - const retryTimer = setTimeout(() => { - pendingRemoteWatcherRetries.delete(key) - pendingRemoteWatcherRetryListeners.delete(key) - const listeners = Array.from(retry.listeners.values()).filter( - (listener) => !listener.isDestroyed() && isCurrentDesiredRemoteWatcher(key, listener) - ) - void Promise.all( - listeners.map((listener) => installRemoteWatcher(listener, connectionId, worktreePath)) - ) - .then((results) => { - if (retry.resyncOnInstall) { - requestRemoteWatcherResync( - key, - worktreePath, - listeners.filter((_, index) => results[index] === 'installed') - ) - } - // Why: don't re-arm on 'cancelled' (renderer stopped watching) — it would fire a stale overflow when the 60s window expires. - if (results.some((result) => result === 'unavailable')) { - for (const listener of listeners) { - scheduleRemoteWatcherRetry( - listener, - connectionId, - worktreePath, - retry.startedAt, - retry.resyncOnInstall - ) - } - } - }) - .catch((error: unknown) => { - if (isWatcherRemovalInProgressError(error)) { - return - } - for (const listener of listeners) { - scheduleRemoteWatcherRetry( - listener, - connectionId, - worktreePath, - retry.startedAt, - retry.resyncOnInstall - ) - } - }) - }, REMOTE_WATCH_RETRY_MS) - pendingRemoteWatcherRetries.set(key, retryTimer) -} +import { + closeLocalWatcherForWorktreePath, + forgetLocalWatcherRemovalSnapshot, + restoreLocalWatcherAfterFailedRemoval +} from './filesystem-watcher-local-removal' +import { + closeRemoteWatcherForWorktreePath, + forgetRemoteWatcherRemovalSnapshot, + restoreRemoteWatcherAfterFailedRemoval +} from './filesystem-watcher-remote-removal' + +export { closeLocalWatcherForWorktreePath } from './filesystem-watcher-local-removal' +export { restoreLocalWatcherAfterFailedRemoval } from './filesystem-watcher-local-removal' +export { forgetLocalWatcherRemovalSnapshot } from './filesystem-watcher-local-removal' +export { closeRemoteWatcherForWorktreePath } from './filesystem-watcher-remote-removal' +export { restoreRemoteWatcherAfterFailedRemoval } from './filesystem-watcher-remote-removal' +export { forgetRemoteWatcherRemovalSnapshot } from './filesystem-watcher-remote-removal' +export { registerFilesystemWatcherHandlers } from './filesystem-watcher-handlers' +export { closeAllWatchers } from './filesystem-watcher-shutdown' // ── Public API ─────────────────────────────────────────────────────── -export function registerFilesystemWatcherHandlers(): void { - // Why: re-registration replaces the handler set, so drop the previous subscription instead of - // stacking a second re-arm on every provider registration. - unsubscribeFromProviderRegistrations?.() - unsubscribeFromProviderRegistrations = onSshFilesystemProviderRegistered( - reinstallRemoteWatchersForConnection - ) - - ipcMain.handle( - 'fs:watchWorktree', - async (event, args: { worktreePath: string; connectionId?: string }): Promise => { - if (args.connectionId) { - // Why: a real new watch reopens the subsystem after closeAllWatchers latched it shut (also resets tests between cases). - remoteWatchersClosed = false - const key = remoteWatcherKey(args.connectionId, args.worktreePath) - // Why: record intent before the install so a provider registering mid-flight (or long after - // this attempt gives up) can still re-arm this listener. - rememberDesiredRemoteWatcher(args.connectionId, args.worktreePath, event.sender) - const result = await installRemoteWatcher( - event.sender, - args.connectionId, - args.worktreePath - ) - if (result === 'unavailable') { - if (!loggedUnavailableRemoteWatchers.has(key)) { - loggedUnavailableRemoteWatchers.add(key) - console.warn( - `[filesystem-watcher] SSH filesystem provider unavailable; retrying watch for ${args.worktreePath} on connection ${args.connectionId}` - ) - } - scheduleRemoteWatcherRetry(event.sender, args.connectionId, args.worktreePath) - return - } - return - } - // Why: reopen the local subsystem for tests and post-shutdown reattachment; stale callers keep the prior generation. - localWatchersClosed = false - await subscribe(args.worktreePath, event.sender) - } - ) - - ipcMain.handle( - 'fs:unwatchWorktree', - (_event, args: { worktreePath: string; connectionId?: string }): void => { - if (args.connectionId) { - const key = remoteWatcherKey(args.connectionId, args.worktreePath) - // 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) - const suspended = suspendedRemoteWatcherListeners.get(key) - suspended?.listeners.delete(_event.sender.id) - if (suspended?.listeners.size === 0) { - suspendedRemoteWatcherListeners.delete(key) - } - const retry = pendingRemoteWatcherRetryListeners.get(key) - retry?.listeners.delete(_event.sender.id) - const retryTimer = pendingRemoteWatcherRetries.get(key) - if (retryTimer && retry?.listeners.size === 0) { - clearTimeout(retryTimer) - pendingRemoteWatcherRetries.delete(key) - pendingRemoteWatcherRetryListeners.delete(key) - } - // Why: a retry-tick provider.watch() may still be in flight; mark cancelled so its resolved unwatch handle is discarded. - const inFlight = inFlightRemoteInstalls.get(key) - if (inFlight) { - inFlight.listeners.delete(_event.sender.id) - cancelInFlightRemoteInstallIfUnowned(inFlight) - } - loggedUnavailableRemoteWatchers.delete(key) - releaseRemoteWatchListener(key, _event?.sender?.id ?? 0) - return - } - const senderId = _event.sender.id - unsubscribe(args.worktreePath, senderId) - } - ) -} - -function remoteWatcherKey(connectionId: string, worktreePath: string): string { - return JSON.stringify([connectionId, normalizeRuntimePathForComparison(worktreePath)]) -} - -function rememberDesiredRemoteWatcher( - connectionId: string, - worktreePath: string, - sender: WebContents -): void { - if (sender.isDestroyed()) { - return - } - const key = remoteWatcherKey(connectionId, worktreePath) - const desired = desiredRemoteWatchers.get(key) ?? { - connectionId, - worktreePath, - listeners: new Map() - } - desired.listeners.set(sender.id, sender) - desiredRemoteWatchers.set(key, desired) - registerSenderCleanup(sender) -} - -function forgetDesiredRemoteWatcher(key: string, senderId: number): void { - const desired = desiredRemoteWatchers.get(key) - if (!desired) { - return - } - desired.listeners.delete(senderId) - if (desired.listeners.size === 0) { - desiredRemoteWatchers.delete(key) - clearRemoteWatcherResync(key) - clearDormantRemoteWatcher(key) - } -} - -function clearDormantRemoteWatcher(key: string): void { - const dormant = dormantRemoteWatchers.get(key) - if (!dormant) { - return - } - clearTimeout(dormant.timer) - dormantRemoteWatchers.delete(key) -} - -function scheduleDormantRemoteWatcherRearm( - connectionId: string, - worktreePath: string, - delayMs = REMOTE_WATCH_DORMANT_RETRY_MS -): void { - const key = remoteWatcherKey(connectionId, worktreePath) - if (remoteWatchersClosed || !desiredRemoteWatchers.has(key) || dormantRemoteWatchers.has(key)) { - return - } - const timer = setTimeout(() => { - dormantRemoteWatchers.delete(key) - void rearmDormantRemoteWatcher(key, connectionId, worktreePath, delayMs) - }, delayMs) - // Why: a half-hour timer shouldn't be what keeps the process alive at quit. - timer.unref?.() - dormantRemoteWatchers.set(key, { delayMs, timer }) -} - -async function rearmDormantRemoteWatcher( - key: string, - connectionId: string, - worktreePath: string, - delayMs: number -): Promise { - const desired = desiredRemoteWatchers.get(key) - if (remoteWatchersClosed || !desired) { - return - } - for (const [senderId, sender] of Array.from(desired.listeners)) { - if (sender.isDestroyed()) { - desired.listeners.delete(senderId) - } - } - if (desired.listeners.size === 0) { - desiredRemoteWatchers.delete(key) - return - } - // Why: a live watch or an in-flight fast retry already owns this key; installing again would - // clobber the entry the running watch reads its listeners from. - if (remoteWatchers.has(key) || pendingRemoteWatcherRetries.has(key)) { - return - } - // Why: no provider means the connection itself is down, and its registration re-arms for free — - // polling would only add wire traffic to a link that is already being rebuilt. - if (!getSshFilesystemProvider(connectionId)) { - return - } - - const listeners = Array.from(desired.listeners.values()) - let results: RemoteWatcherInstallResult[] - try { - results = await Promise.all( - listeners.map((listener) => installRemoteWatcher(listener, connectionId, worktreePath)) - ) - } catch (error) { - if (isWatcherRemovalInProgressError(error)) { - // Why: removal owns the key now and either forgets the intent or restores the watch itself. - return - } - scheduleDormantRemoteWatcherRearm(connectionId, worktreePath, nextDormantDelayMs(delayMs)) - return - } - requestRemoteWatcherResync( - key, - worktreePath, - listeners.filter((_, index) => results[index] === 'installed') - ) - // Why: 'cancelled' means shutdown or the last listener left, so only 'unavailable' stays dormant. - if (results.some((result) => result === 'unavailable')) { - scheduleDormantRemoteWatcherRearm(connectionId, worktreePath, nextDormantDelayMs(delayMs)) - } -} - -function nextDormantDelayMs(delayMs: number): number { - return Math.min(delayMs * 2, REMOTE_WATCH_DORMANT_RETRY_MAX_MS) -} - -/** - * Rebuild remote watches for a connection whose filesystem provider was just (re)registered. - * - * Why: the relay's watch registrations die with the transport they were made on, and the previous - * provider's unwatch handle is scoped to that dead transport. Reinstalling is the only way the - * subscription comes back, and consumers get an overflow so they resync whatever changed while the - * watch was down. - */ -function reinstallRemoteWatchersForConnection(connectionId: string): void { - if (remoteWatchersClosed) { - return - } - for (const [key, desired] of Array.from(desiredRemoteWatchers)) { - if (desired.connectionId !== connectionId) { - continue - } - for (const [senderId, sender] of Array.from(desired.listeners)) { - if (sender.isDestroyed()) { - desired.listeners.delete(senderId) - } - } - if (desired.listeners.size === 0) { - desiredRemoteWatchers.delete(key) - continue - } - - // Why: drop the entry the dead transport left behind first — installRemoteWatcher treats an - // existing entry as already-installed and would hand back a watcher that can never fire again. - const stale = remoteWatchers.get(key) - if (stale) { - remoteWatchers.delete(key) - stale.batch.close() - try { - stale.unwatch() - } catch { - // Why: the handle belongs to the replaced transport; failing to close it is expected. - } - } - const retryTimer = pendingRemoteWatcherRetries.get(key) - if (retryTimer) { - clearTimeout(retryTimer) - pendingRemoteWatcherRetries.delete(key) - pendingRemoteWatcherRetryListeners.delete(key) - } - // Why: a pending watch belongs to the replaced transport; joiners must retry on the new provider. - const inFlight = inFlightRemoteInstalls.get(key) - if (inFlight) { - inFlight.listeners.clear() - inFlight.cancelled = true - inFlight.abortController.abort() - } - // Why: this reinstall supersedes the pending backoff; leaving it armed double-installs the key. - clearDormantRemoteWatcher(key) - loggedUnavailableRemoteWatchers.delete(key) - - const listeners = Array.from(desired.listeners.values()) - void Promise.all( - listeners.map((listener) => - installRemoteWatcher(listener, desired.connectionId, desired.worktreePath) - ) - ) - .then((results) => { - // Why: events between the transport dropping and this reinstall are gone for good. - requestRemoteWatcherResync( - key, - desired.worktreePath, - listeners.filter((_, index) => results[index] === 'installed') - ) - if (results.some((result) => result === 'unavailable')) { - for (const listener of listeners) { - scheduleRemoteWatcherRetry( - listener, - desired.connectionId, - desired.worktreePath, - Date.now(), - true - ) - } - } - }) - .catch((error: unknown) => { - if (isWatcherRemovalInProgressError(error)) { - return - } - for (const listener of listeners) { - scheduleRemoteWatcherRetry( - listener, - desired.connectionId, - desired.worktreePath, - Date.now(), - true - ) - } - }) - } -} - -/** Tear down all watchers on app shutdown. */ -export async function closeAllWatchers(): Promise { - // Why: drop the intent with the rest of the state, but keep the provider-registration - // subscription — a new fs:watchWorktree reopens the subsystem and still needs the re-arm hook. - desiredRemoteWatchers.clear() - senderCleanupRegistered.clear() - unwatchableRoots.clear() - suspendedLocalWatcherListeners.clear() - suspendedRemoteWatcherListeners.clear() - for (const retry of pendingLocalCapacityRetries.values()) { - retry.cancelWait() - } - pendingLocalCapacityRetries.clear() - - // Cancel any pending grace-period teardowns — we're tearing down everything. - for (const timer of pendingTeardowns.values()) { - clearTimeout(timer) - } - pendingTeardowns.clear() - - for (const timer of pendingRemoteWatcherRetries.values()) { - clearTimeout(timer) - } - pendingRemoteWatcherRetries.clear() - pendingRemoteWatcherRetryListeners.clear() - for (const state of remoteWatcherResyncStates.values()) { - if (state.timer) { - clearTimeout(state.timer) - } - } - remoteWatcherResyncStates.clear() - for (const dormant of dormantRemoteWatchers.values()) { - clearTimeout(dormant.timer) - } - dormantRemoteWatchers.clear() - loggedUnavailableRemoteWatchers.clear() - // Why: latch both subsystems shut so late installs can't register; generation bumps reject older-lifecycle waiters. - remoteWatchersClosed = true - remoteWatcherLifecycleGeneration += 1 - localWatchersClosed = true - localWatcherLifecycleGeneration += 1 - pendingRemoteInstallPromises.clear() - // Why: cancel in-flight provider.watch() calls so their resolved unwatch handles aren't installed post-shutdown. - for (const token of inFlightRemoteInstalls.values()) { - token.listeners.clear() - token.cancelled = true - token.abortController.abort() - } - for (const token of inFlightLocalInstalls.values()) { - token.listeners.clear() - token.cancelled = true - token.abortController.abort() - } - - for (const [rootKey, root] of watchedRoots) { - if (root.batch.timer) { - clearTimeout(root.batch.timer) - } - await trackLocalUnsubscribe(rootKey, root).catch(() => undefined) - } - watchedRoots.clear() - await Promise.allSettled(Array.from(pendingLocalUnsubscribes)) - failedLocalUnsubscribes.clear() - // Why: kill the forked watcher process instead of watcher.node's crash-prone async teardown; process death frees native handles. - disposeWatcherProcess() - - // Why: remote watchers are separate from local @parcel/watcher subs; unwatch here or the relay keeps polling FS after shutdown. - for (const [key, state] of remoteWatchers) { - state.batch.close() - try { - state.unwatch() - } catch (err) { - console.error(`[filesystem-watcher] remote unwatch error for ${key}:`, err) - } - } - remoteWatchers.clear() -} - /** The desktop binding for {@link WorktreeWatcherRemoval}. Installed during startup. */ export const desktopWorktreeWatcherRemoval: WorktreeWatcherRemoval = { closeLocal: (worktreePath, deadline) =>