fix(workspaces): await Git watcher teardown

packages/server/src/server/workspace-git-service.observation.test.ts

@@ -24,7 +24,10 @@ interface WatchRecord {
   unsubscribe: ReturnType<typeof vi.fn>;
 }
 
-function createWatcherHarness(harnessOptions?: { failDirectories?: Set<string> }) {
+function createWatcherHarness(harnessOptions?: {
+  failDirectories?: Set<string>;
+  unsubscribe?: () => Promise<void>;
+}) {
   const records: WatchRecord[] = [];
   const subscribe = vi.fn(
     async (
@@ -39,7 +42,7 @@ function createWatcherHarness(harnessOptions?: { failDirectories?: Set<string> }
         const record = records.find((candidate) => candidate.updateIgnore === updateIgnore);
         if (record) record.ignore = paths;
       });
-      const unsubscribe = vi.fn(async () => {});
+      const unsubscribe = vi.fn(harnessOptions?.unsubscribe ?? (async () => {}));
       records.push({
         directory,
         callback,
@@ -248,6 +251,42 @@ describe("WorkspaceGitService checkout observation", () => {
     await disposal;
   });
 
+  test("dispose waits for checkout and repository watchers to finish closing", async () => {
+    const watcherCloseFinished = createDeferred<void>();
+    const watcher = createWatcherHarness({
+      unsubscribe: () => watcherCloseFinished.promise,
+    });
+    const fileObserver: FileObserver = {
+      subscribe: watcher.subscribe as FileObserver["subscribe"],
+      getDiagnostics: () => {
+        throw new Error("Diagnostics are not used by this lifecycle test");
+      },
+      close: async () => {},
+    };
+    const service = createService(watcher, undefined, createLogger(), fileObserver);
+    service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
+    await vi.waitFor(() => {
+      expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(1);
+      expect(getWatcherRecordsForDirectory(watcher, GIT_DIR)).toHaveLength(1);
+    });
+
+    const disposal = service.dispose();
+    await vi.waitFor(() => {
+      for (const { unsubscribe } of watcher.records) {
+        expect(unsubscribe).toHaveBeenCalledTimes(1);
+      }
+    });
+    expect(
+      await Promise.race([
+        disposal.then(() => "disposed" as const),
+        Promise.resolve("pending" as const),
+      ]),
+    ).toBe("pending");
+
+    watcherCloseFinished.resolve();
+    await disposal;
+  });
+
   test("shares one recursive checkout observer between cwd-equivalent consumers", async () => {
     const watcher = createWatcherHarness();
     const runGitCommand = vi.fn(async (args: string[]) => ({

packages/server/src/server/workspace-git-service.ts

@@ -1425,20 +1425,21 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     this.selectedWorkspaces.clear();
     this.legacyWorkspaces.clear();
 
+    const repoTargetDisposals = [...this.repoTargets.values()].map((target) =>
+      this.closeRepoTarget(target),
+    );
+    this.repoTargets.clear();
+
+    const workingTreeWatchDisposals = [...this.workingTreeWatchTargets.values()].map((target) =>
+      this.closeWorkingTreeWatchTarget(target),
+    );
+    this.workingTreeWatchTargets.clear();
+
     const workspaceTargetDisposals = [...this.workspaceTargets.values()].map((target) =>
       this.closeWorkspaceTarget(target),
     );
     this.workspaceTargets.clear();
 
-    for (const target of this.repoTargets.values()) {
-      this.closeRepoTarget(target);
-    }
-    this.repoTargets.clear();
-
-    for (const target of this.workingTreeWatchTargets.values()) {
-      this.closeWorkingTreeWatchTarget(target);
-    }
-    this.workingTreeWatchTargets.clear();
     this.workingTreeWatchSetups.clear();
     this.workingTreeWatchResolutions.clear();
     this.workingTreeWatchAliases.clear();
@@ -1448,6 +1449,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     this.disposePromise = Promise.all([
       this.fileObserver.close(),
       ...selectedWorkspaceDisposals,
+      ...repoTargetDisposals,
+      ...workingTreeWatchDisposals,
       ...workspaceTargetDisposals,
     ]).then(() => undefined);
     return this.disposePromise;
@@ -3744,7 +3747,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       const repoTarget = this.repoTargets.get(target.repoGitRoot);
       repoTarget?.workspaceKeys.delete(target.cwd);
       if (repoTarget && repoTarget.workspaceKeys.size === 0) {
-        this.closeRepoTarget(repoTarget);
+        void this.closeRepoTarget(repoTarget);
         this.repoTargets.delete(target.repoGitRoot);
       } else if (repoTarget?.cwd === target.cwd) {
         repoTarget.cwd = repoTarget.workspaceKeys.values().next().value ?? repoTarget.cwd;
@@ -3766,7 +3769,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       return;
     }
 
-    this.closeWorkingTreeWatchTarget(target);
+    void this.closeWorkingTreeWatchTarget(target);
     this.workingTreeWatchTargets.delete(cwd);
   }
 
@@ -3778,7 +3781,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     if (target.workspaceKeys.size > 0 || target.listeners.size > 0) {
       return;
     }
-    this.closeWorkingTreeWatchTarget(target);
+    void this.closeWorkingTreeWatchTarget(target);
     if (this.workingTreeWatchTargets.get(target.cwd) === target) {
       this.workingTreeWatchTargets.delete(target.cwd);
     }
@@ -3793,7 +3796,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     ) {
       return;
     }
-    this.closeWorkingTreeWatchTarget(target);
+    void this.closeWorkingTreeWatchTarget(target);
     this.workingTreeWatchTargets.delete(target.cwd);
   }
 
@@ -3819,7 +3822,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     return runtimeObservationDisposal;
   }
 
-  private closeWorkingTreeWatchTarget(target: WorkingTreeWatchTarget): void {
+  private closeWorkingTreeWatchTarget(target: WorkingTreeWatchTarget): Promise<void> {
     target.closed = true;
     for (const alias of target.aliases) {
       if (this.workingTreeWatchAliases.get(alias) === target.cwd) {
@@ -3837,18 +3840,20 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       target.recovery.timer = null;
     }
 
+    let subscriptionDisposal = Promise.resolve();
     if (target.subscription) {
       const subscription = target.subscription;
       target.subscription = null;
-      void subscription.unsubscribe().catch((error) => {
+      subscriptionDisposal = subscription.unsubscribe().catch((error) => {
         this.logger.warn({ err: error, cwd: target.cwd }, "Failed to stop working tree watcher");
       });
     }
     target.workspaceKeys.clear();
     target.listeners.clear();
+    return subscriptionDisposal;
   }
 
-  private closeRepoTarget(target: RepoGitTarget): void {
+  private closeRepoTarget(target: RepoGitTarget): Promise<void> {
     target.closed = true;
     if (target.intervalId) {
       clearInterval(target.intervalId);
@@ -3863,10 +3868,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       clearTimeout(target.recovery.timer);
       target.recovery.timer = null;
     }
+    let subscriptionDisposal = Promise.resolve();
     if (target.subscription) {
       const subscription = target.subscription;
       target.subscription = null;
-      void subscription.unsubscribe().catch((error) => {
+      subscriptionDisposal = subscription.unsubscribe().catch((error) => {
         this.logger.warn(
           { err: error, repoGitRoot: target.repoGitRoot },
           "Failed to stop repository metadata watcher",
@@ -3874,6 +3880,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       });
     }
     target.workspaceKeys.clear();
+    return subscriptionDisposal;
   }
 }
 

packages/server/src/server/workspace-runtime/git-observation/internal/integration.test.ts

@@ -1,12 +1,20 @@
 import { PassThrough, Readable } from "node:stream";
 
-import { expect, test } from "vitest";
+import { expect, test, vi } from "vitest";
 
 import { observeWorkspaceGit } from "../../../workspace-git-observation.js";
 import type { BoundWorkspaceRuntime } from "../../index.js";
 import type { WorkspaceRuntimeDriver } from "../../drivers/index.js";
 import { createGitCommonObservationCoordinator } from "./integration.js";
 
+function createDeferred<T>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
 test("selected runtime Git observation prunes paths ignored by that runtime checkout", async () => {
   let subscriptionInput: Parameters<BoundWorkspaceRuntime["files"]["subscribe"]>[0] | null = null;
   const runtime = {
@@ -38,3 +46,41 @@ test("selected runtime Git observation prunes paths ignored by that runtime chec
   });
   await subscription.unsubscribe();
 });
+
+test("closing Git observation waits for and releases an in-flight acquisition", async () => {
+  const physicalUnsubscribe = vi.fn(async () => {});
+  const physical = createDeferred<{ unsubscribe(): Promise<void> }>();
+  const workingTreeUnsubscribe = vi.fn(async () => {});
+  const runtime = {
+    files: {
+      subscribe: vi.fn(async () => ({ unsubscribe: workingTreeUnsubscribe })),
+    },
+  } as unknown as BoundWorkspaceRuntime;
+  const driver = {
+    observeGit: vi.fn(() => physical.promise),
+  } as unknown as WorkspaceRuntimeDriver;
+  const coordinator = createGitCommonObservationCoordinator();
+  coordinator.bind(runtime, "workspace-1", driver);
+
+  const observation = observeWorkspaceGit(runtime, () => undefined);
+  await vi.waitFor(() => expect(driver.observeGit).toHaveBeenCalledTimes(1));
+  const closing = coordinator.close();
+  expect(coordinator.close()).toBe(closing);
+  expect(
+    await Promise.race([
+      closing.then(() => "closed" as const),
+      Promise.resolve("pending" as const),
+    ]),
+  ).toBe("pending");
+
+  physical.resolve({ unsubscribe: physicalUnsubscribe });
+  await expect(observation).rejects.toThrow("Git common observation coordinator is closed");
+  await closing;
+  expect(physicalUnsubscribe).toHaveBeenCalledTimes(1);
+  expect(workingTreeUnsubscribe).toHaveBeenCalledTimes(1);
+
+  await expect(observeWorkspaceGit(runtime, () => undefined)).rejects.toThrow(
+    "Git common observation coordinator is closed",
+  );
+  expect(driver.observeGit).toHaveBeenCalledTimes(1);
+});

packages/server/src/server/workspace-runtime/git-observation/internal/integration.ts

@@ -11,6 +11,8 @@ interface GitObservation {
   physical: { unsubscribe(): Promise<void> } | null;
 }
 
+type PhysicalGitObservation = NonNullable<GitObservation["physical"]>;
+
 export interface ObservationRebindTransaction {
   commit(): Promise<void>;
   rollback(): Promise<void>;
@@ -27,11 +29,17 @@ export interface GitCommonObservationCoordinator {
 /** Git invalidation is workspace-bound. Runtime-private observers never disclose placement. */
 export function createGitCommonObservationCoordinator(): GitCommonObservationCoordinator {
   const observations = new Set<GitObservation>();
+  const acquisitions = new Set<Promise<PhysicalGitObservation>>();
+  const resumeStages = new Set<Map<GitObservation, PhysicalGitObservation>>();
+  let closePromise: Promise<void> | null = null;
+  let closed = false;
 
   return {
     bind(runtime, workspaceId, driver) {
+      assertOpen();
       registerGitCommonObservationCapability(runtime, {
         observe: async (listener) => {
+          assertOpen();
           const observation: GitObservation = {
             workspaceId,
             runtime,
@@ -39,7 +47,7 @@ export function createGitCommonObservationCoordinator(): GitCommonObservationCoo
             listener,
             physical: null,
           };
-          observation.physical = await start(observation);
+          observation.physical = await acquire(observation);
           observations.add(observation);
           let active = true;
           return {
@@ -47,54 +55,66 @@ export function createGitCommonObservationCoordinator(): GitCommonObservationCoo
               if (!active) return;
               active = false;
               observations.delete(observation);
-              const physical = observation.physical;
-              observation.physical = null;
-              await physical?.unsubscribe();
+              await stop(observation);
             },
           };
         },
       });
     },
-    async close() {
-      const selected = [...observations];
-      observations.clear();
-      await Promise.allSettled(selected.map((observation) => observation.physical?.unsubscribe()));
+    close() {
+      if (closePromise) return closePromise;
+      closed = true;
+      closePromise = (async () => {
+        await Promise.allSettled(acquisitions);
+        await Promise.allSettled([...resumeStages].map((stage) => disposeStage(stage)));
+        const selected = [...observations];
+        observations.clear();
+        await Promise.allSettled(selected.map((observation) => stop(observation)));
+      })();
+      return closePromise;
     },
     async pause(workspaceId) {
       for (const observation of owned(workspaceId)) {
-        const physical = observation.physical;
-        observation.physical = null;
-        await physical?.unsubscribe();
+        await stop(observation);
       }
     },
     async stageResume(workspaceId) {
-      const staged = new Map<GitObservation, { unsubscribe(): Promise<void> }>();
+      assertOpen();
+      const staged = new Map<GitObservation, PhysicalGitObservation>();
+      resumeStages.add(staged);
       try {
         for (const observation of owned(workspaceId)) {
-          if (!observation.physical) staged.set(observation, await start(observation));
+          if (!observation.physical) staged.set(observation, await acquire(observation));
         }
       } catch (error) {
-        await Promise.allSettled([...staged.values()].map((item) => item.unsubscribe()));
+        await disposeStage(staged);
         throw error;
       }
       let finished = false;
       return {
         async commit() {
           if (finished) return;
+          if (closed) {
+            finished = true;
+            await disposeStage(staged);
+            throw new Error("Git common observation coordinator is closed");
+          }
           finished = true;
+          resumeStages.delete(staged);
           for (const [observation, physical] of staged) observation.physical = physical;
+          staged.clear();
         },
         async rollback() {
           if (finished) return;
           finished = true;
-          await Promise.allSettled([...staged.values()].map((item) => item.unsubscribe()));
+          await disposeStage(staged);
         },
       };
     },
     async destroy(workspaceId) {
       const selected = owned(workspaceId);
       for (const observation of selected) observations.delete(observation);
-      await Promise.allSettled(selected.map((observation) => observation.physical?.unsubscribe()));
+      await Promise.allSettled(selected.map((observation) => stop(observation)));
     },
   };
 
@@ -102,7 +122,42 @@ export function createGitCommonObservationCoordinator(): GitCommonObservationCoo
     return [...observations].filter((observation) => observation.workspaceId === workspaceId);
   }
 
-  async function start(observation: GitObservation) {
+  function assertOpen(): void {
+    if (closed) throw new Error("Git common observation coordinator is closed");
+  }
+
+  function acquire(observation: GitObservation): Promise<PhysicalGitObservation> {
+    assertOpen();
+    const acquisition = (async () => {
+      const physical = await start(observation);
+      if (closed) {
+        await physical.unsubscribe();
+        throw new Error("Git common observation coordinator is closed");
+      }
+      return physical;
+    })();
+    acquisitions.add(acquisition);
+    void acquisition.then(
+      () => acquisitions.delete(acquisition),
+      () => acquisitions.delete(acquisition),
+    );
+    return acquisition;
+  }
+
+  async function stop(observation: GitObservation): Promise<void> {
+    const physical = observation.physical;
+    observation.physical = null;
+    await physical?.unsubscribe();
+  }
+
+  async function disposeStage(stage: Map<GitObservation, PhysicalGitObservation>): Promise<void> {
+    resumeStages.delete(stage);
+    const physical = [...stage.values()];
+    stage.clear();
+    await Promise.allSettled(physical.map((item) => item.unsubscribe()));
+  }
+
+  async function start(observation: GitObservation): Promise<PhysicalGitObservation> {
     if (observation.driver.observeGit) {
       return observation.driver.observeGit(observation.workspaceId, observation.listener);
     }