fix: auto-heal wedged SQLite connections in place instead of wedging until process restart
A long-lived connection can go SQLITE_NOTADB ('file is not a database' on
every query) while the on-disk file stays intact — observed 2026-07-10 on the
live dashboard, which then failed every API request and poll cycle until the
process was restarted, because all corruption recovery ran at open time only.
The sqlite adapter now detects connection-corruption errors, closes the dead
handle, reopens the same path, replays connection-scoped PRAGMAs, verifies
with quick_check, and retries the failed operation once when outside an
explicit transaction. Prepared statements are generation-tracked and
re-prepare transparently after a reopen; a lost transaction's unwind is
absorbed so the original error propagates cleanly. Reopens are rate-limited,
and real on-disk corruption still defers to the open-time recovery machinery.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
7
.changeset/sqlite-connection-auto-reopen.md
Normal file
7
.changeset/sqlite-connection-auto-reopen.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Auto-heal wedged SQLite connections in place instead of failing every request until restart.
|
||||
category: fix
|
||||
dev: The sqlite adapter now classifies connection-corruption errors (SQLITE_NOTADB "file is not a database" / "database disk image is malformed"), reopens the connection on the same path, replays assignment-style PRAGMAs, verifies with PRAGMA quick_check, and retries the failed operation once when outside an explicit transaction. Statements are generation-tracked so ones prepared before the reopen re-prepare transparently; mid-transaction unwind (ROLLBACK/RELEASE) after a reopen is absorbed as no-ops. Covers fusion.db, fusion-central.db, and archive.db. On-disk corruption (quick_check failure) still defers to the open-time recovery machinery.
|
||||
162
packages/core/src/__tests__/sqlite-adapter-reopen.test.ts
Normal file
162
packages/core/src/__tests__/sqlite-adapter-reopen.test.ts
Normal file
@@ -0,0 +1,162 @@
|
||||
/*
|
||||
FNXC:SqliteConnectionReopen 2026-07-10-22:50:
|
||||
Regression tests for in-place healing of a wedged SQLite connection.
|
||||
Incident 2026-07-10: the live dashboard's fusion.db connection went SQLITE_NOTADB
|
||||
("file is not a database" on every query) while the on-disk file stayed intact,
|
||||
and the only recovery was restarting the whole process. The adapter must reopen
|
||||
the connection in place, replay connection-scoped PRAGMAs, re-prepare cached
|
||||
statements, retry the failed operation once outside transactions, and absorb the
|
||||
unwind of a transaction that died with the old connection.
|
||||
|
||||
The wedge is simulated by swapping the adapter's private `impl` for a stub whose
|
||||
every call throws the corruption error — exactly the observable behavior of the
|
||||
real wedged handle (the real trigger, an inconsistent pager/WAL-index view, is
|
||||
not reproducible deterministically in-process).
|
||||
*/
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { DatabaseSync } from "../sqlite-adapter.js";
|
||||
|
||||
const NOTADB = () => new Error("file is not a database");
|
||||
|
||||
/**
|
||||
* Replace the adapter's live connection with one that throws NOTADB on every
|
||||
* query. close() delegates to the real underlying handle — in production the
|
||||
* reopen path closes the actual wedged connection, releasing any locks it held
|
||||
* (e.g. a write transaction's RESERVED lock).
|
||||
*/
|
||||
function wedge(db: DatabaseSync): void {
|
||||
const holder = db as unknown as {
|
||||
impl: { exec(sql: string): void; prepare(sql: string): unknown; close(): void };
|
||||
};
|
||||
const real = holder.impl;
|
||||
holder.impl = {
|
||||
exec: () => {
|
||||
throw NOTADB();
|
||||
},
|
||||
prepare: () => {
|
||||
throw NOTADB();
|
||||
},
|
||||
close: () => real.close(),
|
||||
};
|
||||
}
|
||||
|
||||
describe("sqlite-adapter corruption reopen", () => {
|
||||
let dir: string;
|
||||
let db: DatabaseSync;
|
||||
|
||||
beforeEach(() => {
|
||||
dir = mkdtempSync(join(tmpdir(), "sqlite-adapter-reopen-test-"));
|
||||
db = new DatabaseSync(join(dir, "test.db"), { reopenCooldownMs: 0 });
|
||||
db.exec("PRAGMA foreign_keys = ON");
|
||||
db.exec("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)");
|
||||
db.prepare("INSERT INTO t (v) VALUES (?)").run("alpha");
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
try {
|
||||
db.close();
|
||||
} catch {
|
||||
// already closed by a test
|
||||
}
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("heals a wedged connection and retries the query once", () => {
|
||||
wedge(db);
|
||||
const row = db.prepare("SELECT v FROM t WHERE id = 1").get() as { v: string };
|
||||
expect(row.v).toBe("alpha");
|
||||
});
|
||||
|
||||
it("re-prepares statements created before the reopen", () => {
|
||||
const stmt = db.prepare("SELECT COUNT(*) AS c FROM t");
|
||||
wedge(db);
|
||||
// Heal via an unrelated query first, then the pre-wedge statement must
|
||||
// transparently re-prepare on the new connection.
|
||||
db.exec("SELECT 1");
|
||||
const row = stmt.get() as { c: number };
|
||||
expect(row.c).toBe(1);
|
||||
});
|
||||
|
||||
it("replays recorded assignment-style PRAGMAs onto the reopened connection", () => {
|
||||
wedge(db);
|
||||
db.exec("SELECT 1");
|
||||
const fk = db.prepare("PRAGMA foreign_keys").get() as { foreign_keys: number };
|
||||
expect(fk.foreign_keys).toBe(1);
|
||||
});
|
||||
|
||||
it("retries writes outside a transaction", () => {
|
||||
wedge(db);
|
||||
db.prepare("INSERT INTO t (v) VALUES (?)").run("beta");
|
||||
const row = db.prepare("SELECT COUNT(*) AS c FROM t").get() as { c: number };
|
||||
expect(row.c).toBe(2);
|
||||
});
|
||||
|
||||
it("does NOT retry inside an explicit transaction, absorbs the unwind, and stays usable", () => {
|
||||
db.exec("BEGIN");
|
||||
db.prepare("INSERT INTO t (v) VALUES (?)").run("in-tx");
|
||||
wedge(db);
|
||||
// The statement inside the broken transaction must fail (no orphan
|
||||
// autocommit retry), even though the connection heals.
|
||||
expect(() => db.prepare("INSERT INTO t (v) VALUES (?)").run("in-tx-2")).toThrow(
|
||||
/file is not a database/,
|
||||
);
|
||||
// The caller's ROLLBACK unwind must be a no-op, not a masking throw.
|
||||
expect(() => db.exec("ROLLBACK")).not.toThrow();
|
||||
// Neither in-tx write survived: the transaction died with the connection.
|
||||
const row = db.prepare("SELECT COUNT(*) AS c FROM t").get() as { c: number };
|
||||
expect(row.c).toBe(1);
|
||||
// Fresh transactions work normally after the unwind.
|
||||
db.exec("BEGIN");
|
||||
db.prepare("INSERT INTO t (v) VALUES (?)").run("post-heal");
|
||||
db.exec("COMMIT");
|
||||
const after = db.prepare("SELECT COUNT(*) AS c FROM t").get() as { c: number };
|
||||
expect(after.c).toBe(2);
|
||||
});
|
||||
|
||||
it("absorbs savepoint unwind (ROLLBACK TO + RELEASE) after a mid-transaction reopen", () => {
|
||||
db.exec("BEGIN");
|
||||
db.exec("SAVEPOINT sp_1");
|
||||
wedge(db);
|
||||
expect(() => db.prepare("SELECT 1").get()).toThrow(/file is not a database/);
|
||||
expect(() => db.exec("ROLLBACK TO sp_1")).not.toThrow();
|
||||
expect(() => db.exec("RELEASE sp_1")).not.toThrow();
|
||||
expect(() => db.exec("ROLLBACK")).not.toThrow();
|
||||
// Connection is healthy afterwards.
|
||||
const row = db.prepare("SELECT COUNT(*) AS c FROM t").get() as { c: number };
|
||||
expect(row.c).toBe(1);
|
||||
});
|
||||
|
||||
it("rethrows the original error while the reopen cooldown is active", () => {
|
||||
const cooled = new DatabaseSync(join(dir, "cooldown.db"), { reopenCooldownMs: 60_000 });
|
||||
try {
|
||||
cooled.exec("CREATE TABLE c (id INTEGER)");
|
||||
wedge(cooled);
|
||||
// First wedge heals (no prior attempt inside the cooldown window)...
|
||||
cooled.exec("SELECT 1");
|
||||
// ...but a second wedge within the window must not reopen again.
|
||||
wedge(cooled);
|
||||
expect(() => cooled.exec("SELECT 1")).toThrow(/file is not a database/);
|
||||
} finally {
|
||||
try {
|
||||
cooled.close();
|
||||
} catch {
|
||||
// wedged stub close may throw
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("does not reopen after the user closed the database", () => {
|
||||
db.close();
|
||||
expect(() => db.prepare("SELECT 1")).toThrow();
|
||||
});
|
||||
|
||||
it("passes unrelated errors through without reopening", () => {
|
||||
expect(() => db.exec("NOT VALID SQL")).toThrow(/syntax error|near/i);
|
||||
// Connection untouched: still generation 0, still works.
|
||||
const row = db.prepare("SELECT COUNT(*) AS c FROM t").get() as { c: number };
|
||||
expect(row.c).toBe(1);
|
||||
});
|
||||
});
|
||||
@@ -68,24 +68,217 @@ function loadDatabaseCtor(): DatabaseCtor {
|
||||
return cachedCtor;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:SqliteConnectionReopen 2026-07-10-22:50:
|
||||
A long-lived connection can wedge in-process: SQLite's pager/WAL-index view goes
|
||||
inconsistent (observed 2026-07-10 during checkpoint activity on the 293MB WAL-mode
|
||||
fusion.db), one query fails "database disk image is malformed", and every query
|
||||
after that returns SQLITE_NOTADB ("file is not a database") forever — while the
|
||||
on-disk file stays fully intact. Before this fix the only recovery was restarting
|
||||
the whole dashboard/engine process, because all corruption-recovery machinery runs
|
||||
at connection-OPEN time only.
|
||||
|
||||
The adapter now heals in place: on a connection-corruption error it closes the dead
|
||||
handle, opens a fresh one on the same path, replays recorded assignment-style
|
||||
PRAGMAs (connection-scoped settings like busy_timeout/foreign_keys/synchronous),
|
||||
verifies the file with PRAGMA quick_check, and retries the failed operation once.
|
||||
Statements returned by prepare() are generation-tracked so ones created before the
|
||||
reopen transparently re-prepare on the new connection.
|
||||
|
||||
Safety rules:
|
||||
- Retry only outside an explicit transaction. A statement inside a broken
|
||||
transaction must NOT be retried on the fresh connection (it would commit as an
|
||||
orphan autocommit write); the connection is still healed but the original error
|
||||
is rethrown so the caller's transaction fails loudly.
|
||||
- After a mid-transaction reopen, the caller's unwind statements
|
||||
(ROLLBACK/ROLLBACK TO/RELEASE/COMMIT) are absorbed as no-ops — the fresh
|
||||
connection has no transaction, and letting ROLLBACK throw would mask the
|
||||
original corruption error in Database.transaction()'s catch path.
|
||||
- quick_check failing on the fresh connection means real on-disk corruption:
|
||||
no retry, the original error propagates, and the open-time recovery machinery
|
||||
(Database.recoverIfCorrupt) remains the owner of that case.
|
||||
- Reopen attempts are rate-limited (default 30s cooldown) so a persistently bad
|
||||
file cannot cause a tight reopen loop.
|
||||
*/
|
||||
|
||||
/** Matches errors indicating the CONNECTION's view of the db is broken (SQLITE_NOTADB / SQLITE_CORRUPT). */
|
||||
function isConnectionCorruptionError(error: unknown): boolean {
|
||||
const message = error instanceof Error ? error.message : String(error ?? "");
|
||||
const text = message.toLowerCase();
|
||||
// FTS5 index corruption is an on-disk shadow-table problem with its own
|
||||
// recovery path (rebuildFts5Index); a connection reopen would not help.
|
||||
if (text.includes("fts5")) return false;
|
||||
return text.includes("file is not a database") || text.includes("database disk image is malformed");
|
||||
}
|
||||
|
||||
type TxControlKind = "begin" | "savepoint" | "commit" | "rollback" | "rollback-to" | "release" | null;
|
||||
|
||||
/** Classify a SQL string that is purely a transaction-control statement (trigger bodies etc. start with CREATE and never match). */
|
||||
function classifyTxControl(sql: string): TxControlKind {
|
||||
const head = sql.trimStart().slice(0, 32).toUpperCase();
|
||||
if (head.startsWith("BEGIN")) return "begin";
|
||||
if (head.startsWith("SAVEPOINT")) return "savepoint";
|
||||
if (head.startsWith("COMMIT") || head.startsWith("END TRANSACTION")) return "commit";
|
||||
if (head.startsWith("ROLLBACK TO")) return "rollback-to";
|
||||
if (head.startsWith("ROLLBACK")) return "rollback";
|
||||
if (head.startsWith("RELEASE")) return "release";
|
||||
return null;
|
||||
}
|
||||
|
||||
/** Assignment-style PRAGMA (`PRAGMA name = value`) — the connection-scoped setup kind worth replaying on reopen. */
|
||||
const SETUP_PRAGMA_RE = /^\s*PRAGMA\s+([A-Za-z_][A-Za-z0-9_]*)\s*=/i;
|
||||
|
||||
const DEFAULT_REOPEN_COOLDOWN_MS = 30_000;
|
||||
|
||||
/**
|
||||
* Drop-in replacement for `node:sqlite`'s `DatabaseSync`. Backed by
|
||||
* `bun:sqlite` under Bun and `node:sqlite` under Node.
|
||||
*/
|
||||
export class DatabaseSync {
|
||||
private impl: RawDatabase;
|
||||
private readonly path: string;
|
||||
private readonly diskBacked: boolean;
|
||||
private readonly reopenCooldownMs: number;
|
||||
/** Bumped on every successful reopen so cached prepared statements re-prepare. */
|
||||
private generation = 0;
|
||||
private userClosed = false;
|
||||
/** Explicit-transaction depth as observed through exec() (BEGIN/SAVEPOINT/...). */
|
||||
private txDepth = 0;
|
||||
/** >0 while absorbing a caller's unwind of a transaction lost to a reopen. */
|
||||
private orphanedTxUnwind = 0;
|
||||
/** Last assignment-style PRAGMA per name, replayed onto a reopened connection. */
|
||||
private readonly setupPragmas = new Map<string, string>();
|
||||
private lastReopenAttemptAt = 0;
|
||||
|
||||
constructor(path: string) {
|
||||
constructor(path: string, options?: { reopenCooldownMs?: number }) {
|
||||
assertOutsideRealFusionPath(path, "SQLite database open");
|
||||
const Ctor = loadDatabaseCtor();
|
||||
this.impl = new Ctor(path);
|
||||
this.path = path;
|
||||
this.diskBacked = path !== ":memory:";
|
||||
this.reopenCooldownMs = Math.max(0, options?.reopenCooldownMs ?? DEFAULT_REOPEN_COOLDOWN_MS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Heal a wedged connection in place. Returns whether the reopen happened and
|
||||
* whether it is safe for the caller to retry the failed operation.
|
||||
*/
|
||||
private attemptCorruptionReopen(cause: unknown): { reopened: boolean; retrySafe: boolean } {
|
||||
if (!this.diskBacked || this.userClosed) return { reopened: false, retrySafe: false };
|
||||
const now = Date.now();
|
||||
if (now - this.lastReopenAttemptAt < this.reopenCooldownMs) return { reopened: false, retrySafe: false };
|
||||
this.lastReopenAttemptAt = now;
|
||||
|
||||
const wasInTransaction = this.txDepth > 0;
|
||||
try {
|
||||
this.impl.close();
|
||||
} catch {
|
||||
// The dead handle may refuse to close; abandon it either way.
|
||||
}
|
||||
|
||||
let fresh: RawDatabase;
|
||||
try {
|
||||
const Ctor = loadDatabaseCtor();
|
||||
fresh = new Ctor(this.path);
|
||||
} catch (openError) {
|
||||
console.error(
|
||||
`[fusion:sqlite] Connection wedged (${cause instanceof Error ? cause.message : String(cause)}) and reopen of ${this.path} failed:`,
|
||||
openError,
|
||||
);
|
||||
return { reopened: false, retrySafe: false };
|
||||
}
|
||||
this.impl = fresh;
|
||||
this.generation++;
|
||||
// The old connection's transaction died with it; absorb the caller's unwind.
|
||||
this.orphanedTxUnwind = wasInTransaction ? this.txDepth : 0;
|
||||
this.txDepth = 0;
|
||||
|
||||
for (const pragma of this.setupPragmas.values()) {
|
||||
try {
|
||||
fresh.exec(pragma);
|
||||
} catch (pragmaError) {
|
||||
console.warn(`[fusion:sqlite] Failed to replay "${pragma}" on reopened ${this.path}:`, pragmaError);
|
||||
}
|
||||
}
|
||||
|
||||
let quickCheckOk = false;
|
||||
try {
|
||||
const row = fresh.prepare("PRAGMA quick_check").get() as { quick_check?: string } | undefined;
|
||||
quickCheckOk = typeof row?.quick_check === "string" && row.quick_check.toLowerCase() === "ok";
|
||||
} catch {
|
||||
quickCheckOk = false;
|
||||
}
|
||||
if (!quickCheckOk) {
|
||||
console.error(
|
||||
`[fusion:sqlite] Reopened ${this.path} after connection corruption, but quick_check failed — on-disk corruption; leaving recovery to the open-time machinery`,
|
||||
);
|
||||
return { reopened: true, retrySafe: false };
|
||||
}
|
||||
|
||||
console.warn(
|
||||
`[fusion:sqlite] Healed wedged connection to ${this.path} (${cause instanceof Error ? cause.message : String(cause)}); reopened in place${wasInTransaction ? "; active transaction was lost and its unwind will be absorbed" : ""}`,
|
||||
);
|
||||
return { reopened: true, retrySafe: !wasInTransaction };
|
||||
}
|
||||
|
||||
/**
|
||||
* Run an operation; on a connection-corruption error, heal the connection and
|
||||
* retry once when safe. `retry` runs against the reopened connection.
|
||||
*/
|
||||
private runWithCorruptionReopen<T>(op: () => T, retry: () => T): T {
|
||||
try {
|
||||
return op();
|
||||
} catch (error) {
|
||||
if (this.userClosed || !isConnectionCorruptionError(error)) throw error;
|
||||
const { reopened, retrySafe } = this.attemptCorruptionReopen(error);
|
||||
if (!reopened || !retrySafe) throw error;
|
||||
return retry();
|
||||
}
|
||||
}
|
||||
|
||||
exec(sql: string): void {
|
||||
this.impl.exec(sql);
|
||||
const txKind = classifyTxControl(sql);
|
||||
if (this.orphanedTxUnwind > 0 && txKind !== null) {
|
||||
// Unwind of a transaction that died with the previous connection: the
|
||||
// fresh connection has no transaction, so these must be no-ops (a real
|
||||
// ROLLBACK here would throw and mask the original corruption error).
|
||||
if (txKind === "commit" || txKind === "rollback") {
|
||||
this.orphanedTxUnwind = 0;
|
||||
} else if (txKind === "release") {
|
||||
this.orphanedTxUnwind--;
|
||||
} else if (txKind === "begin" || txKind === "savepoint") {
|
||||
// A new transaction is starting; the orphaned unwind never completed
|
||||
// (caller swallowed the error) — drop the stale state and execute.
|
||||
this.orphanedTxUnwind = 0;
|
||||
this.execTracked(sql, txKind);
|
||||
}
|
||||
// "rollback-to" keeps the savepoint alive: pure no-op here.
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.diskBacked) {
|
||||
const pragmaMatch = SETUP_PRAGMA_RE.exec(sql);
|
||||
if (pragmaMatch) {
|
||||
this.setupPragmas.set(pragmaMatch[1].toLowerCase(), sql);
|
||||
}
|
||||
}
|
||||
|
||||
this.execTracked(sql, txKind);
|
||||
}
|
||||
|
||||
private execTracked(sql: string, txKind: TxControlKind): void {
|
||||
this.runWithCorruptionReopen(
|
||||
() => this.impl.exec(sql),
|
||||
() => this.impl.exec(sql),
|
||||
);
|
||||
if (txKind === "begin") this.txDepth = 1;
|
||||
else if (txKind === "savepoint") this.txDepth++;
|
||||
else if (txKind === "commit" || txKind === "rollback") this.txDepth = 0;
|
||||
else if (txKind === "release") this.txDepth = Math.max(0, this.txDepth - 1);
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.userClosed = true;
|
||||
this.impl.close();
|
||||
}
|
||||
|
||||
@@ -115,18 +308,42 @@ export class DatabaseSync {
|
||||
}
|
||||
|
||||
prepare(sql: string): SqliteStatement {
|
||||
const stmt = this.impl.prepare(sql);
|
||||
// Prepare eagerly so SQL syntax errors still surface at prepare() time,
|
||||
// but track the connection generation: a statement created before a
|
||||
// corruption reopen transparently re-prepares on the new connection.
|
||||
let stmt = this.runWithCorruptionReopen(
|
||||
() => this.impl.prepare(sql),
|
||||
() => this.impl.prepare(sql),
|
||||
);
|
||||
let stmtGeneration = this.generation;
|
||||
|
||||
const invoke = <T>(call: (s: RawStatement) => T): T =>
|
||||
this.runWithCorruptionReopen(
|
||||
() => {
|
||||
if (stmtGeneration !== this.generation) {
|
||||
stmt = this.impl.prepare(sql);
|
||||
stmtGeneration = this.generation;
|
||||
}
|
||||
return call(stmt);
|
||||
},
|
||||
() => {
|
||||
stmt = this.impl.prepare(sql);
|
||||
stmtGeneration = this.generation;
|
||||
return call(stmt);
|
||||
},
|
||||
);
|
||||
|
||||
// Both node:sqlite and bun:sqlite expose the same .all/.get/.run shape.
|
||||
// Normalize `get` to return undefined (not null) when no row matches, and
|
||||
// pass run() through unchanged — both runtimes already produce the same
|
||||
// { changes, lastInsertRowid } shape.
|
||||
return {
|
||||
all: (...params: unknown[]) => stmt.all(...params),
|
||||
all: (...params: unknown[]) => invoke((s) => s.all(...params)),
|
||||
get: (...params: unknown[]) => {
|
||||
const row = stmt.get(...params);
|
||||
const row = invoke((s) => s.get(...params));
|
||||
return row ?? undefined;
|
||||
},
|
||||
run: (...params: unknown[]) => stmt.run(...params),
|
||||
run: (...params: unknown[]) => invoke((s) => s.run(...params)),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user