diff --git a/packages/server/src/server/workspace-git-service.observation.test.ts b/packages/server/src/server/workspace-git-service.observation.test.ts index e863ebb4a0..09dc9274cd 100644 --- a/packages/server/src/server/workspace-git-service.observation.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.test.ts @@ -24,7 +24,10 @@ interface WatchRecord { unsubscribe: ReturnType; } -function createWatcherHarness(harnessOptions?: { failDirectories?: Set }) { +function createWatcherHarness(harnessOptions?: { + failDirectories?: Set; + unsubscribe?: () => Promise; +}) { const records: WatchRecord[] = []; const subscribe = vi.fn( async ( @@ -39,7 +42,7 @@ function createWatcherHarness(harnessOptions?: { failDirectories?: Set } 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(); + 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[]) => ({ diff --git a/packages/server/src/server/workspace-git-service.ts b/packages/server/src/server/workspace-git-service.ts index 6e8eaa803d..2de18807f2 100644 --- a/packages/server/src/server/workspace-git-service.ts +++ b/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 { 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 { 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; } } diff --git a/packages/server/src/server/workspace-runtime/git-observation/internal/integration.test.ts b/packages/server/src/server/workspace-runtime/git-observation/internal/integration.test.ts index 00a1b40d07..2450b975bf 100644 --- a/packages/server/src/server/workspace-runtime/git-observation/internal/integration.test.ts +++ b/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() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + test("selected runtime Git observation prunes paths ignored by that runtime checkout", async () => { let subscriptionInput: Parameters[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 }>(); + 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); +}); diff --git a/packages/server/src/server/workspace-runtime/git-observation/internal/integration.ts b/packages/server/src/server/workspace-runtime/git-observation/internal/integration.ts index fdb7d9a57f..3e8b2c2c83 100644 --- a/packages/server/src/server/workspace-runtime/git-observation/internal/integration.ts +++ b/packages/server/src/server/workspace-runtime/git-observation/internal/integration.ts @@ -11,6 +11,8 @@ interface GitObservation { physical: { unsubscribe(): Promise } | null; } +type PhysicalGitObservation = NonNullable; + export interface ObservationRebindTransaction { commit(): Promise; rollback(): Promise; @@ -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(); + const acquisitions = new Set>(); + const resumeStages = new Set>(); + let closePromise: Promise | 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 }>(); + assertOpen(); + const staged = new Map(); + 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 { + 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 { + const physical = observation.physical; + observation.physical = null; + await physical?.unsubscribe(); + } + + async function disposeStage(stage: Map): Promise { + resumeStages.delete(stage); + const physical = [...stage.values()]; + stage.clear(); + await Promise.allSettled(physical.map((item) => item.unsubscribe())); + } + + async function start(observation: GitObservation): Promise { if (observation.driver.observeGit) { return observation.driver.observeGit(observation.workspaceId, observation.listener); }