diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 5d26c3d464..380e3618e0 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1779,7 +1779,7 @@ export async function createPaseoDaemon( await pluginRuntime.stopAllPlugins(); unsubscribePluginProviders(); await hubRelationships.stop(); - await workspaceReconciliation.dispose(); + workspaceReconciliation.dispose(); scriptHealthMonitor.stop(); // Freeze both ingress and registration before taking the agent closure snapshot. wsServer?.prepareForShutdown(); diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index 7c2ea1f0a7..84726d0b02 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -1602,7 +1602,6 @@ export class HubRelationshipHarness { appVersion: "0.1.106", }); await client.connect(); - await client.observeAgents().ready; return client; } diff --git a/packages/server/src/server/workspace-reconciliation-observation.test.ts b/packages/server/src/server/workspace-reconciliation-observation.test.ts index 3b9ab5d35e..a5fa20c97f 100644 --- a/packages/server/src/server/workspace-reconciliation-observation.test.ts +++ b/packages/server/src/server/workspace-reconciliation-observation.test.ts @@ -1,4 +1,4 @@ -import { mkdirSync, mkdtempSync, rmSync, watch as watchPath } from "node:fs"; +import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages"; @@ -30,52 +30,6 @@ afterEach(() => { for (const target of cleanupPaths.splice(0)) rmSync(target, { recursive: true, force: true }); }); -test("disposal releases a real project root watch before its directory is removed", async () => { - const root = mkdtempSync(path.join(tmpdir(), "project-watch-close-")); - cleanupPaths.push(root); - const logger = createTestLogger(); - const projects = new FileBackedProjectRegistry(path.join(root, "projects.json"), logger); - const workspaces = new FileBackedWorkspaceRegistry(path.join(root, "workspaces.json"), logger); - await projects.initialize(); - await workspaces.initialize(); - await projects.upsert( - createPersistedProjectRecord({ - projectId: "project-watch-close", - rootPath: root, - kind: "non_git", - displayName: "root watch", - createdAt: TIMESTAMP, - updatedAt: TIMESTAMP, - }), - ); - let released = false; - const service = new WorkspaceReconciliationService({ - projectRegistry: projects, - workspaceRegistry: workspaces, - logger, - watchProjectRoot: (rootPath, options, onChange, onError) => { - const watcher = watchPath(rootPath, options, onChange); - watcher.on("error", onError); - const closed = new Promise((resolve) => - watcher.once("close", () => { - released = true; - resolve(); - }), - ); - return { - close: () => { - watcher.close(); - return closed; - }, - }; - }, - }); - await service.start(); - await service.dispose(); - expect(released).toBe(true); - rmSync(root, { recursive: true }); -}); - interface ProjectSpec { id: string; root: string; @@ -234,11 +188,7 @@ class ObservedPlacements { const watch = { rootPath, onChange, onError, closed: false }; this.watches.push(watch); this.lifecycleEvents.push(`watch installed:${path.relative(this.home, rootPath)}`); - return { - close: async () => { - watch.closed = true; - }, - }; + return { close: () => (watch.closed = true) }; }; this.service = new WorkspaceReconciliationService({ projectRegistry: this.projects, @@ -327,8 +277,8 @@ class ObservedPlacements { await this.clock.advanceBy(elapsedMs); } - async dispose(): Promise { - await this.service.dispose(); + dispose(): void { + this.service.dispose(); } watchedRoots(): string[] { @@ -457,7 +407,7 @@ describe("observed workspace placement", () => { "registry mutation resolved:project-new", ]); expect(observed.gitReads).toBe(0); - await observed.dispose(); + observed.dispose(); }); test("deduplicates active root watches and tears them down on archive and remove", async () => { @@ -484,7 +434,7 @@ describe("observed workspace placement", () => { { kind: "remove", projectId: "project-duplicate" }, { kind: "remove", projectId: "project-remove" }, ]); - await observed.dispose(); + observed.dispose(); }); test("filters unrelated files and coalesces Git change bursts", async () => { @@ -500,7 +450,7 @@ describe("observed workspace placement", () => { observed.change("repo", null); await observed.advanceBy(DEBOUNCE_MS); expect(observed.gitReads).toBe(1); - await observed.dispose(); + observed.dispose(); }); test("recovers errored and temporarily unavailable watchers on the periodic pass", async () => { @@ -518,7 +468,7 @@ describe("observed workspace placement", () => { expect(observed.watchedRoots()).toEqual(["errored", "unavailable"]); expect(observed.closedRoots()).toEqual(["errored"]); - await observed.dispose(); + observed.dispose(); }); test("archives missing workspace directories on the periodic pass", async () => { @@ -532,7 +482,7 @@ describe("observed workspace placement", () => { expect((await observed.placement("workspace-one"))?.archivedAt).toEqual(expect.any(String)); expect(observed.workspaceBatches).toEqual([["workspace-one"]]); - await observed.dispose(); + observed.dispose(); }); test("preserves a periodic full pass queued behind metadata reconciliation", async () => { @@ -554,7 +504,7 @@ describe("observed workspace placement", () => { expect((await observed.placement("workspace-one"))?.archivedAt).toEqual(expect.any(String)); expect(observed.workspaceBatches).toEqual([["workspace-one"]]); - await observed.dispose(); + observed.dispose(); }); test("contains a failed reconciliation and converges on the next change", async () => { @@ -576,7 +526,7 @@ describe("observed workspace placement", () => { branch: "main", displayName: "Durable workspace-one", }); - await observed.dispose(); + observed.dispose(); }); test("deduplicates direct placement and project-derived workspace fanout", async () => { @@ -599,7 +549,7 @@ describe("observed workspace placement", () => { expect(observed.projectUpdates).toHaveLength(1); expect(observed.workspaceBatches).toEqual([["workspace-one", "workspace-two"]]); expect(new Set(observed.workspaceBatches[0]).size).toBe(2); - await observed.dispose(); + observed.dispose(); }); test("disposal suppresses in-flight mutations and reconciliation fanout", async () => { @@ -608,7 +558,7 @@ describe("observed workspace placement", () => { const registryRead = mutation.holdNextRegistryRead(); const adding = mutation.add({ id: "project-late", root: "late" }); await registryRead.started; - await mutation.dispose(); + mutation.dispose(); registryRead.release(); await adding; expect(mutation.projectUpdates).toEqual([]); @@ -623,7 +573,7 @@ describe("observed workspace placement", () => { reconciliation.change("repo", ".git"); const advancing = reconciliation.advanceBy(DEBOUNCE_MS); await workspaceRead.started; - await reconciliation.dispose(); + reconciliation.dispose(); workspaceRead.release(); await advancing; diff --git a/packages/server/src/server/workspace-reconciliation-service.ts b/packages/server/src/server/workspace-reconciliation-service.ts index d42f7334a6..59d2f9c124 100644 --- a/packages/server/src/server/workspace-reconciliation-service.ts +++ b/packages/server/src/server/workspace-reconciliation-service.ts @@ -25,7 +25,7 @@ export type ProjectUpdate = | { kind: "remove"; projectId: string }; interface ProjectRootWatcher { - close(): Promise; + close(): void; } export interface ProjectRootWatch { @@ -58,13 +58,7 @@ const systemClock: ReconciliationClock = { const watchProjectRoot: ProjectRootWatch = (rootPath, options, onChange, onError) => { const watcher = watchPath(rootPath, options, onChange); watcher.on("error", onError); - const closed = new Promise((resolve) => watcher.once("close", resolve)); - return { - close() { - watcher.close(); - return closed; - }, - }; + return watcher; }; export type ReconciliationChange = @@ -184,15 +178,14 @@ export class WorkspaceReconciliationService { this.rescanTimer.unref?.(); } - async dispose(): Promise { + dispose(): void { this.disposed = true; this.unsubscribeRegistry?.(); this.unsubscribeRegistry = null; if (this.rescanTimer) this.clock.clearInterval(this.rescanTimer); if (this.debounceTimer) this.clock.clearTimeout(this.debounceTimer); - const closing = this.watchers.map(({ watcher }) => watcher.close()); + for (const { watcher } of this.watchers) watcher.close(); this.watchers.length = 0; - await Promise.all(closing); } /** Reconciles mutable Git facts only; never archives missing records. */ @@ -420,7 +413,7 @@ export class WorkspaceReconciliationService { areEquivalentPaths(project.rootPath, target.rootPath), ); if (stillActive) continue; - void target.watcher.close(); + target.watcher.close(); this.watchers.splice(index, 1); } @@ -440,7 +433,7 @@ export class WorkspaceReconciliationService { } }, (error) => { - void watcher.close(); + watcher.close(); const index = this.watchers.findIndex((target) => target.watcher === watcher); if (index >= 0) this.watchers.splice(index, 1); this.logger.warn(