mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 20:21:50 +01:00
fix(runtime): defer fixed-port readiness after commit
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
b8637fb3f4
commit
22af9b3203
2 files changed
+238
-40
No files matched your search
@@ -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<typeof startRuntimeServicesForWorkspaceControl> | 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();
|
||||
|
||||
@@ -175,6 +175,10 @@ type LocalRuntimeServiceStart = {
|
||||
readiness: Promise<void>;
|
||||
};
|
||||
|
||||
type PendingRuntimeServiceReadiness = LocalRuntimeServiceStart & {
|
||||
service: Record<string, unknown>;
|
||||
};
|
||||
|
||||
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<string, unknown>;
|
||||
record: RuntimeServiceRecord;
|
||||
} | null;
|
||||
excludedPorts?: ReadonlySet<number>;
|
||||
},
|
||||
): Promise<WorkspaceControlStartBatch> {
|
||||
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<string, unknown>,
|
||||
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<RuntimeServiceRef[]> {
|
||||
@@ -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<number>();
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user