mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-24 11:25:50 -06:00
fix(plugins): preserve install index state across failures (#119228)
* fix(plugins): preserve install index policy config * fix(plugins): restore complete install index state * fix(plugins): fence install index rollback * fix(plugins): fence generic install index rollback * fix(plugins): keep lifecycle lease context private * test(cli): align plugin index rollback mocks * test(plugins): enforce index rollback receipts * fix(plugins): serialize install rollback with config commit * test(plugins): keep legacy index writer mock private
This commit is contained in:
Vincent Koc
@@ -17,7 +17,7 @@ import {
|
||||
runtimeLogs,
|
||||
setInstalledPluginIndexInstallRecords,
|
||||
writeConfigFile,
|
||||
writePersistedInstalledPluginIndexInstallRecords,
|
||||
writePersistedInstalledPluginIndexInstallRecordsWithLease,
|
||||
applyPluginUninstallDirectoryRemoval,
|
||||
} from "../cli/plugins-cli-test-helpers.js";
|
||||
import type { OpenClawConfig } from "../config/config.js";
|
||||
@@ -109,8 +109,8 @@ describe("persistPluginInstall", () => {
|
||||
|
||||
expect(next).toEqual(enabledConfig);
|
||||
const persistedRecords = requireMockCallArg(
|
||||
writePersistedInstalledPluginIndexInstallRecords,
|
||||
"writePersistedInstalledPluginIndexInstallRecords",
|
||||
writePersistedInstalledPluginIndexInstallRecordsWithLease,
|
||||
"writePersistedInstalledPluginIndexInstallRecordsWithLease",
|
||||
);
|
||||
expect(persistedRecords.alpha).toEqual({
|
||||
source: "npm",
|
||||
@@ -881,8 +881,8 @@ describe("persistPluginInstall", () => {
|
||||
'Installed plugin "needs-config" without enabling it because it requires configuration first.',
|
||||
);
|
||||
const persistedRecords = requireMockCallArg(
|
||||
writePersistedInstalledPluginIndexInstallRecords,
|
||||
"writePersistedInstalledPluginIndexInstallRecords",
|
||||
writePersistedInstalledPluginIndexInstallRecordsWithLease,
|
||||
"writePersistedInstalledPluginIndexInstallRecordsWithLease",
|
||||
);
|
||||
expect(persistedRecords["needs-config"]).toMatchObject({
|
||||
source: "npm",
|
||||
@@ -937,7 +937,7 @@ describe("persistPluginInstall", () => {
|
||||
).rejects.toThrow("has invalid configured settings");
|
||||
|
||||
expect(enablePluginInConfig).not.toHaveBeenCalled();
|
||||
expect(writePersistedInstalledPluginIndexInstallRecords).not.toHaveBeenCalled();
|
||||
expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).not.toHaveBeenCalled();
|
||||
expect(writeConfigFile).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
@@ -969,8 +969,8 @@ describe("persistPluginInstall", () => {
|
||||
expect(enablePluginInConfig).not.toHaveBeenCalled();
|
||||
expect(applyExclusiveSlotSelection).not.toHaveBeenCalled();
|
||||
const persistedRecords = requireMockCallArg(
|
||||
writePersistedInstalledPluginIndexInstallRecords,
|
||||
"writePersistedInstalledPluginIndexInstallRecords",
|
||||
writePersistedInstalledPluginIndexInstallRecordsWithLease,
|
||||
"writePersistedInstalledPluginIndexInstallRecordsWithLease",
|
||||
);
|
||||
expect(persistedRecords["memory-lancedb"]).toEqual({
|
||||
source: "path",
|
||||
|
||||
@@ -0,0 +1,192 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js";
|
||||
import { withEnvAsync } from "../test-utils/env.js";
|
||||
import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js";
|
||||
import { writePersistedInstalledPluginIndexInstallRecordsWithLease } from "./installed-plugin-index-records.js";
|
||||
import { readPersistedInstalledPluginIndex } from "./installed-plugin-index-store.js";
|
||||
import { resolveInstalledPluginIndexPolicyHash } from "./installed-plugin-index.js";
|
||||
import { withPluginLifecycleLease } from "./plugin-lifecycle-lease.js";
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
});
|
||||
|
||||
function runChild(scriptPath: string, args: string[]): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawn(process.execPath, ["--import", "tsx", scriptPath, ...args], {
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
let output = "";
|
||||
child.stdout.on("data", (chunk) => (output += chunk));
|
||||
child.stderr.on("data", (chunk) => (output += chunk));
|
||||
child.on("close", (code) => {
|
||||
if (code === 0) {
|
||||
resolve();
|
||||
} else {
|
||||
reject(new Error(`install-record commit child exited ${code}: ${output}`));
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function fileExists(filePath: string): Promise<boolean> {
|
||||
try {
|
||||
await fs.promises.access(filePath);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
|
||||
return false;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForFile(filePath: string, timeoutMs = 15_000): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (await fileExists(filePath)) {
|
||||
return;
|
||||
}
|
||||
await delay(10);
|
||||
}
|
||||
throw new Error(`timed out waiting for ${filePath}`);
|
||||
}
|
||||
|
||||
async function expectFileToStayAbsent(filePath: string, durationMs = 500): Promise<void> {
|
||||
const deadline = Date.now() + durationMs;
|
||||
while (Date.now() < deadline) {
|
||||
expect(await fileExists(filePath)).toBe(false);
|
||||
await delay(10);
|
||||
}
|
||||
}
|
||||
|
||||
describe("plugin install record commit rollback", () => {
|
||||
it("serializes two failing direct config commits and restores the original index", async () => {
|
||||
await withOpenClawTestState({ label: "plugin-record-failing-commits" }, async (state) => {
|
||||
const commitModuleUrl = pathToFileURL(
|
||||
path.resolve("src/plugins/install-record-commit.ts"),
|
||||
).href;
|
||||
const childScript = await state.writeText(
|
||||
"fail-config-commit.mts",
|
||||
`
|
||||
import fs from "node:fs";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { commitConfigWriteWithPendingPluginInstalls } from ${JSON.stringify(commitModuleUrl)};
|
||||
const [stateDir, pluginId, startedPath, enteredPath, releasePath] = process.argv.slice(2);
|
||||
process.env.OPENCLAW_STATE_DIR = stateDir;
|
||||
await fs.promises.writeFile(startedPath, "started");
|
||||
try {
|
||||
await commitConfigWriteWithPendingPluginInstalls({
|
||||
nextConfig: {
|
||||
plugins: {
|
||||
installs: {
|
||||
[pluginId]: {
|
||||
source: "path",
|
||||
spec: pluginId,
|
||||
sourcePath: "/tmp/" + pluginId,
|
||||
installPath: "/tmp/" + pluginId,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
commit: async () => {
|
||||
await fs.promises.writeFile(enteredPath, "entered");
|
||||
while (true) {
|
||||
try {
|
||||
await fs.promises.access(releasePath);
|
||||
break;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
await delay(10);
|
||||
}
|
||||
throw new Error("config failed " + pluginId);
|
||||
},
|
||||
});
|
||||
throw new Error("config commit unexpectedly succeeded");
|
||||
} catch (error) {
|
||||
if (!(error instanceof Error) || error.message !== "config failed " + pluginId) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
`,
|
||||
);
|
||||
const firstStarted = path.join(state.stateDir, "first-started");
|
||||
const firstEntered = path.join(state.stateDir, "first-entered");
|
||||
const firstRelease = path.join(state.stateDir, "first-release");
|
||||
const secondStarted = path.join(state.stateDir, "second-started");
|
||||
const secondEntered = path.join(state.stateDir, "second-entered");
|
||||
const secondRelease = path.join(state.stateDir, "second-release");
|
||||
|
||||
await withEnvAsync(state.env, async () => {
|
||||
await withPluginLifecycleLease({}, async (lease) => {
|
||||
await writePersistedInstalledPluginIndexInstallRecordsWithLease(
|
||||
{
|
||||
original: {
|
||||
source: "path",
|
||||
spec: "original",
|
||||
sourcePath: "/tmp/original",
|
||||
installPath: "/tmp/original",
|
||||
},
|
||||
},
|
||||
{ config: {}, lease },
|
||||
);
|
||||
});
|
||||
|
||||
const firstDone = runChild(childScript, [
|
||||
state.stateDir,
|
||||
"first",
|
||||
firstStarted,
|
||||
firstEntered,
|
||||
firstRelease,
|
||||
]);
|
||||
let secondDone: Promise<void> | undefined;
|
||||
try {
|
||||
await waitForFile(firstEntered);
|
||||
secondDone = runChild(childScript, [
|
||||
state.stateDir,
|
||||
"second",
|
||||
secondStarted,
|
||||
secondEntered,
|
||||
secondRelease,
|
||||
]);
|
||||
await waitForFile(secondStarted);
|
||||
|
||||
// The second writer must stay outside its config commit until the
|
||||
// first writer rolls its tentative index state back.
|
||||
await expectFileToStayAbsent(secondEntered);
|
||||
|
||||
await fs.promises.writeFile(firstRelease, "release");
|
||||
await firstDone;
|
||||
await waitForFile(secondEntered);
|
||||
await fs.promises.writeFile(secondRelease, "release");
|
||||
await secondDone;
|
||||
} finally {
|
||||
await Promise.all([
|
||||
fs.promises.writeFile(firstRelease, "release"),
|
||||
fs.promises.writeFile(secondRelease, "release"),
|
||||
]);
|
||||
await Promise.allSettled([firstDone, ...(secondDone ? [secondDone] : [])]);
|
||||
}
|
||||
});
|
||||
|
||||
const persisted = await readPersistedInstalledPluginIndex({ env: state.env });
|
||||
expect(persisted?.installRecords).toEqual({
|
||||
original: {
|
||||
source: "path",
|
||||
spec: "original",
|
||||
sourcePath: "/tmp/original",
|
||||
installPath: "/tmp/original",
|
||||
},
|
||||
});
|
||||
expect(persisted?.policyHash).toBe(resolveInstalledPluginIndexPolicyHash({}));
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -6,17 +6,38 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import type { PluginInstallRecord } from "../config/types.plugins.js";
|
||||
import { withEnvAsync } from "../test-utils/env.js";
|
||||
import type { InstalledPluginIndex } from "./installed-plugin-index.js";
|
||||
import {
|
||||
hasRetainedManagedNpmInstallMarker,
|
||||
markRetainedManagedNpmInstall,
|
||||
} from "./managed-npm-retention.js";
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
loadInstalledPluginIndexInstallRecords: vi.fn(),
|
||||
replaceConfigFile: vi.fn(),
|
||||
transformConfigFileWithRetry: vi.fn(),
|
||||
writePersistedInstalledPluginIndexInstallRecords: vi.fn(),
|
||||
}));
|
||||
const mocks = vi.hoisted(() => {
|
||||
const lease = {
|
||||
databasePath: "/tmp/openclaw-plugin-index.sqlite",
|
||||
signal: new AbortController().signal,
|
||||
assertOwned: vi.fn(),
|
||||
assertOwnedInTransaction: vi.fn(),
|
||||
};
|
||||
return {
|
||||
lease,
|
||||
loadInstalledPluginIndexInstallRecords: vi.fn(),
|
||||
replaceConfigFile: vi.fn(),
|
||||
restorePersistedInstalledPluginIndexIfCurrent:
|
||||
vi.fn<
|
||||
typeof import("./installed-plugin-index-store.js").restorePersistedInstalledPluginIndexIfCurrent
|
||||
>(),
|
||||
transformConfigFileWithRetry: vi.fn(),
|
||||
withPluginLifecycleLease: vi.fn(
|
||||
async (_options: unknown, run: (activeLease: typeof lease) => Promise<unknown>) =>
|
||||
await run(lease),
|
||||
),
|
||||
writePersistedInstalledPluginIndexInstallRecordsWithLease:
|
||||
vi.fn<
|
||||
typeof import("./installed-plugin-index-records.js").writePersistedInstalledPluginIndexInstallRecordsWithLease
|
||||
>(),
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("../config/config.js", () => ({
|
||||
replaceConfigFile: mocks.replaceConfigFile,
|
||||
@@ -29,20 +50,52 @@ vi.mock("./installed-plugin-index-records.js", async (importOriginal) => {
|
||||
return {
|
||||
...actual,
|
||||
loadInstalledPluginIndexInstallRecords: mocks.loadInstalledPluginIndexInstallRecords,
|
||||
writePersistedInstalledPluginIndexInstallRecords:
|
||||
mocks.writePersistedInstalledPluginIndexInstallRecords,
|
||||
writePersistedInstalledPluginIndexInstallRecordsWithLease:
|
||||
mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease,
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("./installed-plugin-index-store.js", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("./installed-plugin-index-store.js")>();
|
||||
return {
|
||||
...actual,
|
||||
restorePersistedInstalledPluginIndexIfCurrent:
|
||||
mocks.restorePersistedInstalledPluginIndexIfCurrent,
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("./plugin-lifecycle-lease.js", () => ({
|
||||
withPluginLifecycleLease: mocks.withPluginLifecycleLease,
|
||||
}));
|
||||
|
||||
import {
|
||||
commitConfigWithPendingPluginInstalls,
|
||||
commitConfigWriteWithPendingPluginInstalls,
|
||||
commitPluginInstallRecordsOnly,
|
||||
commitPluginInstallRecordsWithConfig,
|
||||
stripPendingPluginInstallRecords,
|
||||
transformConfigWithPendingPluginInstalls,
|
||||
unchangedPendingPluginInstallRecordIds,
|
||||
} from "./install-record-commit.js";
|
||||
|
||||
function createTestInstalledPluginIndex(params: {
|
||||
policyHash: string;
|
||||
installRecords: Record<string, PluginInstallRecord>;
|
||||
}): InstalledPluginIndex {
|
||||
return {
|
||||
version: 1,
|
||||
hostContractVersion: "test",
|
||||
compatRegistryVersion: "test",
|
||||
migrationVersion: 1,
|
||||
policyHash: params.policyHash,
|
||||
generatedAtMs: 0,
|
||||
refreshReason: "source-changed",
|
||||
installRecords: structuredClone(params.installRecords),
|
||||
plugins: [],
|
||||
diagnostics: [],
|
||||
};
|
||||
}
|
||||
|
||||
describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
@@ -56,7 +109,11 @@ describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
afterWrite: { mode: "auto" },
|
||||
followUp: { mode: "auto", requiresRestart: false },
|
||||
}));
|
||||
mocks.writePersistedInstalledPluginIndexInstallRecords.mockResolvedValue(undefined);
|
||||
mocks.restorePersistedInstalledPluginIndexIfCurrent.mockResolvedValue(true);
|
||||
mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease.mockResolvedValue({
|
||||
previous: null,
|
||||
revision: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it("moves pending plugin install records into the plugin index before writing stripped config", async () => {
|
||||
@@ -87,10 +144,23 @@ describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
baseHash: "config-1",
|
||||
});
|
||||
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith({
|
||||
...existingRecords,
|
||||
...pendingRecords,
|
||||
});
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledWith(
|
||||
{
|
||||
...existingRecords,
|
||||
...pendingRecords,
|
||||
},
|
||||
{
|
||||
config: {
|
||||
plugins: {
|
||||
entries: {
|
||||
demo: { enabled: true },
|
||||
},
|
||||
},
|
||||
},
|
||||
filePath: mocks.lease.databasePath,
|
||||
lease: mocks.lease,
|
||||
},
|
||||
);
|
||||
expect(mocks.replaceConfigFile).toHaveBeenCalledWith({
|
||||
nextConfig: {
|
||||
plugins: {
|
||||
@@ -122,6 +192,40 @@ describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("uses the effective config for records-only index commits", async () => {
|
||||
const nextConfig: OpenClawConfig = {
|
||||
plugins: {
|
||||
entries: {
|
||||
demo: { enabled: false },
|
||||
},
|
||||
},
|
||||
};
|
||||
const nextInstallRecords: Record<string, PluginInstallRecord> = {
|
||||
demo: {
|
||||
source: "npm",
|
||||
spec: "demo@2.0.0",
|
||||
},
|
||||
};
|
||||
const verifyConfigFresh = vi.fn(async () => undefined);
|
||||
|
||||
await commitPluginInstallRecordsOnly({
|
||||
nextConfig,
|
||||
nextInstallRecords,
|
||||
verifyConfigFresh,
|
||||
});
|
||||
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledWith(
|
||||
nextInstallRecords,
|
||||
{
|
||||
config: nextConfig,
|
||||
filePath: mocks.lease.databasePath,
|
||||
lease: mocks.lease,
|
||||
},
|
||||
);
|
||||
expect(verifyConfigFresh).toHaveBeenCalledOnce();
|
||||
expect(mocks.replaceConfigFile).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("migrates source records below the canonical index and explicit pending records", async () => {
|
||||
const sourceConfig: OpenClawConfig = {
|
||||
plugins: {
|
||||
@@ -154,12 +258,19 @@ describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
commit,
|
||||
});
|
||||
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith({
|
||||
stale: existingRecords.stale,
|
||||
missing: sourceConfig.plugins?.installs?.missing,
|
||||
codex: nextConfig.plugins?.installs?.codex,
|
||||
concurrent: nextConfig.plugins?.installs?.concurrent,
|
||||
});
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledWith(
|
||||
{
|
||||
stale: existingRecords.stale,
|
||||
missing: sourceConfig.plugins?.installs?.missing,
|
||||
codex: nextConfig.plugins?.installs?.codex,
|
||||
concurrent: nextConfig.plugins?.installs?.concurrent,
|
||||
},
|
||||
{
|
||||
config: {},
|
||||
filePath: mocks.lease.databasePath,
|
||||
lease: mocks.lease,
|
||||
},
|
||||
);
|
||||
expect(commit).toHaveBeenCalledWith(
|
||||
{},
|
||||
{
|
||||
@@ -204,10 +315,17 @@ describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
}),
|
||||
});
|
||||
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith({
|
||||
other: sourceConfig.plugins?.installs?.other,
|
||||
codex: codexRecord,
|
||||
});
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledWith(
|
||||
{
|
||||
other: sourceConfig.plugins?.installs?.other,
|
||||
codex: codexRecord,
|
||||
},
|
||||
{
|
||||
config: {},
|
||||
filePath: mocks.lease.databasePath,
|
||||
lease: mocks.lease,
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it("strips only selected pending plugin install records", () => {
|
||||
@@ -660,7 +778,15 @@ describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
spec: "existing@1.0.0",
|
||||
},
|
||||
};
|
||||
const previousPersistedIndex = createTestInstalledPluginIndex({
|
||||
policyHash: "previous-policy",
|
||||
installRecords: existingRecords,
|
||||
});
|
||||
mocks.loadInstalledPluginIndexInstallRecords.mockResolvedValue(existingRecords);
|
||||
mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease.mockResolvedValue({
|
||||
previous: previousPersistedIndex,
|
||||
revision: 17,
|
||||
});
|
||||
mocks.replaceConfigFile.mockRejectedValue(new Error("config changed"));
|
||||
|
||||
await expect(
|
||||
@@ -678,20 +804,73 @@ describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
}),
|
||||
).rejects.toThrow("config changed");
|
||||
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith(1, {
|
||||
existing: {
|
||||
source: "npm",
|
||||
spec: "existing@1.0.0",
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledWith(
|
||||
{
|
||||
existing: {
|
||||
source: "npm",
|
||||
spec: "existing@1.0.0",
|
||||
},
|
||||
demo: {
|
||||
source: "npm",
|
||||
spec: "demo@1.0.0",
|
||||
},
|
||||
},
|
||||
demo: {
|
||||
source: "npm",
|
||||
spec: "demo@1.0.0",
|
||||
{
|
||||
config: {},
|
||||
filePath: mocks.lease.databasePath,
|
||||
lease: mocks.lease,
|
||||
},
|
||||
});
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
existingRecords,
|
||||
);
|
||||
expect(mocks.restorePersistedInstalledPluginIndexIfCurrent).toHaveBeenCalledWith(
|
||||
previousPersistedIndex,
|
||||
17,
|
||||
{
|
||||
filePath: mocks.lease.databasePath,
|
||||
lease: mocks.lease,
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it("leaves marker state intact when a successor owns the plugin index", async () => {
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-record-commit-"));
|
||||
const installPath = path.join(
|
||||
stateDir,
|
||||
"npm",
|
||||
"projects",
|
||||
"codex-v2",
|
||||
"node_modules",
|
||||
"@openclaw",
|
||||
"codex",
|
||||
);
|
||||
fs.mkdirSync(installPath, { recursive: true });
|
||||
await markRetainedManagedNpmInstall({
|
||||
packageDir: installPath,
|
||||
pluginId: "codex",
|
||||
retainedAt: "2026-04-25T00:00:00.000Z",
|
||||
reason: "test-successor-owned-marker",
|
||||
});
|
||||
mocks.restorePersistedInstalledPluginIndexIfCurrent.mockResolvedValueOnce(false);
|
||||
mocks.replaceConfigFile.mockRejectedValueOnce(new Error("config changed"));
|
||||
|
||||
try {
|
||||
await expect(
|
||||
commitPluginInstallRecordsWithConfig({
|
||||
previousInstallRecords: {},
|
||||
nextInstallRecords: {
|
||||
codex: {
|
||||
source: "npm",
|
||||
spec: "@openclaw/codex@2.0.0",
|
||||
installPath,
|
||||
},
|
||||
},
|
||||
nextConfig: {},
|
||||
}),
|
||||
).rejects.toThrow("config changed");
|
||||
|
||||
expect(hasRetainedManagedNpmInstallMarker(installPath)).toBe(false);
|
||||
} finally {
|
||||
fs.rmSync(stateDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("uses a plain config write when no pending plugin install records exist", async () => {
|
||||
@@ -704,7 +883,7 @@ describe("commitConfigWithPendingPluginInstalls", () => {
|
||||
const result = await commitConfigWithPendingPluginInstalls({ nextConfig });
|
||||
|
||||
expect(mocks.loadInstalledPluginIndexInstallRecords).not.toHaveBeenCalled();
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecords).not.toHaveBeenCalled();
|
||||
expect(mocks.writePersistedInstalledPluginIndexInstallRecordsWithLease).not.toHaveBeenCalled();
|
||||
expect(mocks.replaceConfigFile).toHaveBeenCalledWith({
|
||||
nextConfig,
|
||||
});
|
||||
|
||||
@@ -18,15 +18,21 @@ import { isPathInside } from "../infra/path-guards.js";
|
||||
import {
|
||||
loadInstalledPluginIndexInstallRecords,
|
||||
PLUGIN_INSTALLS_CONFIG_PATH,
|
||||
type InstalledPluginIndexRecordStoreOptions,
|
||||
withoutPluginInstallRecords,
|
||||
writePersistedInstalledPluginIndexInstallRecords,
|
||||
writePersistedInstalledPluginIndexInstallRecordsWithLease,
|
||||
} from "./installed-plugin-index-records.js";
|
||||
import {
|
||||
restorePersistedInstalledPluginIndexIfCurrent,
|
||||
type InstalledPluginIndexWriteReceipt,
|
||||
} from "./installed-plugin-index-store.js";
|
||||
import {
|
||||
clearRetainedManagedNpmInstallMarker,
|
||||
markRetainedManagedNpmInstall,
|
||||
resolveRetainedManagedNpmInstallPackageInfo,
|
||||
resolveRetainedManagedNpmInstallMarkerPath,
|
||||
} from "./managed-npm-retention.js";
|
||||
import { withPluginLifecycleLease } from "./plugin-lifecycle-lease.js";
|
||||
import { planPluginUninstall } from "./uninstall.js";
|
||||
|
||||
function mergeUnsetPaths(
|
||||
@@ -272,33 +278,46 @@ async function restoreClearedRetainedManagedNpmInstallMarkers(
|
||||
}
|
||||
|
||||
async function commitPluginInstallRecordsWithWriter(params: {
|
||||
previousInstallRecords?: Record<string, PluginInstallRecord>;
|
||||
nextInstallRecords: Record<string, PluginInstallRecord>;
|
||||
prepareInstallRecords: (storeOptions: InstalledPluginIndexRecordStoreOptions) => Promise<{
|
||||
previousInstallRecords: Record<string, PluginInstallRecord>;
|
||||
nextInstallRecords: Record<string, PluginInstallRecord>;
|
||||
}>;
|
||||
nextConfig: OpenClawConfig;
|
||||
writeOptions?: ConfigWriteOptions;
|
||||
commit: ConfigCommit;
|
||||
}): Promise<ConfigReplaceResult | void> {
|
||||
const previousInstallRecords =
|
||||
params.previousInstallRecords ?? (await loadInstalledPluginIndexInstallRecords());
|
||||
const retainedMarkerPaths: string[] = [];
|
||||
const clearedMarkerSnapshots: Array<{ markerPath: string; contents: string }> = [];
|
||||
try {
|
||||
await writePersistedInstalledPluginIndexInstallRecords(params.nextInstallRecords);
|
||||
}): Promise<{
|
||||
committed: ConfigReplaceResult | void;
|
||||
nextInstallRecords: Record<string, PluginInstallRecord>;
|
||||
}> {
|
||||
return await withPluginLifecycleLease({}, async (lease) => {
|
||||
let tentativeWrite: InstalledPluginIndexWriteReceipt | undefined;
|
||||
const retainedMarkerPaths: string[] = [];
|
||||
const clearedMarkerSnapshots: Array<{ markerPath: string; contents: string }> = [];
|
||||
try {
|
||||
const storeOptions = { filePath: lease.databasePath };
|
||||
const prepared = await params.prepareInstallRecords(storeOptions);
|
||||
tentativeWrite = await writePersistedInstalledPluginIndexInstallRecordsWithLease(
|
||||
prepared.nextInstallRecords,
|
||||
{
|
||||
...storeOptions,
|
||||
config: params.nextConfig,
|
||||
lease,
|
||||
},
|
||||
);
|
||||
await markRetainedReplacedManagedNpmInstallRecords({
|
||||
previousInstallRecords,
|
||||
nextInstallRecords: params.nextInstallRecords,
|
||||
// Keep partial progress visible to the outer rollback path.
|
||||
previousInstallRecords: prepared.previousInstallRecords,
|
||||
nextInstallRecords: prepared.nextInstallRecords,
|
||||
// Keep partial progress visible to the rollback path.
|
||||
createdMarkerPaths: retainedMarkerPaths,
|
||||
});
|
||||
clearedMarkerSnapshots.push(
|
||||
...(await clearActiveRetainedManagedNpmInstallMarkers(params.nextInstallRecords)),
|
||||
...(await clearActiveRetainedManagedNpmInstallMarkers(prepared.nextInstallRecords)),
|
||||
);
|
||||
const installRecordsChanged = !isDeepStrictEqual(
|
||||
previousInstallRecords,
|
||||
params.nextInstallRecords,
|
||||
prepared.previousInstallRecords,
|
||||
prepared.nextInstallRecords,
|
||||
);
|
||||
return await params.commit(params.nextConfig, {
|
||||
const committed = await params.commit(params.nextConfig, {
|
||||
...params.writeOptions,
|
||||
...(installRecordsChanged && params.writeOptions?.afterWrite === undefined
|
||||
? { afterWrite: { mode: "restart", reason: PLUGIN_SOURCE_CHANGED_RESTART_REASON } }
|
||||
@@ -307,23 +326,35 @@ async function commitPluginInstallRecordsWithWriter(params: {
|
||||
Array.from(PLUGIN_INSTALLS_CONFIG_PATH),
|
||||
]),
|
||||
});
|
||||
return { committed, nextInstallRecords: prepared.nextInstallRecords };
|
||||
} catch (error) {
|
||||
try {
|
||||
// Keep config and install index atomic from the caller's perspective.
|
||||
await writePersistedInstalledPluginIndexInstallRecords(previousInstallRecords);
|
||||
} catch (rollbackError) {
|
||||
throw new Error(
|
||||
"Failed to commit plugin install records and could not restore the previous plugin index",
|
||||
{ cause: rollbackError },
|
||||
);
|
||||
const tentative = tentativeWrite;
|
||||
if (tentative) {
|
||||
try {
|
||||
const restored = await restorePersistedInstalledPluginIndexIfCurrent(
|
||||
tentative.previous,
|
||||
tentative.revision,
|
||||
{
|
||||
filePath: lease.databasePath,
|
||||
lease,
|
||||
},
|
||||
);
|
||||
if (restored) {
|
||||
// Marker compensation belongs to the same tentative revision. A newer
|
||||
// index owner may rely on the current marker state.
|
||||
await restoreClearedRetainedManagedNpmInstallMarkers(clearedMarkerSnapshots);
|
||||
await removeCreatedRetainedManagedNpmInstallMarkers(retainedMarkerPaths);
|
||||
}
|
||||
} catch (rollbackError) {
|
||||
throw new Error(
|
||||
"Failed to commit plugin install records and could not roll back tentative plugin state",
|
||||
{ cause: rollbackError },
|
||||
);
|
||||
}
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
} catch (error) {
|
||||
await restoreClearedRetainedManagedNpmInstallMarkers(clearedMarkerSnapshots);
|
||||
await removeCreatedRetainedManagedNpmInstallMarkers(retainedMarkerPaths);
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Persist plugin install records and commit the matching config update to disk. */
|
||||
@@ -335,7 +366,14 @@ export async function commitPluginInstallRecordsWithConfig(params: {
|
||||
writeOptions?: ConfigWriteOptions;
|
||||
}): Promise<void> {
|
||||
await commitPluginInstallRecordsWithWriter({
|
||||
...params,
|
||||
prepareInstallRecords: async (storeOptions) => ({
|
||||
previousInstallRecords:
|
||||
params.previousInstallRecords ??
|
||||
(await loadInstalledPluginIndexInstallRecords(storeOptions)),
|
||||
nextInstallRecords: params.nextInstallRecords,
|
||||
}),
|
||||
nextConfig: params.nextConfig,
|
||||
...(params.writeOptions ? { writeOptions: params.writeOptions } : {}),
|
||||
commit: async (nextConfig, writeOptions) => {
|
||||
return await replaceConfigFile({
|
||||
nextConfig,
|
||||
@@ -350,12 +388,17 @@ export async function commitPluginInstallRecordsWithConfig(params: {
|
||||
export async function commitPluginInstallRecordsOnly(params: {
|
||||
previousInstallRecords?: Record<string, PluginInstallRecord>;
|
||||
nextInstallRecords: Record<string, PluginInstallRecord>;
|
||||
nextConfig: OpenClawConfig;
|
||||
verifyConfigFresh?: () => Promise<void>;
|
||||
}): Promise<void> {
|
||||
await commitPluginInstallRecordsWithWriter({
|
||||
previousInstallRecords: params.previousInstallRecords,
|
||||
nextInstallRecords: params.nextInstallRecords,
|
||||
nextConfig: {},
|
||||
prepareInstallRecords: async (storeOptions) => ({
|
||||
previousInstallRecords:
|
||||
params.previousInstallRecords ??
|
||||
(await loadInstalledPluginIndexInstallRecords(storeOptions)),
|
||||
nextInstallRecords: params.nextInstallRecords,
|
||||
}),
|
||||
nextConfig: params.nextConfig,
|
||||
commit: async () => {
|
||||
await params.verifyConfigFresh?.();
|
||||
return undefined;
|
||||
@@ -401,25 +444,28 @@ export async function commitConfigWriteWithPendingPluginInstalls(params: {
|
||||
}
|
||||
|
||||
const pendingInstallRecords = nextPendingConfig.plugins?.installs ?? {};
|
||||
const previousInstallRecords = await loadInstalledPluginIndexInstallRecords();
|
||||
const nextInstallRecords = {
|
||||
...sourceInstallRecords,
|
||||
...previousInstallRecords,
|
||||
...pendingInstallRecords,
|
||||
};
|
||||
const strippedConfig = withoutPluginInstallRecords(params.nextConfig);
|
||||
const committed = await commitPluginInstallRecordsWithWriter({
|
||||
previousInstallRecords,
|
||||
nextInstallRecords,
|
||||
const result = await commitPluginInstallRecordsWithWriter({
|
||||
prepareInstallRecords: async (storeOptions) => {
|
||||
const previousInstallRecords = await loadInstalledPluginIndexInstallRecords(storeOptions);
|
||||
return {
|
||||
previousInstallRecords,
|
||||
nextInstallRecords: {
|
||||
...sourceInstallRecords,
|
||||
...previousInstallRecords,
|
||||
...pendingInstallRecords,
|
||||
},
|
||||
};
|
||||
},
|
||||
nextConfig: strippedConfig,
|
||||
...(params.writeOptions ? { writeOptions: params.writeOptions } : {}),
|
||||
commit: params.commit,
|
||||
});
|
||||
return {
|
||||
config: strippedConfig,
|
||||
installRecords: nextInstallRecords,
|
||||
installRecords: result.nextInstallRecords,
|
||||
movedInstallRecords: true,
|
||||
persistedHash: committed?.persistedHash ?? null,
|
||||
persistedHash: result.committed?.persistedHash ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -479,8 +525,12 @@ export async function transformConfigWithPendingPluginInstalls<T = void>(
|
||||
};
|
||||
};
|
||||
|
||||
return await transformConfigFileWithRetry<T>({
|
||||
...params,
|
||||
commit,
|
||||
// The config lock is acquired inside the transform. Own the plugin lifecycle
|
||||
// lease first so pending-record commits keep the canonical lock order.
|
||||
return await withPluginLifecycleLease({}, async () => {
|
||||
return await transformConfigFileWithRetry<T>({
|
||||
...params,
|
||||
commit,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@@ -12,6 +12,9 @@ import { resolveInstalledPluginIndexStorePath } from "./installed-plugin-index-s
|
||||
import {
|
||||
refreshPersistedInstalledPluginIndex,
|
||||
refreshPersistedInstalledPluginIndexSync,
|
||||
refreshPersistedInstalledPluginIndexWithLeaseSync,
|
||||
type InstalledPluginIndexWriteLease,
|
||||
type InstalledPluginIndexWriteReceipt,
|
||||
} from "./installed-plugin-index-store.js";
|
||||
import type { RefreshInstalledPluginIndexParams } from "./installed-plugin-index.js";
|
||||
import { recordPluginInstall, type PluginInstallUpdate } from "./installs.js";
|
||||
@@ -59,6 +62,20 @@ export async function writePersistedInstalledPluginIndexInstallRecords(
|
||||
return resolveInstalledPluginIndexRecordsStorePath(options);
|
||||
}
|
||||
|
||||
/** Refresh persisted install records while holding the plugin lifecycle lease. */
|
||||
export async function writePersistedInstalledPluginIndexInstallRecordsWithLease(
|
||||
records: Record<string, PluginInstallRecord>,
|
||||
options: InstalledPluginIndexRecordRefreshOptions & {
|
||||
lease: InstalledPluginIndexWriteLease;
|
||||
},
|
||||
): Promise<InstalledPluginIndexWriteReceipt> {
|
||||
return refreshPersistedInstalledPluginIndexWithLeaseSync({
|
||||
...options,
|
||||
reason: "source-changed",
|
||||
installRecords: records,
|
||||
});
|
||||
}
|
||||
|
||||
/** Refreshes persisted installed plugin index records synchronously. */
|
||||
export function writePersistedInstalledPluginIndexInstallRecordsSync(
|
||||
records: Record<string, PluginInstallRecord>,
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
// Covers installed plugin index store persistence and recovery behavior.
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { requireNodeSqlite } from "../infra/node-sqlite.js";
|
||||
import {
|
||||
acquireStartupMigrationLease,
|
||||
@@ -12,16 +12,25 @@ import {
|
||||
runOpenClawStateWriteTransaction,
|
||||
} from "../state/openclaw-state-db.js";
|
||||
import type { PluginCandidate } from "./discovery.js";
|
||||
import { readPersistedInstalledPluginIndexInstallRecords } from "./installed-plugin-index-records.js";
|
||||
import {
|
||||
readPersistedInstalledPluginIndexInstallRecords,
|
||||
writePersistedInstalledPluginIndexInstallRecords,
|
||||
writePersistedInstalledPluginIndexInstallRecordsWithLease,
|
||||
} from "./installed-plugin-index-records.js";
|
||||
import {
|
||||
inspectPersistedInstalledPluginIndex,
|
||||
readPersistedInstalledPluginIndex,
|
||||
refreshPersistedInstalledPluginIndex,
|
||||
resolveInstalledPluginIndexStorePath,
|
||||
restorePersistedInstalledPluginIndexIfCurrent,
|
||||
writePersistedInstalledPluginIndex,
|
||||
writePersistedInstalledPluginIndexWithLeaseSync,
|
||||
} from "./installed-plugin-index-store.js";
|
||||
import type { InstalledPluginIndex } from "./installed-plugin-index.js";
|
||||
import {
|
||||
resolveInstalledPluginIndexPolicyHash,
|
||||
type InstalledPluginIndex,
|
||||
} from "./installed-plugin-index.js";
|
||||
import { loadPluginRegistrySnapshotWithMetadata } from "./plugin-registry-snapshot.js";
|
||||
import { cleanupTrackedTempDirs, makeTrackedTempDir } from "./test-helpers/fs-fixtures.js";
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
@@ -103,6 +112,13 @@ function requirePersisted(index: InstalledPluginIndex | null): InstalledPluginIn
|
||||
return index;
|
||||
}
|
||||
|
||||
function requirePersistedRevision(revision: number | null): number {
|
||||
if (revision === null) {
|
||||
throw new Error("Expected persisted installed plugin index revision");
|
||||
}
|
||||
return revision;
|
||||
}
|
||||
|
||||
function expectPluginIds(index: InstalledPluginIndex, expected: string[]) {
|
||||
expect(index.plugins.map((plugin) => plugin.pluginId)).toEqual(expected);
|
||||
}
|
||||
@@ -205,6 +221,24 @@ function insertPersistedIndexRow(
|
||||
);
|
||||
}
|
||||
|
||||
function readPersistedIndexRevision(stateDir: string): number | null {
|
||||
return runOpenClawStateWriteTransaction(
|
||||
({ db }) => {
|
||||
const row = db
|
||||
.prepare(
|
||||
`
|
||||
SELECT updated_at_ms
|
||||
FROM installed_plugin_index
|
||||
WHERE index_key = 'installed-plugin-index'
|
||||
`,
|
||||
)
|
||||
.get() as { updated_at_ms: number | bigint } | undefined;
|
||||
return row ? Number(row.updated_at_ms) : null;
|
||||
},
|
||||
{ env: { ...process.env, OPENCLAW_STATE_DIR: stateDir } },
|
||||
);
|
||||
}
|
||||
|
||||
describe("installed plugin index persistence", () => {
|
||||
it("resolves the persisted index path to the shared state database", () => {
|
||||
const stateDir = makeTempDir();
|
||||
@@ -232,6 +266,102 @@ describe("installed plugin index persistence", () => {
|
||||
expectPluginFields(persisted, "demo", { packageBuild: { bundledDist: false } });
|
||||
});
|
||||
|
||||
it("atomically captures the predecessor and revision for a leased install-record write", async () => {
|
||||
const stateDir = makeTempDir();
|
||||
const lease = { assertOwnedInTransaction: vi.fn() };
|
||||
await writePersistedInstalledPluginIndex(createIndex({ policyHash: "successor" }), {
|
||||
stateDir,
|
||||
});
|
||||
const predecessor = requirePersisted(await readPersistedInstalledPluginIndex({ stateDir }));
|
||||
|
||||
const receipt = await writePersistedInstalledPluginIndexInstallRecordsWithLease(
|
||||
{},
|
||||
{
|
||||
stateDir,
|
||||
candidates: [],
|
||||
lease,
|
||||
},
|
||||
);
|
||||
|
||||
expect(receipt.previous).toEqual(predecessor);
|
||||
expect(receipt.revision).toBe(requirePersistedRevision(readPersistedIndexRevision(stateDir)));
|
||||
expect(lease.assertOwnedInTransaction).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("conditionally restores a matching tentative index revision", async () => {
|
||||
const stateDir = makeTempDir();
|
||||
const lease = { assertOwnedInTransaction: vi.fn() };
|
||||
await writePersistedInstalledPluginIndex(createIndex({ policyHash: "previous" }), {
|
||||
stateDir,
|
||||
});
|
||||
const previous = requirePersisted(await readPersistedInstalledPluginIndex({ stateDir }));
|
||||
await writePersistedInstalledPluginIndex(createIndex({ policyHash: "tentative" }), {
|
||||
stateDir,
|
||||
});
|
||||
const tentativeRevision = requirePersistedRevision(readPersistedIndexRevision(stateDir));
|
||||
|
||||
await expect(
|
||||
restorePersistedInstalledPluginIndexIfCurrent(previous, tentativeRevision, {
|
||||
stateDir,
|
||||
lease,
|
||||
}),
|
||||
).resolves.toBe(true);
|
||||
|
||||
await expect(readPersistedInstalledPluginIndex({ stateDir })).resolves.toEqual(previous);
|
||||
expect(lease.assertOwnedInTransaction).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("conditionally restores matching prior index absence", async () => {
|
||||
const stateDir = makeTempDir();
|
||||
const lease = { assertOwnedInTransaction: vi.fn() };
|
||||
await writePersistedInstalledPluginIndex(createIndex({ policyHash: "tentative" }), {
|
||||
stateDir,
|
||||
});
|
||||
const tentativeRevision = requirePersistedRevision(readPersistedIndexRevision(stateDir));
|
||||
|
||||
await expect(
|
||||
restorePersistedInstalledPluginIndexIfCurrent(null, tentativeRevision, {
|
||||
stateDir,
|
||||
lease,
|
||||
}),
|
||||
).resolves.toBe(true);
|
||||
|
||||
await expect(readPersistedInstalledPluginIndex({ stateDir })).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it("keeps a successor index when conditional rollback sees a newer revision", async () => {
|
||||
const stateDir = makeTempDir();
|
||||
const lease = { assertOwnedInTransaction: vi.fn() };
|
||||
const nowSpy = vi.spyOn(Date, "now").mockReturnValue(1_000);
|
||||
try {
|
||||
await writePersistedInstalledPluginIndex(createIndex({ policyHash: "previous" }), {
|
||||
stateDir,
|
||||
});
|
||||
const previous = requirePersisted(await readPersistedInstalledPluginIndex({ stateDir }));
|
||||
await writePersistedInstalledPluginIndex(createIndex({ policyHash: "tentative" }), {
|
||||
stateDir,
|
||||
});
|
||||
const tentativeRevision = requirePersistedRevision(readPersistedIndexRevision(stateDir));
|
||||
await writePersistedInstalledPluginIndex(createIndex({ policyHash: "successor" }), {
|
||||
stateDir,
|
||||
});
|
||||
const successorRevision = requirePersistedRevision(readPersistedIndexRevision(stateDir));
|
||||
|
||||
expect(successorRevision).toBeGreaterThan(tentativeRevision);
|
||||
await expect(
|
||||
restorePersistedInstalledPluginIndexIfCurrent(previous, tentativeRevision, {
|
||||
stateDir,
|
||||
lease,
|
||||
}),
|
||||
).resolves.toBe(false);
|
||||
expect(
|
||||
requirePersisted(await readPersistedInstalledPluginIndex({ stateDir })).policyHash,
|
||||
).toBe("successor");
|
||||
} finally {
|
||||
nowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects a stale leased write without replacing the successor index", async () => {
|
||||
const stateDir = makeTempDir();
|
||||
const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir };
|
||||
@@ -265,6 +395,41 @@ describe("installed plugin index persistence", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("rereads install-record writes under their non-default policy", async () => {
|
||||
const stateDir = makeTempDir();
|
||||
const pluginDir = path.join(stateDir, "plugins", "demo");
|
||||
fs.mkdirSync(pluginDir, { recursive: true });
|
||||
const candidate = createCandidate(pluginDir);
|
||||
const config = {
|
||||
plugins: {
|
||||
entries: {
|
||||
demo: { enabled: false },
|
||||
},
|
||||
},
|
||||
};
|
||||
const env = {
|
||||
OPENCLAW_BUNDLED_PLUGINS_DIR: undefined,
|
||||
OPENCLAW_VERSION: "2026.4.25",
|
||||
VITEST: "true",
|
||||
};
|
||||
|
||||
await writePersistedInstalledPluginIndexInstallRecords(
|
||||
{ demo: { source: "npm", spec: "demo@1.0.0", installPath: pluginDir } },
|
||||
{ stateDir, candidates: [candidate], config, env },
|
||||
);
|
||||
const result = loadPluginRegistrySnapshotWithMetadata({
|
||||
stateDir,
|
||||
candidates: [candidate],
|
||||
config,
|
||||
env,
|
||||
});
|
||||
|
||||
expect(result.source).toBe("persisted");
|
||||
expect(result.diagnostics).toStrictEqual([]);
|
||||
expect(result.snapshot.policyHash).toBe(resolveInstalledPluginIndexPolicyHash(config));
|
||||
expectPluginFields(result.snapshot, "demo", { enabled: false });
|
||||
});
|
||||
|
||||
it("hashes and persists resolved doctor contract artifacts", async () => {
|
||||
const stateDir = makeTempDir();
|
||||
const pluginDir = path.join(stateDir, "plugins", "demo");
|
||||
|
||||
@@ -54,6 +54,11 @@ export type InstalledPluginIndexWriteLease = {
|
||||
assertOwnedInTransaction(database: DatabaseSync): void;
|
||||
};
|
||||
|
||||
export type InstalledPluginIndexWriteReceipt = {
|
||||
previous: InstalledPluginIndex | null;
|
||||
revision: number;
|
||||
};
|
||||
|
||||
const StringArraySchema = z.array(z.string());
|
||||
const INSTALLED_PLUGIN_INDEX_SQLITE_KEY = "installed-plugin-index";
|
||||
|
||||
@@ -203,6 +208,7 @@ type InstalledPluginIndexSqliteRow = {
|
||||
install_records_json: string;
|
||||
plugins_json: string;
|
||||
diagnostics_json: string;
|
||||
updated_at_ms: number | bigint;
|
||||
};
|
||||
|
||||
function assertWritableInstalledPluginIndexStoreOptions(
|
||||
@@ -244,50 +250,43 @@ function parseInstalledPluginIndexSqliteRow(
|
||||
});
|
||||
}
|
||||
|
||||
function readPersistedInstalledPluginIndexFromSqlite(
|
||||
options: InstalledPluginIndexStoreOptions = {},
|
||||
): InstalledPluginIndex | null {
|
||||
if (options.filePath?.endsWith(".json")) {
|
||||
return null;
|
||||
}
|
||||
if (!existsSync(resolveInstalledPluginIndexStorePath(options))) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return withOpenClawStateDatabaseReadOnly(({ db }) => {
|
||||
const row = db
|
||||
.prepare(
|
||||
`
|
||||
SELECT version, warning, host_contract_version, compat_registry_version,
|
||||
migration_version, policy_hash, generated_at_ms, refresh_reason,
|
||||
install_records_json, plugins_json, diagnostics_json
|
||||
FROM installed_plugin_index
|
||||
WHERE index_key = ?
|
||||
`,
|
||||
)
|
||||
.get(INSTALLED_PLUGIN_INDEX_SQLITE_KEY) as InstalledPluginIndexSqliteRow | undefined;
|
||||
return parseInstalledPluginIndexSqliteRow(row);
|
||||
}, resolveInstalledPluginIndexStateDatabaseOptions(options));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function writePersistedInstalledPluginIndexToSqlite(
|
||||
index: InstalledPluginIndex,
|
||||
options: InstalledPluginIndexStoreOptions = {},
|
||||
lease?: InstalledPluginIndexWriteLease,
|
||||
): void {
|
||||
assertWritableInstalledPluginIndexStoreOptions(options);
|
||||
const persisted = {
|
||||
function preparePersistedInstalledPluginIndex(index: InstalledPluginIndex): InstalledPluginIndex {
|
||||
return {
|
||||
...index,
|
||||
warning: INSTALLED_PLUGIN_INDEX_WARNING,
|
||||
installRecords: copySafeInstallRecords(index.installRecords) ?? {},
|
||||
};
|
||||
const now = Date.now();
|
||||
runOpenClawStateWriteTransaction(({ db }) => {
|
||||
lease?.assertOwnedInTransaction(db);
|
||||
db.prepare(
|
||||
}
|
||||
|
||||
function readInstalledPluginIndexRow(
|
||||
database: DatabaseSync,
|
||||
): InstalledPluginIndexSqliteRow | undefined {
|
||||
return database
|
||||
.prepare(
|
||||
`
|
||||
SELECT version, warning, host_contract_version, compat_registry_version,
|
||||
migration_version, policy_hash, generated_at_ms, refresh_reason,
|
||||
install_records_json, plugins_json, diagnostics_json, updated_at_ms
|
||||
FROM installed_plugin_index
|
||||
WHERE index_key = ?
|
||||
`,
|
||||
)
|
||||
.get(INSTALLED_PLUGIN_INDEX_SQLITE_KEY) as InstalledPluginIndexSqliteRow | undefined;
|
||||
}
|
||||
|
||||
function resolveNextInstalledPluginIndexRevision(current: number | null): number {
|
||||
// Revisions fence rollback across processes, so same-millisecond writes must
|
||||
// still receive distinct values.
|
||||
return Math.max(Date.now(), (current ?? 0) + 1);
|
||||
}
|
||||
|
||||
function writePersistedInstalledPluginIndexRow(
|
||||
database: DatabaseSync,
|
||||
index: InstalledPluginIndex,
|
||||
revision: number,
|
||||
): void {
|
||||
database
|
||||
.prepare(
|
||||
`
|
||||
INSERT INTO installed_plugin_index (
|
||||
index_key, version, host_contract_version, compat_registry_version,
|
||||
@@ -312,24 +311,69 @@ function writePersistedInstalledPluginIndexToSqlite(
|
||||
warning = excluded.warning,
|
||||
updated_at_ms = excluded.updated_at_ms
|
||||
`,
|
||||
).run({
|
||||
)
|
||||
.run({
|
||||
index_key: INSTALLED_PLUGIN_INDEX_SQLITE_KEY,
|
||||
version: persisted.version,
|
||||
host_contract_version: persisted.hostContractVersion,
|
||||
compat_registry_version: persisted.compatRegistryVersion,
|
||||
migration_version: persisted.migrationVersion,
|
||||
policy_hash: persisted.policyHash,
|
||||
generated_at_ms: persisted.generatedAtMs,
|
||||
refresh_reason: persisted.refreshReason ?? null,
|
||||
install_records_json: JSON.stringify(persisted.installRecords),
|
||||
plugins_json: JSON.stringify(persisted.plugins),
|
||||
diagnostics_json: JSON.stringify(persisted.diagnostics),
|
||||
warning: persisted.warning,
|
||||
updated_at_ms: now,
|
||||
version: index.version,
|
||||
host_contract_version: index.hostContractVersion,
|
||||
compat_registry_version: index.compatRegistryVersion,
|
||||
migration_version: index.migrationVersion,
|
||||
policy_hash: index.policyHash,
|
||||
generated_at_ms: index.generatedAtMs,
|
||||
refresh_reason: index.refreshReason ?? null,
|
||||
install_records_json: JSON.stringify(index.installRecords),
|
||||
plugins_json: JSON.stringify(index.plugins),
|
||||
diagnostics_json: JSON.stringify(index.diagnostics),
|
||||
warning: index.warning ?? INSTALLED_PLUGIN_INDEX_WARNING,
|
||||
updated_at_ms: revision,
|
||||
});
|
||||
}
|
||||
|
||||
function readPersistedInstalledPluginIndexFromSqlite(
|
||||
options: InstalledPluginIndexStoreOptions = {},
|
||||
): InstalledPluginIndex | null {
|
||||
if (options.filePath?.endsWith(".json")) {
|
||||
return null;
|
||||
}
|
||||
if (!existsSync(resolveInstalledPluginIndexStorePath(options))) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return withOpenClawStateDatabaseReadOnly(
|
||||
({ db }) => parseInstalledPluginIndexSqliteRow(readInstalledPluginIndexRow(db)),
|
||||
resolveInstalledPluginIndexStateDatabaseOptions(options),
|
||||
);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function writePersistedInstalledPluginIndexToSqlite(
|
||||
index: InstalledPluginIndex,
|
||||
options: InstalledPluginIndexStoreOptions = {},
|
||||
lease?: InstalledPluginIndexWriteLease,
|
||||
): InstalledPluginIndexWriteReceipt {
|
||||
assertWritableInstalledPluginIndexStoreOptions(options);
|
||||
const persisted = preparePersistedInstalledPluginIndex(index);
|
||||
return runOpenClawStateWriteTransaction(({ db }) => {
|
||||
lease?.assertOwnedInTransaction(db);
|
||||
const previousRow = readInstalledPluginIndexRow(db);
|
||||
const revision = resolveNextInstalledPluginIndexRevision(
|
||||
previousRow ? Number(previousRow.updated_at_ms) : null,
|
||||
);
|
||||
writePersistedInstalledPluginIndexRow(db, persisted, revision);
|
||||
return {
|
||||
previous: parseInstalledPluginIndexSqliteRow(previousRow),
|
||||
revision,
|
||||
};
|
||||
}, resolveInstalledPluginIndexStateDatabaseOptions(options));
|
||||
}
|
||||
|
||||
function clearPersistedInstalledPluginIndexCaches(): void {
|
||||
clearPluginMetadataLifecycleCaches();
|
||||
clearLoadInstalledPluginIndexInstallRecordsCache();
|
||||
}
|
||||
|
||||
export async function readPersistedInstalledPluginIndex(
|
||||
options: InstalledPluginIndexStoreOptions = {},
|
||||
): Promise<InstalledPluginIndex | null> {
|
||||
@@ -349,14 +393,55 @@ export async function writePersistedInstalledPluginIndex(
|
||||
return writePersistedInstalledPluginIndexSync(index, options);
|
||||
}
|
||||
|
||||
/** Restore a snapshot only while the caller's tentative write is still current. */
|
||||
export async function restorePersistedInstalledPluginIndexIfCurrent(
|
||||
index: InstalledPluginIndex | null,
|
||||
expectedRevision: number,
|
||||
options: InstalledPluginIndexStoreOptions & {
|
||||
lease: InstalledPluginIndexWriteLease;
|
||||
},
|
||||
): Promise<boolean> {
|
||||
const { lease, ...storeOptions } = options;
|
||||
assertWritableInstalledPluginIndexStoreOptions(storeOptions);
|
||||
if (!existsSync(resolveInstalledPluginIndexStorePath(storeOptions))) {
|
||||
return false;
|
||||
}
|
||||
const restored = runOpenClawStateWriteTransaction(({ db }) => {
|
||||
lease.assertOwnedInTransaction(db);
|
||||
const currentRow = readInstalledPluginIndexRow(db);
|
||||
const currentRevision = currentRow ? Number(currentRow.updated_at_ms) : null;
|
||||
if (currentRevision !== expectedRevision) {
|
||||
return false;
|
||||
}
|
||||
if (index) {
|
||||
writePersistedInstalledPluginIndexRow(
|
||||
db,
|
||||
preparePersistedInstalledPluginIndex(index),
|
||||
resolveNextInstalledPluginIndexRevision(currentRevision),
|
||||
);
|
||||
} else {
|
||||
db.prepare(
|
||||
`
|
||||
DELETE FROM installed_plugin_index
|
||||
WHERE index_key = ?
|
||||
`,
|
||||
).run(INSTALLED_PLUGIN_INDEX_SQLITE_KEY);
|
||||
}
|
||||
return true;
|
||||
}, resolveInstalledPluginIndexStateDatabaseOptions(storeOptions));
|
||||
// A mismatched revision means another process committed, which also makes
|
||||
// this process's cached metadata stale.
|
||||
clearPersistedInstalledPluginIndexCaches();
|
||||
return restored;
|
||||
}
|
||||
|
||||
export function writePersistedInstalledPluginIndexSync(
|
||||
index: InstalledPluginIndex,
|
||||
options: InstalledPluginIndexStoreOptions = {},
|
||||
): string {
|
||||
const filePath = resolveInstalledPluginIndexStorePath(options);
|
||||
writePersistedInstalledPluginIndexToSqlite(index, options);
|
||||
clearPluginMetadataLifecycleCaches();
|
||||
clearLoadInstalledPluginIndexInstallRecordsCache();
|
||||
clearPersistedInstalledPluginIndexCaches();
|
||||
return filePath;
|
||||
}
|
||||
|
||||
@@ -369,8 +454,7 @@ export function writePersistedInstalledPluginIndexWithLeaseSync(
|
||||
const { lease, ...storeOptions } = options;
|
||||
const filePath = resolveInstalledPluginIndexStorePath(storeOptions);
|
||||
writePersistedInstalledPluginIndexToSqlite(index, storeOptions, lease);
|
||||
clearPluginMetadataLifecycleCaches();
|
||||
clearLoadInstalledPluginIndexInstallRecordsCache();
|
||||
clearPersistedInstalledPluginIndexCaches();
|
||||
return filePath;
|
||||
}
|
||||
|
||||
@@ -467,7 +551,7 @@ export async function refreshPersistedInstalledPluginIndex(
|
||||
return refreshPersistedInstalledPluginIndexSync(params);
|
||||
}
|
||||
|
||||
export function refreshPersistedInstalledPluginIndexSync(
|
||||
function resolveRefreshedPersistedInstalledPluginIndex(
|
||||
params: RefreshInstalledPluginIndexParams & InstalledPluginIndexStoreOptions,
|
||||
): InstalledPluginIndex {
|
||||
const persisted =
|
||||
@@ -475,15 +559,32 @@ export function refreshPersistedInstalledPluginIndexSync(
|
||||
? readPersistedInstalledPluginIndexSync(params)
|
||||
: null;
|
||||
if (canRefreshPersistedPolicyState(persisted, params)) {
|
||||
const index = refreshPersistedPolicyState(persisted, params);
|
||||
writePersistedInstalledPluginIndexSync(index, params);
|
||||
return index;
|
||||
return refreshPersistedPolicyState(persisted, params);
|
||||
}
|
||||
const index = refreshInstalledPluginIndex({
|
||||
return refreshInstalledPluginIndex({
|
||||
...params,
|
||||
installRecords:
|
||||
params.installRecords ?? extractPluginInstallRecordsFromInstalledPluginIndex(persisted),
|
||||
});
|
||||
}
|
||||
|
||||
export function refreshPersistedInstalledPluginIndexSync(
|
||||
params: RefreshInstalledPluginIndexParams & InstalledPluginIndexStoreOptions,
|
||||
): InstalledPluginIndex {
|
||||
const index = resolveRefreshedPersistedInstalledPluginIndex(params);
|
||||
writePersistedInstalledPluginIndexSync(index, params);
|
||||
return index;
|
||||
}
|
||||
|
||||
export function refreshPersistedInstalledPluginIndexWithLeaseSync(
|
||||
params: RefreshInstalledPluginIndexParams &
|
||||
InstalledPluginIndexStoreOptions & {
|
||||
lease: InstalledPluginIndexWriteLease;
|
||||
},
|
||||
): InstalledPluginIndexWriteReceipt {
|
||||
const { lease, ...storeParams } = params;
|
||||
const index = resolveRefreshedPersistedInstalledPluginIndex(storeParams);
|
||||
const receipt = writePersistedInstalledPluginIndexToSqlite(index, storeParams, lease);
|
||||
clearPersistedInstalledPluginIndexCaches();
|
||||
return receipt;
|
||||
}
|
||||
|
||||
@@ -272,12 +272,19 @@ describe("plugin lifecycle lease", () => {
|
||||
it("reuses the active lease for nested lifecycle work", async () => {
|
||||
await withOpenClawTestState({ label: "plugin-lifecycle-reentrant" }, async (state) => {
|
||||
const events: string[] = [];
|
||||
await withPluginLifecycleLease({ env: state.env, leaseMs: 1_000, waitMs: 0 }, async () => {
|
||||
events.push("outer");
|
||||
await withPluginLifecycleLease({ env: state.env, leaseMs: 1_000, waitMs: 0 }, async () => {
|
||||
events.push("inner");
|
||||
});
|
||||
});
|
||||
await withPluginLifecycleLease(
|
||||
{ env: state.env, leaseMs: 1_000, waitMs: 0 },
|
||||
async (outerLease) => {
|
||||
events.push("outer");
|
||||
await withPluginLifecycleLease({}, async (innerLease) => {
|
||||
events.push("inner");
|
||||
expect(innerLease).toBe(outerLease);
|
||||
expect(innerLease.databasePath).toBe(
|
||||
path.resolve(state.stateDir, "state", "openclaw.sqlite"),
|
||||
);
|
||||
});
|
||||
},
|
||||
);
|
||||
expect(events).toEqual(["outer", "inner"]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -14,9 +14,13 @@ const PLUGIN_LIFECYCLE_LEASE_KEY = "global";
|
||||
const DEFAULT_PLUGIN_LIFECYCLE_LEASE_MS = 5 * 60_000;
|
||||
const DEFAULT_PLUGIN_LIFECYCLE_WAIT_MS = 10 * 60_000;
|
||||
|
||||
type PluginLifecycleLeaseContext = OpenClawStateLeaseContext & {
|
||||
databasePath: string;
|
||||
};
|
||||
|
||||
type ActivePluginLifecycleLease = {
|
||||
databasePath: string;
|
||||
lease: OpenClawStateLeaseContext;
|
||||
lease: PluginLifecycleLeaseContext;
|
||||
};
|
||||
|
||||
type PluginLifecycleLeaseOptions = Pick<
|
||||
@@ -46,13 +50,24 @@ function resolveLifecycleLeaseEnv(env: NodeJS.ProcessEnv | undefined): NodeJS.Pr
|
||||
/** Serialize plugin artifact, install-index, and config mutations across processes. */
|
||||
export async function withPluginLifecycleLease<T>(
|
||||
options: PluginLifecycleLeaseOptions,
|
||||
run: (lease: OpenClawStateLeaseContext) => Promise<T>,
|
||||
run: (lease: PluginLifecycleLeaseContext) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const active = activePluginLifecycleLease.getStore();
|
||||
if (
|
||||
active &&
|
||||
options.env === undefined &&
|
||||
options.path === undefined &&
|
||||
options.database === undefined
|
||||
) {
|
||||
options.signal?.throwIfAborted();
|
||||
active.lease.assertOwned();
|
||||
return await run(active.lease);
|
||||
}
|
||||
|
||||
const env = resolveLifecycleLeaseEnv(options.env);
|
||||
const databasePath = path.resolve(
|
||||
options.database?.path ?? options.path ?? resolveOpenClawStateSqlitePath(env),
|
||||
);
|
||||
const active = activePluginLifecycleLease.getStore();
|
||||
if (active) {
|
||||
if (active.databasePath !== databasePath) {
|
||||
throw new OpenClawStateLeaseError(
|
||||
@@ -84,11 +99,17 @@ export async function withPluginLifecycleLease<T>(
|
||||
operationLabel: "plugins.lifecycle.lease",
|
||||
},
|
||||
async (lease) => {
|
||||
const pluginLease: PluginLifecycleLeaseContext = {
|
||||
databasePath,
|
||||
signal: lease.signal,
|
||||
assertOwned: () => lease.assertOwned(),
|
||||
assertOwnedInTransaction: (database) => lease.assertOwnedInTransaction(database),
|
||||
};
|
||||
// Another process may have committed while this process waited for ownership.
|
||||
clearLoadInstalledPluginIndexInstallRecordsCache();
|
||||
return await activePluginLifecycleLease.run(
|
||||
{ databasePath, lease },
|
||||
async () => await run(lease),
|
||||
{ databasePath, lease: pluginLease },
|
||||
async () => await run(pluginLease),
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user