diff --git a/server/src/__tests__/workspace-runtime.test.ts b/server/src/__tests__/workspace-runtime.test.ts index f25377efba8..1d97db74918 100644 --- a/server/src/__tests__/workspace-runtime.test.ts +++ b/server/src/__tests__/workspace-runtime.test.ts @@ -5988,6 +5988,96 @@ describeEmbeddedPostgres("workspace runtime service control persistence", () => } }, 20_000); + it("commits fixed-port reservations before readiness windows longer than 60 seconds", async () => { + const fixture = await createRuntimeFixture(); + const cleanupRuntimeHome = await createRuntimeHome(); + const workspace = fixture.workspaces[0]!; + const basePort = await findFreePort(); + const spawnedMarkerPath = path.join(workspace.cwd, "slow-readiness-spawned.marker"); + const releaseMarkerPath = path.join(workspace.cwd, "slow-readiness-release.marker"); + const serverScript = [ + "const fs=require('node:fs');", + "const http=require('node:http');", + `const spawned=${JSON.stringify(spawnedMarkerPath)};`, + `const release=${JSON.stringify(releaseMarkerPath)};`, + "fs.writeFileSync(spawned,'spawned');", + "const timer=setInterval(()=>{", + "if(!fs.existsSync(release))return;", + "clearInterval(timer);", + "http.createServer((_req,res)=>res.end('ok')).listen(Number(process.env.PORT),'127.0.0.1');", + "},25);", + "setInterval(()=>{},1000);", + ].join(""); + const command = `${JSON.stringify(process.execPath)} -e ${JSON.stringify(serverScript)}`; + const config = fixedPortRuntimeConfig(basePort, command); + config.workspaceRuntime.services[0]!.readiness.timeoutSec = 70; + let startPromise: ReturnType | null = null; + + try { + startPromise = startRuntimeServicesForWorkspaceControl({ + db, + actor: fixture.actor, + issue: null, + workspace: fixture.realizedWorkspace(workspace), + executionWorkspaceId: workspace.id, + config, + adapterEnv: {}, + }); + startPromise.catch(() => undefined); + + const markerDeadline = Date.now() + 5_000; + while (!existsSync(spawnedMarkerPath) && Date.now() < markerDeadline) { + await new Promise((resolve) => setTimeout(resolve, 25)); + } + expect(existsSync(spawnedMarkerPath)).toBe(true); + + const parentWrite = (async () => await db + .update(executionWorkspaces) + .set({ updatedAt: new Date() }) + .where(eq(executionWorkspaces.id, workspace.id)))(); + const parentWriteOutcome = await Promise.race([ + parentWrite.then(() => "committed" as const), + new Promise<"locked">((resolve) => setTimeout(() => resolve("locked"), 2_000)), + ]); + + const readinessWaitStartedAt = Date.now(); + await new Promise((resolve) => setTimeout(resolve, 60_250)); + expect(Date.now() - readinessWaitStartedAt).toBeGreaterThanOrEqual(60_000); + + await fs.writeFile(releaseMarkerPath, "release"); + const started = (await startPromise)[0]!; + await parentWrite; + + expect(parentWriteOutcome).toBe("committed"); + expect(started).toMatchObject({ + port: basePort, + url: `http://127.0.0.1:${basePort}`, + status: "running", + healthStatus: "healthy", + }); + const persisted = await db + .select() + .from(workspaceRuntimeServices) + .where(eq(workspaceRuntimeServices.executionWorkspaceId, workspace.id)) + .then((rows) => rows[0] ?? null); + expect(persisted).toMatchObject({ + port: basePort, + url: `http://127.0.0.1:${basePort}`, + status: "running", + }); + } finally { + await fs.writeFile(releaseMarkerPath, "release").catch(() => undefined); + await startPromise?.catch(() => undefined); + await stopRuntimeServicesForExecutionWorkspace({ + db, + executionWorkspaceId: workspace.id, + workspaceCwd: workspace.cwd, + }).catch(() => undefined); + await cleanupRuntimeHome(); + await fixture.cleanup(); + } + }, 80_000); + it("reuses a stopped isolated workspace's actual port when it is free", async () => { const fixture = await createRuntimeFixture(); const cleanupRuntimeHome = await createRuntimeHome(); diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index 82e41b41b98..b3ce928873c 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -175,6 +175,10 @@ type LocalRuntimeServiceStart = { readiness: Promise; }; +type PendingRuntimeServiceReadiness = LocalRuntimeServiceStart & { + service: Record; +}; + type StoppedRuntimeServiceReuseCandidate = { id: string; port: number | null; @@ -4878,7 +4882,7 @@ async function startLocalRuntimeService( input.allowFixedPortFallback && portType !== "auto" && explicitPort > 0, ); const retryBindCollisions = fixedPortFallbackEnabled || portType === "auto"; - const deferReadiness = Boolean(options?.deferReadiness && !fixedPortFallbackEnabled); + const deferReadiness = Boolean(options?.deferReadiness); try { for (let attempt = 0; attempt < WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS; attempt += 1) { @@ -5247,7 +5251,7 @@ type StartRuntimeServicesForWorkspaceControlInput = { type WorkspaceControlStartBatch = { refs: RuntimeServiceRef[]; - pendingReadiness: LocalRuntimeServiceStart[]; + pendingReadiness: PendingRuntimeServiceReadiness[]; startedServiceIds: string[]; }; @@ -5266,10 +5270,11 @@ async function startRuntimeServicesForWorkspaceControlUnlocked( service: Record; record: RuntimeServiceRecord; } | null; + excludedPorts?: ReadonlySet; }, ): Promise { const refs: RuntimeServiceRef[] = []; - const pendingReadiness: LocalRuntimeServiceStart[] = []; + const pendingReadiness: PendingRuntimeServiceReadiness[] = []; const startedServiceIds: string[] = []; for (const service of rawServices) { @@ -5334,6 +5339,7 @@ async function startRuntimeServicesForWorkspaceControlUnlocked( ? options.preparedProvisioning.record : undefined, allowFixedPortFallback: options?.allowFixedPortFallback, + excludedPorts: options?.excludedPorts, reuseKey, scopeType, scopeId, @@ -5352,7 +5358,7 @@ async function startRuntimeServicesForWorkspaceControlUnlocked( // Attach a rejection handler immediately; the caller awaits the same promise after // the DB transaction commits, but transaction failures may skip that wait path. started.readiness.catch(() => undefined); - pendingReadiness.push(started); + pendingReadiness.push({ ...started, service }); startedServiceIds.push(started.record.id); } } @@ -5360,6 +5366,63 @@ async function startRuntimeServicesForWorkspaceControlUnlocked( return { refs, pendingReadiness, startedServiceIds }; } +async function lockWorkspaceRuntimeStartParents( + db: Db, + input: StartRuntimeServicesForWorkspaceControlInput, +) { + let allowFixedPortFallback = false; + if (input.executionWorkspaceId) { + const [lockedExecutionWorkspace] = await db + .select({ id: executionWorkspaces.id, mode: executionWorkspaces.mode }) + .from(executionWorkspaces) + .where( + and( + eq(executionWorkspaces.id, input.executionWorkspaceId), + eq(executionWorkspaces.companyId, input.actor.companyId), + ), + ) + .for("update"); + if (!lockedExecutionWorkspace) throw new Error("Execution workspace not found before starting runtime services"); + allowFixedPortFallback = lockedExecutionWorkspace.mode === "isolated_workspace"; + } + + if (input.workspace.workspaceId) { + const [lockedProjectWorkspace] = await db + .select({ id: projectWorkspaces.id }) + .from(projectWorkspaces) + .where( + and( + eq(projectWorkspaces.id, input.workspace.workspaceId), + eq(projectWorkspaces.companyId, input.actor.companyId), + ), + ) + .for("update"); + if (!lockedProjectWorkspace) throw new Error("Project workspace not found before starting runtime services"); + } + + return allowFixedPortFallback; +} + +function canRetryDeferredPortBindCollision( + service: Record, + allowFixedPortFallback: boolean, +) { + const portConfig = parseObject(service.port); + const portType = asString(portConfig.type, ""); + const explicitPort = asNumber(portConfig.value, asNumber(service.port, 0)); + return portType === "auto" || Boolean(allowFixedPortFallback && explicitPort > 0); +} + +async function discardFailedDeferredRuntimeStart(db: Db, record: RuntimeServiceRecord) { + clearIdleTimer(record); + if (runtimeServicesById.get(record.id) === record) runtimeServicesById.delete(record.id); + if (record.reuseKey && runtimeServicesByReuseKey.get(record.reuseKey) === record.id) { + runtimeServicesByReuseKey.delete(record.reuseKey); + } + await removeLocalServiceRegistryRecord(record.serviceKey).catch(() => undefined); + await persistRuntimeServiceRecord(db, record); +} + export async function startRuntimeServicesForWorkspaceControl( input: StartRuntimeServicesForWorkspaceControlInput, ): Promise { @@ -5446,35 +5509,7 @@ export async function startRuntimeServicesForWorkspaceControl( await input.db.transaction(async (tx) => { const txDb = tx as unknown as Db; - - if (input.executionWorkspaceId) { - const [lockedExecutionWorkspace] = await tx - .select({ id: executionWorkspaces.id, mode: executionWorkspaces.mode }) - .from(executionWorkspaces) - .where( - and( - eq(executionWorkspaces.id, input.executionWorkspaceId), - eq(executionWorkspaces.companyId, input.actor.companyId), - ), - ) - .for("update"); - if (!lockedExecutionWorkspace) throw new Error("Execution workspace not found before starting runtime services"); - allowFixedPortFallback = lockedExecutionWorkspace.mode === "isolated_workspace"; - } - - if (input.workspace.workspaceId) { - const [lockedProjectWorkspace] = await tx - .select({ id: projectWorkspaces.id }) - .from(projectWorkspaces) - .where( - and( - eq(projectWorkspaces.id, input.workspace.workspaceId), - eq(projectWorkspaces.companyId, input.actor.companyId), - ), - ) - .for("update"); - if (!lockedProjectWorkspace) throw new Error("Project workspace not found before starting runtime services"); - } + allowFixedPortFallback = await lockWorkspaceRuntimeStartParents(txDb, input); // Branch reconciliation takes these same parent row locks before mutating // a recorded branch. Persisting a `starting` service row before commit closes @@ -5495,13 +5530,86 @@ export async function startRuntimeServicesForWorkspaceControl( ); }); - for (const pending of startBatch.pendingReadiness) { - try { - await pending.readiness; - await persistRuntimeServiceRecord(input.db, pending.record); - } catch (error) { - await persistRuntimeServiceRecord(input.db, pending.record).catch(() => undefined); - throw error; + // Readiness never uses the transaction-scoped DB handle. A late bind collision is + // recorded after commit, then only the next bounded reservation re-enters a short + // parent-locked transaction. Slow builds therefore cannot retain the parent locks. + for (const initialPending of startBatch.pendingReadiness) { + let pending: PendingRuntimeServiceReadiness | null = initialPending; + const excludedPorts = new Set(); + + while (pending) { + try { + await pending.readiness; + await persistRuntimeServiceRecord(input.db, pending.record); + break; + } catch (error) { + await discardFailedDeferredRuntimeStart(input.db, pending.record); + if ( + !(error instanceof RuntimeServicePortBindCollision) + || !canRetryDeferredPortBindCollision(pending.service, allowFixedPortFallback) + ) { + throw error; + } + + excludedPorts.add(error.port); + if (excludedPorts.size >= WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS) { + const portConfig = parseObject(pending.service.port); + const portType = asString(portConfig.type, ""); + const preferredPort = asNumber(portConfig.value, asNumber(pending.service.port, 0)); + if (allowFixedPortFallback && portType !== "auto" && preferredPort > 0 && input.executionWorkspaceId) { + throw await buildRuntimePortAllocationConflict({ + db: input.db, + companyId: input.actor.companyId, + executionWorkspaceId: input.executionWorkspaceId, + preferredPort, + attemptedPorts: [...excludedPorts], + }); + } + throw conflict("No safe automatically allocated runtime service port is available.", { + code: "workspace_runtime_port_allocation_exhausted", + attemptedPortCount: excludedPorts.size, + remediation: "Retry the start or configure a different runtime service port.", + }); + } + + const failedRecordId = pending.record.id; + let retryBatch: WorkspaceControlStartBatch = { + refs: [], + pendingReadiness: [], + startedServiceIds: [], + }; + let retryAllowsFixedPortFallback = false; + await input.db.transaction(async (tx) => { + const txDb = tx as unknown as Db; + retryAllowsFixedPortFallback = await lockWorkspaceRuntimeStartParents(txDb, input); + if (!canRetryDeferredPortBindCollision(pending!.service, retryAllowsFixedPortFallback)) { + throw error; + } + retryBatch = await startRuntimeServicesForWorkspaceControlUnlocked( + { ...input, db: txDb }, + [pending!.service], + invocationId, + txDb, + input.db, + { + deferReadiness: true, + allowFixedPortFallback: retryAllowsFixedPortFallback, + provisionCoordinator, + excludedPorts, + }, + ); + }); + allowFixedPortFallback = retryAllowsFixedPortFallback; + for (const serviceId of retryBatch.startedServiceIds) { + if (!startBatch.startedServiceIds.includes(serviceId)) { + startBatch.startedServiceIds.push(serviceId); + } + } + const replacementRef = retryBatch.refs[0]; + const failedRefIndex = startBatch.refs.findIndex((ref) => ref.id === failedRecordId); + if (replacementRef && failedRefIndex >= 0) startBatch.refs[failedRefIndex] = replacementRef; + pending = retryBatch.pendingReadiness[0] ?? null; + } } }