diff --git a/server/src/__tests__/runtime-skill-cache.test.ts b/server/src/__tests__/runtime-skill-cache.test.ts index 06ee38f311a..ab63e868ac4 100644 --- a/server/src/__tests__/runtime-skill-cache.test.ts +++ b/server/src/__tests__/runtime-skill-cache.test.ts @@ -6,7 +6,19 @@ import os from "node:os"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { CompanySkill } from "@paperclipai/shared"; -import { removeRuntimeSkillCache, resolveRuntimeSkillCache, runtimeSkillCacheSpec } from "../services/runtime-skill-cache.js"; +import { once } from "node:events"; +import { releaseExitedRuntimeSkillCacheLock, removeRuntimeSkillCache, resolveRuntimeSkillCache, runtimeSkillCacheLockHolderExited, runtimeSkillCacheSpec } from "../services/runtime-skill-cache.js"; +import { currentControllerProcessStamp } from "../services/controller-process-identity.js"; + +/** A PID whose process has exited and been reaped. */ +async function exitedPid(): Promise { + const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + await once(child, "exit"); + return child.pid!; +} + +const lockStamp = (controller: Record, token: string = randomUUID()) => + Buffer.from(JSON.stringify({ pid: controller.pid, host: os.hostname(), token, controller })); async function makeWritable(root: string): Promise { const stat = await fs.lstat(root); @@ -85,6 +97,96 @@ describe("runtime skill revision cache", () => { expect(await launch()).toEqual({ source: first.source, reads: 0 }); }); + it("releases a publication lock at once when its holder provably exited", async () => { + const spec = runtimeSkillCacheSpec(root, skill)!; + const namespace = path.dirname(spec.root); + await fs.mkdir(namespace, { recursive: true }); + const pid = await exitedPid(); + const token = randomUUID(); + const stale = lockStamp({ ...currentControllerProcessStamp(), bootId: randomUUID(), pid, processStartedAt: null }, token); + // The crashed holder's owner file and its hard-linked lock, as publishLocked leaves them. + await fs.writeFile(path.join(namespace, `.owner-${token}`), stale); + await fs.link(path.join(namespace, `.owner-${token}`), path.join(namespace, `${skill.id}.lock`)); + const startedAt = Date.now(); + // Concurrent waiters in one process share one build; separate processes race for the claim. + const sources = await Promise.all([resolveRuntimeSkillCache(spec, reader()), + resolveRuntimeSkillCache(runtimeSkillCacheSpec(root, { ...skill })!, reader())]); + expect(new Set(sources)).toEqual(new Set([path.join(spec.entry, "files")])); + expect(Date.now() - startedAt).toBeLessThan(5_000); + expect(await fs.readdir(namespace)).not.toContain(`${skill.id}.lock`); + }); + + it("takes over a release claim whose claimer exited before removing the stale lock", async () => { + const spec = runtimeSkillCacheSpec(root, skill)!; + const namespace = path.dirname(spec.root); + await fs.mkdir(namespace, { recursive: true }); + const dead = { ...currentControllerProcessStamp(), bootId: randomUUID(), processStartedAt: null }; + const token = randomUUID(); + await fs.writeFile(path.join(namespace, `.owner-${token}`), lockStamp({ ...dead, pid: await exitedPid() }, token)); + await fs.link(path.join(namespace, `.owner-${token}`), path.join(namespace, `${skill.id}.lock`)); + // The waiter that claimed the stale lock crashed before unlinking it, + // leaving its claim published from a kept owner file. + const claimToken = randomUUID(); + await fs.writeFile(path.join(namespace, `.owner-${claimToken}`), + JSON.stringify({ token: claimToken, controller: { ...dead, pid: await exitedPid() } })); + await fs.link(path.join(namespace, `.owner-${claimToken}`), path.join(namespace, `${skill.id}.lock.released-${token}`)); + const startedAt = Date.now(); + expect(await resolveRuntimeSkillCache(spec, reader())).toBe(path.join(spec.entry, "files")); + expect(Date.now() - startedAt).toBeLessThan(5_000); + }); + + it("publishes a release claim only with its complete body", async () => { + const namespace = path.join(root, "claims"); + await fs.mkdir(namespace); + const lock = path.join(namespace, "skill.lock"); + const pid = await exitedPid(); + const token = randomUUID(); + await fs.writeFile(path.join(namespace, `.owner-${token}`), + lockStamp({ ...currentControllerProcessStamp(), bootId: randomUUID(), pid, processStartedAt: null }, token)); + await fs.link(path.join(namespace, `.owner-${token}`), lock); + const stale = await fs.lstat(lock); + await releaseExitedRuntimeSkillCacheLock(namespace, lock, { inode: stale.ino, token }); + const claim = await fs.readFile(`${lock}.released-${token}`, "utf8"); + expect(JSON.parse(claim)).toMatchObject({ token: expect.any(String), controller: { pid: process.pid } }); + // The claim keeps an owner file with the same inode, so its inode is never reused. + const claimInode = (await fs.lstat(`${lock}.released-${token}`)).ino; + const owners = await Promise.all((await fs.readdir(namespace)).filter(name => name.startsWith(".owner-")) + .map(async name => (await fs.lstat(path.join(namespace, name))).ino)); + expect(owners).toContain(claimInode); + await expect(fs.lstat(lock)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("never removes a newer lock for a waiter that read a stale one late", async () => { + const namespace = path.join(root, "locks"); + await fs.mkdir(namespace); + const lock = path.join(namespace, "skill.lock"); + const staleToken = randomUUID(); + const liveToken = randomUUID(); + // The stale lock was already released and a live holder took the path. + await fs.writeFile(path.join(namespace, `.owner-${liveToken}`), lockStamp(currentControllerProcessStamp(), liveToken)); + await fs.link(path.join(namespace, `.owner-${liveToken}`), lock); + const live = await fs.lstat(lock); + // Another waiter's claim exists: a late waiter does nothing, even when the + // live lock happens to reuse the stale inode number. + await fs.writeFile(`${lock}.released-${staleToken}`, "{}"); + await releaseExitedRuntimeSkillCacheLock(namespace, lock, { inode: live.ino, token: staleToken }); + expect((await fs.lstat(lock)).ino).toBe(live.ino); + // Without a claim, a different inode keeps the live lock too. + await releaseExitedRuntimeSkillCacheLock(namespace, lock, { inode: live.ino + 1, token: randomUUID() }); + expect((await fs.lstat(lock)).ino).toBe(live.ino); + }); + + it("never treats age, another host, or a legacy holder as proof that a lock holder exited", async () => { + const pid = await exitedPid(); + const here = { ...currentControllerProcessStamp(), bootId: randomUUID(), pid, processStartedAt: null }; + expect(await runtimeSkillCacheLockHolderExited(lockStamp(here))).toBe(true); + expect(await runtimeSkillCacheLockHolderExited(lockStamp({ ...here, hostHash: "another-host" }))).toBe(false); + expect(await runtimeSkillCacheLockHolderExited(Buffer.from(JSON.stringify({ pid, host: os.hostname() })))).toBe(false); + expect(await runtimeSkillCacheLockHolderExited(lockStamp({ ...here, pid: process.pid }))).toBe(false); + expect(await runtimeSkillCacheLockHolderExited(lockStamp(currentControllerProcessStamp()))).toBe(false); + expect(await runtimeSkillCacheLockHolderExited(Buffer.from("not json"))).toBe(false); + }); + it("fingerprints installed content and ownership, excluding cosmetic metadata", async () => { const spec = runtimeSkillCacheSpec(root, skill)!; expect(runtimeSkillCacheSpec(root, { ...skill, name: "Renamed", updatedAt: new Date(), metadata: { ...skill.metadata, starred: true, ref: "new-branch" } })!.fingerprint).toBe(spec.fingerprint); diff --git a/server/src/services/controller-process-identity.ts b/server/src/services/controller-process-identity.ts index 6065aa41408..29500709a7b 100644 --- a/server/src/services/controller-process-identity.ts +++ b/server/src/services/controller-process-identity.ts @@ -85,10 +85,23 @@ export function currentControllerBootId(): string { return bootId; } +let resolvedProcessStartedAt: string | null = null; + export async function currentControllerProcess(): Promise { const startedAt = await currentProcessStartedAtMs(); - return { bootId, pid: process.pid, hostHash, - processStartedAt: startedAt === null ? null : new Date(startedAt).toISOString() }; + const iso = startedAt === null ? null : new Date(startedAt).toISOString(); + resolvedProcessStartedAt = iso; + return { bootId, pid: process.pid, hostHash, processStartedAt: iso }; +} + +/** + * This process's identity without any I/O, for stamps written on the turn + * path. The start time is present only once an earlier call resolved it. A + * stamp without one still proves exit when its PID is gone in this PID space; + * a live PID then stays ambiguous, which fails closed. + */ +export function currentControllerProcessStamp(): ControllerProcess { + return { bootId, pid: process.pid, hostHash, processStartedAt: resolvedProcessStartedAt }; } export function readControllerProcess(value: unknown): ControllerProcess | null { @@ -137,6 +150,8 @@ export function processStartsDefinitelyDiffer( export async function controllerProcessExited(controller: ControllerProcess): Promise { if (controller.bootId === bootId || hostHash === UNKNOWN_PROCESS_SPACE || controller.hostHash !== hostHash) return false; if (!processExists(controller.pid)) return true; + // Without a recorded start time a live PID can never prove exit; skip the probe. + if (!Number.isFinite(Date.parse(controller.processStartedAt ?? ""))) return false; // The PID now belongs to a live process, possibly this restarted server. // Only a definitely different start time proves the recorded process is gone. const observed = await observedProcessStartedAt(controller.pid); diff --git a/server/src/services/runtime-skill-cache.ts b/server/src/services/runtime-skill-cache.ts index 010fb28bf1f..7c9cef9d7b3 100644 --- a/server/src/services/runtime-skill-cache.ts +++ b/server/src/services/runtime-skill-cache.ts @@ -3,6 +3,7 @@ import { constants, promises as fs } from "node:fs"; import os from "node:os"; import path from "node:path"; import type { CompanySkill } from "@paperclipai/shared"; +import { controllerProcessExited, currentControllerProcessStamp, readControllerProcess } from "./controller-process-identity.js"; const FORMAT = 1; const inFlight = new Map>(); @@ -112,34 +113,122 @@ async function matches(spec: CacheSpec, entry = spec.entry): Promise { } catch { return false; } } +/** + * True only when a lock holder's recorded server process has provably exited + * in this PID space. Lock age is never proof. A holder on another host or PID + * namespace, an unreadable space, or a lock written before holders recorded + * their process identity stays held. + */ +export async function runtimeSkillCacheLockHolderExited(lockContent: Buffer): Promise { + let holder: unknown; + try { holder = JSON.parse(lockContent.toString("utf8")); } catch { return false; } + const controller = readControllerProcess((holder as { controller?: unknown } | null)?.controller); + return controller !== null && await controllerProcessExited(controller); +} + +const OWNER_TOKEN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; + +function lockHolderToken(lockContent: Buffer): string | null { + try { + const token = (JSON.parse(lockContent.toString("utf8")) as { token?: unknown } | null)?.token; + return typeof token === "string" && OWNER_TOKEN.test(token) ? token : null; + } catch { return null; } +} + +/** + * Removes a lock whose holder exited. Every waiter that read the same stale + * lock races for one exclusive claim named by the holder's unique token, so + * only one of them ever unlinks it. While the stale lock exists no one else + * can acquire or remove it (its holder is gone), so the inode check before the + * unlink only confirms the claim is still about that lock. Every lock and + * claim inode is pinned by a kept owner file, so it is never reused while a + * delayed waiter may compare against it, and a waiter that read the stale + * lock late finds the claim and does nothing. A claim is itself a lock with a token and controller: when its + * claimer provably exited before finishing, the same rule releases the claim + * and the next poll claims the stale lock again. + */ +export async function releaseExitedRuntimeSkillCacheLock(root: string, lock: string, + stale: { inode: number; token: string }, depth = 0): Promise { + const claim = `${lock}.released-${stale.token}`; + // Publish the claim the way a lock is published: write a complete owner + // file, then hard-link it into place, so a claim name never exists without + // its body. The owner file is kept: it pins the claim's inode, so a deleted + // claim's inode is never reused by a newer claim that a delayed waiter + // might then compare against. + const claimToken = randomUUID(); + const claimOwner = path.join(root, `.owner-${claimToken}`); + await fs.writeFile(claimOwner, JSON.stringify({ token: claimToken, controller: currentControllerProcessStamp() }), { flag: "wx" }); + try { + await fs.link(claimOwner, claim); + } catch (error) { + await fs.unlink(claimOwner).catch(() => {}); + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + // Bounded: a claimer crash inside this two-syscall window is already rare. + if (depth >= 2) return; + const handle = await fs.open(claim, constants.O_RDONLY | constants.O_NOFOLLOW).catch((openError: NodeJS.ErrnoException) => { + if (openError.code === "ENOENT") return null; + throw openError; + }); + if (!handle) return; + let held: { content: Buffer; inode: number }; + try { held = { content: await handle.readFile(), inode: (await handle.stat()).ino }; } + finally { await handle.close(); } + const heldToken = lockHolderToken(held.content); + if (heldToken && await runtimeSkillCacheLockHolderExited(held.content)) { + await releaseExitedRuntimeSkillCacheLock(root, claim, { inode: held.inode, token: heldToken }, depth + 1); + } + return; + } + const current = await fs.lstat(lock).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; + throw error; + }); + if (current?.ino === stale.inode) await fs.unlink(lock); +} + // Serialize builds and cleanup for one skill across processes as well as callers. -// A hard link publishes complete lock ownership atomically; crashed owners are reported without stealing another publisher’s lock. +// A hard link publishes complete lock ownership atomically. A lock whose holder +// provably exited is released at once; any other holder is waited for. async function publishLocked(root: string, fingerprint: string, action: () => Promise): Promise { const lock = path.join(root, `${fingerprint}.lock`); - const owner = path.join(root, `.owner-${randomUUID()}`); - await fs.writeFile(owner, JSON.stringify({ pid: process.pid, host: os.hostname() }), { flag: "wx" }); + const token = randomUUID(); + const owner = path.join(root, `.owner-${token}`); + // `pid` and `host` stay for older readers; `controller` is the proof input. + // The stamp does no I/O, so preparation adds no wait before the turn. + await fs.writeFile(owner, JSON.stringify({ pid: process.pid, host: os.hostname(), token, + controller: currentControllerProcessStamp() }), { flag: "wx" }); let acquired = false; + // A live holder is re-proved at most once a second, not on every 20 ms poll. + let judged: { inode: number; at: number } | null = null; try { const deadline = Date.now() + 60_000; while (!acquired) { try { await fs.link(owner, lock); acquired = true; } catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; - const lockContent = await readRegularFile(lock).catch((readError: NodeJS.ErrnoException) => { - if (readError.code === "ENOENT") return null; - throw readError; + const handle = await fs.open(lock, constants.O_RDONLY | constants.O_NOFOLLOW).catch((openError: NodeJS.ErrnoException) => { + if (openError.code === "ENOENT") return null; + throw openError; }); - if (!lockContent) continue; - const holder = JSON.parse(lockContent.toString("utf8")); - if (holder.host === os.hostname() && Number.isSafeInteger(holder.pid) && holder.pid > 0) { - try { process.kill(holder.pid, 0); } - catch (probeError) { - if ((probeError as NodeJS.ErrnoException).code === "ESRCH") { - throw new Error("Runtime skill cache publisher exited; remove its stale publication lock before retrying"); - } + if (!handle) continue; + let held: { content: Buffer; inode: number }; + try { + const stat = await handle.stat(); + if (!stat.isFile()) throw new Error("Unsafe runtime skill cache file"); + held = { content: await handle.readFile(), inode: stat.ino }; + } finally { await handle.close(); } + const holderToken = lockHolderToken(held.content); + const due = judged === null || judged.inode !== held.inode || Date.now() - judged.at >= 1_000; + if (holderToken && due) { + judged = { inode: held.inode, at: Date.now() }; + if (await runtimeSkillCacheLockHolderExited(held.content)) { + await releaseExitedRuntimeSkillCacheLock(root, lock, { inode: held.inode, token: holderToken }); + continue; } } - if (Date.now() >= deadline) throw new Error("Runtime skill cache publisher is busy; retry preparation"); + if (Date.now() >= deadline) { + throw new Error(`Runtime skill cache publisher is busy; retry preparation. If no server process holds ${lock}, remove it.`); + } await new Promise((resolve) => setTimeout(resolve, 20)); } }