From 8ffe7c75d5553ae1872c3dd48de6adcd8aa927d9 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 9 Aug 2026 02:14:00 +0200 Subject: [PATCH] Keep the daemon responsive when file watching fails (#3056) * fix(server): keep watcher failures from stalling the daemon Native watcher teardown can block the Node main thread when an error overlaps subscription lifecycle work. Reduce lifecycle churn, abandon errored subscriptions into bounded polling, recover only after setup settles, and expose error callback frequency in runtime metrics. * fix(server): keep watcher recovery bounded --- .../test-utils/workspace-git-service-stub.ts | 1 + .../server/src/server/websocket-server.ts | 1 + .../workspace-git-service.observation.test.ts | 442 +++++++++++++++++- .../src/server/workspace-git-service.ts | 265 +++++++++-- 4 files changed, 660 insertions(+), 49 deletions(-) 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;