pstack orch: serialise stale-lock takeover so two takers cannot both hold the store lock

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();

pstack/skills/poteto-mode/scripts/orch/store.ts

@@ -391,17 +391,61 @@ async function acquireLock(
     await handle.close();
   };
 
+  const heldBy = async (): Promise<never> => {
+    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<void> => {
-    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(() => {});
     }
   };