mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-27 19:35:13 +01:00
Fix workspace runtime service adoption
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
5b6fc549a9
commit
52fd6b7cc6
3 files changed
+95
-7
No files matched your search
@@ -1,6 +1,7 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
@@ -37,7 +38,7 @@ import {
|
||||
stopRuntimeServicesForExecutionWorkspace,
|
||||
type RealizedExecutionWorkspace,
|
||||
} from "../services/workspace-runtime.ts";
|
||||
import { writeLocalServiceRegistryRecord } from "../services/local-service-supervisor.ts";
|
||||
import { readLocalServicePortOwner, writeLocalServiceRegistryRecord } from "../services/local-service-supervisor.ts";
|
||||
import { resolvePaperclipConfigPath } from "../paths.ts";
|
||||
import type { WorkspaceOperation } from "@paperclipai/shared";
|
||||
import type { WorkspaceOperationRecorder } from "../services/workspace-operations.ts";
|
||||
@@ -2915,6 +2916,34 @@ describe("resolveShell (shell fallback)", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("readLocalServicePortOwner", () => {
|
||||
it("detects the owner of a listening TCP port", async () => {
|
||||
try {
|
||||
await execFileAsync("lsof", ["-v"]);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
const server = net.createServer();
|
||||
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
try {
|
||||
const address = server.address();
|
||||
const port = typeof address === "object" && address ? address.port : null;
|
||||
expect(port).toBeTypeOf("number");
|
||||
|
||||
const owner = await readLocalServicePortOwner(port!);
|
||||
expect(owner).toBe(process.pid);
|
||||
} finally {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((err) => {
|
||||
if (err) reject(err);
|
||||
else resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describeEmbeddedPostgres("workspace runtime startup reconciliation", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
@@ -249,12 +249,17 @@ async function isLikelyMatchingCommand(record: LocalServiceRegistryRecord) {
|
||||
|
||||
export async function findAdoptableLocalService(input: {
|
||||
serviceKey: string;
|
||||
profileKind?: string | null;
|
||||
serviceName?: string | null;
|
||||
command?: string | null;
|
||||
cwd?: string | null;
|
||||
envFingerprint?: string | null;
|
||||
port?: number | null;
|
||||
url?: string | null;
|
||||
}) {
|
||||
const record = await readLocalServiceRegistryRecord(input.serviceKey);
|
||||
const record =
|
||||
await readLocalServiceRegistryRecord(input.serviceKey)
|
||||
?? await adoptLocalServiceFromPortOwner(input);
|
||||
if (!record) return null;
|
||||
|
||||
if (!isPidAlive(record.pid)) {
|
||||
@@ -272,6 +277,59 @@ export async function findAdoptableLocalService(input: {
|
||||
return record;
|
||||
}
|
||||
|
||||
async function readProcessGroupId(pid: number) {
|
||||
if (process.platform === "win32") return null;
|
||||
try {
|
||||
const { stdout } = await execFileAsync("ps", ["-o", "pgid=", "-p", String(pid)]);
|
||||
const parsed = Number.parseInt(stdout.trim(), 10);
|
||||
return Number.isInteger(parsed) && parsed > 0 ? parsed : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function adoptLocalServiceFromPortOwner(input: {
|
||||
serviceKey: string;
|
||||
profileKind?: string | null;
|
||||
serviceName?: string | null;
|
||||
command?: string | null;
|
||||
cwd?: string | null;
|
||||
envFingerprint?: string | null;
|
||||
port?: number | null;
|
||||
url?: string | null;
|
||||
}) {
|
||||
if (!input.port) return null;
|
||||
const ownerPid = await readLocalServicePortOwner(input.port);
|
||||
if (!ownerPid) return null;
|
||||
|
||||
const processGroupId = await readProcessGroupId(ownerPid);
|
||||
const pid = processGroupId && isPidAlive(processGroupId) ? processGroupId : ownerPid;
|
||||
const now = new Date().toISOString();
|
||||
const record: LocalServiceRegistryRecord = {
|
||||
version: 1,
|
||||
serviceKey: input.serviceKey,
|
||||
profileKind: input.profileKind ?? "workspace-runtime",
|
||||
serviceName: input.serviceName ?? "service",
|
||||
command: input.command ?? input.serviceName ?? "service",
|
||||
cwd: input.cwd ?? process.cwd(),
|
||||
envFingerprint: input.envFingerprint ?? "",
|
||||
port: input.port,
|
||||
url: input.url ?? null,
|
||||
pid,
|
||||
processGroupId: processGroupId ?? pid,
|
||||
provider: "local_process",
|
||||
runtimeServiceId: null,
|
||||
reuseKey: input.envFingerprint ?? null,
|
||||
startedAt: now,
|
||||
lastSeenAt: now,
|
||||
metadata: null,
|
||||
};
|
||||
|
||||
if (!(await isLikelyMatchingCommand(record))) return null;
|
||||
await writeLocalServiceRegistryRecord(record);
|
||||
return record;
|
||||
}
|
||||
|
||||
export async function touchLocalServiceRegistryRecord(
|
||||
serviceKey: string,
|
||||
patch?: Partial<Omit<LocalServiceRegistryRecord, "serviceKey" | "version">>,
|
||||
@@ -334,7 +392,7 @@ export async function terminateLocalService(
|
||||
export async function readLocalServicePortOwner(port: number) {
|
||||
if (!Number.isInteger(port) || port <= 0 || process.platform === "win32") return null;
|
||||
try {
|
||||
const { stdout } = await execFileAsync("lsof", ["-nPiTCP", `:${port}`, "-sTCP:LISTEN", "-t"]);
|
||||
const { stdout } = await execFileAsync("lsof", ["-nP", `-iTCP:${port}`, "-sTCP:LISTEN", "-t"]);
|
||||
const firstPid = stdout
|
||||
.split("\n")
|
||||
.map((line) => Number.parseInt(line.trim(), 10))
|
||||
|
||||
@@ -2125,9 +2125,7 @@ async function startLocalRuntimeService(input: {
|
||||
});
|
||||
const reusableStoppedPort =
|
||||
asString(portConfig.type, "") === "auto" && stoppedReuseCandidate?.port
|
||||
? (await readLocalServicePortOwner(stoppedReuseCandidate.port))
|
||||
? null
|
||||
: stoppedReuseCandidate.port
|
||||
? stoppedReuseCandidate.port
|
||||
: null;
|
||||
const port =
|
||||
asString(portConfig.type, "") === "auto"
|
||||
@@ -2181,10 +2179,13 @@ async function startLocalRuntimeService(input: {
|
||||
});
|
||||
const adoptedRecord = await findAdoptableLocalService({
|
||||
serviceKey,
|
||||
profileKind: "workspace-runtime",
|
||||
serviceName,
|
||||
command,
|
||||
cwd: serviceCwd,
|
||||
envFingerprint: serviceIdentityFingerprint,
|
||||
port: identityPort,
|
||||
port: port ?? identityPort,
|
||||
url,
|
||||
});
|
||||
if (adoptedRecord) {
|
||||
return {
|
||||
|
||||
Reference in new issue
Block a user