fix: prevent SQLite B-tree corruption with WAL tuning and batched agent logs
The agentLogEntries table was doing individual auto-committed INSERTs, creating extreme WAL pressure that caused recurring B-tree corruption. This adds three WAL PRAGMAs (aggressive autocheckpoint, size limit, synchronous=NORMAL), a write-behind buffer for agent log entries, and a startup integrity check with auto-recovery attempt. Refs: Runfusion/Fusion#24 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Timothy Laurent
@@ -57,7 +57,7 @@ describe("Database", () => {
|
||||
const journalSizeLimit = db.prepare("PRAGMA journal_size_limit").get() as { journal_size_limit: number };
|
||||
|
||||
expect(synchronous.synchronous).toBe(1); // NORMAL
|
||||
expect(autoCheckpoint.wal_autocheckpoint).toBe(1000);
|
||||
expect(autoCheckpoint.wal_autocheckpoint).toBe(100);
|
||||
expect(journalSizeLimit.journal_size_limit).toBe(4_194_304);
|
||||
});
|
||||
|
||||
@@ -184,14 +184,63 @@ describe("Database", () => {
|
||||
it("does not overwrite existing config on re-init", () => {
|
||||
// Update the config
|
||||
db.prepare("UPDATE config SET nextId = 42 WHERE id = 1").run();
|
||||
|
||||
|
||||
// Re-init
|
||||
db.init();
|
||||
|
||||
|
||||
// Should keep updated value
|
||||
const row = db.prepare("SELECT nextId FROM config WHERE id = 1").get() as any;
|
||||
expect(row.nextId).toBe(42);
|
||||
});
|
||||
|
||||
it("sets wal_autocheckpoint to 100", () => {
|
||||
const row = db.prepare("PRAGMA wal_autocheckpoint").get() as { wal_autocheckpoint: number };
|
||||
expect(row.wal_autocheckpoint).toBe(100);
|
||||
});
|
||||
|
||||
it("sets journal_size_limit to 4 MB", () => {
|
||||
const row = db.prepare("PRAGMA journal_size_limit").get() as { journal_size_limit: number };
|
||||
expect(row.journal_size_limit).toBe(4194304);
|
||||
});
|
||||
|
||||
it("sets synchronous to NORMAL (1)", () => {
|
||||
const row = db.prepare("PRAGMA synchronous").get() as { synchronous: number };
|
||||
expect(row.synchronous).toBe(1); // NORMAL = 1
|
||||
});
|
||||
|
||||
it("sets busy_timeout to 5000ms", () => {
|
||||
const row = db.prepare("PRAGMA busy_timeout").get() as Record<string, number>;
|
||||
// node:sqlite returns PRAGMA results as objects; the key name varies
|
||||
const value = Object.values(row)[0];
|
||||
expect(value).toBe(5000);
|
||||
});
|
||||
|
||||
it("skips WAL PRAGMAs for in-memory databases", () => {
|
||||
const memDb = new Database(":memory:", { inMemory: true });
|
||||
memDb.init();
|
||||
// journal_mode for :memory: is "memory", not "wal"
|
||||
const row = memDb.prepare("PRAGMA journal_mode").get() as { journal_mode: string };
|
||||
expect(row.journal_mode).toBe("memory");
|
||||
memDb.close();
|
||||
});
|
||||
});
|
||||
|
||||
describe("startup integrity check", () => {
|
||||
it("passes silently on a healthy database", () => {
|
||||
// db was already init'd in beforeEach — no warning means pass
|
||||
const result = db.prepare("PRAGMA integrity_check").get() as { integrity_check: string };
|
||||
expect(result.integrity_check).toBe("ok");
|
||||
});
|
||||
|
||||
it("init completes without throwing even on a fresh database", () => {
|
||||
const freshDir = makeTmpDir();
|
||||
const freshFusionDir = join(freshDir, ".fusion");
|
||||
const freshDb = new Database(freshFusionDir);
|
||||
|
||||
// init includes the integrity check — should not throw
|
||||
expect(() => freshDb.init()).not.toThrow();
|
||||
freshDb.close();
|
||||
});
|
||||
});
|
||||
|
||||
describe("change detection", () => {
|
||||
|
||||
@@ -4286,6 +4286,7 @@ Task with acceptance criteria
|
||||
|
||||
await store.appendAgentLog(task.id, "Hello world", "text");
|
||||
await store.appendAgentLog(task.id, "Read", "tool");
|
||||
(store as any).flushAgentLogBuffer();
|
||||
|
||||
const rows = (store as any).db.prepare(`
|
||||
SELECT taskId, text, type FROM agentLogEntries
|
||||
@@ -4773,6 +4774,7 @@ Task with acceptance criteria
|
||||
it("deleting a task cascades agent log entry deletion", async () => {
|
||||
const task = await createTestTask();
|
||||
await store.appendAgentLog(task.id, "cascade me", "text");
|
||||
(store as any).flushAgentLogBuffer();
|
||||
|
||||
const before = (store as any).db.prepare(
|
||||
"SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?",
|
||||
@@ -4872,6 +4874,120 @@ Task with acceptance criteria
|
||||
).get("agentLogLegacyFileImportVersion") as { value: string } | undefined;
|
||||
expect(migrationRow?.value).toBe("1");
|
||||
});
|
||||
|
||||
describe("agent log buffering", () => {
|
||||
it("buffers entries and flushes in a single transaction when buffer is full", async () => {
|
||||
const task = await createTestTask();
|
||||
|
||||
// Fill the buffer to its max size (50)
|
||||
for (let i = 0; i < 50; i++) {
|
||||
await store.appendAgentLog(task.id, `entry ${i}`, "text");
|
||||
}
|
||||
|
||||
// All 50 should be in the DB now (auto-flush at buffer size)
|
||||
const count = await store.getAgentLogCount(task.id);
|
||||
expect(count).toBe(50);
|
||||
});
|
||||
|
||||
it("auto-flushes buffered entries when getAgentLogs is called", async () => {
|
||||
const task = await createTestTask();
|
||||
|
||||
// Write fewer than BUFFER_SIZE entries — these stay buffered
|
||||
await store.appendAgentLog(task.id, "buffered 1", "text");
|
||||
await store.appendAgentLog(task.id, "buffered 2", "text");
|
||||
|
||||
// getAgentLogs triggers a flush
|
||||
const logs = await store.getAgentLogs(task.id);
|
||||
expect(logs).toHaveLength(2);
|
||||
expect(logs[0].text).toBe("buffered 1");
|
||||
expect(logs[1].text).toBe("buffered 2");
|
||||
});
|
||||
|
||||
it("auto-flushes buffered entries when getAgentLogCount is called", async () => {
|
||||
const task = await createTestTask();
|
||||
|
||||
await store.appendAgentLog(task.id, "counted", "text");
|
||||
const count = await store.getAgentLogCount(task.id);
|
||||
expect(count).toBe(1);
|
||||
});
|
||||
|
||||
it("auto-flushes before deleteTask so FK cascade finds the rows", async () => {
|
||||
const task = await createTestTask();
|
||||
|
||||
await store.appendAgentLog(task.id, "to be cascaded", "text");
|
||||
// deleteTask should flush first, then cascade-delete the entry
|
||||
await store.deleteTask(task.id);
|
||||
|
||||
const after = (store as any).db.prepare(
|
||||
"SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?",
|
||||
).get(task.id) as { count: number };
|
||||
expect(after.count).toBe(0);
|
||||
});
|
||||
|
||||
it("flushes remaining entries on close without throwing", async () => {
|
||||
// Disk-backed store required — in-memory data doesn't survive close+reopen
|
||||
store.close();
|
||||
store = new TaskStore(rootDir, globalDir); // no inMemoryDb
|
||||
await store.init();
|
||||
|
||||
const task = await createTestTask();
|
||||
|
||||
await store.appendAgentLog(task.id, "flush on close", "text");
|
||||
// close() should flush the buffer gracefully
|
||||
expect(() => store.close()).not.toThrow();
|
||||
|
||||
// Re-open and verify the entry was persisted
|
||||
store = new TaskStore(rootDir, globalDir);
|
||||
await store.init();
|
||||
const logs = await store.getAgentLogs(task.id);
|
||||
expect(logs).toHaveLength(1);
|
||||
expect(logs[0].text).toBe("flush on close");
|
||||
});
|
||||
|
||||
it("close does not throw when flushing entries for already-deleted tasks", async () => {
|
||||
const task = await createTestTask();
|
||||
|
||||
await store.appendAgentLog(task.id, "orphaned entry", "text");
|
||||
// Flush so the entry is in the DB, then delete the task
|
||||
(store as any).flushAgentLogBuffer();
|
||||
await store.deleteTask(task.id);
|
||||
|
||||
// Now buffer another entry for the deleted task
|
||||
await store.appendAgentLog(task.id, "ghost entry", "text");
|
||||
// close() should not throw despite FK constraint violation on flush
|
||||
expect(() => store.close()).not.toThrow();
|
||||
});
|
||||
|
||||
it("emits agent:log event immediately even when buffered", async () => {
|
||||
const task = await createTestTask();
|
||||
const events: any[] = [];
|
||||
store.on("agent:log", (entry) => events.push(entry));
|
||||
|
||||
await store.appendAgentLog(task.id, "immediate event", "text");
|
||||
|
||||
// Event fires immediately, even though DB write is deferred
|
||||
expect(events).toHaveLength(1);
|
||||
expect(events[0].text).toBe("immediate event");
|
||||
expect(events[0].taskId).toBe(task.id);
|
||||
});
|
||||
|
||||
it("groups entries from multiple tasks into a single flush", async () => {
|
||||
const taskA = await createTestTask();
|
||||
const taskB = await store.createTask({ description: "Task B" });
|
||||
|
||||
// Interleave entries for two tasks
|
||||
for (let i = 0; i < 25; i++) {
|
||||
await store.appendAgentLog(taskA.id, `A-${i}`, "text");
|
||||
await store.appendAgentLog(taskB.id, `B-${i}`, "text");
|
||||
}
|
||||
// 50 total = buffer full, triggers flush
|
||||
|
||||
const countA = await store.getAgentLogCount(taskA.id);
|
||||
const countB = await store.getAgentLogCount(taskB.id);
|
||||
expect(countA).toBe(25);
|
||||
expect(countB).toBe(25);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("task comments", () => {
|
||||
|
||||
Reference in New Issue
Block a user