mirror of
https://github.com/stablyai/orca.git
synced 2026-10-26 00:03:55 +00:00
Split filesystem watcher responsibilities
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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<void> => {
|
||||
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)
|
||||
}
|
||||
)
|
||||
}
|
||||
@@ -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<number, WebContents>
|
||||
abortController: AbortController
|
||||
}
|
||||
|
||||
export type LocalWatcherInstallResult = 'installed' | 'unavailable' | 'capacity' | 'cancelled'
|
||||
|
||||
export type LocalWatcherCapacityRetry = {
|
||||
listeners: Map<number, WebContents>
|
||||
cancelWait: () => void
|
||||
}
|
||||
|
||||
export type RemoteWatcherInstallToken = {
|
||||
cancelled: boolean
|
||||
listeners: Map<number, WebContents>
|
||||
abortController: AbortController
|
||||
abortScheduled: boolean
|
||||
terminalError?: Error
|
||||
}
|
||||
|
||||
export type RemoteWatcherState = {
|
||||
unwatch: () => void
|
||||
listeners: Map<number, WebContents>
|
||||
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<number, WebContents>
|
||||
timer?: ReturnType<typeof setTimeout>
|
||||
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<string, WatchedRoot>(),
|
||||
unwatchableRoots: new Set<string>(),
|
||||
// Why: key cleanup by sender WebContents (not per root) to avoid MaxListeners warnings when a workspace has many worktrees open.
|
||||
senderCleanupRegistered: new Set<number>(),
|
||||
pendingTeardowns: new Map<string, ReturnType<typeof setTimeout>>(),
|
||||
// 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<Promise<void>>(),
|
||||
pendingLocalUnsubscribesByRoot: new Map<string, Set<Promise<void>>>(),
|
||||
suspendedLocalWatcherListeners: new Map<
|
||||
string,
|
||||
{ worktreePath: string; listeners: Map<number, WebContents> }
|
||||
>(),
|
||||
// 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<string, unknown>(),
|
||||
// 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<Promise<void>>(),
|
||||
// Why: watcher creation is async; concurrent watch requests for the same root must share one install or later resolves orphan listeners.
|
||||
inFlightLocalInstalls: new Map<string, LocalWatcherInstallToken>(),
|
||||
pendingLocalInstallPromises: new Map<string, Promise<LocalWatcherInstallResult>>(),
|
||||
pendingLocalCapacityRetries: new Map<string, LocalWatcherCapacityRetry>(),
|
||||
// Remote watcher state
|
||||
// Key: `${connectionId}:${worktreePath}`, Value: shared remote watch state.
|
||||
remoteWatchers: new Map<string, RemoteWatcherState>(),
|
||||
suspendedRemoteWatcherListeners: new Map<
|
||||
string,
|
||||
{ connectionId: string; worktreePath: string; listeners: Map<number, WebContents> }
|
||||
>(),
|
||||
// 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<number, WebContents> }
|
||||
>(),
|
||||
// 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<typeof setTimeout> }
|
||||
>(),
|
||||
loggedUnavailableRemoteWatchers: new Set<string>(),
|
||||
pendingRemoteWatcherRetries: new Map<string, ReturnType<typeof setTimeout>>(),
|
||||
pendingRemoteWatcherRetryListeners: new Map<
|
||||
string,
|
||||
{ listeners: Map<number, WebContents>; startedAt: number; resyncOnInstall: boolean }
|
||||
>(),
|
||||
remoteWatcherResyncStates: new Map<string, RemoteWatcherResyncState>(),
|
||||
// Why: last-listener cleanup aborts relay setup; late success is unwatched rather than installed after the renderer stopped watching.
|
||||
inFlightRemoteInstalls: new Map<string, RemoteWatcherInstallToken>(),
|
||||
// Why: dedupe concurrent installRemoteWatcher calls per key so overlapping watches share one watcher instead of clobbering per-key state.
|
||||
pendingRemoteInstallPromises: new Map<string, Promise<RemoteWatcherInstallResult>>(),
|
||||
// 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
|
||||
}
|
||||
@@ -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<void> {
|
||||
const rootUnsubscribes =
|
||||
watcherLifecycleState.pendingLocalUnsubscribesByRoot.get(rootKey) ?? new Set<Promise<void>>()
|
||||
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<Promise<void>>): 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)
|
||||
}
|
||||
}
|
||||
@@ -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<void>
|
||||
|
||||
export function scheduleLocalCapacityRetry(
|
||||
rootKey: string,
|
||||
worktreePath: string,
|
||||
listeners: Map<number, WebContents>,
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -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<string, { type: 'create' | 'update' | 'delete'; index: number }>()
|
||||
const deleteBeforeCreate = new Set<string>()
|
||||
|
||||
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<boolean | undefined> {
|
||||
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<void> {
|
||||
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<WatchedRoot> {
|
||||
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
|
||||
}
|
||||
@@ -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<number, Electron.WebContents>) => void
|
||||
): Promise<LocalWatcherInstallResult> {
|
||||
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'
|
||||
}
|
||||
@@ -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<void> {
|
||||
const { key: rootKey } = getLocalWatcherRoot(worktreePath)
|
||||
const suspended = watcherLifecycleState.suspendedLocalWatcherListeners.get(rootKey) ?? {
|
||||
worktreePath,
|
||||
listeners: new Map<number, WebContents>()
|
||||
}
|
||||
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<void> {
|
||||
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<number, WebContents>()
|
||||
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)
|
||||
}
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -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)])
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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<number, WebContents>()
|
||||
}
|
||||
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
|
||||
)
|
||||
}
|
||||
@@ -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<void> {
|
||||
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<ReturnType<InstallRemoteWatcher>>[]
|
||||
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)
|
||||
}
|
||||
@@ -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<RemoteWatcherInstallResult> {
|
||||
// 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<RemoteWatcherInstallResult> {
|
||||
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<ReturnType<typeof getSshFilesystemProvider>>,
|
||||
key: string,
|
||||
connectionId: string,
|
||||
worktreePath: string,
|
||||
cancelToken: RemoteWatcherInstallToken,
|
||||
onTerminalError: RemoteWatcherTerminalErrorHandler
|
||||
): Promise<RemoteWatcherInstallResult> {
|
||||
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'
|
||||
}
|
||||
@@ -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
|
||||
)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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<void> {
|
||||
const key = getRemoteWatcherKey(connectionId, worktreePath)
|
||||
const suspended = watcherLifecycleState.suspendedRemoteWatcherListeners.get(key) ?? {
|
||||
connectionId,
|
||||
worktreePath,
|
||||
listeners: new Map<number, WebContents>()
|
||||
}
|
||||
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<void> {
|
||||
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)
|
||||
}
|
||||
@@ -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<WebContents>
|
||||
): void {
|
||||
const state = watcherLifecycleState.remoteWatcherResyncStates.get(key) ?? {
|
||||
lastSentAt: Number.NEGATIVE_INFINITY,
|
||||
listeners: new Map<number, WebContents>(),
|
||||
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?.()
|
||||
}
|
||||
}
|
||||
@@ -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<RemoteWatcherInstallResult>
|
||||
|
||||
export type RequestRemoteWatcherResync = (
|
||||
key: string,
|
||||
worktreePath: string,
|
||||
listeners: Iterable<WebContents>
|
||||
) => 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)
|
||||
}
|
||||
@@ -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<void> {
|
||||
// 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()
|
||||
}
|
||||
+19
-1959
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user