From d57096a4568cd2db106f35a0c5011627de53a646 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Wed, 14 Oct 2026 11:07:06 +0200 Subject: [PATCH] Start newly configured workspace services without restarting (#6542) * fix(server): start newly configured workspace services without restarting Extend the workspace port plan when paseo.json declares another service, preserving existing peer ports. Serialize extensions and roll back only their new reservations on failure. * fix(server): serialize service restarts with port plan extensions Keep concurrent restart and newly added service allocations in the same workspace update transaction so neither overwrites the other. * fix(server): keep service port update failures scoped and release cancelled reservations * test(hub): show creation errors in re-enrollment failures Assert the error field along with success so failed CI reports include the reason agent creation was rejected. --------- Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../hub/relationship-controller.test.ts | 8 +- .../workspace-service-port-registry.test.ts | 224 +++++++++++++++++- .../server/workspace-service-port-registry.ts | 142 +++++++---- .../src/server/worktree-bootstrap.test.ts | 48 ++++ 4 files changed, 365 insertions(+), 57 deletions(-) diff --git a/packages/server/src/server/hub/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index 51596b2384..d994a7280e 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -773,8 +773,12 @@ describe("Hub relationship", () => { const second = await relationship.ownedCreateResult("second-create"); expect(relationship.relationshipFile()?.relationship.daemonId).toBe(daemonId); - expect(first).toMatchObject({ payload: { success: true, executionId: "first-execution" } }); - expect(second).toMatchObject({ payload: { success: true, executionId: "second-execution" } }); + expect(first).toMatchObject({ + payload: { success: true, executionId: "first-execution", error: null }, + }); + expect(second).toMatchObject({ + payload: { success: true, executionId: "second-execution", error: null }, + }); expect(await relationship.durableOwnedAgentIds()).toHaveLength(2); }); diff --git a/packages/server/src/server/workspace-service-port-registry.test.ts b/packages/server/src/server/workspace-service-port-registry.test.ts index 3bf455ba57..b61bbb0f1d 100644 --- a/packages/server/src/server/workspace-service-port-registry.test.ts +++ b/packages/server/src/server/workspace-service-port-registry.test.ts @@ -33,7 +33,7 @@ describe("ensureWorkspaceServicePortPlan", () => { expect(allocatedServices).toEqual(["api", "web", "worker"]); }); - it("returns the existing plan without calling the allocator on later calls", async () => { + it("extends the existing plan with newly declared services without reallocating peers", async () => { let allocationCount = 0; const firstPlan = await ensureWorkspaceServicePortPlan({ @@ -54,9 +54,11 @@ describe("ensureWorkspaceServicePortPlan", () => { }, }); - expect(Array.from(secondPlan.entries())).toEqual(Array.from(firstPlan.entries())); - expect(secondPlan.has("new-service")).toBe(false); - expect(allocationCount).toBe(2); + expect(Array.from(secondPlan.entries())).toEqual([ + ...firstPlan.entries(), + ["new-service", 4300], + ]); + expect(allocationCount).toBe(3); }); it("shares one first-plan build across concurrent callers", async () => { @@ -101,6 +103,103 @@ describe("ensureWorkspaceServicePortPlan", () => { expect(allocationCount).toBe(2); }); + it("extends a pending plan for a concurrent caller with another service", async () => { + const allocation = createDeferredPort(); + const workspaceId = "registry-concurrent-extension-workspace"; + const firstPlan = ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }], + allocatePort: async () => allocation.promise, + }); + let allocations = 0; + const secondPlan = ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }, { scriptName: "web" }], + allocatePort: async () => { + allocations += 1; + return 6101; + }, + }); + allocation.resolve(6100); + expect(Array.from((await firstPlan).entries())).toEqual([["backend-dev", 6100]]); + expect(Array.from((await secondPlan).entries())).toEqual([ + ["backend-dev", 6100], + ["web", 6101], + ]); + expect(allocations).toBe(1); + releaseWorkspaceServicePortPlan(workspaceId); + }); + + it("keeps the previous plan and reservations when an extension fails", async () => { + const workspaceId = "registry-failed-extension-workspace"; + await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }], + allocatePort: async () => 6200, + }); + let allocations = 0; + await expect( + ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }, { scriptName: "web" }, { scriptName: "worker" }], + allocatePort: async () => { + allocations += 1; + if (allocations === 1) return 6201; + throw new Error("Allocation failed"); + }, + }), + ).rejects.toThrow("Allocation failed"); + const priorPlan = await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }], + allocatePort: async () => { + throw new Error("Existing port must survive"); + }, + }); + expect(Array.from(priorPlan.entries())).toEqual([["backend-dev", 6200]]); + const otherWorkspace = "registry-failed-extension-other-workspace"; + const reusedPlan = await ensureWorkspaceServicePortPlan({ + workspaceId: otherWorkspace, + services: [{ scriptName: "web" }], + allocatePort: async ({ reservedPorts }) => { + expect(reservedPorts.has(6200)).toBe(true); + expect(reservedPorts.has(6201)).toBe(false); + return 6201; + }, + }); + expect(reusedPlan.get("web")).toBe(6201); + releaseWorkspaceServicePortPlan(workspaceId); + releaseWorkspaceServicePortPlan(otherWorkspace); + }); + + it("rejects a new explicit port that collides with an existing service", async () => { + const workspaceId = "registry-extension-explicit-collision-workspace"; + await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }], + allocatePort: async () => 6300, + }); + await expect( + ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }, { scriptName: "web", port: 6300 }], + allocatePort: async () => 6301, + }), + ).rejects.toThrow("Service 'web' has a duplicate port 6300"); + const plan = await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }, { scriptName: "web", port: 6301 }], + allocatePort: async () => { + throw new Error("No allocation needed"); + }, + }); + expect(Array.from(plan.entries())).toEqual([ + ["backend-dev", 6300], + ["web", 6301], + ]); + releaseWorkspaceServicePortPlan(workspaceId); + }); + it("uses explicit configured ports without calling the allocator", async () => { let allocationCount = 0; @@ -312,6 +411,123 @@ describe("refreshWorkspaceServicePort", () => { ]); }); + it("preserves a concurrent service addition while refreshing a stopped service", async () => { + const workspaceId = "registry-refresh-and-extend-workspace"; + await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }], + allocatePort: async () => 6400, + }); + const allocation = createDeferredPort(); + const refreshing = refreshWorkspaceServicePort({ + workspaceId, + service: { scriptName: "backend-dev" }, + allocatePort: async () => allocation.promise, + }); + const extending = ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "backend-dev" }, { scriptName: "web", port: 6402 }], + allocatePort: async () => { + throw new Error("Explicit addition needs no allocation"); + }, + }); + allocation.resolve(6401); + expect(await refreshing).toBe(6401); + await extending; + const plan = await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [], + allocatePort: async () => { + throw new Error("No allocation needed"); + }, + }); + expect(Array.from(plan.entries())).toEqual([ + ["backend-dev", 6401], + ["web", 6402], + ]); + releaseWorkspaceServicePortPlan(workspaceId); + }); + + it.each(["reused port", "allocation failure"])( + "cancels a pending refresh and its waiter after workspace release (%s)", + async (outcome) => { + const workspaceId = "registry-cancelled-refresh-workspace"; + await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "api" }], + allocatePort: async () => 6500, + }); + const allocation = createDeferredPort(); + const refreshing = refreshWorkspaceServicePort({ + workspaceId, + service: { scriptName: "api" }, + allocatePort: async () => { + const port = await allocation.promise; + if (outcome === "allocation failure") throw new Error("Allocation failed"); + return port; + }, + }); + const waiting = refreshWorkspaceServicePort({ + workspaceId, + service: { scriptName: "web", port: 6501 }, + allocatePort: async () => 6501, + }); + const assertions = Promise.all([ + expect(refreshing).rejects.toThrow("Workspace service port plan was released"), + expect(waiting).rejects.toThrow("Workspace service port plan was released"), + ]); + releaseWorkspaceServicePortPlan(workspaceId); + allocation.resolve(6500); + await assertions; + const reusedPlan = await ensureWorkspaceServicePortPlan({ + workspaceId: "registry-after-cancelled-refresh-workspace", + services: [{ scriptName: "api" }], + allocatePort: async () => 6500, + }); + expect(reusedPlan.get("api")).toBe(6500); + releaseWorkspaceServicePortPlan("registry-after-cancelled-refresh-workspace"); + }, + ); + + it("lets a waiting service restart allocate after another service's restart fails", async () => { + const workspaceId = "registry-independent-refresh-failure-workspace"; + await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [{ scriptName: "api" }, { scriptName: "web" }], + allocatePort: createSequentialPortAllocator(6600), + }); + const allocation = createDeferredPort(); + const failedRefresh = refreshWorkspaceServicePort({ + workspaceId, + service: { scriptName: "api" }, + allocatePort: async () => { + await allocation.promise; + throw new Error("API port script failed"); + }, + }); + const failedAssertion = expect(failedRefresh).rejects.toThrow("API port script failed"); + const waitingRefresh = refreshWorkspaceServicePort({ + workspaceId, + service: { scriptName: "web" }, + allocatePort: async () => 6602, + }); + allocation.resolve(0); + await failedAssertion; + expect(await waitingRefresh).toBe(6602); + const plan = await ensureWorkspaceServicePortPlan({ + workspaceId, + services: [], + allocatePort: async () => { + throw new Error("No allocation needed"); + }, + }); + expect(Array.from(plan.entries())).toEqual([ + ["api", 6600], + ["web", 6602], + ]); + releaseWorkspaceServicePortPlan(workspaceId); + }); + it("uses an explicit configured port without calling the allocator", async () => { let allocationCount = 0; diff --git a/packages/server/src/server/workspace-service-port-registry.ts b/packages/server/src/server/workspace-service-port-registry.ts index be41a8d123..ea6cebfa35 100644 --- a/packages/server/src/server/workspace-service-port-registry.ts +++ b/packages/server/src/server/workspace-service-port-registry.ts @@ -24,6 +24,24 @@ interface PendingWorkspaceServicePortPlanToken { isReleased: boolean; } +class WorkspaceServicePortPlanReleasedError extends Error { + constructor(workspaceId: string) { + super(`Workspace service port plan was released while being created for '${workspaceId}'`); + } +} + +async function waitForWorkspaceServicePortPlanUpdate( + pendingPlan: Promise>, +): Promise { + try { + await pendingPlan; + } catch (error) { + // Allocation failure belongs to the original request. A waiting request + // retries against the rolled-back plan; workspace archival cancels both. + if (error instanceof WorkspaceServicePortPlanReleasedError) throw error; + } +} + const workspaceServicePortPlans = new Map>(); const pendingWorkspaceServicePortPlans = new Map>>(); const pendingWorkspaceServicePortPlanTokens = new Map< @@ -37,25 +55,19 @@ const MAX_DYNAMIC_PORT_ALLOCATION_ATTEMPTS = 10; export async function ensureWorkspaceServicePortPlan( options: EnsureWorkspaceServicePortPlanOptions, ): Promise> { - const existingPlan = workspaceServicePortPlans.get(options.workspaceId); - if (existingPlan) { + const pendingPlan = pendingWorkspaceServicePortPlans.get(options.workspaceId); + if (pendingPlan) { + await waitForWorkspaceServicePortPlanUpdate(pendingPlan); + return ensureWorkspaceServicePortPlan(options); + } + + const existingPlan = + workspaceServicePortPlans.get(options.workspaceId) ?? new Map(); + if (options.services.every((service) => existingPlan.has(service.scriptName))) { return new Map(existingPlan); } - let pendingPlan = pendingWorkspaceServicePortPlans.get(options.workspaceId); - if (!pendingPlan) { - const token: PendingWorkspaceServicePortPlanToken = { isReleased: false }; - pendingPlan = createPendingWorkspaceServicePortPlan({ - workspaceId: options.workspaceId, - services: options.services, - allocatePort: options.allocatePort, - token, - }); - pendingWorkspaceServicePortPlans.set(options.workspaceId, pendingPlan); - pendingWorkspaceServicePortPlanTokens.set(options.workspaceId, token); - } - - return new Map(await pendingPlan); + return new Map(await startWorkspaceServicePortPlanUpdate({ ...options, existingPlan })); } export function requirePlannedWorkspaceServicePort( @@ -82,27 +94,62 @@ export function releaseWorkspaceServicePortPlan(workspaceId: string): void { dynamicPortsByWorkspace.delete(workspaceId); } -async function createPendingWorkspaceServicePortPlan(options: { - workspaceId: string; - services: readonly WorkspaceServicePortDeclaration[]; - allocatePort: (request: WorkspaceServicePortAllocationRequest) => Promise; - token: PendingWorkspaceServicePortPlanToken; -}): Promise> { +interface WorkspaceServicePortPlanUpdateOptions extends EnsureWorkspaceServicePortPlanOptions { + existingPlan: ReadonlyMap; + refreshScriptName?: string; +} + +function startWorkspaceServicePortPlanUpdate( + options: WorkspaceServicePortPlanUpdateOptions, +): Promise> { + const token: PendingWorkspaceServicePortPlanToken = { isReleased: false }; + const plan = createPendingWorkspaceServicePortPlan({ ...options, token }); + pendingWorkspaceServicePortPlans.set(options.workspaceId, plan); + pendingWorkspaceServicePortPlanTokens.set(options.workspaceId, token); + return plan; +} + +async function createPendingWorkspaceServicePortPlan( + options: WorkspaceServicePortPlanUpdateOptions & { token: PendingWorkspaceServicePortPlanToken }, +): Promise> { + const previousDynamicPorts = new Map(dynamicPortsByWorkspace.get(options.workspaceId)); + const retainedPlan = new Map(options.existingPlan); + if (options.refreshScriptName !== undefined) retainedPlan.delete(options.refreshScriptName); try { - const plan = await buildWorkspaceServicePortPlan({ + const updatedPlan = await buildWorkspaceServicePortPlan({ workspaceId: options.workspaceId, services: options.services, allocatePort: options.allocatePort, + existingPlan: retainedPlan, }); + const plan = new Map(options.existingPlan); + for (const [scriptName, port] of updatedPlan) plan.set(scriptName, port); if (options.token.isReleased) { - throw new Error( - `Workspace service port plan was released while being created for '${options.workspaceId}'`, - ); + throw new WorkspaceServicePortPlanReleasedError(options.workspaceId); + } + for (const [scriptName, port] of previousDynamicPorts) { + if (plan.get(scriptName) !== port) { + releaseDynamicPort({ workspaceId: options.workspaceId, scriptName, port }); + } } workspaceServicePortPlans.set(options.workspaceId, plan); return plan; } catch (error) { - releaseWorkspaceServicePortPlan(options.workspaceId); + const dynamicPorts = dynamicPortsByWorkspace.get(options.workspaceId); + if (dynamicPorts) { + for (const [scriptName, port] of dynamicPorts) { + if (options.token.isReleased || previousDynamicPorts.get(scriptName) !== port) { + releaseDynamicPort({ workspaceId: options.workspaceId, scriptName, port }); + } + } + } + if (!options.token.isReleased) { + for (const [scriptName, port] of previousDynamicPorts) { + reserveDynamicPort({ workspaceId: options.workspaceId, scriptName, port }); + } + } + if (options.token.isReleased) + throw new WorkspaceServicePortPlanReleasedError(options.workspaceId); throw error; } finally { pendingWorkspaceServicePortPlans.delete(options.workspaceId); @@ -111,12 +158,16 @@ async function createPendingWorkspaceServicePortPlan(options: { } async function buildWorkspaceServicePortPlan(options: { + existingPlan: ReadonlyMap; workspaceId: string; services: readonly WorkspaceServicePortDeclaration[]; allocatePort: (request: WorkspaceServicePortAllocationRequest) => Promise; }): Promise> { - const explicitPortOwners = new Map(); + const explicitPortOwners = new Map( + Array.from(options.existingPlan, ([scriptName, port]) => [port, scriptName]), + ); for (const service of options.services) { + if (options.existingPlan.has(service.scriptName)) continue; if (service.port === undefined) continue; if (explicitPortOwners.has(service.port)) { throw new Error(`Service '${service.scriptName}' has a duplicate port ${service.port}`); @@ -124,8 +175,9 @@ async function buildWorkspaceServicePortPlan(options: { explicitPortOwners.set(service.port, service.scriptName); } - const plan = new Map(); + const plan = new Map(options.existingPlan); for (const service of options.services) { + if (plan.has(service.scriptName)) continue; if (service.port !== undefined) { plan.set(service.scriptName, service.port); continue; @@ -148,32 +200,20 @@ async function buildWorkspaceServicePortPlan(options: { export async function refreshWorkspaceServicePort( options: RefreshWorkspaceServicePortOptions, ): Promise { - const plan = workspaceServicePortPlans.get(options.workspaceId) ?? new Map(); + const pendingPlan = pendingWorkspaceServicePortPlans.get(options.workspaceId); + if (pendingPlan) { + await waitForWorkspaceServicePortPlanUpdate(pendingPlan); + return refreshWorkspaceServicePort(options); + } - const reservedPorts = new Set(plan.values()); - const previousPort = plan.get(options.service.scriptName); - if (previousPort !== undefined) { - reservedPorts.delete(previousPort); - } - const oldDynamicPort = dynamicPortsByWorkspace - .get(options.workspaceId) - ?.get(options.service.scriptName); - const port = await resolveServicePort({ - service: options.service, + const plan = await startWorkspaceServicePortPlanUpdate({ workspaceId: options.workspaceId, + services: [options.service], allocatePort: options.allocatePort, - reservedPorts, + existingPlan: workspaceServicePortPlans.get(options.workspaceId) ?? new Map(), + refreshScriptName: options.service.scriptName, }); - if (oldDynamicPort !== undefined && oldDynamicPort !== port) { - releaseDynamicPort({ - workspaceId: options.workspaceId, - scriptName: options.service.scriptName, - port: oldDynamicPort, - }); - } - plan.set(options.service.scriptName, port); - workspaceServicePortPlans.set(options.workspaceId, plan); - return port; + return requirePlannedWorkspaceServicePort(plan, options.service.scriptName); } async function resolveServicePort(options: { diff --git a/packages/server/src/server/worktree-bootstrap.test.ts b/packages/server/src/server/worktree-bootstrap.test.ts index c878ef1387..5e5ac1762e 100644 --- a/packages/server/src/server/worktree-bootstrap.test.ts +++ b/packages/server/src/server/worktree-bootstrap.test.ts @@ -837,6 +837,54 @@ describe("runAsyncWorktreeBootstrap", () => { }); }); + it("starts a service added to paseo.json without restarting the daemon", async () => { + const backend = { + type: "service" as const, + command: "uv run uvicorn app.main:app --reload --port $PASEO_PORT", + }; + commitPaseoScripts({ "backend-dev": backend }); + + const serviceProxy = new ScriptRouteStore(); + const runtimeStore = new WorkspaceScriptRuntimeStore(); + const createTerminalCalls: CreateTerminalCall[] = []; + const terminalRecords: StubTerminalRecord[] = []; + const options = { + repoRoot: repoDir, + workspaceId: repoDir, + projectSlug: "repo", + branchName: "feature-added-service", + daemonPort: 6767, + serviceProxy, + runtimeStore, + terminalManager: createStubTerminalManager(createTerminalCalls, terminalRecords), + }; + const backendResult = await spawnWorkspaceScript({ ...options, scriptName: "backend-dev" }); + + writeFileSync( + join(repoDir, "paseo.json"), + JSON.stringify({ + scripts: { + "backend-dev": backend, + web: { type: "service", command: "pnpm run dev -- --port $PASEO_PORT" }, + }, + }), + ); + const webResult = await spawnWorkspaceScript({ ...options, scriptName: "web" }); + + expect(webResult.port).toEqual(expect.any(Number)); + expect(webResult.port).not.toBe(backendResult.port); + expect(terminalRecords[1]?.sentInputs).toEqual(["pnpm run dev -- --port $PASEO_PORT\r"]); + expect(createTerminalCalls[1]?.env?.PASEO_PORT).toBe(String(webResult.port)); + expect(createTerminalCalls[1]?.env?.PASEO_SERVICE_BACKEND_DEV_PORT).toBe( + String(backendResult.port), + ); + expect(serviceProxy.listRoutes()).toHaveLength(2); + expect(runtimeStore.get({ workspaceId: repoDir, scriptName: "backend-dev" })).toMatchObject({ + lifecycle: "running", + terminalId: backendResult.terminalId, + }); + }); + it("spawns services with public aliases and public service URLs", async () => { commitPaseoScripts( {