mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 04:02:18 +01:00
fix(server): claim stale instruction locks atomically
Move the observed owner record out of the lock before deleting it so competing stale-lock recovery cannot remove a replacement owner's lock. Apply the same atomic claim to normal release and cover eight concurrent stale-lock contenders. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
bbdf358049
commit
fab8c2af2a
2 files changed
+116
-8
No files matched your search
@@ -489,6 +489,48 @@ describe("agent instructions service", () => {
|
||||
await expect(fs.stat(lockPath)).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("lets only one contender reap a stale managed materialization lock", async () => {
|
||||
const paperclipHome = await makeTempDir("paperclip-agent-instructions-stale-lock-race-");
|
||||
cleanupDirs.add(paperclipHome);
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "test-instance";
|
||||
const agent = makeAgent({});
|
||||
const managedRoot = path.join(
|
||||
paperclipHome,
|
||||
"instances",
|
||||
"test-instance",
|
||||
"companies",
|
||||
"company-1",
|
||||
"agents",
|
||||
"agent-1",
|
||||
"instructions",
|
||||
);
|
||||
const lockPath = `${managedRoot}.paperclip-materialize.lock`;
|
||||
await fs.mkdir(lockPath, { recursive: true });
|
||||
await fs.writeFile(
|
||||
path.join(lockPath, "owner.json"),
|
||||
`${JSON.stringify({
|
||||
pid: 2_147_483_647,
|
||||
token: "00000000-0000-4000-8000-000000000000",
|
||||
createdAt: new Date().toISOString(),
|
||||
})}\n`,
|
||||
"utf8",
|
||||
);
|
||||
|
||||
const results = await Promise.all(
|
||||
Array.from({ length: 8 }, (_, index) => agentInstructionsService().materializeManagedBundle(
|
||||
agent,
|
||||
{ "AGENTS.md": `# Stock ${index}\n` },
|
||||
)),
|
||||
);
|
||||
|
||||
expect(results.filter((result) => result.materialization.action === "added")).toHaveLength(1);
|
||||
expect(results.filter((result) => result.materialization.action === "skipped")).toHaveLength(7);
|
||||
await expect(fs.stat(lockPath)).rejects.toThrow();
|
||||
const live = await fs.readFile(path.join(managedRoot, "AGENTS.md"), "utf8");
|
||||
expect(Array.from({ length: 8 }, (_, index) => `# Stock ${index}\n`)).toContain(live);
|
||||
});
|
||||
|
||||
it("does not evict an old materialization lock owned by a live process", async () => {
|
||||
const paperclipHome = await makeTempDir("paperclip-agent-instructions-live-lock-");
|
||||
cleanupDirs.add(paperclipHome);
|
||||
|
||||
@@ -223,6 +223,46 @@ async function managedBundleLockAgeMs(targetPath: string): Promise<number> {
|
||||
return stat ? Date.now() - stat.mtimeMs : Number.POSITIVE_INFINITY;
|
||||
}
|
||||
|
||||
async function restoreManagedBundleLockOwner(ownerPath: string, claimedOwnerPath: string): Promise<void> {
|
||||
await fs.rename(claimedOwnerPath, ownerPath).catch(() => undefined);
|
||||
}
|
||||
|
||||
async function claimManagedBundleLockOwner(
|
||||
lockPath: string,
|
||||
expectedOwner: string,
|
||||
purpose: "release" | "reap",
|
||||
): Promise<string | null> {
|
||||
const ownerPath = path.join(lockPath, "owner.json");
|
||||
const claimedOwnerPath = path.join(lockPath, `owner.${purpose}-${randomUUID()}.json`);
|
||||
try {
|
||||
await fs.rename(ownerPath, claimedOwnerPath);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
|
||||
const claimedOwner = await fs.readFile(claimedOwnerPath, "utf8").catch(() => null);
|
||||
if (claimedOwner !== expectedOwner) {
|
||||
await restoreManagedBundleLockOwner(ownerPath, claimedOwnerPath);
|
||||
return null;
|
||||
}
|
||||
return claimedOwnerPath;
|
||||
}
|
||||
|
||||
async function claimOwnerlessManagedBundleLock(lockPath: string): Promise<string | null> {
|
||||
const ownerPath = path.join(lockPath, "owner.json");
|
||||
const claimPath = path.join(lockPath, "owner.reaping");
|
||||
try {
|
||||
await fs.writeFile(claimPath, randomUUID(), { encoding: "utf8", flag: "wx" });
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (await statIfExists(ownerPath)) {
|
||||
await fs.rm(claimPath, { force: true }).catch(() => undefined);
|
||||
return null;
|
||||
}
|
||||
return claimPath;
|
||||
}
|
||||
|
||||
function processStartMarkersComparable(left: string, right: string): boolean {
|
||||
const leftSeparator = left.indexOf(":");
|
||||
const rightSeparator = right.indexOf(":");
|
||||
@@ -233,8 +273,10 @@ function processStartMarkersComparable(left: string, right: string): boolean {
|
||||
|
||||
async function removeStaleManagedBundleLock(lockPath: string): Promise<boolean> {
|
||||
let shouldRemove = false;
|
||||
let ownerSnapshot: string | null = null;
|
||||
try {
|
||||
const owner = JSON.parse(await fs.readFile(path.join(lockPath, "owner.json"), "utf8")) as {
|
||||
ownerSnapshot = await fs.readFile(path.join(lockPath, "owner.json"), "utf8");
|
||||
const owner = JSON.parse(ownerSnapshot) as {
|
||||
pid?: unknown;
|
||||
token?: unknown;
|
||||
processStartMarker?: unknown;
|
||||
@@ -272,8 +314,20 @@ async function removeStaleManagedBundleLock(lockPath: string): Promise<boolean>
|
||||
shouldRemove = await managedBundleLockAgeMs(lockPath) > MANAGED_BUNDLE_LOCK_STALE_MS;
|
||||
}
|
||||
if (!shouldRemove) return false;
|
||||
await fs.rm(lockPath, { recursive: true, force: true }).catch(() => undefined);
|
||||
return true;
|
||||
const claimPath = ownerSnapshot === null
|
||||
? await claimOwnerlessManagedBundleLock(lockPath)
|
||||
: await claimManagedBundleLockOwner(lockPath, ownerSnapshot, "reap");
|
||||
if (!claimPath) return false;
|
||||
return fs.rm(lockPath, { recursive: true, force: true })
|
||||
.then(() => true)
|
||||
.catch(async () => {
|
||||
if (ownerSnapshot !== null) {
|
||||
await restoreManagedBundleLockOwner(path.join(lockPath, "owner.json"), claimPath);
|
||||
} else {
|
||||
await fs.rm(claimPath, { force: true }).catch(() => undefined);
|
||||
}
|
||||
return false;
|
||||
});
|
||||
}
|
||||
|
||||
async function withManagedBundleMaterializationLock<T>(
|
||||
@@ -326,11 +380,23 @@ async function withManagedBundleMaterializationLock<T>(
|
||||
return await operation();
|
||||
} finally {
|
||||
clearInterval(heartbeat);
|
||||
const owner = await fs.readFile(ownerPath, "utf8")
|
||||
.then((raw) => JSON.parse(raw) as { token?: unknown })
|
||||
.catch(() => null);
|
||||
if (owner?.token === token) {
|
||||
await fs.rm(lockPath, { recursive: true, force: true }).catch(() => undefined);
|
||||
const ownerSnapshot = await fs.readFile(ownerPath, "utf8").catch(() => null);
|
||||
const owner = ownerSnapshot === null
|
||||
? null
|
||||
: (() => {
|
||||
try {
|
||||
return JSON.parse(ownerSnapshot) as { token?: unknown };
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
})();
|
||||
if (owner?.token === token && ownerSnapshot !== null) {
|
||||
const claimPath = await claimManagedBundleLockOwner(lockPath, ownerSnapshot, "release");
|
||||
if (claimPath) {
|
||||
await fs.rm(lockPath, { recursive: true, force: true }).catch(async () => {
|
||||
await restoreManagedBundleLockOwner(ownerPath, claimPath);
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user