diff --git a/src/main/ipc/worktree-git-common-narrow-watch.ts b/src/main/ipc/worktree-git-common-narrow-watch.ts new file mode 100644 index 00000000000..a0cbd9c670b --- /dev/null +++ b/src/main/ipc/worktree-git-common-narrow-watch.ts @@ -0,0 +1,272 @@ +import { stat } from 'node:fs/promises' +import { join } from 'node:path' +import { subscribeViaWatcherProcess } from './parcel-watcher-process' +import { isWatcherProcessFailure } from './parcel-watcher-process-failure' +import type { WorktreeBaseWatchTarget } from './worktree-base-directory-event-filter' +import type { + WorktreeBasePollEvent, + WorktreeBaseSubscription, + WorktreePollerWindowVisibility +} from './worktree-base-directory-poller' +import { createGitCommonWatchReconciliation } from './worktree-git-common-watch-reconciliation' +import { startGitCommonPolling } from './worktree-git-common-polling' + +// The native stream is hosted in the crash-isolated watcher child, never the +// Electron main process: watcher.node teardown races heap-corrupt the hosting +// process when unsubscribe overlaps in-flight callbacks (issue #8732), and +// root deletion via `git worktree prune` makes that overlap routine here. + +export async function startGitCommonNarrowWatch( + target: WorktreeBaseWatchTarget, + onEvents: (events: WorktreeBasePollEvent[]) => void, + pollIntervalMs: number, + visibility: WorktreePollerWindowVisibility, + onFullScan?: () => void, + onWatchError?: (error: Error) => void +): Promise { + const worktreesDir = join(target.path, 'worktrees') + let disposed = false + let subscription: WorktreeBaseSubscription | null = null + let existenceTimer: ReturnType | null = null + let pollingFallbackPromise: Promise | null = null + let subscribing = false + let parkedWhileHidden = false + let usingPollingFallback = false + let nativeSubscriptionGeneration = 0 + const reconciliation = createGitCommonWatchReconciliation({ + commonDirPath: target.path, + pollIntervalMs, + visibility, + canStart: () => !disposed && !usingPollingFallback && subscription !== null, + shouldKeep: () => !disposed && !usingPollingFallback, + onRootReplacement: () => { + nativeSubscriptionGeneration++ + const current = subscription + subscription = null + if (current) { + void current.unsubscribe().catch(() => {}) + } + armExistencePoll() + }, + onEvents + }) + + const stopExistencePoll = (): void => { + if (existenceTimer) { + clearInterval(existenceTimer) + existenceTimer = null + } + } + + const shouldUsePollingFallback = (error: unknown): boolean => + isWatcherProcessFailure(error) && + (error.code === 'supervisor_crash_fuse' || error.code === 'process_unavailable') + + const ensurePollingFallback = (): Promise => { + if (pollingFallbackPromise) { + return pollingFallbackPromise + } + stopExistencePoll() + usingPollingFallback = true + const pending = reconciliation + .unsubscribe() + .catch(() => {}) + .then(() => + startGitCommonPolling(target.path, onEvents, pollIntervalMs, visibility, onFullScan, false) + ) + .then(async (fallback) => { + if (disposed || subscription) { + await fallback.unsubscribe() + return + } + subscription = fallback + }) + const tracked = pending.finally(() => { + if (pollingFallbackPromise === tracked) { + pollingFallbackPromise = null + } + }) + pollingFallbackPromise = tracked + return pollingFallbackPromise + } + + const tryUpgradeToNarrowWatch = async (): Promise => { + if (disposed || subscribing || subscription) { + return + } + subscribing = true + try { + const installed = await trySubscribe() + if (installed && !disposed) { + stopExistencePoll() + await reconciliation.ensureStarted() + // The dir appearing means a first linked worktree was just + // registered; surface it so the repo's worktree list refreshes. + onEvents([{ type: 'create', path: worktreesDir }]) + } + } finally { + subscribing = false + } + } + + const armExistencePoll = (): void => { + if (disposed || existenceTimer || subscription) { + return + } + if (!visibility.isWindowVisible()) { + parkedWhileHidden = true + return + } + existenceTimer = setInterval(() => { + if (disposed) { + return + } + if (!visibility.isWindowVisible()) { + parkedWhileHidden = true + stopExistencePoll() + return + } + void tryUpgradeToNarrowWatch() + }, pollIntervalMs) + existenceTimer.unref?.() + } + + const unsubscribeVisibility = visibility.onWindowBecameVisible(() => { + if (!disposed && parkedWhileHidden) { + parkedWhileHidden = false + void tryUpgradeToNarrowWatch().finally(() => { + armExistencePoll() + }) + } + reconciliation.notifyWindowBecameVisible() + }) + + const trySubscribe = async (): Promise => { + try { + const s = await stat(worktreesDir) + if (!s.isDirectory()) { + return false + } + } catch { + return false + } + const generation = ++nativeSubscriptionGeneration + let errored = false + let active = true + // Why: parcel tears its native stream down when the watched root is + // deleted (e.g. `git worktree prune` removing an empty worktrees dir) — + // sometimes surfaced as an error, sometimes as a delete event for the + // root. Either way: notify, drop the dead stream, and let the existence + // poll re-arm when a future worktree add recreates the dir. + const teardown = (): void => { + active = false + errored = true + if (generation === nativeSubscriptionGeneration) { + nativeSubscriptionGeneration++ + } + const current = subscription + subscription = null + if (current) { + void current.unsubscribe().catch(() => {}) + } + } + const teardownAndRearm = (): void => { + teardown() + armExistencePoll() + } + try { + const sub = await subscribeViaWatcherProcess( + worktreesDir, + (error, events) => { + if (disposed || !active || generation !== nativeSubscriptionGeneration) { + return + } + if (error) { + if (onWatchError) { + onWatchError(error) + } else { + onEvents([{ type: 'update', path: worktreesDir }]) + } + if (shouldUsePollingFallback(error)) { + teardown() + void ensurePollingFallback().catch(() => { + if (!disposed) { + armExistencePoll() + } + }) + } else { + teardownAndRearm() + } + return + } + if (events.length > 0) { + const rootGone = events.some( + (event) => event.type === 'delete' && event.path === worktreesDir + ) + onEvents(events.map((event) => ({ type: event.type, path: event.path }))) + if (rootGone) { + teardownAndRearm() + } + } + }, + {}, + { + // Why: a watcher-child crash drops events during the automatic + // resubscribe gap; report a structural change so worktrees re-sync. + onInterruption: () => { + if (!disposed && active && generation === nativeSubscriptionGeneration) { + if (onWatchError) { + onWatchError(new Error('Git common watcher interrupted')) + } else { + onEvents([{ type: 'update', path: worktreesDir }]) + } + } + } + } + ) + if (generation !== nativeSubscriptionGeneration) { + void sub.unsubscribe().catch(() => {}) + return false + } + if (disposed || errored) { + void sub.unsubscribe().catch(() => {}) + await pollingFallbackPromise?.catch(() => {}) + return !errored || subscription !== null + } + subscription = { unsubscribe: () => sub.unsubscribe() } + return true + } catch (error) { + if (disposed || generation !== nativeSubscriptionGeneration) { + return false + } + if (shouldUsePollingFallback(error)) { + await ensurePollingFallback() + return subscription !== null + } + return false + } + } + + if (!(await trySubscribe())) { + // Why: repos commonly start without linked worktrees; retrying the narrow + // subscription lets macOS upgrade to native events when the directory appears. + armExistencePoll() + } + await reconciliation.ensureStarted() + + return { + unsubscribe: async () => { + disposed = true + stopExistencePoll() + unsubscribeVisibility() + await pollingFallbackPromise?.catch(() => {}) + nativeSubscriptionGeneration++ + const current = subscription + subscription = null + await Promise.all([ + current?.unsubscribe().catch(() => {}), + reconciliation.unsubscribe().catch(() => {}) + ]) + } + } +} diff --git a/src/main/ipc/worktree-git-common-watch-reconciliation.ts b/src/main/ipc/worktree-git-common-watch-reconciliation.ts new file mode 100644 index 00000000000..fbf75a5e05b --- /dev/null +++ b/src/main/ipc/worktree-git-common-watch-reconciliation.ts @@ -0,0 +1,106 @@ +import { dirname, join } from 'node:path' +import type { + WorktreeBasePollEvent, + WorktreeBaseSubscription, + WorktreePollerWindowVisibility +} from './worktree-base-directory-poller' +import { startGitCommonPolling } from './worktree-git-common-polling' + +// The native stream is still the fast path. A scheduled 15-tick reconciliation +// bounds silent watcher loss at the existing 30-second backstop without joining +// the per-repo 2-second timer fleet. +const NARROW_WATCH_RECONCILIATION_TICKS = 15 + +type GitCommonWatchReconciliationOptions = { + commonDirPath: string + pollIntervalMs: number + visibility: WorktreePollerWindowVisibility + canStart: () => boolean + shouldKeep: () => boolean + onRootReplacement: () => void + onEvents: (events: WorktreeBasePollEvent[]) => void +} + +type GitCommonWatchReconciliation = { + ensureStarted: () => Promise + notifyWindowBecameVisible: () => void + unsubscribe: () => Promise +} + +export function createGitCommonWatchReconciliation({ + commonDirPath, + pollIntervalMs, + visibility, + canStart, + shouldKeep, + onRootReplacement, + onEvents +}: GitCommonWatchReconciliationOptions): GitCommonWatchReconciliation { + const worktreesDir = join(commonDirPath, 'worktrees') + let subscription: WorktreeBaseSubscription | null = null + const visibilityListeners = new Set<() => void>() + const pollVisibility: WorktreePollerWindowVisibility = { + isWindowVisible: visibility.isWindowVisible, + onWindowBecameVisible: (listener) => { + visibilityListeners.add(listener) + return () => { + visibilityListeners.delete(listener) + } + } + } + + return { + ensureStarted: async () => { + if (subscription || !canStart()) { + return + } + const reconciliation = await startGitCommonPolling( + commonDirPath, + (events) => { + const rootWasReplaced = + events.some((event) => event.type === 'delete' && event.path === worktreesDir) && + events.some((event) => event.type === 'create' && event.path === worktreesDir) + const coarseRootReplacement = events.some( + (event) => + event.type === 'create' && + event.path !== worktreesDir && + dirname(event.path) === worktreesDir + ) + if (rootWasReplaced || coarseRootReplacement) { + onRootReplacement() + } + onEvents( + coarseRootReplacement + ? events.map((event) => + event.type === 'update' && event.path === worktreesDir + ? { ...event, type: 'create' } + : event + ) + : events + ) + }, + pollIntervalMs * NARROW_WATCH_RECONCILIATION_TICKS, + pollVisibility, + undefined, + false, + () => [], + { forceFullScanEveryTick: true } + ) + if (!shouldKeep()) { + await reconciliation.unsubscribe() + } else { + subscription = reconciliation + } + }, + notifyWindowBecameVisible: () => { + for (const listener of visibilityListeners) { + listener() + } + }, + unsubscribe: async () => { + const current = subscription + subscription = null + await current?.unsubscribe() + } + } +} diff --git a/src/main/ipc/worktree-git-common-watch.test.ts b/src/main/ipc/worktree-git-common-watch.test.ts index 88ed8474546..1e6f4bb1938 100644 --- a/src/main/ipc/worktree-git-common-watch.test.ts +++ b/src/main/ipc/worktree-git-common-watch.test.ts @@ -37,6 +37,9 @@ vi.mock('node:fs/promises', async (importOriginal) => { }) const POLL_MS = 25 +// Native re-arm is intentionally deferred to the 15-tick reconciliation. Give +// loaded CI shards enough wall-clock slack without changing the exact call count. +const RECONCILIATION_WAIT_TIMEOUT_MS = 5_000 const alwaysVisible: WorktreePollerWindowVisibility = { isWindowVisible: () => true, @@ -522,9 +525,12 @@ describe('worktree git-common narrow watch (darwin)', () => { await rm(worktreesDir, { recursive: true }) await mkdir(retainedEntry, { recursive: true }) - await vi.waitFor(() => { - expect(subscribeMock).toHaveBeenCalledTimes(2) - }) + await vi.waitFor( + () => { + expect(subscribeMock).toHaveBeenCalledTimes(2) + }, + { timeout: RECONCILIATION_WAIT_TIMEOUT_MS } + ) expect(staleSubscription.unsubscribe).toHaveBeenCalledOnce() const beforeStaleEvent = received.length @@ -543,9 +549,7 @@ describe('worktree git-common narrow watch (darwin)', () => { subscribeMock.mockImplementation(async (dir, callback, _opts, hooks = {}) => { const unsubscribe = vi.fn(async () => {}) childSubscriptions.push({ dir, callback, hooks, unsubscribe }) - return childSubscriptions.length === 2 - ? deferredSubscribe.promise - : { unsubscribe } + return childSubscriptions.length === 2 ? deferredSubscribe.promise : { unsubscribe } }) const commonDir = await makeCommonDir(true) const worktreesDir = join(commonDir, 'worktrees') @@ -557,20 +561,24 @@ describe('worktree git-common narrow watch (darwin)', () => { await rm(worktreesDir, { recursive: true }) await mkdir(retainedEntry, { recursive: true }) - await vi.waitFor(() => { - expect(subscribeMock).toHaveBeenCalledTimes(2) - }) + await vi.waitFor( + () => { + expect(subscribeMock).toHaveBeenCalledTimes(2) + }, + { timeout: RECONCILIATION_WAIT_TIMEOUT_MS } + ) const stalePendingSubscription = childSubscriptions[1] await rm(worktreesDir, { recursive: true }) await mkdir(retainedEntry, { recursive: true }) - await vi.waitFor(() => { - expect( - received - .flat() - .filter((event) => event.type === 'create' && event.path === worktreesDir) - ).toHaveLength(2) - }, { timeout: 2_000 }) + await vi.waitFor( + () => { + expect( + received.flat().filter((event) => event.type === 'create' && event.path === worktreesDir) + ).toHaveLength(2) + }, + { timeout: RECONCILIATION_WAIT_TIMEOUT_MS } + ) const beforeStaleInterruption = received.length stalePendingSubscription.hooks.onInterruption?.() @@ -697,9 +705,12 @@ describe('worktree git-common polling gate (non-darwin)', () => { // Real filesystem stats cannot be flushed by a fake clock. Two scans must // arrive before the ordinary 15-tick (750ms) index backstop could fire. - await vi.waitFor(() => { - expect(fullScans.mock.calls.length).toBeGreaterThanOrEqual(2) - }, { timeout: 500 }) + await vi.waitFor( + () => { + expect(fullScans.mock.calls.length).toBeGreaterThanOrEqual(2) + }, + { timeout: 500 } + ) }) it('detects linked worktree add and remove from the every-tick readdir', async () => { diff --git a/src/main/ipc/worktree-git-common-watch.ts b/src/main/ipc/worktree-git-common-watch.ts index c0dc3f57b1d..44b51b9eea7 100644 --- a/src/main/ipc/worktree-git-common-watch.ts +++ b/src/main/ipc/worktree-git-common-watch.ts @@ -1,13 +1,10 @@ -import { stat } from 'node:fs/promises' -import { dirname, join } from 'node:path' -import { subscribeViaWatcherProcess } from './parcel-watcher-process' -import { isWatcherProcessFailure } from './parcel-watcher-process-failure' import type { WorktreeBaseWatchTarget } from './worktree-base-directory-event-filter' import type { WorktreeBasePollEvent, WorktreeBaseSubscription, WorktreePollerWindowVisibility } from './worktree-base-directory-poller' +import { startGitCommonNarrowWatch } from './worktree-git-common-narrow-watch' import { startGitCommonPolling } from './worktree-git-common-polling' import { startGitCommonPrimaryPolling } from './worktree-git-common-primary-polling' @@ -21,340 +18,6 @@ import { startGitCommonPrimaryPolling } from './worktree-git-common-primary-poll // included). Other platforms: dir-listing poll (no fseventsd to protect, and // on Windows an open directory handle on `worktrees/` could interfere with // `git worktree prune` removing it). -// The native stream is hosted in the crash-isolated watcher child, never the -// Electron main process: watcher.node teardown races heap-corrupt the hosting -// process when unsubscribe overlaps in-flight callbacks (issue #8732), and -// root deletion via `git worktree prune` makes that overlap routine here. - -// The native stream is still the fast path. A scheduled 15-tick reconciliation -// bounds silent watcher loss at the existing 30-second backstop without joining -// the per-repo 2-second timer fleet. -const NARROW_WATCH_RECONCILIATION_TICKS = 15 - -async function startGitCommonNarrowWatch( - target: WorktreeBaseWatchTarget, - onEvents: (events: WorktreeBasePollEvent[]) => void, - pollIntervalMs: number, - visibility: WorktreePollerWindowVisibility, - onFullScan?: () => void, - onWatchError?: (error: Error) => void -): Promise { - const worktreesDir = join(target.path, 'worktrees') - let disposed = false - let subscription: WorktreeBaseSubscription | null = null - let existenceTimer: ReturnType | null = null - let pollingFallbackPromise: Promise | null = null - let subscribing = false - let parkedWhileHidden = false - let reconciliationSubscription: WorktreeBaseSubscription | null = null - let usingPollingFallback = false - let nativeSubscriptionGeneration = 0 - const reconciliationVisibilityListeners = new Set<() => void>() - const reconciliationVisibility: WorktreePollerWindowVisibility = { - isWindowVisible: visibility.isWindowVisible, - onWindowBecameVisible: (listener) => { - reconciliationVisibilityListeners.add(listener) - return () => { - reconciliationVisibilityListeners.delete(listener) - } - } - } - - const stopExistencePoll = (): void => { - if (existenceTimer) { - clearInterval(existenceTimer) - existenceTimer = null - } - } - - const shouldUsePollingFallback = (error: unknown): boolean => - isWatcherProcessFailure(error) && - (error.code === 'supervisor_crash_fuse' || error.code === 'process_unavailable') - - const ensurePollingFallback = (): Promise => { - if (pollingFallbackPromise) { - return pollingFallbackPromise - } - stopExistencePoll() - usingPollingFallback = true - const previousReconciliation = reconciliationSubscription - reconciliationSubscription = null - const pending = ( - previousReconciliation - ? previousReconciliation.unsubscribe().catch(() => {}) - : Promise.resolve() - ) - .then(() => - startGitCommonPolling( - target.path, - onEvents, - pollIntervalMs, - visibility, - onFullScan, - false - ) - ) - .then(async (fallback) => { - if (disposed || subscription) { - await fallback.unsubscribe() - return - } - subscription = fallback - }) - const tracked = pending.finally(() => { - if (pollingFallbackPromise === tracked) { - pollingFallbackPromise = null - } - }) - pollingFallbackPromise = tracked - return pollingFallbackPromise - } - - // Reconciliation also repairs a stream whose root was deleted and recreated - // between scans. In that coarse race the signature is present at both samples; - // a newly listed direct child is the evidence that consumers need a root-create - // signal and the native subscription needs to be re-armed. - const ensureReconciliation = async (): Promise => { - if (disposed || usingPollingFallback || reconciliationSubscription || !subscription) { - return - } - const reconciliation = await startGitCommonPolling( - target.path, - (events) => { - const rootWasReplaced = - events.some((event) => event.type === 'delete' && event.path === worktreesDir) && - events.some((event) => event.type === 'create' && event.path === worktreesDir) - const coarseRootReplacement = events.some( - (event) => - event.type === 'create' && - event.path !== worktreesDir && - dirname(event.path) === worktreesDir - ) - if (rootWasReplaced || coarseRootReplacement) { - nativeSubscriptionGeneration++ - const current = subscription - subscription = null - if (current) { - void current.unsubscribe().catch(() => {}) - } - armExistencePoll() - } - onEvents( - coarseRootReplacement - ? events.map((event) => - event.type === 'update' && event.path === worktreesDir - ? { ...event, type: 'create' } - : event - ) - : events - ) - }, - pollIntervalMs * NARROW_WATCH_RECONCILIATION_TICKS, - reconciliationVisibility, - undefined, - false, - () => [], - { forceFullScanEveryTick: true } - ) - if (disposed || usingPollingFallback) { - await reconciliation.unsubscribe() - } else { - reconciliationSubscription = reconciliation - } - } - - const tryUpgradeToNarrowWatch = async (): Promise => { - if (disposed || subscribing || subscription) { - return - } - subscribing = true - try { - const installed = await trySubscribe() - if (installed && !disposed) { - stopExistencePoll() - await ensureReconciliation() - // The dir appearing means a first linked worktree was just - // registered; surface it so the repo's worktree list refreshes. - onEvents([{ type: 'create', path: worktreesDir }]) - } - } finally { - subscribing = false - } - } - - const armExistencePoll = (): void => { - if (disposed || existenceTimer || subscription) { - return - } - if (!visibility.isWindowVisible()) { - parkedWhileHidden = true - return - } - existenceTimer = setInterval(() => { - if (disposed) { - return - } - if (!visibility.isWindowVisible()) { - parkedWhileHidden = true - stopExistencePoll() - return - } - void tryUpgradeToNarrowWatch() - }, pollIntervalMs) - existenceTimer.unref?.() - } - - const unsubscribeVisibility = visibility.onWindowBecameVisible(() => { - if (!disposed && parkedWhileHidden) { - parkedWhileHidden = false - void tryUpgradeToNarrowWatch().finally(() => { - armExistencePoll() - }) - } - for (const listener of reconciliationVisibilityListeners) { - listener() - } - }) - - const trySubscribe = async (): Promise => { - try { - const s = await stat(worktreesDir) - if (!s.isDirectory()) { - return false - } - } catch { - return false - } - const generation = ++nativeSubscriptionGeneration - let errored = false - let active = true - // Why: parcel tears its native stream down when the watched root is - // deleted (e.g. `git worktree prune` removing an empty worktrees dir) — - // sometimes surfaced as an error, sometimes as a delete event for the - // root. Either way: notify, drop the dead stream, and let the existence - // poll re-arm when a future worktree add recreates the dir. - const teardown = (): void => { - active = false - errored = true - if (generation === nativeSubscriptionGeneration) { - nativeSubscriptionGeneration++ - } - const current = subscription - subscription = null - if (current) { - void current.unsubscribe().catch(() => {}) - } - } - const teardownAndRearm = (): void => { - teardown() - armExistencePoll() - } - try { - const sub = await subscribeViaWatcherProcess( - worktreesDir, - (error, events) => { - if ( - disposed || - !active || - generation !== nativeSubscriptionGeneration - ) { - return - } - if (error) { - if (onWatchError) { - onWatchError(error) - } else { - onEvents([{ type: 'update', path: worktreesDir }]) - } - if (shouldUsePollingFallback(error)) { - teardown() - void ensurePollingFallback().catch(() => { - if (!disposed) { - armExistencePoll() - } - }) - } else { - teardownAndRearm() - } - return - } - if (events.length > 0) { - const rootGone = events.some( - (event) => event.type === 'delete' && event.path === worktreesDir - ) - onEvents(events.map((event) => ({ type: event.type, path: event.path }))) - if (rootGone) { - teardownAndRearm() - } - } - }, - {}, - { - // Why: a watcher-child crash drops events during the automatic - // resubscribe gap; report a structural change so worktrees re-sync. - onInterruption: () => { - if ( - !disposed && - active && - generation === nativeSubscriptionGeneration - ) { - if (onWatchError) { - onWatchError(new Error('Git common watcher interrupted')) - } else { - onEvents([{ type: 'update', path: worktreesDir }]) - } - } - } - } - ) - if (generation !== nativeSubscriptionGeneration) { - void sub.unsubscribe().catch(() => {}) - return false - } - if (disposed || errored) { - void sub.unsubscribe().catch(() => {}) - await pollingFallbackPromise?.catch(() => {}) - return !errored || subscription !== null - } - subscription = { unsubscribe: () => sub.unsubscribe() } - return true - } catch (error) { - if (disposed || generation !== nativeSubscriptionGeneration) { - return false - } - if (shouldUsePollingFallback(error)) { - await ensurePollingFallback() - return subscription !== null - } - return false - } - } - - if (!(await trySubscribe())) { - // Why: repos commonly start without linked worktrees; retrying the narrow - // subscription lets macOS upgrade to native events when the directory appears. - armExistencePoll() - } - await ensureReconciliation() - - return { - unsubscribe: async () => { - disposed = true - stopExistencePoll() - unsubscribeVisibility() - await pollingFallbackPromise?.catch(() => {}) - nativeSubscriptionGeneration++ - const current = subscription - const reconciliation = reconciliationSubscription - subscription = null - reconciliationSubscription = null - await Promise.all([ - current?.unsubscribe().catch(() => {}), - reconciliation?.unsubscribe().catch(() => {}) - ]) - } - } -} export async function startGitCommonWatch( target: WorktreeBaseWatchTarget,