From aa8fbb7f3c20d6bc080c00a03fe6ac366a2f935d Mon Sep 17 00:00:00 2001 From: Benjamin Badejo Date: Thu, 30 Jul 2026 19:45:26 +0300 Subject: [PATCH] fix(matrix): reserve dedupe migration capacity --- .../doctor-contract-api.capacity.test.ts | 183 ++++++++++++++++++ extensions/matrix/doctor-contract-api.test.ts | 16 +- extensions/matrix/doctor-contract-api.ts | 9 + .../monitor/inbound-dedupe-migration.ts | 26 ++- src/plugin-sdk/plugin-state-test-runtime.ts | 1 + 5 files changed, 229 insertions(+), 6 deletions(-) create mode 100644 extensions/matrix/doctor-contract-api.capacity.test.ts diff --git a/extensions/matrix/doctor-contract-api.capacity.test.ts b/extensions/matrix/doctor-contract-api.capacity.test.ts new file mode 100644 index 000000000000..d7821b60c9e7 --- /dev/null +++ b/extensions/matrix/doctor-contract-api.capacity.test.ts @@ -0,0 +1,183 @@ +// Matrix tests cover plugin-wide capacity during inbound dedupe migration. +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { + createPersistentDedupeImportEntry, + type PersistentDedupeEntry, +} from "openclaw/plugin-sdk/persistent-dedupe"; +import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime"; +import { + createPluginStateKeyedStoreForTests, + getPluginStateCapacityForTests, + importPluginStateEntriesForDoctorForTests, + resetPluginStateStoreForTests, + setMaxPluginStateEntriesPerPluginForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import type { PluginDoctorStateMigrationContext } from "openclaw/plugin-sdk/runtime-doctor"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { stateMigrations } from "./doctor-contract-api.js"; +import { + MATRIX_INBOUND_DEDUPE_TTL_MS, + resolveMatrixInboundDedupeStateNamespace, +} from "./src/matrix/monitor/inbound-dedupe.js"; + +function createMigrationParams(stateDir: string) { + const env = { OPENCLAW_STATE_DIR: stateDir }; + const context: PluginDoctorStateMigrationContext = { + getPluginStateCapacity() { + return getPluginStateCapacityForTests("matrix", env); + }, + importPluginStateEntries(options, entries) { + importPluginStateEntriesForDoctorForTests("matrix", options, entries); + }, + openPluginStateKeyedStore(options: OpenKeyedStoreOptions) { + return createPluginStateKeyedStoreForTests("matrix", options); + }, + }; + return { + config: {} as OpenClawConfig, + env, + stateDir, + oauthDir: path.join(stateDir, "oauth"), + context, + }; +} + +function getMigration() { + const migration = stateMigrations.find( + (entry) => entry.id === "matrix-inbound-dedupe-to-claimable-dedupe", + ); + if (!migration) { + throw new Error("missing Matrix inbound dedupe migration"); + } + return migration; +} + +function writeLegacyDedupeSource(stateDir: string, now: number, withMetadata = false) { + const root = path.join( + stateDir, + "matrix", + "accounts", + "home", + "matrix.example.org__bot", + "token-a", + ); + fs.mkdirSync(root, { recursive: true }); + const jsonPath = path.join(root, "inbound-dedupe.json"); + fs.writeFileSync( + jsonPath, + JSON.stringify({ + version: 1, + entries: [{ key: "!room:example.org|$legacy", ts: now - 60_000 }], + }), + ); + if (withMetadata) { + fs.writeFileSync( + path.join(root, "storage-meta.json"), + JSON.stringify({ accountId: "home", userId: "@home:example.org" }), + ); + } + return jsonPath; +} + +describe("matrix inbound dedupe migration capacity", () => { + const tempDirs: string[] = []; + + beforeEach(() => { + resetPluginStateStoreForTests(); + }); + + afterEach(() => { + setMaxPluginStateEntriesPerPluginForTests(undefined); + resetPluginStateStoreForTests(); + for (const dir of tempDirs.splice(0)) { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); + + it("keeps sources when completion capacity cannot be reserved", async () => { + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-capacity-")); + tempDirs.push(stateDir); + const jsonPath = writeLegacyDedupeSource(stateDir, Date.now()); + const params = createMigrationParams(stateDir); + setMaxPluginStateEntriesPerPluginForTests(3); + const dedupeStore = params.context.openPluginStateKeyedStore({ + namespace: resolveMatrixInboundDedupeStateNamespace(), + maxEntries: 20_000, + defaultTtlMs: MATRIX_INBOUND_DEDUPE_TTL_MS, + env: params.env, + }); + const canonicalEntry = createPersistentDedupeImportEntry({ + key: "ops\0!room:example.org\0$runtime", + seenAt: Date.now(), + }); + await dedupeStore.register(canonicalEntry.key, canonicalEntry.value); + const siblingStore = params.context.openPluginStateKeyedStore<{ value: number }>({ + namespace: "capacity-sibling", + maxEntries: 10, + env: params.env, + }); + await siblingStore.register("one", { value: 1 }); + await siblingStore.register("two", { value: 2 }); + + const result = await getMigration().migrateLegacyState(params); + + expect(result.changes).toEqual([]); + expect(result.warnings).toEqual([ + expect.stringContaining("Failed reserving Matrix inbound dedupe migration completion:"), + ]); + expect(fs.existsSync(jsonPath)).toBe(true); + expect(fs.existsSync(`${jsonPath}.migrated`)).toBe(false); + await expect(dedupeStore.lookup(canonicalEntry.key)).resolves.toEqual(canonicalEntry.value); + await expect(siblingStore.entries()).resolves.toHaveLength(2); + }); + + it("reserves completion capacity before bounded import", async () => { + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-capacity-")); + tempDirs.push(stateDir); + const now = Date.now(); + const jsonPath = writeLegacyDedupeSource(stateDir, now, true); + const params = createMigrationParams(stateDir); + setMaxPluginStateEntriesPerPluginForTests(5); + const dedupeStore = params.context.openPluginStateKeyedStore({ + namespace: resolveMatrixInboundDedupeStateNamespace(), + maxEntries: 20_000, + defaultTtlMs: MATRIX_INBOUND_DEDUPE_TTL_MS, + env: params.env, + }); + const canonicalEntry = createPersistentDedupeImportEntry({ + key: "ops\0!room:example.org\0$runtime", + seenAt: now, + }); + await dedupeStore.register(canonicalEntry.key, canonicalEntry.value); + const siblingStore = params.context.openPluginStateKeyedStore<{ value: number }>({ + namespace: "capacity-sibling", + maxEntries: 10, + env: params.env, + }); + await siblingStore.register("one", { value: 1 }); + await siblingStore.register("two", { value: 2 }); + + await expect(getMigration().migrateLegacyState(params)).resolves.toEqual({ + changes: [ + "Migrated Matrix inbound dedupe markers to the claimable dedupe store (1 of 1 entries)", + `Archived Matrix inbound dedupe legacy source -> ${jsonPath}.migrated`, + "Recorded Matrix inbound dedupe migration completion (0 SQLite roots, 1 JSON roots scanned)", + ], + warnings: [], + }); + await expect(dedupeStore.lookup(canonicalEntry.key)).resolves.toEqual(canonicalEntry.value); + const legacyEntry = createPersistentDedupeImportEntry({ + key: "home\0!room:example.org\0$legacy", + seenAt: now - 60_000, + }); + await expect(dedupeStore.lookup(legacyEntry.key)).resolves.toEqual(legacyEntry.value); + expect(getPluginStateCapacityForTests("matrix", params.env)).toEqual({ + liveEntries: 5, + maxEntries: 5, + }); + await expect(getMigration().detectLegacyState(params)).resolves.toBeNull(); + }); +}); diff --git a/extensions/matrix/doctor-contract-api.test.ts b/extensions/matrix/doctor-contract-api.test.ts index f2e89aa2bb13..ddf6dc5a5c28 100644 --- a/extensions/matrix/doctor-contract-api.test.ts +++ b/extensions/matrix/doctor-contract-api.test.ts @@ -17,6 +17,7 @@ import type { } from "openclaw/plugin-sdk/plugin-state-runtime"; import { createPluginStateKeyedStoreForTests, + getPluginStateCapacityForTests, importPluginStateEntriesForDoctorForTests, resetPluginStateStoreForTests, } from "openclaw/plugin-sdk/plugin-state-test-runtime"; @@ -57,8 +58,11 @@ import { installMatrixTestRuntime } from "./src/test-runtime.js"; const DOCTOR_IDB_DATABASE_PREFIX = "openclaw-matrix-doctor-test"; -function createContext(): PluginDoctorStateMigrationContext { +function createContext(env?: NodeJS.ProcessEnv): PluginDoctorStateMigrationContext { return { + getPluginStateCapacity() { + return getPluginStateCapacityForTests("matrix", env); + }, importPluginStateEntries(options, entries) { importPluginStateEntriesForDoctorForTests("matrix", options, entries); }, @@ -68,12 +72,13 @@ function createContext(): PluginDoctorStateMigrationContext { } function createMigrationParams(stateDir: string) { + const env = { OPENCLAW_STATE_DIR: stateDir }; return { config: {} as OpenClawConfig, - env: { OPENCLAW_STATE_DIR: stateDir }, + env, stateDir, oauthDir: path.join(stateDir, "oauth"), - context: createContext(), + context: createContext(env), }; } @@ -962,7 +967,8 @@ describe("matrix doctor contract state migrations", () => { it("keeps newer runtime dedupe rows when legacy imports hit capacity", async () => { const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-doctor-")); tempDirs.push(stateDir); - const io = { context: createContext(), env: { OPENCLAW_STATE_DIR: stateDir } }; + const env = { OPENCLAW_STATE_DIR: stateDir }; + const io = { context: createContext(env), env }; const roomId = "!room:example.org"; const now = Date.now(); const store = createPluginStateKeyedStoreForTests("matrix", { @@ -1011,7 +1017,7 @@ describe("matrix doctor contract state migrations", () => { const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-doctor-")); tempDirs.push(stateDir); const env = { OPENCLAW_STATE_DIR: stateDir }; - const io = { context: createContext(), env }; + const io = { context: createContext(env), env }; const now = 2_000_000_000_000; const remainingTtlMs = 1_000; const roomId = "!room:example.org"; diff --git a/extensions/matrix/doctor-contract-api.ts b/extensions/matrix/doctor-contract-api.ts index 70ac22edba1e..aadfe1570a0c 100644 --- a/extensions/matrix/doctor-contract-api.ts +++ b/extensions/matrix/doctor-contract-api.ts @@ -58,6 +58,7 @@ import { readLegacyInboundDedupeJsonSource, readLegacyInboundDedupeSqliteSource, recordMatrixInboundDedupeMigrationCompletion, + reserveMatrixInboundDedupeMigrationCompletion, retireLegacyInboundDedupeSqliteRows, verifyMatrixInboundDedupeSourcesRetired, type LegacyInboundDedupeMarker, @@ -320,6 +321,14 @@ export const stateMigrations: PluginDoctorStateMigration[] = [ if (await hasCompletedMatrixInboundDedupeMigration(params.context, params.env)) { return { changes, warnings }; } + try { + await reserveMatrixInboundDedupeMigrationCompletion(params.context, params.env); + } catch (err) { + warnings.push( + `Failed reserving Matrix inbound dedupe migration completion: ${String(err)}; left legacy sources in place`, + ); + return { changes, warnings }; + } const sources = await collectMatrixInboundDedupeSources(params.stateDir); if (sources.status === "incomplete") { warnings.push(...sources.warnings); diff --git a/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts b/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts index e599366a5a48..5f39b77f27d6 100644 --- a/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts +++ b/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts @@ -113,6 +113,21 @@ export async function recordMatrixInboundDedupeMigrationCompletion( ); } +/** + * Reserves the durable completion row before any legacy source is changed. + * The invalid timestamp keeps detection active after an interrupted run, while + * updating this same key after retirement remains possible at plugin capacity. + */ +export async function reserveMatrixInboundDedupeMigrationCompletion( + context: PluginDoctorStateMigrationContext, + env: NodeJS.ProcessEnv, +): Promise { + await openMatrixInboundDedupeMigrationCompletionStore(context, env).register( + MIGRATION_COMPLETION_KEY, + { version: 1, completedAt: -1 }, + ); +} + function loadNodeSqlite(): typeof import("node:sqlite") { const req = createRequire(import.meta.url); return req("node:sqlite") as typeof import("node:sqlite"); @@ -431,7 +446,16 @@ export async function importNewestInboundDedupeMarkers(params: { }); return existingKeys.has(entry.key) ? [] : [{ marker, entry }]; }); - const capacity = Math.max(0, stateMaxEntries - existingEntries.length); + const namespaceCapacity = Math.max(0, stateMaxEntries - existingEntries.length); + const pluginCapacity = + missingEntries.length > 0 ? params.io.context.getPluginStateCapacity?.() : undefined; + if (missingEntries.length > 0 && !pluginCapacity) { + throw new Error("plugin-wide Matrix inbound dedupe import capacity is unavailable"); + } + const pluginRemainingCapacity = pluginCapacity + ? Math.max(0, pluginCapacity.maxEntries - pluginCapacity.liveEntries) + : namespaceCapacity; + const capacity = Math.min(namespaceCapacity, pluginRemainingCapacity); const selectedEntries = missingEntries.slice(0, capacity); if (selectedEntries.length > 0) { const importEntries = params.io.context.importPluginStateEntries; diff --git a/src/plugin-sdk/plugin-state-test-runtime.ts b/src/plugin-sdk/plugin-state-test-runtime.ts index 69949122f6c9..54fce1380c8d 100644 --- a/src/plugin-sdk/plugin-state-test-runtime.ts +++ b/src/plugin-sdk/plugin-state-test-runtime.ts @@ -8,6 +8,7 @@ export { importPluginStateEntriesForDoctor as importPluginStateEntriesForDoctorForTests, resetPluginStateStoreForTests, } from "../plugin-state/plugin-state-store.js"; +export { setMaxPluginStateEntriesPerPluginForTests } from "../plugin-state/plugin-state-store.test-helpers.js"; export { setMaxMemoryHostEventsForTests } from "../memory-host-sdk/event-store.js"; export { createPluginBlobStoreForTests,