From 2d8ca452614c8f9bb1fe0f56c045d5bf76e8a170 Mon Sep 17 00:00:00 2001 From: GautamSharma99 Date: Fri, 31 Jul 2026 17:48:05 +0530 Subject: [PATCH] fix: preserve parallel connector runs --- .changeset/unique-connector-runs.md | 5 ++ src/connectors/io.ts | 38 ++++++++++-- src/connectors/mcp-runtime.ts | 13 ++-- test/connector-io.test.ts | 94 +++++++++++++++++++++++++++++ 4 files changed, 137 insertions(+), 13 deletions(-) create mode 100644 .changeset/unique-connector-runs.md create mode 100644 test/connector-io.test.ts diff --git a/.changeset/unique-connector-runs.md b/.changeset/unique-connector-runs.md new file mode 100644 index 000000000..75b3277c2 --- /dev/null +++ b/.changeset/unique-connector-runs.md @@ -0,0 +1,5 @@ +--- +"openwiki": patch +--- + +fix: preserve parallel connector runs and state updates diff --git a/src/connectors/io.ts b/src/connectors/io.ts index e02d13cc5..5fd0ebff7 100644 --- a/src/connectors/io.ts +++ b/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>(); + +/** Read, transform, and atomically commit connector state as one local transaction. */ +export async function updateConnectorState( + connectorId: ConnectorId, + updater: (state: ConnectorState) => ConnectorState | Promise, +): Promise { + const previous = stateUpdateQueues.get(connectorId) ?? Promise.resolve(); + let release!: () => void; + const current = new Promise((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 { diff --git a/src/connectors/mcp-runtime.ts b/src/connectors/mcp-runtime.ts index 1140baaa2..1b7a90208 100644 --- a/src/connectors/mcp-runtime.ts +++ b/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 { 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, ): Promise { 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,7 +171,6 @@ async function readMcpConnectorConfig( async function recordMcpRun( connectorId: McpConnectorId, - state: Awaited>, run: { rawFiles: string[]; runId: string; @@ -182,8 +178,7 @@ async function recordMcpRun( warnings: string[]; }, ): Promise { - await writeConnectorState( - connectorId, + await updateConnectorState(connectorId, (state) => updateStateWithRun(state, { at: new Date().toISOString(), rawFiles: run.rawFiles, diff --git a/test/connector-io.test.ts b/test/connector-io.test.ts new file mode 100644 index 000000000..614abaec0 --- /dev/null +++ b/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("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((resolve) => { + firstStarted = resolve; + }); + const firstReleased = new Promise((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"]); + }); +});