From ec35bb57afd3d0d2002dd6eff9cb5a8723437c15 Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 6 Aug 2026 17:57:07 +0000 Subject: [PATCH] fix(server): probe instruction lock liveness Co-Authored-By: Paperclip --- .../agent-instructions-service.test.ts | 21 +++- server/src/services/agent-instructions.ts | 101 +++++++++++++++++- 2 files changed, 116 insertions(+), 6 deletions(-) diff --git a/server/src/__tests__/agent-instructions-service.test.ts b/server/src/__tests__/agent-instructions-service.test.ts index 1c1811f72fb..61c58716ed7 100644 --- a/server/src/__tests__/agent-instructions-service.test.ts +++ b/server/src/__tests__/agent-instructions-service.test.ts @@ -1,4 +1,6 @@ +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 { afterEach, describe, expect, it, vi } from "vitest"; @@ -688,13 +690,21 @@ describe("agent instructions service", () => { "instructions", ); const lockPath = `${managedRoot}.paperclip-materialize.lock`; - const token = "00000000-0000-4000-8000-000000000000"; + const token = randomUUID(); + const probePath = `\0paperclip-managed-bundle-${token}`; + const probe = net.createServer((socket) => socket.destroy()); + await new Promise((resolve, reject) => { + probe.once("error", reject); + probe.listen(probePath, resolve); + }); + probe.unref(); await fs.mkdir(lockPath, { recursive: true }); await fs.writeFile( path.join(lockPath, "owner.json"), `${JSON.stringify({ pid: process.pid, token, + probePath, processStartMarker: "unavailable:same-live-process-in-another-format", createdAt: new Date().toISOString(), })}\n`, @@ -714,6 +724,7 @@ describe("agent instructions service", () => { expect(firstOutcome).toBe("waiting"); const releasedLockPath = `${lockPath}.released`; await fs.rename(lockPath, releasedLockPath); + await new Promise((resolve) => probe.close(() => resolve())); await fs.rm(releasedLockPath, { recursive: true, force: true }); await expect(materialization).resolves.toMatchObject({ materialization: { action: "added" }, @@ -722,7 +733,7 @@ describe("agent instructions service", () => { ); it.skipIf(process.platform !== "linux")( - "recovers a reused PID when the owner recorded the ps fallback scheme", + "recovers a reused PID after its liveness probe is released", async () => { const paperclipHome = await makeTempDir("paperclip-agent-instructions-ps-reused-pid-"); cleanupDirs.add(paperclipHome); @@ -741,14 +752,16 @@ describe("agent instructions service", () => { "instructions", ); const lockPath = `${managedRoot}.paperclip-materialize.lock`; - const token = "00000000-0000-4000-8000-000000000000"; + const token = randomUUID(); + const probePath = `\0paperclip-managed-bundle-${token}`; await fs.mkdir(lockPath, { recursive: true }); await fs.writeFile( path.join(lockPath, "owner.json"), `${JSON.stringify({ pid: process.pid, token, - processStartMarker: "ps:not-the-current-process-start", + probePath, + processStartMarker: "unavailable:not-the-current-process-start", createdAt: new Date().toISOString(), })}\n`, "utf8", diff --git a/server/src/services/agent-instructions.ts b/server/src/services/agent-instructions.ts index d7c9a8997b5..a41b09aafb3 100644 --- a/server/src/services/agent-instructions.ts +++ b/server/src/services/agent-instructions.ts @@ -1,6 +1,7 @@ import { createHash, randomUUID } from "node:crypto"; import { execFile } from "node:child_process"; import fs from "node:fs/promises"; +import net from "node:net"; import path from "node:path"; import { notFound, unprocessable } from "../errors.js"; import { resolveHomeAwarePath, resolvePaperclipInstanceRoot } from "../home-paths.js"; @@ -33,6 +34,8 @@ const MANAGED_BUNDLE_LOCK_SUFFIX = ".paperclip-materialize.lock"; const MANAGED_BUNDLE_LOCK_TIMEOUT_MS = 30_000; const MANAGED_BUNDLE_LOCK_HEARTBEAT_MS = 1_000; const MANAGED_BUNDLE_LOCK_STALE_MS = 10_000; +const MANAGED_BUNDLE_LOCK_PROBE_CACHE_MS = 1_000; +const managedBundleProbeConfirmedAt = new Map(); type BundleMode = "managed" | "external"; @@ -237,6 +240,75 @@ async function managedBundleLockAgeMs(targetPath: string): Promise { return stat ? Date.now() - stat.mtimeMs : Number.POSITIVE_INFINITY; } +function managedBundleLockProbePath(token: string): string { + if (process.platform === "win32") { + return `\\\\.\\pipe\\paperclip-managed-bundle-${token}`; + } + if (process.platform === "linux") { + return `\0paperclip-managed-bundle-${token}`; + } + return path.join("/tmp", `.paperclip-managed-bundle-${token}.sock`); +} + +function managedBundleLockProbeHasFilesystemPath(): boolean { + return process.platform !== "win32" && process.platform !== "linux"; +} + +async function startManagedBundleLockProbe(token: string): Promise<{ + path: string; + server: net.Server; +}> { + const probePath = managedBundleLockProbePath(token); + const server = net.createServer((socket) => socket.destroy()); + await new Promise((resolve, reject) => { + const onError = (error: Error) => reject(error); + server.once("error", onError); + server.listen(probePath, () => { + server.off("error", onError); + server.on("error", () => undefined); + server.unref(); + resolve(); + }); + }); + return { path: probePath, server }; +} + +async function stopManagedBundleLockProbe(probe: { + path: string; + server: net.Server; +}): Promise { + await new Promise((resolve) => probe.server.close(() => resolve())); + managedBundleProbeConfirmedAt.delete(probe.path); + if (managedBundleLockProbeHasFilesystemPath()) { + await fs.rm(probe.path, { force: true }).catch(() => undefined); + } +} + +async function managedBundleLockProbeIsAlive(probePath: string): Promise { + const confirmedAt = managedBundleProbeConfirmedAt.get(probePath); + if (confirmedAt !== undefined && Date.now() - confirmedAt < MANAGED_BUNDLE_LOCK_PROBE_CACHE_MS) { + return true; + } + + const alive = await new Promise((resolve) => { + const socket = net.createConnection(probePath); + let settled = false; + const finish = (result: boolean) => { + if (settled) return; + settled = true; + socket.destroy(); + resolve(result); + }; + socket.unref(); + socket.once("connect", () => finish(true)); + socket.once("error", () => finish(false)); + socket.setTimeout(250, () => finish(false)); + }); + if (alive) managedBundleProbeConfirmedAt.set(probePath, Date.now()); + else managedBundleProbeConfirmedAt.delete(probePath); + return alive; +} + async function restoreManagedBundleLockOwner(ownerPath: string, claimedOwnerPath: string): Promise { await fs.rename(claimedOwnerPath, ownerPath).catch(() => undefined); } @@ -288,16 +360,29 @@ function processStartMarkersComparable(left: string, right: string): boolean { async function removeStaleManagedBundleLock(lockPath: string): Promise { let shouldRemove = false; let ownerSnapshot: string | null = null; + let staleProbePath: string | null = null; try { ownerSnapshot = await fs.readFile(path.join(lockPath, "owner.json"), "utf8"); const owner = JSON.parse(ownerSnapshot) as { pid?: unknown; + token?: unknown; + probePath?: unknown; processStartMarker?: unknown; }; const pid = typeof owner.pid === "number" && Number.isInteger(owner.pid) && owner.pid > 0 ? owner.pid : null; - if (pid !== null && isProcessAlive(pid)) { + const token = typeof owner.token === "string" && /^[0-9a-f-]{36}$/i.test(owner.token) + ? owner.token + : null; + const probePath = token !== null + && owner.probePath === managedBundleLockProbePath(token) + ? owner.probePath + : null; + if (probePath !== null) { + shouldRemove = !(await managedBundleLockProbeIsAlive(probePath)); + if (shouldRemove) staleProbePath = probePath; + } else if (pid !== null && isProcessAlive(pid)) { const recordedStart = typeof owner.processStartMarker === "string" ? owner.processStartMarker : null; @@ -329,7 +414,15 @@ async function removeStaleManagedBundleLock(lockPath: string): Promise : await claimManagedBundleLockOwner(lockPath, ownerSnapshot, "reap"); if (!claimPath) return false; return fs.rm(lockPath, { recursive: true, force: true }) - .then(() => true) + .then(async () => { + if (staleProbePath !== null) { + managedBundleProbeConfirmedAt.delete(staleProbePath); + if (managedBundleLockProbeHasFilesystemPath()) { + await fs.rm(staleProbePath, { force: true }).catch(() => undefined); + } + } + return true; + }) .catch(async () => { if (ownerSnapshot !== null) { await restoreManagedBundleLockOwner(path.join(lockPath, "owner.json"), claimPath); @@ -351,6 +444,7 @@ async function withManagedBundleMaterializationLock( const pendingLockPath = `${lockPath}.pending-${token}`; const ownerProcessStartMarker = await processStartMarker(process.pid); const deadline = Date.now() + MANAGED_BUNDLE_LOCK_TIMEOUT_MS; + const probe = await startManagedBundleLockProbe(token); await fs.mkdir(path.dirname(rootPath), { recursive: true }); try { await fs.mkdir(pendingLockPath); @@ -359,6 +453,7 @@ async function withManagedBundleMaterializationLock( `${JSON.stringify({ pid: process.pid, token, + probePath: probe.path, processStartMarker: ownerProcessStartMarker, createdAt: new Date().toISOString(), })}\n`, @@ -380,6 +475,7 @@ async function withManagedBundleMaterializationLock( } } catch (error) { await fs.rm(pendingLockPath, { recursive: true, force: true }).catch(() => undefined); + await stopManagedBundleLockProbe(probe); throw error; } @@ -410,6 +506,7 @@ async function withManagedBundleMaterializationLock( }); } } + await stopManagedBundleLockProbe(probe); } }