diff --git a/config/knip.config.ts b/config/knip.config.ts index c8ed8ae14e01..66a34da01105 100644 --- a/config/knip.config.ts +++ b/config/knip.config.ts @@ -157,6 +157,8 @@ const rootEntries = [ // Loaded by URL from the SQLite lifecycle archive owner. "src/config/sessions/session-accessor.sqlite-archive.worker.ts!", "src/state/openclaw-database-verify.worker.ts!", + // Spawned by path from sqlite-readonly-location.ts to isolate raw-fd snapshot preparation. + "src/infra/sqlite-readonly-location.worker.ts!", // Loaded by URL from tailscale.ts to outlive abrupt Gateway process exit. "src/infra/tailscale-route-owner.worker.ts!", "src/agents/model-provider-auth.worker.ts!", diff --git a/src/infra/sqlite-readonly-location.test.ts b/src/infra/sqlite-readonly-location.test.ts index 5c294f270173..b335d2e848bd 100644 --- a/src/infra/sqlite-readonly-location.test.ts +++ b/src/infra/sqlite-readonly-location.test.ts @@ -1,3 +1,4 @@ +import { spawnSync } from "node:child_process"; import fs from "node:fs"; import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; @@ -5,7 +6,12 @@ import { Worker } from "node:worker_threads"; import { afterEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; import { requireNodeSqlite } from "./node-sqlite.js"; -import { prepareSqliteReadOnlyLocation } from "./sqlite-readonly-location.js"; +import { + prepareSqliteReadOnlyLocation, + prepareSqliteReadOnlyLocationInProcess, + prepareSqliteReadOnlyLocationSync, + prepareSqliteReadOnlyLocationSyncInProcess, +} from "./sqlite-readonly-location.js"; const workers: Worker[] = []; const tempDirs = useAutoCleanupTempDirTracker((cleanup) => { @@ -21,6 +27,42 @@ function createTempDatabasePath(): string { return path.join(tempDir, "state.sqlite"); } +async function expectPublicSnapshot( + prepare: ( + pathname: string, + ) => + | { cleanup: () => boolean; location: string } + | Promise<{ cleanup: () => boolean; location: string }>, +): Promise { + const sqlite = requireNodeSqlite(); + const databasePath = createTempDatabasePath(); + const writer = new sqlite.DatabaseSync(databasePath); + let cleanup: (() => boolean) | undefined; + try { + writer.exec(` + PRAGMA journal_mode = WAL; + PRAGMA wal_autocheckpoint = 0; + CREATE TABLE probe (value TEXT NOT NULL); + INSERT INTO probe VALUES ('from-wal'); + `); + expect(fs.existsSync(`${databasePath}-wal`)).toBe(true); + + const prepared = await prepare(databasePath); + cleanup = prepared.cleanup; + const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); + try { + expect(snapshot.prepare("SELECT value FROM probe").all()).toEqual([{ value: "from-wal" }]); + } finally { + snapshot.close(); + } + expect(prepared.cleanup()).toBe(true); + cleanup = undefined; + } finally { + cleanup?.(); + writer.close(); + } +} + function readFamily(pathname: string): Map { const family = new Map(); for (const suffix of ["", "-journal", "-shm", "-wal"]) { @@ -56,7 +98,105 @@ function waitForWorkerMessage(worker: Worker, expected: string): Promise { }); } +type PosixLock = { + length: number; + pid: number; + start: number; + type: string; +}; + +function readMainDatabasePosixLocks(pathname: string): PosixLock[] { + const result = spawnSync( + "python3", + [ + "-c", + ` +import fcntl, json, os, struct, sys +layout = struct.Struct("hhqqi4x") +request = layout.pack(fcntl.F_WRLCK, os.SEEK_SET, 1073741826, 510, 0) +with open(sys.argv[1], "rb") as database: + result = layout.unpack(fcntl.fcntl(database.fileno(), fcntl.F_GETLK, request)) +lock_type, _, start, length, pid = result +locks = [] if lock_type == fcntl.F_UNLCK else [{ + "length": length, + "pid": pid, + "start": start, + "type": "read" if lock_type == fcntl.F_RDLCK else "write", +}] +print(json.dumps(locks)) +`, + pathname, + ], + { encoding: "utf8" }, + ); + if (result.status !== 0) { + throw new Error(result.stderr || "POSIX lock probe failed"); + } + return JSON.parse(result.stdout) as PosixLock[]; +} + describe("prepareSqliteReadOnlyLocation", () => { + it("prepares a readable WAL snapshot through the async public entry point", async () => { + await expectPublicSnapshot(prepareSqliteReadOnlyLocation); + }); + + it("prepares a readable WAL snapshot through the sync public entry point", async () => { + await expectPublicSnapshot(prepareSqliteReadOnlyLocationSync); + }); + + it("propagates async public entry point failures", async () => { + const missingPath = path.join(tempDirs.make("openclaw-sqlite-readonly-missing-"), "missing.db"); + await expect(prepareSqliteReadOnlyLocation(missingPath)).rejects.toThrow( + /SQLite read-only worker .*ENOENT/u, + ); + }); + + it("propagates sync public entry point failures", () => { + const missingPath = path.join(tempDirs.make("openclaw-sqlite-readonly-missing-"), "missing.db"); + expect(() => prepareSqliteReadOnlyLocationSync(missingPath)).toThrow( + /SQLite read-only worker .*ENOENT/u, + ); + }); + + it.runIf(process.platform === "linux")( + "keeps a live WAL connection's POSIX locks in the owning process", + async () => { + const sqlite = requireNodeSqlite(); + const databasePath = createTempDatabasePath(); + const writer = new sqlite.DatabaseSync(databasePath); + writer.exec(` + PRAGMA journal_mode = WAL; + CREATE TABLE writes (id INTEGER PRIMARY KEY); + INSERT INTO writes DEFAULT VALUES; + `); + const locksBefore = readMainDatabasePosixLocks(databasePath); + const cleanups: Array<() => boolean> = []; + try { + expect(locksBefore).toHaveLength(1); + + const preparedAsync = await prepareSqliteReadOnlyLocation(databasePath); + cleanups.push(preparedAsync.cleanup); + expect(readMainDatabasePosixLocks(databasePath)).toEqual(locksBefore); + expect(preparedAsync.cleanup()).toBe(true); + + const preparedSync = prepareSqliteReadOnlyLocationSync(databasePath); + cleanups.push(preparedSync.cleanup); + expect(readMainDatabasePosixLocks(databasePath)).toEqual(locksBefore); + expect(preparedSync.cleanup()).toBe(true); + + const characterized = prepareSqliteReadOnlyLocationSyncInProcess(databasePath); + cleanups.push(characterized.cleanup); + expect(readMainDatabasePosixLocks(databasePath)).toEqual([]); + expect(characterized.cleanup()).toBe(true); + } finally { + for (const cleanup of cleanups) { + cleanup(); + } + writer.close(); + } + }, + ); + it("retries a same-size WAL reset instead of accepting an impossible pair", async () => { const sqlite = requireNodeSqlite(); const livePath = createTempDatabasePath(); @@ -92,7 +232,7 @@ describe("prepareSqliteReadOnlyLocation", () => { } }); - const prepared = await prepareSqliteReadOnlyLocation(databasePath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(databasePath); expect(injected).toBe(true); expect(fs.statSync(`${databasePath}-wal`).size).toBe(walSizeBeforeReset); const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); @@ -152,7 +292,7 @@ describe("prepareSqliteReadOnlyLocation", () => { try { await waitForWorkerMessage(worker, "ready"); - const prepared = await prepareSqliteReadOnlyLocation(databasePath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(databasePath); const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); expect(snapshot.prepare("PRAGMA integrity_check").get()).toEqual({ integrity_check: "ok" }); expect(snapshot.prepare("SELECT COUNT(*) AS count FROM payload").get()).toEqual({ count: 1 }); @@ -196,7 +336,7 @@ describe("prepareSqliteReadOnlyLocation", () => { return statSync(pathname, options as never); }) as typeof fs.statSync); - const prepared = await prepareSqliteReadOnlyLocation(databasePath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(databasePath); const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); expect(snapshot.prepare("SELECT value FROM probe").all()).toEqual([{ value: "ok" }]); snapshot.close(); @@ -249,7 +389,7 @@ describe("prepareSqliteReadOnlyLocation", () => { try { await waitForWorkerMessage(worker, "ready"); - const prepared = await prepareSqliteReadOnlyLocation(databasePath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(databasePath); const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); const values = snapshot .prepare("SELECT value FROM pair ORDER BY name") @@ -290,7 +430,7 @@ describe("prepareSqliteReadOnlyLocation", () => { } }); - const prepared = await prepareSqliteReadOnlyLocation(databasePath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(databasePath); expect(injected).toBe(true); expect(path.resolve(prepared.location)).not.toBe(path.resolve(databasePath)); const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); @@ -316,7 +456,7 @@ describe("prepareSqliteReadOnlyLocation", () => { const symlinkPath = path.join(path.dirname(databasePath), "state-link.sqlite"); fs.symlinkSync(path.basename(databasePath), symlinkPath); - const prepared = await prepareSqliteReadOnlyLocation(symlinkPath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(symlinkPath); const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); expect(snapshot.prepare("SELECT value FROM probe").all()).toEqual([{ value: "from-wal" }]); snapshot.close(); @@ -346,7 +486,7 @@ describe("prepareSqliteReadOnlyLocation", () => { return statSync(pathname, options as never); }) as typeof fs.statSync); - const prepared = await prepareSqliteReadOnlyLocation(databasePath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(databasePath); expect(injected).toBe(true); const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); expect(snapshot.prepare("SELECT value FROM probe").all()).toEqual([{ value: "ok" }]); @@ -360,7 +500,7 @@ describe("prepareSqliteReadOnlyLocation", () => { const seed = new sqlite.DatabaseSync(databasePath); seed.exec("CREATE TABLE probe (value TEXT);"); seed.close(); - const prepared = await prepareSqliteReadOnlyLocation(databasePath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(databasePath); const privateDirectory = path.dirname(prepared.location); const rmSync = fs.rmSync.bind(fs); let failRemoval = true; @@ -397,7 +537,7 @@ describe("prepareSqliteReadOnlyLocation", () => { const beforeMain = fs.readFileSync(databasePath); const beforeEntries = fs.readdirSync(path.dirname(databasePath)).toSorted(); - const prepared = await prepareSqliteReadOnlyLocation(databasePath); + const prepared = await prepareSqliteReadOnlyLocationInProcess(databasePath); const snapshot = new sqlite.DatabaseSync(prepared.location, { readOnly: true }); expect(snapshot.prepare("SELECT value FROM probe").all()).toEqual([{ value: "committed" }]); snapshot.close(); diff --git a/src/infra/sqlite-readonly-location.ts b/src/infra/sqlite-readonly-location.ts index b5502830858b..e9ad295df491 100644 --- a/src/infra/sqlite-readonly-location.ts +++ b/src/infra/sqlite-readonly-location.ts @@ -1,4 +1,5 @@ // Prepares consistent private SQLite read-only snapshots. +import { execFile, spawnSync } from "node:child_process"; import fs, { type BigIntStats } from "node:fs"; import path from "node:path"; import { sameFileIdentity } from "./fs-safe-advanced.js"; @@ -7,6 +8,7 @@ import { requireNodeSqlite, resolveSqliteFilesystemPath, } from "./node-sqlite.js"; +import { resolveRuntimeWorkerArgv, resolveRuntimeWorkerUrl } from "./runtime-worker-url.js"; import { createPrivateSqliteTempDirectory, createPrivateSqliteTempDirectorySync, @@ -19,6 +21,8 @@ const SQLITE_HEADER_BYTES = 20; const SQLITE_READONLY_RESULT_CODE = 8; const SQLITE_RESULT_CODE_MASK = 0xff; const SQLITE_JOURNAL_MAGIC = Buffer.from([0xd9, 0xd5, 0x05, 0xf9, 0x20, 0xa1, 0x63, 0xd7]); +export const SQLITE_READONLY_CHILD_ARG = "--openclaw-sqlite-readonly-child"; +const SQLITE_READONLY_STDERR_TAIL_CHARS = 4_000; const pendingTempDirectoryCleanup = new Set(); let cleanupExitHandlerInstalled = false; @@ -41,6 +45,8 @@ type PreparedSqliteReadOnlyLocation = { location: string; }; +type SqliteReadOnlyWorkerResult = { ok: true; location: string } | { ok: false; message: string }; + class SqliteSourceChangedError extends Error {} function statIfPresent(pathname: string): BigIntStats | undefined { @@ -281,6 +287,24 @@ function removeTempDirectory(tempDir: string): boolean { } } +function adoptPreparedLocation(location: string): PreparedSqliteReadOnlyLocation { + const tempDir = path.dirname(location); + let active = true; + return { + location, + cleanup: () => { + if (!active) { + return true; + } + const removed = removeTempDirectory(tempDir); + if (removed) { + active = false; + } + return removed; + }, + }; +} + function recoverPrivateRollbackCopy(snapshotPath: string): void { if (rollbackJournalReferencesSuperJournal(`${snapshotPath}-journal`)) { throw new Error( @@ -355,20 +379,7 @@ function createStableReadOnlyCopyInTempDirectory( // a later writable open can perform SQLite's normal crash recovery. recoverPrivateRollbackCopy(snapshotPath); } - let active = true; - return { - location: snapshotPath, - cleanup: () => { - if (!active) { - return true; - } - const removed = removeTempDirectory(tempDir); - if (removed) { - active = false; - } - return removed; - }, - }; + return adoptPreparedLocation(snapshotPath); } catch (error) { removeTempDirectory(tempDir); throw error; @@ -435,20 +446,7 @@ async function createOnlineReadOnlyBackup( } finally { fs.closeSync(descriptor); } - let active = true; - return { - location: snapshotPath, - cleanup: () => { - if (!active) { - return true; - } - const removed = removeTempDirectory(tempDir); - if (removed) { - active = false; - } - return removed; - }, - }; + return adoptPreparedLocation(snapshotPath); } catch (error) { removeTempDirectory(tempDir); throw error; @@ -459,8 +457,10 @@ async function createOnlineReadOnlyBackup( * Active rollback and WAL state use SQLite's locking and backup protocol. * Crash residue that cannot be opened read-only is copied and recovered * privately so inspection never mutates coordination files beside the source. + * The InProcess exports are child-only: POSIX close() can release every lock + * the calling process holds on the same source inode. */ -export async function prepareSqliteReadOnlyLocation( +export async function prepareSqliteReadOnlyLocationInProcess( pathname: string, ): Promise { const canonicalPath = fs.realpathSync.native(pathname); @@ -540,8 +540,7 @@ export async function prepareSqliteReadOnlyLocation( }); } -/** Synchronously prepares a stable private family for lock-free public inspection. */ -export function prepareSqliteReadOnlyLocationSync( +export function prepareSqliteReadOnlyLocationSyncInProcess( pathname: string, ): PreparedSqliteReadOnlyLocation { const canonicalPath = fs.realpathSync.native(pathname); @@ -577,6 +576,124 @@ export function prepareSqliteReadOnlyLocationSync( }); } +function resolveSqliteReadOnlyWorkerUrl(): URL { + return resolveRuntimeWorkerUrl({ + currentModuleUrl: import.meta.url, + sourceWorkerName: "sqlite-readonly-location.worker", + distWorkerPath: "infra/sqlite-readonly-location.worker.js", + }); +} + +function isSqliteReadOnlyWorkerResult(value: unknown): value is SqliteReadOnlyWorkerResult { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return false; + } + if (Object.keys(value).length !== 2 || !("ok" in value)) { + return false; + } + return ( + (value.ok === true && "location" in value && typeof value.location === "string") || + (value.ok === false && "message" in value && typeof value.message === "string") + ); +} + +function createSqliteReadOnlyWorkerError(message: string, stderr: string): Error { + const stderrTail = stderr.trim().slice(-SQLITE_READONLY_STDERR_TAIL_CHARS); + return new Error( + `SQLite read-only worker ${message}${stderrTail ? `\nstderr (tail): ${stderrTail}` : ""}`, + ); +} + +function parseSqliteReadOnlyWorkerResult( + stdout: string, + stderr: string, +): SqliteReadOnlyWorkerResult { + if (!stdout.trim()) { + throw createSqliteReadOnlyWorkerError("returned no JSON result", stderr); + } + let message: unknown; + try { + message = JSON.parse(stdout); + } catch { + throw createSqliteReadOnlyWorkerError("returned invalid JSON", stderr); + } + if (!isSqliteReadOnlyWorkerResult(message)) { + throw createSqliteReadOnlyWorkerError("returned an invalid result", stderr); + } + return message; +} + +function adoptSqliteReadOnlyWorkerResult(params: { + failure?: string; + stderr: string; + stdout: string; +}): PreparedSqliteReadOnlyLocation { + let result: SqliteReadOnlyWorkerResult; + try { + result = parseSqliteReadOnlyWorkerResult(params.stdout, params.stderr); + } catch (error) { + if (params.failure) { + throw createSqliteReadOnlyWorkerError(params.failure, params.stderr); + } + throw error; + } + if (params.failure || !result.ok) { + throw createSqliteReadOnlyWorkerError( + !result.ok ? result.message : (params.failure ?? "failed"), + params.stderr, + ); + } + return adoptPreparedLocation(result.location); +} + +export async function prepareSqliteReadOnlyLocation( + pathname: string, +): Promise { + const workerUrl = resolveSqliteReadOnlyWorkerUrl(); + return await new Promise((resolve, reject) => { + execFile( + process.execPath, + [ + ...resolveRuntimeWorkerArgv(workerUrl), + SQLITE_READONLY_CHILD_ARG, + "async", + path.resolve(pathname), + ], + { encoding: "utf8" }, + (error, stdout, stderr) => { + try { + const failure = error ? `exited unsuccessfully: ${error.message}` : undefined; + resolve(adoptSqliteReadOnlyWorkerResult({ failure, stderr, stdout })); + } catch (workerError) { + reject(workerError instanceof Error ? workerError : new Error(String(workerError))); + } + }, + ); + }); +} + +export function prepareSqliteReadOnlyLocationSync( + pathname: string, +): PreparedSqliteReadOnlyLocation { + const workerUrl = resolveSqliteReadOnlyWorkerUrl(); + const result = spawnSync( + process.execPath, + [ + ...resolveRuntimeWorkerArgv(workerUrl), + SQLITE_READONLY_CHILD_ARG, + "sync", + path.resolve(pathname), + ], + { encoding: "utf8" }, + ); + const failure = result.error + ? `failed to start: ${result.error.message}` + : result.status === 0 + ? undefined + : `exited with ${result.signal ? `signal ${result.signal}` : `code ${result.status}`}`; + return adoptSqliteReadOnlyWorkerResult({ failure, stderr: result.stderr, stdout: result.stdout }); +} + async function prepareSqliteSnapshotSource( pathname: string, ): Promise { diff --git a/src/infra/sqlite-readonly-location.worker.ts b/src/infra/sqlite-readonly-location.worker.ts new file mode 100644 index 000000000000..f3daf36cd824 --- /dev/null +++ b/src/infra/sqlite-readonly-location.worker.ts @@ -0,0 +1,41 @@ +import { + SQLITE_READONLY_CHILD_ARG, + prepareSqliteReadOnlyLocationInProcess, + prepareSqliteReadOnlyLocationSyncInProcess, +} from "./sqlite-readonly-location.js"; + +function formatWorkerError(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +// The sync strategy raw-copies without attaching SQLite to the source, so sync +// callers stay byte-neutral on the live family; the async strategy holds a read +// transaction on the source and may update its WAL index. +async function runWorker(): Promise { + const mode = process.argv[3]; + const pathname = process.argv[4]; + if ((mode !== "sync" && mode !== "async") || !pathname) { + process.exitCode = 1; + process.stdout.write( + JSON.stringify({ + ok: false, + message: "SQLite read-only worker requires a mode and a database path", + }), + ); + return; + } + try { + const prepared = + mode === "sync" + ? prepareSqliteReadOnlyLocationSyncInProcess(pathname) + : await prepareSqliteReadOnlyLocationInProcess(pathname); + process.stdout.write(JSON.stringify({ ok: true, location: prepared.location })); + } catch (error) { + process.exitCode = 1; + process.stdout.write(JSON.stringify({ ok: false, message: formatWorkerError(error) })); + } +} + +if (process.argv[2] === SQLITE_READONLY_CHILD_ARG) { + void runWorker(); +} diff --git a/src/infra/sqlite-wal.test.ts b/src/infra/sqlite-wal.test.ts index 4e55ab5de950..0d5094700573 100644 --- a/src/infra/sqlite-wal.test.ts +++ b/src/infra/sqlite-wal.test.ts @@ -564,6 +564,112 @@ describe("sqlite WAL maintenance", () => { expect(db["exec"]).toHaveBeenCalledTimes(4); }); + it.runIf(process.platform === "linux")( + "invalidates an unlinked WAL family and permits a clean reopen", + () => { + vi.useFakeTimers(); + const tempDir = tempDirs.make("openclaw-sqlite-wal-split-brain-"); + const databasePath = path.join(tempDir, "state.sqlite"); + const { DatabaseSync } = requireNodeSqlite(); + const writer = new DatabaseSync(databasePath); + const events: unknown[] = []; + let reopened: InstanceType | undefined; + let reopenedMaintenance: ReturnType | undefined; + const maintenance = configureSqliteWalMaintenance(writer, { + checkpointIntervalMs: 100, + databaseLabel: "split-brain-test", + databasePath, + onWalSplitBrain: (event) => events.push(event), + }); + try { + writer.exec("CREATE TABLE events (id INTEGER PRIMARY KEY, value TEXT NOT NULL);"); + writer.prepare("INSERT INTO events (value) VALUES (?)").run("before-unlink"); + expect(maintenance.checkpoint()).toBe(true); + fs.unlinkSync(`${databasePath}-wal`); + fs.unlinkSync(`${databasePath}-shm`); + + vi.advanceTimersByTime(100); + + expect(events).toEqual([ + expect.objectContaining({ + event: "sqlite_wal_sidecar_identity_mismatch", + databasePath, + sidecarPath: expect.stringMatching(/-wal$|-shm$/u), + }), + ]); + expect(writer.isOpen).toBe(false); + expect(() => writer.prepare("INSERT INTO events (value) VALUES ('stale')").run()).toThrow(); + + const fresh = new DatabaseSync(databasePath); + reopened = fresh; + reopenedMaintenance = configureSqliteWalMaintenance(fresh, { + checkpointIntervalMs: 0, + databasePath, + }); + expect(() => + fresh.prepare("INSERT INTO events (value) VALUES (?)").run("after-reopen"), + ).not.toThrow(); + } finally { + maintenance.close(); + reopenedMaintenance?.close(); + if (reopened?.isOpen) { + reopened.close(); + } + if (writer.isOpen) { + writer.close(); + } + } + }, + ); + + it.runIf(process.platform === "linux").each(["EACCES", "EPERM"] as const)( + "disables split-brain detection after a %s scan error", + (code) => { + vi.useFakeTimers(); + const tempDir = tempDirs.make("openclaw-sqlite-wal-tripwire-error-"); + const databasePath = path.join(tempDir, "state.sqlite"); + const { DatabaseSync } = requireNodeSqlite(); + const writer = new DatabaseSync(databasePath); + const events: unknown[] = []; + const maintenance = configureSqliteWalMaintenance(writer, { + checkpointIntervalMs: 100, + databasePath, + onWalSplitBrain: (event) => events.push(event), + }); + const prepare = vi.spyOn(writer, "prepare"); + const readdir = vi.spyOn(fs, "readdirSync").mockImplementationOnce(() => { + const error = new Error("restricted procfs"); + (error as NodeJS.ErrnoException).code = code; + throw error; + }); + try { + writer.exec("CREATE TABLE events (value TEXT NOT NULL);"); + + expect(() => vi.advanceTimersByTime(100)).not.toThrow(); + expect(readdir).toHaveBeenCalledTimes(1); + expect(() => + writer.prepare("INSERT INTO events VALUES (?)").run("still-open"), + ).not.toThrow(); + + fs.unlinkSync(`${databasePath}-wal`); + fs.unlinkSync(`${databasePath}-shm`); + expect(() => vi.advanceTimersByTime(100)).not.toThrow(); + + expect(readdir).toHaveBeenCalledTimes(1); + expect(events).toEqual([]); + expect( + prepare.mock.calls.filter(([sql]) => sql === "PRAGMA wal_checkpoint(PASSIVE);"), + ).toHaveLength(2); + expect(writer.isOpen).toBe(true); + } finally { + maintenance.close(); + if (writer.isOpen) { + writer.close(); + } + } + }, + ); + it("clamps oversized checkpoint intervals before arming timers", () => { vi.useFakeTimers(); const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); diff --git a/src/infra/sqlite-wal.ts b/src/infra/sqlite-wal.ts index 6054ebe09c53..0a0a5d75aecf 100644 --- a/src/infra/sqlite-wal.ts +++ b/src/infra/sqlite-wal.ts @@ -1,9 +1,11 @@ // Configures SQLite WAL and related pragmas for local stores. -import fs from "node:fs"; +import fs, { type BigIntStats } from "node:fs"; import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; import type { Result } from "@openclaw/normalization-core/result"; +import { createSubsystemLogger } from "../logging/subsystem.js"; +import { hasErrnoCode } from "./errno.js"; import { normalizeSqliteNonNegativeInteger } from "./sqlite-busy-timeout.js"; import { isSqliteLockError } from "./sqlite-transaction.js"; @@ -28,6 +30,9 @@ const MOUNT_COMMAND_TIMEOUT_MS = 1_000; const NETWORK_FILESYSTEM_TYPES = new Set(["cifs", "smbfs", "smb2", "smb3"]); const JOURNAL_MODE_RETRY_INTERVAL_MS = 10; const JOURNAL_MODE_RETRY_SLEEP = new Int32Array(new SharedArrayBuffer(4)); +const PROC_SELF_FD_PATH = "/proc/self/fd"; + +const log = createSubsystemLogger("infra/sqlite-wal"); type IntervalHandle = ReturnType & { unref?: () => void; @@ -37,6 +42,16 @@ type SqliteWalCheckpointMode = "PASSIVE" | "FULL" | "RESTART" | "TRUNCATE"; type SqliteFilesystemJournalPolicy = "rollback" | "unsupported" | "wal"; type MountEntry = { mountPoint: string; fsType: string; source?: string }; +type SqliteWalSplitBrainEvent = { + event: "sqlite_wal_sidecar_identity_mismatch"; + databasePath: string; + descriptorDevice: string; + descriptorInode: string; + sidecarPath: string; + targetDevice?: string; + targetInode?: string; +}; + export type SqliteWalMaintenance = { checkpoint: () => boolean; close: (options?: { checkpointMode?: SqliteWalCheckpointMode }) => boolean; @@ -51,6 +66,7 @@ export type SqliteWalMaintenanceOptions = { databaseLabel?: string; databasePath?: string; onCheckpointError?: (error: unknown) => void; + onWalSplitBrain?: (event: SqliteWalSplitBrainEvent) => void; }; export type SqliteConnectionPragmaOptions = SqliteWalMaintenanceOptions & { @@ -341,6 +357,97 @@ function readCheckpointBusyResult(row: unknown): boolean { return value === 1 || value === 1n; } +function statSqliteSidecarTarget(pathname: string): BigIntStats | undefined { + try { + return fs.statSync(pathname, { bigint: true }); + } catch (error) { + if (hasErrnoCode(error, "ENOENT")) { + return undefined; + } + throw error; + } +} + +function isSqliteWalSidecarSplitBrain( + descriptor: BigIntStats, + target: BigIntStats | undefined, +): boolean { + return ( + descriptor.nlink === 0n || + !target || + descriptor.dev !== target.dev || + descriptor.ino !== target.ino + ); +} + +function detectSqliteWalSplitBrain(databasePath: string): SqliteWalSplitBrainEvent | undefined { + let descriptors: string[]; + try { + descriptors = fs.readdirSync(PROC_SELF_FD_PATH); + } catch (error) { + if (hasErrnoCode(error, "ENOENT")) { + return undefined; + } + throw error; + } + const sidecarPaths = [`${databasePath}-wal`, `${databasePath}-shm`]; + for (const descriptorName of descriptors) { + const descriptorPath = path.join(PROC_SELF_FD_PATH, descriptorName); + let linkedPath: string; + try { + linkedPath = fs.readlinkSync(descriptorPath); + } catch (error) { + if (hasErrnoCode(error, "ENOENT")) { + continue; + } + throw error; + } + const sidecarPath = sidecarPaths.find( + (candidate) => linkedPath === candidate || linkedPath === `${candidate} (deleted)`, + ); + if (!sidecarPath) { + continue; + } + let descriptor: BigIntStats; + try { + descriptor = fs.fstatSync(Number(descriptorName), { bigint: true }); + } catch (error) { + if (hasErrnoCode(error, "EBADF") || hasErrnoCode(error, "ENOENT")) { + continue; + } + throw error; + } + try { + if (fs.readlinkSync(descriptorPath) !== linkedPath) { + continue; + } + } catch (error) { + if (hasErrnoCode(error, "ENOENT")) { + continue; + } + throw error; + } + const target = statSqliteSidecarTarget(sidecarPath); + if (!isSqliteWalSidecarSplitBrain(descriptor, target)) { + continue; + } + return { + event: "sqlite_wal_sidecar_identity_mismatch", + databasePath, + descriptorDevice: descriptor.dev.toString(), + descriptorInode: descriptor.ino.toString(), + sidecarPath, + ...(target + ? { + targetDevice: target.dev.toString(), + targetInode: target.ino.toString(), + } + : {}), + }; + } + return undefined; +} + function requireRollbackJournalMode(db: DatabaseSync, options: SqliteWalMaintenanceOptions): void { const row = db.prepare("PRAGMA journal_mode = DELETE;").get(); const journalMode = readJournalModeResult(row); @@ -466,6 +573,13 @@ export function configureSqliteWalMaintenance( enableMacosCheckpointFullfsync(db); db.exec(`PRAGMA wal_autocheckpoint = ${autoCheckpointPages};`); db.exec(`PRAGMA journal_size_limit = ${DEFAULT_SQLITE_WAL_JOURNAL_SIZE_LIMIT_BYTES};`); + const tripwireDatabasePath = + process.platform === "linux" && options.databasePath && fs.existsSync(options.databasePath) + ? fs.realpathSync.native(options.databasePath) + : undefined; + let invalidated = false; + let splitBrainDetectionEnabled = Boolean(tripwireDatabasePath); + let splitBrainDetectionWarningLogged = false; const runCheckpoint = (mode: SqliteWalCheckpointMode): boolean => { try { @@ -494,11 +608,58 @@ export function configureSqliteWalMaintenance( } }; - const checkpoint = (): boolean => runCheckpoint(checkpointMode); + const checkpoint = (): boolean => !invalidated && runCheckpoint(checkpointMode); let timer: IntervalHandle | null = null; if (timerIntervalMs > 0) { timer = setInterval(() => { + if (tripwireDatabasePath && splitBrainDetectionEnabled) { + try { + const splitBrain = detectSqliteWalSplitBrain(tripwireDatabasePath); + if (splitBrain) { + invalidated = true; + if (timer) { + clearInterval(timer); + timer = null; + } + log.error("SQLite WAL sidecar identity mismatch", { + ...splitBrain, + databaseLabel: options.databaseLabel, + }); + try { + options.onWalSplitBrain?.(splitBrain); + } catch (error) { + log.error("SQLite WAL split-brain hook failed", { + databaseLabel: options.databaseLabel, + databasePath: tripwireDatabasePath, + error: error instanceof Error ? error.message : String(error), + }); + } + try { + if (db.isOpen) { + db.close(); + } + } catch (error) { + log.error("SQLite WAL split-brain close failed", { + databaseLabel: options.databaseLabel, + databasePath: tripwireDatabasePath, + error: error instanceof Error ? error.message : String(error), + }); + } + return; + } + } catch (error) { + splitBrainDetectionEnabled = false; + if (!splitBrainDetectionWarningLogged) { + splitBrainDetectionWarningLogged = true; + log.warn("SQLite WAL split-brain detection disabled", { + databaseLabel: options.databaseLabel, + databasePath: tripwireDatabasePath, + error: error instanceof Error ? error.message : String(error), + }); + } + } + } runCheckpoint(periodicCheckpointMode); runIncrementalVacuum(); }, timerIntervalMs) as IntervalHandle; @@ -512,6 +673,9 @@ export function configureSqliteWalMaintenance( clearInterval(timer); timer = null; } + if (invalidated) { + return false; + } // Cache eviction passes PASSIVE: a TRUNCATE close-checkpoint waits on // readers and has starved the event loop for seconds under fleet churn. // Orderly dispose/delete keeps TRUNCATE so sidecars are flushed for unlink. diff --git a/src/infra/tsdown-config.test.ts b/src/infra/tsdown-config.test.ts index a2a0fbd6e931..665aac68eacb 100644 --- a/src/infra/tsdown-config.test.ts +++ b/src/infra/tsdown-config.test.ts @@ -177,6 +177,7 @@ describe("tsdown config", () => { "agents/compaction-planning.worker", "agents/model-provider-auth.worker", "config/sessions/session-accessor.sqlite-archive.worker", + "infra/sqlite-readonly-location.worker", "state/openclaw-database-verify.worker", "system-agent/setup-inference-detection.worker", "plugins/memory-state", diff --git a/src/state/openclaw-database-verify.worker.ts b/src/state/openclaw-database-verify.worker.ts index 00230983b8c6..7eaf27bbe1ae 100644 --- a/src/state/openclaw-database-verify.worker.ts +++ b/src/state/openclaw-database-verify.worker.ts @@ -3,7 +3,7 @@ import { assertSqliteIntegrity, isTerminalSqliteIntegrityError, } from "../infra/sqlite-integrity.js"; -import { prepareSqliteReadOnlyLocation } from "../infra/sqlite-readonly-location.js"; +import { prepareSqliteReadOnlyLocationInProcess } from "../infra/sqlite-readonly-location.js"; import { OPENCLAW_SQLITE_BUSY_TIMEOUT_MS } from "./openclaw-state-db.js"; const DATABASE_VERIFY_CHILD_ARG = "--openclaw-database-verify-child"; @@ -44,7 +44,7 @@ async function verifyOpenClawDatabase( let database: import("node:sqlite").DatabaseSync | undefined; let result = await (async (): Promise => { try { - const prepared = await prepareSqliteReadOnlyLocation(target.path); + const prepared = await prepareSqliteReadOnlyLocationInProcess(target.path); cleanup = prepared.cleanup; database = openNodeSqliteDatabase(prepared.location, { readOnly: true, diff --git a/src/state/openclaw-state-db-open.ts b/src/state/openclaw-state-db-open.ts index aa26522fc14d..0f43a519ecde 100644 --- a/src/state/openclaw-state-db-open.ts +++ b/src/state/openclaw-state-db-open.ts @@ -56,6 +56,7 @@ export function openUnpublishedStateDatabase(params: { busyTimeoutMs: number; lockFailureReporting: SqliteLockFailureReporting; ensureSchema: (database: DatabaseSync) => void; + onWalSplitBrain: () => void; recordOpenFailure: (pathname: string, error: Error) => void; }): OpenClawStateDatabase { const { busyTimeoutMs, lockFailureReporting } = params; @@ -77,6 +78,7 @@ export function openUnpublishedStateDatabase(params: { databaseLabel: "openclaw-state", databasePath: params.pathname, foreignKeys: true, + onWalSplitBrain: params.onWalSplitBrain, synchronous: "NORMAL", }); params.ensureSchema(db); diff --git a/src/state/openclaw-state-db.test.ts b/src/state/openclaw-state-db.test.ts index d812c9165299..2687aba2849f 100644 --- a/src/state/openclaw-state-db.test.ts +++ b/src/state/openclaw-state-db.test.ts @@ -1480,6 +1480,39 @@ afterEach(() => { }); describe("openclaw state database", () => { + it.runIf(process.platform === "linux")( + "evicts a detached WAL family before reopening the shared state database", + () => { + vi.useFakeTimers(); + try { + const stateDir = createTempStateDir(); + const options = { env: { OPENCLAW_STATE_DIR: stateDir } }; + const opened = openOpenClawStateDatabase(options); + expect(opened.walMaintenance.checkpoint()).toBe(true); + fs.unlinkSync(`${opened.path}-wal`); + fs.unlinkSync(`${opened.path}-shm`); + + vi.advanceTimersByTime(30 * 60 * 1000); + + expect(opened.db.isOpen).toBe(false); + expect(() => opened.db.prepare("PRAGMA schema_version").get()).toThrow(); + const reopened = openOpenClawStateDatabase(options); + expect(reopened).not.toBe(opened); + expect(reopened.db.isOpen).toBe(true); + expect(() => + reopened.db + .prepare( + "UPDATE schema_meta SET updated_at = updated_at + 1 WHERE meta_key = 'primary'", + ) + .run(), + ).not.toThrow(); + } finally { + closeOpenClawStateDatabaseForTest(); + vi.useRealTimers(); + } + }, + ); + it("resolves under the shared state database directory", () => { const stateDir = createTempStateDir(); diff --git a/src/state/openclaw-state-db.ts b/src/state/openclaw-state-db.ts index 68948b3f57d9..5c712ef1d551 100644 --- a/src/state/openclaw-state-db.ts +++ b/src/state/openclaw-state-db.ts @@ -607,6 +607,11 @@ function openOpenClawStateDatabaseWithBusyTimeout( busyTimeoutMs, lockFailureReporting, ensureSchema: (database) => ensureSchema(database, pathname, env, busyTimeoutMs), + onWalSplitBrain: () => { + if (unpublished) { + stateDbCache.evictCachedOpenClawStateDatabase(unpublished); + } + }, recordOpenFailure: recordOpenClawStateDatabaseOpenFailure, })); }, diff --git a/tsdown.config.ts b/tsdown.config.ts index 8920d52434dd..ab2a855324ee 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -362,6 +362,7 @@ function buildCoreDistEntries(): Record { "src/config/sessions/session-accessor.sqlite-archive.worker.ts", "config/sessions/session-transcript-reconcile.worker": "src/config/sessions/session-transcript-reconcile.worker.ts", + "infra/sqlite-readonly-location.worker": "src/infra/sqlite-readonly-location.worker.ts", "state/openclaw-database-verify.worker": "src/state/openclaw-database-verify.worker.ts", "infra/tailscale-route-owner.worker": "src/infra/tailscale-route-owner.worker.ts", "system-agent/setup-inference-detection.worker":