diff --git a/server/src/__tests__/runtime-skill-cache.test.ts b/server/src/__tests__/runtime-skill-cache.test.ts index ab63e868ac4..d03475037a5 100644 --- a/server/src/__tests__/runtime-skill-cache.test.ts +++ b/server/src/__tests__/runtime-skill-cache.test.ts @@ -187,6 +187,62 @@ describe("runtime skill revision cache", () => { expect(await runtimeSkillCacheLockHolderExited(Buffer.from("not json"))).toBe(false); }); + it("releases a lock whose holder ran in a deployment the platform replaced", async () => { + // A Railway redeploy starts the server in a new process space, so a PID + // proof never applies. The platform stopped the earlier deployment, which + // held the same exclusive volume, before it started this one. + const railway = { RAILWAY_PROJECT_ID: "project", RAILWAY_ENVIRONMENT_ID: "environment", RAILWAY_SERVICE_ID: "service", + RAILWAY_DEPLOYMENT_ID: "deployment-new", RAILWAY_VOLUME_ID: "volume", RAILWAY_VOLUME_MOUNT_PATH: "/data" }; + for (const [key, value] of Object.entries(railway)) vi.stubEnv(key, value); + try { + const worker = { provider: "railway", projectId: "project", environmentId: "environment", serviceId: "service" }; + const holder = { bootId: randomUUID(), pid: 3, hostHash: "f".repeat(32), processStartedAt: null, + deployment: { worker: { ...worker, deploymentId: "deployment-old" }, exclusiveVolume: true, volumeId: "volume" } }; + expect(await runtimeSkillCacheLockHolderExited(lockStamp(holder))).toBe(true); + // Only a replaced deployment on the same exclusive volume is proof. + const deployment = holder.deployment; + for (const changed of [{ ...deployment, worker: { ...worker, deploymentId: "deployment-new" } }, + { ...deployment, volumeId: "another-volume" }, { ...deployment, exclusiveVolume: false }, + { ...deployment, worker: { ...worker, serviceId: "another-service", deploymentId: "deployment-old" } }]) { + expect(await runtimeSkillCacheLockHolderExited(lockStamp({ ...holder, deployment: changed }))).toBe(false); + } + const { deployment: _omitted, ...withoutDeployment } = holder; + expect(await runtimeSkillCacheLockHolderExited(lockStamp(withoutDeployment))).toBe(false); + } finally { + vi.unstubAllEnvs(); + } + }); + + it("keeps a lock held when the holder proof stalls or fails", async () => { + // A distinct holder per case: a still-pending proof is joined by later waits on the same holder. + const holder = () => ({ bootId: randomUUID(), pid: 3, hostHash: "f".repeat(32), processStartedAt: null }); + const startedAt = Date.now(); + expect(await runtimeSkillCacheLockHolderExited(lockStamp(holder()), + { timeoutMs: 50, prove: () => new Promise(() => {}) })).toBe(false); + expect(Date.now() - startedAt).toBeLessThan(2_000); + expect(await runtimeSkillCacheLockHolderExited(lockStamp(holder()), + { prove: async () => { throw new Error("database unavailable"); } })).toBe(false); + expect(await runtimeSkillCacheLockHolderExited(lockStamp(holder()), + { prove: async () => ({ verdict: "gone", reason: "controller_stopped" }) })).toBe(true); + }); + + it("joins a stalled holder proof instead of starting another read on every timeout", async () => { + const holder = { bootId: randomUUID(), pid: 3, hostHash: "f".repeat(32), processStartedAt: null }; + let settle!: (proof: { verdict: "gone"; reason: "controller_stopped" }) => void; + const prove = vi.fn(() => new Promise<{ verdict: "gone"; reason: "controller_stopped" }>(resolve => { settle = resolve; })); + for (let attempt = 0; attempt < 5; attempt++) { + expect(await runtimeSkillCacheLockHolderExited(lockStamp(holder), { timeoutMs: 10, prove })).toBe(false); + } + expect(prove).toHaveBeenCalledTimes(1); + settle({ verdict: "gone", reason: "controller_stopped" }); + expect(await runtimeSkillCacheLockHolderExited(lockStamp(holder), { timeoutMs: 1_000, prove })).toBe(true); + // A settled proof is not reused: the next wait asks again. + await new Promise(resolve => setImmediate(resolve)); + const fresh = vi.fn(async () => ({ verdict: "alive" as const, reason: "controller_lease_held" as const })); + expect(await runtimeSkillCacheLockHolderExited(lockStamp(holder), { prove: fresh })).toBe(false); + expect(fresh).toHaveBeenCalledTimes(1); + }); + 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 338c7485b60..a5649a2f51a 100644 --- a/server/src/services/controller-process-identity.ts +++ b/server/src/services/controller-process-identity.ts @@ -200,20 +200,6 @@ export function processStartsDefinitelyDiffer( return Number.isFinite(left) && Number.isFinite(right) && Math.abs(left - right) >= PROCESS_START_PRECISION_MS; } -/** True only with proof, in this process's own PID space, that the recorded - * server process has exited. Another space, an unreadable space, this server, - * or an ambiguous live PID returns false. */ -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); - return observed !== null && processStartsDefinitelyDiffer(controller.processStartedAt, new Date(observed)); -} - /** This process's PID-space hash. An unreadable space is stored as * "unknown", which `sameProcessSpace` never accepts. */ export function currentProcessSpaceHash(): string { diff --git a/server/src/services/controller-process-space-identity.test.ts b/server/src/services/controller-process-space-identity.test.ts index bbf3b9046f4..858e7bef487 100644 --- a/server/src/services/controller-process-space-identity.test.ts +++ b/server/src/services/controller-process-space-identity.test.ts @@ -50,10 +50,8 @@ describe("controller process-space identity", () => { expect(state.sysctl).toHaveBeenCalledTimes(2); expect(JSON.stringify(first)).not.toContain(firstBoot); // Old weak/foreign receipts retain their original deadline: no local probe. - const kill = vi.spyOn(process, "kill"); - expect(await owner.controllerProcessExited({ bootId: "old", pid: 2_000_000_001, - hostHash: "old-rounded-clock-hash", processStartedAt: null })).toBe(false); - expect(kill).not.toHaveBeenCalled(); + expect(owner.sameProcessSpace("old-rounded-clock-hash")).toBe(false); + expect(owner.sameProcessSpace(first.hostHash)).toBe(true); }); it.each(["unreadable", "malformed", "unsupported"] as const)("fails closed without an owner PID probe for %s identity", async mode => { @@ -62,10 +60,7 @@ describe("controller process-space identity", () => { if (mode === "malformed") state.sysctl.mockReturnValue("not-a-kernel-uuid"); const owner = await fresh(); expect((await owner.currentControllerProcess()).hostHash).toBe("unknown"); - const kill = vi.spyOn(process, "kill"); - expect(await owner.controllerProcessExited({ bootId: "old", pid: 2_000_000_001, - hostHash: "unknown", processStartedAt: null })).toBe(false); - expect(kill).not.toHaveBeenCalled(); + expect(owner.sameProcessSpace("unknown")).toBe(false); if (mode === "unsupported") expect(state.sysctl).not.toHaveBeenCalled(); }); }); diff --git a/server/src/services/runtime-skill-cache.ts b/server/src/services/runtime-skill-cache.ts index 7c9cef9d7b3..0366a774c37 100644 --- a/server/src/services/runtime-skill-cache.ts +++ b/server/src/services/runtime-skill-cache.ts @@ -3,7 +3,11 @@ 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"; +import { currentControllerProcessStamp, readControllerProcess } from "./controller-process-identity.js"; +import { proveOwnerGone, type LocalOwnerStamp, type OwnerProof } from "./owner-proof.js"; + +/** Bounds one lock-holder proof, which can read the database across process spaces. */ +const LOCK_PROOF_TIMEOUT_MS = 5_000; const FORMAT = 1; const inFlight = new Map>(); @@ -114,18 +118,51 @@ async function matches(spec: CacheSpec, entry = spec.entry): Promise { } /** - * 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. + * True only when the shared owner prover shows that a lock holder's recorded + * server process is gone. Lock age is never proof. In this PID space the PID + * decides. A holder in another process space (for example before a redeploy) + * is gone only when the platform replaced its deployment on the same exclusive + * volume, or when its server boot stopped or was fenced. A lock written before + * holders recorded their process identity stays held. */ -export async function runtimeSkillCacheLockHolderExited(lockContent: Buffer): Promise { +export async function runtimeSkillCacheLockHolderExited(lockContent: Buffer, options: { + /** This wait gives up (the lock stays held) after this long. A stalled read keeps running; later waits on the same holder join it. */ + timeoutMs?: number; + prove?: (stamp: LocalOwnerStamp) => Promise; +} = {}): 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); + if (!controller) return false; + const stamp: LocalOwnerStamp = { + kind: "local", scope: "runtime_skill_cache_lock", generation: null, host: null, processSpace: controller.hostHash, + bootId: controller.bootId, pid: controller.pid, processStartedAt: controller.processStartedAt, processGroupId: null, + ...(controller.deployment ? { worker: controller.deployment.worker, exclusiveVolume: controller.deployment.exclusiveVolume, + volumeId: controller.deployment.volumeId } : {}), + }; + // A proof that outlives its timeout keeps running (its database read cannot + // be cancelled), so later waits on the same holder join it instead of adding + // another read to the shared pool. + const key = JSON.stringify(stamp); + let proved = pendingLockProofs.get(key); + if (!proved) { + proved = (options.prove ?? (value => proveOwnerGone(value)))(stamp) + .then(proof => proof.verdict === "gone", () => false) + .finally(() => { if (pendingLockProofs.get(key) === proved) pendingLockProofs.delete(key); }); + pendingLockProofs.set(key, proved); + } + let timer: NodeJS.Timeout | undefined; + const timedOut = new Promise(resolve => { timer = setTimeout(() => resolve(false), options.timeoutMs ?? LOCK_PROOF_TIMEOUT_MS); }); + try { + return await Promise.race([proved, timedOut]); + } finally { + clearTimeout(timer); + } } +/** Lock-holder proofs still running, by holder stamp; at most one per holder. */ +const pendingLockProofs = new Map>(); + 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 { @@ -221,7 +258,8 @@ async function publishLocked(root: string, fingerprint: string, action: () => 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)) { + if (await runtimeSkillCacheLockHolderExited(held.content, + { timeoutMs: Math.max(1, Math.min(LOCK_PROOF_TIMEOUT_MS, deadline - Date.now())) })) { await releaseExitedRuntimeSkillCacheLock(root, lock, { inode: held.inode, token: holderToken }); continue; }