mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-21 10:16:44 +00:00
refactor(matrix): adopt core claimable dedupe with doctor-owned state migration (#104391)
* refactor(matrix): adopt core claimable dedupe for inbound events * fix(matrix): migrate legacy inbound dedupe markers via doctor * fix(matrix): single-pool dedupe namespace and capacity-aware legacy import * fix(matrix): keep unreadable legacy dedupe JSON in place during doctor import
This commit is contained in:
Peter Steinberger
@@ -1303,7 +1303,11 @@ sessionId})`; create, branch, continue, list, and fork flows live in their
|
||||
credentials, and recovery keys now use shared SQLite plugin state/blob
|
||||
tables. Runtime path structs no longer expose a `storage-meta.json` metadata
|
||||
path; that filename is a legacy migration input only. Their legacy JSON import
|
||||
plan lives in the Matrix plugin setup/doctor migration surface.
|
||||
plan lives in the Matrix plugin setup/doctor migration surface. Inbound
|
||||
dedupe markers ride the core claimable dedupe (`matrix.inbound-dedupe.*`
|
||||
namespaces in the shared state DB); the Matrix doctor state migration imports
|
||||
the retired per-root `inbound-dedupe` rows and `inbound-dedupe.json` once,
|
||||
then the runtime reads only the claimable-dedupe store.
|
||||
- Matrix startup no longer scans, reports, or completes legacy Matrix file
|
||||
state. Matrix file detection, legacy crypto snapshot creation, room-key
|
||||
restore migration state, import, and source removal are all doctor-owned.
|
||||
@@ -1615,13 +1619,14 @@ Move these into the global database:
|
||||
- Matrix sync cache, storage metadata, thread bindings, inbound dedupe markers,
|
||||
startup verification cooldown state, credentials, recovery keys, and SDK
|
||||
IndexedDB crypto snapshots now use SQLite plugin state/blob namespaces under
|
||||
`matrix` (`sync-store`, `storage-meta`, `thread-bindings`, `inbound-dedupe`,
|
||||
`matrix` (`sync-store`, `storage-meta`, `thread-bindings`,
|
||||
`matrix.inbound-dedupe.*` via the core claimable dedupe,
|
||||
`startup-verification`, `credentials`, `recovery-key`, `idb-snapshots`)
|
||||
instead of `bot-storage.json`, `storage-meta.json`, `thread-bindings.json`,
|
||||
`inbound-dedupe.json`, `startup-verification.json`, `credentials.json`,
|
||||
`recovery-key.json`, and `crypto-idb-snapshot.json`; the Matrix doctor/setup
|
||||
migration imports and removes those legacy files from account-scoped Matrix
|
||||
storage roots.
|
||||
migration imports and removes those legacy files (and the retired per-root
|
||||
`inbound-dedupe` SQLite rows) from account-scoped Matrix storage roots.
|
||||
- Nostr bus cursors and profile publish state now use SQLite plugin state under
|
||||
`nostr` namespaces (`bus-state`, `profile-state`) instead of
|
||||
`bus-state-*.json` and `profile-state-*.json`; the Nostr doctor/setup
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
// Matrix tests cover doctor contract state migrations.
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
@@ -21,6 +22,12 @@ import {
|
||||
readMatrixLegacyCryptoMigrationState,
|
||||
readMatrixRecoveryKeyState,
|
||||
} from "./src/matrix/crypto-state-store.js";
|
||||
import { importNewestInboundDedupeMarkers } from "./src/matrix/monitor/inbound-dedupe-migration.js";
|
||||
import {
|
||||
createMatrixInboundEventDeduper,
|
||||
MATRIX_INBOUND_DEDUPE_TTL_MS,
|
||||
resolveMatrixInboundDedupeStateNamespace,
|
||||
} from "./src/matrix/monitor/inbound-dedupe.js";
|
||||
import { installMatrixTestRuntime } from "./src/test-runtime.js";
|
||||
|
||||
function createContext(): PluginDoctorStateMigrationContext {
|
||||
@@ -316,4 +323,188 @@ describe("matrix doctor contract state migrations", () => {
|
||||
expect(readMatrixLegacyCryptoMigrationState(storageRootDir)?.restoreStatus).toBe("pending");
|
||||
expect(fs.existsSync(path.join(storageRootDir, "legacy-crypto-migration.json"))).toBe(false);
|
||||
});
|
||||
|
||||
it("migrates legacy inbound dedupe markers into the claimable dedupe store", async () => {
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-doctor-"));
|
||||
tempDirs.push(stateDir);
|
||||
const sqliteRoot = path.join(
|
||||
stateDir,
|
||||
"matrix",
|
||||
"accounts",
|
||||
"ops",
|
||||
"matrix.example.org__bot",
|
||||
"token-a",
|
||||
);
|
||||
const jsonRoot = path.join(
|
||||
stateDir,
|
||||
"matrix",
|
||||
"accounts",
|
||||
"home",
|
||||
"matrix.example.org__bot",
|
||||
"token-b",
|
||||
);
|
||||
fs.mkdirSync(sqliteRoot, { recursive: true });
|
||||
fs.mkdirSync(jsonRoot, { recursive: true });
|
||||
const roomId = "!room:example.org";
|
||||
const now = Date.now();
|
||||
const legacyKey = (accountId: string, eventId: string) =>
|
||||
`${accountId}:${createHash("sha256")
|
||||
.update(accountId)
|
||||
.update("\0")
|
||||
.update(roomId)
|
||||
.update("\0")
|
||||
.update(eventId)
|
||||
.digest("hex")}`;
|
||||
|
||||
// >=2026.6 shape: per-storage-root SQLite rows plus JSON-import markers.
|
||||
const legacyStore = createPluginStateKeyedStoreForTests<{
|
||||
roomId: string;
|
||||
eventId: string;
|
||||
ts: number;
|
||||
}>("matrix", {
|
||||
namespace: "inbound-dedupe",
|
||||
maxEntries: 20_000,
|
||||
env: { OPENCLAW_STATE_DIR: sqliteRoot },
|
||||
});
|
||||
await legacyStore.register(legacyKey("ops", "$committed"), {
|
||||
roomId,
|
||||
eventId: "$committed",
|
||||
ts: now - 60_000,
|
||||
});
|
||||
await legacyStore.register(legacyKey("ops", "$expired"), {
|
||||
roomId,
|
||||
eventId: "$expired",
|
||||
ts: now - 31 * 24 * 60 * 60 * 1000,
|
||||
});
|
||||
const legacyMarkersStore = createPluginStateKeyedStoreForTests<{ importedAt: number }>(
|
||||
"matrix",
|
||||
{
|
||||
namespace: "inbound-dedupe-migrations",
|
||||
maxEntries: 1_000,
|
||||
env: { OPENCLAW_STATE_DIR: sqliteRoot },
|
||||
},
|
||||
);
|
||||
await legacyMarkersStore.register("ops:legacy-json-marker", { importedAt: now });
|
||||
|
||||
// <=2026.5 shape: raw inbound-dedupe.json plus storage-meta.json identity.
|
||||
fs.writeFileSync(
|
||||
path.join(jsonRoot, "inbound-dedupe.json"),
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
entries: [{ key: `${roomId}|$json-committed`, ts: now - 60_000 }],
|
||||
}),
|
||||
);
|
||||
fs.writeFileSync(
|
||||
path.join(jsonRoot, "storage-meta.json"),
|
||||
JSON.stringify({ accountId: "home", userId: "@home:example.org" }),
|
||||
);
|
||||
|
||||
const migration = migrationById("matrix-inbound-dedupe-to-claimable-dedupe");
|
||||
await expect(migration.detectLegacyState(createMigrationParams(stateDir))).resolves.toEqual({
|
||||
preview: [
|
||||
`Matrix inbound dedupe rows can migrate to the claimable dedupe store: ${sqliteRoot}`,
|
||||
`Matrix inbound dedupe JSON can migrate to the claimable dedupe store: ${path.join(jsonRoot, "inbound-dedupe.json")}`,
|
||||
],
|
||||
});
|
||||
|
||||
await expect(migration.migrateLegacyState(createMigrationParams(stateDir))).resolves.toEqual({
|
||||
changes: [
|
||||
"Migrated Matrix inbound dedupe markers to the claimable dedupe store (2 of 3 entries)",
|
||||
`Retired Matrix inbound dedupe rows for ${sqliteRoot}`,
|
||||
`Archived Matrix inbound dedupe legacy source -> ${path.join(jsonRoot, "inbound-dedupe.json")}.migrated`,
|
||||
],
|
||||
warnings: [],
|
||||
});
|
||||
|
||||
// Pre-upgrade markers must keep deduping through the new runtime guard.
|
||||
const dedupeEnv = { ...process.env, OPENCLAW_STATE_DIR: stateDir };
|
||||
const opsDeduper = createMatrixInboundEventDeduper({
|
||||
auth: { accountId: "ops" },
|
||||
env: dedupeEnv,
|
||||
});
|
||||
await expect(opsDeduper.claimEvent({ roomId, eventId: "$committed" })).resolves.toBe(false);
|
||||
await expect(opsDeduper.claimEvent({ roomId, eventId: "$expired" })).resolves.toBe(true);
|
||||
const homeDeduper = createMatrixInboundEventDeduper({
|
||||
auth: { accountId: "home" },
|
||||
env: dedupeEnv,
|
||||
});
|
||||
await expect(homeDeduper.claimEvent({ roomId, eventId: "$json-committed" })).resolves.toBe(
|
||||
false,
|
||||
);
|
||||
|
||||
// Legacy sources are retired and the migration is idempotent.
|
||||
await expect(legacyStore.entries()).resolves.toEqual([]);
|
||||
await expect(legacyMarkersStore.entries()).resolves.toEqual([]);
|
||||
expect(fs.existsSync(path.join(jsonRoot, "inbound-dedupe.json"))).toBe(false);
|
||||
expect(fs.existsSync(path.join(jsonRoot, "inbound-dedupe.json.migrated"))).toBe(true);
|
||||
await expect(migration.detectLegacyState(createMigrationParams(stateDir))).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it("archives malformed inbound dedupe JSON without importing it", async () => {
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-doctor-"));
|
||||
tempDirs.push(stateDir);
|
||||
const jsonRoot = path.join(
|
||||
stateDir,
|
||||
"matrix",
|
||||
"accounts",
|
||||
"home",
|
||||
"matrix.example.org__bot",
|
||||
"token-a",
|
||||
);
|
||||
fs.mkdirSync(jsonRoot, { recursive: true });
|
||||
const jsonPath = path.join(jsonRoot, "inbound-dedupe.json");
|
||||
fs.writeFileSync(jsonPath, "not-json");
|
||||
|
||||
const migration = migrationById("matrix-inbound-dedupe-to-claimable-dedupe");
|
||||
await expect(migration.migrateLegacyState(createMigrationParams(stateDir))).resolves.toEqual({
|
||||
changes: [
|
||||
"Migrated Matrix inbound dedupe markers to the claimable dedupe store (0 of 0 entries)",
|
||||
`Archived Matrix inbound dedupe legacy source -> ${jsonPath}.migrated`,
|
||||
],
|
||||
warnings: [
|
||||
`Matrix inbound dedupe JSON for ${jsonRoot} is malformed; archived without import`,
|
||||
],
|
||||
});
|
||||
expect(fs.existsSync(jsonPath)).toBe(false);
|
||||
expect(fs.existsSync(`${jsonPath}.migrated`)).toBe(true);
|
||||
});
|
||||
|
||||
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 roomId = "!room:example.org";
|
||||
const now = Date.now();
|
||||
// Simulate the row the upgraded runtime already committed post-upgrade.
|
||||
await expect(
|
||||
importNewestInboundDedupeMarkers({
|
||||
io,
|
||||
now,
|
||||
stateMaxEntries: 2,
|
||||
markers: [{ accountId: "ops", roomId, eventId: "$runtime", ts: now - 1_000 }],
|
||||
}),
|
||||
).resolves.toEqual({ imported: 1, total: 1 });
|
||||
|
||||
// Only one slot remains: the newest legacy marker wins, the runtime row survives.
|
||||
await expect(
|
||||
importNewestInboundDedupeMarkers({
|
||||
io,
|
||||
now,
|
||||
stateMaxEntries: 2,
|
||||
markers: [
|
||||
{ accountId: "ops", roomId, eventId: "$old", ts: now - 60_000 },
|
||||
{ accountId: "ops", roomId, eventId: "$newer", ts: now - 30_000 },
|
||||
],
|
||||
}),
|
||||
).resolves.toEqual({ imported: 1, total: 2 });
|
||||
|
||||
const store = createPluginStateKeyedStoreForTests<{ key: string }>("matrix", {
|
||||
namespace: resolveMatrixInboundDedupeStateNamespace(),
|
||||
maxEntries: 2,
|
||||
defaultTtlMs: MATRIX_INBOUND_DEDUPE_TTL_MS,
|
||||
env: io.env,
|
||||
});
|
||||
const keys = (await store.entries()).map((entry) => entry.value.key).toSorted();
|
||||
expect(keys).toEqual([`ops\0${roomId}\0$runtime`, `ops\0${roomId}\0$newer`].toSorted());
|
||||
});
|
||||
});
|
||||
|
||||
@@ -38,6 +38,16 @@ import {
|
||||
type MatrixIdbSnapshotRecord,
|
||||
type MatrixLegacyCryptoMigrationState,
|
||||
} from "./src/matrix/crypto-state-store.js";
|
||||
import {
|
||||
collectMatrixInboundDedupeSources,
|
||||
importNewestInboundDedupeMarkers,
|
||||
MATRIX_LEGACY_INBOUND_DEDUPE_FILENAME,
|
||||
readLegacyInboundDedupeJsonSource,
|
||||
readLegacyInboundDedupeSqliteSource,
|
||||
retireLegacyInboundDedupeSqliteRows,
|
||||
type LegacyInboundDedupeMarker,
|
||||
type MatrixInboundDedupeMigrationIo,
|
||||
} from "./src/matrix/monitor/inbound-dedupe-migration.js";
|
||||
import { readLegacyMatrixIdbSnapshotState } from "./src/matrix/sdk/idb-persistence.js";
|
||||
import type { MatrixStoredRecoveryKey } from "./src/matrix/sdk/types.js";
|
||||
|
||||
@@ -137,6 +147,118 @@ async function archiveLegacyMatrixStateFile(params: {
|
||||
}
|
||||
|
||||
export const stateMigrations: PluginDoctorStateMigration[] = [
|
||||
{
|
||||
id: "matrix-inbound-dedupe-to-claimable-dedupe",
|
||||
label: "Matrix inbound dedupe markers",
|
||||
async detectLegacyState(params) {
|
||||
const io: MatrixInboundDedupeMigrationIo = { context: params.context, env: params.env };
|
||||
const preview: string[] = [];
|
||||
const sources = await collectMatrixInboundDedupeSources(params.stateDir);
|
||||
for (const storageRootDir of sources.sqliteRoots) {
|
||||
try {
|
||||
if (
|
||||
(await readLegacyInboundDedupeSqliteSource(io, storageRootDir)).legacyRowCount === 0
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
preview.push(
|
||||
`Matrix inbound dedupe rows can migrate to the claimable dedupe store: ${storageRootDir}`,
|
||||
);
|
||||
}
|
||||
for (const storageRootDir of sources.jsonRoots) {
|
||||
preview.push(
|
||||
`Matrix inbound dedupe JSON can migrate to the claimable dedupe store: ${path.join(storageRootDir, MATRIX_LEGACY_INBOUND_DEDUPE_FILENAME)}`,
|
||||
);
|
||||
}
|
||||
return preview.length > 0 ? { preview } : null;
|
||||
},
|
||||
async migrateLegacyState(params) {
|
||||
const io: MatrixInboundDedupeMigrationIo = { context: params.context, env: params.env };
|
||||
const changes: string[] = [];
|
||||
const warnings: string[] = [];
|
||||
const sources = await collectMatrixInboundDedupeSources(params.stateDir);
|
||||
|
||||
// Gather every marker first so the capacity-aware import keeps the
|
||||
// globally newest ones instead of whichever storage root imports last.
|
||||
const gathered: LegacyInboundDedupeMarker[] = [];
|
||||
const sqliteRootsToRetire: string[] = [];
|
||||
for (const storageRootDir of sources.sqliteRoots) {
|
||||
try {
|
||||
const source = await readLegacyInboundDedupeSqliteSource(io, storageRootDir);
|
||||
if (source.legacyRowCount === 0) {
|
||||
continue;
|
||||
}
|
||||
gathered.push(...source.markers);
|
||||
sqliteRootsToRetire.push(storageRootDir);
|
||||
} catch (err) {
|
||||
warnings.push(
|
||||
`Failed reading Matrix inbound dedupe rows for ${storageRootDir}: ${String(err)}; left legacy rows in place`,
|
||||
);
|
||||
}
|
||||
}
|
||||
const jsonRootsToRetire: string[] = [];
|
||||
for (const storageRootDir of sources.jsonRoots) {
|
||||
try {
|
||||
const markers = await readLegacyInboundDedupeJsonSource(storageRootDir);
|
||||
if (markers === null) {
|
||||
// Nothing recoverable, but archiving (rename, not delete) resolves
|
||||
// the pending detection while preserving the bytes for inspection.
|
||||
warnings.push(
|
||||
`Matrix inbound dedupe JSON for ${storageRootDir} is malformed; archived without import`,
|
||||
);
|
||||
} else {
|
||||
gathered.push(...markers);
|
||||
}
|
||||
jsonRootsToRetire.push(storageRootDir);
|
||||
} catch (err) {
|
||||
warnings.push(
|
||||
`Failed reading Matrix inbound dedupe JSON for ${storageRootDir}: ${String(err)}; left legacy file in place`,
|
||||
);
|
||||
}
|
||||
}
|
||||
if (sqliteRootsToRetire.length + jsonRootsToRetire.length === 0) {
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await importNewestInboundDedupeMarkers({ io, markers: gathered });
|
||||
changes.push(
|
||||
`Migrated Matrix inbound dedupe markers to the claimable dedupe store (${result.imported} of ${result.total} entries)`,
|
||||
);
|
||||
} catch (err) {
|
||||
warnings.push(
|
||||
`Failed importing Matrix inbound dedupe markers: ${String(err)}; left legacy sources in place`,
|
||||
);
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
||||
// Retire the legacy sources only after the import succeeded so a failed
|
||||
// run keeps them for the next doctor attempt.
|
||||
for (const storageRootDir of sqliteRootsToRetire) {
|
||||
try {
|
||||
await retireLegacyInboundDedupeSqliteRows(io, storageRootDir);
|
||||
changes.push(`Retired Matrix inbound dedupe rows for ${storageRootDir}`);
|
||||
} catch (err) {
|
||||
warnings.push(
|
||||
`Failed retiring Matrix inbound dedupe rows for ${storageRootDir}: ${String(err)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
for (const storageRootDir of jsonRootsToRetire) {
|
||||
await archiveLegacyMatrixStateFile({
|
||||
storageRootDir,
|
||||
filename: MATRIX_LEGACY_INBOUND_DEDUPE_FILENAME,
|
||||
label: "Matrix inbound dedupe",
|
||||
changes,
|
||||
warnings,
|
||||
});
|
||||
}
|
||||
return { changes, warnings };
|
||||
},
|
||||
},
|
||||
{
|
||||
id: "matrix-storage-meta-json-to-plugin-state",
|
||||
label: "Matrix storage metadata",
|
||||
|
||||
@@ -2562,7 +2562,7 @@ describe("matrix monitor handler live allowlist reload", () => {
|
||||
describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
it("skips replayed inbound events before session recording", async () => {
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => false),
|
||||
claimEvent: vi.fn(async () => false),
|
||||
commitEvent: vi.fn(async () => undefined),
|
||||
releaseEvent: vi.fn(),
|
||||
};
|
||||
@@ -2594,7 +2594,7 @@ describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
it("commits inbound events only after queued replies finish delivering", async () => {
|
||||
const callOrder: string[] = [];
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => {
|
||||
claimEvent: vi.fn(async () => {
|
||||
callOrder.push("claim");
|
||||
return true;
|
||||
}),
|
||||
@@ -2661,7 +2661,7 @@ describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
|
||||
it("commits a claimed event when bot loop protection suppresses dispatch", async () => {
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => true),
|
||||
claimEvent: vi.fn(async () => true),
|
||||
commitEvent: vi.fn(async () => undefined),
|
||||
releaseEvent: vi.fn(),
|
||||
};
|
||||
@@ -2703,7 +2703,7 @@ describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
|
||||
it("releases a claimed event when reply dispatch fails before completion", async () => {
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => true),
|
||||
claimEvent: vi.fn(async () => true),
|
||||
commitEvent: vi.fn(async () => undefined),
|
||||
releaseEvent: vi.fn(),
|
||||
};
|
||||
@@ -2740,7 +2740,7 @@ describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
|
||||
it("keeps replay committed when queued final delivery fails after a generic error", async () => {
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => true),
|
||||
claimEvent: vi.fn(async () => true),
|
||||
commitEvent: vi.fn(async () => undefined),
|
||||
releaseEvent: vi.fn(),
|
||||
};
|
||||
@@ -2787,7 +2787,7 @@ describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
"keeps replay committed when queued %s delivery fails after a generic error and no final reply exists",
|
||||
async (kind) => {
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => true),
|
||||
claimEvent: vi.fn(async () => true),
|
||||
commitEvent: vi.fn(async () => undefined),
|
||||
releaseEvent: vi.fn(),
|
||||
};
|
||||
@@ -2837,7 +2837,7 @@ describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
|
||||
it("releases a claimed event when queued final delivery fails with an explicit retryable error", async () => {
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => true),
|
||||
claimEvent: vi.fn(async () => true),
|
||||
commitEvent: vi.fn(async () => undefined),
|
||||
releaseEvent: vi.fn(),
|
||||
};
|
||||
@@ -2883,7 +2883,7 @@ describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
"releases a claimed event when queued %s delivery fails with an explicit retryable error and no final reply exists",
|
||||
async (kind) => {
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => true),
|
||||
claimEvent: vi.fn(async () => true),
|
||||
commitEvent: vi.fn(async () => undefined),
|
||||
releaseEvent: vi.fn(),
|
||||
};
|
||||
@@ -2929,7 +2929,7 @@ describe("matrix monitor handler durable inbound dedupe", () => {
|
||||
it("commits a claimed event when dispatch completes without a final reply", async () => {
|
||||
const callOrder: string[] = [];
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => {
|
||||
claimEvent: vi.fn(async () => {
|
||||
callOrder.push("claim");
|
||||
return true;
|
||||
}),
|
||||
|
||||
@@ -670,7 +670,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
|
||||
return undefined;
|
||||
}
|
||||
if (eventId && inboundDeduper) {
|
||||
claimedInboundEvent = inboundDeduper.claimEvent({ roomId, eventId });
|
||||
claimedInboundEvent = await inboundDeduper.claimEvent({ roomId, eventId });
|
||||
if (!claimedInboundEvent) {
|
||||
logVerboseMessage(`matrix: skip duplicate inbound event room=${roomId} id=${eventId}`);
|
||||
return undefined;
|
||||
|
||||
@@ -0,0 +1,294 @@
|
||||
// Legacy-source readers and importer for the one-time doctor migration of
|
||||
// pre-claimable-dedupe inbound replay markers. Losing those markers would
|
||||
// re-dispatch already-handled events on the first stale /sync or decrypt
|
||||
// replay after upgrade. Two shipped sources exist:
|
||||
// - >=2026.6 tags persisted rows in each account storage root's SQLite DB
|
||||
// (namespace `inbound-dedupe`, key `<accountId>:<sha256>`, value
|
||||
// `{roomId, eventId, ts}`), plus `inbound-dedupe-migrations` import markers.
|
||||
// - <=2026.5 tags wrote `inbound-dedupe.json` beside the account sync store;
|
||||
// the retired runtime importer read it lazily, so upgrades that skip the
|
||||
// SQLite era can still carry the raw file.
|
||||
// The PluginDoctorStateMigration itself lives in doctor-contract-api.ts, which
|
||||
// also owns the legacy-file archival write.
|
||||
import { createHash } from "node:crypto";
|
||||
import type { Dirent } from "node:fs";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import {
|
||||
createPersistentDedupeImportEntry,
|
||||
type PersistentDedupeEntry,
|
||||
} from "openclaw/plugin-sdk/persistent-dedupe";
|
||||
import type { PluginDoctorStateMigrationContext } from "openclaw/plugin-sdk/runtime-doctor";
|
||||
import { isRecord } from "../../record-shared.js";
|
||||
import { normalizeMatrixStorageMetadata } from "../client/storage.js";
|
||||
import {
|
||||
buildMatrixInboundDedupeEventKey,
|
||||
MATRIX_INBOUND_DEDUPE_STATE_MAX_ENTRIES,
|
||||
MATRIX_INBOUND_DEDUPE_TTL_MS,
|
||||
resolveMatrixInboundDedupeStateNamespace,
|
||||
} from "./inbound-dedupe.js";
|
||||
|
||||
const LEGACY_SQLITE_NAMESPACE = "inbound-dedupe";
|
||||
const LEGACY_SQLITE_MAX_ENTRIES = 20_000;
|
||||
const LEGACY_MARKERS_NAMESPACE = "inbound-dedupe-migrations";
|
||||
const LEGACY_MARKERS_MAX_ENTRIES = 1_000;
|
||||
const LEGACY_JSON_VERSION = 1;
|
||||
const STORAGE_META_FILENAME = "storage-meta.json";
|
||||
|
||||
export const MATRIX_LEGACY_INBOUND_DEDUPE_FILENAME = "inbound-dedupe.json";
|
||||
|
||||
export type MatrixInboundDedupeMigrationIo = {
|
||||
context: PluginDoctorStateMigrationContext;
|
||||
env: NodeJS.ProcessEnv;
|
||||
};
|
||||
|
||||
export type LegacyInboundDedupeMarker = {
|
||||
accountId: string;
|
||||
roomId: string;
|
||||
eventId: string;
|
||||
ts: number;
|
||||
};
|
||||
|
||||
export async function collectMatrixInboundDedupeSources(stateDir: string): Promise<{
|
||||
sqliteRoots: string[];
|
||||
jsonRoots: string[];
|
||||
}> {
|
||||
const matrixRoot = path.join(stateDir, "matrix");
|
||||
const sqliteRoots = new Set<string>();
|
||||
const jsonRoots = new Set<string>();
|
||||
async function visit(dir: string): Promise<void> {
|
||||
let entries: Dirent[];
|
||||
try {
|
||||
entries = await fs.readdir(dir, { withFileTypes: true });
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
for (const entry of entries) {
|
||||
const entryPath = path.join(dir, entry.name);
|
||||
if (entry.isFile()) {
|
||||
// Legacy per-root dedupe rows live in `<storageRoot>/state/openclaw.sqlite`.
|
||||
if (entry.name === "openclaw.sqlite" && path.basename(dir) === "state") {
|
||||
sqliteRoots.add(path.dirname(dir));
|
||||
} else if (entry.name === MATRIX_LEGACY_INBOUND_DEDUPE_FILENAME) {
|
||||
jsonRoots.add(dir);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (entry.isDirectory()) {
|
||||
await visit(entryPath);
|
||||
}
|
||||
}
|
||||
}
|
||||
await visit(matrixRoot);
|
||||
const matrixRootResolved = path.resolve(matrixRoot);
|
||||
const isAccountRoot = (root: string) => path.resolve(root) !== matrixRootResolved;
|
||||
return {
|
||||
sqliteRoots: [...sqliteRoots].filter(isAccountRoot).toSorted(),
|
||||
jsonRoots: [...jsonRoots].filter(isAccountRoot).toSorted(),
|
||||
};
|
||||
}
|
||||
|
||||
function openLegacySqliteStore(io: MatrixInboundDedupeMigrationIo, storageRootDir: string) {
|
||||
return io.context.openPluginStateKeyedStore<unknown>({
|
||||
namespace: LEGACY_SQLITE_NAMESPACE,
|
||||
maxEntries: LEGACY_SQLITE_MAX_ENTRIES,
|
||||
env: { ...io.env, OPENCLAW_STATE_DIR: storageRootDir },
|
||||
});
|
||||
}
|
||||
|
||||
function openLegacyMarkersStore(io: MatrixInboundDedupeMigrationIo, storageRootDir: string) {
|
||||
return io.context.openPluginStateKeyedStore<unknown>({
|
||||
namespace: LEGACY_MARKERS_NAMESPACE,
|
||||
maxEntries: LEGACY_MARKERS_MAX_ENTRIES,
|
||||
env: { ...io.env, OPENCLAW_STATE_DIR: storageRootDir },
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeLegacyTimestamp(raw: unknown): number | null {
|
||||
if (typeof raw !== "number" || !Number.isFinite(raw)) {
|
||||
return null;
|
||||
}
|
||||
return Math.max(0, Math.floor(raw));
|
||||
}
|
||||
|
||||
function parseLegacySqliteRow(row: {
|
||||
key: string;
|
||||
value: unknown;
|
||||
}): LegacyInboundDedupeMarker | null {
|
||||
const value = isRecord(row.value) ? row.value : {};
|
||||
const roomId = typeof value.roomId === "string" ? value.roomId.trim() : "";
|
||||
const eventId = typeof value.eventId === "string" ? value.eventId.trim() : "";
|
||||
const ts = normalizeLegacyTimestamp(value.ts);
|
||||
const separator = row.key.lastIndexOf(":");
|
||||
if (!roomId || !eventId || ts === null || separator <= 0) {
|
||||
return null;
|
||||
}
|
||||
const accountId = row.key.slice(0, separator);
|
||||
// Legacy keys embed sha256(accountId\0roomId\0eventId); recomputing it
|
||||
// validates the account-prefix parse and drops corrupt rows, mirroring the
|
||||
// retired runtime loader's expected-key check.
|
||||
const digest = createHash("sha256")
|
||||
.update(accountId)
|
||||
.update("\0")
|
||||
.update(roomId)
|
||||
.update("\0")
|
||||
.update(eventId)
|
||||
.digest("hex");
|
||||
if (row.key.slice(separator + 1) !== digest) {
|
||||
return null;
|
||||
}
|
||||
return { accountId, roomId, eventId, ts };
|
||||
}
|
||||
|
||||
/** Reads one storage root's legacy SQLite dedupe rows; throws on store errors. */
|
||||
export async function readLegacyInboundDedupeSqliteSource(
|
||||
io: MatrixInboundDedupeMigrationIo,
|
||||
storageRootDir: string,
|
||||
): Promise<{ markers: LegacyInboundDedupeMarker[]; legacyRowCount: number }> {
|
||||
const rows = await openLegacySqliteStore(io, storageRootDir).entries();
|
||||
const markerRows = await openLegacyMarkersStore(io, storageRootDir).entries();
|
||||
const markers: LegacyInboundDedupeMarker[] = [];
|
||||
for (const row of rows) {
|
||||
const marker = parseLegacySqliteRow(row);
|
||||
if (marker) {
|
||||
markers.push(marker);
|
||||
}
|
||||
}
|
||||
return { markers, legacyRowCount: rows.length + markerRows.length };
|
||||
}
|
||||
|
||||
/** Clears one storage root's retired legacy dedupe namespaces after import. */
|
||||
export async function retireLegacyInboundDedupeSqliteRows(
|
||||
io: MatrixInboundDedupeMigrationIo,
|
||||
storageRootDir: string,
|
||||
): Promise<void> {
|
||||
await openLegacySqliteStore(io, storageRootDir).clear();
|
||||
await openLegacyMarkersStore(io, storageRootDir).clear();
|
||||
}
|
||||
|
||||
async function resolveJsonRootAccountId(storageRootDir: string): Promise<string> {
|
||||
// The JSON era predates the per-root SQLite stores, so account identity comes
|
||||
// from storage-meta.json (or its doctor-archived copy when the metadata
|
||||
// migration already ran). Pre-metadata roots belong to the legacy single
|
||||
// account, which used the literal "default" account id.
|
||||
for (const filename of [STORAGE_META_FILENAME, `${STORAGE_META_FILENAME}.migrated`]) {
|
||||
try {
|
||||
const metadata = normalizeMatrixStorageMetadata(
|
||||
JSON.parse(await fs.readFile(path.join(storageRootDir, filename), "utf8")) as unknown,
|
||||
);
|
||||
if (metadata?.accountId) {
|
||||
return metadata.accountId;
|
||||
}
|
||||
} catch {
|
||||
// Try the next metadata source.
|
||||
}
|
||||
}
|
||||
return "default";
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads one storage root's legacy inbound-dedupe.json markers. Throws on file
|
||||
* read errors so a transiently unreadable file is never retired unread, and
|
||||
* returns null for malformed content so the caller can archive it explicitly.
|
||||
*/
|
||||
export async function readLegacyInboundDedupeJsonSource(
|
||||
storageRootDir: string,
|
||||
): Promise<LegacyInboundDedupeMarker[] | null> {
|
||||
const jsonPath = path.join(storageRootDir, MATRIX_LEGACY_INBOUND_DEDUPE_FILENAME);
|
||||
const raw = await fs.readFile(jsonPath, "utf8");
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(raw) as unknown;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (
|
||||
!isRecord(parsed) ||
|
||||
parsed.version !== LEGACY_JSON_VERSION ||
|
||||
!Array.isArray(parsed.entries)
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
const accountId = await resolveJsonRootAccountId(storageRootDir);
|
||||
const markers: LegacyInboundDedupeMarker[] = [];
|
||||
for (const entry of parsed.entries) {
|
||||
if (!isRecord(entry) || typeof entry.key !== "string") {
|
||||
continue;
|
||||
}
|
||||
// Legacy JSON keys are `roomId|eventId`; event ids never contain "|".
|
||||
const separator = entry.key.indexOf("|");
|
||||
if (separator <= 0) {
|
||||
continue;
|
||||
}
|
||||
const roomId = entry.key.slice(0, separator).trim();
|
||||
const eventId = entry.key.slice(separator + 1).trim();
|
||||
const ts = normalizeLegacyTimestamp(entry.ts);
|
||||
if (!roomId || !eventId || ts === null) {
|
||||
continue;
|
||||
}
|
||||
markers.push({ accountId, roomId, eventId, ts });
|
||||
}
|
||||
return markers;
|
||||
}
|
||||
|
||||
/**
|
||||
* Imports the globally newest legacy markers into the claimable-dedupe store.
|
||||
* Never exceeds capacity: eviction is by row creation time, so letting fresh
|
||||
* imports overflow the namespace would evict the newer rows the runtime
|
||||
* committed since the upgrade. Throws on store errors so the caller keeps the
|
||||
* legacy sources for the next doctor attempt.
|
||||
*/
|
||||
export async function importNewestInboundDedupeMarkers(params: {
|
||||
io: MatrixInboundDedupeMigrationIo;
|
||||
markers: Iterable<LegacyInboundDedupeMarker>;
|
||||
now?: number;
|
||||
stateMaxEntries?: number;
|
||||
}): Promise<{ imported: number; total: number }> {
|
||||
const now = params.now ?? Date.now();
|
||||
const stateMaxEntries = params.stateMaxEntries ?? MATRIX_INBOUND_DEDUPE_STATE_MAX_ENTRIES;
|
||||
const newestByKey = new Map<string, LegacyInboundDedupeMarker & { key: string }>();
|
||||
for (const marker of params.markers) {
|
||||
const key = buildMatrixInboundDedupeEventKey(marker);
|
||||
if (!key) {
|
||||
continue;
|
||||
}
|
||||
const existing = newestByKey.get(key);
|
||||
if (!existing || marker.ts > existing.ts) {
|
||||
newestByKey.set(key, { ...marker, key });
|
||||
}
|
||||
}
|
||||
// Newest first so the capacity limit drops the least replay-relevant markers.
|
||||
const markers = [...newestByKey.values()].toSorted((left, right) => right.ts - left.ts);
|
||||
const store = params.io.context.openPluginStateKeyedStore<PersistentDedupeEntry>({
|
||||
namespace: resolveMatrixInboundDedupeStateNamespace(),
|
||||
maxEntries: stateMaxEntries,
|
||||
defaultTtlMs: MATRIX_INBOUND_DEDUPE_TTL_MS,
|
||||
env: params.io.env,
|
||||
});
|
||||
let capacity = Math.max(0, stateMaxEntries - (await store.entries()).length);
|
||||
let imported = 0;
|
||||
for (const marker of markers) {
|
||||
if (capacity <= 0) {
|
||||
break;
|
||||
}
|
||||
const remainingTtlMs = MATRIX_INBOUND_DEDUPE_TTL_MS - (now - marker.ts);
|
||||
if (remainingTtlMs <= 0) {
|
||||
continue;
|
||||
}
|
||||
const entry = createPersistentDedupeImportEntry({
|
||||
key: marker.key,
|
||||
seenAt: marker.ts,
|
||||
ttlMs: Math.max(1, Math.floor(remainingTtlMs)),
|
||||
});
|
||||
// Rows committed after the upgrade are newer than any legacy marker, so
|
||||
// registerIfAbsent keeps them and only fills the missing keys.
|
||||
const registered = await store.registerIfAbsent(entry.key, entry.value, {
|
||||
ttlMs: entry.ttlMs ?? MATRIX_INBOUND_DEDUPE_TTL_MS,
|
||||
});
|
||||
if (registered) {
|
||||
imported += 1;
|
||||
capacity -= 1;
|
||||
}
|
||||
}
|
||||
return { imported, total: markers.length };
|
||||
}
|
||||
@@ -2,219 +2,106 @@
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime";
|
||||
import {
|
||||
createPluginStateKeyedStoreForTests,
|
||||
resetPluginStateStoreForTests,
|
||||
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { PluginRuntime } from "../../runtime-api.js";
|
||||
import { setMatrixRuntime } from "../../runtime.js";
|
||||
import { resetPluginStateStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { LogService } from "../sdk/logger.js";
|
||||
import { createMatrixInboundEventDeduper } from "./inbound-dedupe.js";
|
||||
|
||||
describe("Matrix inbound event dedupe", () => {
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
beforeEach(() => {
|
||||
setMatrixRuntime({
|
||||
state: {
|
||||
openKeyedStore: (options: OpenKeyedStoreOptions) =>
|
||||
createPluginStateKeyedStoreForTests("matrix", options),
|
||||
},
|
||||
} as unknown as PluginRuntime);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
vi.useRealTimers();
|
||||
resetPluginStateStoreForTests();
|
||||
for (const dir of tempDirs.splice(0)) {
|
||||
fs.rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
function createStoragePath(): string {
|
||||
function createStateEnv(): NodeJS.ProcessEnv {
|
||||
const dir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-inbound-dedupe-"));
|
||||
tempDirs.push(dir);
|
||||
return path.join(dir, "inbound-dedupe.json");
|
||||
return { ...process.env, OPENCLAW_STATE_DIR: dir };
|
||||
}
|
||||
|
||||
const auth = {
|
||||
accountId: "ops",
|
||||
homeserver: "https://matrix.example.org",
|
||||
userId: "@bot:example.org",
|
||||
accessToken: "token",
|
||||
deviceId: "DEVICE",
|
||||
} as const;
|
||||
const persistenceTestTtlMs = 60_000;
|
||||
const auth = { accountId: "ops" } as const;
|
||||
const event = { roomId: "!room:example.org", eventId: "$event-1" } as const;
|
||||
|
||||
it("drops a duplicate event after commit", async () => {
|
||||
const deduper = createMatrixInboundEventDeduper({ auth, env: createStateEnv() });
|
||||
|
||||
await expect(deduper.claimEvent(event)).resolves.toBe(true);
|
||||
await deduper.commitEvent(event);
|
||||
await expect(deduper.claimEvent(event)).resolves.toBe(false);
|
||||
});
|
||||
|
||||
it("reports an in-flight claim as a duplicate", async () => {
|
||||
const deduper = createMatrixInboundEventDeduper({ auth, env: createStateEnv() });
|
||||
|
||||
await expect(deduper.claimEvent(event)).resolves.toBe(true);
|
||||
await expect(deduper.claimEvent(event)).resolves.toBe(false);
|
||||
});
|
||||
|
||||
it("persists committed events across restarts", async () => {
|
||||
const storagePath = createStoragePath();
|
||||
const first = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
});
|
||||
const env = createStateEnv();
|
||||
const first = createMatrixInboundEventDeduper({ auth, env });
|
||||
await expect(first.claimEvent(event)).resolves.toBe(true);
|
||||
await first.commitEvent(event);
|
||||
|
||||
expect(first.claimEvent({ roomId: "!room:example.org", eventId: "$event-1" })).toBe(true);
|
||||
await first.commitEvent({
|
||||
roomId: "!room:example.org",
|
||||
eventId: "$event-1",
|
||||
});
|
||||
await first.stop();
|
||||
|
||||
const second = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
});
|
||||
expect(second.claimEvent({ roomId: "!room:example.org", eventId: "$event-1" })).toBe(false);
|
||||
// A fresh instance has an empty memory layer, so the duplicate verdict
|
||||
// must come from the persisted plugin-state SQLite rows.
|
||||
const second = createMatrixInboundEventDeduper({ auth, env });
|
||||
await expect(second.claimEvent(event)).resolves.toBe(false);
|
||||
});
|
||||
|
||||
it("does not persist released pending claims", async () => {
|
||||
const storagePath = createStoragePath();
|
||||
const first = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
});
|
||||
it("lets a released claim retry, including after a restart", async () => {
|
||||
const env = createStateEnv();
|
||||
const first = createMatrixInboundEventDeduper({ auth, env });
|
||||
await expect(first.claimEvent(event)).resolves.toBe(true);
|
||||
first.releaseEvent(event);
|
||||
await expect(first.claimEvent(event)).resolves.toBe(true);
|
||||
first.releaseEvent(event);
|
||||
|
||||
expect(first.claimEvent({ roomId: "!room:example.org", eventId: "$event-2" })).toBe(true);
|
||||
first.releaseEvent({ roomId: "!room:example.org", eventId: "$event-2" });
|
||||
await first.stop();
|
||||
|
||||
const second = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
});
|
||||
expect(second.claimEvent({ roomId: "!room:example.org", eventId: "$event-2" })).toBe(true);
|
||||
const second = createMatrixInboundEventDeduper({ auth, env });
|
||||
await expect(second.claimEvent(event)).resolves.toBe(true);
|
||||
});
|
||||
|
||||
it("prunes expired and overflowed entries on load", async () => {
|
||||
const storagePath = createStoragePath();
|
||||
fs.writeFileSync(
|
||||
storagePath,
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
entries: [
|
||||
{ key: "!room:example.org|$old", ts: 10 },
|
||||
{ key: "!room:example.org|$keep-1", ts: 90 },
|
||||
{ key: "!room:example.org|$keep-2", ts: 95 },
|
||||
{ key: "!room:example.org|$keep-3", ts: 100 },
|
||||
],
|
||||
}),
|
||||
"utf8",
|
||||
);
|
||||
it("scopes dedupe state per account", async () => {
|
||||
const env = createStateEnv();
|
||||
const ops = createMatrixInboundEventDeduper({ auth: { accountId: "ops" }, env });
|
||||
await expect(ops.claimEvent(event)).resolves.toBe(true);
|
||||
await ops.commitEvent(event);
|
||||
|
||||
const deduper = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
ttlMs: 20,
|
||||
maxEntries: 2,
|
||||
nowMs: () => 100,
|
||||
});
|
||||
|
||||
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$old" })).toBe(true);
|
||||
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$keep-1" })).toBe(true);
|
||||
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$keep-2" })).toBe(false);
|
||||
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$keep-3" })).toBe(false);
|
||||
const home = createMatrixInboundEventDeduper({ auth: { accountId: "home" }, env });
|
||||
await expect(home.claimEvent(event)).resolves.toBe(true);
|
||||
});
|
||||
|
||||
it("retains replayed backlog events based on processing time", async () => {
|
||||
const storagePath = createStoragePath();
|
||||
let now = 100;
|
||||
const first = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
// Plugin-state TTL uses real wall-clock time; keep restart/import tests
|
||||
// away from millisecond expiry races while fake nowMs drives dedupe pruning.
|
||||
ttlMs: persistenceTestTtlMs,
|
||||
nowMs: () => now,
|
||||
});
|
||||
it("fails open for events without usable identifiers", async () => {
|
||||
const deduper = createMatrixInboundEventDeduper({ auth, env: createStateEnv() });
|
||||
|
||||
expect(first.claimEvent({ roomId: "!room:example.org", eventId: "$backlog" })).toBe(true);
|
||||
await first.commitEvent({
|
||||
roomId: "!room:example.org",
|
||||
eventId: "$backlog",
|
||||
});
|
||||
await first.stop();
|
||||
|
||||
now = 110;
|
||||
const second = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
ttlMs: persistenceTestTtlMs,
|
||||
nowMs: () => now,
|
||||
});
|
||||
expect(second.claimEvent({ roomId: "!room:example.org", eventId: "$backlog" })).toBe(false);
|
||||
});
|
||||
|
||||
it("imports legacy JSON entries into plugin state", async () => {
|
||||
const storagePath = createStoragePath();
|
||||
fs.writeFileSync(
|
||||
storagePath,
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
entries: [{ key: "!room:example.org|$legacy", ts: 90 }],
|
||||
}),
|
||||
"utf8",
|
||||
);
|
||||
|
||||
const first = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
// Plugin-state TTL uses real wall-clock time; this test proves migration
|
||||
// durability after the legacy JSON file is gone, not expiry behavior.
|
||||
ttlMs: persistenceTestTtlMs,
|
||||
nowMs: () => 100,
|
||||
});
|
||||
expect(first.claimEvent({ roomId: "!room:example.org", eventId: "$legacy" })).toBe(false);
|
||||
|
||||
fs.rmSync(storagePath, { force: true });
|
||||
const second = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
ttlMs: persistenceTestTtlMs,
|
||||
nowMs: () => 100,
|
||||
});
|
||||
expect(second.claimEvent({ roomId: "!room:example.org", eventId: "$legacy" })).toBe(false);
|
||||
await expect(deduper.claimEvent({ roomId: " ", eventId: "$x" })).resolves.toBe(true);
|
||||
await expect(deduper.claimEvent({ roomId: " ", eventId: "$x" })).resolves.toBe(true);
|
||||
await expect(deduper.commitEvent({ roomId: "!r:x", eventId: "" })).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it("keeps committed events in memory when plugin-state persistence fails", async () => {
|
||||
const storagePath = createStoragePath();
|
||||
const warnSpy = vi.spyOn(LogService, "warn").mockImplementation(() => {});
|
||||
setMatrixRuntime({
|
||||
state: {
|
||||
openKeyedStore: () => ({
|
||||
entries: async () => [],
|
||||
register: async () => {
|
||||
throw new Error("sqlite unavailable");
|
||||
},
|
||||
registerIfAbsent: async () => false,
|
||||
lookup: async () => undefined,
|
||||
consume: async () => undefined,
|
||||
delete: async () => false,
|
||||
clear: async () => {},
|
||||
}),
|
||||
},
|
||||
} as unknown as PluginRuntime);
|
||||
const deduper = await createMatrixInboundEventDeduper({
|
||||
auth: auth as never,
|
||||
storagePath,
|
||||
const blockedDir = createStateEnv().OPENCLAW_STATE_DIR as string;
|
||||
// A regular file where the state dir should be makes every SQLite open fail.
|
||||
const filePath = path.join(blockedDir, "not-a-dir");
|
||||
fs.writeFileSync(filePath, "x", "utf8");
|
||||
const deduper = createMatrixInboundEventDeduper({
|
||||
auth,
|
||||
env: { ...process.env, OPENCLAW_STATE_DIR: path.join(filePath, "nested") },
|
||||
});
|
||||
|
||||
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$best-effort" })).toBe(true);
|
||||
await expect(
|
||||
deduper.commitEvent({
|
||||
roomId: "!room:example.org",
|
||||
eventId: "$best-effort",
|
||||
}),
|
||||
).resolves.toBeUndefined();
|
||||
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$best-effort" })).toBe(
|
||||
false,
|
||||
);
|
||||
await expect(deduper.claimEvent(event)).resolves.toBe(true);
|
||||
await expect(deduper.commitEvent(event)).resolves.toBeUndefined();
|
||||
await expect(deduper.claimEvent(event)).resolves.toBe(false);
|
||||
expect(warnSpy).toHaveBeenCalledWith(
|
||||
"MatrixInboundDedupe",
|
||||
"Failed persisting Matrix inbound dedupe entry:",
|
||||
expect.any(Error),
|
||||
"Matrix inbound dedupe persistence failed:",
|
||||
expect.anything(),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,312 +1,103 @@
|
||||
// Matrix plugin module implements inbound dedupe behavior.
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { readJsonFileWithFallback } from "openclaw/plugin-sdk/json-store";
|
||||
import { getMatrixRuntime } from "../../runtime.js";
|
||||
import { resolveMatrixStateFilePath } from "../client/storage.js";
|
||||
// Matrix inbound replay protection: /sync replays events after an unclean
|
||||
// shutdown and the decrypt bridge re-emits decrypted events, so each
|
||||
// (account, room, event) is claimed before handling and committed only after
|
||||
// reply dispatch succeeds; release on retryable failure reopens the event.
|
||||
import {
|
||||
createClaimableDedupe,
|
||||
resolvePersistentDedupePluginStateNamespace,
|
||||
} from "openclaw/plugin-sdk/persistent-dedupe";
|
||||
import type { MatrixAuth } from "../client/types.js";
|
||||
import { LogService } from "../sdk/logger.js";
|
||||
import { resolveMatrixSqliteStateEnv } from "../sqlite-state.js";
|
||||
|
||||
const INBOUND_DEDUPE_FILENAME = "inbound-dedupe.json";
|
||||
const INBOUND_DEDUPE_NAMESPACE = "inbound-dedupe";
|
||||
const INBOUND_DEDUPE_MIGRATIONS_NAMESPACE = "inbound-dedupe-migrations";
|
||||
const STORE_VERSION = 1;
|
||||
const DEFAULT_MAX_ENTRIES = 20_000;
|
||||
const DEFAULT_TTL_MS = 30 * 24 * 60 * 60 * 1000;
|
||||
|
||||
type StoredMatrixInboundDedupeEntry = {
|
||||
roomId: string;
|
||||
eventId: string;
|
||||
ts: number;
|
||||
};
|
||||
|
||||
type LegacyMatrixInboundDedupeEntry = {
|
||||
key: string;
|
||||
ts: number;
|
||||
};
|
||||
|
||||
type LegacyMatrixInboundDedupeState = {
|
||||
version: number;
|
||||
entries: LegacyMatrixInboundDedupeEntry[];
|
||||
};
|
||||
|
||||
type MatrixInboundDedupeMigrationMarker = {
|
||||
importedAt: number;
|
||||
};
|
||||
const MATRIX_INBOUND_DEDUPE_PLUGIN_ID = "matrix";
|
||||
// One shared "global" namespace with the account baked into each key: the
|
||||
// plugin-state fuse sheds only the writing namespace, so per-account
|
||||
// namespaces could starve a new account once older ones fill the per-plugin
|
||||
// row budget. A single bounded pool stays far under that fuse. The qa-matrix
|
||||
// runtime-state probe mirrors this prefix and key shape when asserting commits.
|
||||
const MATRIX_INBOUND_DEDUPE_NAMESPACE_PREFIX = "matrix.inbound-dedupe";
|
||||
const MATRIX_INBOUND_DEDUPE_NAMESPACE = "global";
|
||||
// 30d window: a /sync backlog after long downtime can resurface old events.
|
||||
export const MATRIX_INBOUND_DEDUPE_TTL_MS = 30 * 24 * 60 * 60 * 1000;
|
||||
const MATRIX_INBOUND_DEDUPE_MEMORY_MAX = 5_000;
|
||||
export const MATRIX_INBOUND_DEDUPE_STATE_MAX_ENTRIES = 20_000;
|
||||
|
||||
export type MatrixInboundEventDeduper = {
|
||||
claimEvent: (params: { roomId: string; eventId: string }) => boolean;
|
||||
/** True when the caller now owns the event; false for committed or in-flight duplicates. */
|
||||
claimEvent: (params: { roomId: string; eventId: string }) => Promise<boolean>;
|
||||
/** Records a handled event so restart/replay cannot dispatch it again. */
|
||||
commitEvent: (params: { roomId: string; eventId: string }) => Promise<void>;
|
||||
/** Drops an uncommitted claim so a failed dispatch can retry the event. */
|
||||
releaseEvent: (params: { roomId: string; eventId: string }) => void;
|
||||
flush: () => Promise<void>;
|
||||
stop: () => Promise<void>;
|
||||
};
|
||||
|
||||
function normalizeEventPart(value: string): string {
|
||||
return value.trim();
|
||||
export function resolveMatrixInboundDedupeAccountId(accountId: string): string {
|
||||
return accountId.trim() || "default";
|
||||
}
|
||||
|
||||
function buildEventKey(params: { auth: MatrixAuth; roomId: string; eventId: string }): string {
|
||||
const accountId = normalizeEventPart(params.auth.accountId) || "default";
|
||||
const roomId = normalizeEventPart(params.roomId);
|
||||
const eventId = normalizeEventPart(params.eventId);
|
||||
export function buildMatrixInboundDedupeEventKey(params: {
|
||||
accountId: string;
|
||||
roomId: string;
|
||||
eventId: string;
|
||||
}): string | null {
|
||||
const roomId = params.roomId.trim();
|
||||
const eventId = params.eventId.trim();
|
||||
if (!roomId || !eventId) {
|
||||
return "";
|
||||
}
|
||||
const digest = createHash("sha256")
|
||||
.update(accountId)
|
||||
.update("\0")
|
||||
.update(roomId)
|
||||
.update("\0")
|
||||
.update(eventId)
|
||||
.digest("hex");
|
||||
return `${accountId}:${digest}`;
|
||||
}
|
||||
|
||||
function resolveInboundDedupeStatePath(params: {
|
||||
auth: MatrixAuth;
|
||||
env?: NodeJS.ProcessEnv;
|
||||
stateDir?: string;
|
||||
}): string {
|
||||
return resolveMatrixStateFilePath({
|
||||
auth: params.auth,
|
||||
env: params.env,
|
||||
stateDir: params.stateDir,
|
||||
filename: INBOUND_DEDUPE_FILENAME,
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeTimestamp(raw: unknown): number | null {
|
||||
if (typeof raw !== "number" || !Number.isFinite(raw)) {
|
||||
return null;
|
||||
}
|
||||
return Math.max(0, Math.floor(raw));
|
||||
// NUL separators: room-version 1/2 event ids may contain ":", so a printable
|
||||
// separator could collide two distinct (account, room, event) triples.
|
||||
return `${resolveMatrixInboundDedupeAccountId(params.accountId)}\0${roomId}\0${eventId}`;
|
||||
}
|
||||
|
||||
function pruneSeenEvents(params: {
|
||||
seen: Map<string, number>;
|
||||
ttlMs: number;
|
||||
maxEntries: number;
|
||||
nowMs: number;
|
||||
}) {
|
||||
const { seen, ttlMs, maxEntries, nowMs } = params;
|
||||
if (ttlMs > 0) {
|
||||
const cutoff = nowMs - ttlMs;
|
||||
for (const [key, ts] of seen) {
|
||||
if (ts < cutoff) {
|
||||
seen.delete(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
const max = Math.max(0, Math.floor(maxEntries));
|
||||
if (max <= 0) {
|
||||
seen.clear();
|
||||
return;
|
||||
}
|
||||
while (seen.size > max) {
|
||||
const oldestKey = [...seen.entries()].toSorted(
|
||||
(a, b) => a[1] - b[1] || a[0].localeCompare(b[0]),
|
||||
)[0]?.[0];
|
||||
if (typeof oldestKey !== "string") {
|
||||
break;
|
||||
}
|
||||
seen.delete(oldestKey);
|
||||
}
|
||||
}
|
||||
|
||||
export function openMatrixInboundDedupeStoreOptions(params: {
|
||||
env?: NodeJS.ProcessEnv;
|
||||
stateDir?: string;
|
||||
}) {
|
||||
return {
|
||||
namespace: INBOUND_DEDUPE_NAMESPACE,
|
||||
maxEntries: DEFAULT_MAX_ENTRIES,
|
||||
env: resolveMatrixSqliteStateEnv(params),
|
||||
};
|
||||
}
|
||||
|
||||
function createInboundDedupeStore(params: { env?: NodeJS.ProcessEnv; stateDir?: string }) {
|
||||
return getMatrixRuntime().state.openKeyedStore<StoredMatrixInboundDedupeEntry>(
|
||||
openMatrixInboundDedupeStoreOptions(params),
|
||||
);
|
||||
}
|
||||
|
||||
function createInboundDedupeMigrationStore(params: { env?: NodeJS.ProcessEnv; stateDir?: string }) {
|
||||
return getMatrixRuntime().state.openKeyedStore<MatrixInboundDedupeMigrationMarker>({
|
||||
namespace: INBOUND_DEDUPE_MIGRATIONS_NAMESPACE,
|
||||
maxEntries: 1_000,
|
||||
env: resolveMatrixSqliteStateEnv(params),
|
||||
/** Persisted plugin-state namespace holding the inbound dedupe rows. */
|
||||
export function resolveMatrixInboundDedupeStateNamespace(): string {
|
||||
return resolvePersistentDedupePluginStateNamespace({
|
||||
namespace: MATRIX_INBOUND_DEDUPE_NAMESPACE,
|
||||
namespacePrefix: MATRIX_INBOUND_DEDUPE_NAMESPACE_PREFIX,
|
||||
});
|
||||
}
|
||||
|
||||
function buildLegacyImportKey(params: { auth: MatrixAuth; storagePath: string }): string {
|
||||
const accountId = normalizeEventPart(params.auth.accountId) || "default";
|
||||
const digest = createHash("sha256")
|
||||
.update(accountId)
|
||||
.update("\0")
|
||||
.update(params.storagePath)
|
||||
.digest("hex");
|
||||
return `${accountId}:${digest}`;
|
||||
}
|
||||
|
||||
async function loadLegacyEntries(storagePath: string): Promise<StoredMatrixInboundDedupeEntry[]> {
|
||||
const { value } = await readJsonFileWithFallback<LegacyMatrixInboundDedupeState | null>(
|
||||
storagePath,
|
||||
null,
|
||||
);
|
||||
if (value?.version !== STORE_VERSION || !Array.isArray(value.entries)) {
|
||||
return [];
|
||||
}
|
||||
const entries: StoredMatrixInboundDedupeEntry[] = [];
|
||||
for (const entry of value.entries) {
|
||||
if (!entry || typeof entry.key !== "string") {
|
||||
continue;
|
||||
}
|
||||
const separatorIndex = entry.key.indexOf("|");
|
||||
if (separatorIndex <= 0) {
|
||||
continue;
|
||||
}
|
||||
const roomId = entry.key.slice(0, separatorIndex).trim();
|
||||
const eventId = entry.key.slice(separatorIndex + 1).trim();
|
||||
const ts = normalizeTimestamp(entry.ts);
|
||||
if (!roomId || !eventId || ts === null) {
|
||||
continue;
|
||||
}
|
||||
entries.push({ roomId, eventId, ts });
|
||||
}
|
||||
return entries;
|
||||
}
|
||||
|
||||
export async function createMatrixInboundEventDeduper(params: {
|
||||
auth: MatrixAuth;
|
||||
export function createMatrixInboundEventDeduper(params: {
|
||||
auth: Pick<MatrixAuth, "accountId">;
|
||||
env?: NodeJS.ProcessEnv;
|
||||
stateDir?: string;
|
||||
storagePath?: string;
|
||||
ttlMs?: number;
|
||||
maxEntries?: number;
|
||||
nowMs?: () => number;
|
||||
}): Promise<MatrixInboundEventDeduper> {
|
||||
const nowMs = params.nowMs ?? (() => Date.now());
|
||||
const ttlMs =
|
||||
typeof params.ttlMs === "number" && Number.isFinite(params.ttlMs)
|
||||
? Math.max(0, Math.floor(params.ttlMs))
|
||||
: DEFAULT_TTL_MS;
|
||||
const maxEntries =
|
||||
typeof params.maxEntries === "number" && Number.isFinite(params.maxEntries)
|
||||
? Math.max(0, Math.floor(params.maxEntries))
|
||||
: DEFAULT_MAX_ENTRIES;
|
||||
const storagePath =
|
||||
params.storagePath ??
|
||||
resolveInboundDedupeStatePath({
|
||||
auth: params.auth,
|
||||
env: params.env,
|
||||
stateDir: params.stateDir,
|
||||
});
|
||||
const stateDir = params.stateDir ?? path.dirname(storagePath);
|
||||
const store = createInboundDedupeStore({ env: params.env, stateDir });
|
||||
const migrationStore = createInboundDedupeMigrationStore({ env: params.env, stateDir });
|
||||
|
||||
const seen = new Map<string, number>();
|
||||
const pending = new Set<string>();
|
||||
|
||||
try {
|
||||
for (const entry of await store.entries()) {
|
||||
const value = entry.value;
|
||||
const roomId = typeof value?.roomId === "string" ? value.roomId.trim() : "";
|
||||
const eventId = typeof value?.eventId === "string" ? value.eventId.trim() : "";
|
||||
const ts = normalizeTimestamp(value?.ts);
|
||||
const expectedKey = buildEventKey({ auth: params.auth, roomId, eventId });
|
||||
if (expectedKey && expectedKey === entry.key && ts !== null) {
|
||||
seen.set(entry.key, ts);
|
||||
}
|
||||
}
|
||||
const legacyImportKey = buildLegacyImportKey({ auth: params.auth, storagePath });
|
||||
const legacyAlreadyImported = await migrationStore.lookup(legacyImportKey);
|
||||
if (!legacyAlreadyImported) {
|
||||
const legacyEntries = await loadLegacyEntries(storagePath);
|
||||
let migratedLegacyEntries = 0;
|
||||
for (const entry of legacyEntries) {
|
||||
const key = buildEventKey({
|
||||
auth: params.auth,
|
||||
roomId: entry.roomId,
|
||||
eventId: entry.eventId,
|
||||
});
|
||||
if (!key) {
|
||||
continue;
|
||||
}
|
||||
if (seen.has(key)) {
|
||||
migratedLegacyEntries += 1;
|
||||
continue;
|
||||
}
|
||||
seen.set(key, entry.ts);
|
||||
await store
|
||||
.register(key, entry, ttlMs > 0 ? { ttlMs } : undefined)
|
||||
.then(() => {
|
||||
migratedLegacyEntries += 1;
|
||||
})
|
||||
.catch(() => {});
|
||||
}
|
||||
if (legacyEntries.length > 0 && migratedLegacyEntries === legacyEntries.length) {
|
||||
await migrationStore.register(legacyImportKey, { importedAt: nowMs() });
|
||||
await fs.rm(storagePath, { force: true }).catch(() => {});
|
||||
}
|
||||
}
|
||||
pruneSeenEvents({ seen, ttlMs, maxEntries, nowMs: nowMs() });
|
||||
} catch (err) {
|
||||
LogService.warn("MatrixInboundDedupe", "Failed loading Matrix inbound dedupe store:", err);
|
||||
}
|
||||
|
||||
}): MatrixInboundEventDeduper {
|
||||
const guard = createClaimableDedupe({
|
||||
pluginId: MATRIX_INBOUND_DEDUPE_PLUGIN_ID,
|
||||
namespacePrefix: MATRIX_INBOUND_DEDUPE_NAMESPACE_PREFIX,
|
||||
ttlMs: MATRIX_INBOUND_DEDUPE_TTL_MS,
|
||||
memoryMaxSize: MATRIX_INBOUND_DEDUPE_MEMORY_MAX,
|
||||
stateMaxEntries: MATRIX_INBOUND_DEDUPE_STATE_MAX_ENTRIES,
|
||||
...(params.env ? { env: params.env } : {}),
|
||||
// Persistence is best effort: a broken state DB must never block inbound
|
||||
// handling, so disk errors log and the memory layer keeps deduping.
|
||||
onDiskError: (err) => {
|
||||
LogService.warn("MatrixInboundDedupe", "Matrix inbound dedupe persistence failed:", err);
|
||||
},
|
||||
});
|
||||
const accountId = params.auth.accountId;
|
||||
const namespace = MATRIX_INBOUND_DEDUPE_NAMESPACE;
|
||||
return {
|
||||
claimEvent: ({ roomId, eventId }) => {
|
||||
const key = buildEventKey({ auth: params.auth, roomId, eventId });
|
||||
claimEvent: async (ids) => {
|
||||
const key = buildMatrixInboundDedupeEventKey({ accountId, ...ids });
|
||||
if (!key) {
|
||||
// Fail open: never suppress an event we cannot identify.
|
||||
return true;
|
||||
}
|
||||
pruneSeenEvents({ seen, ttlMs, maxEntries, nowMs: nowMs() });
|
||||
if (seen.has(key) || pending.has(key)) {
|
||||
return false;
|
||||
}
|
||||
pending.add(key);
|
||||
return true;
|
||||
return (await guard.claim(key, { namespace })).kind === "claimed";
|
||||
},
|
||||
commitEvent: async ({ roomId, eventId }) => {
|
||||
const key = buildEventKey({ auth: params.auth, roomId, eventId });
|
||||
commitEvent: async (ids) => {
|
||||
const key = buildMatrixInboundDedupeEventKey({ accountId, ...ids });
|
||||
if (!key) {
|
||||
return;
|
||||
}
|
||||
pending.delete(key);
|
||||
const ts = nowMs();
|
||||
seen.delete(key);
|
||||
seen.set(key, ts);
|
||||
pruneSeenEvents({ seen, ttlMs, maxEntries, nowMs: nowMs() });
|
||||
await store
|
||||
.register(
|
||||
key,
|
||||
{
|
||||
roomId: normalizeEventPart(roomId),
|
||||
eventId: normalizeEventPart(eventId),
|
||||
ts,
|
||||
},
|
||||
ttlMs > 0 ? { ttlMs } : undefined,
|
||||
)
|
||||
.catch((err: unknown) => {
|
||||
LogService.warn(
|
||||
"MatrixInboundDedupe",
|
||||
"Failed persisting Matrix inbound dedupe entry:",
|
||||
err,
|
||||
);
|
||||
});
|
||||
await guard.commit(key, { namespace });
|
||||
},
|
||||
releaseEvent: ({ roomId, eventId }) => {
|
||||
const key = buildEventKey({ auth: params.auth, roomId, eventId });
|
||||
if (!key) {
|
||||
return;
|
||||
releaseEvent: (ids) => {
|
||||
const key = buildMatrixInboundDedupeEventKey({ accountId, ...ids });
|
||||
if (key) {
|
||||
guard.release(key, { namespace });
|
||||
}
|
||||
pending.delete(key);
|
||||
},
|
||||
flush: async () => {},
|
||||
stop: async () => {},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -50,13 +50,11 @@ const hoisted = vi.hoisted(() => {
|
||||
dm: {},
|
||||
};
|
||||
const inboundDeduper = {
|
||||
claimEvent: vi.fn(() => true),
|
||||
claimEvent: vi.fn(async () => true),
|
||||
commitEvent: vi.fn(async () => undefined),
|
||||
releaseEvent: vi.fn(),
|
||||
flush: vi.fn(async () => undefined),
|
||||
stop: vi.fn(async () => undefined),
|
||||
};
|
||||
const createMatrixInboundEventDeduper = vi.fn(async () => inboundDeduper);
|
||||
const createMatrixInboundEventDeduper = vi.fn(() => inboundDeduper);
|
||||
const client = Object.assign(createEmitter(), {
|
||||
id: "matrix-client",
|
||||
hasPersistedSyncState: vi.fn(() => false),
|
||||
@@ -503,12 +501,10 @@ describe("monitorMatrixProvider", () => {
|
||||
hoisted.client.hasPersistedSyncState.mockReset().mockReturnValue(false);
|
||||
hoisted.client.stopSyncWithoutPersist.mockReset();
|
||||
hoisted.client.drainPendingDecryptions.mockReset().mockResolvedValue(undefined);
|
||||
hoisted.inboundDeduper.claimEvent.mockReset().mockReturnValue(true);
|
||||
hoisted.inboundDeduper.claimEvent.mockReset().mockResolvedValue(true);
|
||||
hoisted.inboundDeduper.commitEvent.mockReset().mockResolvedValue(undefined);
|
||||
hoisted.inboundDeduper.releaseEvent.mockReset();
|
||||
hoisted.inboundDeduper.flush.mockReset().mockResolvedValue(undefined);
|
||||
hoisted.inboundDeduper.stop.mockReset().mockResolvedValue(undefined);
|
||||
hoisted.createMatrixInboundEventDeduper.mockReset().mockResolvedValue(hoisted.inboundDeduper);
|
||||
hoisted.createMatrixInboundEventDeduper.mockReset().mockReturnValue(hoisted.inboundDeduper);
|
||||
hoisted.backfillMatrixAuthDeviceIdAfterStartup.mockReset().mockResolvedValue(undefined);
|
||||
hoisted.runMatrixStartupMaintenance.mockReset().mockResolvedValue(undefined);
|
||||
hoisted.createMatrixRoomMessageHandler.mockReset().mockReturnValue(vi.fn());
|
||||
@@ -728,7 +724,9 @@ describe("monitorMatrixProvider", () => {
|
||||
});
|
||||
|
||||
it("releases the prepared client when startup fails before later resources exist", async () => {
|
||||
hoisted.createMatrixInboundEventDeduper.mockRejectedValue(new Error("deduper failed"));
|
||||
hoisted.createMatrixInboundEventDeduper.mockImplementation(() => {
|
||||
throw new Error("deduper failed");
|
||||
});
|
||||
|
||||
await expect(
|
||||
monitorMatrixProvider({
|
||||
@@ -737,7 +735,6 @@ describe("monitorMatrixProvider", () => {
|
||||
).rejects.toThrow("deduper failed");
|
||||
|
||||
expect(hoisted.releaseSharedClientInstance).toHaveBeenCalledWith(hoisted.client, "persist");
|
||||
expect(hoisted.inboundDeduper.stop).not.toHaveBeenCalled();
|
||||
expectLastStatusFields({
|
||||
accountId: "default",
|
||||
connected: false,
|
||||
@@ -909,9 +906,6 @@ describe("monitorMatrixProvider", () => {
|
||||
hoisted.callOrder.push("release-client");
|
||||
return true;
|
||||
});
|
||||
hoisted.inboundDeduper.stop.mockImplementation(async () => {
|
||||
hoisted.callOrder.push("stop-deduper");
|
||||
});
|
||||
|
||||
const monitorPromise = monitorMatrixProvider({ abortSignal: abortController.signal });
|
||||
await waitForCallOrderEntry("start-client");
|
||||
@@ -923,7 +917,7 @@ describe("monitorMatrixProvider", () => {
|
||||
const roomMessagePromise = onRoomMessage("!room:example.org", { event_id: "$event" });
|
||||
abortController.abort();
|
||||
await waitForCallOrderEntry("pause-client");
|
||||
expect(hoisted.callOrder).not.toContain("stop-deduper");
|
||||
expect(hoisted.callOrder).not.toContain("stop-manager");
|
||||
|
||||
if (resolveHandler === null) {
|
||||
throw new Error("expected in-flight handler to be pending");
|
||||
@@ -942,9 +936,6 @@ describe("monitorMatrixProvider", () => {
|
||||
hoisted.callOrder.indexOf("stop-manager"),
|
||||
);
|
||||
expect(hoisted.callOrder.indexOf("stop-manager")).toBeLessThan(
|
||||
hoisted.callOrder.indexOf("stop-deduper"),
|
||||
);
|
||||
expect(hoisted.callOrder.indexOf("stop-deduper")).toBeLessThan(
|
||||
hoisted.callOrder.indexOf("release-client"),
|
||||
);
|
||||
});
|
||||
|
||||
@@ -46,10 +46,7 @@ import { resolveMatrixMonitorConfig } from "./config.js";
|
||||
import { createDirectRoomTracker } from "./direct.js";
|
||||
import { registerMatrixMonitorEvents } from "./events.js";
|
||||
import { createMatrixRoomMessageHandler } from "./handler.js";
|
||||
import {
|
||||
createMatrixInboundEventDeduper,
|
||||
type MatrixInboundEventDeduper,
|
||||
} from "./inbound-dedupe.js";
|
||||
import { createMatrixInboundEventDeduper } from "./inbound-dedupe.js";
|
||||
import { shouldPromoteRecentInviteRoom } from "./recent-invite.js";
|
||||
import { createMatrixRoomInfoResolver } from "./room-info.js";
|
||||
import { resolveMatrixRoomConfig } from "./rooms.js";
|
||||
@@ -228,7 +225,6 @@ export async function monitorMatrixProvider(opts: MonitorMatrixOpts = {}): Promi
|
||||
let cleanedUp = false;
|
||||
let client: MatrixClient | null = null;
|
||||
let threadBindingManager: { accountId: string; stop: () => void } | null = null;
|
||||
let inboundDeduper: MatrixInboundEventDeduper | null = null;
|
||||
const monitorTaskRunner = createMatrixMonitorTaskRunner({
|
||||
logger,
|
||||
logVerboseMessage,
|
||||
@@ -248,7 +244,6 @@ export async function monitorMatrixProvider(opts: MonitorMatrixOpts = {}): Promi
|
||||
await monitorTaskRunner.waitForIdle();
|
||||
}
|
||||
threadBindingManager?.stop();
|
||||
await inboundDeduper?.stop();
|
||||
if (client) {
|
||||
await releaseSharedClientInstance(client, mode);
|
||||
}
|
||||
@@ -331,7 +326,7 @@ export async function monitorMatrixProvider(opts: MonitorMatrixOpts = {}): Promi
|
||||
accountId: auth.accountId,
|
||||
});
|
||||
setActiveMatrixClient(client, auth.accountId);
|
||||
inboundDeduper = await createMatrixInboundEventDeduper({
|
||||
const inboundDeduper = createMatrixInboundEventDeduper({
|
||||
auth,
|
||||
env: process.env,
|
||||
});
|
||||
|
||||
@@ -10,7 +10,6 @@ export {
|
||||
openMatrixStorageMetaStoreOptions,
|
||||
} from "./src/matrix/client/storage.js";
|
||||
export type { MatrixStorageMetadata } from "./src/matrix/client/storage.js";
|
||||
export { openMatrixInboundDedupeStoreOptions } from "./src/matrix/monitor/inbound-dedupe.js";
|
||||
export type {
|
||||
EncryptedFile,
|
||||
MatrixDeviceVerificationStatus,
|
||||
|
||||
@@ -236,7 +236,6 @@ export async function runStaleSyncReplayDedupeScenario(context: MatrixQaScenario
|
||||
});
|
||||
|
||||
await waitForMatrixInboundDedupeEntry({
|
||||
context,
|
||||
eventId: replayDriverEventId,
|
||||
roomId,
|
||||
stateDir,
|
||||
|
||||
@@ -1,35 +1,12 @@
|
||||
import { createHash } from "node:crypto";
|
||||
// Qa Matrix tests cover persisted runtime state probes.
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import {
|
||||
createPluginStateSyncKeyedStoreForTests,
|
||||
resetPluginStateStoreForTests,
|
||||
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
|
||||
import { createClaimableDedupe } from "openclaw/plugin-sdk/persistent-dedupe";
|
||||
import { resetPluginStateStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import type { MatrixQaScenarioContext } from "./scenario-runtime-shared.js";
|
||||
import { waitForMatrixInboundDedupeEntry } from "./scenario-runtime-state-files.js";
|
||||
|
||||
const dedupeStoreRuntime = {
|
||||
openMatrixInboundDedupeStoreOptions(params: { stateDir?: string }) {
|
||||
return {
|
||||
namespace: "inbound-dedupe",
|
||||
maxEntries: 20_000,
|
||||
env: { ...process.env, OPENCLAW_STATE_DIR: params.stateDir },
|
||||
};
|
||||
},
|
||||
};
|
||||
|
||||
function buildDedupeKey(params: { accountId: string; eventId: string; roomId: string }) {
|
||||
return `${params.accountId}:${createHash("sha256")
|
||||
.update(params.accountId)
|
||||
.update("\0")
|
||||
.update(params.roomId)
|
||||
.update("\0")
|
||||
.update(params.eventId)
|
||||
.digest("hex")}`;
|
||||
}
|
||||
|
||||
describe("Matrix QA persisted state probes", () => {
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
@@ -38,27 +15,30 @@ describe("Matrix QA persisted state probes", () => {
|
||||
await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { force: true, recursive: true })));
|
||||
});
|
||||
|
||||
it("observes inbound dedupe entries through the canonical plugin-state store", async () => {
|
||||
it("observes inbound dedupe entries committed through the core claimable dedupe", async () => {
|
||||
const stateDir = await mkdtemp(path.join(os.tmpdir(), "matrix-qa-dedupe-"));
|
||||
tempDirs.push(stateDir);
|
||||
const accountRoot = path.join(stateDir, "matrix", "accounts", "sut", "server", "token");
|
||||
const accountId = "sut";
|
||||
const eventId = "$event";
|
||||
const roomId = "!room:matrix-qa.test";
|
||||
const options = dedupeStoreRuntime.openMatrixInboundDedupeStoreOptions({
|
||||
stateDir: accountRoot,
|
||||
// Mirrors the matrix monitor's guard configuration so the probe is proven
|
||||
// against the exact persisted row shape the runtime writes, including the
|
||||
// account-scoped key the probe must match by suffix.
|
||||
const guard = createClaimableDedupe({
|
||||
pluginId: "matrix",
|
||||
namespacePrefix: "matrix.inbound-dedupe",
|
||||
ttlMs: 30 * 24 * 60 * 60 * 1000,
|
||||
memoryMaxSize: 100,
|
||||
stateMaxEntries: 100,
|
||||
env: { ...process.env, OPENCLAW_STATE_DIR: accountRoot },
|
||||
});
|
||||
const runtimeAccountId = "runtime-default";
|
||||
createPluginStateSyncKeyedStoreForTests("matrix", options).register(
|
||||
buildDedupeKey({ accountId: runtimeAccountId, eventId, roomId }),
|
||||
{ eventId, roomId, ts: Date.now() },
|
||||
);
|
||||
const key = `runtime-default\0${roomId}\0${eventId}`;
|
||||
await guard.claim(key);
|
||||
await guard.commit(key);
|
||||
resetPluginStateStoreForTests();
|
||||
|
||||
await expect(
|
||||
waitForMatrixInboundDedupeEntry({
|
||||
context: { sutAccountId: accountId } as MatrixQaScenarioContext,
|
||||
dedupeStoreRuntime,
|
||||
eventId,
|
||||
roomId,
|
||||
stateDir,
|
||||
|
||||
@@ -3,28 +3,21 @@ import { createHash, randomUUID } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { setTimeout as sleep } from "node:timers/promises";
|
||||
import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime";
|
||||
import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
import { loadMatrixQaE2eeRuntime } from "../../substrate/e2ee-client.js";
|
||||
import type { MatrixQaScenarioContext } from "./scenario-runtime-shared.js";
|
||||
|
||||
const MATRIX_SYNC_STORE_FILENAME = "bot-storage.json";
|
||||
const MATRIX_INBOUND_DEDUPE_FILENAME = "inbound-dedupe.json";
|
||||
const MATRIX_PLUGIN_ID = "matrix";
|
||||
const MATRIX_SYNC_CACHE_NAMESPACE = "sync-cache";
|
||||
// Mirrors the matrix plugin's core claimable-dedupe state namespace:
|
||||
// `matrix.inbound-dedupe.<hash>` (extensions/matrix monitor/inbound-dedupe.ts).
|
||||
const MATRIX_INBOUND_DEDUPE_NAMESPACE_LIKE = "matrix.inbound-dedupe.%";
|
||||
const MATRIX_STATE_POLL_INTERVAL_MS = 100;
|
||||
const MATRIX_SYNC_CACHE_MAX_ENTRIES = 20_000;
|
||||
const MATRIX_SYNC_CACHE_MAX_CHUNKS = Math.floor((MATRIX_SYNC_CACHE_MAX_ENTRIES - 1) / 2);
|
||||
// PluginState serializes this string inside a row object; 24KB leaves room for JSON escaping.
|
||||
const MATRIX_SYNC_CACHE_CHUNK_BYTES = 24_000;
|
||||
|
||||
type MatrixQaInboundDedupeStoreRuntime = {
|
||||
openMatrixInboundDedupeStoreOptions: (params: {
|
||||
env?: NodeJS.ProcessEnv;
|
||||
stateDir?: string;
|
||||
}) => OpenKeyedStoreOptions;
|
||||
};
|
||||
|
||||
type MatrixSyncStoreCursor = {
|
||||
cursor: string;
|
||||
pathname: string;
|
||||
@@ -497,49 +490,15 @@ export async function waitForMatrixSyncStoreWithCursor(params: {
|
||||
);
|
||||
}
|
||||
|
||||
function hasPersistedMatrixDedupeEntry(params: {
|
||||
parsed: unknown;
|
||||
roomId: string;
|
||||
eventId: string;
|
||||
}) {
|
||||
if (!isRecord(params.parsed) || !Array.isArray(params.parsed.entries)) {
|
||||
return false;
|
||||
}
|
||||
const expectedKey = `${params.roomId}|${params.eventId}`;
|
||||
return params.parsed.entries.some((entry) => isRecord(entry) && entry.key === expectedKey);
|
||||
}
|
||||
|
||||
function buildMatrixInboundDedupePluginStateKey(params: {
|
||||
accountId: string;
|
||||
eventId: string;
|
||||
roomId: string;
|
||||
}): string {
|
||||
const accountId = params.accountId.trim() || "sut";
|
||||
const roomId = params.roomId.trim();
|
||||
const eventId = params.eventId.trim();
|
||||
const digest = createHash("sha256")
|
||||
.update(accountId)
|
||||
.update("\0")
|
||||
.update(roomId)
|
||||
.update("\0")
|
||||
.update(eventId)
|
||||
.digest("hex");
|
||||
return `${accountId}:${digest}`;
|
||||
}
|
||||
|
||||
async function hasPersistedMatrixPluginStateDedupeEntry(params: {
|
||||
accountId: string;
|
||||
dedupeStoreRuntime?: MatrixQaInboundDedupeStoreRuntime;
|
||||
eventId: string;
|
||||
roomId: string;
|
||||
stateDir: string;
|
||||
}): Promise<string | null> {
|
||||
const entryKey = buildMatrixInboundDedupePluginStateKey({
|
||||
accountId: params.accountId,
|
||||
eventId: params.eventId,
|
||||
roomId: params.roomId,
|
||||
});
|
||||
const dedupeStoreRuntime = params.dedupeStoreRuntime ?? (await loadMatrixQaE2eeRuntime());
|
||||
// Mirrors extensions/matrix monitor/inbound-dedupe.ts: the persisted entry
|
||||
// value records the NUL-joined (account, room, event) dedupe key; suffix
|
||||
// matching keeps the probe independent of the runtime account id.
|
||||
const expectedKeySuffix = `\0${params.roomId.trim()}\0${params.eventId.trim()}`;
|
||||
const databasePaths = await findFilesByName({
|
||||
filename: "openclaw.sqlite",
|
||||
rootDir: params.stateDir,
|
||||
@@ -553,38 +512,25 @@ async function hasPersistedMatrixPluginStateDedupeEntry(params: {
|
||||
}
|
||||
for (const databasePath of databasePaths) {
|
||||
try {
|
||||
const storageRootDir = path.dirname(path.dirname(databasePath));
|
||||
const options = dedupeStoreRuntime.openMatrixInboundDedupeStoreOptions({
|
||||
stateDir: storageRootDir,
|
||||
});
|
||||
const stateRoot = options.env?.OPENCLAW_STATE_DIR?.trim();
|
||||
if (
|
||||
!stateRoot ||
|
||||
path.resolve(stateRoot, "state", "openclaw.sqlite") !== path.resolve(databasePath)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const db = new sqlite.DatabaseSync(databasePath, { readOnly: true });
|
||||
try {
|
||||
const rows = db
|
||||
.prepare(
|
||||
`SELECT entry_key AS entryKey, value_json AS valueJson
|
||||
`SELECT value_json AS valueJson
|
||||
FROM plugin_state_entries
|
||||
WHERE plugin_id = ?
|
||||
AND namespace = ?
|
||||
AND namespace LIKE ?
|
||||
AND (expires_at IS NULL OR expires_at > ?)`,
|
||||
)
|
||||
.all(MATRIX_PLUGIN_ID, options.namespace, Date.now()) as Array<{
|
||||
entryKey?: unknown;
|
||||
.all(MATRIX_PLUGIN_ID, MATRIX_INBOUND_DEDUPE_NAMESPACE_LIKE, Date.now()) as Array<{
|
||||
valueJson?: unknown;
|
||||
}>;
|
||||
const matched = rows.some((row) => {
|
||||
if (row.entryKey === entryKey) {
|
||||
return true;
|
||||
}
|
||||
const entry = parsePluginStateJson(row.valueJson);
|
||||
return (
|
||||
isRecord(entry) && entry.roomId === params.roomId && entry.eventId === params.eventId
|
||||
isRecord(entry) &&
|
||||
typeof entry.key === "string" &&
|
||||
entry.key.endsWith(expectedKeySuffix)
|
||||
);
|
||||
});
|
||||
if (matched) {
|
||||
@@ -601,8 +547,6 @@ async function hasPersistedMatrixPluginStateDedupeEntry(params: {
|
||||
}
|
||||
|
||||
export async function waitForMatrixInboundDedupeEntry(params: {
|
||||
context: MatrixQaScenarioContext;
|
||||
dedupeStoreRuntime?: MatrixQaInboundDedupeStoreRuntime;
|
||||
eventId: string;
|
||||
roomId: string;
|
||||
stateDir: string;
|
||||
@@ -611,8 +555,6 @@ export async function waitForMatrixInboundDedupeEntry(params: {
|
||||
const startedAt = Date.now();
|
||||
while (Date.now() - startedAt < params.timeoutMs) {
|
||||
const sqlitePath = await hasPersistedMatrixPluginStateDedupeEntry({
|
||||
accountId: params.context.sutAccountId ?? "sut",
|
||||
...(params.dedupeStoreRuntime ? { dedupeStoreRuntime: params.dedupeStoreRuntime } : {}),
|
||||
eventId: params.eventId,
|
||||
roomId: params.roomId,
|
||||
stateDir: params.stateDir,
|
||||
@@ -620,23 +562,6 @@ export async function waitForMatrixInboundDedupeEntry(params: {
|
||||
if (sqlitePath) {
|
||||
return sqlitePath;
|
||||
}
|
||||
const pathname = await resolveBestMatrixStateFile({
|
||||
context: params.context,
|
||||
filename: MATRIX_INBOUND_DEDUPE_FILENAME,
|
||||
stateDir: params.stateDir,
|
||||
});
|
||||
if (pathname) {
|
||||
const parsed = await readJsonFile(pathname);
|
||||
if (
|
||||
hasPersistedMatrixDedupeEntry({
|
||||
parsed,
|
||||
roomId: params.roomId,
|
||||
eventId: params.eventId,
|
||||
})
|
||||
) {
|
||||
return pathname;
|
||||
}
|
||||
}
|
||||
await sleep(MATRIX_STATE_POLL_INTERVAL_MS);
|
||||
}
|
||||
throw new Error(
|
||||
|
||||
@@ -68,22 +68,13 @@ import {
|
||||
type MatrixQaScenarioContext,
|
||||
} from "./scenarios.js";
|
||||
|
||||
function matrixInboundDedupePluginStateKey(params: {
|
||||
accountId: string;
|
||||
eventId: string;
|
||||
roomId: string;
|
||||
}): string {
|
||||
const accountId = params.accountId.trim() || "sut";
|
||||
const digest = createHash("sha256")
|
||||
.update(accountId)
|
||||
.update("\0")
|
||||
.update(params.roomId.trim())
|
||||
.update("\0")
|
||||
.update(params.eventId.trim())
|
||||
.digest("hex");
|
||||
return `${accountId}:${digest}`;
|
||||
function sha256Hex32(value: string): string {
|
||||
return createHash("sha256").update(value).digest("hex").slice(0, 32);
|
||||
}
|
||||
|
||||
// Mirrors the matrix plugin's core claimable-dedupe rows: the shared "global"
|
||||
// namespace under `matrix.inbound-dedupe.`, a hashed `k.` entry key, and a
|
||||
// `{key, seenAt}` value recording the NUL-joined (account, room, event) key.
|
||||
async function writeMatrixInboundDedupePluginStateEntry(params: {
|
||||
accountId: string;
|
||||
eventId: string;
|
||||
@@ -94,6 +85,7 @@ async function writeMatrixInboundDedupePluginStateEntry(params: {
|
||||
const databasePath = path.join(params.stateRoot, "state", "openclaw.sqlite");
|
||||
await mkdir(path.dirname(databasePath), { recursive: true });
|
||||
const db = new sqlite.DatabaseSync(databasePath);
|
||||
const eventKey = `${params.accountId.trim() || "default"}\0${params.roomId.trim()}\0${params.eventId.trim()}`;
|
||||
try {
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS plugin_state_entries (
|
||||
@@ -116,12 +108,11 @@ async function writeMatrixInboundDedupePluginStateEntry(params: {
|
||||
expires_at = excluded.expires_at
|
||||
`).run(
|
||||
"matrix",
|
||||
"inbound-dedupe",
|
||||
matrixInboundDedupePluginStateKey(params),
|
||||
`matrix.inbound-dedupe.${sha256Hex32("global")}`,
|
||||
`k.${sha256Hex32(eventKey)}`,
|
||||
JSON.stringify({
|
||||
roomId: params.roomId,
|
||||
eventId: params.eventId,
|
||||
ts: Date.now(),
|
||||
key: eventKey,
|
||||
seenAt: Date.now(),
|
||||
}),
|
||||
Date.now(),
|
||||
null,
|
||||
@@ -370,13 +361,7 @@ describe("matrix live qa scenarios", () => {
|
||||
beforeEach(() => {
|
||||
createMatrixQaClient.mockReset();
|
||||
createMatrixQaE2eeScenarioClient.mockReset();
|
||||
loadMatrixQaE2eeRuntime.mockReset().mockResolvedValue({
|
||||
openMatrixInboundDedupeStoreOptions: ({ stateDir }: { stateDir?: string }) => ({
|
||||
namespace: "inbound-dedupe",
|
||||
maxEntries: 20_000,
|
||||
env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
|
||||
}),
|
||||
});
|
||||
loadMatrixQaE2eeRuntime.mockReset();
|
||||
runMatrixQaE2eeBootstrap.mockReset();
|
||||
runMatrixQaOpenClawCli.mockReset();
|
||||
startMatrixQaOpenClawCli.mockReset();
|
||||
|
||||
Reference in New Issue
Block a user