From 0952ddbd4bca85ea0a870256afde97275e59e22b Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 4 Nov 2026 13:10:21 +0700 Subject: [PATCH] fix(server): recover workspace Git watchers after setup timeouts * fix(server): recover git watchers after setup timeouts Setup failures entered degraded polling without scheduling the existing recovery ladder. Retry both watcher types so successful subscriptions stop fallback reads. * fix(server): isolate callbacks from expired git watchers An expired subscription can report an error after recovery. Ignore abandoned callbacks and callbacks that no longer own the active subscription so they cannot tear down a replacement. * fix(server): refresh metadata when git observation recovers * fix(server): settle watcher recovery during workspace teardown * fix(server): reconcile exclusions when metadata watching recovers --- .../src/server/workspace-git-service.test.ts | 237 +++++++++++++++++- .../src/server/workspace-git-service.ts | 62 ++++- 2 files changed, 293 insertions(+), 6 deletions(-) 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 {