Stop idle workspace Git refreshes from starving daemon requests (#3323)

* fix(server): trust healthy workspace git watchers

Quiet observed workspaces were periodically scheduling full Git refreshes, saturating the daemon-wide process pool as workspace count grew. Keep only cheap observation re-ensure work, refresh ignore paths from watcher events, and detect silent metadata watchers with a one-shot canary before accepting the subscription.

* fix(server): refresh watcher ignores from git metadata

Repository exclude and Git configuration changes also affect the standard ignore set. Route those metadata events back to linked working-tree observations so their excluded directory lists stay current without periodic Git polling.

* test(server): drive ignore updates through watcher events

The integration contract now exercises the production event-driven ignore refresh instead of waiting for the removed periodic audit.

* fix(server): normalize git metadata paths on windows

File observation returns platform-native relative separators. Normalize those paths before matching Git metadata names so repository exclusion changes refresh working-tree ignores on Windows too.

* fix(server): trail concurrent ignore refreshes

Remember ignore-source events that arrive during an in-flight Git query and recompute once more before considering the working-tree observer current.
This commit is contained in:
Mohamed Boudra
2026-08-13 22:08:23 +08:00
committed by GitHub
parent 20bccd31a3
commit 1a6054138d
9 changed files with 381 additions and 183 deletions
@@ -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");
});
@@ -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<void>;
}
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<void>((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<never>((_resolve, reject) => {
timeout = setTimeout(() => {
reject(
new Error(
`Watcher for ${watchRoot} did not report its liveness canary within ${timeoutMs}ms`,
),
);
}, timeoutMs);
});
const abortPromise = new Promise<never>((_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 });
}
},
};
}
@@ -153,6 +153,18 @@ async function startObservedFixture(siblingCount: number): Promise<ReturnType<ty
return fixture;
}
test("quiet observed workspaces spawn no git commands beyond the old self-heal interval", async () => {
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))),
@@ -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);
@@ -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({
@@ -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<string, number[]>();
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<CheckoutSnapshotFacts>>()
@@ -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<CheckoutStatusGit>();
const getCheckoutStatus = vi
.fn<() => Promise<CheckoutStatusGit>>()
.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", () => {
@@ -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"),
};
}
@@ -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<T>(
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<WorkspaceGitFetchResult>;
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<string>;
ignoredDirectoriesRefreshPromise: Promise<void> | null;
ignoredDirectoriesRefreshRequested: boolean;
aliases: Set<string>;
workspaceKeys: Set<string>;
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<string>();
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();