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( {