From e2de050dab781487876d718f5614298e84736a55 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 26 Oct 2026 15:29:32 +0100 Subject: [PATCH] fix(server): wait for healthy reconciliation inspection capacity --- .../workspace-reconciliation-service.test.ts | 65 +++++++++++++++++++ .../workspace-reconciliation-service.ts | 27 +++++++- 2 files changed, 89 insertions(+), 3 deletions(-) diff --git a/packages/server/src/server/workspace-reconciliation-service.test.ts b/packages/server/src/server/workspace-reconciliation-service.test.ts index 2cb40dd782..e1a8860e3b 100644 --- a/packages/server/src/server/workspace-reconciliation-service.test.ts +++ b/packages/server/src/server/workspace-reconciliation-service.test.ts @@ -404,6 +404,71 @@ describe("WorkspaceReconciliationService", () => { } }); + test("inspects every healthy path when one hung inspection times out during a longer scan", async () => { + const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-long-hung-"))); + tempDirs.push(rootPath); + const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries(); + projects.set( + "p1", + createPersistedProjectRecord({ + projectId: "p1", + rootPath, + projectKey: canonicalLocalProjectKey(rootPath), + kind: "non_git", + displayName: "project", + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + const healthyIds = Array.from({ length: 12 }, (_, i) => `healthy${i}`); + for (const workspaceId of ["hung", ...healthyIds]) { + workspaces.set( + workspaceId, + createPersistedWorkspaceRecord({ + workspaceId, + projectId: "p1", + cwd: path.join(rootPath, workspaceId), + kind: "directory", + displayName: workspaceId, + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + } + const calls: string[] = []; + let outstanding = 0; + let maximumOutstanding = 0; + const { logger, warnRecords } = createCapturingLogger(); + const service = new WorkspaceReconciliationService({ + projectRegistry, + workspaceRegistry, + logger, + inspectionTimeoutMs: 100, + inspectDirectory: async (targetPath) => { + calls.push(targetPath); + maximumOutstanding = Math.max(maximumOutstanding, ++outstanding); + if (path.basename(targetPath) === "hung") return new Promise(() => {}); + await new Promise((resolve) => setTimeout(resolve, 20)); + outstanding--; + return "directory"; + }, + }); + try { + const result = await service.runOnce(); + expect(result.durationMs).toBeGreaterThan(100); + expect(calls).toEqual( + ["hung", ...healthyIds].map((id) => path.join(rootPath, id)).concat(rootPath), + ); + expect(maximumOutstanding).toBe(2); + expect(outstanding).toBe(1); + expect(workspaces.get("hung")?.archivedAt).toBeNull(); + expect(result.changesApplied).toEqual([]); + expect(warnRecords.filter((record) => record.message.includes("saturated"))).toEqual([]); + } finally { + service.dispose(); + } + }); + test("times out a hung inspection without archiving it or blocking healthy paths, and retains its IO slot", async () => { const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-hung-"))); tempDirs.push(rootPath); diff --git a/packages/server/src/server/workspace-reconciliation-service.ts b/packages/server/src/server/workspace-reconciliation-service.ts index 6225d16ff5..5333da8f83 100644 --- a/packages/server/src/server/workspace-reconciliation-service.ts +++ b/packages/server/src/server/workspace-reconciliation-service.ts @@ -123,6 +123,7 @@ export class WorkspaceReconciliationService { // the deadline to admit more work into libuv's shared threadpool. private readonly timedOutPaths = new Set(); private inspectionSaturated = false; + private readonly inspectionWaiters = new Set<() => void>(); private readonly pendingInspections = new Map>(); private readonly directoryInspection: (targetPath: string) => Promise; private readonly inspectionTimeoutMs: number; @@ -197,6 +198,7 @@ export class WorkspaceReconciliationService { dispose(): void { this.disposed = true; + this.wakeInspectionWaiters(); this.unsubscribeRegistry?.(); this.unsubscribeRegistry = null; if (this.rescanTimer) this.clock.clearInterval(this.rescanTimer); @@ -246,6 +248,9 @@ export class WorkspaceReconciliationService { const allWorkspaces = await this.workspaceRegistry.list(); const activeProjects = allProjects.filter((p) => !p.archivedAt); + const activeProjectsById = new Map( + activeProjects.map((project) => [project.projectId, project]), + ); const activeWorkspaces = allWorkspaces.filter((w) => !w.archivedAt); const workspaceDirectoryStates = await mapBounded(activeWorkspaces, async (workspace) => ({ workspace, @@ -293,7 +298,7 @@ export class WorkspaceReconciliationService { current.projectId !== workspace.projectId || !project || project.archivedAt || - project.rootPath !== activeProjects.find((p) => p.projectId === project.projectId)?.rootPath + project.rootPath !== activeProjectsById.get(project.projectId)?.rootPath ) return; const timestamp = new Date().toISOString(); @@ -579,11 +584,25 @@ export class WorkspaceReconciliationService { return this.workspaceGitService.getCheckout(cwd); } + private wakeInspectionWaiters(): void { + for (const wake of this.inspectionWaiters) wake(); + this.inspectionWaiters.clear(); + } + private async inspectDirectory(targetPath: string): Promise { + // A healthy in-flight inspection owns capacity temporarily. Wait for it; + // only unresolved operations that have all timed out justify skipping paths. + while ( + !this.disposed && + !this.pendingInspections.has(targetPath) && + this.pendingInspections.size >= INSPECTION_CONCURRENCY + ) { + if (this.timedOutPaths.size === INSPECTION_CONCURRENCY) return "unreadable"; + await new Promise((resolve) => this.inspectionWaiters.add(resolve)); + } + if (this.disposed) return "unreadable"; let pending = this.pendingInspections.get(targetPath); if (!pending) { - if (this.pendingInspections.size >= INSPECTION_CONCURRENCY || this.disposed) - return "unreadable"; pending = Promise.resolve() .then(() => this.directoryInspection(targetPath)) .catch((error) => { @@ -597,6 +616,7 @@ export class WorkspaceReconciliationService { .finally(() => { this.pendingInspections.delete(targetPath); this.timedOutPaths.delete(targetPath); + this.wakeInspectionWaiters(); if (this.inspectionSaturated && this.timedOutPaths.size === 0) { this.inspectionSaturated = false; this.logger.info( @@ -614,6 +634,7 @@ export class WorkspaceReconciliationService { new Promise((resolve) => { timer = setTimeout(() => { this.timedOutPaths.add(targetPath); + this.wakeInspectionWaiters(); if (!this.inspectionSaturated && this.timedOutPaths.size === INSPECTION_CONCURRENCY) { this.inspectionSaturated = true; this.logger.warn(