diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 53172b2835..4473a4004d 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -358,7 +358,7 @@ function createNoopWorkspaceRegistry(): WorkspaceRegistry { get: async () => null, update: async () => null, upsert: async () => {}, - archive: async () => {}, + archive: async () => false, remove: async () => {}, }; } diff --git a/packages/server/src/server/workspace-reconciliation-observation.test.ts b/packages/server/src/server/workspace-reconciliation-observation.test.ts index 8977641bab..44bb9de08e 100644 --- a/packages/server/src/server/workspace-reconciliation-observation.test.ts +++ b/packages/server/src/server/workspace-reconciliation-observation.test.ts @@ -495,6 +495,10 @@ describe("observed workspace placement", () => { expect((await observed.placement("workspace-one"))?.archivedAt).toEqual(expect.any(String)); expect(observed.workspaceBatches).toEqual([["workspace-one"]]); + const archived = await observed.placement("workspace-one"); + await observed.advanceBy(RESCAN_INTERVAL_MS); + expect(await observed.placement("workspace-one")).toEqual(archived); + expect(observed.workspaceBatches).toEqual([["workspace-one"]]); observed.dispose(); }); diff --git a/packages/server/src/server/workspace-reconciliation-service.test.ts b/packages/server/src/server/workspace-reconciliation-service.test.ts index 52c7d41065..e1a8860e3b 100644 --- a/packages/server/src/server/workspace-reconciliation-service.test.ts +++ b/packages/server/src/server/workspace-reconciliation-service.test.ts @@ -31,6 +31,14 @@ function canonicalLocalProjectKey(rootPath: string): string { } function createTestRegistries() { + let archiveGate: { + started: ReturnType; + release: ReturnType; + } | null = null; + const holdNextArchive = () => { + archiveGate = { started: deferred(), release: deferred() }; + return archiveGate; + }; const projects = new Map(); const workspaces = new Map(); @@ -55,6 +63,13 @@ function createTestRegistries() { projects.set(record.projectId, record); return record; }, + update: async (id, updater) => { + const current = projects.get(id); + if (!current) return null; + const updated = updater(current); + projects.set(id, updated); + return updated; + }, upsert: async (record: PersistedProjectRecord) => { projects.set(record.projectId, record); }, @@ -85,17 +100,23 @@ function createTestRegistries() { workspaces.set(record.workspaceId, record); }, archive: async (id: string, archivedAt: string) => { - const existing = workspaces.get(id); - if (existing) { - workspaces.set(id, { ...existing, archivedAt, updatedAt: archivedAt }); + const gate = archiveGate; + archiveGate = null; + if (gate) { + gate.started.resolve(); + await gate.release.promise; } + const existing = workspaces.get(id); + if (!existing || existing.archivedAt) return false; + workspaces.set(id, { ...existing, archivedAt, updatedAt: archivedAt }); + return true; }, remove: async (id: string) => { workspaces.delete(id); }, }; - return { projects, workspaces, projectRegistry, workspaceRegistry }; + return { projects, workspaces, projectRegistry, workspaceRegistry, holdNextArchive }; } function createTestLogger() { @@ -117,6 +138,7 @@ interface CapturedLogRecord { function createCapturingLogger() { const infoRecords: CapturedLogRecord[] = []; + const warnRecords: CapturedLogRecord[] = []; const logger = { child: () => logger, trace: () => undefined, @@ -124,10 +146,12 @@ function createCapturingLogger() { info: (payload: unknown, message?: string) => { infoRecords.push({ payload, message: message ?? "" }); }, - warn: () => undefined, + warn: (payload: unknown, message?: string) => { + warnRecords.push({ payload, message: message ?? "" }); + }, error: () => undefined, }; - return { logger: logger as unknown as pino.Logger, infoRecords }; + return { logger: logger as unknown as pino.Logger, infoRecords, warnRecords }; } function createWorkspaceGitServiceStub( @@ -241,6 +265,419 @@ describe("WorkspaceReconciliationService", () => { tempDirs.length = 0; }); + test("does not emit a duplicate archive when user archival commits before reconciliation's archive", async () => { + const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-archive-commit-"))); + tempDirs.push(rootPath); + const { projects, workspaces, projectRegistry, workspaceRegistry, holdNextArchive } = + createTestRegistries(); + projects.set( + "p1", + createPersistedProjectRecord({ + projectId: "p1", + rootPath, + projectKey: canonicalLocalProjectKey(rootPath), + kind: "non_git", + displayName: "project", + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + workspaces.set( + "w1", + createPersistedWorkspaceRecord({ + workspaceId: "w1", + projectId: "p1", + cwd: path.join(rootPath, "missing"), + kind: "directory", + displayName: "workspace", + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + const callbacks: string[] = []; + const service = new WorkspaceReconciliationService({ + projectRegistry, + workspaceRegistry, + logger: createTestLogger(), + onWorkspaceArchived: (id) => { + callbacks.push(id); + }, + }); + const gate = holdNextArchive(); + try { + const scan = service.runOnce(); + await gate.started.promise; + expect(await workspaceRegistry.archive("w1", timestamp)).toBe(true); + gate.release.resolve(); + expect((await scan).changesApplied).toEqual([]); + expect(callbacks).toEqual([]); + expect(workspaces.get("w1")?.archivedAt).toBe(timestamp); + } finally { + gate.release.resolve(); + service.dispose(); + } + }); + + test("preserves paths while two inspections saturate capacity, logs once, and recovers after they drain", async () => { + const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-saturated-"))); + tempDirs.push(rootPath); + const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries(); + const { logger, warnRecords, infoRecords } = createCapturingLogger(); + projects.set( + "p1", + createPersistedProjectRecord({ + projectId: "p1", + rootPath, + projectKey: canonicalLocalProjectKey(rootPath), + kind: "non_git", + displayName: "project", + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + const ids = ["hung1", "hung2", "healthy"]; + for (const workspaceId of ids) { + workspaces.set( + workspaceId, + createPersistedWorkspaceRecord({ + workspaceId, + projectId: "p1", + cwd: path.join(rootPath, workspaceId), + kind: "directory", + displayName: workspaceId, + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + } + const release = deferred(); + const calls: string[] = []; + let recovered = false; + let ticks = 0; + const probe = setInterval(() => ticks++, 1); + const service = new WorkspaceReconciliationService({ + projectRegistry, + workspaceRegistry, + logger, + inspectionTimeoutMs: 30, + inspectDirectory: async (targetPath) => { + calls.push(targetPath); + if (!recovered && path.basename(targetPath).startsWith("hung")) await release.promise; + return "directory"; + }, + }); + try { + const result = await service.runOnce(); + expect(ticks).toBeGreaterThan(0); + expect(result.durationMs).toBeLessThan(1000); + expect(result.changesApplied).toEqual([]); + expect(calls).toEqual(ids.slice(0, 2).map((id) => path.join(rootPath, id))); + expect([...workspaces.values()].map((workspace) => workspace.archivedAt)).toEqual([ + null, + null, + null, + ]); + expect(warnRecords.filter((record) => record.message.includes("saturated"))).toEqual([ + { + message: "Workspace directory inspection saturated; preserving uninspected workspaces", + payload: { paths: ids.slice(0, 2).map((id) => path.join(rootPath, id)) }, + }, + ]); + recovered = true; + release.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + expect(infoRecords).toEqual([ + { + message: + "Timed-out workspace directory inspections drained; reconciliation capacity restored", + payload: {}, + }, + ]); + calls.length = 0; + await service.runOnce(); + expect(calls).toEqual([...ids.map((id) => path.join(rootPath, id)), rootPath]); + expect(warnRecords.filter((record) => record.message.includes("saturated"))).toHaveLength(1); + } finally { + release.resolve(); + clearInterval(probe); + service.dispose(); + } + }); + + 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); + 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, + }), + ); + for (const workspaceId of ["hung", "healthy", "healthy2", "healthy3"]) { + 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 healthyInspected = deferred(); + const service = new WorkspaceReconciliationService({ + projectRegistry, + workspaceRegistry, + logger: createTestLogger(), + inspectionTimeoutMs: 30, + inspectDirectory: async (targetPath) => { + calls.push(targetPath); + maximumOutstanding = Math.max(maximumOutstanding, ++outstanding); + if (path.basename(targetPath) === "hung") return new Promise(() => {}); + healthyInspected.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + outstanding--; + return "directory"; + }, + }); + try { + const scan = service.runOnce(); + await Promise.race([ + healthyInspected.promise, + scan.then(() => { + throw new Error("Reconciliation did not use its directory inspection capability"); + }), + ]); + expect(outstanding).toBeLessThanOrEqual(2); + expect(workspaces.get("hung")?.archivedAt).toBeNull(); + const result = await scan; + expect(result.changesApplied).toEqual([]); + await service.runOnce(); + expect(calls.filter((cwd) => path.basename(cwd) === "hung")).toHaveLength(1); + expect(calls.filter((cwd) => path.basename(cwd) === "healthy3")).toHaveLength(2); + expect(maximumOutstanding).toBe(2); + expect([...workspaces.values()].map((workspace) => workspace.archivedAt)).toEqual([ + null, + null, + null, + null, + ]); + } finally { + service.dispose(); + } + }); + + test.each(["missing", "directory"] as const)( + "does not process or archive a workspace archived during %s inspection", + async (state) => { + const rootPath = realpathSync( + mkdtempSync(path.join(tmpdir(), "reconcile-inspection-archive-")), + ); + 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 cwd = path.join(rootPath, "missing"); + workspaces.set( + "w1", + createPersistedWorkspaceRecord({ + workspaceId: "w1", + projectId: "p1", + cwd, + kind: "directory", + displayName: "workspace", + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + const started = deferred(); + const release = deferred(); + const archivedCallbacks: string[] = []; + const checkouts = new TestCheckouts(); + const service = new WorkspaceReconciliationService({ + projectRegistry, + workspaceRegistry, + logger: createTestLogger(), + workspaceGitService: checkouts, + onWorkspaceArchived: (id) => { + archivedCallbacks.push(id); + }, + inspectDirectory: async (targetPath) => { + if (targetPath === cwd) { + started.resolve(); + await release.promise; + return state; + } + return "directory"; + }, + }); + try { + const scan = service.runOnce(); + await Promise.race([ + started.promise, + scan.then(() => { + throw new Error("Reconciliation did not use its directory inspection capability"); + }), + ]); + await workspaceRegistry.archive("w1", "2026-01-01T00:00:00.000Z"); + const archived = workspaces.get("w1"); + release.resolve(); + const result = await scan; + expect(workspaces.get("w1")).toEqual(archived); + expect(archivedCallbacks).toEqual([]); + expect(checkouts.reads).toEqual([rootPath]); + expect(result.changesApplied).toEqual([]); + } finally { + release.resolve(); + service.dispose(); + } + }, + ); + + test("keeps the event loop responsive while inspecting 20000 real workspace paths", async () => { + const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-many-"))); + 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, + }), + ); + for (let i = 0; i < 20000; i++) { + const workspaceId = `w${i}`; + workspaces.set( + workspaceId, + createPersistedWorkspaceRecord({ + workspaceId, + projectId: "p1", + cwd: path.join(rootPath, workspaceId), + kind: "directory", + displayName: workspaceId, + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + } + const service = new WorkspaceReconciliationService({ + projectRegistry, + workspaceRegistry, + logger: createTestLogger(), + }); + let last = performance.now(); + let longestGap = 0; + const probe = setInterval(() => { + const now = performance.now(); + longestGap = Math.max(longestGap, now - last); + last = now; + }, 1); + let result; + try { + result = await service.runOnce(); + await new Promise((resolve) => setTimeout(resolve, 1)); + } finally { + clearInterval(probe); + service.dispose(); + } + expect(longestGap).toBeLessThan(20); + expect(result.changesApplied).toHaveLength(20000); + expect([...workspaces.values()].every((workspace) => workspace.archivedAt !== null)).toBe(true); + }, 30000); + test("preserves workspace archival that lands during boot reconciliation", async () => { const workspaceRoot = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-archive-race-"))); tempDirs.push(workspaceRoot); diff --git a/packages/server/src/server/workspace-reconciliation-service.ts b/packages/server/src/server/workspace-reconciliation-service.ts index eb79dbacd2..5333da8f83 100644 --- a/packages/server/src/server/workspace-reconciliation-service.ts +++ b/packages/server/src/server/workspace-reconciliation-service.ts @@ -1,4 +1,6 @@ -import { statSync, watch as watchPath } from "node:fs"; +import { watch as watchPath } from "node:fs"; +import { stat } from "node:fs/promises"; +import { setImmediate as yieldToLoop } from "node:timers/promises"; import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages"; import type pino from "pino"; import type { @@ -19,6 +21,8 @@ import { deriveProjectKey } from "./project-key.js"; const DEFAULT_RESCAN_INTERVAL_MS = 5 * 60_000; const DEFAULT_DEBOUNCE_MS = 100; +const INSPECTION_CONCURRENCY = 2; +const DEFAULT_INSPECTION_TIMEOUT_MS = 5000; export type ProjectUpdate = | { kind: "upsert"; project: PersistedProjectRecord } @@ -95,6 +99,8 @@ export interface WorkspaceReconciliationServiceOptions { clock?: ReconciliationClock; rescanIntervalMs?: number; debounceMs?: number; + inspectDirectory?: (targetPath: string) => Promise; + inspectionTimeoutMs?: number; } interface ProjectReconciliationInput { @@ -110,9 +116,17 @@ interface CachedCheckoutRead { checkout: Promise; } -type DirectoryState = "directory" | "missing" | "unreadable"; +export type DirectoryState = "directory" | "missing" | "unreadable"; export class WorkspaceReconciliationService { + // Timed-out fs operations retain their slot until the OS returns. Never use + // 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; private readonly serverId: string | undefined; private readonly projectRegistry: ProjectRegistry; private readonly workspaceRegistry: WorkspaceRegistry; @@ -136,6 +150,10 @@ export class WorkspaceReconciliationService { private reconcileQueuedMode: "metadata" | "full" | null = null; constructor(options: WorkspaceReconciliationServiceOptions) { + this.directoryInspection = + options.inspectDirectory ?? + (async (targetPath) => ((await stat(targetPath)).isDirectory() ? "directory" : "missing")); + this.inspectionTimeoutMs = options.inspectionTimeoutMs ?? DEFAULT_INSPECTION_TIMEOUT_MS; this.serverId = options.serverId; this.projectRegistry = options.projectRegistry; this.workspaceRegistry = options.workspaceRegistry; @@ -180,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); @@ -197,16 +216,23 @@ export class WorkspaceReconciliationService { this.workspaceRegistry.list(), ]); const workspacesByProject = new Map(); - for (const workspace of workspaces) { - if (workspace.archivedAt || this.inspectDirectory(workspace.cwd) !== "directory") continue; + const workspaceStates = await mapBounded( + workspaces.filter((w) => !w.archivedAt), + async (workspace) => ({ workspace, state: await this.inspectDirectory(workspace.cwd) }), + ); + for (const { workspace, state } of workspaceStates) { + if (state !== "directory") continue; + if ((await this.workspaceRegistry.get(workspace.workspaceId))?.archivedAt !== null) continue; const siblings = workspacesByProject.get(workspace.projectId) ?? []; siblings.push(workspace); workspacesByProject.set(workspace.projectId, siblings); } + const projectStates = await mapBounded( + projects.filter((p) => !p.archivedAt), + async (project) => ({ project, state: await this.inspectDirectory(project.rootPath) }), + ); await this.reconcileGitMetadataForProjects( - projects.filter( - (project) => !project.archivedAt && this.inspectDirectory(project.rootPath) === "directory", - ), + projectStates.filter(({ state }) => state === "directory").map(({ project }) => project), workspacesByProject, changes, ); @@ -222,23 +248,30 @@ 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 = activeWorkspaces.map((workspace) => ({ + const workspaceDirectoryStates = await mapBounded(activeWorkspaces, async (workspace) => ({ workspace, - state: this.inspectDirectory(workspace.cwd), + state: await this.inspectDirectory(workspace.cwd), + })); + // Read project roots after workspaces: losing a volume can withhold an + // archive, never turn a workspace's absence into proof of deletion. + const projectStates = await mapBounded(activeProjects, async (project) => ({ + project, + state: await this.inspectDirectory(project.rootPath), })); - // Project roots are read after the workspace directories, so a volume that - // goes away mid-pass leaves its project unreachable rather than its workspaces - // alone. The skew can only withhold an archive, never produce one. const reachableProjectIds = new Set( - activeProjects - .filter((project) => this.inspectDirectory(project.rootPath) === "directory") - .map((project) => project.projectId), + projectStates + .filter(({ state }) => state === "directory") + .map(({ project }) => project.projectId), ); const workspacesByProject = new Map(); for (const { workspace, state } of workspaceDirectoryStates) { if (state !== "directory") continue; + if ((await this.workspaceRegistry.get(workspace.workspaceId))?.archivedAt !== null) continue; const list = workspacesByProject.get(workspace.projectId) ?? []; list.push(workspace); workspacesByProject.set(workspace.projectId, list); @@ -255,26 +288,29 @@ export class WorkspaceReconciliationService { state === "missing" && reachableProjectIds.has(workspace.projectId), ) .map(({ workspace }) => workspace); - await Promise.all( - missingWorkspaces.map(async (workspace) => { - const timestamp = new Date().toISOString(); - await this.workspaceRegistry.archive(workspace.workspaceId, timestamp); - await this.onWorkspaceArchived?.(workspace.workspaceId); - changes.push({ - kind: "workspace_archived", - workspaceId: workspace.workspaceId, - directory: workspace.cwd, - reason: "directory_missing", - }); - - // Update the in-memory list for the project orphan check below - const siblings = workspacesByProject.get(workspace.projectId); - if (siblings) { - const updated = siblings.filter((w) => w.workspaceId !== workspace.workspaceId); - workspacesByProject.set(workspace.projectId, updated); - } - }), - ); + await mapBounded(missingWorkspaces, async (workspace) => { + const current = await this.workspaceRegistry.get(workspace.workspaceId); + const project = await this.projectRegistry.get(workspace.projectId); + if ( + !current || + current.archivedAt || + current.cwd !== workspace.cwd || + current.projectId !== workspace.projectId || + !project || + project.archivedAt || + project.rootPath !== activeProjectsById.get(project.projectId)?.rootPath + ) + return; + const timestamp = new Date().toISOString(); + if (!(await this.workspaceRegistry.archive(workspace.workspaceId, timestamp))) return; + await this.onWorkspaceArchived?.(workspace.workspaceId); + changes.push({ + kind: "workspace_archived", + workspaceId: workspace.workspaceId, + directory: workspace.cwd, + reason: "directory_missing", + }); + }); // 2. Reconcile mutable git metadata without changing identity or membership. // Projects persist until explicitly removed, even when they currently have @@ -351,13 +387,17 @@ export class WorkspaceReconciliationService { } private async reconcileProject(input: ProjectReconciliationInput): Promise { - const { project, siblings, currentGit, readCheckout, changes } = input; + const { siblings, currentGit, readCheckout, changes } = input; const workspaceCheckouts = await Promise.all( - siblings.map(async (workspace) => ({ - workspace, - checkout: await readCheckout(workspace.cwd), - })), + siblings.map(async (workspace) => { + const current = await this.workspaceRegistry.get(workspace.workspaceId); + if (!current || current.cwd !== workspace.cwd || current.projectId !== workspace.projectId) + return null; + return { workspace: current, checkout: await readCheckout(current.cwd) }; + }), ); + const project = await this.projectRegistry.get(input.project.projectId); + if (!project || project.archivedAt || project.rootPath !== input.project.rootPath) return; const projectUpdates: Partial> = {}; const mappedKind = deriveProjectKind(currentGit); const projectKey = deriveProjectKey({ @@ -377,40 +417,47 @@ export class WorkspaceReconciliationService { if (Object.keys(projectUpdates).length > 0) { const timestamp = new Date().toISOString(); - await this.projectRegistry.upsert({ - ...project, - ...projectUpdates, - updatedAt: timestamp, - }); - changes.push({ - kind: "project_updated", - projectId: project.projectId, - directory: project.rootPath, - fields: projectUpdates, + let applied = false; + await this.projectRegistry.update(project.projectId, (current) => { + if (current.archivedAt || current.rootPath !== project.rootPath) return current; + applied = true; + return { ...current, ...projectUpdates, updatedAt: timestamp }; }); + if (applied) + changes.push({ + kind: "project_updated", + projectId: project.projectId, + directory: project.rootPath, + fields: projectUpdates, + }); } await Promise.all( - workspaceCheckouts.map(async ({ workspace, checkout: wsGit }) => { + workspaceCheckouts.map(async (entry) => { + if (!entry) return; + const { workspace, checkout: wsGit } = entry; const timestamp = new Date().toISOString(); - const update = reconcileWorkspacePlacement({ - workspace, - checkout: wsGit, - updatedAt: timestamp, + if (!reconcileWorkspacePlacement({ workspace, checkout: wsGit, updatedAt: timestamp })) + return; + let fields: Partial | null = null; + await this.workspaceRegistry.update(workspace.workspaceId, (current) => { + if (current.cwd !== workspace.cwd || current.projectId !== workspace.projectId) + return current; + const update = reconcileWorkspacePlacement({ + workspace: current, + checkout: wsGit, + updatedAt: timestamp, + }); + if (!update) return current; + fields = update.fields; + return { ...current, ...fields, updatedAt: timestamp }; }); - if (!update) return; - - const updated = await this.workspaceRegistry.update(workspace.workspaceId, (current) => ({ - ...current, - ...update.fields, - updatedAt: timestamp, - })); - if (!updated) return; + if (!fields) return; changes.push({ kind: "workspace_updated", workspaceId: workspace.workspaceId, directory: workspace.cwd, - fields: update.fields, + fields, }); }), ); @@ -537,16 +584,74 @@ export class WorkspaceReconciliationService { return this.workspaceGitService.getCheckout(cwd); } - private inspectDirectory(targetPath: string): DirectoryState { + 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) { + pending = Promise.resolve() + .then(() => this.directoryInspection(targetPath)) + .catch((error) => { + if (isMissingPathError(error)) return "missing" as const; + this.logger.warn( + { err: error, targetPath }, + "Skipped workspace reconciliation after directory inspection failed", + ); + return "unreadable" as const; + }) + .finally(() => { + this.pendingInspections.delete(targetPath); + this.timedOutPaths.delete(targetPath); + this.wakeInspectionWaiters(); + if (this.inspectionSaturated && this.timedOutPaths.size === 0) { + this.inspectionSaturated = false; + this.logger.info( + {}, + "Timed-out workspace directory inspections drained; reconciliation capacity restored", + ); + } + }); + this.pendingInspections.set(targetPath, pending); + } + let timer: ReturnType | undefined; try { - return statSync(targetPath).isDirectory() ? "directory" : "missing"; - } catch (error) { - if (isMissingPathError(error)) return "missing"; - this.logger.warn( - { err: error, targetPath }, - "Skipped workspace reconciliation after directory inspection failed", - ); - return "unreadable"; + return await Promise.race([ + pending, + 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( + { paths: [...this.timedOutPaths] }, + "Workspace directory inspection saturated; preserving uninspected workspaces", + ); + } + this.logger.warn( + { targetPath }, + "Skipped workspace reconciliation after directory inspection timed out", + ); + resolve("unreadable"); + }, this.inspectionTimeoutMs); + }), + ]); + } finally { + clearTimeout(timer); } } } @@ -555,3 +660,21 @@ function isMissingPathError(error: unknown): boolean { if (typeof error !== "object" || error === null || !("code" in error)) return false; return error.code === "ENOENT" || error.code === "ENOTDIR"; } + +// Two workers bound admission; yielding also bounds passes whose adapters settle +// immediately (including unavailable paths after both IO slots time out). +async function mapBounded(items: T[], visit: (item: T) => Promise): Promise { + const results: R[] = []; + let next = 0; + await Promise.all( + Array.from({ length: Math.min(INSPECTION_CONCURRENCY, items.length) }, async () => { + let count = 0; + while (next < items.length) { + const index = next++; + results[index] = await visit(items[index]!); + if (++count % 32 === 0) await yieldToLoop(); + } + }), + ); + return results; +} diff --git a/packages/server/src/server/workspace-registry.test.ts b/packages/server/src/server/workspace-registry.test.ts index 6c27027edc..d72da7c072 100644 --- a/packages/server/src/server/workspace-registry.test.ts +++ b/packages/server/src/server/workspace-registry.test.ts @@ -474,7 +474,7 @@ describe("workspace registries", () => { expect(await workspaceRegistry.list()).toEqual([]); }); - test("refreshes workspace archive timestamps when an archive is repeated", async () => { + test("preserves the first workspace archive and reports whether a queued archive applied", async () => { await workspaceRegistry.initialize(); await workspaceRegistry.upsert( createPersistedWorkspaceRecord({ @@ -488,12 +488,24 @@ describe("workspace registries", () => { }), ); - await workspaceRegistry.archive("workspace-one", "2026-03-02T00:00:00.000Z"); - await workspaceRegistry.archive("workspace-one", "2026-03-03T00:00:00.000Z"); - - expect(await workspaceRegistry.get("workspace-one")).toMatchObject({ - archivedAt: "2026-03-03T00:00:00.000Z", - updatedAt: "2026-03-03T00:00:00.000Z", + const events: WorkspaceMutation[] = []; + workspaceRegistry.subscribeToMutations((event) => { + events.push(event); + }); + expect( + await Promise.all([ + workspaceRegistry.archive("workspace-one", "2026-03-02T00:00:00.000Z"), + workspaceRegistry.archive("workspace-one", "2026-03-03T00:00:00.000Z"), + ]), + ).toEqual([true, false]); + expect(events.map((event) => event.kind)).toEqual(["archive"]); + const reloaded = new FileBackedWorkspaceRegistry( + path.join(tmpDir, "projects", "workspaces.json"), + logger, + ); + expect(await reloaded.get("workspace-one")).toMatchObject({ + archivedAt: "2026-03-02T00:00:00.000Z", + updatedAt: "2026-03-02T00:00:00.000Z", }); }); diff --git a/packages/server/src/server/workspace-registry.ts b/packages/server/src/server/workspace-registry.ts index 0e0fb4e62c..61fcb4f720 100644 --- a/packages/server/src/server/workspace-registry.ts +++ b/packages/server/src/server/workspace-registry.ts @@ -167,7 +167,7 @@ export interface WorkspaceRegistry { workspaceId: string, archivedAt: string, context?: WorkspaceArchiveContext, - ): Promise; + ): Promise; remove(workspaceId: string): Promise; /** Central lifecycle seam for daemon-global workspace observers. */ subscribeToMutations?( @@ -247,20 +247,6 @@ class FileBackedRegistry { }); } - async archive(id: string, archivedAt: string): Promise { - await this.archiveIfPresent(id, archivedAt); - } - - protected async archiveIfPresent(id: string, archivedAt: string): Promise { - return this.mutateCache((records) => { - const existing = records.get(id); - if (!existing) return null; - const next = this.schema.parse({ ...existing, updatedAt: archivedAt, archivedAt }); - records.set(id, next); - return next; - }); - } - protected async archiveIfActive(id: string, archivedAt: string): Promise { return this.mutateCache((records) => { const existing = records.get(id); @@ -485,7 +471,7 @@ export class FileBackedProjectRegistry return project; } - override async archive(projectId: string, archivedAt: string): Promise { + async archive(projectId: string, archivedAt: string): Promise { const project = await this.archiveIfActive(projectId, archivedAt); if (!project) return; await this.notifyMutation({ kind: "archive", projectId, project }); @@ -565,21 +551,28 @@ export class FileBackedWorkspaceRegistry }); } - override async archive( + async archive( workspaceId: string, archivedAt: string, context?: WorkspaceArchiveContext, - ): Promise { - const workspace = await super.update(workspaceId, (existing) => ({ - ...existing, - updatedAt: archivedAt, - archivedAt, - ...(context?.autoArchivedChangeRequestUrl - ? { autoArchivedChangeRequestUrl: context.autoArchivedChangeRequestUrl } - : {}), - })); - if (!workspace) return; + ): Promise { + const [workspace] = await this.mutateMany((records) => { + const existing = records.get(workspaceId); + if (!existing || existing.archivedAt) return []; + return [ + { + ...existing, + updatedAt: archivedAt, + archivedAt, + ...(context?.autoArchivedChangeRequestUrl + ? { autoArchivedChangeRequestUrl: context.autoArchivedChangeRequestUrl } + : {}), + }, + ]; + }); + if (!workspace) return false; await this.notifyMutation({ kind: "archive", workspaceId, workspace }); + return true; } override async remove(workspaceId: string): Promise {