diff --git a/packages/server/src/server/workspace-git-service.test.ts b/packages/server/src/server/workspace-git-service.test.ts index 744cce2930..490807f8d6 100644 --- a/packages/server/src/server/workspace-git-service.test.ts +++ b/packages/server/src/server/workspace-git-service.test.ts @@ -1560,6 +1560,241 @@ describe("WorkspaceGitServiceImpl", () => { service.dispose(); }); + test.each(["repository metadata", "working tree"])( + "recovers after a %s watcher subscription times out during setup", + async (kind) => { + const watchPath = kind === "repository metadata" ? join(REPO_CWD, ".git") : REPO_CWD; + let attempts = 0; + const subscribe = vi.fn(async (directory: string) => { + if (directory === watchPath && ++attempts === 1) { + return new Promise>(() => {}); + } + return createAsyncSubscription(); + }); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const service = createService({ subscribe, getCheckoutStatus }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + try { + await settle(() => expect(attempts).toBe(1)); + await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + await waitForObservationToSettle(service, 1); + const readsBeforePoll = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(5_000); + await waitForRefreshesToSettle(service); + expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(readsBeforePoll); + await vi.advanceTimersByTimeAsync(24_999); + expect(attempts).toBe(1); + await vi.advanceTimersByTimeAsync(1); + await settle(() => expect(attempts).toBe(2)); + // Recovery emits a debounced working-tree refresh before observation takes over. + await vi.advanceTimersByTimeAsync(1_000); + await waitForRefreshesToSettle(service); + const readsAfterRecovery = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(20_000); + await waitForRefreshesToSettle(service); + expect(getCheckoutStatus.mock.calls.length).toBe(readsAfterRecovery); + } finally { + subscription.unsubscribe(); + await service.dispose(); + } + }, + 15_000, + ); + + test.each(["repository metadata", "working tree"])( + "ignores expired %s watcher callbacks after recovery", + async (kind) => { + const watchPath = kind === "repository metadata" ? join(REPO_CWD, ".git") : REPO_CWD; + let attempts = 0; + const callbacks: Array< + (error: Error | null, events: Array<{ path: string; type: "update" }>) => void + > = []; + const recovered = createAsyncSubscription(); + const subscribe = vi.fn(async (directory: string, callback: (typeof callbacks)[number]) => { + if (directory === watchPath) callbacks.push(callback); + if (directory === watchPath && ++attempts === 1) { + return new Promise>(() => {}); + } + return directory === watchPath ? recovered : createAsyncSubscription(); + }); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const service = createService({ subscribe, getCheckoutStatus }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + try { + await settle(() => expect(attempts).toBe(1)); + await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + await waitForObservationToSettle(service, 1); + const readsBeforePoll = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(5_000); + await waitForRefreshesToSettle(service); + expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(readsBeforePoll); + await vi.advanceTimersByTimeAsync(24_999); + expect(attempts).toBe(1); + await vi.advanceTimersByTimeAsync(1); + await settle(() => expect(attempts).toBe(2)); + // Recovery emits a debounced working-tree refresh before observation takes over. + await vi.advanceTimersByTimeAsync(1_000); + await waitForRefreshesToSettle(service); + const readsAfterRecovery = getCheckoutStatus.mock.calls.length; + callbacks[0]!(new Error("expired watcher stopped"), []); + callbacks[0]!(null, [{ path: join(watchPath, "stale.ts"), type: "update" }]); + expect(recovered.unsubscribe).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(20_000); + await waitForRefreshesToSettle(service); + expect(getCheckoutStatus.mock.calls.length).toBe(readsAfterRecovery); + expect(recovered.unsubscribe).not.toHaveBeenCalled(); + } finally { + subscription.unsubscribe(); + await service.dispose(); + } + }, + 15_000, + ); + + test.each(["repository metadata", "working tree"])( + "refreshes changes missed before %s watcher recovery", + async (kind) => { + const watchPath = kind === "repository metadata" ? join(REPO_CWD, ".git") : REPO_CWD; + let attempts = 0; + let changed = false; + const subscribe = vi.fn(async (directory: string) => { + if (directory === watchPath && ++attempts === 1) { + return new Promise>(() => {}); + } + return createAsyncSubscription(); + }); + const getCheckoutStatus = vi.fn(async (cwd: string) => + createCheckoutStatus(cwd, { + isDirty: changed, + currentBranch: changed ? "feature" : "main", + }), + ); + const service = createService({ + subscribe, + getCheckoutStatus, + getCheckoutSnapshotFacts: async (cwd) => ({ + ...createCheckoutSnapshotFacts(cwd), + currentBranch: changed ? "feature" : "main", + }), + }); + const snapshots: WorkspaceGitRuntimeSnapshot[] = []; + service.onSnapshotUpdated((snapshot) => snapshots.push(snapshot)); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + try { + await settle(() => expect(attempts).toBe(1)); + await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + await waitForObservationToSettle(service, 1); + await vi.advanceTimersByTimeAsync(29_999); + await waitForRefreshesToSettle(service); + expect(snapshots.at(-1)?.git).toMatchObject({ currentBranch: "main", isDirty: false }); + changed = true; + await vi.advanceTimersByTimeAsync(1); + await settle(() => expect(attempts).toBe(2)); + await vi.advanceTimersByTimeAsync(1_000); + await waitForRefreshesToSettle(service); + expect(snapshots.at(-1)?.git).toMatchObject( + kind === "repository metadata" ? { currentBranch: "feature" } : { isDirty: true }, + ); + const readsAfterRecovery = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(20_000); + await waitForRefreshesToSettle(service); + expect(getCheckoutStatus.mock.calls.length).toBe(readsAfterRecovery); + } finally { + subscription.unsubscribe(); + await service.dispose(); + } + }, + ); + + test.each(["repository metadata", "working tree"])( + "settles a pending %s watcher retry after the final consumer leaves", + async (kind) => { + const watchPath = kind === "repository metadata" ? join(REPO_CWD, ".git") : REPO_CWD; + let attempts = 0; + const subscribe = vi.fn(async (directory: string) => { + if (directory === watchPath) { + attempts += 1; + return new Promise>(() => {}); + } + return createAsyncSubscription(); + }); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const service = createService({ subscribe, getCheckoutStatus }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + try { + await settle(() => expect(attempts).toBe(1)); + await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + await waitForObservationToSettle(service, 1); + await vi.advanceTimersByTimeAsync(30_000); + await settle(() => expect(attempts).toBe(2)); + subscription.unsubscribe(); + await waitForRefreshesToSettle(service); + const readsAfterUnsubscribe = getCheckoutStatus.mock.calls.length; + // Vitest reports any unhandled rejection as a suite failure. + await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + await vi.advanceTimersByTimeAsync(300_000); + expect(attempts).toBe(2); + expect(vi.getTimerCount()).toBe(0); + expect(getCheckoutStatus.mock.calls.length).toBe(readsAfterUnsubscribe); + } finally { + subscription.unsubscribe(); + await service.dispose(); + } + }, + ); + + test("metadata recovery reconciles exclusions changed during the outage", async () => { + let ignoredDirectories = "generated/\n"; + let attempts = 0; + let workingTreeCallback!: ( + error: Error | null, + events: Array<{ path: string; type: "delete" }>, + ) => void; + const workingTreeSubscription = createAsyncSubscription(); + const subscribe = vi.fn(async (directory: string, callback: typeof workingTreeCallback) => { + if (directory === join(REPO_CWD, ".git") && ++attempts === 1) { + return new Promise>(() => {}); + } + if (directory === REPO_CWD) { + workingTreeCallback = callback; + return workingTreeSubscription; + } + return createAsyncSubscription(); + }); + const runGitCommand = vi.fn(async (args: string[]) => ({ + stdout: args[0] === "rev-parse" ? `${REPO_CWD}\n` : ignoredDirectories, + stderr: "", + truncated: false, + exitCode: 0, + signal: null, + })); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const service = createService({ subscribe, runGitCommand, getCheckoutStatus }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + try { + await settle(() => expect(attempts).toBe(1)); + await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS); + await waitForObservationToSettle(service, 1); + expect(subscribe.mock.calls.find(([directory]) => directory === REPO_CWD)).toBeDefined(); + ignoredDirectories = ""; // info/exclude changed while metadata observation was unavailable. + await vi.advanceTimersByTimeAsync(30_000); + await settle(() => expect(attempts).toBe(2)); + await vi.advanceTimersByTimeAsync(1_000); + await waitForRefreshesToSettle(service); + const readsAfterRecovery = getCheckoutStatus.mock.calls.length; + workingTreeCallback(null, [ + { path: join(REPO_CWD, "generated", "output.ts"), type: "delete" }, + ]); + await vi.advanceTimersByTimeAsync(1_000); + await waitForRefreshesToSettle(service); + expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(readsAfterRecovery); + expect(workingTreeSubscription.updateIgnore).toHaveBeenCalledWith([join(REPO_CWD, ".git")]); + } finally { + subscription.unsubscribe(); + await service.dispose(); + } + }); + test("sets a bounded polling fallback when recursive observation is unavailable", async () => { const subscribe = vi.fn(async () => { throw new Error("watcher unavailable"); @@ -1567,7 +1802,7 @@ describe("WorkspaceGitServiceImpl", () => { const service = createService({ subscribe }); const subscription = await service.requestWorkingTreeWatch(REPO_CWD, vi.fn()); - expect(vi.getTimerCount()).toBe(1); + expect(vi.getTimerCount()).toBe(2); subscription.unsubscribe(); service.dispose(); diff --git a/packages/server/src/server/workspace-git-service.ts b/packages/server/src/server/workspace-git-service.ts index 295d579265..668373e998 100644 --- a/packages/server/src/server/workspace-git-service.ts +++ b/packages/server/src/server/workspace-git-service.ts @@ -1552,6 +1552,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { options?: { replaceFallback?: boolean }, ): Promise { const ignore = this.workingTreeIgnoreList(target); + let abandoned = false; + let activeSubscription: FileObserverSubscription | null = null; let watcherErrored = false; let subscribeSettled = false; const markSubscribeSettled = () => { @@ -1564,6 +1566,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const subscription = await this.subscribeWithDeadline( target.watchPath, (error, events) => { + if ( + abandoned || + target.closed || + (activeSubscription && target.subscription !== activeSubscription) + ) { + return; + } if (error) { if (watcherErrored) { return; @@ -1597,6 +1606,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { markSubscribeSettled, ); if (watcherErrored) { + abandoned = true; await this.unsubscribeWatcherSubscription(subscription, target.watchPath); return false; } @@ -1605,9 +1615,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { (target.fallbackPolling && !options?.replaceFallback) || target.subscription ) { + abandoned = true; await this.unsubscribeWatcherSubscription(subscription, target.watchPath); return false; } + activeSubscription = subscription; target.subscription = subscription; target.appliedIgnoreList = ignore; // Do not reset `attemptCount` here — `subscribe()` resolving proves @@ -1625,11 +1637,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } return true; } catch (error) { + abandoned = true; if (watcherErrored) { return false; } if (this.disposed || target.closed) { - throw error; + return false; } this.logger.warn( { err: error, cwd: target.cwd }, @@ -1638,6 +1651,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { if (!options?.replaceFallback) { this.startWorkingTreeWatchFallback(target, "watcher_setup_failed"); } + this.scheduleWorkingTreeWatchRecovery(target); return false; } } @@ -1868,7 +1882,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const delayMs = this.advanceWatchRecoveryLadder(target.recovery); target.recovery.timer = setTimeout(() => { target.recovery.timer = null; - void this.recoverWorkingTreeWatch(target); + void this.recoverWorkingTreeWatch(target).catch((error) => { + if (this.disposed || target.closed) { + return; + } + this.logger.warn({ err: error }, "Git watcher recovery failed; retrying with backoff"); + this.scheduleWorkingTreeWatchRecovery(target); + }); }, delayMs); } @@ -2374,6 +2394,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const matchesRepoGitRoot = paths.matches; const canary = this.deps.createWatcherLivenessCanary(target.repoGitRoot); let openedSubscription: FileObserverSubscription | null = null; + let abandoned = false; + let activeSubscription: FileObserverSubscription | null = null; let watcherErrored = false; let subscribeSettled = false; const markSubscribeSettled = () => { @@ -2386,6 +2408,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const subscription = await this.subscribeWithDeadline( target.repoGitRoot, (error, events) => { + if ( + abandoned || + target.closed || + (activeSubscription && target.subscription !== activeSubscription) + ) { + return; + } const liveEvents = canary.filterEvents( events.map((event) => ({ ...event, path: paths.toWatchPath(event.path) })), ); @@ -2444,6 +2473,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { openedSubscription = subscription; await canary.verify(this.disposeController.signal); if (watcherErrored) { + abandoned = true; await this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot); return false; } @@ -2452,9 +2482,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { (target.fallbackPolling && !options?.replaceFallback) || this.repoTargets.get(target.repoGitRoot) !== target ) { + abandoned = true; await this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot); return false; } + activeSubscription = subscription; target.subscription = subscription; // See `advanceWatchRecoveryLadder`: do not reset `attemptCount` on // subscribe success alone, only record when this subscription was @@ -2469,6 +2501,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } return true; } catch (error) { + abandoned = true; if (openedSubscription) { await this.unsubscribeWatcherSubscription(openedSubscription, target.repoGitRoot); } @@ -2476,7 +2509,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return false; } if (this.disposed || target.closed) { - throw error; + return false; } this.logger.warn( { err: error, repoGitRoot: target.repoGitRoot }, @@ -2485,6 +2518,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { if (!options?.replaceFallback) { this.startRepoMetadataFallback(target); } + this.scheduleRepoMetadataWatchRecovery(target); return false; } } @@ -2506,7 +2540,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const delayMs = this.advanceWatchRecoveryLadder(target.recovery); target.recovery.timer = setTimeout(() => { target.recovery.timer = null; - void this.recoverRepoMetadataWatch(target); + void this.recoverRepoMetadataWatch(target).catch((error) => { + if (this.disposed || target.closed) { + return; + } + this.logger.warn({ err: error }, "Git watcher recovery failed; retrying with backoff"); + this.scheduleRepoMetadataWatchRecovery(target); + }); }, delayMs); } @@ -2520,6 +2560,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return; } this.logger.info({ repoGitRoot: target.repoGitRoot }, "Repository metadata watcher recovered"); + await this.refreshWorkingTreeIgnoresForWorkspaces(target.workspaceKeys); + this.scheduleRepoMetadataRefresh(target, "git-metadata-watch-recovered", true); } private routeRepoMetadataEvents( @@ -2561,13 +2603,23 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } } } + void this.refreshWorkingTreeIgnoresForWorkspaces(workspaceKeys); + } + + private async refreshWorkingTreeIgnoresForWorkspaces( + workspaceKeys: Iterable, + ): Promise { + const targets = new Set(); for (const workspaceKey of workspaceKeys) { const workspaceTarget = this.workspaceTargets.get(workspaceKey); const workingTreeTarget = workspaceTarget ? this.getWorkingTreeWatchTargetForWorkspace(workspaceTarget) : null; - if (workingTreeTarget) void this.refreshWorkingTreeIgnoredDirectories(workingTreeTarget); + if (workingTreeTarget) targets.add(workingTreeTarget); } + await Promise.all( + [...targets].map((target) => this.refreshWorkingTreeIgnoredDirectories(target)), + ); } private isFetchRemoteMetadataEvent(target: RepoGitTarget, event: FileChange): boolean {