test(FN-4822): complete Step 5 — add cross-node claim mutex integration race test

Fusion-Task-Id: FN-4822
Fusion-Task-Lineage: 08cc29e8-114a-48dc-80de-8d7fd2ce0e69

packages/engine/src/__tests__/cross-node-claim-mutex.integration.test.ts

@@ -0,0 +1,83 @@
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { mkdtempSync } from "node:fs";
+import { rm } from "node:fs/promises";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { AgentStore, CheckoutConflictError, TaskStore, createCentralDatabase } from "@fusion/core";
+
+function makeTmpDir(): string {
+  return mkdtempSync(join(tmpdir(), "fn-cross-node-claim-test-"));
+}
+
+describe("cross-node claim mutex integration", () => {
+  let rootDir: string;
+  let globalDir: string;
+  let taskStore: TaskStore;
+  let centralDb: CentralDatabase;
+  let storeA: AgentStore;
+  let storeB: AgentStore;
+  let agentA: string;
+  let agentB: string;
+  let taskId: string;
+
+  beforeEach(async () => {
+    rootDir = makeTmpDir();
+    globalDir = join(rootDir, ".fusion-global");
+    taskStore = new TaskStore(rootDir, globalDir);
+    await taskStore.init();
+    centralDb = createCentralDatabase(globalDir);
+    centralDb.init();
+
+    storeA = new AgentStore({ rootDir, taskStore, claimStore: centralDb, projectId: "P-1", nodeId: "node-a" });
+    storeB = new AgentStore({ rootDir, taskStore, claimStore: centralDb, projectId: "P-1", nodeId: "node-b" });
+    await storeA.init();
+    await storeB.init();
+
+    agentA = (await storeA.createAgent({ name: "agent-a", role: "executor" })).id;
+    agentB = (await storeB.createAgent({ name: "agent-b", role: "executor" })).id;
+    taskId = (await taskStore.createTask({ description: "cross-node claim race" })).id;
+  });
+
+  afterEach(async () => {
+    storeA?.close();
+    storeB?.close();
+    taskStore?.close();
+    centralDb?.close();
+    await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
+  });
+
+  it("allows one winner per race and bumps epoch once per successful ownership acquisition", async () => {
+    const [first, second] = await Promise.allSettled([
+      storeA.checkoutTask(agentA, taskId, { runId: "run-a" }),
+      storeB.checkoutTask(agentB, taskId, { runId: "run-b" }),
+    ]);
+
+    const fulfilled = [first, second].filter((entry): entry is PromiseFulfilledResult<Awaited<ReturnType<AgentStore["checkoutTask"]>>> => entry.status === "fulfilled");
+    const rejected = [first, second].filter((entry): entry is PromiseRejectedResult => entry.status === "rejected");
+
+    expect(fulfilled).toHaveLength(1);
+    expect(rejected).toHaveLength(1);
+    expect(rejected[0].reason).toBeInstanceOf(CheckoutConflictError);
+
+    const winner = fulfilled[0].value;
+    expect(rejected[0].reason.currentHolderId).toBe(winner.checkedOutBy);
+    expect(centralDb.getTaskClaim("P-1", taskId)?.leaseEpoch).toBe(1);
+    expect(winner.checkoutLeaseEpoch).toBe(1);
+    expect(["node-a", "node-b"]).toContain(winner.checkoutNodeId);
+
+    await (winner.checkedOutBy === agentA ? storeA : storeB).releaseTask(winner.checkedOutBy ?? "", taskId);
+
+    const [third, fourth] = await Promise.allSettled([
+      storeA.checkoutTask(agentA, taskId, { runId: "run-c" }),
+      storeB.checkoutTask(agentB, taskId, { runId: "run-d" }),
+    ]);
+
+    const fulfilled2 = [third, fourth].filter((entry): entry is PromiseFulfilledResult<Awaited<ReturnType<AgentStore["checkoutTask"]>>> => entry.status === "fulfilled");
+    const rejected2 = [third, fourth].filter((entry): entry is PromiseRejectedResult => entry.status === "rejected");
+
+    expect(fulfilled2).toHaveLength(1);
+    expect(rejected2).toHaveLength(1);
+    expect(rejected2[0].reason).toBeInstanceOf(CheckoutConflictError);
+    expect(centralDb.getTaskClaim("P-1", taskId)?.leaseEpoch).toBe(1);
+  });
+});