From d07a5d6ce5400311fc89b61ba4eafe62c7cef6d2 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sat, 20 Jun 2026 14:03:55 -0700 Subject: [PATCH] Offload background integrity_check off the main event loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The boot-time integrity check (scheduleBackgroundIntegrityCheck, ~60s after init) ran PRAGMA integrity_check on the live connection, walking every page and freezing the event loop for several seconds — the largest single read-contention stall in normal operation. Offload it to the sqlite3 CLI in a child process (async spawn), matching the existing out-of-process pattern used by quickCheckSqliteFile and .recover. The CLI connection is opened -readonly so it can never checkpoint or write the live WAL; this works because the live process holds the DB open (so the -shm exists). When the CLI is unavailable or can't open read-only, fall back to the in-process check (verified=false), preserving today's behavior on those environments. - New integrityCheckSqliteFileAsync(dbPath, limit) module helper. - New private runBackgroundIntegrityCheck() seam (offload + fallback) so the scheduler has one testable, deterministic policy point. - Background scheduler callback is now async (IIFE + finally) with errors swallowed so a background timer can't crash the process. VACUUM is intentionally NOT offloaded: the call graph shows it is invoked only by the `fn db vacuum` CLI command and tests, never from the periodic maintenance loop, so it is not a background event-loop stall — and an out-of-process VACUUM on a live WAL DB would add corruption surface for no hot-path benefit. Tests updated to the async/offloaded seam (deterministic regardless of whether the sqlite3 CLI exists in the environment), plus coverage for the new helper. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/core/src/__tests__/db.test.ts | 86 +++++++++---- packages/core/src/db.ts | 160 +++++++++++++++++++++---- 2 files changed, 203 insertions(+), 43 deletions(-) diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index f8999259f3..0f2aff1ad3 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -3,6 +3,7 @@ import { Database, createDatabase, quickCheckSqliteFile, + integrityCheckSqliteFileAsync, toJson, toJsonNullable, fromJson, @@ -446,9 +447,25 @@ describe("Database", () => { }); describe("startup integrity check", () => { - it("schedules full integrity check after init instead of blocking startup", () => { + // The background check is offloaded to the sqlite3 CLI off the event loop + // (db.ts runBackgroundIntegrityCheck → integrityCheckSqliteFileAsync). Spy on + // that seam so these tests stay deterministic regardless of whether the + // sqlite3 CLI exists in the environment, and advance timers with the async + // variant so the awaited check resolves before assertions run. + type BackgroundCheckResult = { ok: true } | { ok: false; errors: string[] }; + const spyBackgroundCheck = (result: BackgroundCheckResult) => + vi + .spyOn( + Database.prototype as unknown as { + runBackgroundIntegrityCheck: () => Promise; + }, + "runBackgroundIntegrityCheck", + ) + .mockResolvedValue(result); + + it("schedules full integrity check after init instead of blocking startup", async () => { vi.useFakeTimers(); - const integritySpy = vi.spyOn(Database.prototype, "integrityCheck"); + const checkSpy = spyBackgroundCheck({ ok: true }); const freshDir = makeTmpDir(); const freshFusionDir = join(freshDir, ".fusion"); @@ -457,24 +474,24 @@ describe("Database", () => { try { expect(() => freshDb.init()).not.toThrow(); expect(freshDb.integrityCheckPending).toBe(true); - expect(integritySpy).not.toHaveBeenCalled(); + expect(checkSpy).not.toHaveBeenCalled(); - vi.advanceTimersByTime(60_000); + await vi.advanceTimersByTimeAsync(60_000); - expect(integritySpy).toHaveBeenCalledTimes(1); + expect(checkSpy).toHaveBeenCalledTimes(1); expect(freshDb.integrityCheckPending).toBe(false); expect(freshDb.integrityCheckLastRunAt).toBeTruthy(); } finally { freshDb.close(); removeTrackedTmpDirSync(freshDir); - integritySpy.mockRestore(); + checkSpy.mockRestore(); vi.useRealTimers(); } }); - it("does not schedule duplicate background integrity checks across repeated init calls", () => { + it("does not schedule duplicate background integrity checks across repeated init calls", async () => { vi.useFakeTimers(); - const integritySpy = vi.spyOn(Database.prototype, "integrityCheck"); + const checkSpy = spyBackgroundCheck({ ok: true }); const freshDir = makeTmpDir(); const freshFusionDir = join(freshDir, ".fusion"); const freshDb = new Database(freshFusionDir); @@ -484,20 +501,20 @@ describe("Database", () => { expect(freshDb.integrityCheckPending).toBe(true); freshDb.init(); - vi.advanceTimersByTime(60_000); + await vi.advanceTimersByTimeAsync(60_000); - expect(integritySpy).toHaveBeenCalledTimes(1); + expect(checkSpy).toHaveBeenCalledTimes(1); } finally { freshDb.close(); removeTrackedTmpDirSync(freshDir); - integritySpy.mockRestore(); + checkSpy.mockRestore(); vi.useRealTimers(); } }); - it("deduplicates background integrity check across multiple instances sharing a db path", () => { + it("deduplicates background integrity check across multiple instances sharing a db path", async () => { vi.useFakeTimers(); - const integritySpy = vi.spyOn(Database.prototype, "integrityCheck"); + const checkSpy = spyBackgroundCheck({ ok: true }); const freshDir = makeTmpDir(); const freshFusionDir = join(freshDir, ".fusion"); const dbA = new Database(freshFusionDir); @@ -510,9 +527,9 @@ describe("Database", () => { expect(dbA.integrityCheckPending).toBe(true); expect(dbB.integrityCheckPending).toBe(true); - vi.advanceTimersByTime(60_000); + await vi.advanceTimersByTimeAsync(60_000); - expect(integritySpy).toHaveBeenCalledTimes(1); + expect(checkSpy).toHaveBeenCalledTimes(1); expect(dbA.integrityCheckPending).toBe(false); expect(dbB.integrityCheckPending).toBe(false); expect(dbA.integrityCheckLastRunAt).toBeTruthy(); @@ -525,14 +542,14 @@ describe("Database", () => { dbA.close(); dbB.close(); removeTrackedTmpDirSync(freshDir); - integritySpy.mockRestore(); + checkSpy.mockRestore(); vi.useRealTimers(); } }); - it("fans out corruption detection to all instances participating in shared background check", () => { + it("fans out corruption detection to all instances participating in shared background check", async () => { vi.useFakeTimers(); - const integritySpy = vi.spyOn(Database.prototype, "integrityCheck").mockReturnValue({ + const checkSpy = spyBackgroundCheck({ ok: false, errors: ["malformed database", "broken index"], }); @@ -545,9 +562,9 @@ describe("Database", () => { dbA.init(); dbB.init(); - vi.advanceTimersByTime(60_000); + await vi.advanceTimersByTimeAsync(60_000); - expect(integritySpy).toHaveBeenCalledTimes(1); + expect(checkSpy).toHaveBeenCalledTimes(1); expect(dbA.integrityCheckPending).toBe(false); expect(dbB.integrityCheckPending).toBe(false); expect(dbA.integrityCheckLastRunAt).toBeTruthy(); @@ -560,7 +577,7 @@ describe("Database", () => { dbA.close(); dbB.close(); removeTrackedTmpDirSync(freshDir); - integritySpy.mockRestore(); + checkSpy.mockRestore(); vi.useRealTimers(); } }); @@ -732,6 +749,33 @@ describe("Database", () => { }); }); + describe("integrityCheckSqliteFileAsync (off-event-loop integrity check)", () => { + it("verifies a healthy live DB via the sqlite3 CLI", async () => { + const now = new Date().toISOString(); + db.prepare( + "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)", + ).run("FN-IC-OK", "integrity ok", "todo", now, now); + + // The harness keeps `db` open, so the -readonly CLI connection can attach + // to the live WAL (its -shm exists) — the production scenario. + const result = await integrityCheckSqliteFileAsync(join(fusionDir, "fusion.db")); + + // If the sqlite3 CLI is unavailable in this environment, the helper reports + // verified:false so the caller falls back to the in-process check. + if (result.verified) { + expect(result.ok).toBe(true); + expect(result.errors).toBeUndefined(); + } else { + expect(result.ok).toBe(true); + } + }); + + it("returns a verified failure for a non-existent file without spawning", async () => { + const result = await integrityCheckSqliteFileAsync(join(fusionDir, "does-not-exist.db")); + expect(result).toEqual({ ok: false, verified: true, errors: ["file does not exist"] }); + }); + }); + describe("transactions", () => { it("commits on success", () => { db.transaction(() => { diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 0334c4b171..bf03b82009 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -11,7 +11,7 @@ import { DatabaseSync } from "./sqlite-adapter.js"; import { basename, isAbsolute, join } from "node:path"; import { mkdirSync, existsSync, statSync, renameSync, rmSync } from "node:fs"; -import { spawnSync } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; import { createHash, randomUUID } from "node:crypto"; import { DEFAULT_PROJECT_SETTINGS } from "./types.js"; import type { PluginOnSchemaInit } from "./plugin-types.js"; @@ -1689,6 +1689,88 @@ export function quickCheckSqliteFile(dbPath: string): { ok: boolean; verified: b return { ok: false, verified: true, errors: stdout.split("\n").slice(0, 5) }; } +/** + * Run `PRAGMA integrity_check(limit)` against a SQLite file via the `sqlite3` + * CLI in a child process, so the full page-walk (several seconds on a large DB) + * runs OFF the main event loop instead of freezing it the way the in-process + * `Database.integrityCheck()` does. + * + * The CLI connection is opened `-readonly` so it can never checkpoint or write + * the live WAL out from under the in-process connection. This relies on the + * caller's process holding the DB open (so the `-shm` exists) — which is exactly + * the case for the background check scheduled at init. `verified=false` means the + * check could not be run out-of-process (sqlite3 CLI absent, or the file could + * not be opened read-only) and the caller should fall back to the in-process + * `integrityCheck()`. Matches the non-blocking-on-failure contract of + * `quickCheckSqliteFile`. + */ +export function integrityCheckSqliteFileAsync( + dbPath: string, + limit = 100, +): Promise<{ ok: boolean; verified: boolean; errors?: string[] }> { + return new Promise((resolve) => { + if (!existsSync(dbPath)) { + resolve({ ok: false, verified: true, errors: ["file does not exist"] }); + return; + } + + let child: ReturnType; + try { + child = spawn("sqlite3", ["-readonly", dbPath, `PRAGMA integrity_check(${limit});`], { + stdio: ["ignore", "pipe", "pipe"], + }); + } catch { + resolve({ ok: true, verified: false }); + return; + } + + let stdout = ""; + let settled = false; + const finish = (result: { ok: boolean; verified: boolean; errors?: string[] }) => { + if (!settled) { + settled = true; + resolve(result); + } + }; + + child.stdout?.setEncoding("utf-8"); + child.stdout?.on("data", (chunk: string) => { + stdout += chunk; + // integrity_check(limit) bounds the row count, but guard against a + // pathological file so a runaway child can't exhaust memory. + if (stdout.length > 16 * 1024 * 1024) { + child.kill(); + } + }); + // Drain stderr so the pipe never fills and stalls the child. + child.stderr?.resume(); + + child.on("error", () => finish({ ok: true, verified: false })); + child.on("close", (code) => { + if (code !== 0) { + // integrity_check itself exits 0 and prints any problems to stdout, so a + // non-zero exit almost always means the DB could not be opened + // (locked / read-only -shm unavailable) rather than corruption. Report + // "could not verify" so the caller falls back to the in-process check + // instead of misreporting healthy data as corrupt. + finish({ ok: true, verified: false }); + return; + } + const text = stdout.trim(); + if (text.toLowerCase() === "ok") { + finish({ ok: true, verified: true }); + return; + } + const errors = text + .split("\n") + .map((line) => line.trim()) + .filter((line) => line.length > 0 && line.toLowerCase() !== "ok") + .slice(0, limit); + finish({ ok: errors.length === 0, verified: true, errors: errors.length ? errors : undefined }); + }); + }); +} + // ── Database Class ─────────────────────────────────────────────────── type SharedIntegrityCheckState = { @@ -1993,6 +2075,27 @@ export class Database { return { ok: true }; } + /** + * Resolve the background integrity-check result, preferring the off-event-loop + * `sqlite3` CLI (`integrityCheckSqliteFileAsync`) and falling back to the + * in-process `integrityCheck()` page-walk only when the CLI cannot run it. + * + * Kept as a single instance method so the background scheduler has one + * testable seam (and so the offload/fallback policy lives in one place). + * In-memory DBs have no on-disk file to hand the CLI, so they use the + * in-process check directly. + */ + private async runBackgroundIntegrityCheck(): Promise<{ ok: true } | { ok: false; errors: string[] }> { + if (this.inMemory) { + return this.integrityCheck(); + } + const offloaded = await integrityCheckSqliteFileAsync(this.dbPath); + if (offloaded.verified) { + return offloaded.ok ? { ok: true } : { ok: false, errors: offloaded.errors ?? [] }; + } + return this.integrityCheck(); + } + /** * Synchronously re-run `integrityCheck()` and update the cached corruption * state (`corruptionDetected`, `integrityCheckErrors`, `integrityCheckLastRunAt`). @@ -5277,30 +5380,43 @@ export class Database { shared.timer = null; shared.running = true; - const participants = [...shared.subscribers].filter((instance) => !instance.closed); - const primary = participants[0]; - const startedAt = new Date().toISOString(); + // FNXC:Database 2026-06-20-13:30: + // Offload the integrity-check page-walk to the sqlite3 CLI in a child + // process so it no longer blocks the event loop for several seconds. The + // in-process check (primary.integrityCheck()) remains the fallback for + // environments without the sqlite3 CLI. Wrapped in an async IIFE because + // setTimeout callbacks can't be async; errors must be swallowed here so an + // unhandled rejection can't crash the process from a background timer. + void (async () => { + const participants = [...shared.subscribers].filter((instance) => !instance.closed); + const primary = participants[0]; + const startedAt = new Date().toISOString(); - let integrity: ReturnType = { ok: true }; - if (primary) { - integrity = primary.integrityCheck(); - } + let integrity: ReturnType = { ok: true }; + if (primary) { + integrity = await primary.runBackgroundIntegrityCheck(); + } - for (const participant of participants) { - participant.integrityCheckPending = false; - participant.integrityCheckLastRunAt = startedAt; - participant.corruptionDetected = !integrity.ok; - participant.integrityCheckErrors = integrity.ok ? [] : [...integrity.errors]; - } + for (const participant of participants) { + participant.integrityCheckPending = false; + participant.integrityCheckLastRunAt = startedAt; + participant.corruptionDetected = !integrity.ok; + participant.integrityCheckErrors = integrity.ok ? [] : [...integrity.errors]; + } - if (!integrity.ok) { - const errorSummary = integrity.errors.slice(0, 3).join(" | "); - console.error( - `[fusion:db] Background integrity check detected corruption for ${this.dbPath}: ${errorSummary}`, - ); - } - - Database.sharedIntegrityChecks.delete(this.dbPath); + if (!integrity.ok) { + const errorSummary = integrity.errors.slice(0, 3).join(" | "); + console.error( + `[fusion:db] Background integrity check detected corruption for ${this.dbPath}: ${errorSummary}`, + ); + } + })() + .catch((error) => { + console.warn(`[fusion:db] Background integrity check failed for ${this.dbPath}`, error); + }) + .finally(() => { + Database.sharedIntegrityChecks.delete(this.dbPath); + }); }, 60_000); Database.sharedIntegrityChecks.set(this.dbPath, shared);