diff --git a/packages/server/src/server/test-utils/workspace-git-service-stub.ts b/packages/server/src/server/test-utils/workspace-git-service-stub.ts index 5a28a5f49..8c82bad82 100644 --- a/packages/server/src/server/test-utils/workspace-git-service-stub.ts +++ b/packages/server/src/server/test-utils/workspace-git-service-stub.ts @@ -84,6 +84,7 @@ export function createNoopWorkspaceGitService( workspaceRefreshQueuedCount: 0, fetchInFlightCount: 0, snapshotUpdatedListenerCount: 0, + watcherErrorCallbackCount: 0, }), dispose: () => {}, ...overrides, diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 18e9ac02a..0b95c9a4b 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -248,6 +248,7 @@ function createFallbackWorkspaceGitService(): WorkspaceGitService { workspaceRefreshQueuedCount: 0, fetchInFlightCount: 0, snapshotUpdatedListenerCount: 0, + watcherErrorCallbackCount: 0, }), dispose: () => {}, }; diff --git a/packages/server/src/server/workspace-git-service.observation.test.ts b/packages/server/src/server/workspace-git-service.observation.test.ts index 97e6c1c76..99b6802eb 100644 --- a/packages/server/src/server/workspace-git-service.observation.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.test.ts @@ -103,6 +103,7 @@ function createLogger(): pino.Logger { const logger = { child: () => logger, debug: vi.fn(), + info: vi.fn(), warn: vi.fn(), }; return logger as unknown as pino.Logger; @@ -122,9 +123,25 @@ function getCalledCwds(mock: ReturnType): string[] { return mock.mock.calls.map(([cwd]) => cwd as string); } +function getWatcherRecordsForDirectory( + watcher: ReturnType, + directory: string, +): WatchRecord[] { + return watcher.records.filter((record) => record.directory === directory); +} + +function getWatcherSubscribeCallCount( + watcher: ReturnType, + directory: string, +): number { + return watcher.subscribe.mock.calls.filter(([calledDirectory]) => calledDirectory === directory) + .length; +} + function createService( watcher: ReturnType, overrides?: Record, + logger: pino.Logger = createLogger(), ) { const defaultGetCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const defaultGetCheckoutShortstat = vi.fn(async () => null); @@ -135,7 +152,7 @@ function createService( (overrides?.getCheckoutShortstat as typeof defaultGetCheckoutShortstat | undefined) ?? defaultGetCheckoutShortstat; return new WorkspaceGitServiceImpl({ - logger: createLogger(), + logger, paseoHome: "/tmp/paseo-home", deps: { subscribe: watcher.subscribe, @@ -1003,13 +1020,14 @@ describe("WorkspaceGitService checkout observation", () => { service.dispose(); }); - test("watcher runtime error switches to scoped polling", async () => { + test("watcher runtime error is abandoned, counted, and switches to scoped polling", async () => { const watcher = createWatcherHarness(); const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd)); const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const service = createService(watcher, { getCheckoutSnapshotFacts, getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000_000, }); const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); @@ -1021,10 +1039,9 @@ describe("WorkspaceGitService checkout observation", () => { expect(checkoutWatcher).toBeDefined(); checkoutWatcher?.callback(new Error("watcher stopped"), []); - await vi.waitFor(() => { - expect(checkoutWatcher?.unsubscribe).toHaveBeenCalledTimes(1); - }); - await vi.advanceTimersByTimeAsync(5_000); + expect(checkoutWatcher?.unsubscribe).not.toHaveBeenCalled(); + expect(service.getMetrics().watcherErrorCallbackCount).toBe(1); + await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { expect(getCheckoutStatus).toHaveBeenCalledTimes(2); }); @@ -1032,6 +1049,419 @@ describe("WorkspaceGitService checkout observation", () => { expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0); + await vi.advanceTimersByTimeAsync(29_000); + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(2); + }); + const recoveredWatcher = getWatcherRecordsForDirectory(watcher, REPO_CWD)[1]; + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); + }); + const statusCallsAfterRecovery = getCheckoutStatus.mock.calls.length; + recoveredWatcher?.callback(null, [ + { path: path.join(REPO_CWD, "recovered.txt"), type: "update" }, + ]); + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsAfterRecovery + 1); + }); + const statusCallsAfterEvent = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(10_000); + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsAfterEvent); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("setup-time watcher error defers recovery until subscribe settles", async () => { + const watcher = createWatcherHarness(); + const openedSubscription = createDeferred<{ unsubscribe: () => Promise }>(); + const erroredUnsubscribe = vi.fn(async () => {}); + watcher.subscribe.mockImplementationOnce(async (_directory, callback) => { + callback(new Error("watcher stopped during setup"), []); + return openedSubscription.promise; + }); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const service = createService(watcher, { + getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000_000, + }); + + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + await vi.waitFor(() => { + expect(watcher.subscribe).toHaveBeenCalledTimes(1); + }); + await vi.advanceTimersByTimeAsync(35_000); + expect(service.getMetrics().workingTreeWatchSetupInFlightCount).toBe(1); + expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(1); + + openedSubscription.resolve({ unsubscribe: erroredUnsubscribe }); + await vi.waitFor(() => { + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + const statusCallsAfterSetup = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(5_000); + await vi.waitFor(() => { + expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(statusCallsAfterSetup); + }); + await vi.advanceTimersByTimeAsync(24_000); + expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(1); + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(2); + }); + expect(erroredUnsubscribe).not.toHaveBeenCalled(); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("watcher error during subscription setup is abandoned without native teardown", async () => { + const watcher = createWatcherHarness(); + const erroredUnsubscribe = vi.fn(async () => {}); + watcher.subscribe.mockImplementationOnce(async (_directory, callback) => { + callback(new Error("watcher stopped during setup"), []); + return { unsubscribe: erroredUnsubscribe }; + }); + const service = createService(watcher); + + const subscription = await service.requestWorkingTreeWatch(REPO_CWD, vi.fn()); + + expect(erroredUnsubscribe).not.toHaveBeenCalled(); + expect(service.getMetrics().watcherErrorCallbackCount).toBe(1); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("repository watcher runtime error is abandoned, counted, and recovered", async () => { + const watcher = createWatcherHarness(); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const service = createService(watcher, { + getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000_000, + }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + + await vi.waitFor(() => { + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + const repositoryWatcher = watcher.records.find((record) => record.directory === GIT_DIR); + expect(repositoryWatcher).toBeDefined(); + + repositoryWatcher?.callback(new Error("repository watcher stopped"), []); + + expect(repositoryWatcher?.unsubscribe).not.toHaveBeenCalled(); + expect(service.getMetrics().watcherErrorCallbackCount).toBe(1); + + const statusCallsBeforeReconciliation = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsBeforeReconciliation + 1); + }); + + await vi.advanceTimersByTimeAsync(29_000); + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, GIT_DIR)).toHaveLength(2); + }); + const recoveredWatcher = getWatcherRecordsForDirectory(watcher, GIT_DIR)[1]; + const statusCallsAfterRecovery = getCheckoutStatus.mock.calls.length; + recoveredWatcher?.callback(null, [{ path: path.join(GIT_DIR, "HEAD"), type: "update" }]); + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsAfterRecovery + 1); + }); + const statusCallsAfterEvent = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(10_000); + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsAfterEvent); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("watcher recovery remains capped when recovered subscriptions emit events before failing", async () => { + const watcher = createWatcherHarness(); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const service = createService(watcher, { + getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000_000, + }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + + await vi.waitFor(() => { + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + const checkoutWatcher = getWatcherRecordsForDirectory(watcher, REPO_CWD)[0]; + + checkoutWatcher?.callback(new Error("watcher stopped"), []); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(2); + }); + getWatcherRecordsForDirectory(watcher, REPO_CWD)[1]?.callback(null, [ + { path: path.join(REPO_CWD, "recovered-1.txt"), type: "update" }, + ]); + getWatcherRecordsForDirectory(watcher, REPO_CWD)[1]?.callback( + new Error("recovered watcher stopped"), + [], + ); + await vi.advanceTimersByTimeAsync(60_000); + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(3); + }); + getWatcherRecordsForDirectory(watcher, REPO_CWD)[2]?.callback(null, [ + { path: path.join(REPO_CWD, "recovered-2.txt"), type: "update" }, + ]); + getWatcherRecordsForDirectory(watcher, REPO_CWD)[2]?.callback( + new Error("recovered watcher stopped again"), + [], + ); + await vi.advanceTimersByTimeAsync(120_000); + await vi.waitFor(() => { + expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(4); + }); + getWatcherRecordsForDirectory(watcher, REPO_CWD)[3]?.callback(null, [ + { path: path.join(REPO_CWD, "recovered-3.txt"), type: "update" }, + ]); + getWatcherRecordsForDirectory(watcher, REPO_CWD)[3]?.callback( + new Error("last recovered watcher stopped"), + [], + ); + + const statusCallsAtCap = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(300_000); + expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(4); + expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(statusCallsAtCap); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("repository watcher recovery remains capped after recovered subscriptions emit events", async () => { + const watcher = createWatcherHarness(); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const service = createService(watcher, { + getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000_000, + }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + + await vi.waitFor(() => { + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + getWatcherRecordsForDirectory(watcher, GIT_DIR)[0]?.callback( + new Error("repository watcher stopped"), + [], + ); + + for (const [recoveryIndex, delayMs] of [30_000, 60_000, 120_000].entries()) { + await vi.advanceTimersByTimeAsync(delayMs); + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, GIT_DIR)).toHaveLength(recoveryIndex + 2); + }); + const recoveredWatcher = getWatcherRecordsForDirectory(watcher, GIT_DIR)[recoveryIndex + 1]; + recoveredWatcher?.callback(null, [{ path: path.join(GIT_DIR, "HEAD"), type: "update" }]); + recoveredWatcher?.callback(new Error("recovered repository watcher stopped"), []); + } + + const statusCallsAtCap = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(300_000); + expect(getWatcherSubscribeCallCount(watcher, GIT_DIR)).toBe(4); + expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(statusCallsAtCap); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("non-Git discovery polling survives watcher recovery", async () => { + const watcher = createWatcherHarness(); + const getCheckoutSnapshotFacts = vi.fn( + async (): Promise => ({ + isGit: false, + }), + ); + const getCheckoutStatus = vi.fn(async () => ({ isGit: false }) as const); + const runGitCommand = vi.fn(async () => { + throw new Error("not a git repository"); + }); + const service = createService(watcher, { + getCheckoutSnapshotFacts, + getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000_000, + runGitCommand, + }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + + await vi.waitFor(() => { + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + const checkoutWatcher = getWatcherRecordsForDirectory(watcher, REPO_CWD)[0]; + checkoutWatcher?.callback(new Error("watcher stopped"), []); + + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(2); + }); + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); + }); + const statusCallsAfterRecovery = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(5_000); + await vi.waitFor(() => { + expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(statusCallsAfterRecovery); + }); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("adding ignored directories updates filtering without replacing the watcher", async () => { + const watcher = createWatcherHarness(); + let ignoredDirectories = "node_modules/\n"; + const runGitCommand = vi.fn(async (args: string[]) => { + return { + 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(watcher, { + getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000, + runGitCommand, + }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(1); + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + const checkoutWatcher = watcher.records.find((record) => record.directory === REPO_CWD); + expect(checkoutWatcher?.ignore).toContain(path.join(REPO_CWD, "node_modules")); + + ignoredDirectories = "node_modules/\nbuild/\n"; + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(runGitCommand).toHaveBeenCalledTimes(3); + }); + + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(1); + expect(checkoutWatcher?.unsubscribe).not.toHaveBeenCalled(); + const statusCallsAfterRefresh = getCheckoutStatus.mock.calls.length; + checkoutWatcher?.callback(null, [ + { path: path.join(REPO_CWD, "build", "output.js"), type: "update" }, + ]); + await vi.advanceTimersByTimeAsync(2_000); + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsAfterRefresh); + + checkoutWatcher?.callback(null, [{ path: path.join(REPO_CWD, "tracked.txt"), type: "update" }]); + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsAfterRefresh + 1); + }); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("removing an ignored directory replaces the watcher", async () => { + const watcher = createWatcherHarness(); + let ignoredDirectories = "node_modules/\nbuild/\n"; + const runGitCommand = vi.fn(async (args: string[]) => { + return { + 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(watcher, { + getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000, + runGitCommand, + }); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(1); + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + const originalWatcher = watcher.records.find((record) => record.directory === REPO_CWD); + + ignoredDirectories = "node_modules/\n"; + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(2); + }); + + expect(originalWatcher?.unsubscribe).toHaveBeenCalledTimes(1); + const replacementWatcher = getWatcherRecordsForDirectory(watcher, REPO_CWD)[1]; + expect(replacementWatcher?.ignore).not.toContain(path.join(REPO_CWD, "build")); + const statusCallsAfterRefresh = getCheckoutStatus.mock.calls.length; + replacementWatcher?.callback(null, [ + { path: path.join(REPO_CWD, "build", "output.js"), type: "update" }, + ]); + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(getCheckoutStatus).toHaveBeenCalledTimes(statusCallsAfterRefresh + 1); + }); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("ignore watcher teardown failure enters polling with a distinct reason", async () => { + const watcher = createWatcherHarness(); + const logger = createLogger(); + let ignoredDirectories = "node_modules/\nbuild/\n"; + 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( + watcher, + { + getCheckoutStatus, + getWorkspaceGitSelfHealPhaseMs: () => 1_000, + runGitCommand, + }, + logger, + ); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(1); + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + const checkoutWatcher = getWatcherRecordsForDirectory(watcher, REPO_CWD)[0]; + checkoutWatcher?.unsubscribe.mockRejectedValueOnce(new Error("teardown failed")); + ignoredDirectories = "node_modules/\n"; + + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ reason: "watcher_teardown_failed" }), + "Working tree watcher unavailable; using bounded polling fallback", + ); + }); + expect(service.getMetrics().watcherErrorCallbackCount).toBe(0); + const statusCallsBeforePoll = getCheckoutStatus.mock.calls.length; + await vi.advanceTimersByTimeAsync(5_000); + await vi.waitFor(() => { + expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(statusCallsBeforePoll); + }); + 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 3fe51becb..83ca876ac 100644 --- a/packages/server/src/server/workspace-git-service.ts +++ b/packages/server/src/server/workspace-git-service.ts @@ -58,6 +58,8 @@ const FORGE_PR_STATUS_POLL_FAST_INTERVAL_MS = 20_000; const FORGE_PR_STATUS_POLL_SLOW_INTERVAL_MS = 120_000; const FORGE_PR_STATUS_POLL_ERROR_BACKOFF_CAP_MS = 300_000; const DEGRADED_GIT_POLL_INTERVAL_MS = 5_000; +const WATCH_RECOVERY_BASE_DELAY_MS = 30_000; +const WATCH_RECOVERY_MAX_ATTEMPTS = 3; // Auxiliary reads may reuse cached values within this window; snapshots do not expire on read. const WORKSPACE_GIT_AUXILIARY_READ_TTL_MS = 15_000; // Non-forced refresh triggers share this minimum gap to absorb watcher/self-heal bursts; force bypasses it. @@ -196,6 +198,7 @@ export interface WorkspaceGitServiceMetrics { workspaceRefreshQueuedCount: number; fetchInFlightCount: number; snapshotUpdatedListenerCount: number; + watcherErrorCallbackCount: number; } export type WorkspaceGitListener = (snapshot: WorkspaceGitRuntimeSnapshot) => void; @@ -371,7 +374,9 @@ interface RepoGitTarget { cwd: string; workspaceKeys: Set; subscription: parcelWatcher.AsyncSubscription | null; + fallbackPolling: boolean; fallbackPollTimer: NodeJS.Timeout | null; + recovery: WatchRecoveryState; intervalId: NodeJS.Timeout | null; fetchInFlight: boolean; closed: boolean; @@ -390,11 +395,24 @@ interface WorkingTreeWatchTarget { ignoredDirectoriesRefreshPromise: Promise | null; aliases: Set; workspaceKeys: Set; + fallbackPolling: boolean; fallbackPollTimer: NodeJS.Timeout | null; + recovery: WatchRecoveryState; listeners: Set<() => void>; closed: boolean; } +interface WatchRecoveryState { + attemptCount: number; + timer: NodeJS.Timeout | null; +} + +type WorkingTreeWatchFallbackReason = + | "not_a_git_checkout" + | "watcher_error" + | "watcher_setup_failed" + | "watcher_teardown_failed"; + interface WorkspaceGitAuxiliaryReadCacheEntry { value: T | null; loadedAtMs: number | null; @@ -477,6 +495,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { string, WorkspaceGitAuxiliaryReadCacheEntry >({ max: WORKSPACE_GIT_CHECKOUT_DIFF_CACHE_MAX }); + private watcherErrorCallbackCount = 0; constructor(options: WorkspaceGitServiceOptions) { this.logger = options.logger.child({ module: "workspace-git-service" }); this.paseoHome = options.paseoHome; @@ -566,6 +585,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { workspaceRefreshQueuedCount, fetchInFlightCount, snapshotUpdatedListenerCount: this.snapshotUpdatedListeners.size, + watcherErrorCallbackCount: this.watcherErrorCallbackCount, }; } @@ -1153,34 +1173,52 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { ignoredDirectoriesRefreshPromise: null, aliases: new Set([cwd]), workspaceKeys: new Set(), + fallbackPolling: false, fallbackPollTimer: null, + recovery: { attemptCount: 0, timer: null }, listeners: new Set(), closed: false, }; + this.workingTreeWatchTargets.set(cwd, target); + this.workingTreeWatchAliases.set(cwd, cwd); await this.startWorkingTreeSubscription(target); if (repoRoot === null) { this.startWorkingTreeWatchFallback(target, "not_a_git_checkout"); } - this.workingTreeWatchTargets.set(cwd, target); - this.workingTreeWatchAliases.set(cwd, cwd); return target; } - private async startWorkingTreeSubscription(target: WorkingTreeWatchTarget): Promise { + private async startWorkingTreeSubscription( + target: WorkingTreeWatchTarget, + options?: { replaceFallback?: boolean }, + ): Promise { const ignore = [join(target.watchPath, ".git"), ...target.ignoredDirectories]; + let watcherErrored = false; + let subscribeSettled = false; try { const subscription = await this.deps.subscribe( target.watchPath, (error, events) => { if (error) { + if (watcherErrored) { + return; + } + watcherErrored = true; + this.watcherErrorCallbackCount += 1; this.logger.warn( { err: error, cwd: target.cwd }, "Working tree watcher error; using degraded polling", ); this.degradeWorkingTreeWatch(target, "watcher_error"); + if (subscribeSettled) { + this.scheduleWorkingTreeWatchRecovery(target); + } + return; + } + if (watcherErrored) { return; } if (!this.hasRelevantWorkingTreeEvent(target, events)) { @@ -1190,27 +1228,53 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }, { ignore }, ); - if (target.closed || target.fallbackPollTimer || target.subscription) { - await subscription.unsubscribe(); - } else { - target.subscription = subscription; + subscribeSettled = true; + if (watcherErrored) { + this.scheduleWorkingTreeWatchRecovery(target); + return false; } + if ( + target.closed || + (target.fallbackPolling && !options?.replaceFallback) || + target.subscription + ) { + await subscription.unsubscribe(); + return false; + } + target.subscription = subscription; + if (options?.replaceFallback && target.repoRoot !== null) { + target.fallbackPolling = false; + if (target.fallbackPollTimer) { + clearTimeout(target.fallbackPollTimer); + target.fallbackPollTimer = null; + } + } + return true; } catch (error) { + subscribeSettled = true; + if (watcherErrored) { + this.scheduleWorkingTreeWatchRecovery(target); + return false; + } this.logger.warn( { err: error, cwd: target.cwd }, "Failed to start working tree watcher; using degraded polling", ); - this.startWorkingTreeWatchFallback(target, "watcher_setup_failed"); + if (!options?.replaceFallback) { + this.startWorkingTreeWatchFallback(target, "watcher_setup_failed"); + } + return false; } } private startWorkingTreeWatchFallback( target: WorkingTreeWatchTarget, - reason: "not_a_git_checkout" | "watcher_error" | "watcher_setup_failed", + reason: WorkingTreeWatchFallbackReason, ): void { - if (target.fallbackPollTimer) { + if (target.fallbackPolling) { return; } + target.fallbackPolling = true; const { cwd } = target; const poll = async () => { target.fallbackPollTimer = null; @@ -1241,6 +1305,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { this.notifyWorkingTreeConsumers(target); if (!target.closed && (target.subscription === null || target.repoRoot === null)) { target.fallbackPollTimer = setTimeout(poll, DEGRADED_GIT_POLL_INTERVAL_MS); + } else { + target.fallbackPolling = false; } }; target.fallbackPollTimer = setTimeout(poll, DEGRADED_GIT_POLL_INTERVAL_MS); @@ -1255,16 +1321,44 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private degradeWorkingTreeWatch(target: WorkingTreeWatchTarget, reason: "watcher_error"): void { - if (target.subscription) { - const subscription = target.subscription; - target.subscription = null; - void subscription.unsubscribe().catch((error) => { - this.logger.warn({ err: error, cwd: target.cwd }, "Failed to stop working tree watcher"); - }); - } + // COMPAT(parcel-watcher-eintr): added in v0.3.0, remove after 2027-02-09. Unsubscribing an + // errored subscription races native teardown and can strand the Node main thread in + // InotifyBackend::~InotifyBackend(): https://github.com/parcel-bundler/watcher/issues/253 + target.subscription = null; + this.notifyWorkingTreeChanged(target, "working-tree-watch-error"); this.startWorkingTreeWatchFallback(target, reason); } + private scheduleWorkingTreeWatchRecovery(target: WorkingTreeWatchTarget): void { + if ( + target.closed || + target.subscription || + target.recovery.timer || + target.recovery.attemptCount >= WATCH_RECOVERY_MAX_ATTEMPTS + ) { + return; + } + target.recovery.attemptCount += 1; + const delayMs = WATCH_RECOVERY_BASE_DELAY_MS * 2 ** (target.recovery.attemptCount - 1); + target.recovery.timer = setTimeout(() => { + target.recovery.timer = null; + void this.recoverWorkingTreeWatch(target); + }, delayMs); + } + + private async recoverWorkingTreeWatch(target: WorkingTreeWatchTarget): Promise { + if (target.closed || target.subscription) { + return; + } + const recovered = await this.startWorkingTreeSubscription(target, { replaceFallback: true }); + if (!recovered) { + this.scheduleWorkingTreeWatchRecovery(target); + return; + } + this.logger.info({ cwd: target.cwd }, "Working tree watcher recovered"); + this.notifyWorkingTreeChanged(target, "working-tree-watch-recovered"); + } + private async promoteWorkingTreeWatchTarget( target: WorkingTreeWatchTarget, repoRoot: string, @@ -1273,9 +1367,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return; } target.repoRoot = repoRoot; - if (target.subscription && target.fallbackPollTimer) { - clearTimeout(target.fallbackPollTimer); - target.fallbackPollTimer = null; + if (target.subscription && target.fallbackPolling) { + target.fallbackPolling = false; + if (target.fallbackPollTimer) { + clearTimeout(target.fallbackPollTimer); + target.fallbackPollTimer = null; + } } await this.refreshWorkingTreeIgnoredDirectories(target); } @@ -1305,7 +1402,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private refreshWorkingTreeIgnoredDirectories(target: WorkingTreeWatchTarget): Promise { - if (target.closed || target.repoRoot === null || target.fallbackPollTimer) { + if (target.closed || target.repoRoot === null || target.fallbackPolling) { return Promise.resolve(); } if (target.ignoredDirectoriesRefreshPromise) { @@ -1334,20 +1431,28 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const ignoredDirectories = await this.loadIgnoredDirs(target.watchPath); if ( target.closed || - target.fallbackPollTimer || + target.fallbackPolling || this.haveSamePaths(target.ignoredDirectories, ignoredDirectories) ) { return; } + const removedIgnoredDirectory = Array.from(target.ignoredDirectories).some( + (path) => !ignoredDirectories.has(path), + ); + target.ignoredDirectories = ignoredDirectories; + if (!removedIgnoredDirectory) { + return; + } + const subscription = target.subscription; if (subscription) { target.subscription = null; try { await subscription.unsubscribe(); } catch (error) { - if (!target.closed && !target.fallbackPollTimer) { - target.subscription = subscription; + if (!target.closed && !target.fallbackPolling) { + this.startWorkingTreeWatchFallback(target, "watcher_teardown_failed"); } this.logger.warn( { err: error, cwd: target.cwd }, @@ -1356,11 +1461,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return; } } - if (target.closed || target.fallbackPollTimer) { + if (target.closed || target.fallbackPolling) { return; } - target.ignoredDirectories = ignoredDirectories; await this.startWorkingTreeSubscription(target); this.notifyWorkingTreeChanged(target, "working-tree-watch-reconfigured"); } @@ -1452,7 +1556,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { cwd: workspaceTarget.cwd, workspaceKeys: new Set([workspaceTarget.cwd]), subscription: null, + fallbackPolling: false, fallbackPollTimer: null, + recovery: { attemptCount: 0, timer: null }, intervalId: null, fetchInFlight: false, closed: false, @@ -1490,21 +1596,37 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { void this.runRepoFetch(repoTarget); } - private async startRepoMetadataObservation(target: RepoGitTarget): Promise { + private async startRepoMetadataObservation( + target: RepoGitTarget, + options?: { replaceFallback?: boolean }, + ): Promise { const ignore = getPrunedGitMetadataPaths("common").map((path) => join(target.repoGitRoot, path), ); const matchesRepoGitRoot = createRealpathAwarePathMatcher(target.repoGitRoot); + let watcherErrored = false; + let subscribeSettled = false; try { const subscription = await this.deps.subscribe( target.repoGitRoot, (error, events) => { if (error) { + if (watcherErrored) { + return; + } + watcherErrored = true; + this.watcherErrorCallbackCount += 1; this.logger.warn( { err: error, repoGitRoot: target.repoGitRoot }, "Repository metadata watcher error; using degraded polling", ); this.degradeRepoMetadataWatch(target); + if (subscribeSettled) { + this.scheduleRepoMetadataWatchRecovery(target); + } + return; + } + if (watcherErrored) { return; } const relevantEvents = events.filter( @@ -1523,38 +1645,82 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }, { ignore }, ); + subscribeSettled = true; + if (watcherErrored) { + this.scheduleRepoMetadataWatchRecovery(target); + return false; + } if ( target.closed || - target.fallbackPollTimer || + (target.fallbackPolling && !options?.replaceFallback) || this.repoTargets.get(target.repoGitRoot) !== target ) { await subscription.unsubscribe(); - } else { - target.subscription = subscription; + return false; } + target.subscription = subscription; + if (options?.replaceFallback) { + target.fallbackPolling = false; + if (target.fallbackPollTimer) { + clearTimeout(target.fallbackPollTimer); + target.fallbackPollTimer = null; + } + } + return true; } catch (error) { + subscribeSettled = true; + if (watcherErrored) { + this.scheduleRepoMetadataWatchRecovery(target); + return false; + } this.logger.warn( { err: error, repoGitRoot: target.repoGitRoot }, "Failed to start repository metadata watcher; using degraded polling", ); - this.startRepoMetadataFallback(target); + if (!options?.replaceFallback) { + this.startRepoMetadataFallback(target); + } + return false; } } private degradeRepoMetadataWatch(target: RepoGitTarget): void { - if (target.subscription) { - const subscription = target.subscription; - target.subscription = null; - void subscription.unsubscribe().catch((error) => { - this.logger.warn( - { err: error, repoGitRoot: target.repoGitRoot }, - "Failed to stop repository metadata watcher", - ); - }); - } + // COMPAT(parcel-watcher-eintr): added in v0.3.0, remove after 2027-02-09. See + // degradeWorkingTreeWatch for why this errored subscription must be abandoned. + target.subscription = null; + this.scheduleRepoMetadataRefresh(target, "git-metadata-watch-error", true); this.startRepoMetadataFallback(target); } + private scheduleRepoMetadataWatchRecovery(target: RepoGitTarget): void { + if ( + target.closed || + target.subscription || + target.recovery.timer || + target.recovery.attemptCount >= WATCH_RECOVERY_MAX_ATTEMPTS + ) { + return; + } + target.recovery.attemptCount += 1; + const delayMs = WATCH_RECOVERY_BASE_DELAY_MS * 2 ** (target.recovery.attemptCount - 1); + target.recovery.timer = setTimeout(() => { + target.recovery.timer = null; + void this.recoverRepoMetadataWatch(target); + }, delayMs); + } + + private async recoverRepoMetadataWatch(target: RepoGitTarget): Promise { + if (target.closed || target.subscription) { + return; + } + const recovered = await this.startRepoMetadataObservation(target, { replaceFallback: true }); + if (!recovered) { + this.scheduleRepoMetadataWatchRecovery(target); + return; + } + this.logger.info({ repoGitRoot: target.repoGitRoot }, "Repository metadata watcher recovered"); + } + private routeRepoMetadataEvents( target: RepoGitTarget, events: parcelWatcher.Event[], @@ -1743,9 +1909,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private startRepoMetadataFallback(target: RepoGitTarget): void { - if (target.fallbackPollTimer || target.closed) { + if (target.fallbackPolling || target.closed) { return; } + target.fallbackPolling = true; const poll = async () => { target.fallbackPollTimer = null; if (target.closed || this.repoTargets.get(target.repoGitRoot) !== target) { @@ -1781,8 +1948,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { for (const workingTreeTarget of workingTreeTargets) { this.notifyWorkingTreeConsumers(workingTreeTarget); } - if (!target.closed) { + if (!target.closed && target.subscription === null) { target.fallbackPollTimer = setTimeout(poll, DEGRADED_GIT_POLL_INTERVAL_MS); + } else { + target.fallbackPolling = false; } }; target.fallbackPollTimer = setTimeout(poll, DEGRADED_GIT_POLL_INTERVAL_MS); @@ -2647,6 +2816,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { clearTimeout(target.fallbackPollTimer); target.fallbackPollTimer = null; } + target.fallbackPolling = false; + if (target.recovery.timer) { + clearTimeout(target.recovery.timer); + target.recovery.timer = null; + } if (target.subscription) { const subscription = target.subscription; @@ -2669,6 +2843,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { clearTimeout(target.fallbackPollTimer); target.fallbackPollTimer = null; } + target.fallbackPolling = false; + if (target.recovery.timer) { + clearTimeout(target.recovery.timer); + target.recovery.timer = null; + } if (target.subscription) { const subscription = target.subscription; target.subscription = null;