diff --git a/docs/file-observation.md b/docs/file-observation.md index 1418422ea..f246372f8 100644 --- a/docs/file-observation.md +++ b/docs/file-observation.md @@ -16,4 +16,6 @@ Read aggregate health from the owning observer service. Runtime metrics include Git owns Git-ignore evaluation. The observer accepts absolute excluded roots and applies updates without replacing the observation or exposing its watcher topology. This keeps tracked files inside otherwise ignored directories observable and keeps Git policy out of the filesystem module. +Workspace Git verifies each repository metadata subscription with a one-shot canary inside the Git directory. If the event does not round-trip through the subscription callback, treat the watcher as unavailable and enter degraded polling. Refresh working-tree Git-ignore exclusions from ignore-file events and watcher recovery, never from a healthy-watcher timer. + The real-filesystem contracts and daemon auto-archive lifecycle run in the normal server test suite. Use the scripts only for manual performance and soak work: `npm run measure:file-observer --workspace=@getpaseo/server` measures burst and sustained-create behavior, and `npm run repro:file-observer-teardown --workspace=@getpaseo/server` runs the teardown soak. diff --git a/packages/server/src/server/watcher-liveness-canary.test.ts b/packages/server/src/server/watcher-liveness-canary.test.ts new file mode 100644 index 000000000..4da148140 --- /dev/null +++ b/packages/server/src/server/watcher-liveness-canary.test.ts @@ -0,0 +1,33 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, test } from "vitest"; +import { createWatcherLivenessCanary } from "./watcher-liveness-canary.js"; + +const cleanupPaths: string[] = []; + +afterEach(async () => { + await Promise.all( + cleanupPaths.splice(0).map((path) => rm(path, { recursive: true, force: true })), + ); +}); + +test("requires the canary event to round-trip through the watcher callback", async () => { + const watchRoot = await mkdtemp(join(tmpdir(), "paseo-watcher-canary-")); + cleanupPaths.push(watchRoot); + const canary = createWatcherLivenessCanary(watchRoot, { timeoutMs: 1_000 }); + + const verification = canary.verify(); + const canaryPath = canary.path; + expect(canary.filterEvents([{ path: canaryPath, type: "create" }])).toEqual([]); + + await expect(verification).resolves.toBeUndefined(); +}); + +test("rejects when the watcher never reports the canary", async () => { + const watchRoot = await mkdtemp(join(tmpdir(), "paseo-watcher-canary-")); + cleanupPaths.push(watchRoot); + const canary = createWatcherLivenessCanary(watchRoot, { timeoutMs: 10 }); + + await expect(canary.verify()).rejects.toThrow("did not report its liveness canary"); +}); diff --git a/packages/server/src/server/watcher-liveness-canary.ts b/packages/server/src/server/watcher-liveness-canary.ts new file mode 100644 index 000000000..a366f0229 --- /dev/null +++ b/packages/server/src/server/watcher-liveness-canary.ts @@ -0,0 +1,66 @@ +import { randomUUID } from "node:crypto"; +import { rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import type { FileChange } from "./file-observer/index.js"; + +export const WATCHER_LIVENESS_CANARY_TIMEOUT_MS = 10_000; + +export interface WatcherLivenessCanary { + readonly path: string; + filterEvents(events: FileChange[]): FileChange[]; + verify(signal?: AbortSignal): Promise; +} + +export function createWatcherLivenessCanary( + watchRoot: string, + options: { timeoutMs?: number } = {}, +): WatcherLivenessCanary { + const canaryPath = join(watchRoot, `.paseo-watcher-canary-${randomUUID()}`); + const timeoutMs = options.timeoutMs ?? WATCHER_LIVENESS_CANARY_TIMEOUT_MS; + let reportCanary!: () => void; + const reported = new Promise((resolve) => { + reportCanary = resolve; + }); + + return { + path: canaryPath, + filterEvents(events) { + const filtered = events.filter((event) => event.path !== canaryPath); + if (filtered.length !== events.length) { + reportCanary(); + } + return filtered; + }, + async verify(signal) { + await writeFile(canaryPath, "paseo watcher liveness canary\n", { flag: "wx" }); + let timeout: NodeJS.Timeout | null = null; + let removeAbortListener = () => {}; + try { + const timeoutPromise = new Promise((_resolve, reject) => { + timeout = setTimeout(() => { + reject( + new Error( + `Watcher for ${watchRoot} did not report its liveness canary within ${timeoutMs}ms`, + ), + ); + }, timeoutMs); + }); + const abortPromise = new Promise((_resolve, reject) => { + if (!signal) return; + const rejectForAbort = () => reject(signal.reason); + if (signal.aborted) { + rejectForAbort(); + return; + } + signal.addEventListener("abort", rejectForAbort, { once: true }); + removeAbortListener = () => signal.removeEventListener("abort", rejectForAbort); + }); + await Promise.race([reported, timeoutPromise, abortPromise]); + } finally { + if (timeout) clearTimeout(timeout); + removeAbortListener(); + await rm(canaryPath, { force: true }); + } + }, + }; +} diff --git a/packages/server/src/server/workspace-create-git-fanout.local.e2e.test.ts b/packages/server/src/server/workspace-create-git-fanout.local.e2e.test.ts index 7037d7f17..82f580ef1 100644 --- a/packages/server/src/server/workspace-create-git-fanout.local.e2e.test.ts +++ b/packages/server/src/server/workspace-create-git-fanout.local.e2e.test.ts @@ -153,6 +153,18 @@ async function startObservedFixture(siblingCount: number): Promise { + await startObservedFixture(2); + await settleGitCommands(); + + startGitCommandMetrics(); + await new Promise((resolve) => setTimeout(resolve, 61_000)); + await waitForGitCommandMetricsIdle({ quietMs: 1_500, timeoutMs: 5_000 }); + const idle = stopGitCommandMetrics(); + + expect(idle.submissions).toEqual([]); +}, 90_000); + interface GitRuntimeExpectation { currentBranch?: string; remoteUrl?: string; @@ -377,21 +389,21 @@ test.each([1, 2, 10])( const creation = stopGitCommandMetrics(); expect(response?.workspace?.name).toBe("created-during-measurement"); - expect(creation.submitted).toBe(90); expect(countGitOperations(creation.submissions)).toEqual({ branch: 4, - config: 19, - diff: 2, - "for-each-ref": 6, - "ls-files": 3, - "merge-base": 2, - "rev-list": 2, - "rev-parse": 34, - "show-ref": 6, - status: 6, + config: 22, + diff: 3, + "for-each-ref": 7, + "ls-files": 4, + "merge-base": 3, + "rev-list": 3, + "rev-parse": 38, + "show-ref": 7, + status: 7, "symbolic-ref": 4, - worktree: 2, + worktree: 1, }); + expect(creation.submitted).toBe(103); const existingWorktrees = new Set(fixture.siblingWorktrees.map((cwd) => realpathSync(cwd))); expect( creation.submissions.filter((command) => existingWorktrees.has(realpathSync(command.cwd))), diff --git a/packages/server/src/server/workspace-git-service.observation.integration.test.ts b/packages/server/src/server/workspace-git-service.observation.integration.test.ts index 95d490e91..eb54262b7 100644 --- a/packages/server/src/server/workspace-git-service.observation.integration.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.integration.test.ts @@ -180,7 +180,6 @@ test("recursive observation updates tracked state and prunes ignored storms", as getCheckoutWorktreeState, getCheckoutDiff, runGitCommand, - getWorkspaceGitSelfHealPhaseMs: () => 7_000, } as never, }); const diffManager = new CheckoutDiffManager({ @@ -320,7 +319,7 @@ test("recursive observation updates tracked state and prunes ignored storms", as editedDuringIgnoreUpdate = true; writeFileSync(newlyTrackedPath, "first\nsecond\n"); }; - writeFileSync(path.join(repoDir, ".git", "index"), "force-added\n"); + writeFileSync(path.join(repoDir, ".gitignore"), "cache/\n"); await vi.waitFor( () => { expect(editedDuringIgnoreUpdate).toBe(true); 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 165435a1f..e863ebb4a 100644 --- a/packages/server/src/server/workspace-git-service.observation.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.test.ts @@ -191,6 +191,11 @@ function createService( exitCode: 0, signal: null, })), + createWatcherLivenessCanary: vi.fn(() => ({ + path: "", + filterEvents: (events) => events, + verify: vi.fn(async () => {}), + })), ...overrides, } as never, }); @@ -1583,6 +1588,47 @@ describe("WorkspaceGitService checkout observation", () => { service.dispose(); }); + test("repository watcher canary timeout closes the subscription and starts fallback polling", async () => { + const watcher = createWatcherHarness(); + const logger = createLogger(); + const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); + const verifyCanary = vi.fn(async () => { + throw new Error("watcher did not report its liveness canary"); + }); + const service = createService( + watcher, + { + getCheckoutStatus, + createWatcherLivenessCanary: vi.fn(() => ({ + path: path.join(GIT_DIR, "paseo", ".watcher-canary-timeout"), + filterEvents: (events: WatchEvent[]) => events, + verify: verifyCanary, + })), + }, + logger, + ); + const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); + + await vi.waitFor(() => { + expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); + }); + const repositoryWatcher = getWatcherRecordsForDirectory(watcher, GIT_DIR)[0]; + expect(repositoryWatcher?.unsubscribe).toHaveBeenCalledTimes(1); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ repoGitRoot: GIT_DIR }), + "Failed to start repository metadata watcher; using degraded polling", + ); + + 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(); + }); + test("setup-time watcher error defers recovery until subscribe settles", async () => { const watcher = createWatcherHarness(); const openedSubscription = createDeferred<{ @@ -1862,7 +1908,7 @@ describe("WorkspaceGitService checkout observation", () => { expect(checkoutWatcher?.ignore).toContain(path.join(REPO_CWD, "node_modules")); ignoredDirectories = "node_modules/\nbuild/\n"; - await vi.advanceTimersByTimeAsync(1_000); + checkoutWatcher?.callback(null, [{ path: path.join(REPO_CWD, ".gitignore"), type: "update" }]); await vi.waitFor(() => { expect(runGitCommand).toHaveBeenCalledTimes(3); expect(checkoutWatcher?.updateIgnore).toHaveBeenCalledTimes(1); @@ -1918,7 +1964,7 @@ describe("WorkspaceGitService checkout observation", () => { const originalWatcher = watcher.records.find((record) => record.directory === REPO_CWD); ignoredDirectories = "node_modules/\n"; - await vi.advanceTimersByTimeAsync(1_000); + originalWatcher?.callback(null, [{ path: path.join(REPO_CWD, ".gitignore"), type: "update" }]); await vi.waitFor(() => { expect(originalWatcher?.updateIgnore).toHaveBeenCalledTimes(1); expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); @@ -1940,6 +1986,93 @@ describe("WorkspaceGitService checkout observation", () => { service.dispose(); }); + test("repository exclude changes update working-tree filtering", async () => { + const watcher = createWatcherHarness(); + let ignoredDirectories = "node_modules/\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 service = createService(watcher, { 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]; + const repositoryWatcher = getWatcherRecordsForDirectory(watcher, GIT_DIR)[0]; + ignoredDirectories = "node_modules/\nbuild/\n"; + + repositoryWatcher?.callback(null, [ + { path: path.join(GIT_DIR, "info", "exclude"), type: "update" }, + ]); + await vi.waitFor(() => { + expect(checkoutWatcher?.updateIgnore).toHaveBeenCalledTimes(1); + }); + expect(checkoutWatcher?.ignore).toContain(path.join(REPO_CWD, "build")); + + subscription.unsubscribe(); + service.dispose(); + }); + + test("ignore events during a refresh trigger a trailing recomputation", async () => { + const watcher = createWatcherHarness(); + const blockedRefresh = createDeferred<{ + stdout: string; + stderr: string; + truncated: boolean; + exitCode: number; + signal: null; + }>(); + let lsFilesCallCount = 0; + const runGitCommand = vi.fn(async (args: string[]) => { + if (args[0] === "rev-parse") { + return { + stdout: `${REPO_CWD}\n`, + stderr: "", + truncated: false, + exitCode: 0, + signal: null, + }; + } + lsFilesCallCount += 1; + if (lsFilesCallCount === 2) return blockedRefresh.promise; + return { + stdout: lsFilesCallCount === 1 ? "node_modules/\n" : "node_modules/\nbuild/\n", + stderr: "", + truncated: false, + exitCode: 0, + signal: null, + }; + }); + const service = createService(watcher, { 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(null, [{ path: path.join(REPO_CWD, ".gitignore"), type: "update" }]); + await vi.waitFor(() => expect(lsFilesCallCount).toBe(2)); + checkoutWatcher?.callback(null, [{ path: path.join(REPO_CWD, ".gitignore"), type: "update" }]); + blockedRefresh.resolve({ + stdout: "node_modules/\n", + stderr: "", + truncated: false, + exitCode: 0, + signal: null, + }); + + await vi.waitFor(() => expect(lsFilesCallCount).toBe(3)); + expect(checkoutWatcher?.ignore).toContain(path.join(REPO_CWD, "build")); + + subscription.unsubscribe(); + service.dispose(); + }); + test("ignore watcher update failure enters polling with a distinct reason", async () => { const watcher = createWatcherHarness(); const logger = createLogger(); @@ -1971,7 +2104,7 @@ describe("WorkspaceGitService checkout observation", () => { checkoutWatcher?.updateIgnore.mockRejectedValueOnce(new Error("update failed")); ignoredDirectories = "node_modules/\n"; - await vi.advanceTimersByTimeAsync(1_000); + checkoutWatcher?.callback(null, [{ path: path.join(REPO_CWD, ".gitignore"), type: "update" }]); await vi.waitFor(() => { expect(logger.warn).toHaveBeenCalledWith( expect.objectContaining({ reason: "watcher_update_failed" }), @@ -1993,7 +2126,7 @@ describe("WorkspaceGitService checkout observation", () => { service.dispose(); }); - test("a missed watcher event converges through the deterministic audit", async () => { + test("a working-tree watcher event refreshes summary and diff subscribers", async () => { const watcher = createWatcherHarness(); let isDirty = false; let diffStat = { additions: 0, deletions: 0 }; @@ -2010,7 +2143,6 @@ describe("WorkspaceGitService checkout observation", () => { getCheckoutStatus, getCheckoutShortstat, getCheckoutDiff, - getWorkspaceGitSelfHealPhaseMs: () => 10_000, }); const diffManager = new CheckoutDiffManager({ logger: createLogger(), @@ -2033,7 +2165,10 @@ describe("WorkspaceGitService checkout observation", () => { diffStat = { additions: 7, deletions: 3 }; diffFiles = [{ path: "tracked.txt", additions: 7, deletions: 3, status: "modified" }]; - await vi.advanceTimersByTimeAsync(70_000); + getWatcherRecordsForDirectory(watcher, REPO_CWD)[0]?.callback(null, [ + { path: path.join(REPO_CWD, "tracked.txt"), type: "update" }, + ]); + await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { expect(listener).toHaveBeenLastCalledWith( expect.objectContaining({ diff --git a/packages/server/src/server/workspace-git-service.primitive.test.ts b/packages/server/src/server/workspace-git-service.primitive.test.ts index 017ed831d..4cccaa6e5 100644 --- a/packages/server/src/server/workspace-git-service.primitive.test.ts +++ b/packages/server/src/server/workspace-git-service.primitive.test.ts @@ -370,6 +370,11 @@ function buildDefaultServiceDeps() { signal: null, })), getWorkspaceGitSelfHealPhaseMs: vi.fn(() => 30_000), + createWatcherLivenessCanary: vi.fn(() => ({ + path: "", + filterEvents: (events) => events, + verify: vi.fn(async () => {}), + })), now: () => new Date("2026-04-12T00:00:00.000Z"), }; } @@ -867,7 +872,7 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { service.dispose(); }); - test("self-heal timer refreshes git without refreshing GitHub", async () => { + test("quiet observed workspaces do not refresh git on the observation re-ensure timer", async () => { let nowMs = 0; const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const getPullRequestStatus = vi.fn(async () => createPullRequestStatusResult()); @@ -886,7 +891,7 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { await vi.advanceTimersByTimeAsync(120_000); await flushPromises(); - expect(getCheckoutStatus).toHaveBeenCalledTimes(2); + expect(getCheckoutStatus).toHaveBeenCalledTimes(1); expect(getPullRequestStatus).toHaveBeenCalledTimes(1); expect(getPullRequestStatus).toHaveBeenCalledWith( REPO_CWD, @@ -899,50 +904,12 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { service.dispose(); }); - test("self-heal phase is stable across process restarts", () => { + test("observation re-ensure phase is stable across process restarts", () => { expect(getWorkspaceGitSelfHealPhaseMs("/tmp/repo")).toBe(54_185); expect(getWorkspaceGitSelfHealPhaseMs("/tmp/staggered-repo")).toBe(9_817); }); - test("self-heal staggers workspaces and settles into a 120 second cadence", async () => { - vi.setSystemTime(0); - const otherRepoCwd = resolvePath("/tmp/staggered-repo"); - const refreshes = new Map(); - const getCheckoutStatus = vi.fn(async (cwd: string) => { - const timestamps = refreshes.get(cwd) ?? []; - timestamps.push(Date.now()); - refreshes.set(cwd, timestamps); - return createCheckoutStatus(cwd, { hasRemote: false, remoteUrl: null }); - }); - const service = createService({ - getCheckoutSnapshotFacts: vi.fn(async (cwd: string) => - createCheckoutFacts(cwd, { remoteUrl: null }), - ), - getCheckoutStatus, - getWorkspaceGitSelfHealPhaseMs: (cwd) => (cwd === REPO_CWD ? 10_000 : 40_000), - now: () => new Date(Date.now()), - }); - const first = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); - const second = service.registerWorkspace({ cwd: otherRepoCwd }, vi.fn()); - await Promise.all([service.getSnapshot(REPO_CWD), service.getSnapshot(otherRepoCwd)]); - - await vi.advanceTimersByTimeAsync(300_000); - await flushPromises(); - - const firstSelfHeals = refreshes.get(REPO_CWD)?.slice(1) ?? []; - const secondSelfHeals = refreshes.get(otherRepoCwd)?.slice(1) ?? []; - expect(firstSelfHeals).toHaveLength(2); - expect(secondSelfHeals).toHaveLength(2); - expect(firstSelfHeals[0]).not.toBe(secondSelfHeals[0]); - expect(firstSelfHeals[1] - firstSelfHeals[0]).toBe(120_000); - expect(secondSelfHeals[1] - secondSelfHeals[0]).toBe(120_000); - - first.unsubscribe(); - second.unsubscribe(); - service.dispose(); - }); - - test("self-heal retries observation setup even when the git snapshot is still recent", async () => { + test("observation re-ensure retries setup even when the git snapshot is still recent", async () => { let nowMs = 0; const getCheckoutSnapshotFacts = vi .fn<(cwd: string) => Promise>() @@ -1483,7 +1450,7 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { service.dispose(); }); - test("multiple subscribers on the same target share one self-heal timer", async () => { + test("multiple subscribers on the same target do not duplicate periodic git work", async () => { let nowMs = 0; const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd, { remoteUrl: null }), @@ -1504,14 +1471,14 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { await vi.advanceTimersByTimeAsync(120_000); await flushPromises(); - expect(getCheckoutStatus).toHaveBeenCalledTimes(2); + expect(getCheckoutStatus).toHaveBeenCalledTimes(1); first.unsubscribe(); second.unsubscribe(); service.dispose(); }); - test("unsubscribe with no remaining subscribers clears the self-heal timer", async () => { + test("unsubscribe with no remaining subscribers clears the observation re-ensure timer", async () => { let nowMs = 0; const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const service = createService({ @@ -1531,7 +1498,7 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { service.dispose(); }); - test("service disposal clears all self-heal timers", async () => { + test("service disposal clears all observation re-ensure timers", async () => { let nowMs = 0; const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const service = createService({ @@ -1547,42 +1514,6 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { expect(getCheckoutStatus).toHaveBeenCalledTimes(0); }); - - test("direct getSnapshot returns current snapshot during a self-heal refresh", async () => { - let nowMs = 0; - const selfHealRefresh = createDeferred(); - const getCheckoutStatus = vi - .fn<() => Promise>() - .mockImplementationOnce(async () => createCheckoutStatus(REPO_CWD)) - .mockImplementationOnce(async () => selfHealRefresh.promise); - const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => - createCheckoutFacts(cwd, { remoteUrl: null }), - ); - const service = createService({ - getCheckoutSnapshotFacts, - getCheckoutStatus, - now: () => new Date(nowMs), - }); - const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); - await service.getSnapshot(REPO_CWD); - - nowMs = 120_000; - await vi.advanceTimersByTimeAsync(120_000); - await flushPromises(); - const directRead = service.getSnapshot(REPO_CWD); - await flushPromises(); - - expect(getCheckoutStatus).toHaveBeenCalledTimes(2); - await expect(directRead).resolves.toEqual(createSnapshot(REPO_CWD)); - - selfHealRefresh.resolve(createCheckoutStatus(REPO_CWD)); - await flushPromises(); - - expect(getCheckoutStatus).toHaveBeenCalledTimes(2); - - subscription.unsubscribe(); - service.dispose(); - }); }); describe("WorkspaceGitServiceImpl D2 read methods", () => { diff --git a/packages/server/src/server/workspace-git-service.test.ts b/packages/server/src/server/workspace-git-service.test.ts index 914fde1ff..a1d6b392e 100644 --- a/packages/server/src/server/workspace-git-service.test.ts +++ b/packages/server/src/server/workspace-git-service.test.ts @@ -293,6 +293,11 @@ function buildDefaultTestServiceDeps() { signal: null, })), getWorkspaceGitSelfHealPhaseMs: vi.fn(() => 30_000), + createWatcherLivenessCanary: vi.fn(() => ({ + path: "", + filterEvents: (events) => events, + verify: vi.fn(async () => {}), + })), now: () => new Date("2026-04-12T00:00:00.000Z"), }; } diff --git a/packages/server/src/server/workspace-git-service.ts b/packages/server/src/server/workspace-git-service.ts index 33fddb1d8..6b39ddc1b 100644 --- a/packages/server/src/server/workspace-git-service.ts +++ b/packages/server/src/server/workspace-git-service.ts @@ -1,6 +1,6 @@ import { createHash } from "node:crypto"; import { readFile } from "node:fs/promises"; -import { join, resolve } from "node:path"; +import { basename, join, resolve } from "node:path"; import { LRUCache } from "lru-cache"; import pLimit from "p-limit"; import type pino from "pino"; @@ -68,11 +68,12 @@ import { createFileObserver, } from "./file-observer/index.js"; import { checkoutLiteFromGitSnapshot } from "./workspace-registry-model.js"; +import { createWatcherLivenessCanary } from "./watcher-liveness-canary.js"; const WORKSPACE_GIT_WATCH_DEBOUNCE_MS = 1_000; const BACKGROUND_GIT_FETCH_INTERVAL_MS = 180_000; const FETCH_METADATA_ECHO_TTL_MS = 5_000; -export const WORKSPACE_GIT_SELF_HEAL_INTERVAL_MS = 60_000; +export const WORKSPACE_GIT_OBSERVATION_REENSURE_INTERVAL_MS = 60_000; 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; @@ -108,12 +109,16 @@ function mergeSets( return { merged, added }; } -export function getWorkspaceGitSelfHealPhaseMs(cwd: string): number { +export function getWorkspaceGitObservationReensurePhaseMs(cwd: string): number { return ( - createHash("sha256").update(cwd).digest().readUInt32BE(0) % WORKSPACE_GIT_SELF_HEAL_INTERVAL_MS + createHash("sha256").update(cwd).digest().readUInt32BE(0) % + WORKSPACE_GIT_OBSERVATION_REENSURE_INTERVAL_MS ); } +// Kept for local diagnostic fixtures that only use the stable phase calculation. +export const getWorkspaceGitSelfHealPhaseMs = getWorkspaceGitObservationReensurePhaseMs; + export interface WorkspaceGitRuntimeSnapshot { cwd: string; git: { @@ -353,7 +358,8 @@ interface WorkspaceGitServiceDependencies { observer: WorkspaceGitFetchObserver, ) => Promise; runGitCommand: typeof runGitCommand; - getWorkspaceGitSelfHealPhaseMs: typeof getWorkspaceGitSelfHealPhaseMs; + getWorkspaceGitSelfHealPhaseMs: typeof getWorkspaceGitObservationReensurePhaseMs; + createWatcherLivenessCanary: typeof createWatcherLivenessCanary; now: () => Date; } @@ -387,15 +393,12 @@ interface WorkspaceGitTarget { workingTreeWatchTarget: WorkingTreeWatchTarget | null; debounceTimer: NodeJS.Timeout | null; pendingDebounceRequest: WorkspaceGitRefreshRequest | null; - selfHealTimer: NodeJS.Timeout | null; + observationReensureTimer: NodeJS.Timeout | null; forgePrStatusPollSubscription: { unsubscribe: () => void } | null; forgePrStatusPollKey: string | null; refreshState: WorkspaceGitRefreshState; latestGit: WorkspaceGitRuntimeSnapshot["git"] | null; latestGitLoadedAtMs: number | null; - latestStructuralRefreshAtMs: number | null; - latestWorktreeRefreshAtMs: number | null; - auditWindowStartedAtMs: number | null; latestForge: WorkspaceGitRuntimeSnapshot["forge"] | null; latestForgeLoadedAtMs: number | null; latestSnapshot: WorkspaceGitRuntimeSnapshot | null; @@ -443,6 +446,7 @@ interface WorkingTreeWatchTarget { subscription: FileObserverSubscription | null; ignoredDirectories: Set; ignoredDirectoriesRefreshPromise: Promise | null; + ignoredDirectoriesRefreshRequested: boolean; aliases: Set; workspaceKeys: Set; fallbackPolling: boolean; @@ -500,7 +504,8 @@ function buildDefaultWorkspaceGitServiceDeps( hasOriginRemote, runGitFetch: fetchWorkspaceGitRemote, runGitCommand, - getWorkspaceGitSelfHealPhaseMs, + getWorkspaceGitSelfHealPhaseMs: getWorkspaceGitObservationReensurePhaseMs, + createWatcherLivenessCanary, now: () => new Date(), }; } @@ -1130,15 +1135,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { workingTreeWatchTarget: null, debounceTimer: null, pendingDebounceRequest: null, - selfHealTimer: null, + observationReensureTimer: null, forgePrStatusPollSubscription: null, forgePrStatusPollKey: null, refreshState: { status: "idle" }, latestGit: null, latestGitLoadedAtMs: null, - latestStructuralRefreshAtMs: null, - latestWorktreeRefreshAtMs: null, - auditWindowStartedAtMs: null, latestForge: null, latestForgeLoadedAtMs: null, latestSnapshot: null, @@ -1295,6 +1297,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { subscription: null, ignoredDirectories, ignoredDirectoriesRefreshPromise: null, + ignoredDirectoriesRefreshRequested: false, aliases: new Set([cwd]), workspaceKeys: new Set(), fallbackPolling: false, @@ -1438,6 +1441,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { if (watcherErrored) { return; } + if (events.some((event) => basename(event.path) === ".gitignore")) { + void this.refreshWorkingTreeIgnoredDirectories(target); + } if (!this.hasRelevantWorkingTreeEvent(target, events)) { return; } @@ -1629,22 +1635,28 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { if (target.closed || target.repoRoot === null) { return Promise.resolve(); } + target.ignoredDirectoriesRefreshRequested = true; if (target.ignoredDirectoriesRefreshPromise) { return target.ignoredDirectoriesRefreshPromise; } - const refreshPromise = this.replaceWorkingTreeIgnoredDirectories(target) - .catch((error) => { - this.logger.warn( - { err: error, cwd: target.cwd }, - "Failed to refresh working tree watcher ignore paths", - ); - }) - .finally(() => { - if (target.ignoredDirectoriesRefreshPromise === refreshPromise) { - target.ignoredDirectoriesRefreshPromise = null; + const refreshPromise = (async () => { + while (!target.closed && target.ignoredDirectoriesRefreshRequested) { + target.ignoredDirectoriesRefreshRequested = false; + try { + await this.replaceWorkingTreeIgnoredDirectories(target); + } catch (error) { + this.logger.warn( + { err: error, cwd: target.cwd }, + "Failed to refresh working tree watcher ignore paths", + ); } - }); + } + })().finally(() => { + if (target.ignoredDirectoriesRefreshPromise === refreshPromise) { + target.ignoredDirectoriesRefreshPromise = null; + } + }); target.ignoredDirectoriesRefreshPromise = refreshPromise; return refreshPromise; } @@ -1822,6 +1834,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { join(target.repoGitRoot, path), ); const matchesRepoGitRoot = createRealpathAwarePathMatcher(target.repoGitRoot); + const canary = this.deps.createWatcherLivenessCanary(target.repoGitRoot); + let openedSubscription: FileObserverSubscription | null = null; let watcherErrored = false; let subscribeSettled = false; const markSubscribeSettled = () => { @@ -1834,6 +1848,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { const subscription = await this.subscribeWithDeadline( target.repoGitRoot, (error, events) => { + const liveEvents = canary.filterEvents(events); if (error) { if (watcherErrored) { return; @@ -1853,7 +1868,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { if (watcherErrored) { return; } - const relevantEvents = events.filter( + if (liveEvents.length === 0) { + return; + } + const relevantEvents = liveEvents.filter( (event) => !matchesRepoGitRoot(event.path) && ignore.every((ignoredPath) => !isRealpathInsideRoot(ignoredPath, event.path)), @@ -1869,6 +1887,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }) : relevantEvents; if (immediateEvents.length > 0) { + this.refreshWorkingTreeIgnoresFromRepoMetadataEvents(target, immediateEvents); const routedRefreshes = this.routeRepoMetadataEvents(target, immediateEvents); this.scheduleRepoMetadataRefresh( target, @@ -1882,6 +1901,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { { ignore }, markSubscribeSettled, ); + openedSubscription = subscription; + await canary.verify(this.disposeController.signal); if (watcherErrored) { await this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot); return false; @@ -1904,6 +1925,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } return true; } catch (error) { + if (openedSubscription) { + await this.unsubscribeWatcherSubscription(openedSubscription, target.repoGitRoot); + } if (watcherErrored) { return false; } @@ -1974,6 +1998,41 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return refreshes; } + private refreshWorkingTreeIgnoresFromRepoMetadataEvents( + target: RepoGitTarget, + events: FileChange[], + ): void { + const workspaceKeys = new Set(); + for (const event of events) { + const commonRelativePath = getRealpathAwareRelativePath( + target.repoGitRoot, + event.path, + )?.replaceAll("\\", "/"); + if (commonRelativePath === "config" || commonRelativePath === "info/exclude") { + for (const workspaceKey of target.workspaceKeys) workspaceKeys.add(workspaceKey); + continue; + } + for (const workspaceKey of target.workspaceKeys) { + const facts = this.workspaceTargets.get(workspaceKey)?.latestFacts; + if ( + facts?.isGit && + facts.absoluteGitDir && + getRealpathAwareRelativePath(facts.absoluteGitDir, event.path)?.replaceAll("\\", "/") === + "config.worktree" + ) { + workspaceKeys.add(workspaceKey); + } + } + } + for (const workspaceKey of workspaceKeys) { + const workspaceTarget = this.workspaceTargets.get(workspaceKey); + const workingTreeTarget = workspaceTarget + ? this.getWorkingTreeWatchTargetForWorkspace(workspaceTarget) + : null; + if (workingTreeTarget) void this.refreshWorkingTreeIgnoredDirectories(workingTreeTarget); + } + } + private isFetchRemoteMetadataEvent(target: RepoGitTarget, event: FileChange): boolean { const relativePath = getRealpathAwareRelativePath(target.repoGitRoot, event.path); const effect = classifyGitMetadataPath("common", relativePath ?? ""); @@ -2309,57 +2368,20 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private startWorkspaceSubscriptionTimers(target: WorkspaceGitTarget): void { - if (!target.selfHealTimer) { - target.auditWindowStartedAtMs = this.deps.now().getTime(); - const runSelfHealTick = () => { + if (!target.observationReensureTimer) { + const reensureObservation = () => { if (!this.isActiveObservedWorkspaceTarget(target)) { - target.selfHealTimer = null; + target.observationReensureTimer = null; return; } - target.selfHealTimer = setTimeout(runSelfHealTick, WORKSPACE_GIT_SELF_HEAL_INTERVAL_MS); + target.observationReensureTimer = setTimeout( + reensureObservation, + WORKSPACE_GIT_OBSERVATION_REENSURE_INTERVAL_MS, + ); this.scheduleWorkspaceObservationSetup(target); - const auditWindowStartedAtMs = target.auditWindowStartedAtMs; - target.auditWindowStartedAtMs = this.deps.now().getTime(); - if (auditWindowStartedAtMs === null) { - return; - } - const workingTreeTarget = this.getWorkingTreeWatchTargetForWorkspace(target); - if (workingTreeTarget) { - void this.refreshWorkingTreeIgnoredDirectories(workingTreeTarget); - } - const refreshStructure = - target.latestStructuralRefreshAtMs === null || - target.latestStructuralRefreshAtMs < auditWindowStartedAtMs; - const refreshWorktree = - refreshStructure || - target.latestWorktreeRefreshAtMs === null || - target.latestWorktreeRefreshAtMs < auditWindowStartedAtMs; - if (!refreshStructure && !refreshWorktree) { - return; - } - if (refreshWorktree) { - if (workingTreeTarget) { - this.notifyWorkingTreeConsumers(workingTreeTarget); - } - } - this.refreshWorkspaceTarget(target, { - force: false, - refreshStructure, - refreshWorktree, - includeForge: false, - reason: "self-heal-git", - notify: true, - queueIfBusy: true, - movedRemoteRefs: new Set(), - }).catch((error) => { - this.logger.warn( - { err: error, cwd: target.cwd, reason: "self-heal-git" }, - "Failed to run workspace git self-heal refresh", - ); - }); }; - target.selfHealTimer = setTimeout( - runSelfHealTick, + target.observationReensureTimer = setTimeout( + reensureObservation, this.deps.getWorkspaceGitSelfHealPhaseMs(target.cwd), ); } @@ -2891,7 +2913,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }; const loadedAtMs = this.deps.now().getTime(); target.latestGitLoadedAtMs = loadedAtMs; - target.latestWorktreeRefreshAtMs = loadedAtMs; } private async refreshGitSnapshot( @@ -2915,8 +2936,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { target.latestGit = buildNotGitSnapshot(cwd).git; const loadedAtMs = this.deps.now().getTime(); target.latestGitLoadedAtMs = loadedAtMs; - target.latestStructuralRefreshAtMs = loadedAtMs; - target.latestWorktreeRefreshAtMs = loadedAtMs; target.latestForge = buildForgeUnavailableSnapshot(); target.latestForgeLoadedAtMs = target.latestGitLoadedAtMs; return facts; @@ -2951,10 +2970,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }; const loadedAtMs = this.deps.now().getTime(); target.latestGitLoadedAtMs = loadedAtMs; - target.latestStructuralRefreshAtMs = loadedAtMs; - if (refreshWorktree) { - target.latestWorktreeRefreshAtMs = loadedAtMs; - } if (previousForgePrStatusPollKey !== this.getForgePrStatusPollKey(target)) { target.latestForge = buildForgeUnavailableSnapshot(); @@ -3266,9 +3281,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { clearTimeout(target.debounceTimer); target.debounceTimer = null; } - if (target.selfHealTimer) { - clearTimeout(target.selfHealTimer); - target.selfHealTimer = null; + if (target.observationReensureTimer) { + clearTimeout(target.observationReensureTimer); + target.observationReensureTimer = null; } this.stopForgePrStatusPollForTarget(target); target.listeners.clear();