diff --git a/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/orch.test.ts b/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/orch.test.ts index 2ef435d7d..c3dd99481 100644 --- a/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/orch.test.ts +++ b/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/orch.test.ts @@ -6,8 +6,10 @@ import { readFile, readdir, rm, + unlink, writeFile, } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; import { realpathSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -436,6 +438,50 @@ describe("Store", () => { expect(await readdir(directory)).not.toContain(".orch.lock"); }); + it("lets exactly one of two concurrent takers replace a stale lock", async () => { + const { directory } = await initializedStore(); + const exited = Bun.spawn(["true"]); + await exited.exited; + await writeFile(join(directory, ".orch.lock"), `${exited.pid}\n`); + + // Both stores stay open until both attempts settle, so the winner still + // holds the lock while the loser retries. + const stores = [useStore(directory, {}), useStore(directory, {})]; + const results = await Promise.allSettled( + stores.map((store) => + store.units.add({ id: `u-${randomUUID()}`, track: "build" }) + ) + ); + const fulfilled = results.filter((r) => r.status === "fulfilled"); + const rejected = results.filter( + (r): r is PromiseRejectedResult => r.status === "rejected" + ); + expect(fulfilled).toHaveLength(1); + expect(rejected).toHaveLength(1); + expect(String(rejected[0].reason)).toMatch(/store lock held by pid/); + expect(await readdir(directory)).not.toContain(".orch.lock.takeover"); + await Promise.all(stores.map((store) => store.close())); + }); + + it("does not unlink a stale lock while another process is mid-takeover", async () => { + const { directory } = await initializedStore(); + const exited = Bun.spawn(["true"]); + await exited.exited; + await writeFile(join(directory, ".orch.lock"), `${exited.pid}\n`); + // Simulate a competing process that has claimed the takeover but not yet + // installed its lock. + await writeFile(join(directory, ".orch.lock.takeover"), ""); + + const store = useStore(directory, {}); + await expect( + store.units.add({ id: "u1", track: "build" }) + ).rejects.toThrow(/store lock held by pid/); + expect(await readFile(join(directory, ".orch.lock"), "utf8")).toBe( + `${exited.pid}\n` + ); + await unlink(join(directory, ".orch.lock.takeover")); + }); + it("blocks a writer and steals the pid lock only with force", async () => { const { directory, store } = await initializedStore(); await store.close(); diff --git a/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/store.ts b/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/store.ts index 417368918..9ba876cb1 100644 --- a/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/store.ts +++ b/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/store.ts @@ -391,17 +391,61 @@ async function acquireLock( await handle.close(); }; + const heldBy = async (): Promise => { + let retryHolder = "unknown"; + try { + retryHolder = (await readFile(path, "utf8")).trim() || "unknown"; + } catch { + retryHolder = "unknown"; + } + throw new UserError(`store lock held by pid ${retryHolder}`); + }; + + // Replacing a stale lock is unlink-then-create, which is not atomic: two + // processes that both read the same dead holder could each unlink and the + // second unlink would remove the first one's fresh lock. Serialise the + // takeover itself through a second `wx` file so only one process performs + // the unlink; the other falls back to a plain create and, on EEXIST, reports + // the winner's pid. const takeOver = async (): Promise => { - await unlink(path); + const takeoverPath = `${path}.takeover`; + let takeoverHandle; + try { + takeoverHandle = await open(takeoverPath, "wx"); + } catch (takeoverError) { + if (errorCode(takeoverError) !== "EEXIST") { + throw takeoverError; + } + // Another process is mid-takeover. Give it a moment to install its + // lock, then try the normal path once. + await new Promise((resolve) => setTimeout(resolve, 20)); + try { + await create(); + return; + } catch (retryError) { + if (errorCode(retryError) === "EEXIST") { + return heldBy(); + } + throw retryError; + } + } try { - await create(); - } catch (retryError) { - if (errorCode(retryError) === "EEXIST") { - const retryHolder = - (await readFile(path, "utf8")).trim() || "unknown"; - throw new UserError(`store lock held by pid ${retryHolder}`); + await takeoverHandle.close(); + await unlink(path).catch((unlinkError) => { + if (errorCode(unlinkError) !== "ENOENT") { + throw unlinkError; + } + }); + try { + await create(); + } catch (retryError) { + if (errorCode(retryError) === "EEXIST") { + return heldBy(); + } + throw retryError; } - throw retryError; + } finally { + await unlink(takeoverPath).catch(() => {}); } }; diff --git a/pstack/skills/poteto-mode/scripts/orch/orch.test.ts b/pstack/skills/poteto-mode/scripts/orch/orch.test.ts index 2ef435d7d..c3dd99481 100644 --- a/pstack/skills/poteto-mode/scripts/orch/orch.test.ts +++ b/pstack/skills/poteto-mode/scripts/orch/orch.test.ts @@ -6,8 +6,10 @@ import { readFile, readdir, rm, + unlink, writeFile, } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; import { realpathSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -436,6 +438,50 @@ describe("Store", () => { expect(await readdir(directory)).not.toContain(".orch.lock"); }); + it("lets exactly one of two concurrent takers replace a stale lock", async () => { + const { directory } = await initializedStore(); + const exited = Bun.spawn(["true"]); + await exited.exited; + await writeFile(join(directory, ".orch.lock"), `${exited.pid}\n`); + + // Both stores stay open until both attempts settle, so the winner still + // holds the lock while the loser retries. + const stores = [useStore(directory, {}), useStore(directory, {})]; + const results = await Promise.allSettled( + stores.map((store) => + store.units.add({ id: `u-${randomUUID()}`, track: "build" }) + ) + ); + const fulfilled = results.filter((r) => r.status === "fulfilled"); + const rejected = results.filter( + (r): r is PromiseRejectedResult => r.status === "rejected" + ); + expect(fulfilled).toHaveLength(1); + expect(rejected).toHaveLength(1); + expect(String(rejected[0].reason)).toMatch(/store lock held by pid/); + expect(await readdir(directory)).not.toContain(".orch.lock.takeover"); + await Promise.all(stores.map((store) => store.close())); + }); + + it("does not unlink a stale lock while another process is mid-takeover", async () => { + const { directory } = await initializedStore(); + const exited = Bun.spawn(["true"]); + await exited.exited; + await writeFile(join(directory, ".orch.lock"), `${exited.pid}\n`); + // Simulate a competing process that has claimed the takeover but not yet + // installed its lock. + await writeFile(join(directory, ".orch.lock.takeover"), ""); + + const store = useStore(directory, {}); + await expect( + store.units.add({ id: "u1", track: "build" }) + ).rejects.toThrow(/store lock held by pid/); + expect(await readFile(join(directory, ".orch.lock"), "utf8")).toBe( + `${exited.pid}\n` + ); + await unlink(join(directory, ".orch.lock.takeover")); + }); + it("blocks a writer and steals the pid lock only with force", async () => { const { directory, store } = await initializedStore(); await store.close(); diff --git a/pstack/skills/poteto-mode/scripts/orch/store.ts b/pstack/skills/poteto-mode/scripts/orch/store.ts index 417368918..9ba876cb1 100644 --- a/pstack/skills/poteto-mode/scripts/orch/store.ts +++ b/pstack/skills/poteto-mode/scripts/orch/store.ts @@ -391,17 +391,61 @@ async function acquireLock( await handle.close(); }; + const heldBy = async (): Promise => { + let retryHolder = "unknown"; + try { + retryHolder = (await readFile(path, "utf8")).trim() || "unknown"; + } catch { + retryHolder = "unknown"; + } + throw new UserError(`store lock held by pid ${retryHolder}`); + }; + + // Replacing a stale lock is unlink-then-create, which is not atomic: two + // processes that both read the same dead holder could each unlink and the + // second unlink would remove the first one's fresh lock. Serialise the + // takeover itself through a second `wx` file so only one process performs + // the unlink; the other falls back to a plain create and, on EEXIST, reports + // the winner's pid. const takeOver = async (): Promise => { - await unlink(path); + const takeoverPath = `${path}.takeover`; + let takeoverHandle; + try { + takeoverHandle = await open(takeoverPath, "wx"); + } catch (takeoverError) { + if (errorCode(takeoverError) !== "EEXIST") { + throw takeoverError; + } + // Another process is mid-takeover. Give it a moment to install its + // lock, then try the normal path once. + await new Promise((resolve) => setTimeout(resolve, 20)); + try { + await create(); + return; + } catch (retryError) { + if (errorCode(retryError) === "EEXIST") { + return heldBy(); + } + throw retryError; + } + } try { - await create(); - } catch (retryError) { - if (errorCode(retryError) === "EEXIST") { - const retryHolder = - (await readFile(path, "utf8")).trim() || "unknown"; - throw new UserError(`store lock held by pid ${retryHolder}`); + await takeoverHandle.close(); + await unlink(path).catch((unlinkError) => { + if (errorCode(unlinkError) !== "ENOENT") { + throw unlinkError; + } + }); + try { + await create(); + } catch (retryError) { + if (errorCode(retryError) === "EEXIST") { + return heldBy(); + } + throw retryError; } - throw retryError; + } finally { + await unlink(takeoverPath).catch(() => {}); } };