fix(runtime): verify mutation lock publication

loopx/control_plane/effect_runtime_io.ts

@@ -56,6 +56,18 @@ async function readFileIdentity(path: string): Promise<CreatedFileIdentity | nul
   }
 }
 
+async function publishedMutationLockMatches(
+  lockPath: string,
+  identity: CreatedFileIdentity,
+  ownerPid: number,
+  token: string,
+): Promise<boolean> {
+  if (!sameFileIdentity(identity, await readFileIdentity(lockPath))) return false;
+  const owner = await readMutationLockOwner(lockPath);
+  if (owner?.pid !== ownerPid || owner.token !== token) return false;
+  return sameFileIdentity(identity, await readFileIdentity(lockPath));
+}
+
 function processIsAlive(pid: number): boolean {
   if (!Number.isSafeInteger(pid) || pid <= 0) return false;
   try {
@@ -322,6 +334,23 @@ export async function acquireFileMutationLock(
         await removeCreatedFile(lockPath, identity);
         throw error;
       }
+      if (!identity) {
+        throw new Error("mutation lock identity was not captured");
+      }
+      if (
+        !(await publishedMutationLockMatches(
+          lockPath,
+          identity,
+          ownerPid,
+          token,
+        ))
+      ) {
+        await removeCreatedFile(lockPath, identity);
+        if (Date.now() >= deadline) {
+          throw new EffectRuntimeLockTimeoutError();
+        }
+        continue;
+      }
       return { targetPath, lockPath, token };
     } catch (error) {
       if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;

tests/control_plane_ts/effect_runtime_io.test.ts

@@ -1,5 +1,5 @@
 import assert from "node:assert/strict";
-import { mkdtemp, rm, stat, utimes, writeFile } from "node:fs/promises";
+import { mkdtemp, open, rm, stat, utimes, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import test, { type TestContext } from "node:test";
@@ -93,3 +93,38 @@ test("blank mutation lock tokens are not treated as valid owners", async (t) =>
   );
   assert.equal(await mutationLockOwner(target), null);
 });
+
+test("acquire rejects success when its lock inode is replaced before owner publication", async (t) => {
+  const root = await workspace(t);
+  const target = join(root, "state");
+  const lockPath = `${target}.ts-effect.lock`;
+  const replacement = { pid: process.pid, token: "replacement-token" };
+  const probe = await open(join(root, "probe"), "w");
+  const prototype = Object.getPrototypeOf(probe) as {
+    writeFile(data: string, encoding: BufferEncoding): Promise<void>;
+  };
+  await probe.close();
+  const writeFileToHandle = prototype.writeFile;
+  let replaced = false;
+
+  prototype.writeFile = async function (data, encoding) {
+    if (!replaced) {
+      replaced = true;
+      await rm(lockPath, { force: true });
+      await writeFile(lockPath, JSON.stringify(replacement), "utf8");
+    }
+    await writeFileToHandle.call(this, data, encoding);
+  };
+
+  try {
+    await assert.rejects(
+      acquireFileMutationLock(target, process.pid, 0),
+      { code: "mutation_lock_timeout" },
+    );
+    assert.equal(replaced, true);
+    assert.deepEqual(await mutationLockOwner(target), replacement);
+  } finally {
+    prototype.writeFile = writeFileToHandle;
+    await rm(lockPath, { force: true });
+  }
+});