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

Offload background integrity_check off the main event loop

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-06-20 14:03:55 -07:00
parent 849eefa35e
commit d07a5d6ce5
2 changed files with 203 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<BackgroundCheckResult>;
},
"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(() => {

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

@@ -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<typeof spawn>;
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<Database["integrityCheck"]> = { ok: true };
if (primary) {
integrity = primary.integrityCheck();
}
let integrity: ReturnType<Database["integrityCheck"]> = { 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);