mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-27 19:52:49 +00:00
fix(workspaces): release runtime resources on shutdown
This commit is contained in:
@@ -1990,6 +1990,7 @@ export async function createPaseoDaemon(
|
||||
if (wsServer) {
|
||||
await wsServer.close();
|
||||
}
|
||||
await workspaceRuntime.close();
|
||||
await serviceProxy.stopStandalone();
|
||||
// Force-drop remaining sockets so httpServer.close() resolves promptly.
|
||||
// We've already closed wsServer (which sent ws-layer close frames) and
|
||||
|
||||
@@ -18,6 +18,7 @@ export interface ObservationRebindTransaction {
|
||||
|
||||
export interface GitCommonObservationCoordinator {
|
||||
bind(runtime: BoundWorkspaceRuntime, workspaceId: string, driver: WorkspaceRuntimeDriver): void;
|
||||
close(): Promise<void>;
|
||||
pause(workspaceId: string): Promise<void>;
|
||||
stageResume(workspaceId: string): Promise<ObservationRebindTransaction>;
|
||||
destroy(workspaceId: string): Promise<void>;
|
||||
@@ -54,6 +55,11 @@ export function createGitCommonObservationCoordinator(): GitCommonObservationCoo
|
||||
},
|
||||
});
|
||||
},
|
||||
async close() {
|
||||
const selected = [...observations];
|
||||
observations.clear();
|
||||
await Promise.allSettled(selected.map((observation) => observation.physical?.unsubscribe()));
|
||||
},
|
||||
async pause(workspaceId) {
|
||||
for (const observation of owned(workspaceId)) {
|
||||
const physical = observation.physical;
|
||||
|
||||
@@ -136,6 +136,7 @@ export interface WorkspaceRuntimeProviderCapability {
|
||||
export interface WorkspaceRuntimeService {
|
||||
listRuntimes(): readonly WorkspaceRuntimeCatalogEntry[];
|
||||
reconcile(): Promise<void>;
|
||||
close(): Promise<void>;
|
||||
create(input: CreateWorkspaceInput): Promise<WorkspaceRuntimePlacement>;
|
||||
run(input: WorkspaceProcessInput): Promise<WorkspaceProcess>;
|
||||
openTerminal(input: WorkspaceTerminalInput): Promise<WorkspaceTerminal>;
|
||||
|
||||
@@ -183,6 +183,28 @@ export function createService(
|
||||
if (failures.length > 0)
|
||||
throw new AggregateError(failures, "Workspace reconciliation failed");
|
||||
},
|
||||
async close() {
|
||||
const workspaceIds = new Set([
|
||||
...processesByWorkspaceId.keys(),
|
||||
...fileClients.keys(),
|
||||
...fileSubscriptions.keys(),
|
||||
...boundRuntimes.keys(),
|
||||
]);
|
||||
boundRuntimes.clear();
|
||||
await Promise.all(
|
||||
[...fileSubscriptions.values()].flatMap((subscriptions) =>
|
||||
[...subscriptions].map(async (subscription) => {
|
||||
await subscription.bound?.unsubscribe();
|
||||
subscription.bound = null;
|
||||
}),
|
||||
),
|
||||
);
|
||||
await gitCommonObservations.close();
|
||||
await Promise.all(
|
||||
[...workspaceIds].map((workspaceId) => closeFiles(workspaceId, true, null)),
|
||||
);
|
||||
await Promise.all([...workspaceIds].map((workspaceId) => stopProcesses(workspaceId)));
|
||||
},
|
||||
async create(input) {
|
||||
return sequence(input.workspaceId, async () => {
|
||||
const selectedRuntimeId = await records.resolveRuntimeId(input.workspaceId);
|
||||
@@ -520,7 +542,11 @@ export function createService(
|
||||
return client;
|
||||
}
|
||||
|
||||
async function closeFiles(workspaceId: string, forgetSubscriptions = false): Promise<void> {
|
||||
async function closeFiles(
|
||||
workspaceId: string,
|
||||
forgetSubscriptions = false,
|
||||
reason: Error | null = new Error("Workspace files client is closed"),
|
||||
): Promise<void> {
|
||||
const cached = fileClients.get(workspaceId);
|
||||
fileClients.delete(workspaceId);
|
||||
const subscriptions = fileSubscriptions.get(workspaceId);
|
||||
@@ -528,7 +554,7 @@ export function createService(
|
||||
for (const subscription of subscriptions) subscription.bound = null;
|
||||
if (forgetSubscriptions) fileSubscriptions.delete(workspaceId);
|
||||
}
|
||||
if (cached) await cached.client.close();
|
||||
if (cached) await cached.client.close(reason ?? undefined);
|
||||
}
|
||||
|
||||
async function stageSubscriptionRebind(
|
||||
|
||||
@@ -399,7 +399,7 @@ test("a failed second subscription rebind rolls back the staged observer set bef
|
||||
),
|
||||
).toBe(false);
|
||||
await fixture.service.destroy(fixture.workspaceId);
|
||||
});
|
||||
}, 10_000);
|
||||
|
||||
test("a file operation racing pause cannot reconstruct the closing helper client", async () => {
|
||||
const fixture = await createFixture("files-pause-race", true);
|
||||
|
||||
@@ -22,6 +22,7 @@ import {
|
||||
type WorkspaceRuntimeOptions,
|
||||
type WorkspaceRuntimeService,
|
||||
} from "./index.js";
|
||||
import { observeWorkspaceGit } from "../workspace-git-observation.js";
|
||||
|
||||
const cleanupRoots: string[] = [];
|
||||
const posixDescribe = describe.runIf(process.platform !== "win32");
|
||||
@@ -150,6 +151,31 @@ posixDescribe("setup lifecycle environment", () => {
|
||||
});
|
||||
});
|
||||
|
||||
posixDescribe("service lifecycle", () => {
|
||||
test("close releases runtime processes and observations without destroying backing", async () => {
|
||||
const fixture = await createFixture("worktree");
|
||||
await fixture.service.create(fixture.createInput);
|
||||
const runtime = await fixture.service.bind(fixture.workspaceId);
|
||||
const observation = await observeWorkspaceGit(runtime, () => undefined);
|
||||
const child = await fixture.service.run({
|
||||
workspaceId: fixture.workspaceId,
|
||||
argv: [process.execPath, "-e", "setInterval(() => {}, 1000)"],
|
||||
env: {},
|
||||
purpose: { kind: "workspace-script", script: "daemon-shutdown" },
|
||||
});
|
||||
child.stdin.end();
|
||||
|
||||
await fixture.service.close();
|
||||
|
||||
await expect(child.exited).resolves.toEqual({ code: null, signal: "SIGTERM" });
|
||||
expect(existsSync(fixture.runtimeRoot())).toBe(true);
|
||||
expect(fixture.runtimeIds.get(fixture.workspaceId)).toBe("worktree");
|
||||
await observation.unsubscribe();
|
||||
|
||||
await createWorkspaceRuntimeService(fixture.options).destroy(fixture.workspaceId);
|
||||
});
|
||||
});
|
||||
|
||||
posixDescribe.each(runtimeContractIds)("%s runtime public contract", (runtimeId) => {
|
||||
test("returns after materialization without waiting for configured setup", async () => {
|
||||
const fixture = await createFixture(runtimeId);
|
||||
|
||||
@@ -15,7 +15,7 @@ export interface WorkspaceFilesOwner {
|
||||
files: WorkspaceFiles;
|
||||
resolveCommand(command: string): Promise<string | null>;
|
||||
verify(): Promise<void>;
|
||||
close(): Promise<void>;
|
||||
close(reason?: Error): Promise<void>;
|
||||
}
|
||||
|
||||
export function bindWorkspaceHelper(options: {
|
||||
|
||||
@@ -164,12 +164,12 @@ export function createClient(options: {
|
||||
);
|
||||
}
|
||||
},
|
||||
async close() {
|
||||
async close(reason) {
|
||||
closePromise ??= (async () => {
|
||||
closing = true;
|
||||
watchSubscriptions.clear();
|
||||
if (watcher) {
|
||||
await stopWatcher(watcher, new Error("Workspace files client is closed"));
|
||||
await stopWatcher(watcher, reason);
|
||||
}
|
||||
await watcherRecovery?.catch(() => undefined);
|
||||
await Promise.all([...processes].map((child) => terminate(child)));
|
||||
@@ -390,7 +390,11 @@ function handleWatchEvent(state: WatchState, line: string): void {
|
||||
}
|
||||
|
||||
function acknowledgement(state: WatchState, key: string): Promise<void> {
|
||||
return new Promise((resolve, reject) => state.acknowledgements.set(key, { resolve, reject }));
|
||||
const pending = new Promise<void>((resolve, reject) =>
|
||||
state.acknowledgements.set(key, { resolve, reject }),
|
||||
);
|
||||
void pending.catch(() => undefined);
|
||||
return pending;
|
||||
}
|
||||
|
||||
async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T> {
|
||||
|
||||
Reference in New Issue
Block a user