diff --git a/.changeset/fn-7091-messaging-corruption-recovery.md b/.changeset/fn-7091-messaging-corruption-recovery.md new file mode 100644 index 0000000000..1a1cce176e --- /dev/null +++ b/.changeset/fn-7091-messaging-corruption-recovery.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Recover corrupt messaging indexes during send or report the exact repair command. +category: fix +dev: MessageStore now runs a scoped REINDEX messages retry on SQLite corruption during send. diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index ff22b388ac..fe82717047 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -2,6 +2,7 @@ import { describe, it, expect, beforeAll, beforeEach, afterEach, afterAll, vi } import { Database, createDatabase, + isSqliteCorruptionError, quickCheckSqliteFile, integrityCheckSqliteFileAsync, toJson, @@ -255,6 +256,43 @@ describe("Database", () => { await cleanupTmpDirsAsync(); }); + describe("SQLite corruption recovery helpers", () => { + it("classifies SQLite corruption errors without matching ordinary SQLite errors", () => { + const corruptByCode = Object.assign(new Error("constraint failed"), { code: "SQLITE_CORRUPT" }); + + expect(isSqliteCorruptionError(corruptByCode)).toBe(true); + expect(isSqliteCorruptionError(new Error("database disk image is malformed"))).toBe(true); + expect(isSqliteCorruptionError(new Error("corruption found reading blob from fts5 table"))).toBe(true); + expect(isSqliteCorruptionError(new Error("fts5 segment is corrupt"))).toBe(true); + expect(isSqliteCorruptionError(Object.assign(new Error("database is locked"), { code: "SQLITE_BUSY" }))).toBe(false); + expect(isSqliteCorruptionError(new Error("plain application failure"))).toBe(false); + }); + + it("reindexes messages indexes for populated disk and in-memory databases", () => { + db.prepare(` + INSERT INTO messages (id, fromId, fromType, toId, toType, content, type, read, metadata, createdAt, updatedAt) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `).run("msg-disk", "agent-1", "agent", "user-1", "user", "hello", "agent-to-user", 0, null, "2026-06-26T00:00:00.000Z", "2026-06-26T00:00:00.000Z"); + + expect(() => db.reindexMessages()).not.toThrow(); + + const memDb = new Database(fusionDir, { inMemory: true }); + try { + memDb.init(); + memDb.prepare(` + INSERT INTO messages (id, fromId, fromType, toId, toType, content, type, read, metadata, createdAt, updatedAt) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `).run("msg-memory", "agent-1", "agent", "user-1", "user", "hello", "agent-to-user", 0, null, "2026-06-26T00:00:00.000Z", "2026-06-26T00:00:00.000Z"); + + expect(() => memDb.reindexMessages()).not.toThrow(); + memDb.close(); + expect(() => memDb.reindexMessages()).not.toThrow(); + } finally { + memDb.close(); + } + }); + }); + describe("initialization", () => { it("creates the database file", () => { expect(existsSync(join(fusionDir, "fusion.db"))).toBe(true); diff --git a/packages/core/src/__tests__/message-store.test.ts b/packages/core/src/__tests__/message-store.test.ts index 21193802cf..e595f2e890 100644 --- a/packages/core/src/__tests__/message-store.test.ts +++ b/packages/core/src/__tests__/message-store.test.ts @@ -7,6 +7,10 @@ import { MessageStore } from "../message-store.js"; import { DASHBOARD_USER_ID } from "../types.js"; import type { Message, Mailbox } from "../types.js"; +function makeSqliteCorruptError(): Error & { code: string } { + return Object.assign(new Error("database disk image is malformed"), { code: "SQLITE_CORRUPT" }); +} + describe("MessageStore", () => { let store: MessageStore; let db: Database; @@ -21,6 +25,7 @@ describe("MessageStore", () => { }); afterEach(() => { + vi.restoreAllMocks(); db.close(); try { rmSync(tempDir, { recursive: true, force: true }); @@ -150,6 +155,101 @@ describe("MessageStore", () => { }).toThrow("metadata.replyTo.messageId must be a non-empty string"); }); + it("reindexes messages indexes once and retries when an insert reports SQLite corruption", () => { + const privateStore = store as unknown as { stmtInsert: { run: (...args: unknown[]) => unknown } }; + const originalInsertRun = privateStore.stmtInsert.run.bind(privateStore.stmtInsert); + let attempts = 0; + privateStore.stmtInsert = { + run: vi.fn((...args: unknown[]) => { + attempts += 1; + if (attempts === 1) { + throw makeSqliteCorruptError(); + } + return originalInsertRun(...args); + }), + }; + const reindexMessages = vi.spyOn(db, "reindexMessages"); + + const message = store.sendMessage({ + fromId: "agent-1", + fromType: "agent", + toId: "user-1", + toType: "user", + content: "Recovered send", + type: "agent-to-user", + }); + + expect(reindexMessages).toHaveBeenCalledTimes(1); + expect(attempts).toBe(2); + expect(message.id).toMatch(/^msg-/); + expect(store.getMessage(message.id)).toEqual(message); + }); + + it("throws a repair-specific remediation error when REINDEX itself reports corruption", () => { + const privateStore = store as unknown as { stmtInsert: { run: (...args: unknown[]) => unknown } }; + privateStore.stmtInsert = { + run: vi.fn(() => { + throw makeSqliteCorruptError(); + }), + }; + const reindexMessages = vi.spyOn(db, "reindexMessages").mockImplementation(() => { + throw makeSqliteCorruptError(); + }); + + let thrown: unknown; + try { + store.sendMessage({ + fromId: "agent-1", + fromType: "agent", + toId: "user-1", + toType: "user", + content: "Reindex fails", + type: "agent-to-user", + }); + } catch (error) { + thrown = error; + } + + expect(reindexMessages).toHaveBeenCalledTimes(1); + expect(thrown).toBeInstanceOf(Error); + const message = (thrown as Error).message; + expect(message).toContain("Messages store index repair failed (table=messages, db=:memory:)"); + expect(message).toContain('run "fn db --vacuum" and inspect with "PRAGMA integrity_check"'); + expect(message).not.toBe("database disk image is malformed"); + }); + + it("throws a table/database remediation error when corruption persists after successful reindex", () => { + const privateStore = store as unknown as { stmtInsert: { run: (...args: unknown[]) => unknown } }; + privateStore.stmtInsert = { + run: vi.fn(() => { + throw makeSqliteCorruptError(); + }), + }; + const reindexMessages = vi.spyOn(db, "reindexMessages"); + + let thrown: unknown; + try { + store.sendMessage({ + fromId: "agent-1", + fromType: "agent", + toId: "user-1", + toType: "user", + content: "Still corrupt", + type: "agent-to-user", + }); + } catch (error) { + thrown = error; + } + + expect(reindexMessages).toHaveBeenCalledTimes(1); + expect(thrown).toBeInstanceOf(Error); + const message = (thrown as Error).message; + expect(message).toContain("Messages store table/database corruption after REINDEX (table=messages, db=:memory:)"); + expect(message).toContain('run "fn db --vacuum" and inspect with "PRAGMA integrity_check"'); + expect(message).not.toContain('run "REINDEX messages" or "fn db --vacuum" to repair'); + expect(message).not.toBe("database disk image is malformed"); + }); + it("returns null for non-existent message", () => { const result = store.getMessage("msg-nonexistent"); expect(result).toBeNull(); diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 755e23a49c..2ead1cab8e 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -152,17 +152,33 @@ export function probeFts5(db: DatabaseSync): boolean { } } +function getSqliteErrorText(error: unknown): string { + const err = error as { message?: unknown; code?: unknown; name?: unknown } | null | undefined; + return [err?.code, err?.name, err?.message, error] + .map((part) => (typeof part === "string" ? part : "")) + .filter(Boolean) + .join(" ") + .toLowerCase(); +} + +/** + * Check whether an error appears to be a SQLite corruption/integrity failure. + */ +export function isSqliteCorruptionError(error: unknown): boolean { + const text = getSqliteErrorText(error); + return ( + text.includes("sqlite_corrupt") || + text.includes("database disk image is malformed") || + text.includes("corruption found reading blob") || + (text.includes("fts5") && text.includes("corrupt")) + ); +} + /** * Check whether an error appears to be an FTS5 corruption/integrity failure. */ export function isFts5CorruptionError(error: unknown): boolean { - const message = error instanceof Error ? error.message : String(error ?? ""); - const lower = message.toLowerCase(); - return ( - lower.includes("corruption found reading blob") || - lower.includes("database disk image is malformed") || - (lower.includes("fts5") && lower.includes("corrupt")) - ); + return isSqliteCorruptionError(error); } // ── Schema Definition ──────────────────────────────────────────────── @@ -5546,6 +5562,19 @@ export class Database { return isFts5CorruptionError(error); } + /** + * Rebuild every SQLite index attached to the messages table. + * + * FNXC:Database 2026-06-26-00:00: + * `PRAGMA quick_check` can report ok while a messages-table index is out of sync with its table; INSERT/UPDATE/DELETE then fail because SQLite must maintain idxMessagesTo, idxMessagesFrom, and idxMessagesCreatedAt. A scoped `REINDEX messages` repairs only the messaging lookup indexes without invoking whole-file recovery from the send path. + */ + reindexMessages(): void { + if (this.closed) { + return; + } + this.db.exec("REINDEX messages"); + } + /** * Read the declared columns for a table. */ diff --git a/packages/core/src/message-store.ts b/packages/core/src/message-store.ts index c9f4415014..7d91764729 100644 --- a/packages/core/src/message-store.ts +++ b/packages/core/src/message-store.ts @@ -13,7 +13,7 @@ import { EventEmitter } from "node:events"; import { randomUUID } from "node:crypto"; import type { Database } from "./db.js"; -import { fromJson, toJsonNullable } from "./db.js"; +import { fromJson, isSqliteCorruptionError, toJsonNullable } from "./db.js"; import { createLogger } from "./logger.js"; import { DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, type Message, type MessageCreateInput, type MessageFilter, type MessageType, type Mailbox, type ParticipantType } from "./types.js"; @@ -151,6 +151,57 @@ export class MessageStore extends EventEmitter { updatedAt: now, }; + this.runInsertWithMessagesIndexRecovery(message); + + this.db.bumpLastModified(); + messageStoreLog.log(`MessageStore emitting message:sent id=${message.id} type=${message.type} fromId=${message.fromId} toId=${message.toId}`); + this.emit("message:sent", message); + this.emit("message:received", message); + + if (message.toType === "agent" && this.onMessageToAgent) { + this.onMessageToAgent(message); + } + + return message; + } + + /** + * Insert a message, repairing messages-table indexes once when SQLite reports corruption. + * + * FNXC:Messaging 2026-06-26-00:00: + * Sending a message must not leak a bare `database disk image is malformed` when only the `messages` lookup indexes are corrupt. The send path runs one scoped `REINDEX messages` repair and one retry, then distinguishes repair failure from post-repair table/database corruption so operator remediation targets the right store and database path. + */ + private runInsertWithMessagesIndexRecovery(message: Message): void { + try { + this.runInsert(message); + return; + } catch (error) { + if (!isSqliteCorruptionError(error)) { + throw error; + } + + try { + this.db.reindexMessages(); + } catch (reindexError) { + if (isSqliteCorruptionError(reindexError)) { + throw this.createMessagesReindexFailureError(reindexError, error); + } + throw reindexError; + } + + try { + this.runInsert(message); + return; + } catch (retryError) { + if (isSqliteCorruptionError(retryError)) { + throw this.createMessagesPostReindexCorruptionError(retryError, error); + } + throw retryError; + } + } + } + + private runInsert(message: Message): void { this.stmtInsert.run( message.id, message.fromId, @@ -164,17 +215,22 @@ export class MessageStore extends EventEmitter { message.createdAt, message.updatedAt, ); + } - this.db.bumpLastModified(); - messageStoreLog.log(`MessageStore emitting message:sent id=${message.id} type=${message.type} fromId=${message.fromId} toId=${message.toId}`); - this.emit("message:sent", message); - this.emit("message:received", message); + private createMessagesReindexFailureError(error: unknown, originalError: unknown): Error { + const detail = error instanceof Error ? error.message : String(error); + const original = originalError instanceof Error ? originalError.message : String(originalError); + return new Error( + `Messages store index repair failed (table=messages, db=${this.db.getPath()}) — REINDEX messages could not complete; run "fn db --vacuum" and inspect with "PRAGMA integrity_check" before retrying (original insert: ${original}; reindex: ${detail})`, + ); + } - if (message.toType === "agent" && this.onMessageToAgent) { - this.onMessageToAgent(message); - } - - return message; + private createMessagesPostReindexCorruptionError(error: unknown, originalError: unknown): Error { + const detail = error instanceof Error ? error.message : String(error); + const original = originalError instanceof Error ? originalError.message : String(originalError); + return new Error( + `Messages store table/database corruption after REINDEX (table=messages, db=${this.db.getPath()}) — REINDEX messages completed but sending still failed; run "fn db --vacuum" and inspect with "PRAGMA integrity_check" for messages table or database-file damage (original insert: ${original}; retry: ${detail})`, + ); } /** @@ -280,6 +336,7 @@ export class MessageStore extends EventEmitter { if (existing.read) return existing; const now = new Date().toISOString(); + // Deferral: markAsRead updates idxMessagesTo but is not the reported fn_send_message failure path; the reusable recovery helper keeps this path ready for a follow-up without expanding this focused fix. this.stmtUpdateRead.run(now, messageId); this.db.bumpLastModified(); @@ -310,7 +367,7 @@ export class MessageStore extends EventEmitter { `).get(...participantIds, ownerType) as { count: number } | undefined; const count = unreadRow?.count ?? 0; - // Mark all as read + // Deferral: markAllAsRead updates idxMessagesTo, but sendMessage is the operator-blocking repro; leave bulk read-state recovery for a dedicated follow-up if observed. this.db.prepare(` UPDATE messages SET read = 1, updatedAt = ? WHERE ${toIdPredicate} AND toType = ? AND read = 0 `).run(now, ...participantIds, ownerType); @@ -331,6 +388,7 @@ export class MessageStore extends EventEmitter { throw new Error(`Message ${id} not found`); } + // Deferral: deleteMessage mutates messages indexes but is not on the fn_send_message delivery path; avoid adding extra retry semantics outside the scoped send repair. this.stmtDelete.run(id); this.db.bumpLastModified(); this.emit("message:deleted", id); @@ -349,6 +407,7 @@ export class MessageStore extends EventEmitter { const cutoff = new Date(Date.now() - maxAgeMs).toISOString(); const deletedIds = this.db.transaction(() => { + // Deferral: cleanupOldMessages deletes through messages indexes in retention maintenance, not interactive messaging send; keep recovery limited to the reported INSERT path. const rows = this.db.prepare(` DELETE FROM messages WHERE updatedAt < ?