Fix adapter plugin conflicts caused by concurrent install/reload/delete operations

Add an async mutex (promise-chain FIFO queue) to serialise all adapter
mutation routes (install, reinstall, reload, delete) so that concurrent
requests cannot race on npm, the adapter-plugins.json store, or the
in-memory adapter registry.

Also switch adapter-plugin-store file writes to atomic write-tmp-then-rename
to prevent partial/corrupted reads from concurrent processes.

Includes the packageName.trim() fix for whitespace-induced npm failures.

9 new tests covering mutex serialisation, error recovery, FIFO ordering,
and atomic store operations.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
2026-04-12 15:54:09 +00:00
parent 9c7a17e16c
commit 92afa0fb67
3 changed files with 410 additions and 198 deletions
@@ -0,0 +1,177 @@
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { withAdapterMutex } from "../routes/adapters.js";
// ---------------------------------------------------------------------------
// withAdapterMutex — serialisation tests
// ---------------------------------------------------------------------------
describe("withAdapterMutex", () => {
it("serialises concurrent calls so they do not overlap", async () => {
const log: string[] = [];
const taskA = withAdapterMutex(async () => {
log.push("A-start");
await new Promise((r) => setTimeout(r, 50));
log.push("A-end");
return "a";
});
const taskB = withAdapterMutex(async () => {
log.push("B-start");
await new Promise((r) => setTimeout(r, 10));
log.push("B-end");
return "b";
});
const [resultA, resultB] = await Promise.all([taskA, taskB]);
expect(resultA).toBe("a");
expect(resultB).toBe("b");
// A must fully complete before B starts (FIFO serialisation)
expect(log).toEqual(["A-start", "A-end", "B-start", "B-end"]);
});
it("a failed task does not block subsequent tasks", async () => {
const failing = withAdapterMutex(async () => {
throw new Error("boom");
});
await expect(failing).rejects.toThrow("boom");
const ok = await withAdapterMutex(async () => "ok");
expect(ok).toBe("ok");
});
it("preserves FIFO order for many concurrent callers", async () => {
const order: number[] = [];
const tasks = Array.from({ length: 5 }, (_, i) =>
withAdapterMutex(async () => {
order.push(i);
}),
);
await Promise.all(tasks);
expect(order).toEqual([0, 1, 2, 3, 4]);
});
it("returns the value from the inner function", async () => {
const result = await withAdapterMutex(async () => ({ answer: 42 }));
expect(result).toEqual({ answer: 42 });
});
it("propagates errors from the inner function", async () => {
await expect(
withAdapterMutex(async () => {
throw new TypeError("bad input");
}),
).rejects.toThrow("bad input");
});
});
// ---------------------------------------------------------------------------
// adapter-plugin-store — atomic write tests
// ---------------------------------------------------------------------------
describe("adapter-plugin-store atomic writes", () => {
// We test the store in an isolated temp directory by stubbing HOME
const originalHome = process.env.HOME;
let tmpDir: string;
beforeEach(() => {
tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "adapter-store-test-"));
process.env.HOME = tmpDir;
// Force the store module to pick up the new HOME by resetting its cache.
// We re-import it fresh for each test.
});
afterEach(() => {
process.env.HOME = originalHome;
fs.rmSync(tmpDir, { recursive: true, force: true });
});
it("writeStore creates the file atomically via rename", async () => {
// Dynamically import the store so it uses the stubbed HOME
vi.resetModules();
const store = await import("../services/adapter-plugin-store.js");
const record = {
packageName: "test-adapter",
type: "test_type",
installedAt: new Date().toISOString(),
};
store.addAdapterPlugin(record);
// The store file should exist and contain the record
const storePath = path.join(tmpDir, ".paperclip", "adapter-plugins.json");
expect(fs.existsSync(storePath)).toBe(true);
const contents = JSON.parse(fs.readFileSync(storePath, "utf-8"));
expect(contents).toHaveLength(1);
expect(contents[0].packageName).toBe("test-adapter");
// No lingering temp files
const dir = path.dirname(storePath);
const tmpFiles = fs.readdirSync(dir).filter((f) => f.endsWith(".tmp"));
expect(tmpFiles).toHaveLength(0);
});
it("addAdapterPlugin is idempotent for the same type", async () => {
vi.resetModules();
const store = await import("../services/adapter-plugin-store.js");
const record1 = {
packageName: "pkg-a",
type: "my_adapter",
version: "1.0.0",
installedAt: new Date().toISOString(),
};
const record2 = {
packageName: "pkg-a",
type: "my_adapter",
version: "2.0.0",
installedAt: new Date().toISOString(),
};
store.addAdapterPlugin(record1);
store.addAdapterPlugin(record2);
const plugins = store.listAdapterPlugins();
expect(plugins).toHaveLength(1);
expect(plugins[0].version).toBe("2.0.0");
});
it("removeAdapterPlugin removes the correct record", async () => {
vi.resetModules();
const store = await import("../services/adapter-plugin-store.js");
store.addAdapterPlugin({
packageName: "a",
type: "type_a",
installedAt: new Date().toISOString(),
});
store.addAdapterPlugin({
packageName: "b",
type: "type_b",
installedAt: new Date().toISOString(),
});
expect(store.listAdapterPlugins()).toHaveLength(2);
const removed = store.removeAdapterPlugin("type_a");
expect(removed).toBe(true);
expect(store.listAdapterPlugins()).toHaveLength(1);
expect(store.listAdapterPlugins()[0].type).toBe("type_b");
});
it("removeAdapterPlugin returns false for unknown type", async () => {
vi.resetModules();
const store = await import("../services/adapter-plugin-store.js");
expect(store.removeAdapterPlugin("nonexistent")).toBe(false);
});
});
+30 -2
View File
@@ -46,6 +46,26 @@ import { BUILTIN_ADAPTER_TYPES } from "../adapters/builtin-adapter-types.js";
const execFileAsync = promisify(execFile); const execFileAsync = promisify(execFile);
// ---------------------------------------------------------------------------
// Concurrency control — serialise all install / reinstall / reload / delete
// operations so that concurrent requests cannot race on npm, the plugin store
// JSON file, or the in-memory adapter registry.
// ---------------------------------------------------------------------------
let mutexQueue: Promise<void> = Promise.resolve();
/**
* Enqueue `fn` behind any in-flight adapter mutation. Only one mutation runs
* at a time; the rest wait in FIFO order.
*/
export function withAdapterMutex<T>(fn: () => Promise<T>): Promise<T> {
const ticket = mutexQueue.then(fn, fn); // always run `fn` after the queue settles
// Swallow rejections on the queue itself so a failed operation doesn't
// permanently block subsequent ones.
mutexQueue = ticket.then(() => {}, () => {});
return ticket;
}
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Request / Response types // Request / Response types
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -211,7 +231,7 @@ export function adapterRoutes() {
// Strip version suffix if the UI sends "pkg@1.2.3" instead of separating it // Strip version suffix if the UI sends "pkg@1.2.3" instead of separating it
// e.g. "@henkey/hermes-paperclip-adapter@0.3.0" → packageName + version // e.g. "@henkey/hermes-paperclip-adapter@0.3.0" → packageName + version
let canonicalName = packageName; let canonicalName = packageName.trim();
let explicitVersion = version; let explicitVersion = version;
const versionSuffix = packageName.match(/@(\d+\.\d+\.\d+.*)$/); const versionSuffix = packageName.match(/@(\d+\.\d+\.\d+.*)$/);
if (versionSuffix) { if (versionSuffix) {
@@ -224,6 +244,7 @@ export function adapterRoutes() {
} }
} }
await withAdapterMutex(async () => {
try { try {
let installedVersion: string | undefined; let installedVersion: string | undefined;
let moduleLocalPath: string | undefined; let moduleLocalPath: string | undefined;
@@ -322,6 +343,7 @@ export function adapterRoutes() {
} }
} }
}); });
});
/** /**
* PATCH /api/adapters/:type * PATCH /api/adapters/:type
@@ -412,6 +434,7 @@ export function adapterRoutes() {
return; return;
} }
await withAdapterMutex(async () => {
// Check that the adapter exists in the registry // Check that the adapter exists in the registry
const existing = findServerAdapter(adapterType); const existing = findServerAdapter(adapterType);
if (!existing) { if (!existing) {
@@ -460,6 +483,7 @@ export function adapterRoutes() {
res.json({ type: adapterType, removed: true }); res.json({ type: adapterType, removed: true });
}); });
});
/** /**
* POST /api/adapters/:type/reload * POST /api/adapters/:type/reload
@@ -480,6 +504,7 @@ export function adapterRoutes() {
return; return;
} }
await withAdapterMutex(async () => {
// Reload the adapter module (busts ESM cache, re-imports) // Reload the adapter module (busts ESM cache, re-imports)
try { try {
const newModule = await reloadExternalAdapter(type); const newModule = await reloadExternalAdapter(type);
@@ -514,6 +539,7 @@ export function adapterRoutes() {
res.status(500).json({ error: `Failed to reload adapter: ${message}` }); res.status(500).json({ error: `Failed to reload adapter: ${message}` });
} }
}); });
});
// ── POST /api/adapters/:type/reinstall ────────────────────────────────── // ── POST /api/adapters/:type/reinstall ──────────────────────────────────
// Reinstall an npm-sourced external adapter (pulls latest from registry). // Reinstall an npm-sourced external adapter (pulls latest from registry).
@@ -542,12 +568,13 @@ export function adapterRoutes() {
return; return;
} }
await withAdapterMutex(async () => {
try { try {
const pluginsDir = getAdapterPluginsDir(); const pluginsDir = getAdapterPluginsDir();
logger.info({ type, packageName: record.packageName }, "Reinstalling adapter package via npm"); logger.info({ type, packageName: record.packageName }, "Reinstalling adapter package via npm");
await execFileAsync("npm", ["install", "--no-save", record.packageName], { await execFileAsync("npm", ["install", "--no-save", record.packageName.trim()], {
cwd: pluginsDir, cwd: pluginsDir,
timeout: 120_000, timeout: 120_000,
}); });
@@ -582,6 +609,7 @@ export function adapterRoutes() {
res.status(500).json({ error: `Reinstall failed: ${message}` }); res.status(500).json({ error: `Reinstall failed: ${message}` });
} }
}); });
});
// ── GET /api/adapters/:type/config-schema ──────────────────────────────── // ── GET /api/adapters/:type/config-schema ────────────────────────────────
// Serve a declarative config schema for an adapter's UI form fields. // Serve a declarative config schema for an adapter's UI form fields.
+9 -2
View File
@@ -86,7 +86,12 @@ function readStore(): AdapterPluginRecord[] {
function writeStore(records: AdapterPluginRecord[]): void { function writeStore(records: AdapterPluginRecord[]): void {
ensureDirs(); ensureDirs();
fs.writeFileSync(ADAPTER_PLUGINS_STORE_PATH, JSON.stringify(records, null, 2), "utf-8"); // Atomic write: write to a temp file in the same directory then rename.
// rename() is atomic on POSIX when source and target are on the same
// filesystem, preventing partial/corrupted reads from concurrent processes.
const tmpPath = `${ADAPTER_PLUGINS_STORE_PATH}.${process.pid}.tmp`;
fs.writeFileSync(tmpPath, JSON.stringify(records, null, 2), "utf-8");
fs.renameSync(tmpPath, ADAPTER_PLUGINS_STORE_PATH);
storeCache = records; storeCache = records;
} }
@@ -106,7 +111,9 @@ function readSettings(): AdapterSettings {
function writeSettings(settings: AdapterSettings): void { function writeSettings(settings: AdapterSettings): void {
ensureDirs(); ensureDirs();
fs.writeFileSync(ADAPTER_SETTINGS_PATH, JSON.stringify(settings, null, 2), "utf-8"); const tmpPath = `${ADAPTER_SETTINGS_PATH}.${process.pid}.tmp`;
fs.writeFileSync(tmpPath, JSON.stringify(settings, null, 2), "utf-8");
fs.renameSync(tmpPath, ADAPTER_SETTINGS_PATH);
settingsCache = settings; settingsCache = settings;
} }