mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
Harden workspace runtime service recovery
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
9a20303072
commit
bf75361d86
2 files changed
+335
-43
No files matched your search
@@ -1,5 +1,5 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { execFile, spawn } from "node:child_process";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
@@ -48,6 +48,17 @@ import {
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
|
||||
function stableStringifyForTest(value: unknown): string {
|
||||
if (Array.isArray(value)) {
|
||||
return `[${value.map((entry) => stableStringifyForTest(entry)).join(",")}]`;
|
||||
}
|
||||
if (value && typeof value === "object") {
|
||||
const rec = value as Record<string, unknown>;
|
||||
return `{${Object.keys(rec).sort().map((key) => `${JSON.stringify(key)}:${stableStringifyForTest(rec[key])}`).join(",")}}`;
|
||||
}
|
||||
return JSON.stringify(value);
|
||||
}
|
||||
const leasedRunIds = new Set<string>();
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
@@ -3053,6 +3064,7 @@ describeEmbeddedPostgres("workspace runtime startup reconciliation", () => {
|
||||
expect(service?.url).toMatch(/^http:\/\/127\.0\.0\.1:\d+$/);
|
||||
await expect(fetch(service!.url!)).resolves.toMatchObject({ ok: true });
|
||||
|
||||
await fs.rm(paperclipHome, { recursive: true, force: true });
|
||||
await resetRuntimeServicesForTests();
|
||||
|
||||
const result = await reconcilePersistedRuntimeServicesOnStartup(db);
|
||||
@@ -3075,6 +3087,222 @@ describeEmbeddedPostgres("workspace runtime startup reconciliation", () => {
|
||||
await expect(fetch(service!.url!)).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("replaces unhealthy adopted Paperclip dev services on a reused auto port", async () => {
|
||||
const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-unhealthy-adopt-"));
|
||||
const paperclipHome = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-home-"));
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = `runtime-unhealthy-adopt-${randomUUID()}`;
|
||||
|
||||
const portProbe = net.createServer();
|
||||
await new Promise<void>((resolve) => portProbe.listen(0, "127.0.0.1", resolve));
|
||||
const address = portProbe.address();
|
||||
const stalePort = typeof address === "object" && address ? address.port : null;
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
portProbe.close((err) => {
|
||||
if (err) reject(err);
|
||||
else resolve();
|
||||
});
|
||||
});
|
||||
expect(stalePort).toBeTypeOf("number");
|
||||
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const projectId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
const executionWorkspaceId = randomUUID();
|
||||
const stoppedServiceId = randomUUID();
|
||||
const serviceCommand =
|
||||
"node -e \"const http=require('node:http'); const stale=process.env.STALE_HEALTH==='1'; http.createServer((req,res)=>{ if (req.url==='/api/health' && stale) { res.statusCode=503; res.end('database_unreachable'); return; } res.end('ok'); }).listen(Number(process.env.PORT), '127.0.0.1')\"";
|
||||
const scopeType = "agent";
|
||||
const scopeId = agentId;
|
||||
const reuseKey = createHash("sha256")
|
||||
.update(
|
||||
stableStringifyForTest({
|
||||
scopeType,
|
||||
scopeId,
|
||||
serviceName: "paperclip-dev",
|
||||
command: serviceCommand,
|
||||
cwd: workspaceRoot,
|
||||
port: null,
|
||||
env: {},
|
||||
}),
|
||||
)
|
||||
.digest("hex");
|
||||
|
||||
const staleProcess = spawn(resolveShell(), ["-lc", serviceCommand], {
|
||||
cwd: workspaceRoot,
|
||||
env: {
|
||||
...process.env,
|
||||
PORT: String(stalePort),
|
||||
STALE_HEALTH: "1",
|
||||
},
|
||||
detached: process.platform !== "win32",
|
||||
stdio: "ignore",
|
||||
});
|
||||
staleProcess.unref();
|
||||
|
||||
try {
|
||||
const rootUrl = `http://127.0.0.1:${stalePort}`;
|
||||
const healthUrl = `${rootUrl}/api/health`;
|
||||
for (let attempt = 0; attempt < 50; attempt += 1) {
|
||||
try {
|
||||
const response = await fetch(rootUrl);
|
||||
if (response.ok) break;
|
||||
} catch {
|
||||
// Keep polling until the stale process has bound its port.
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
}
|
||||
await expect(fetch(rootUrl)).resolves.toMatchObject({ ok: true });
|
||||
await expect(fetch(healthUrl)).resolves.toMatchObject({ ok: false, status: 503 });
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Codex Coder",
|
||||
role: "engineer",
|
||||
status: "active",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
invocationSource: "manual",
|
||||
status: "running",
|
||||
startedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
});
|
||||
await db.insert(projects).values({
|
||||
id: projectId,
|
||||
companyId,
|
||||
name: "Runtime unhealthy adoption test",
|
||||
status: "in_progress",
|
||||
});
|
||||
await db.insert(executionWorkspaces).values({
|
||||
id: executionWorkspaceId,
|
||||
companyId,
|
||||
projectId,
|
||||
projectWorkspaceId: null,
|
||||
mode: "isolated_workspace",
|
||||
strategyType: "git_worktree",
|
||||
name: "Runtime unhealthy adoption",
|
||||
status: "active",
|
||||
cwd: workspaceRoot,
|
||||
providerType: "git_worktree",
|
||||
providerRef: workspaceRoot,
|
||||
});
|
||||
await db.insert(workspaceRuntimeServices).values({
|
||||
id: stoppedServiceId,
|
||||
companyId,
|
||||
projectId,
|
||||
projectWorkspaceId: null,
|
||||
executionWorkspaceId,
|
||||
issueId: null,
|
||||
scopeType,
|
||||
scopeId,
|
||||
serviceName: "paperclip-dev",
|
||||
status: "stopped",
|
||||
lifecycle: "shared",
|
||||
reuseKey,
|
||||
command: serviceCommand,
|
||||
cwd: workspaceRoot,
|
||||
port: stalePort,
|
||||
url: rootUrl,
|
||||
provider: "local_process",
|
||||
providerRef: String(staleProcess.pid ?? ""),
|
||||
ownerAgentId: null,
|
||||
startedByRunId: null,
|
||||
lastUsedAt: new Date(),
|
||||
startedAt: new Date(),
|
||||
stoppedAt: new Date(),
|
||||
stopPolicy: { type: "manual" },
|
||||
healthStatus: "unknown",
|
||||
});
|
||||
|
||||
leasedRunIds.add(runId);
|
||||
const services = await ensureRuntimeServicesForRun({
|
||||
db,
|
||||
runId,
|
||||
agent: {
|
||||
id: agentId,
|
||||
name: "Codex Coder",
|
||||
companyId,
|
||||
},
|
||||
issue: null,
|
||||
workspace: {
|
||||
...buildWorkspace(workspaceRoot),
|
||||
projectId,
|
||||
workspaceId: null,
|
||||
},
|
||||
executionWorkspaceId,
|
||||
config: {
|
||||
workspaceRuntime: {
|
||||
services: [
|
||||
{
|
||||
name: "paperclip-dev",
|
||||
command: serviceCommand,
|
||||
cwd: ".",
|
||||
port: { type: "auto" },
|
||||
readiness: {
|
||||
type: "http",
|
||||
urlTemplate: "http://127.0.0.1:{{port}}",
|
||||
timeoutSec: 10,
|
||||
intervalMs: 100,
|
||||
},
|
||||
expose: {
|
||||
type: "url",
|
||||
urlTemplate: "http://127.0.0.1:{{port}}",
|
||||
},
|
||||
lifecycle: "shared",
|
||||
reuseScope: "agent",
|
||||
stopPolicy: {
|
||||
type: "manual",
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
adapterEnv: {},
|
||||
});
|
||||
|
||||
expect(services).toHaveLength(1);
|
||||
expect(services[0]?.reused).toBe(false);
|
||||
expect(services[0]?.id).toBe(stoppedServiceId);
|
||||
expect(services[0]?.port).toBe(stalePort);
|
||||
await expect(fetch(healthUrl)).resolves.toMatchObject({ ok: true });
|
||||
expect(await readLocalServicePortOwner(stalePort!)).not.toBe(staleProcess.pid);
|
||||
} finally {
|
||||
leasedRunIds.delete(runId);
|
||||
await releaseRuntimeServicesForRun(runId);
|
||||
await stopRuntimeServicesForExecutionWorkspace({
|
||||
db,
|
||||
executionWorkspaceId,
|
||||
workspaceCwd: workspaceRoot,
|
||||
});
|
||||
if (staleProcess.pid) {
|
||||
try {
|
||||
process.kill(-staleProcess.pid, "SIGKILL");
|
||||
} catch {
|
||||
try {
|
||||
process.kill(staleProcess.pid, "SIGKILL");
|
||||
} catch {
|
||||
// Ignore cleanup races.
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}, 20_000);
|
||||
|
||||
it("marks persisted local services stopped when the registry pid is stale", async () => {
|
||||
const companyId = randomUUID();
|
||||
const runtimeServiceId = randomUUID();
|
||||
|
||||
@@ -1901,10 +1901,43 @@ async function waitForReadiness(input: {
|
||||
throw new Error(`Readiness check failed for ${input.url}: ${lastError}`);
|
||||
}
|
||||
|
||||
async function isRuntimeServiceUrlHealthy(url: string | null) {
|
||||
if (!url) return true;
|
||||
function isPaperclipDevRuntimeService(input: { serviceName?: string | null; command?: string | null }) {
|
||||
const serviceName = (input.serviceName ?? "").trim().toLowerCase();
|
||||
const command = (input.command ?? "").trim().toLowerCase();
|
||||
return (
|
||||
serviceName === "paperclip-dev"
|
||||
|| serviceName === "paperclip-dev-once"
|
||||
|| (command.includes("dev:once") && command.includes("tailscale-auth"))
|
||||
);
|
||||
}
|
||||
|
||||
function resolveRuntimeServiceHealthUrl(
|
||||
url: string | null,
|
||||
input?: { serviceName?: string | null; command?: string | null },
|
||||
) {
|
||||
if (!url || !isPaperclipDevRuntimeService(input ?? {})) return url;
|
||||
try {
|
||||
const response = await fetch(url, { signal: AbortSignal.timeout(2_000) });
|
||||
const parsed = new URL(url);
|
||||
if (parsed.pathname === "/" || parsed.pathname === "") {
|
||||
parsed.pathname = "/api/health";
|
||||
parsed.search = "";
|
||||
parsed.hash = "";
|
||||
return parsed.toString();
|
||||
}
|
||||
} catch {
|
||||
return url;
|
||||
}
|
||||
return url;
|
||||
}
|
||||
|
||||
async function isRuntimeServiceUrlHealthy(
|
||||
url: string | null,
|
||||
input?: { serviceName?: string | null; command?: string | null },
|
||||
) {
|
||||
if (!url) return true;
|
||||
const healthUrl = resolveRuntimeServiceHealthUrl(url, input);
|
||||
try {
|
||||
const response = await fetch(healthUrl, { signal: AbortSignal.timeout(2_000) });
|
||||
return response.ok;
|
||||
} catch {
|
||||
return false;
|
||||
@@ -2188,42 +2221,48 @@ async function startLocalRuntimeService(input: {
|
||||
url,
|
||||
});
|
||||
if (adoptedRecord) {
|
||||
return {
|
||||
id: adoptedRecord.runtimeServiceId ?? randomUUID(),
|
||||
companyId: input.agent.companyId,
|
||||
projectId: input.workspace.projectId,
|
||||
projectWorkspaceId: input.workspace.workspaceId,
|
||||
executionWorkspaceId: input.executionWorkspaceId ?? null,
|
||||
issueId: input.issue?.id ?? null,
|
||||
serviceName,
|
||||
status: "running",
|
||||
lifecycle,
|
||||
scopeType: input.scopeType,
|
||||
scopeId: input.scopeId,
|
||||
reuseKey: input.reuseKey,
|
||||
command,
|
||||
cwd: serviceCwd,
|
||||
port: adoptedRecord.port ?? port,
|
||||
url: adoptedRecord.url ?? url,
|
||||
provider: "local_process",
|
||||
providerRef: String(adoptedRecord.pid),
|
||||
ownerAgentId: input.agent.id ?? null,
|
||||
startedByRunId,
|
||||
lastUsedAt: new Date().toISOString(),
|
||||
startedAt: adoptedRecord.startedAt,
|
||||
stoppedAt: null,
|
||||
stopPolicy,
|
||||
healthStatus: "healthy",
|
||||
reused: true,
|
||||
db: input.db,
|
||||
child: null,
|
||||
leaseRunIds: leaseRunId ? new Set([leaseRunId]) : new Set(),
|
||||
idleTimer: null,
|
||||
envFingerprint,
|
||||
serviceKey,
|
||||
profileKind: "workspace-runtime",
|
||||
processGroupId: adoptedRecord.processGroupId ?? null,
|
||||
};
|
||||
const adoptedUrl = adoptedRecord.url ?? url;
|
||||
if (!(await isRuntimeServiceUrlHealthy(adoptedUrl, { serviceName, command }))) {
|
||||
await terminateLocalService(adoptedRecord);
|
||||
await removeLocalServiceRegistryRecord(adoptedRecord.serviceKey);
|
||||
} else {
|
||||
return {
|
||||
id: adoptedRecord.runtimeServiceId ?? randomUUID(),
|
||||
companyId: input.agent.companyId,
|
||||
projectId: input.workspace.projectId,
|
||||
projectWorkspaceId: input.workspace.workspaceId,
|
||||
executionWorkspaceId: input.executionWorkspaceId ?? null,
|
||||
issueId: input.issue?.id ?? null,
|
||||
serviceName,
|
||||
status: "running",
|
||||
lifecycle,
|
||||
scopeType: input.scopeType,
|
||||
scopeId: input.scopeId,
|
||||
reuseKey: input.reuseKey,
|
||||
command,
|
||||
cwd: serviceCwd,
|
||||
port: adoptedRecord.port ?? port,
|
||||
url: adoptedRecord.url ?? url,
|
||||
provider: "local_process",
|
||||
providerRef: String(adoptedRecord.pid),
|
||||
ownerAgentId: input.agent.id ?? null,
|
||||
startedByRunId,
|
||||
lastUsedAt: new Date().toISOString(),
|
||||
startedAt: adoptedRecord.startedAt,
|
||||
stoppedAt: null,
|
||||
stopPolicy,
|
||||
healthStatus: "healthy",
|
||||
reused: true,
|
||||
db: input.db,
|
||||
child: null,
|
||||
leaseRunIds: leaseRunId ? new Set([leaseRunId]) : new Set(),
|
||||
idleTimer: null,
|
||||
envFingerprint,
|
||||
serviceKey,
|
||||
profileKind: "workspace-runtime",
|
||||
processGroupId: adoptedRecord.processGroupId ?? null,
|
||||
};
|
||||
}
|
||||
}
|
||||
if (identityPort) {
|
||||
const ownerPid = await readLocalServicePortOwner(identityPort);
|
||||
@@ -2841,13 +2880,38 @@ export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) {
|
||||
let adopted = 0;
|
||||
let stopped = 0;
|
||||
for (const row of rows) {
|
||||
const adoptedRecord = await findLocalServiceRegistryRecordByRuntimeServiceId({
|
||||
let adoptedRecord = await findLocalServiceRegistryRecordByRuntimeServiceId({
|
||||
runtimeServiceId: row.id,
|
||||
profileKind: "workspace-runtime",
|
||||
});
|
||||
if (!adoptedRecord && row.command && row.cwd) {
|
||||
adoptedRecord = await findAdoptableLocalService({
|
||||
serviceKey: createLocalServiceKey({
|
||||
profileKind: "workspace-runtime",
|
||||
serviceName: row.serviceName,
|
||||
cwd: row.cwd,
|
||||
command: row.command,
|
||||
envFingerprint: row.reuseKey ?? "",
|
||||
port: null,
|
||||
scope: {
|
||||
scopeType: row.scopeType as RuntimeServiceRecord["scopeType"],
|
||||
scopeId: row.scopeId ?? null,
|
||||
executionWorkspaceId: row.executionWorkspaceId ?? null,
|
||||
reuseKey: row.reuseKey ?? null,
|
||||
},
|
||||
}),
|
||||
profileKind: "workspace-runtime",
|
||||
serviceName: row.serviceName,
|
||||
command: row.command,
|
||||
cwd: row.cwd,
|
||||
envFingerprint: row.reuseKey ?? "",
|
||||
port: row.port ?? null,
|
||||
url: row.url ?? null,
|
||||
});
|
||||
}
|
||||
if (adoptedRecord) {
|
||||
const adoptedUrl = adoptedRecord.url ?? row.url ?? null;
|
||||
if (!(await isRuntimeServiceUrlHealthy(adoptedUrl))) {
|
||||
if (!(await isRuntimeServiceUrlHealthy(adoptedUrl, { serviceName: row.serviceName, command: row.command }))) {
|
||||
await removeLocalServiceRegistryRecord(adoptedRecord.serviceKey);
|
||||
} else {
|
||||
const record: RuntimeServiceRecord = {
|
||||
|
||||
Reference in new issue
Block a user