diff --git a/src/main/ipc/worktree-git-common-entry-snapshot.ts b/src/main/ipc/worktree-git-common-entry-snapshot.ts index 513a03e02d4..0ea028d20ce 100644 --- a/src/main/ipc/worktree-git-common-entry-snapshot.ts +++ b/src/main/ipc/worktree-git-common-entry-snapshot.ts @@ -10,12 +10,25 @@ function statSignature(value: { mtimeMs: number; ctimeMs: number; ino: number }) return `${value.mtimeMs}:${value.ctimeMs}:${value.ino}` } -export async function gitCommonDirectorySignature(path: string): Promise { +export async function gitCommonDirectorySignature( + path: string, + previousSignature = 'missing', + onMissing?: () => void +): Promise { try { const value = await stat(path) return `${statSignature(value)}:${value.size}` - } catch { - return 'missing' + } catch (error) { + if ( + typeof error === 'object' && + error !== null && + 'code' in error && + (error.code === 'ENOENT' || error.code === 'ENOTDIR') + ) { + onMissing?.() + return 'missing' + } + return previousSignature } } diff --git a/src/main/ipc/worktree-git-common-polling.ts b/src/main/ipc/worktree-git-common-polling.ts index 74486546673..306638be450 100644 --- a/src/main/ipc/worktree-git-common-polling.ts +++ b/src/main/ipc/worktree-git-common-polling.ts @@ -60,11 +60,12 @@ async function snapshotGitCommon( forceFullScan = false, statusRefPaths = new Set(), statInPlaceLeaves = false, - onDirGateMiss?: () => void + onDirGateMiss?: () => void, + onRootMissing?: () => void ): Promise { const worktreesDir = join(commonDirPath, 'worktrees') const [worktreesDirSignature, primarySignatures, statusRefSignatures] = await Promise.all([ - gitCommonDirectorySignature(worktreesDir), + gitCommonDirectorySignature(worktreesDir, previous?.worktreesDirSignature, onRootMissing), includePrimary ? snapshotPrimaryCheckoutSignatures(commonDirPath) : new Map(), snapshotStatusRefSignatures(statusRefPaths) ]) @@ -113,6 +114,8 @@ async function snapshotGitCommon( export type GitCommonPollingOptions = { /** Reconciliation directly probes in-place leaves each tick and forces a full sweep every N ticks. */ reconciliationFullScanEveryTicks?: number + /** Missing snapshots can outlive the native owner they originally retired. */ + onRootMissing?: (initial: boolean) => void } // Require misses on separate ticks before distrusting the directory gate for the whole catalog. @@ -141,7 +144,10 @@ export async function startGitCommonPolling( undefined, includePrimary, false, - new Set(getStatusRefPaths()) + new Set(getStatusRefPaths()), + false, + undefined, + () => options.onRootMissing?.(true) ) let timer: ReturnType | null = null let parkedWhileHidden = false @@ -179,6 +185,10 @@ export async function startGitCommonPolling( ? () => { dirGateMissed = true } + : undefined, + // A present-to-missing edge already retires through the normal diff. + snapshot.worktreesDirSignature === 'missing' + ? () => options.onRootMissing?.(false) : undefined ) if (disposed) { diff --git a/src/main/ipc/worktree-git-common-root-lifetime.test.ts b/src/main/ipc/worktree-git-common-root-lifetime.test.ts new file mode 100644 index 00000000000..1af4d7ca211 --- /dev/null +++ b/src/main/ipc/worktree-git-common-root-lifetime.test.ts @@ -0,0 +1,658 @@ +import { mkdir, mkdtemp, realpath, rm, writeFile } from 'node:fs/promises' +import type * as NodeFsPromises from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { subscribeViaWatcherProcess } from './filesystem/watcher-process/parcel-watcher-process' +import { WatcherProcessFailure } from './filesystem/watcher-process/parcel-watcher-process-failure' +import type { + WatcherProcessCallback, + WatcherProcessHooks +} from './filesystem/watcher-process/parcel-watcher-process-subscription' +import { startGitCommonNarrowWatch } from './worktree-git-common-narrow-watch' +import { startGitCommonPolling } from './worktree-git-common-polling' +import { handleLocalWatchEvents, type ActiveWatch } from './worktree-base-directory-watch-events' +import { + createWorktreeHeadIdentityRefreshState, + refreshWorktreeHeadIdentities +} from './worktree-head-identity-refresh' +import * as headReader from './worktree-head-identity-reader' +import { EMPTY_HEAD_IDENTITY_SCOPE } from './worktree-head-identity-scope' +import type { WorktreeBaseRepoWatchConfig } from './worktree-base-directory-event-filter' +import { WorktreeWatcherFailureRefreshCooldown } from './worktree-watcher-failure-refresh-cooldown' +import { notifyWatchedWorktreeCatalogChanged } from './watched-worktree-catalog-notification' +import { + notifyWorktreeHeadIdentitiesChanged, + notifyWorktreeGitStatusMetadataChanged +} from './worktree-remote' +import type { + WorktreeBasePollEvent, + WorktreeBaseSubscription +} from './worktree-base-directory-poller' + +type RootFilesystemControl = { + root: string + stat: (() => Promise) | null + read: (() => Promise) | null + reads: number + files: string[] + upgrade: () => Promise +} +const fsControl = vi.hoisted(() => ({ + root: '', + stat: null, + read: null, + reads: 0, + files: [], + upgrade: async () => {} +})) +vi.mock('node:fs/promises', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + readFile: async (...args: Parameters) => { + if (typeof args[0] === 'string') { + fsControl.files.push(args[0]) + } + return actual.readFile(...args) + }, + stat: async (...args: Parameters) => { + if (args[0] === fsControl.root) { + await fsControl.stat?.() + } + return actual.stat(...args) + }, + readdir: async (...args: Parameters) => { + if (args[0] === fsControl.root) { + fsControl.reads++ + await fsControl.read?.() + } + return actual.readdir(...args) + } + } +}) +vi.mock('./worktree-remote', () => ({ + notifyWorktreeGitStatusMetadataChanged: vi.fn(), + notifyWorktreeHeadIdentitiesChanged: vi.fn() +})) +vi.mock('./watched-worktree-catalog-notification', () => ({ + notifyWatchedWorktreeCatalogChanged: vi.fn() +})) +vi.mock('./filesystem/watcher-process/parcel-watcher-process', () => ({ + subscribeViaWatcherProcess: vi.fn() +})) +// Drive the production upgrade directly without advancing five minutes of fallback polling. +vi.mock('./worktree-git-common-native-upgrade-poll', () => ({ + createNativeUpgradePoll: (attempt: () => Promise) => { + fsControl.upgrade = attempt + return { arm: () => {}, stop: () => {} } + } +})) + +const POLL_MS = 10 +const RECONCILE_MS = POLL_MS * 15 +const visible = { isWindowVisible: () => true, onWindowBecameVisible: () => () => {} } +const native = vi.mocked(subscribeViaWatcherProcess) +const subscriptions: { + callback: WatcherProcessCallback + hooks: WatcherProcessHooks + unsubscribe: ReturnType Promise>> +}[] = [] +const cleanups: (() => Promise)[] = [] +let common: string +const installNative: typeof subscribeViaWatcherProcess = async ( + _path, + callback, + _options, + hooks = {} +) => { + const unsubscribe = vi.fn(async () => {}) + subscriptions.push({ callback, hooks, unsubscribe }) + return { unsubscribe } +} + +beforeEach(async () => { + common = await realpath(await mkdtemp(join(tmpdir(), 'orca-root-lifetime-'))) + fsControl.root = join(common, 'worktrees') + fsControl.stat = null + fsControl.read = null + fsControl.reads = 0 + fsControl.files = [] + vi.clearAllMocks() + await recreate() + native.mockReset() + native.mockImplementation(installNative) + vi.useFakeTimers() +}) + +afterEach(async () => { + fsControl.stat = null + fsControl.read = null + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())) + await rm(common, { recursive: true, force: true }) + subscriptions.length = 0 + vi.useRealTimers() +}) + +async function recreate(): Promise { + await mkdir(join(fsControl.root, 'same-child'), { recursive: true }) +} + +function start(events: WorktreeBasePollEvent[][], errors = vi.fn(), overflow = vi.fn()) { + return startGitCommonNarrowWatch( + { key: 'root-lifetime', kind: 'git-common', path: common, repos: new Map() }, + (batch) => events.push(batch), + POLL_MS, + 'darwin', + visible, + undefined, + errors, + overflow + ) +} + +function lateDelivery(index: number): void { + const old = subscriptions[index] + old.callback(null, [{ type: 'update', path: join(fsControl.root, 'stale') }]) + old.hooks.onInterruption?.() + old.hooks.onOverflow?.() +} + +it('fences a pending native subscribe when root deletion and creation are separate snapshots', async () => { + const pending = Promise.withResolvers() + native.mockImplementation(async (...args) => { + const subscription = await installNative(...args) + return subscriptions.length === 2 ? pending.promise : subscription + }) + const events: WorktreeBasePollEvent[][] = [] + const errors = vi.fn() + const overflow = vi.fn() + const watch = await start(events, errors, overflow) + cleanups.push(() => watch.unsubscribe()) + try { + subscriptions[0].callback(null, [{ type: 'delete', path: fsControl.root }]) + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => expect(subscriptions).toHaveLength(2)) + events.length = 0 + fsControl.stat = async () => { + fsControl.stat = null + await rm(fsControl.root, { recursive: true }) + } + fsControl.read = async () => { + fsControl.read = null + await recreate() + } + await vi.advanceTimersByTimeAsync(RECONCILE_MS) + await vi.waitFor(() => + expect(events.flat()).toContainEqual({ type: 'delete', path: fsControl.root }) + ) + expect(events.flat()).not.toContainEqual({ type: 'create', path: fsControl.root }) + const before = events.length + lateDelivery(1) + expect(events).toHaveLength(before) + expect(errors).not.toHaveBeenCalled() + expect(overflow).not.toHaveBeenCalled() + pending.resolve({ unsubscribe: subscriptions[1].unsubscribe }) + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => expect(subscriptions).toHaveLength(3)) + expect(subscriptions[1].unsubscribe).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(RECONCILE_MS) + await vi.waitFor(() => + expect(events.flat()).toContainEqual({ type: 'create', path: fsControl.root }) + ) + subscriptions[2].callback(null, [{ type: 'update', path: join(fsControl.root, 'fresh') }]) + expect(events.flat()).toContainEqual({ type: 'update', path: join(fsControl.root, 'fresh') }) + } finally { + pending.resolve({ unsubscribe: subscriptions[1].unsubscribe }) + } +}) + +it.each([ + ['startup', 'ENOENT'], + ['upgrade', 'ENOENT'], + ['startup', 'ENOTDIR'], + ['upgrade', 'ENOTDIR'], + ['startup', 'EIO'], + ['upgrade', 'EIO'] +] as const)('handles %s baseline root observation %s before recreation', async (entry, code) => { + const absent = code === 'ENOENT' || code === 'ENOTDIR' + const blocked = Promise.withResolvers() + const reachedRead = vi.fn() + native.mockImplementation(async (...args) => { + if (entry === 'upgrade' && native.mock.calls.length === 1) { + throw new WatcherProcessFailure('test fuse', 'supervisor', 'supervisor_crash_fuse') + } + const subscription = await installNative(...args) + if (subscriptions.length === 1) { + if (absent) { + await rm(fsControl.root, { recursive: true }) + } + fsControl.stat = async () => { + fsControl.stat = null + throw Object.assign(new Error('controlled root stat'), { code }) + } + fsControl.read = async () => { + reachedRead() + await blocked.promise + } + } + return subscription + }) + const events: WorktreeBasePollEvent[][] = [] + const errors = vi.fn() + const overflow = vi.fn() + const starting = start(events, errors, overflow) + let upgrading: Promise | undefined + try { + if (entry === 'upgrade') { + await starting + upgrading = fsControl.upgrade() + } + await vi.waitFor(() => expect(reachedRead).toHaveBeenCalled()) + expect(events).toEqual(absent ? [[{ type: 'delete', path: fsControl.root }]] : []) + const before = events.length + lateDelivery(0) + if (absent) { + expect(events).toHaveLength(before) + expect(errors).not.toHaveBeenCalled() + expect(overflow).not.toHaveBeenCalled() + expect(subscriptions[0].unsubscribe).toHaveBeenCalledOnce() + } else { + expect(subscriptions[0].unsubscribe).not.toHaveBeenCalled() + expect(errors).toHaveBeenCalledOnce() + expect(overflow).toHaveBeenCalledOnce() + expect(events).toHaveLength(before + 1) + } + await recreate() + blocked.resolve() + fsControl.read = null + await starting + await upgrading + await vi.advanceTimersByTimeAsync(POLL_MS) + if (absent) { + await vi.waitFor(() => expect(subscriptions).toHaveLength(2)) + } + const current = subscriptions[absent ? 1 : 0] + current.callback(null, [{ type: 'update', path: join(fsControl.root, 'fresh') }]) + expect(events.flat()).toContainEqual({ type: 'update', path: join(fsControl.root, 'fresh') }) + if (absent) { + expect(subscriptions[0].unsubscribe).toHaveBeenCalledOnce() + } + } finally { + await recreate() + blocked.resolve() + fsControl.read = null + await upgrading + await (await starting).unsubscribe() + } +}) + +it('converges when a delayed deletion snapshot retires an already fresh subscription', async () => { + const events: WorktreeBasePollEvent[][] = [] + const errors = vi.fn() + const overflow = vi.fn() + const watch = await start(events, errors, overflow) + cleanups.push(() => watch.unsubscribe()) + const blocked = Promise.withResolvers() + const reachedRead = vi.fn() + fsControl.stat = async () => { + fsControl.stat = null + await rm(fsControl.root, { recursive: true }) + } + fsControl.read = async () => { + reachedRead() + await blocked.promise + } + try { + await vi.advanceTimersByTimeAsync(RECONCILE_MS) + await vi.waitFor(() => expect(reachedRead).toHaveBeenCalled()) + subscriptions[0].callback(null, [{ type: 'delete', path: fsControl.root }]) + await recreate() + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => expect(subscriptions).toHaveLength(2)) + blocked.resolve() + fsControl.read = null + await vi.waitFor(() => expect(subscriptions[1].unsubscribe).toHaveBeenCalledOnce()) + const before = events.length + lateDelivery(1) + expect(events).toHaveLength(before) + expect(errors).not.toHaveBeenCalled() + expect(overflow).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => expect(subscriptions).toHaveLength(3)) + const reads = fsControl.reads + await vi.advanceTimersByTimeAsync(RECONCILE_MS * 3) + await vi.waitFor(() => expect(fsControl.reads).toBeGreaterThan(reads)) + expect(subscriptions).toHaveLength(3) + expect(subscriptions[0].unsubscribe).toHaveBeenCalledOnce() + expect(subscriptions[1].unsubscribe).toHaveBeenCalledOnce() + subscriptions[2].callback(null, [{ type: 'update', path: join(fsControl.root, 'fresh') }]) + expect(events.flat()).toContainEqual({ type: 'update', path: join(fsControl.root, 'fresh') }) + } finally { + blocked.resolve() + } +}) + +it.each(['EACCES', 'EIO', undefined, 'ENOENT', 'ENOTDIR'])( + 'preserves root evidence on stat error %s', + async (code) => { + const events: WorktreeBasePollEvent[][] = [] + const scanned = vi.fn() + const watch = await startGitCommonPolling( + common, + (batch) => events.push(batch), + POLL_MS, + visible, + scanned, + false, + () => [], + { reconciliationFullScanEveryTicks: 1 } + ) + cleanups.push(() => watch.unsubscribe()) + const absent = code === 'ENOENT' || code === 'ENOTDIR' + fsControl.stat = async () => { + throw Object.assign(new Error('controlled root stat'), { code }) + } + const reads = fsControl.reads + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => expect(fsControl.reads).toBeGreaterThan(reads)) + await vi.waitFor(() => expect(scanned).toHaveBeenCalled()) + expect( + events.flat().some((event) => event.type === 'delete' && event.path === fsControl.root) + ).toBe(absent) + fsControl.stat = null + const afterError = join(fsControl.root, 'after-error') + await mkdir(afterError) + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => + expect(events.flat()).toContainEqual({ type: 'create', path: afterError }) + ) + } +) + +it.each(['unknown baseline', 'reattachment'] as const)( + 'fences confirmed absence after %s leaves a missing root snapshot', + async (schedule) => { + native.mockImplementation(async (...args) => { + const subscription = await installNative(...args) + if (subscriptions.length === 1) { + fsControl.stat = async () => { + fsControl.stat = null + if (schedule === 'unknown baseline') { + throw Object.assign(new Error('controlled baseline root stat'), { code: 'EIO' }) + } + await rm(fsControl.root, { recursive: true }) + } + if (schedule === 'reattachment') { + fsControl.read = async () => { + fsControl.read = null + await recreate() + } + } + } + return subscription + }) + const events: WorktreeBasePollEvent[][] = [] + const errors = vi.fn() + const overflow = vi.fn() + const watch = await start(events, errors, overflow) + cleanups.push(() => watch.unsubscribe()) + const owner = schedule === 'unknown baseline' ? 0 : 1 + if (schedule === 'unknown baseline') { + expect(subscriptions[0].unsubscribe).not.toHaveBeenCalled() + expect(events).toHaveLength(0) + } else { + expect(subscriptions[0].unsubscribe).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => expect(subscriptions).toHaveLength(2), { + interval: 1, + timeout: RECONCILE_MS / 2 + }) + } + expect(fsControl.reads).toBe(1) + subscriptions[owner].callback(null, [ + { type: 'update', path: join(fsControl.root, 'before-loss') } + ]) + expect(events.flat()).toContainEqual({ + type: 'update', + path: join(fsControl.root, 'before-loss') + }) + const beforeLoss = events.length + await rm(fsControl.root, { recursive: true }) + // Drop native deletion; a child-delete batch proves the missing-to-missing snapshot settled. + await vi.advanceTimersByTimeAsync(RECONCILE_MS) + await vi.waitFor(() => + expect(events.slice(beforeLoss).flat()).toContainEqual({ + type: 'delete', + path: join(fsControl.root, 'same-child') + }) + ) + const beforeLate = events.length + lateDelivery(owner) + expect.soft(events).toHaveLength(beforeLate) + expect.soft(errors).not.toHaveBeenCalled() + expect.soft(overflow).not.toHaveBeenCalled() + expect.soft(subscriptions[owner].unsubscribe).toHaveBeenCalledOnce() + await recreate() + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => expect(subscriptions).toHaveLength(owner + 2), { + interval: 1, + timeout: RECONCILE_MS / 2 + }) + subscriptions[owner + 1].callback(null, [ + { type: 'update', path: join(fsControl.root, 'fresh-after-loss') } + ]) + expect(events.flat()).toContainEqual({ + type: 'update', + path: join(fsControl.root, 'fresh-after-loss') + }) + expect(subscriptions[owner].unsubscribe).toHaveBeenCalledOnce() + } +) + +it('keeps an ordinary polling baseline silent when the root is absent', async () => { + await rm(fsControl.root, { recursive: true }) + const events = vi.fn() + const watch = await startGitCommonPolling(common, events, POLL_MS, visible) + cleanups.push(() => watch.unsubscribe()) + expect(events).not.toHaveBeenCalled() +}) + +it.each([ + [4, 4, 'ENOENT'], + [40, 4, 'ENOENT'], + [4, 40, 'ENOENT'], + [4, 4, 'ENOTDIR'], + [40, 4, 'ENOTDIR'], + [4, 40, 'ENOTDIR'] +] as const)( + 'bounds upgrade absence fanout with %i repos and %i entries (%s)', + async (repoCount, entryCount, code) => { + const makeConsumer = (path: string, count: number): ActiveWatch => ({ + key: path, + kind: 'git-common', + path, + repos: new Map( + Array.from({ length: count }, (_, i): [string, WorktreeBaseRepoWatchConfig] => { + const repoId = `${path}:${i}` + return [repoId, { repoId, repoName: `repo-${i}`, nestWorkspaces: false }] + }) + ), + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the consumer only calls isDestroyed. + mainWindow: { isDestroyed: () => false } as never, + subscription: { unsubscribe: async () => {} }, + notifyTimer: null, + pendingStructureRepoIds: new Set(), + pendingGitStatusRepoIds: new Set(), + pendingHeadIdentityRepoIds: new Set(), + pendingHeadIdentityScope: EMPTY_HEAD_IDENTITY_SCOPE, + pendingGitStatusScope: EMPTY_HEAD_IDENTITY_SCOPE, + headIdentityRefresh: createWorktreeHeadIdentityRefreshState(), + gitStatusRefPaths: new Set(), + watcherFailureRefresh: new WorktreeWatcherFailureRefreshCooldown(), + disposed: false + }) + const seedEntries = async () => { + await rm(fsControl.root, { recursive: true, force: true }) + await Promise.all( + Array.from({ length: entryCount }, async (_, i) => { + const entry = join(fsControl.root, `entry-${i}`) + await mkdir(entry, { recursive: true }) + await writeFile(join(entry, 'HEAD'), 'a'.repeat(40)) + await writeFile(join(entry, 'gitdir'), join(common, `checkout-${i}`, '.git')) + }) + ) + } + await seedEntries() + const consumer = makeConsumer(common, repoCount) + const unrelated = makeConsumer(join(common, 'unrelated'), 1) + const events: WorktreeBasePollEvent[][] = [] + const errors = vi.fn() + const overflow = vi.fn() + const reader = vi.spyOn(headReader, 'readGitCommonHeadIdentities') + const catalog = vi.mocked(notifyWatchedWorktreeCatalogChanged) + const counts = () => ({ + batches: events.length, + catalogs: catalog.mock.calls.length, + heads: reader.mock.calls.length, + files: fsControl.files.length, + statuses: vi.mocked(notifyWorktreeGitStatusMetadataChanged).mock.calls.length, + identities: vi.mocked(notifyWorktreeHeadIdentitiesChanged).mock.calls.length + }) + const blocked = Promise.withResolvers() + const reachedRead = vi.fn() + native.mockRejectedValueOnce( + new WatcherProcessFailure('test fuse', 'supervisor', 'supervisor_crash_fuse') + ) + consumer.subscription = await startGitCommonNarrowWatch( + consumer, + (batch) => { + events.push(batch) + handleLocalWatchEvents(consumer, null, batch, () => [consumer, unrelated]) + }, + POLL_MS, + 'darwin', + visible, + undefined, + errors, + overflow + ) + cleanups.push(() => consumer.subscription.unsubscribe()) + await refreshWorktreeHeadIdentities(consumer, consumer.headIdentityRefresh, false) + native.mockImplementation(async (...args) => { + const installed = await installNative(...args) + await rm(fsControl.root, { recursive: true }) + fsControl.stat = async () => { + fsControl.stat = null + throw Object.assign(new Error('controlled root absence'), { code }) + } + fsControl.read = async () => { + reachedRead() + await blocked.promise + } + return installed + }) + const before = counts() + const upgrading = fsControl.upgrade() + try { + await vi.waitFor(() => expect(reachedRead).toHaveBeenCalledOnce()) + expect(events.slice(before.batches)).toEqual([[{ type: 'delete', path: fsControl.root }]]) + expect(subscriptions[0].unsubscribe).toHaveBeenCalledOnce() + // Keep the real baseline blocked, but allow the real head reader its own listing. + fsControl.read = null + await seedEntries() + const reads = fsControl.reads + await vi.advanceTimersByTimeAsync(250) + await vi.waitFor(() => expect(consumer.headIdentityRefresh.inFlight).toBe(false)) + expect( + catalog.mock.calls + .slice(before.catalogs) + .map(([, id]) => id) + .sort() + ).toEqual([...consumer.repos.keys()].sort()) + expect(reader.mock.calls.slice(before.heads)).toHaveLength(1) + expect(reader.mock.calls.at(-1)?.[0]).toBe(common) + expect(reader.mock.calls.at(-1)?.[2]?.all).toBe(true) + expect(fsControl.reads - reads).toBe(1) + expect(fsControl.files.slice(before.files)).toHaveLength(entryCount * 2) + expect(consumer.headIdentityRefresh.baseline?.size).toBe(entryCount) + expect(unrelated.headIdentityRefresh.baseline).toBeNull() + const beforeLate = counts() + lateDelivery(0) + await vi.advanceTimersByTimeAsync(250) + expect(counts()).toEqual(beforeLate) + expect(errors).not.toHaveBeenCalled() + expect(overflow).not.toHaveBeenCalled() + blocked.resolve() + await upgrading + // Upgrade resync and the two root-return signals may flush separately. + native.mockImplementation(installNative) + await vi.advanceTimersByTimeAsync(300) + await vi.waitFor(() => expect(subscriptions).toHaveLength(2)) + await vi.waitFor(() => + expect( + events.flat().filter((event) => event.type === 'create' && event.path === fsControl.root) + ).toHaveLength(2) + ) + await vi.advanceTimersByTimeAsync(300) + await vi.waitFor(() => expect(consumer.headIdentityRefresh.inFlight).toBe(false)) + expect(consumer.notifyTimer).toBeNull() + expect(catalog.mock.calls.length - beforeLate.catalogs).toBeGreaterThanOrEqual(repoCount) + expect(catalog.mock.calls.length - beforeLate.catalogs).toBeLessThanOrEqual(3 * repoCount) + expect(reader.mock.calls.length - beforeLate.heads).toBeGreaterThanOrEqual(1) + expect(reader.mock.calls.length - beforeLate.heads).toBeLessThanOrEqual(3) + expect(subscriptions[0].unsubscribe).toHaveBeenCalledOnce() + // Real later loss gets one diff notification, then missing-to-missing ticks stay quiet. + const beforeLoss = counts() + await rm(fsControl.root, { recursive: true }) + await vi.advanceTimersByTimeAsync(RECONCILE_MS) + await vi.waitFor(() => expect(subscriptions[1].unsubscribe).toHaveBeenCalledOnce()) + await vi.advanceTimersByTimeAsync(250) + await vi.waitFor(() => expect(consumer.headIdentityRefresh.baseline?.size).toBe(0)) + const absent = counts() + expect(absent.catalogs - beforeLoss.catalogs).toBe(repoCount) + expect(absent.heads - beforeLoss.heads).toBe(1) + expect( + events + .slice(beforeLoss.batches) + .flat() + .filter((event) => event.path === fsControl.root && event.type === 'delete') + ).toHaveLength(1) + lateDelivery(1) + await vi.advanceTimersByTimeAsync(RECONCILE_MS * 4) + expect(counts()).toEqual(absent) + expect(errors).not.toHaveBeenCalled() + expect(overflow).not.toHaveBeenCalled() + await seedEntries() + await vi.advanceTimersByTimeAsync(POLL_MS) + await vi.waitFor(() => expect(subscriptions).toHaveLength(3)) + await writeFile(join(fsControl.root, 'entry-0', 'HEAD'), 'b'.repeat(40)) + subscriptions[2].callback(null, [ + { type: 'update', path: join(fsControl.root, 'entry-0', 'HEAD') } + ]) + await vi.advanceTimersByTimeAsync(300) + await vi.waitFor(() => expect(consumer.headIdentityRefresh.inFlight).toBe(false)) + expect(catalog.mock.calls.length - absent.catalogs).toBe(repoCount) + expect(reader.mock.calls.length).toBeGreaterThan(absent.heads) + expect(consumer.headIdentityRefresh.baseline?.get(join(common, 'checkout-0'))).toBe( + `${'b'.repeat(40)} ` + ) + expect(catalog.mock.calls.every(([, id]) => consumer.repos.has(id))).toBe(true) + await consumer.subscription.unsubscribe() + consumer.disposed = true + const stopped = counts() + lateDelivery(2) + await vi.advanceTimersByTimeAsync(RECONCILE_MS * 4) + expect(counts()).toEqual(stopped) + expect(subscriptions[1].unsubscribe).toHaveBeenCalledOnce() + expect(subscriptions[2].unsubscribe).toHaveBeenCalledOnce() + } finally { + fsControl.read = null + blocked.resolve() + await upgrading + consumer.disposed = true + clearTimeout(consumer.notifyTimer ?? undefined) + reader.mockRestore() + } + } +) diff --git a/src/main/ipc/worktree-git-common-watch-reconciliation.ts b/src/main/ipc/worktree-git-common-watch-reconciliation.ts index 8fa93be51b0..5aa52bf1f9f 100644 --- a/src/main/ipc/worktree-git-common-watch-reconciliation.ts +++ b/src/main/ipc/worktree-git-common-watch-reconciliation.ts @@ -52,6 +52,36 @@ export function createGitCommonWatchReconciliation({ } } + const handleEvents = (events: WorktreeBasePollEvent[]): void => { + const rootDeleted = events.some( + (event) => event.type === 'delete' && event.path === worktreesDir + ) + // Ordinary child creates repeat native events; delayed root loss may conservatively rearm a fresh stream. + const rootRecreated = events.some( + (event) => event.type === 'create' && event.path === worktreesDir + ) + const coarseRootReplacement = + rootRecreated && + events.some( + (event) => + event.type === 'create' && + event.path !== worktreesDir && + dirname(event.path) === worktreesDir + ) + if (rootDeleted || coarseRootReplacement) { + onRootReplacement() + } + onEvents( + coarseRootReplacement + ? events.map((event) => + event.type === 'update' && event.path === worktreesDir + ? { ...event, type: 'create' } + : event + ) + : events + ) + } + return { ensureStarted: () => { if (subscription || !canStart()) { @@ -59,45 +89,29 @@ export function createGitCommonWatchReconciliation({ } return starting.run(async () => { const startGeneration = generation + const shouldKeepBaseline = (): boolean => startGeneration === generation && shouldKeep() 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) - // Delayed reconciliation repeats native creates; only recreated roots justify replacing the healthy stream. - const rootRecreated = events.some( - (event) => event.type === 'create' && event.path === worktreesDir - ) - const coarseRootReplacement = - rootRecreated && - 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 - ) - }, + handleEvents, pollIntervalMs * NARROW_WATCH_RECONCILIATION_TICKS, pollVisibility, undefined, false, () => [], - { reconciliationFullScanEveryTicks: RECONCILIATION_FULL_SCAN_EVERY_TICKS } + { + reconciliationFullScanEveryTicks: RECONCILIATION_FULL_SCAN_EVERY_TICKS, + onRootMissing: (initial) => { + if (shouldKeepBaseline()) { + if (initial) { + handleEvents([{ type: 'delete', path: worktreesDir }]) + } else { + onRootReplacement() + } + } + } + } ) - if (startGeneration !== generation || !shouldKeep()) { + if (!shouldKeepBaseline()) { await reconciliation.unsubscribe() } else { subscription = reconciliation diff --git a/src/main/ipc/worktree-git-common-watch.test.ts b/src/main/ipc/worktree-git-common-watch.test.ts index 73772794658..a48c40877d9 100644 --- a/src/main/ipc/worktree-git-common-watch.test.ts +++ b/src/main/ipc/worktree-git-common-watch.test.ts @@ -767,16 +767,16 @@ describe('worktree git-common narrow watch (local native platforms)', () => { { timeout: 1_000 } ) const stalePendingSubscription = narrowSubscriptions()[1] + const beforeReplacement = received.length await replaceWorktreesRoot(commonDir, worktreesDir, retainedEntry) await vi.advanceTimersByTimeAsync(POLL_MS * RECONCILIATION_TICKS * 4) // Timer advancement starts the replacement check; wait separately for its filesystem work. await vi.waitFor( () => { - expect( - received - .flat() - .filter((event) => event.type === 'create' && event.path === worktreesDir) - ).toHaveLength(2) + expect(received.slice(beforeReplacement).flat()).toContainEqual({ + type: 'delete', + path: worktreesDir + }) }, { timeout: 2_000 } )