From ef11eae39b29ad9c4123ba211ef4fa4ac005221c Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 5 Aug 2026 01:04:17 +0800 Subject: [PATCH] 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 --- src/cli/plugins-cli-test-helpers.ts | 107 +++++++- src/cli/plugins-cli.install.test.ts | 10 +- src/cli/plugins-cli.uninstall.test.ts | 67 +++-- src/cli/plugins-cli.update.test.ts | 102 ++++--- src/cli/plugins-update-command.ts | 1 + src/cli/update-cli.test.ts | 148 ++++++++++- .../update-cli/update-command-post-core.ts | 38 ++- src/commands/agents.commands.add.ts | 17 +- src/plugins/install-persistence.test.ts | 16 +- .../install-record-commit.sqlite.test.ts | 192 ++++++++++++++ src/plugins/install-record-commit.test.ts | 249 +++++++++++++++--- src/plugins/install-record-commit.ts | 146 ++++++---- src/plugins/installed-plugin-index-records.ts | 17 ++ .../installed-plugin-index-store.test.ts | 171 +++++++++++- src/plugins/installed-plugin-index-store.ts | 225 +++++++++++----- src/plugins/plugin-lifecycle-lease.test.ts | 19 +- src/plugins/plugin-lifecycle-lease.ts | 31 ++- 17 files changed, 1314 insertions(+), 242 deletions(-) create mode 100644 src/plugins/install-record-commit.sqlite.test.ts diff --git a/src/cli/plugins-cli-test-helpers.ts b/src/cli/plugins-cli-test-helpers.ts index aa3be5f2bc30..55650b42479e 100644 --- a/src/cli/plugins-cli-test-helpers.ts +++ b/src/cli/plugins-cli-test-helpers.ts @@ -6,6 +6,7 @@ import { getRuntimeConfig } from "../config/config.js"; import type { HookInstallRecord } from "../config/types.hooks.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import type { PluginInstallRecord } from "../config/types.plugins.js"; +import type { InstalledPluginIndex } from "../plugins/installed-plugin-index.js"; import type { CliMockOutputRuntime } from "./test-runtime-capture.js"; type UnknownMock = Mock<(...args: unknown[]) => unknown>; @@ -27,6 +28,14 @@ type UpdateNpmInstalledPluginsFn = (typeof import("../plugins/update.js"))["updateNpmInstalledPlugins"]; type UpdateNpmInstalledHookPacksFn = (typeof import("../hooks/update.js"))["updateNpmInstalledHookPacks"]; +type ReadPersistedInstalledPluginIndexFn = + (typeof import("../plugins/installed-plugin-index-store.js"))["readPersistedInstalledPluginIndex"]; +type RestorePersistedInstalledPluginIndexIfCurrentFn = + (typeof import("../plugins/installed-plugin-index-store.js"))["restorePersistedInstalledPluginIndexIfCurrent"]; +type WritePersistedInstalledPluginIndexInstallRecordsFn = + (typeof import("../plugins/installed-plugin-index-records.js"))["writePersistedInstalledPluginIndexInstallRecords"]; +type WritePersistedInstalledPluginIndexInstallRecordsWithLeaseFn = + (typeof import("../plugins/installed-plugin-index-records.js"))["writePersistedInstalledPluginIndexInstallRecordsWithLease"]; type PluginInstallRecordMap = Record; function createEmptyUninstallActions() { @@ -45,6 +54,7 @@ function createEmptyUninstallActions() { let mockInstalledPluginIndexInstallRecords: PluginInstallRecordMap = {}; let mockHookInstallRecords: Record = {}; +let mockInstalledPluginIndexRevision = 0; export function setHookInstallRecords(records: Record): void { mockHookInstallRecords = structuredClone(records); @@ -55,6 +65,24 @@ function clonePluginInstallRecords(records: PluginInstallRecordMap): PluginInsta return structuredClone(records); } +export function createTestInstalledPluginIndex(params: { + policyHash: string; + installRecords: PluginInstallRecordMap; +}): InstalledPluginIndex { + return { + version: 1, + hostContractVersion: "test", + compatRegistryVersion: "test", + migrationVersion: 1, + policyHash: params.policyHash, + generatedAtMs: 0, + refreshReason: "source-changed", + installRecords: clonePluginInstallRecords(params.installRecords), + plugins: [], + diagnostics: [], + }; +} + // oxlint-disable-next-line typescript/no-unnecessary-type-parameters -- Test helper preserves mock call and result types. function invokeMock(mock: unknown, ...args: TArgs): TResult { return (mock as (...args: TArgs) => TResult)(...args); @@ -78,13 +106,31 @@ export const recordPluginInstall: UnknownMock = vi.fn(); const loadInstalledPluginIndexInstallRecords: AsyncUnknownMock = vi.fn(async () => clonePluginInstallRecords(mockInstalledPluginIndexInstallRecords), ); -export const writePersistedInstalledPluginIndexInstallRecords: AsyncUnknownMock = vi.fn( - async (records: unknown) => { +const writePersistedInstalledPluginIndexInstallRecords: Mock = + vi.fn(async (records) => { + mockInstalledPluginIndexInstallRecords = clonePluginInstallRecords(records); + return "/tmp/openclaw-state/openclaw.sqlite"; + }); +export const readPersistedInstalledPluginIndex: Mock = + vi.fn(async () => null); +export const writePersistedInstalledPluginIndexInstallRecordsWithLease: Mock = + vi.fn(async (records) => { + const previous = await readPersistedInstalledPluginIndex(); + mockInstalledPluginIndexInstallRecords = clonePluginInstallRecords(records); + mockInstalledPluginIndexRevision += 1; + return { previous, revision: mockInstalledPluginIndexRevision }; + }); +export const restorePersistedInstalledPluginIndexIfCurrent: Mock = + vi.fn(async (index, expectedRevision) => { + if (mockInstalledPluginIndexRevision !== expectedRevision) { + return false; + } mockInstalledPluginIndexInstallRecords = clonePluginInstallRecords( - (records ?? {}) as PluginInstallRecordMap, + (index?.installRecords ?? {}) as PluginInstallRecordMap, ); - }, -); + mockInstalledPluginIndexRevision += 1; + return true; + }); export const loadPluginManifestRegistry: UnknownMock = vi.fn(); export const buildPluginSnapshotReport: UnknownMock = vi.fn(); export const buildPluginRegistrySnapshotReport: UnknownMock = vi.fn(); @@ -319,6 +365,11 @@ vi.mock("../plugins/installed-plugin-index-records.js", async (importOriginal) = writePersistedInstalledPluginIndexInstallRecords, ...args, )) as (...args: unknown[]) => unknown, + writePersistedInstalledPluginIndexInstallRecordsWithLease: ((...args: unknown[]) => + invokeMock( + writePersistedInstalledPluginIndexInstallRecordsWithLease, + ...args, + )) as (...args: unknown[]) => unknown, recordPluginInstallInRecords: ( records: Record, update: { pluginId: string; installedAt?: string } & Record, @@ -336,6 +387,22 @@ vi.mock("../plugins/installed-plugin-index-records.js", async (importOriginal) = }; }); +vi.mock("../plugins/installed-plugin-index-store.js", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + readPersistedInstalledPluginIndex: ((...args: unknown[]) => + invokeMock(readPersistedInstalledPluginIndex, ...args)) as ( + ...args: unknown[] + ) => unknown, + restorePersistedInstalledPluginIndexIfCurrent: ((...args: unknown[]) => + invokeMock(restorePersistedInstalledPluginIndexIfCurrent, ...args)) as ( + ...args: unknown[] + ) => unknown, + }; +}); + vi.mock("../plugins/manifest-registry.js", () => ({ loadPluginManifestRegistry: ((...args: unknown[]) => invokeMock(loadPluginManifestRegistry, ...args)) as ( @@ -751,8 +818,12 @@ export function resetPluginsCliTestState() { enablePluginInConfig.mockReset(); recordPluginInstall.mockReset(); mockInstalledPluginIndexInstallRecords = {}; + mockInstalledPluginIndexRevision = 0; loadInstalledPluginIndexInstallRecords.mockReset(); writePersistedInstalledPluginIndexInstallRecords.mockReset(); + writePersistedInstalledPluginIndexInstallRecordsWithLease.mockReset(); + readPersistedInstalledPluginIndex.mockReset(); + restorePersistedInstalledPluginIndexIfCurrent.mockReset(); loadPluginManifestRegistry.mockReset(); buildPluginSnapshotReport.mockReset(); buildPluginRegistrySnapshotReport.mockReset(); @@ -837,11 +908,29 @@ export function resetPluginsCliTestState() { loadInstalledPluginIndexInstallRecords.mockImplementation(async () => clonePluginInstallRecords(mockInstalledPluginIndexInstallRecords), ); - writePersistedInstalledPluginIndexInstallRecords.mockImplementation(async (records: unknown) => { - mockInstalledPluginIndexInstallRecords = clonePluginInstallRecords( - (records ?? {}) as PluginInstallRecordMap, - ); + writePersistedInstalledPluginIndexInstallRecords.mockImplementation(async (records) => { + mockInstalledPluginIndexInstallRecords = clonePluginInstallRecords(records); + return "/tmp/openclaw-state/openclaw.sqlite"; }); + readPersistedInstalledPluginIndex.mockResolvedValue(null); + writePersistedInstalledPluginIndexInstallRecordsWithLease.mockImplementation(async (records) => { + const previous = await readPersistedInstalledPluginIndex(); + mockInstalledPluginIndexInstallRecords = clonePluginInstallRecords(records); + mockInstalledPluginIndexRevision += 1; + return { previous, revision: mockInstalledPluginIndexRevision }; + }); + restorePersistedInstalledPluginIndexIfCurrent.mockImplementation( + async (index, expectedRevision) => { + if (mockInstalledPluginIndexRevision !== expectedRevision) { + return false; + } + mockInstalledPluginIndexInstallRecords = clonePluginInstallRecords( + (index?.installRecords ?? {}) as PluginInstallRecordMap, + ); + mockInstalledPluginIndexRevision += 1; + return true; + }, + ); loadPluginManifestRegistry.mockReturnValue({ plugins: [], diagnostics: [], diff --git a/src/cli/plugins-cli.install.test.ts b/src/cli/plugins-cli.install.test.ts index 0bb08c64ca5a..c71b35d841bb 100644 --- a/src/cli/plugins-cli.install.test.ts +++ b/src/cli/plugins-cli.install.test.ts @@ -40,7 +40,7 @@ import { runtimeErrors, runtimeLogs, writeConfigFile, - writePersistedInstalledPluginIndexInstallRecords, + writePersistedInstalledPluginIndexInstallRecordsWithLease, } from "./plugins-cli-test-helpers.js"; const CLI_STATE_ROOT = "/tmp/openclaw-state"; @@ -430,10 +430,10 @@ function hookNpmInstallCall(callIndex = 0): PluginInstallCall { } function persistedInstallRecords(callIndex = 0): Record { - return mockCallArg(writePersistedInstalledPluginIndexInstallRecords, callIndex) as Record< - string, - PersistedInstallRecord - >; + return mockCallArg( + writePersistedInstalledPluginIndexInstallRecordsWithLease, + callIndex, + ) as Record; } function persistedInstallRecord(pluginId: string, callIndex = 0): PersistedInstallRecord { diff --git a/src/cli/plugins-cli.uninstall.test.ts b/src/cli/plugins-cli.uninstall.test.ts index 049e83f65049..c5bc4a3ad8bb 100644 --- a/src/cli/plugins-cli.uninstall.test.ts +++ b/src/cli/plugins-cli.uninstall.test.ts @@ -10,19 +10,22 @@ import { applyPluginUninstallDirectoryRemoval, buildPluginDiagnosticsReport, buildPluginSnapshotReport, + createTestInstalledPluginIndex, loadConfig, planPluginUninstall, PromptInputClosedError, promptYesNo, + readPersistedInstalledPluginIndex, refreshPluginRegistry, replaceConfigFile, resetPluginsCliTestState, + restorePersistedInstalledPluginIndexIfCurrent, runPluginsCommand, runtimeErrors, runtimeLogs, setInstalledPluginIndexInstallRecords, writeConfigFile, - writePersistedInstalledPluginIndexInstallRecords, + writePersistedInstalledPluginIndexInstallRecordsWithLease, } from "./plugins-cli-test-helpers.js"; const CLI_STATE_ROOT = "/tmp/openclaw-state"; @@ -34,6 +37,17 @@ function expectRuntimeLogIncludes(fragment: string) { expect(runtimeLogs.join("\n")).toContain(fragment); } +function expectInstallRecordsWrittenWithLease(records: unknown, config: unknown) { + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledWith( + records, + expect.objectContaining({ + config, + filePath: expect.any(String), + lease: expect.anything(), + }), + ); +} + function expectLatestUninstallPlanParams(expected: { pluginId: string; deleteFiles: boolean; @@ -186,7 +200,7 @@ describe("plugins cli uninstall", () => { expect(promptYesNo).not.toHaveBeenCalled(); expectLatestUninstallPlanParams({ pluginId: "alpha", deleteFiles: false }); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith({}); + expectInstallRecordsWrittenWithLease({}, { plugins: { entries: {} } }); expect(writeConfigFile).toHaveBeenCalledWith({ plugins: { entries: {}, @@ -267,9 +281,16 @@ describe("plugins cli uninstall", () => { await runPluginsCommand(["plugins", "uninstall", "calendar", "--force", "--keep-files"]); expectLatestUninstallPlanParams({ pluginId: "calendar", deleteFiles: false }); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith({ - "unrelated-plugin": { source: "npm", spec: "unrelated-plugin@1.0.0" }, - }); + expectInstallRecordsWrittenWithLease( + { + "unrelated-plugin": { source: "npm", spec: "unrelated-plugin@1.0.0" }, + }, + { + plugins: { + entries: { "unrelated-plugin": { enabled: true } }, + }, + }, + ); }); it("rejects an ambiguous display name before planning or mutating installed plugins", async () => { @@ -304,7 +325,7 @@ describe("plugins cli uninstall", () => { expect(planPluginUninstall).not.toHaveBeenCalled(); expect(promptYesNo).not.toHaveBeenCalled(); expect(applyPluginUninstallDirectoryRemoval).not.toHaveBeenCalled(); - expect(writePersistedInstalledPluginIndexInstallRecords).not.toHaveBeenCalled(); + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).not.toHaveBeenCalled(); expect(writeConfigFile).not.toHaveBeenCalled(); expect(refreshPluginRegistry).not.toHaveBeenCalled(); }); @@ -368,7 +389,7 @@ describe("plugins cli uninstall", () => { expectRuntimeLogIncludes('Warning: plugin "alpha" is referenced by Claw: @owner/audit-claw.'); expectRuntimeLogIncludes("Uninstalling it may break those Claws"); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith({}); + expectInstallRecordsWrittenWithLease({}, { plugins: { entries: {} } }); } finally { if (previousStateDir === undefined) { delete process.env.OPENCLAW_STATE_DIR; @@ -424,7 +445,7 @@ describe("plugins cli uninstall", () => { expect(runtimeErrors).toContain( "Error: plugins uninstall requires confirmation input. Re-run in an interactive TTY or pass --force.", ); - expect(writePersistedInstalledPluginIndexInstallRecords).not.toHaveBeenCalled(); + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).not.toHaveBeenCalled(); expect(writeConfigFile).not.toHaveBeenCalled(); expect(refreshPluginRegistry).not.toHaveBeenCalled(); expect(applyPluginUninstallDirectoryRemoval).not.toHaveBeenCalled(); @@ -452,9 +473,14 @@ describe("plugins cli uninstall", () => { installs: {}, }, } as OpenClawConfig; + const previousPersistedIndex = createTestInstalledPluginIndex({ + policyHash: "previous-policy", + installRecords, + }); loadConfig.mockReturnValue(baseConfig); setInstalledPluginIndexInstallRecords(installRecords); + readPersistedInstalledPluginIndex.mockResolvedValue(previousPersistedIndex); buildPluginSnapshotReport.mockReturnValue({ plugins: [{ id: "alpha", name: "alpha" }], diagnostics: [], @@ -480,10 +506,14 @@ describe("plugins cli uninstall", () => { runPluginsCommand(["plugins", "uninstall", "alpha", "--force", "--keep-files"]), ).rejects.toThrow("config changed"); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith(1, {}); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith( - 2, - installRecords, + expectInstallRecordsWrittenWithLease({}, { plugins: { entries: {} } }); + expect(restorePersistedInstalledPluginIndexIfCurrent).toHaveBeenCalledWith( + previousPersistedIndex, + expect.any(Number), + expect.objectContaining({ + filePath: expect.any(String), + lease: expect.anything(), + }), ); expect(refreshPluginRegistry).not.toHaveBeenCalled(); expect(applyPluginUninstallDirectoryRemoval).not.toHaveBeenCalled(); @@ -613,7 +643,7 @@ describe("plugins cli uninstall", () => { installs: installRecords, }, }); - expect(writePersistedInstalledPluginIndexInstallRecords).not.toHaveBeenCalled(); + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).not.toHaveBeenCalled(); expect(refreshPluginRegistry).not.toHaveBeenCalled(); }); @@ -762,7 +792,14 @@ describe("plugins cli uninstall", () => { await runPluginsCommand(["plugins", "uninstall", pluginId, "--force", "--keep-files"]); expectLatestUninstallPlanParams({ pluginId, channelIds, deleteFiles: false }); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith({}); + expectInstallRecordsWrittenWithLease( + {}, + expect.objectContaining({ + channels: Object.fromEntries( + Object.entries(channels).filter(([channelId]) => !channelIds.includes(channelId)), + ), + }), + ); expect(writeConfigFile).toHaveBeenCalledWith( expect.objectContaining({ channels: Object.fromEntries( @@ -837,7 +874,7 @@ describe("plugins cli uninstall", () => { channelIds: undefined, deleteFiles: false, }); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith({}); + expectInstallRecordsWrittenWithLease({}, nextConfig); expect(writeConfigFile).toHaveBeenCalledWith(nextConfig); expectRuntimeLogIncludes("channel config (channels.alpha)"); expect(runtimeLogs.at(-2)).toContain('Uninstalled plugin "alpha"'); diff --git a/src/cli/plugins-cli.update.test.ts b/src/cli/plugins-cli.update.test.ts index 364e8d7e5950..739b5c6a06f0 100644 --- a/src/cli/plugins-cli.update.test.ts +++ b/src/cli/plugins-cli.update.test.ts @@ -8,13 +8,16 @@ import { resolveRegistryUpdateChannel } from "../infra/update-channels.js"; import { CLAWHUB_INSTALL_ERROR_CODE } from "../plugins/clawhub-error-codes.js"; import { VERSION } from "../version.js"; import { + createTestInstalledPluginIndex, loadConfig, notifyGatewayPluginMetadataChanged, readConfigFileSnapshotForWrite, + readPersistedInstalledPluginIndex, refreshPluginRegistry, registerPluginsCli, replaceConfigFile, resetPluginsCliTestState, + restorePersistedInstalledPluginIndexIfCurrent, runPluginsCommand, runtimeErrors, runtimeLogs, @@ -23,7 +26,7 @@ import { updateNpmInstalledHookPacks, updateNpmInstalledPlugins, writeConfigFile, - writePersistedInstalledPluginIndexInstallRecords, + writePersistedInstalledPluginIndexInstallRecordsWithLease, } from "./plugins-cli-test-helpers.js"; const ORIGINAL_OPENCLAW_NIX_MODE = process.env.OPENCLAW_NIX_MODE; @@ -81,6 +84,17 @@ function expectRestartNoticeLogged() { ).toBe(true); } +function expectInstallRecordsWrittenWithLease(records: unknown, config: unknown) { + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledWith( + records, + expect.objectContaining({ + config, + filePath: expect.any(String), + lease: expect.anything(), + }), + ); +} + function expectSingleCallParams(mockFn: ReturnType) { expect(mockFn).toHaveBeenCalledTimes(1); const params = mockFn.mock.calls[0]?.[0] as Record | undefined; @@ -218,7 +232,7 @@ async function expectSkippedClawHubPluginUpdate(params: { await expect(runPluginsCommand(["plugins", "update", "demo"])).rejects.toThrow("__exit__:1"); - expect(writePersistedInstalledPluginIndexInstallRecords).not.toHaveBeenCalled(); + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).not.toHaveBeenCalled(); expect(runtimeLogs.at(-1)).toContain(params.expectedLog); } @@ -301,7 +315,7 @@ describe("plugins cli update", () => { expect(acquireLease).not.toHaveBeenCalled(); expect(writeConfigFile).not.toHaveBeenCalled(); expect(replaceConfigFile).not.toHaveBeenCalled(); - expect(writePersistedInstalledPluginIndexInstallRecords).not.toHaveBeenCalled(); + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).not.toHaveBeenCalled(); expect(refreshPluginRegistry).not.toHaveBeenCalled(); expect(runtimeLogs).toContain("Would update alpha: 1.0.0 -> 1.1.0."); } finally { @@ -591,9 +605,7 @@ describe("plugins cli update", () => { expect(runtimeErrors).toEqual([]); expect(updateNpmInstalledPlugins).toHaveBeenCalledOnce(); expect(updateNpmInstalledHookPacks).not.toHaveBeenCalled(); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith( - nextConfig.plugins?.installs, - ); + expectInstallRecordsWrittenWithLease(nextConfig.plugins?.installs, cfg); expect(writeConfigFile).not.toHaveBeenCalled(); }); @@ -632,7 +644,7 @@ describe("plugins cli update", () => { expect(runtimeErrors).toEqual([]); expect(updateNpmInstalledPlugins).toHaveBeenCalledOnce(); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith(pluginRecords); + expectInstallRecordsWrittenWithLease(pluginRecords, cfg); expect(writeConfigFile).not.toHaveBeenCalled(); }); @@ -671,7 +683,7 @@ describe("plugins cli update", () => { await runPluginsCommand(["plugins", "update", "brave"]); expect(runtimeErrors).toEqual([]); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith(nextRecords); + expectInstallRecordsWrittenWithLease(nextRecords, sourceCfg); expect(writeConfigFile).not.toHaveBeenCalled(); expect(replaceConfigFile).not.toHaveBeenCalled(); expect(refreshPluginRegistry).toHaveBeenCalledWith({ @@ -722,7 +734,11 @@ describe("plugins cli update", () => { await runPluginsCommand(["plugins", "update", "brave"]); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith(nextRecords); + expectInstallRecordsWrittenWithLease(nextRecords, { + plugins: { + load: { paths: [nextInstallPath, customPath] }, + }, + }); expect(replaceConfigFile).toHaveBeenCalledWith({ nextConfig: { plugins: { @@ -783,18 +799,24 @@ describe("plugins cli update", () => { .mockResolvedValueOnce(initialSnapshot) .mockResolvedValueOnce(changedSnapshot); const { previousRecords, nextRecords } = primeBravePluginRecordUpdate(cfg); + const previousPersistedIndex = createTestInstalledPluginIndex({ + policyHash: "previous-policy", + installRecords: previousRecords, + }); + readPersistedInstalledPluginIndex.mockResolvedValue(previousPersistedIndex); await expect(runPluginsCommand(["plugins", "update", "brave"])).rejects.toThrow( "config changed since last load", ); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith( - 1, - nextRecords, - ); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith( - 2, - previousRecords, + expectInstallRecordsWrittenWithLease(nextRecords, cfg); + expect(restorePersistedInstalledPluginIndexIfCurrent).toHaveBeenCalledWith( + previousPersistedIndex, + expect.any(Number), + expect.objectContaining({ + filePath: expect.any(String), + lease: expect.anything(), + }), ); expect(writeConfigFile).not.toHaveBeenCalled(); expect(replaceConfigFile).not.toHaveBeenCalled(); @@ -839,18 +861,24 @@ describe("plugins cli update", () => { .mockResolvedValueOnce(initialSnapshot) .mockResolvedValueOnce(changedSnapshot); const { previousRecords, nextRecords } = primeBravePluginRecordUpdate(cfg); + const previousPersistedIndex = createTestInstalledPluginIndex({ + policyHash: "previous-policy", + installRecords: previousRecords, + }); + readPersistedInstalledPluginIndex.mockResolvedValue(previousPersistedIndex); await expect(runPluginsCommand(["plugins", "update", "brave"])).rejects.toThrow( "included config changed since last load", ); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith( - 1, - nextRecords, - ); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith( - 2, - previousRecords, + expectInstallRecordsWrittenWithLease(nextRecords, cfg); + expect(restorePersistedInstalledPluginIndexIfCurrent).toHaveBeenCalledWith( + previousPersistedIndex, + expect.any(Number), + expect.objectContaining({ + filePath: expect.any(String), + lease: expect.anything(), + }), ); expect(writeConfigFile).not.toHaveBeenCalled(); expect(replaceConfigFile).not.toHaveBeenCalled(); @@ -888,18 +916,24 @@ describe("plugins cli update", () => { .mockResolvedValueOnce(initialSnapshot) .mockResolvedValueOnce(invalidSnapshot); const { previousRecords, nextRecords } = primeBravePluginRecordUpdate(cfg); + const previousPersistedIndex = createTestInstalledPluginIndex({ + policyHash: "previous-policy", + installRecords: previousRecords, + }); + readPersistedInstalledPluginIndex.mockResolvedValue(previousPersistedIndex); await expect(runPluginsCommand(["plugins", "update", "brave"])).rejects.toThrow( "invalid config for plugin brave", ); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith( - 1, - nextRecords, - ); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenNthCalledWith( - 2, - previousRecords, + expectInstallRecordsWrittenWithLease(nextRecords, cfg); + expect(restorePersistedInstalledPluginIndexIfCurrent).toHaveBeenCalledWith( + previousPersistedIndex, + expect.any(Number), + expect.objectContaining({ + filePath: expect.any(String), + lease: expect.anything(), + }), ); expect(writeConfigFile).not.toHaveBeenCalled(); expect(replaceConfigFile).not.toHaveBeenCalled(); @@ -1420,9 +1454,7 @@ describe("plugins cli update", () => { expect(updateParams.config).toEqual(runtimeConfig); expect(updateParams.pluginIds).toEqual(["alpha"]); expect(updateParams.dryRun).toBe(false); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith( - nextConfig.plugins?.installs, - ); + expectInstallRecordsWrittenWithLease(nextConfig.plugins?.installs, {}); expect(updateNpmInstalledHookPacks).not.toHaveBeenCalled(); expect(writeConfigFile).toHaveBeenCalledWith({}); expect(replaceConfigFile).toHaveBeenCalledWith({ @@ -1489,9 +1521,7 @@ describe("plugins cli update", () => { await expect(runPluginsCommand(["plugins", "update", "--all"])).rejects.toThrow("__exit__:1"); - expect(writePersistedInstalledPluginIndexInstallRecords).toHaveBeenCalledWith( - nextConfig.plugins?.installs, - ); + expectInstallRecordsWrittenWithLease(nextConfig.plugins?.installs, {}); expect(refreshPluginRegistry).toHaveBeenCalledWith({ config: {}, installRecords: nextConfig.plugins?.installs, diff --git a/src/cli/plugins-update-command.ts b/src/cli/plugins-update-command.ts index 57947393dcf5..ecbecb8ceb82 100644 --- a/src/cli/plugins-update-command.ts +++ b/src/cli/plugins-update-command.ts @@ -437,6 +437,7 @@ async function runPluginUpdateCommandUnlocked(params: RunPluginUpdateCommandPara await commitPluginInstallRecordsOnly({ previousInstallRecords: persistedPluginInstallRecords, nextInstallRecords: nextPluginInstallRecords, + nextConfig, verifyConfigFresh: async () => { await assertRecordsOnlyUpdateConfigFresh({ baseHash: sourceSnapshot?.snapshot.hash, diff --git a/src/cli/update-cli.test.ts b/src/cli/update-cli.test.ts index cbcddc5c77a1..2fd3fdddb21b 100644 --- a/src/cli/update-cli.test.ts +++ b/src/cli/update-cli.test.ts @@ -67,6 +67,16 @@ const updateNpmInstalledPlugins = vi.fn(); const loadInstalledPluginIndexInstallRecords = vi.fn( async (params: { config?: OpenClawConfig } = {}) => params.config?.plugins?.installs ?? {}, ); +const readPersistedInstalledPluginIndex = vi.fn(async () => null); +const restorePersistedInstalledPluginIndex = vi.fn(async () => undefined); +const restorePersistedInstalledPluginIndexIfCurrent = vi.fn< + typeof import("../plugins/installed-plugin-index-store.js").restorePersistedInstalledPluginIndexIfCurrent +>(async () => true); +const writePersistedInstalledPluginIndexInstallRecords = vi.fn(async () => undefined); +const writePersistedInstalledPluginIndexInstallRecordsWithLease = vi.fn(async () => ({ + previous: null, + revision: 1, +})); const checkShellCompletionStatus = vi.fn(); const ensureCompletionCacheExists = vi.fn(); const installCompletion = vi.fn(); @@ -292,7 +302,19 @@ vi.mock("../plugins/installed-plugin-index-records.js", async (importOriginal) = return { ...actual, loadInstalledPluginIndexInstallRecords, - writePersistedInstalledPluginIndexInstallRecords: vi.fn(async () => undefined), + writePersistedInstalledPluginIndexInstallRecords, + writePersistedInstalledPluginIndexInstallRecordsWithLease, + }; +}); + +vi.mock("../plugins/installed-plugin-index-store.js", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + readPersistedInstalledPluginIndex, + restorePersistedInstalledPluginIndex, + restorePersistedInstalledPluginIndexIfCurrent, }; }); @@ -1310,6 +1332,14 @@ describe("update-cli", () => { delete process.env[GATEWAY_SERVICE_RUNTIME_PID_ENV]; restartHealthTestControl.snapshot = undefined; vi.clearAllMocks(); + readPersistedInstalledPluginIndex.mockResolvedValue(null); + restorePersistedInstalledPluginIndex.mockResolvedValue(undefined); + restorePersistedInstalledPluginIndexIfCurrent.mockResolvedValue(true); + writePersistedInstalledPluginIndexInstallRecords.mockResolvedValue(undefined); + writePersistedInstalledPluginIndexInstallRecordsWithLease.mockResolvedValue({ + previous: null, + revision: 1, + }); resetRuntimeCapture(); spawn.mockImplementation(() => { const child = new EventEmitter() as EventEmitter & { @@ -1777,6 +1807,21 @@ describe("update-cli", () => { readPackageVersion.mockImplementation(async (pkgRoot: string) => pkgRoot === root ? "0.0.1" : "2026.5.28", ); + const preUpdateConfig = { + plugins: { + entries: { + msteams: { enabled: false }, + }, + }, + } as OpenClawConfig; + vi.mocked(readConfigFileSnapshot).mockResolvedValue({ + ...baseSnapshot, + parsed: preUpdateConfig, + sourceConfig: preUpdateConfig, + resolved: preUpdateConfig, + config: preUpdateConfig, + runtimeConfig: preUpdateConfig, + }); const pluginInstallRecords = { msteams: { source: "npm", @@ -1819,6 +1864,107 @@ describe("update-cli", () => { integrity: "sha512-newer", }, }); + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledWith( + capturedRecords, + { + config: preUpdateConfig, + lease: expect.anything(), + }, + ); + expect(restorePersistedInstalledPluginIndexIfCurrent).not.toHaveBeenCalled(); + }); + + it("restores the exact plugin index revision when post-core handoff fails", async () => { + const { root } = setupUpdatedRootRefresh(); + readPackageVersion.mockImplementation(async (pkgRoot: string) => + pkgRoot === root ? "0.0.1" : "2026.5.28", + ); + const previousPersistedIndex = { + policyHash: "previous-policy", + installRecords: { + msteams: { + source: "npm", + spec: "@openclaw/msteams", + resolvedVersion: "1.0.0", + }, + } satisfies Record, + }; + writePersistedInstalledPluginIndexInstallRecordsWithLease.mockResolvedValue({ + previous: previousPersistedIndex as never, + revision: 17, + }); + loadInstalledPluginIndexInstallRecords.mockResolvedValueOnce( + previousPersistedIndex.installRecords, + ); + spawn.mockImplementationOnce(() => { + throw new Error("post-core spawn failed"); + }); + + await expect(updateCommand({ yes: true, restart: false })).rejects.toThrow( + "post-core spawn failed", + ); + + expect(writePersistedInstalledPluginIndexInstallRecordsWithLease).toHaveBeenCalledTimes(1); + expect(restorePersistedInstalledPluginIndexIfCurrent).toHaveBeenCalledWith( + previousPersistedIndex, + 17, + { lease: expect.anything() }, + ); + }); + + it("keeps a child-committed plugin index when the post-core handoff is signaled", async () => { + const { root } = setupUpdatedRootRefresh(); + readPackageVersion.mockImplementation(async (pkgRoot: string) => + pkgRoot === root ? "0.0.1" : "2026.5.28", + ); + const previousPersistedIndex = { + policyHash: "previous-policy", + installRecords: { + msteams: { + source: "npm", + spec: "@openclaw/msteams", + resolvedVersion: "1.0.0", + }, + } satisfies Record, + }; + let currentRevision = 17; + writePersistedInstalledPluginIndexInstallRecordsWithLease.mockResolvedValue({ + previous: previousPersistedIndex as never, + revision: currentRevision, + }); + restorePersistedInstalledPluginIndexIfCurrent.mockImplementation( + async (_index, expectedRevision) => { + if (currentRevision !== expectedRevision) { + return false; + } + currentRevision += 1; + return true; + }, + ); + loadInstalledPluginIndexInstallRecords.mockResolvedValueOnce( + previousPersistedIndex.installRecords, + ); + spawn.mockImplementationOnce(() => { + const child = new EventEmitter() as EventEmitter & { + once: EventEmitter["once"]; + }; + currentRevision = 18; + queueMicrotask(() => { + child.emit("exit", null, "SIGTERM"); + }); + return child; + }); + + await expect(updateCommand({ yes: true, restart: false })).rejects.toThrow( + "post-update process terminated by signal SIGTERM", + ); + + expect(restorePersistedInstalledPluginIndexIfCurrent).toHaveBeenCalledWith( + previousPersistedIndex, + 17, + { lease: expect.anything() }, + ); + expect(currentRevision).toBe(18); }); it("respawns into the updated git root before requested channel persistence", async () => { diff --git a/src/cli/update-cli/update-command-post-core.ts b/src/cli/update-cli/update-command-post-core.ts index 7f65ba7217b9..5a426fde894f 100644 --- a/src/cli/update-cli/update-command-post-core.ts +++ b/src/cli/update-cli/update-command-post-core.ts @@ -43,8 +43,9 @@ import type { UpdateRunResult } from "../../infra/update-runner.js"; import { getWindowsSystem32ExePath } from "../../infra/windows-install-roots.js"; import { loadInstalledPluginIndexInstallRecords, - writePersistedInstalledPluginIndexInstallRecords, + writePersistedInstalledPluginIndexInstallRecordsWithLease, } from "../../plugins/installed-plugin-index-records.js"; +import { restorePersistedInstalledPluginIndexIfCurrent } from "../../plugins/installed-plugin-index-store.js"; import { withPluginLifecycleLease } from "../../plugins/plugin-lifecycle-lease.js"; import { runExec } from "../../process/exec.js"; import { defaultRuntime } from "../../runtime.js"; @@ -513,10 +514,33 @@ export async function continuePostCoreUpdateInFreshProcess(params: { records: params.pluginInstallRecords, targetVersion: postCoreHostVersion, }); + let tentativePluginIndex: + | Awaited> + | undefined; + const restoreTentativePluginIndex = async () => { + const tentative = tentativePluginIndex; + if (!tentative) { + return; + } + await withPluginLifecycleLease({}, async (lease) => { + await restorePersistedInstalledPluginIndexIfCurrent(tentative.previous, tentative.revision, { + lease, + }); + }); + tentativePluginIndex = undefined; + }; try { if (pluginInstallRecords && pluginInstallRecords !== params.pluginInstallRecords) { - await writePersistedInstalledPluginIndexInstallRecords(pluginInstallRecords); + await withPluginLifecycleLease({}, async (lease) => { + tentativePluginIndex = await writePersistedInstalledPluginIndexInstallRecordsWithLease( + pluginInstallRecords, + { + ...(params.preUpdateConfig ? { config: params.preUpdateConfig.sourceConfig } : {}), + lease, + }, + ); + }); } await writePostCorePluginInstallRecordsFile(installRecordsPath, pluginInstallRecords); await writePostCoreSourceConfigFile(sourceConfigPath, params.preUpdateConfig); @@ -608,9 +632,19 @@ export async function continuePostCoreUpdateInFreshProcess(params: { if (pluginUpdate) { return { resumed: true, pluginUpdate }; } + await restoreTentativePluginIndex(); return { resumed: false, exitCode }; } return { resumed: true, ...(pluginUpdate ? { pluginUpdate } : {}) }; + } catch (error) { + try { + await restoreTentativePluginIndex(); + } catch (rollbackError) { + throw new Error("Post-core update failed and could not restore the previous plugin index", { + cause: rollbackError, + }); + } + throw error; } finally { await fs.rm(resultDir, { recursive: true, force: true }).catch(() => undefined); } diff --git a/src/commands/agents.commands.add.ts b/src/commands/agents.commands.add.ts index a0537c4dfcf6..41be31f6dc67 100644 --- a/src/commands/agents.commands.add.ts +++ b/src/commands/agents.commands.add.ts @@ -24,6 +24,7 @@ import { commitConfigWithPendingPluginInstalls, transformConfigWithPendingPluginInstalls, } from "../plugins/install-record-commit.js"; +import { withPluginLifecycleLease } from "../plugins/plugin-lifecycle-lease.js"; import { LEGACY_IMPLICIT_AGENT_ID, normalizeAgentId } from "../routing/session-key.js"; import { type RuntimeEnv, writeRuntimeJson } from "../runtime.js"; import { defaultRuntime } from "../runtime.js"; @@ -129,13 +130,15 @@ export async function agentsAddCommand( runtime.log(`Normalized agent id to "${agentId}".`); } - const created = await createAgent({ - name: nameInput, - workspace: workspaceFlag, - ...(opts.agentDir ? { agentDir: opts.agentDir } : {}), - ...(opts.model ? { model: opts.model } : {}), - ...(opts.bind?.length ? { bindingSpecs: opts.bind } : {}), - transformConfig: transformConfigWithPendingPluginInstalls, + const created = await withPluginLifecycleLease({}, async () => { + return await createAgent({ + name: nameInput, + workspace: workspaceFlag, + ...(opts.agentDir ? { agentDir: opts.agentDir } : {}), + ...(opts.model ? { model: opts.model } : {}), + ...(opts.bind?.length ? { bindingSpecs: opts.bind } : {}), + transformConfig: transformConfigWithPendingPluginInstalls, + }); }); if (created.status === "error") { runtime.error( diff --git a/src/plugins/install-persistence.test.ts b/src/plugins/install-persistence.test.ts index fe3f34075ddd..4de243ab3b7d 100644 --- a/src/plugins/install-persistence.test.ts +++ b/src/plugins/install-persistence.test.ts @@ -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", diff --git a/src/plugins/install-record-commit.sqlite.test.ts b/src/plugins/install-record-commit.sqlite.test.ts new file mode 100644 index 000000000000..2a882fab6e4a --- /dev/null +++ b/src/plugins/install-record-commit.sqlite.test.ts @@ -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 { + 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 { + 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 { + 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 { + 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 | 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({})); + }); + }); +}); diff --git a/src/plugins/install-record-commit.test.ts b/src/plugins/install-record-commit.test.ts index a36a73243233..1d10cee072da 100644 --- a/src/plugins/install-record-commit.test.ts +++ b/src/plugins/install-record-commit.test.ts @@ -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) => + 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(); + 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; +}): 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 = { + 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, }); diff --git a/src/plugins/install-record-commit.ts b/src/plugins/install-record-commit.ts index b5fe1cdf2a1e..1fae6acef00b 100644 --- a/src/plugins/install-record-commit.ts +++ b/src/plugins/install-record-commit.ts @@ -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; - nextInstallRecords: Record; + prepareInstallRecords: (storeOptions: InstalledPluginIndexRecordStoreOptions) => Promise<{ + previousInstallRecords: Record; + nextInstallRecords: Record; + }>; nextConfig: OpenClawConfig; writeOptions?: ConfigWriteOptions; commit: ConfigCommit; -}): Promise { - 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; +}> { + 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 { 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; nextInstallRecords: Record; + nextConfig: OpenClawConfig; verifyConfigFresh?: () => Promise; }): Promise { 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( }; }; - return await transformConfigFileWithRetry({ - ...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({ + ...params, + commit, + }); }); } diff --git a/src/plugins/installed-plugin-index-records.ts b/src/plugins/installed-plugin-index-records.ts index 1a8ef64fdda4..63d486d49bdd 100644 --- a/src/plugins/installed-plugin-index-records.ts +++ b/src/plugins/installed-plugin-index-records.ts @@ -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, + options: InstalledPluginIndexRecordRefreshOptions & { + lease: InstalledPluginIndexWriteLease; + }, +): Promise { + return refreshPersistedInstalledPluginIndexWithLeaseSync({ + ...options, + reason: "source-changed", + installRecords: records, + }); +} + /** Refreshes persisted installed plugin index records synchronously. */ export function writePersistedInstalledPluginIndexInstallRecordsSync( records: Record, diff --git a/src/plugins/installed-plugin-index-store.test.ts b/src/plugins/installed-plugin-index-store.test.ts index f1b0f0b132ca..18ebae13d451 100644 --- a/src/plugins/installed-plugin-index-store.test.ts +++ b/src/plugins/installed-plugin-index-store.test.ts @@ -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"); diff --git a/src/plugins/installed-plugin-index-store.ts b/src/plugins/installed-plugin-index-store.ts index 9892eca6dd05..16c1583e34df 100644 --- a/src/plugins/installed-plugin-index-store.ts +++ b/src/plugins/installed-plugin-index-store.ts @@ -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 { @@ -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 { + 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; +} diff --git a/src/plugins/plugin-lifecycle-lease.test.ts b/src/plugins/plugin-lifecycle-lease.test.ts index c0292492fdd6..abe9757f28ce 100644 --- a/src/plugins/plugin-lifecycle-lease.test.ts +++ b/src/plugins/plugin-lifecycle-lease.test.ts @@ -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"]); }); }); diff --git a/src/plugins/plugin-lifecycle-lease.ts b/src/plugins/plugin-lifecycle-lease.ts index ae6a4a630199..a28c0a564afe 100644 --- a/src/plugins/plugin-lifecycle-lease.ts +++ b/src/plugins/plugin-lifecycle-lease.ts @@ -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( options: PluginLifecycleLeaseOptions, - run: (lease: OpenClawStateLeaseContext) => Promise, + run: (lease: PluginLifecycleLeaseContext) => Promise, ): Promise { + 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( 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), ); }, );