FN-7091: recover messaging index corruption on send
Repair the messaging send path when SQLite reports corrupt messages-table indexes despite quick_check passing. - Detect SQLite corruption errors from codes, names, and messages instead of only FTS5 text. - Add a scoped REINDEX messages repair and single retry around message inserts. - Return actionable repair guidance when reindexing or the retry still fails. - Cover send recovery and corruption classification with regression tests, plus a patch changeset. Files changed: .changeset/fn-7091-messaging-corruption-recovery.md | 7 ++ packages/core/src/__tests__/db.test.ts | 38 ++++++++ packages/core/src/__tests__/message-store.test.ts | 100 +++++++++++++++++++++ packages/core/src/db.ts | 43 +++++++-- packages/core/src/message-store.ts | 81 ++++++++++++++--- 5 files changed, 251 insertions(+), 18 deletions(-) Fusion-Task-Id: FN-7091 Fusion-Task-Lineage: 47bcbe17-b428-4702-9835-59cb9e6ec164 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-7091-messaging-corruption-recovery.md
Normal file
7
.changeset/fn-7091-messaging-corruption-recovery.md
Normal file
@@ -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.
|
||||
@@ -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);
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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.
|
||||
*/
|
||||
|
||||
@@ -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<MessageStoreEvents> {
|
||||
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<MessageStoreEvents> {
|
||||
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<MessageStoreEvents> {
|
||||
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<MessageStoreEvents> {
|
||||
`).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<MessageStoreEvents> {
|
||||
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<MessageStoreEvents> {
|
||||
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 < ?
|
||||
|
||||
Reference in New Issue
Block a user