root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(core): address all PR review feedback

Browse Source 
- Move integrity check before SCHEMA_SQL writes (was after, contradicting comment)
- Log error message in checkpoint recovery catch block
- Make flushAgentLogBuffer private
- Move deleteTask flush inside withTaskLock to prevent race
- Use try/finally in flush so splice always runs on failure
- Filter stale task entries during flush to prevent buffer poisoning
- Add flush before getAgentLogsByTimeRange for read consistency
- Clean up leaked temp dir in fresh DB test
- Fix false-positive buffer-capacity test (query DB directly)
- Rename misleading "single flush" test title

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Timothy Laurent
2026-05-03 13:53:53 -07:00
parent 71f26c07b0
commit b68eee04b7
4 changed files with 64 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
packages/core/src/__tests__/db.test.ts
View File

@@ -1,7 +1,7 @@
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { Database, createDatabase, toJson, toJsonNullable, fromJson, normalizeTaskComments } from "../db.js";
import { DEFAULT_PROJECT_SETTINGS } from "../types.js";
import { mkdtempSync, existsSync, readFileSync } from "node:fs";
import { mkdtempSync, existsSync, readFileSync, rmSync } from "node:fs";
import { join, dirname } from "node:path";
import { tmpdir } from "node:os";
import { fileURLToPath } from "node:url";
@@ -237,9 +237,13 @@ describe("Database", () => {
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();
try {
// init includes the integrity check — should not throw
expect(() => freshDb.init()).not.toThrow();
} finally {
freshDb.close();
rmSync(freshDir, { recursive: true, force: true });
}
});
});

10
packages/core/src/__tests__/store.test.ts
View File

@@ -4884,9 +4884,11 @@ Task with acceptance criteria
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);
// Validate DB persistence without invoking read-path auto-flush helpers.
const row = (store as any).db
.prepare("SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?")
.get(task.id) as { count: number };
expect(row.count).toBe(50);
});
it("auto-flushes buffered entries when getAgentLogs is called", async () => {
@@ -4971,7 +4973,7 @@ Task with acceptance criteria
expect(events[0].taskId).toBe(task.id);
});
it("groups entries from multiple tasks into a single flush", async () => {
it("flushes interleaved entries from multiple tasks correctly", async () => {
const taskA = await createTestTask();
const taskB = await store.createTask({ description: "Task B" });

31
packages/core/src/db.ts
View File

@@ -910,20 +910,8 @@ export class Database {
* and seed meta values.
*/
init(): void {
this.db.exec(SCHEMA_SQL);
// Seed schemaVersion and lastModified idempotently
this.db.exec(
`INSERT OR IGNORE INTO __meta (key, value) VALUES ('schemaVersion', '1')`,
);
this.db.exec(
`INSERT OR IGNORE INTO __meta (key, value) VALUES ('lastModified', '${Date.now()}')`,
);
// Startup integrity check — run BEFORE migrations and seeds to avoid
// writing to a corrupted database and making it worse. Uses the
// structured integrityCheck() method and attempts WAL checkpoint
// recovery on failure.
// Startup integrity check — run BEFORE any writes to avoid
// compounding corruption. Attempts WAL checkpoint recovery on failure.
const integrity = this.integrityCheck();
if (!integrity.ok) {
this.corruptionDetected = true;
@@ -941,14 +929,25 @@ export class Database {
`Run: sqlite3 ${this.dbPath} ".recover" | sqlite3 ${this.dbPath}.recovered`,
);
}
} catch {
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
console.error(
`[fusion:db] Database corruption detected for ${this.dbPath} and checkpoint recovery failed. ` +
`[fusion:db] Database corruption detected for ${this.dbPath} and checkpoint recovery failed: ${errMsg}. ` +
"Manual recovery required.",
);
}
}
this.db.exec(SCHEMA_SQL);
// Seed schemaVersion and lastModified idempotently
this.db.exec(
`INSERT OR IGNORE INTO __meta (key, value) VALUES ('schemaVersion', '1')`,
);
this.db.exec(
`INSERT OR IGNORE INTO __meta (key, value) VALUES ('lastModified', '${Date.now()}')`,
);
// Run schema migrations
this.migrate();

51
packages/core/src/store.ts
View File

@@ -3752,9 +3752,10 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
async deleteTask(id: string, options?: { removeDependencyReferences?: boolean }): Promise<Task> {
// Flush buffered agent logs so FK cascade deletes can find them.
this.flushAgentLogBuffer();
return this.withTaskLock(id, async () => {
// Flush buffered agent logs inside the lock so no new appends for this
// task can sneak in between flush and DELETE.
this.flushAgentLogBuffer();
const task = this.readTaskFromDb(id);
if (!task) {
throw new Error(`Task ${id} not found`);
@@ -4727,7 +4728,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
* Flush all buffered agent log entries in a single transaction.
* Called when the buffer is full or on a timer.
*/
flushAgentLogBuffer(): void {
private flushAgentLogBuffer(): void {
if (this.agentLogFlushTimer) {
clearTimeout(this.agentLogFlushTimer);
this.agentLogFlushTimer = null;
@@ -4740,21 +4741,37 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
const batch = this.agentLogBuffer.slice();
const flushCount = batch.length;
this.db.transaction(() => {
const stmt = this.db.prepare(`
INSERT INTO agentLogEntries (taskId, timestamp, text, type, detail, agent)
VALUES (?, ?, ?, ?, ?, ?)
`);
for (const entry of batch) {
stmt.run(entry.taskId, entry.timestamp, entry.text, entry.type, entry.detail, entry.agent);
try {
// Filter out entries for deleted tasks to prevent FK violations
// from poisoning the entire buffer.
const liveTaskIds = new Set(
(this.db.prepare("SELECT id FROM tasks").all() as Array<{ id: string }>).map((r) => r.id),
);
const validEntries = batch.filter((e) => liveTaskIds.has(e.taskId));
const dropped = batch.length - validEntries.length;
if (dropped > 0) {
console.warn(
`[fusion] Dropped ${dropped} buffered agent log entries for deleted tasks (${this.db.path})`,
);
}
this.db.bumpLastModified();
});
// Remove only the flushed entries. If appendAgentLog added entries
// during the transaction (can't happen in single-threaded Node, but
// defensive), they remain in the buffer.
this.agentLogBuffer.splice(0, flushCount);
if (validEntries.length > 0) {
this.db.transaction(() => {
const stmt = this.db.prepare(`
INSERT INTO agentLogEntries (taskId, timestamp, text, type, detail, agent)
VALUES (?, ?, ?, ?, ?, ?)
`);
for (const entry of validEntries) {
stmt.run(entry.taskId, entry.timestamp, entry.text, entry.type, entry.detail, entry.agent);
}
this.db.bumpLastModified();
});
}
} finally {
// Always drain the flushed slice so a failed transaction doesn't
// cause the same entries to block every future flush.
this.agentLogBuffer.splice(0, flushCount);
}
}
async appendAgentLogBatch(
@@ -5522,6 +5539,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
startIso: string,
endIso: string | null,
): Promise<AgentLogEntry[]> {
// Ensure buffered entries are visible before reading.
this.flushAgentLogBuffer();
const end = endIso ?? new Date().toISOString();
const selectClause = this.getAgentLogSelectClause();
const rows = this.db.prepare(`