puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-04 07:33:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(server): recover workspace Git watchers after setup timeouts

Browse Source 
* fix(server): recover git watchers after setup timeouts

Setup failures entered degraded polling without scheduling the existing recovery ladder. Retry both watcher types so successful subscriptions stop fallback reads.

* fix(server): isolate callbacks from expired git watchers

An expired subscription can report an error after recovery. Ignore abandoned callbacks and callbacks that no longer own the active subscription so they cannot tear down a replacement.

* fix(server): refresh metadata when git observation recovers

* fix(server): settle watcher recovery during workspace teardown

* fix(server): reconcile exclusions when metadata watching recovers
This commit is contained in:
Mohamed Boudra
2026-11-04 13:10:21 +07:00
committed by GitHub
parent 67628b48fe
commit 0952ddbd4b
2 changed files with 293 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/workspace-git-service.test.ts
+236 -1
View File
@@ -1560,6 +1560,241 @@ describe("WorkspaceGitServiceImpl", () => {
service.dispose();
});
test.each(["repository metadata", "working tree"])(
"recovers after a %s watcher subscription times out during setup",
async (kind) => {
const watchPath = kind === "repository metadata" ? join(REPO_CWD, ".git") : REPO_CWD;
let attempts = 0;
const subscribe = vi.fn(async (directory: string) => {
if (directory === watchPath && ++attempts === 1) {
return new Promise<ReturnType<typeof createAsyncSubscription>>(() => {});
}
return createAsyncSubscription();
});
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService({ subscribe, getCheckoutStatus });
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
try {
await settle(() => expect(attempts).toBe(1));
await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
await waitForObservationToSettle(service, 1);
const readsBeforePoll = getCheckoutStatus.mock.calls.length;
await vi.advanceTimersByTimeAsync(5_000);
await waitForRefreshesToSettle(service);
expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(readsBeforePoll);
await vi.advanceTimersByTimeAsync(24_999);
expect(attempts).toBe(1);
await vi.advanceTimersByTimeAsync(1);
await settle(() => expect(attempts).toBe(2));
// Recovery emits a debounced working-tree refresh before observation takes over.
await vi.advanceTimersByTimeAsync(1_000);
await waitForRefreshesToSettle(service);
const readsAfterRecovery = getCheckoutStatus.mock.calls.length;
await vi.advanceTimersByTimeAsync(20_000);
await waitForRefreshesToSettle(service);
expect(getCheckoutStatus.mock.calls.length).toBe(readsAfterRecovery);
} finally {
subscription.unsubscribe();
await service.dispose();
}
},
15_000,
);
test.each(["repository metadata", "working tree"])(
"ignores expired %s watcher callbacks after recovery",
async (kind) => {
const watchPath = kind === "repository metadata" ? join(REPO_CWD, ".git") : REPO_CWD;
let attempts = 0;
const callbacks: Array<
(error: Error | null, events: Array<{ path: string; type: "update" }>) => void
> = [];
const recovered = createAsyncSubscription();
const subscribe = vi.fn(async (directory: string, callback: (typeof callbacks)[number]) => {
if (directory === watchPath) callbacks.push(callback);
if (directory === watchPath && ++attempts === 1) {
return new Promise<ReturnType<typeof createAsyncSubscription>>(() => {});
}
return directory === watchPath ? recovered : createAsyncSubscription();
});
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService({ subscribe, getCheckoutStatus });
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
try {
await settle(() => expect(attempts).toBe(1));
await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
await waitForObservationToSettle(service, 1);
const readsBeforePoll = getCheckoutStatus.mock.calls.length;
await vi.advanceTimersByTimeAsync(5_000);
await waitForRefreshesToSettle(service);
expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(readsBeforePoll);
await vi.advanceTimersByTimeAsync(24_999);
expect(attempts).toBe(1);
await vi.advanceTimersByTimeAsync(1);
await settle(() => expect(attempts).toBe(2));
// Recovery emits a debounced working-tree refresh before observation takes over.
await vi.advanceTimersByTimeAsync(1_000);
await waitForRefreshesToSettle(service);
const readsAfterRecovery = getCheckoutStatus.mock.calls.length;
callbacks[0]!(new Error("expired watcher stopped"), []);
callbacks[0]!(null, [{ path: join(watchPath, "stale.ts"), type: "update" }]);
expect(recovered.unsubscribe).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(20_000);
await waitForRefreshesToSettle(service);
expect(getCheckoutStatus.mock.calls.length).toBe(readsAfterRecovery);
expect(recovered.unsubscribe).not.toHaveBeenCalled();
} finally {
subscription.unsubscribe();
await service.dispose();
}
},
15_000,
);
test.each(["repository metadata", "working tree"])(
"refreshes changes missed before %s watcher recovery",
async (kind) => {
const watchPath = kind === "repository metadata" ? join(REPO_CWD, ".git") : REPO_CWD;
let attempts = 0;
let changed = false;
const subscribe = vi.fn(async (directory: string) => {
if (directory === watchPath && ++attempts === 1) {
return new Promise<ReturnType<typeof createAsyncSubscription>>(() => {});
}
return createAsyncSubscription();
});
const getCheckoutStatus = vi.fn(async (cwd: string) =>
createCheckoutStatus(cwd, {
isDirty: changed,
currentBranch: changed ? "feature" : "main",
}),
);
const service = createService({
subscribe,
getCheckoutStatus,
getCheckoutSnapshotFacts: async (cwd) => ({
...createCheckoutSnapshotFacts(cwd),
currentBranch: changed ? "feature" : "main",
}),
});
const snapshots: WorkspaceGitRuntimeSnapshot[] = [];
service.onSnapshotUpdated((snapshot) => snapshots.push(snapshot));
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
try {
await settle(() => expect(attempts).toBe(1));
await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
await waitForObservationToSettle(service, 1);
await vi.advanceTimersByTimeAsync(29_999);
await waitForRefreshesToSettle(service);
expect(snapshots.at(-1)?.git).toMatchObject({ currentBranch: "main", isDirty: false });
changed = true;
await vi.advanceTimersByTimeAsync(1);
await settle(() => expect(attempts).toBe(2));
await vi.advanceTimersByTimeAsync(1_000);
await waitForRefreshesToSettle(service);
expect(snapshots.at(-1)?.git).toMatchObject(
kind === "repository metadata" ? { currentBranch: "feature" } : { isDirty: true },
);
const readsAfterRecovery = getCheckoutStatus.mock.calls.length;
await vi.advanceTimersByTimeAsync(20_000);
await waitForRefreshesToSettle(service);
expect(getCheckoutStatus.mock.calls.length).toBe(readsAfterRecovery);
} finally {
subscription.unsubscribe();
await service.dispose();
}
},
);
test.each(["repository metadata", "working tree"])(
"settles a pending %s watcher retry after the final consumer leaves",
async (kind) => {
const watchPath = kind === "repository metadata" ? join(REPO_CWD, ".git") : REPO_CWD;
let attempts = 0;
const subscribe = vi.fn(async (directory: string) => {
if (directory === watchPath) {
attempts += 1;
return new Promise<ReturnType<typeof createAsyncSubscription>>(() => {});
}
return createAsyncSubscription();
});
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService({ subscribe, getCheckoutStatus });
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
try {
await settle(() => expect(attempts).toBe(1));
await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
await waitForObservationToSettle(service, 1);
await vi.advanceTimersByTimeAsync(30_000);
await settle(() => expect(attempts).toBe(2));
subscription.unsubscribe();
await waitForRefreshesToSettle(service);
const readsAfterUnsubscribe = getCheckoutStatus.mock.calls.length;
// Vitest reports any unhandled rejection as a suite failure.
await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
await vi.advanceTimersByTimeAsync(300_000);
expect(attempts).toBe(2);
expect(vi.getTimerCount()).toBe(0);
expect(getCheckoutStatus.mock.calls.length).toBe(readsAfterUnsubscribe);
} finally {
subscription.unsubscribe();
await service.dispose();
}
},
);
test("metadata recovery reconciles exclusions changed during the outage", async () => {
let ignoredDirectories = "generated/\n";
let attempts = 0;
let workingTreeCallback!: (
error: Error | null,
events: Array<{ path: string; type: "delete" }>,
) => void;
const workingTreeSubscription = createAsyncSubscription();
const subscribe = vi.fn(async (directory: string, callback: typeof workingTreeCallback) => {
if (directory === join(REPO_CWD, ".git") && ++attempts === 1) {
return new Promise<ReturnType<typeof createAsyncSubscription>>(() => {});
}
if (directory === REPO_CWD) {
workingTreeCallback = callback;
return workingTreeSubscription;
}
return createAsyncSubscription();
});
const runGitCommand = vi.fn(async (args: string[]) => ({
stdout: args[0] === "rev-parse" ? `${REPO_CWD}\n` : ignoredDirectories,
stderr: "",
truncated: false,
exitCode: 0,
signal: null,
}));
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService({ subscribe, runGitCommand, getCheckoutStatus });
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
try {
await settle(() => expect(attempts).toBe(1));
await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS);
await waitForObservationToSettle(service, 1);
expect(subscribe.mock.calls.find(([directory]) => directory === REPO_CWD)).toBeDefined();
ignoredDirectories = ""; // info/exclude changed while metadata observation was unavailable.
await vi.advanceTimersByTimeAsync(30_000);
await settle(() => expect(attempts).toBe(2));
await vi.advanceTimersByTimeAsync(1_000);
await waitForRefreshesToSettle(service);
const readsAfterRecovery = getCheckoutStatus.mock.calls.length;
workingTreeCallback(null, [
{ path: join(REPO_CWD, "generated", "output.ts"), type: "delete" },
]);
await vi.advanceTimersByTimeAsync(1_000);
await waitForRefreshesToSettle(service);
expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(readsAfterRecovery);
expect(workingTreeSubscription.updateIgnore).toHaveBeenCalledWith([join(REPO_CWD, ".git")]);
} finally {
subscription.unsubscribe();
await service.dispose();
}
});
test("sets a bounded polling fallback when recursive observation is unavailable", async () => {
const subscribe = vi.fn(async () => {
throw new Error("watcher unavailable");
@@ -1567,7 +1802,7 @@ describe("WorkspaceGitServiceImpl", () => {
const service = createService({ subscribe });
const subscription = await service.requestWorkingTreeWatch(REPO_CWD, vi.fn());
expect(vi.getTimerCount()).toBe(1);
expect(vi.getTimerCount()).toBe(2);
subscription.unsubscribe();
service.dispose();
packages/server/src/server/workspace-git-service.ts
+57 -5
View File
@@ -1552,6 +1552,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
options?: { replaceFallback?: boolean },
): Promise<boolean> {
const ignore = this.workingTreeIgnoreList(target);
let abandoned = false;
let activeSubscription: FileObserverSubscription | null = null;
let watcherErrored = false;
let subscribeSettled = false;
const markSubscribeSettled = () => {
@@ -1564,6 +1566,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const subscription = await this.subscribeWithDeadline(
target.watchPath,
(error, events) => {
if (
abandoned ||
target.closed ||
(activeSubscription && target.subscription !== activeSubscription)
) {
return;
}
if (error) {
if (watcherErrored) {
return;
@@ -1597,6 +1606,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
markSubscribeSettled,
);
if (watcherErrored) {
abandoned = true;
await this.unsubscribeWatcherSubscription(subscription, target.watchPath);
return false;
}
@@ -1605,9 +1615,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
(target.fallbackPolling && !options?.replaceFallback) ||
target.subscription
) {
abandoned = true;
await this.unsubscribeWatcherSubscription(subscription, target.watchPath);
return false;
}
activeSubscription = subscription;
target.subscription = subscription;
target.appliedIgnoreList = ignore;
// Do not reset `attemptCount` here — `subscribe()` resolving proves
@@ -1625,11 +1637,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
return true;
} catch (error) {
abandoned = true;
if (watcherErrored) {
return false;
}
if (this.disposed || target.closed) {
throw error;
return false;
}
this.logger.warn(
{ err: error, cwd: target.cwd },
@@ -1638,6 +1651,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
if (!options?.replaceFallback) {
this.startWorkingTreeWatchFallback(target, "watcher_setup_failed");
}
this.scheduleWorkingTreeWatchRecovery(target);
return false;
}
}
@@ -1868,7 +1882,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const delayMs = this.advanceWatchRecoveryLadder(target.recovery);
target.recovery.timer = setTimeout(() => {
target.recovery.timer = null;
void this.recoverWorkingTreeWatch(target);
void this.recoverWorkingTreeWatch(target).catch((error) => {
if (this.disposed || target.closed) {
return;
}
this.logger.warn({ err: error }, "Git watcher recovery failed; retrying with backoff");
this.scheduleWorkingTreeWatchRecovery(target);
});
}, delayMs);
}
@@ -2374,6 +2394,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const matchesRepoGitRoot = paths.matches;
const canary = this.deps.createWatcherLivenessCanary(target.repoGitRoot);
let openedSubscription: FileObserverSubscription | null = null;
let abandoned = false;
let activeSubscription: FileObserverSubscription | null = null;
let watcherErrored = false;
let subscribeSettled = false;
const markSubscribeSettled = () => {
@@ -2386,6 +2408,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const subscription = await this.subscribeWithDeadline(
target.repoGitRoot,
(error, events) => {
if (
abandoned ||
target.closed ||
(activeSubscription && target.subscription !== activeSubscription)
) {
return;
}
const liveEvents = canary.filterEvents(
events.map((event) => ({ ...event, path: paths.toWatchPath(event.path) })),
);
@@ -2444,6 +2473,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
openedSubscription = subscription;
await canary.verify(this.disposeController.signal);
if (watcherErrored) {
abandoned = true;
await this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot);
return false;
}
@@ -2452,9 +2482,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
(target.fallbackPolling && !options?.replaceFallback) ||
this.repoTargets.get(target.repoGitRoot) !== target
) {
abandoned = true;
await this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot);
return false;
}
activeSubscription = subscription;
target.subscription = subscription;
// See `advanceWatchRecoveryLadder`: do not reset `attemptCount` on
// subscribe success alone, only record when this subscription was
@@ -2469,6 +2501,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
return true;
} catch (error) {
abandoned = true;
if (openedSubscription) {
await this.unsubscribeWatcherSubscription(openedSubscription, target.repoGitRoot);
}
@@ -2476,7 +2509,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return false;
}
if (this.disposed || target.closed) {
throw error;
return false;
}
this.logger.warn(
{ err: error, repoGitRoot: target.repoGitRoot },
@@ -2485,6 +2518,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
if (!options?.replaceFallback) {
this.startRepoMetadataFallback(target);
}
this.scheduleRepoMetadataWatchRecovery(target);
return false;
}
}
@@ -2506,7 +2540,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const delayMs = this.advanceWatchRecoveryLadder(target.recovery);
target.recovery.timer = setTimeout(() => {
target.recovery.timer = null;
void this.recoverRepoMetadataWatch(target);
void this.recoverRepoMetadataWatch(target).catch((error) => {
if (this.disposed || target.closed) {
return;
}
this.logger.warn({ err: error }, "Git watcher recovery failed; retrying with backoff");
this.scheduleRepoMetadataWatchRecovery(target);
});
}, delayMs);
}
@@ -2520,6 +2560,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return;
}
this.logger.info({ repoGitRoot: target.repoGitRoot }, "Repository metadata watcher recovered");
await this.refreshWorkingTreeIgnoresForWorkspaces(target.workspaceKeys);
this.scheduleRepoMetadataRefresh(target, "git-metadata-watch-recovered", true);
}
private routeRepoMetadataEvents(
@@ -2561,13 +2603,23 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
}
}
void this.refreshWorkingTreeIgnoresForWorkspaces(workspaceKeys);
}
private async refreshWorkingTreeIgnoresForWorkspaces(
workspaceKeys: Iterable<string>,
): Promise<void> {
const targets = new Set<WorkingTreeWatchTarget>();
for (const workspaceKey of workspaceKeys) {
const workspaceTarget = this.workspaceTargets.get(workspaceKey);
const workingTreeTarget = workspaceTarget
? this.getWorkingTreeWatchTargetForWorkspace(workspaceTarget)
: null;
if (workingTreeTarget) void this.refreshWorkingTreeIgnoredDirectories(workingTreeTarget);
if (workingTreeTarget) targets.add(workingTreeTarget);
}
await Promise.all(
[...targets].map((target) => this.refreshWorkingTreeIgnoredDirectories(target)),
);
}
private isFetchRemoteMetadataEvent(target: RepoGitTarget, event: FileChange): boolean {