From a7fada4b61b6bc76fe031877e524b64aa293a97a Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Tue, 16 Jun 2026 12:45:22 +0200 Subject: [PATCH] fix(memory): invalidate stale shadow indexes --- .../memory-core/src/memory/index.test.ts | 14 +- .../memory-core/src/memory/manager-db.test.ts | 97 ++++++++++++++ .../memory-core/src/memory/manager-db.ts | 124 +++++++++++++++++- .../src/memory/manager-reindex-lock.ts | 46 ++++--- .../src/memory/manager-sync-ops.ts | 41 ++---- .../shared/legacy-config-migrate.test.ts | 2 +- 6 files changed, 269 insertions(+), 55 deletions(-) create mode 100644 extensions/memory-core/src/memory/manager-db.test.ts diff --git a/extensions/memory-core/src/memory/index.test.ts b/extensions/memory-core/src/memory/index.test.ts index 73b7b2e4bb20..7b8da6034d68 100644 --- a/extensions/memory-core/src/memory/index.test.ts +++ b/extensions/memory-core/src/memory/index.test.ts @@ -341,15 +341,17 @@ describe("memory index", () => { }; } ).db; - (manager as unknown as { resetIndex: () => void }).resetIndex(); - const embeddingCacheTable = db - .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?") - .get("embedding_cache"); - if (embeddingCacheTable?.name === "embedding_cache") { - db.exec("DELETE FROM embedding_cache"); + for (const table of ["files", "chunks", "embedding_cache", "chunks_fts", "chunks_vec"]) { + const existingTable = db + .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?") + .get(table); + if (existingTable?.name === table) { + db.exec(`DELETE FROM ${table}`); + } } (manager as unknown as { dirty: boolean }).dirty = true; (manager as unknown as { sessionsDirty: boolean }).sessionsDirty = false; + (manager as unknown as { sessionsDirtyFiles: Set }).sessionsDirtyFiles.clear(); } type TestCfg = Parameters[0]["cfg"]; diff --git a/extensions/memory-core/src/memory/manager-db.test.ts b/extensions/memory-core/src/memory/manager-db.test.ts new file mode 100644 index 000000000000..2188f99afe88 --- /dev/null +++ b/extensions/memory-core/src/memory/manager-db.test.ts @@ -0,0 +1,97 @@ +// Memory Core tests cover shared agent database publication and shadow cleanup. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { ensureMemoryIndexSchema } from "openclaw/plugin-sdk/memory-core-host-engine-storage"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { cleanupAgedMemoryReindexTempFiles, publishMemoryDatabaseTables } from "./manager-db.js"; +import { acquireMemoryReindexLock } from "./manager-reindex-lock.js"; + +function ensureTestMemorySchema(db: DatabaseSync): void { + ensureMemoryIndexSchema({ + db, + embeddingCacheTable: "embedding_cache", + cacheEnabled: true, + ftsTable: "chunks_fts", + ftsEnabled: false, + }); +} + +async function expectPathMissing(targetPath: string): Promise { + await expect(fs.access(targetPath)).rejects.toThrow("ENOENT"); +} + +describe("memory manager database publication", () => { + let fixtureRoot = ""; + + beforeEach(async () => { + fixtureRoot = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-memory-db-")); + }); + + afterEach(async () => { + await fs.rm(fixtureRoot, { recursive: true, force: true }); + }); + + it("removes a stale vector table when the shadow index has no vectors", () => { + const targetPath = path.join(fixtureRoot, "target.sqlite"); + const sourcePath = path.join(fixtureRoot, "source.sqlite"); + const targetDb = new DatabaseSync(targetPath); + const sourceDb = new DatabaseSync(sourcePath); + try { + ensureTestMemorySchema(targetDb); + ensureTestMemorySchema(sourceDb); + targetDb.exec("CREATE TABLE chunks_vec (id TEXT PRIMARY KEY, embedding BLOB)"); + targetDb.prepare("INSERT INTO chunks_vec (id, embedding) VALUES (?, ?)").run("stale", "[]"); + sourceDb.close(); + + publishMemoryDatabaseTables({ + targetDb, + sourcePath, + metaKey: "memory_index_meta", + }); + + expect( + targetDb + .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'chunks_vec'") + .get(), + ).toBeUndefined(); + } finally { + try { + sourceDb.close(); + } catch {} + targetDb.close(); + } + }); + + it("removes aged orphan shadows but preserves young and locked shadows", async () => { + const databasePath = path.join(fixtureRoot, "agent.sqlite"); + const database = new DatabaseSync(databasePath); + database.close(); + const oldShadow = `${databasePath}.memory-reindex-11111111-2222-3333-4444-555555555555`; + const youngShadow = `${databasePath}.memory-reindex-aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee`; + const lockedShadow = `${databasePath}.memory-reindex-99999999-aaaa-bbbb-cccc-dddddddddddd`; + const old = new Date(Date.now() - 48 * 60 * 60_000); + + for (const suffix of ["", "-wal", "-journal"]) { + await fs.writeFile(`${oldShadow}${suffix}`, "orphan"); + await fs.utimes(`${oldShadow}${suffix}`, old, old); + } + await fs.writeFile(youngShadow, "active"); + await fs.writeFile(lockedShadow, "locked"); + await fs.utimes(lockedShadow, old, old); + + const lock = acquireMemoryReindexLock(databasePath); + cleanupAgedMemoryReindexTempFiles(databasePath); + await expect(fs.access(lockedShadow)).resolves.toBeUndefined(); + lock.release(); + + cleanupAgedMemoryReindexTempFiles(databasePath); + + await expectPathMissing(oldShadow); + await expectPathMissing(`${oldShadow}-wal`); + await expectPathMissing(`${oldShadow}-journal`); + await expectPathMissing(lockedShadow); + await expect(fs.access(youngShadow)).resolves.toBeUndefined(); + }); +}); diff --git a/extensions/memory-core/src/memory/manager-db.ts b/extensions/memory-core/src/memory/manager-db.ts index d882998f54ea..fc30b474dac7 100644 --- a/extensions/memory-core/src/memory/manager-db.ts +++ b/extensions/memory-core/src/memory/manager-db.ts @@ -12,9 +12,45 @@ import { ensureOpenClawAgentDatabaseSchema, runSqliteImmediateTransactionSync, } from "openclaw/plugin-sdk/sqlite-runtime"; +import { + tryAcquireMemoryReindexLock, + type MemoryReindexLockHandle, +} from "./manager-reindex-lock.js"; const MEMORY_REINDEX_SCHEMA = "memory_reindex"; const MEMORY_DATABASE_FILE_SUFFIXES = ["", "-wal", "-shm", "-journal"] as const; +const MEMORY_REINDEX_ENTRY_SUFFIXES = ["-wal", "-shm", "-journal", ""] as const; +const MEMORY_REINDEX_UUID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; +const MEMORY_REINDEX_ORPHAN_MIN_AGE_MS = 24 * 60 * 60_000; + +function resolveMemoryReindexBaseName( + databaseBaseName: string, + entryName: string, +): string | undefined { + for (const suffix of MEMORY_REINDEX_ENTRY_SUFFIXES) { + if (!entryName.endsWith(suffix)) { + continue; + } + const baseName = entryName.slice(0, entryName.length - suffix.length); + const prefix = `${databaseBaseName}.memory-reindex-`; + if ( + baseName.startsWith(prefix) && + MEMORY_REINDEX_UUID_PATTERN.test(baseName.slice(prefix.length)) + ) { + return baseName; + } + } + return undefined; +} + +function isRegularFile(filePath: string): boolean { + try { + return fs.statSync(filePath).isFile(); + } catch { + return false; + } +} function tableExists(db: DatabaseSync, schema: string, tableName: string): boolean { const row = db @@ -34,13 +70,17 @@ function replaceVirtualTable(params: { db: DatabaseSync; tableName: "chunks_fts" | "chunks_vec"; columns: string; - dropWhenSourceMissing?: boolean; + ignoreDropErrorWhenSourceMissing?: boolean; }): void { const { db, tableName, columns } = params; const createSql = readTableSql(db, MEMORY_REINDEX_SCHEMA, tableName); if (!createSql) { - if (params.dropWhenSourceMissing !== false) { + try { db.exec(`DROP TABLE IF EXISTS main.${tableName}`); + } catch (err) { + if (!params.ignoreDropErrorWhenSourceMissing) { + throw err; + } } return; } @@ -104,8 +144,9 @@ export function publishMemoryDatabaseTables(params: { tableName: "chunks_vec", columns: "id, embedding", // A vector-disabled connection may not have sqlite-vec loaded and cannot - // drop an old virtual table. It is unused and can remain until vec loads. - dropWhenSourceMissing: false, + // drop an old virtual table. Missing vector metadata forces a strict + // rebuild before that table can be queried again. + ignoreDropErrorWhenSourceMissing: true, }); }); } finally { @@ -120,6 +161,81 @@ export function removeMemoryDatabaseFiles(dbPath: string): void { } } +/** Remove crash-left shadow databases only when no full reindex is active. */ +export function cleanupAgedMemoryReindexTempFiles(dbPath: string, nowMs = Date.now()): void { + if (!isRegularFile(dbPath)) { + return; + } + + let reindexLock: MemoryReindexLockHandle | undefined; + try { + reindexLock = tryAcquireMemoryReindexLock(dbPath); + } catch { + return; + } + if (!reindexLock) { + return; + } + + try { + const dir = path.dirname(dbPath); + const databaseBaseName = path.basename(dbPath); + const shadowBaseNames = new Set(); + let entries: fs.Dirent[]; + try { + entries = fs.readdirSync(dir, { withFileTypes: true }); + } catch { + return; + } + + for (const entry of entries) { + if (!entry.isFile()) { + continue; + } + const shadowBaseName = resolveMemoryReindexBaseName(databaseBaseName, entry.name); + if (shadowBaseName) { + shadowBaseNames.add(shadowBaseName); + } + } + + for (const shadowBaseName of shadowBaseNames) { + const filePaths = MEMORY_DATABASE_FILE_SUFFIXES.map((suffix) => + path.join(dir, `${shadowBaseName}${suffix}`), + ); + const stats: fs.Stats[] = []; + let hasUnknownFileState = false; + for (const filePath of filePaths) { + try { + stats.push(fs.statSync(filePath)); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") { + hasUnknownFileState = true; + break; + } + } + } + if (hasUnknownFileState || stats.length === 0) { + continue; + } + if ( + nowMs - Math.max(...stats.map((stat) => stat.mtimeMs)) < + MEMORY_REINDEX_ORPHAN_MIN_AGE_MS + ) { + continue; + } + for (const filePath of filePaths) { + try { + fs.rmSync(filePath, { force: true }); + } catch {} + } + } + } finally { + try { + reindexLock.release(); + } catch {} + } +} + export function openMemoryDatabaseAtPath( dbPath: string, allowExtension: boolean, diff --git a/extensions/memory-core/src/memory/manager-reindex-lock.ts b/extensions/memory-core/src/memory/manager-reindex-lock.ts index 3e9f38b69923..5907a87df2ba 100644 --- a/extensions/memory-core/src/memory/manager-reindex-lock.ts +++ b/extensions/memory-core/src/memory/manager-reindex-lock.ts @@ -33,22 +33,7 @@ function openMemoryLockDatabase(lockPath: string): DatabaseSync { } } -/** Acquire an exclusive build lock without locking readers of the live agent database. */ -export function acquireMemoryReindexLock(dbPath: string): MemoryReindexLockHandle { - const lockPath = resolveMemoryReindexLockPath(dbPath); - const lockDb = openMemoryLockDatabase(lockPath); - try { - lockDb.exec("BEGIN EXCLUSIVE"); - } catch (err) { - lockDb.close(); - if (isSqliteBusyError(err)) { - throw Object.assign( - new Error(`Memory reindex lock is held at ${lockPath}; another reindex is active.`), - { code: "SQLITE_BUSY" }, - ); - } - throw err; - } +function createMemoryReindexLockHandle(lockDb: DatabaseSync): MemoryReindexLockHandle { return { release: () => { let releaseError: unknown; @@ -68,3 +53,32 @@ export function acquireMemoryReindexLock(dbPath: string): MemoryReindexLockHandl }, }; } + +/** Try to acquire the build lock without locking readers of the live agent database. */ +export function tryAcquireMemoryReindexLock(dbPath: string): MemoryReindexLockHandle | undefined { + const lockDb = openMemoryLockDatabase(resolveMemoryReindexLockPath(dbPath)); + try { + lockDb.exec("BEGIN EXCLUSIVE"); + } catch (err) { + lockDb.close(); + if (isSqliteBusyError(err)) { + return undefined; + } + throw err; + } + return createMemoryReindexLockHandle(lockDb); +} + +/** Acquire an exclusive build lock without locking readers of the live agent database. */ +export function acquireMemoryReindexLock(dbPath: string): MemoryReindexLockHandle { + const lock = tryAcquireMemoryReindexLock(dbPath); + if (lock) { + return lock; + } + throw Object.assign( + new Error( + `Memory reindex lock is held at ${resolveMemoryReindexLockPath(dbPath)}; another reindex is active.`, + ), + { code: "SQLITE_BUSY" }, + ); +} diff --git a/extensions/memory-core/src/memory/manager-sync-ops.ts b/extensions/memory-core/src/memory/manager-sync-ops.ts index b225e0ecd4ac..157423157a6e 100644 --- a/extensions/memory-core/src/memory/manager-sync-ops.ts +++ b/extensions/memory-core/src/memory/manager-sync-ops.ts @@ -47,6 +47,7 @@ import { type EmbeddingProviderRuntime, } from "./embeddings.js"; import { + cleanupAgedMemoryReindexTempFiles, closeMemoryDatabase, openMemoryDatabaseAtPath, publishMemoryDatabaseTables, @@ -608,6 +609,12 @@ export abstract class MemoryManagerSyncOps { return false; } if (ready && typeof dimensions === "number" && dimensions > 0) { + // Another process may have published a vectorless index while this + // connection retained the previous dimensions in memory. + const persistedMeta = this.readMeta(); + if (persistedMeta && persistedMeta.vectorDims !== this.vector.dims) { + this.vector.dims = persistedMeta.vectorDims; + } this.ensureVectorTable(dimensions); } return ready; @@ -645,8 +652,8 @@ export abstract class MemoryManagerSyncOps { if (this.vector.dims === dimensions) { return; } - if (this.vector.dims && this.vector.dims !== dimensions) { - this.dropVectorTable(); + if (!this.dropVectorTable()) { + throw new Error(`Failed to reset ${VECTOR_TABLE} before rebuilding vector dimensions`); } this.db.exec( `CREATE VIRTUAL TABLE IF NOT EXISTS ${VECTOR_TABLE} USING vec0(\n` + @@ -657,12 +664,14 @@ export abstract class MemoryManagerSyncOps { this.vector.dims = dimensions; } - private dropVectorTable(): void { + private dropVectorTable(): boolean { try { this.db.exec(`DROP TABLE IF EXISTS ${VECTOR_TABLE}`); + return true; } catch (err) { const message = formatErrorMessage(err); log.debug(`Failed to drop ${VECTOR_TABLE}: ${message}`); + return false; } } @@ -2442,6 +2451,7 @@ export abstract class MemoryManagerSyncOps { this.vectorReady = originalState.vectorReady; }; try { + cleanupAgedMemoryReindexTempFiles(dbPath); reindexLock = acquireMemoryReindexLock(dbPath); tempDb = openMemoryDatabaseAtPath(tempDbPath, this.settings.store.vector.enabled); this.db = tempDb; @@ -2562,31 +2572,6 @@ export abstract class MemoryManagerSyncOps { } } - private resetIndex() { - this.db.exec(`DELETE FROM files`); - this.db.exec(`DELETE FROM chunks`); - if (this.fts.enabled && this.fts.available) { - try { - this.db.exec(`DROP TABLE IF EXISTS ${FTS_TABLE}`); - } catch {} - } - this.ensureSchema(); - if (this.vector.enabled && this.vector.available) { - try { - this.db.exec(`DELETE FROM ${VECTOR_TABLE}`); - } catch { - this.dropVectorTable(); - this.vector.dims = undefined; - this.vector.available = null; - this.vectorReady = null; - } - } else { - this.dropVectorTable(); - this.vector.dims = undefined; - } - this.sessionsDirtyFiles.clear(); - } - protected readMeta(): MemoryIndexMeta | null { const row = this.db.prepare(`SELECT value FROM meta WHERE key = ?`).get(META_KEY) as | { value: string } diff --git a/src/commands/doctor/shared/legacy-config-migrate.test.ts b/src/commands/doctor/shared/legacy-config-migrate.test.ts index 21a24dd54eb2..fe6ce1e76d7c 100644 --- a/src/commands/doctor/shared/legacy-config-migrate.test.ts +++ b/src/commands/doctor/shared/legacy-config-migrate.test.ts @@ -96,7 +96,7 @@ describe("legacy memory search config migrate", () => { }, }); - expect(res.config?.memorySearch).toBeUndefined(); + expect((res.config as Record | undefined)?.memorySearch).toBeUndefined(); expect(res.config?.agents?.defaults?.memorySearch?.store).toEqual({ fts: { tokenizer: "trigram" }, vector: { enabled: false },