fix: preserve parallel connector runs and state

.changeset/unique-connector-runs.md

@@ -0,0 +1,5 @@
+---
+"openwiki": patch
+---
+
+fix: preserve parallel connector runs and state updates

src/connectors/io.ts

@@ -1,4 +1,5 @@
-import { chmod, readFile, writeFile } from "node:fs/promises";
+import { chmod, readFile, rename, writeFile } from "node:fs/promises";
+import { randomUUID } from "node:crypto";
 import path from "node:path";
 import {
   ensureConnectorHome,
@@ -56,6 +57,33 @@ export async function writeConnectorState(
   await writePrivateJson(getConnectorStatePath(connectorId), state);
 }
 
+const stateUpdateQueues = new Map<ConnectorId, Promise<void>>();
+
+/** Read, transform, and atomically commit connector state as one local transaction. */
+export async function updateConnectorState(
+  connectorId: ConnectorId,
+  updater: (state: ConnectorState) => ConnectorState | Promise<ConnectorState>,
+): Promise<ConnectorState> {
+  const previous = stateUpdateQueues.get(connectorId) ?? Promise.resolve();
+  let release!: () => void;
+  const current = new Promise<void>((resolve) => {
+    release = resolve;
+  });
+  stateUpdateQueues.set(connectorId, current);
+
+  await previous;
+  try {
+    const nextState = await updater(await readConnectorState(connectorId));
+    await writeConnectorState(connectorId, nextState);
+    return nextState;
+  } finally {
+    release();
+    if (stateUpdateQueues.get(connectorId) === current) {
+      stateUpdateQueues.delete(connectorId);
+    }
+  }
+}
+
 export async function writeRawJson(
   connectorId: ConnectorId,
   runId: string,
@@ -70,7 +98,7 @@ export async function writeRawJson(
 }
 
 export function createRunId(): string {
-  return new Date().toISOString().replace(/[:.]/gu, "-");
+  return `${new Date().toISOString().replace(/[:.]/gu, "-")}-${randomUUID()}`;
 }
 
 export function updateStateWithRun(
@@ -92,11 +120,13 @@ async function writePrivateJson(
   await import("node:fs/promises").then(({ mkdir }) =>
     mkdir(path.dirname(filePath), { recursive: true, mode: 0o700 }),
   );
-  await writeFile(filePath, `${JSON.stringify(value, null, 2)}\n`, {
+  const temporaryPath = `${filePath}.${process.pid}.${randomUUID()}.tmp`;
+  await writeFile(temporaryPath, `${JSON.stringify(value, null, 2)}\n`, {
     encoding: "utf8",
     mode: 0o600,
   });
-  await chmod(filePath, 0o600);
+  await chmod(temporaryPath, 0o600);
+  await rename(temporaryPath, filePath);
 }
 
 function isFileNotFoundError(error: unknown): boolean {

src/connectors/mcp-runtime.ts

@@ -1,9 +1,8 @@
 import {
   createRunId,
   readConnectorConfig,
-  readConnectorState,
+  updateConnectorState,
   updateStateWithRun,
-  writeConnectorState,
   writeRawJson,
 } from "./io.js";
 import {
@@ -48,7 +47,6 @@ export async function discoverMcpConnectorTools(
   connectorId: McpConnectorId,
 ): Promise<McpToolDiscoveryResult> {
   const runId = createRunId();
-  const state = await readConnectorState(connectorId);
   const config = await readMcpConnectorConfig(connectorId);
   const discovery = await listMcpTools(config);
   const rawFile = await writeRawJson(connectorId, runId, "mcp-tools.json", {
@@ -59,7 +57,7 @@ export async function discoverMcpConnectorTools(
     transport: sanitizeMcpTransport(config.transport),
   });
 
-  await recordMcpRun(connectorId, state, {
+  await recordMcpRun(connectorId, {
     rawFiles: [rawFile],
     runId,
     status: "success",
@@ -80,7 +78,6 @@ export async function callMcpConnectorTool(
   args: Record<string, unknown>,
 ): Promise<McpToolCallResult> {
   const runId = createRunId();
-  const state = await readConnectorState(connectorId);
   const config = await readMcpConnectorConfig(connectorId);
   const discovery = await listMcpTools(config);
   const tool = discovery.tools.find((candidate) => candidate.name === toolName);
@@ -112,7 +109,7 @@ export async function callMcpConnectorTool(
     },
   );
 
-  await recordMcpRun(connectorId, state, {
+  await recordMcpRun(connectorId, {
     rawFiles: [rawFile],
     runId,
     status: "success",
@@ -174,16 +171,14 @@ async function readMcpConnectorConfig(
 
 async function recordMcpRun(
   connectorId: McpConnectorId,
-  state: Awaited<ReturnType<typeof readConnectorState>>,
   run: {
     rawFiles: string[];
     runId: string;
     status: ConnectorIngestResult["status"];
     warnings: string[];
   },
 ): Promise<void> {
-  await writeConnectorState(
-    connectorId,
+  await updateConnectorState(connectorId, (state) =>
     updateStateWithRun(state, {
       at: new Date().toISOString(),
       rawFiles: run.rawFiles,

test/connector-io.test.ts

@@ -0,0 +1,94 @@
+import { mkdtemp, readFile, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+
+type ConnectorIo = typeof import("../src/connectors/io.ts");
+
+let tempHome: string;
+let io: ConnectorIo;
+
+beforeEach(async () => {
+  tempHome = await mkdtemp(path.join(tmpdir(), "openwiki-connector-io-"));
+  vi.resetModules();
+  vi.doMock("node:os", async () => {
+    const actual = await vi.importActual<typeof import("node:os")>("node:os");
+    return {
+      ...actual,
+      homedir: () => tempHome,
+      default: {
+        ...(actual.default as typeof import("node:os")),
+        homedir: () => tempHome,
+      },
+    };
+  });
+  io = await import("../src/connectors/io.ts");
+});
+
+afterEach(async () => {
+  vi.doUnmock("node:os");
+  vi.resetModules();
+  await rm(tempHome, { recursive: true, force: true });
+});
+
+describe("connector run identity", () => {
+  test("adds a collision-resistant suffix to timestamp IDs", () => {
+    const first = io.createRunId();
+    const second = io.createRunId();
+
+    expect(first).not.toBe(second);
+    expect(first).toMatch(/^\d{4}-\d{2}-\d{2}T.*-[0-9a-f-]{36}$/u);
+  });
+});
+
+describe("updateConnectorState", () => {
+  test("serializes concurrent read-transform-write transactions", async () => {
+    let releaseFirst!: () => void;
+    let firstStarted!: () => void;
+    const firstReady = new Promise<void>((resolve) => {
+      firstStarted = resolve;
+    });
+    const firstReleased = new Promise<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+    let updaterCalls = 0;
+
+    const first = io.updateConnectorState("notion", async (state) => {
+      updaterCalls += 1;
+      firstStarted();
+      await firstReleased;
+      return io.updateStateWithRun(state, {
+        at: "first",
+        rawFiles: ["first.json"],
+        runId: "first",
+        status: "success",
+        warnings: [],
+      });
+    });
+    await firstReady;
+
+    const second = io.updateConnectorState("notion", (state) => {
+      updaterCalls += 1;
+      return io.updateStateWithRun(state, {
+        at: "second",
+        rawFiles: ["second.json"],
+        runId: "second",
+        status: "success",
+        warnings: [],
+      });
+    });
+    await Promise.resolve();
+    expect(updaterCalls).toBe(1);
+
+    releaseFirst();
+    await Promise.all([first, second]);
+
+    const state = JSON.parse(
+      await readFile(
+        path.join(tempHome, ".openwiki/connectors/notion/state.json"),
+        "utf8",
+      ),
+    ) as { runs: Array<{ runId: string }> };
+    expect(state.runs.map((run) => run.runId)).toEqual(["second", "first"]);
+  });
+});