From 080cbb16938eec0c45c063be09f2a0b3f81ef5e9 Mon Sep 17 00:00:00 2001 From: Benjamin Badejo Date: Wed, 22 Jul 2026 22:04:42 +0300 Subject: [PATCH] fix(matrix): fail closed on incomplete dedupe census --- extensions/matrix/doctor-contract-api.test.ts | 62 ++++++++++++++++++- extensions/matrix/doctor-contract-api.ts | 3 + .../monitor/inbound-dedupe-migration.ts | 35 ++++++++--- 3 files changed, 91 insertions(+), 9 deletions(-) diff --git a/extensions/matrix/doctor-contract-api.test.ts b/extensions/matrix/doctor-contract-api.test.ts index 3ce8363552a3..56a0c34d829a 100644 --- a/extensions/matrix/doctor-contract-api.test.ts +++ b/extensions/matrix/doctor-contract-api.test.ts @@ -2,6 +2,7 @@ import "fake-indexeddb/auto"; import { createHash } from "node:crypto"; import fs from "node:fs"; +import fsPromises from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { DatabaseSync } from "node:sqlite"; @@ -15,7 +16,7 @@ import { resetPluginStateStoreForTests, } 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 { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { stateMigrations } from "./doctor-contract-api.js"; import { SqliteBackedMatrixSyncStore } from "./src/matrix/client/file-sync-store.js"; import { openMatrixStorageMetaStoreOptions } from "./src/matrix/client/storage.js"; @@ -86,6 +87,7 @@ describe("matrix doctor contract state migrations", () => { afterEach(async () => { await clearAllIndexedDbState({ databasePrefix: DOCTOR_IDB_DATABASE_PREFIX }); + vi.restoreAllMocks(); resetPluginStateStoreForTests(); for (const dir of tempDirs.splice(0)) { fs.rmSync(dir, { recursive: true, force: true }); @@ -797,6 +799,64 @@ describe("matrix doctor contract state migrations", () => { }); }); + it("withholds completion after a directory read failure and imports the source on retry", async () => { + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-doctor-")); + tempDirs.push(stateDir); + const blockedDir = path.join(stateDir, "matrix", "accounts", "home"); + const jsonRoot = path.join(blockedDir, "matrix.example.org__bot", "token-a"); + const jsonPath = path.join(jsonRoot, "inbound-dedupe.json"); + const roomId = "!room:example.org"; + const eventId = "$found-on-retry"; + fs.mkdirSync(jsonRoot, { recursive: true }); + fs.writeFileSync( + jsonPath, + JSON.stringify({ + version: 1, + entries: [{ key: `${roomId}|${eventId}`, ts: Date.now() - 60_000 }], + }), + ); + fs.writeFileSync( + path.join(jsonRoot, "storage-meta.json"), + JSON.stringify({ accountId: "home", userId: "@home:example.org" }), + ); + + const originalReaddir = fsPromises.readdir.bind(fsPromises); + const readdirSpy = vi.spyOn(fsPromises, "readdir").mockImplementation(async (...args) => { + if (path.resolve(String(args[0])) === path.resolve(blockedDir)) { + throw Object.assign(new Error("injected directory read failure"), { code: "EACCES" }); + } + return originalReaddir(...args); + }); + const migration = migrationById("matrix-inbound-dedupe-to-claimable-dedupe"); + const params = createMigrationParams(stateDir); + + await expect(migration.migrateLegacyState(params)).resolves.toEqual({ + changes: [], + warnings: [ + `Failed scanning Matrix inbound dedupe sources under ${blockedDir}: Error: injected directory read failure`, + ], + }); + await expect(migration.detectLegacyState(params)).resolves.toEqual({ + preview: ["Matrix inbound dedupe legacy sources need a one-time migration scan"], + }); + + readdirSpy.mockRestore(); + await expect(migration.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: [], + }); + const deduper = createMatrixInboundEventDeduper({ + auth: { accountId: "home" }, + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + await expect(deduper.claim({ roomId, eventId })).resolves.toEqual({ kind: "duplicate" }); + await expect(migration.detectLegacyState(params)).resolves.toBeNull(); + }); + it("ignores an invalid legacy-scan completion receipt", async () => { const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-doctor-")); tempDirs.push(stateDir); diff --git a/extensions/matrix/doctor-contract-api.ts b/extensions/matrix/doctor-contract-api.ts index c7ba542aef10..70ac22edba1e 100644 --- a/extensions/matrix/doctor-contract-api.ts +++ b/extensions/matrix/doctor-contract-api.ts @@ -321,6 +321,9 @@ export const stateMigrations: PluginDoctorStateMigration[] = [ return { changes, warnings }; } const sources = await collectMatrixInboundDedupeSources(params.stateDir); + if (sources.status === "incomplete") { + warnings.push(...sources.warnings); + } const recordCompletionIfClean = async (verifyRetirement = false) => { if (warnings.length > 0) { diff --git a/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts b/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts index abd5e72c6fb0..9ac5ff4aae81 100644 --- a/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts +++ b/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts @@ -54,6 +54,15 @@ export type LegacyInboundDedupeMarker = { ts: number; }; +type MatrixInboundDedupeSourceRoots = { + sqliteRoots: string[]; + jsonRoots: string[]; +}; + +export type MatrixInboundDedupeSourceCensus = + | ({ status: "complete" } & MatrixInboundDedupeSourceRoots) + | ({ status: "incomplete"; warnings: string[] } & MatrixInboundDedupeSourceRoots); + type LegacySqliteRow = { namespace: string; entry_key: string; @@ -109,18 +118,22 @@ function loadNodeSqlite(): typeof import("node:sqlite") { return req("node:sqlite") as typeof import("node:sqlite"); } -export async function collectMatrixInboundDedupeSources(stateDir: string): Promise<{ - sqliteRoots: string[]; - jsonRoots: string[]; -}> { +export async function collectMatrixInboundDedupeSources( + stateDir: string, +): Promise { const matrixRoot = path.join(stateDir, "matrix"); const sqliteRoots = new Set(); const jsonRoots = new Set(); - async function visit(dir: string): Promise { + const warnings: string[] = []; + async function visit(dir: string, allowMissing = false): Promise { let entries: Dirent[]; try { entries = await fs.readdir(dir, { withFileTypes: true }); - } catch { + } catch (err) { + if (allowMissing && (err as NodeJS.ErrnoException).code === "ENOENT") { + return; + } + warnings.push(`Failed scanning Matrix inbound dedupe sources under ${dir}: ${String(err)}`); return; } for (const entry of entries) { @@ -139,13 +152,16 @@ export async function collectMatrixInboundDedupeSources(stateDir: string): Promi } } } - await visit(matrixRoot); + await visit(matrixRoot, true); const matrixRootResolved = path.resolve(matrixRoot); const isAccountRoot = (root: string) => path.resolve(root) !== matrixRootResolved; - return { + const roots = { sqliteRoots: [...sqliteRoots].filter(isAccountRoot).toSorted(), jsonRoots: [...jsonRoots].filter(isAccountRoot).toSorted(), }; + return warnings.length === 0 + ? { status: "complete", ...roots } + : { status: "incomplete", ...roots, warnings }; } function selectLegacySqliteRows(db: DatabaseSync): LegacySqliteRow[] { @@ -282,6 +298,9 @@ export async function retireLegacyInboundDedupeSqliteRows(storageRootDir: string export async function verifyMatrixInboundDedupeSourcesRetired(stateDir: string): Promise { const warnings: string[] = []; const remaining = await collectMatrixInboundDedupeSources(stateDir); + if (remaining.status === "incomplete") { + warnings.push(...remaining.warnings); + } for (const storageRootDir of remaining.sqliteRoots) { try { if ((await readLegacyInboundDedupeSqliteSource(storageRootDir)).legacyRowCount > 0) {