From ef196b68b48dbf45c2384ff4f9a40a654338f65c Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 12 Jul 2026 15:47:03 +0200 Subject: [PATCH] refactor(sqlite): centralize verified snapshot publication (#105412) Co-authored-by: Gio Della-Libera --- src/commands/doctor-state-sqlite-compact.ts | 44 +--- src/infra/backup-create.ts | 47 +--- src/infra/sqlite-snapshot.test.ts | 271 ++++++++++++++++++++ src/infra/sqlite-snapshot.ts | 268 +++++++++++++++++++ src/state/openclaw-state-db.ts | 38 +++ 5 files changed, 590 insertions(+), 78 deletions(-) create mode 100644 src/infra/sqlite-snapshot.test.ts create mode 100644 src/infra/sqlite-snapshot.ts diff --git a/src/commands/doctor-state-sqlite-compact.ts b/src/commands/doctor-state-sqlite-compact.ts index a8b6b1cbb393..1bbe34084d3a 100644 --- a/src/commands/doctor-state-sqlite-compact.ts +++ b/src/commands/doctor-state-sqlite-compact.ts @@ -1,14 +1,9 @@ /** Explicit doctor maintenance for the canonical shared state SQLite database. */ import fs from "node:fs"; -import type { DatabaseSync } from "node:sqlite"; -import { - createNewerSqliteSchemaVersionError, - readSqliteUserVersion, -} from "../infra/sqlite-user-version.js"; import { + assertOpenClawStateDatabaseForMaintenance, ensureOpenClawStatePermissions, isOpenClawStateDatabaseOpen, - OPENCLAW_STATE_SCHEMA_VERSION, } from "../state/openclaw-state-db.js"; import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js"; import { @@ -65,7 +60,8 @@ export function runDoctorStateSqliteCompact( const compact = compactDoctorSqliteFile({ afterMutation: () => ensureOpenClawStatePermissions(sqlitePath, env), sqlitePath, - validateBeforeMutation: (database) => validateCanonicalStateDatabase(database, sqlitePath), + validateBeforeMutation: (database) => + assertOpenClawStateDatabaseForMaintenance(database, { pathname: sqlitePath }), }); return { ...compact, @@ -85,37 +81,3 @@ function readCanonicalStateDatabaseStat(sqlitePath: string): fs.Stats | undefine throw error; } } - -function validateCanonicalStateDatabase(database: DatabaseSync, sqlitePath: string): void { - const userVersion = readSqliteUserVersion(database); - if (userVersion > OPENCLAW_STATE_SCHEMA_VERSION) { - throw createNewerSqliteSchemaVersionError( - "OpenClaw state database", - sqlitePath, - userVersion, - OPENCLAW_STATE_SCHEMA_VERSION, - ); - } - if (userVersion !== OPENCLAW_STATE_SCHEMA_VERSION) { - throw new Error( - `OpenClaw state database ${sqlitePath} uses schema version ${userVersion}; run openclaw doctor --fix before compacting it.`, - ); - } - - const metadata = database - .prepare("SELECT role, schema_version FROM schema_meta WHERE meta_key = 'primary' LIMIT 1") - .get() as { role?: unknown; schema_version?: unknown } | undefined; - if (metadata?.role !== "global") { - const role = typeof metadata?.role === "string" ? metadata.role : "missing"; - throw new Error( - `OpenClaw state database ${sqlitePath} has schema role ${role}; expected global.`, - ); - } - if (metadata.schema_version !== OPENCLAW_STATE_SCHEMA_VERSION) { - const schemaVersion = - typeof metadata.schema_version === "number" ? metadata.schema_version : "invalid"; - throw new Error( - `OpenClaw state database ${sqlitePath} metadata schema version ${schemaVersion} does not match ${OPENCLAW_STATE_SCHEMA_VERSION}; run openclaw doctor --fix before compacting it.`, - ); - } -} diff --git a/src/infra/backup-create.ts b/src/infra/backup-create.ts index 234a2d6ccf60..322b3841b5cb 100644 --- a/src/infra/backup-create.ts +++ b/src/infra/backup-create.ts @@ -7,7 +7,6 @@ import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; import { pipeline } from "node:stream/promises"; import { resolveDateTimestampMs } from "@openclaw/normalization-core/number-coercion"; -import { loadSqliteVecExtension } from "../../packages/memory-host-sdk/src/engine-storage.js"; import { buildBackupArchiveBasename, buildBackupArchivePath, @@ -24,8 +23,7 @@ import { resolveRuntimeServiceVersion } from "../version.js"; import { isVolatileBackupPath } from "./backup-volatile-filter.js"; import { formatErrorMessage } from "./errors.js"; import { writeJson } from "./json-files.js"; -import { requireNodeSqlite } from "./node-sqlite.js"; -import { assertSqliteIntegrity } from "./sqlite-integrity.js"; +import { createVerifiedSqliteSnapshot } from "./sqlite-snapshot.js"; const loadTarRuntime = createLazyRuntimeModule(() => import("tar")); @@ -631,7 +629,6 @@ function tableExistsSql(db: DatabaseSync, tableName: string): boolean { function sanitizeGlobalStateSqliteSnapshot(db: DatabaseSync): void { if (tableExistsSql(db, "delivery_queue_entries")) { db.prepare("DELETE FROM delivery_queue_entries").run(); - db.exec("VACUUM;"); } } @@ -738,7 +735,6 @@ async function createStateSqliteBackupPlan(params: { stateDir: params.stateDir, globalStateSqlitePath, }); - const sqlite = requireNodeSqlite(); const snapshots: SqliteBackupAsset[] = []; for (const archiveSourcePath of discovery.snapshotPaths) { // A discovered *.sqlite file that SQLite cannot snapshot aborts backup. @@ -749,44 +745,21 @@ async function createStateSqliteBackupPlan(params: { path.resolve(archiveSourcePath) === globalStateSqlitePath ? await fs.realpath(archiveSourcePath) : archiveSourcePath; - const source = new sqlite.DatabaseSync(sourceDatabasePath, { - allowExtension: true, - readOnly: true, - }); const sourcePath = path.join(params.tempDir, `openclaw-state-db-${snapshots.length}.sqlite`); try { - source.exec("PRAGMA busy_timeout = 30000;"); - try { - // VACUUM INTO removes deleted-page remnants before the snapshot enters - // the archive. Load known bundled extensions, but fail closed when an - // owner schema needs capabilities core cannot safely reproduce. - await loadSqliteVecExtension({ db: source }); - assertSqliteIntegrity(source, archiveSourcePath); - source.prepare("VACUUM INTO ?").run(sourcePath); - } catch (err) { - throw new Error( - `SQLite database cannot be compacted safely for backup: ${archiveSourcePath}. ${formatErrorMessage(err)}. The source must pass full integrity checks and VACUUM INTO with its required SQLite capabilities; raw page backup was refused because it can retain deleted data.`, - { cause: err }, - ); - } - } finally { - source.close(); - } - await fs.chmod(sourcePath, 0o600); - const snapshot = new sqlite.DatabaseSync(sourcePath, { allowExtension: true }); - try { - await loadSqliteVecExtension({ db: snapshot }); - if (path.resolve(archiveSourcePath) === globalStateSqlitePath) { - sanitizeGlobalStateSqliteSnapshot(snapshot); - } - assertSqliteIntegrity(snapshot, sourcePath); + await createVerifiedSqliteSnapshot({ + sourcePath: sourceDatabasePath, + targetPath: sourcePath, + transform: + path.resolve(archiveSourcePath) === globalStateSqlitePath + ? sanitizeGlobalStateSqliteSnapshot + : undefined, + }); } catch (err) { throw new Error( - `SQLite backup snapshot failed verification for ${archiveSourcePath}: ${formatErrorMessage(err)}`, + `SQLite database cannot be compacted safely for backup: ${archiveSourcePath}. ${formatErrorMessage(err)}. The source must pass full integrity checks and VACUUM INTO with its required SQLite capabilities; raw page backup was refused because it can retain deleted data.`, { cause: err }, ); - } finally { - snapshot.close(); } snapshots.push({ sourcePath, diff --git a/src/infra/sqlite-snapshot.test.ts b/src/infra/sqlite-snapshot.test.ts new file mode 100644 index 000000000000..53639637a883 --- /dev/null +++ b/src/infra/sqlite-snapshot.test.ts @@ -0,0 +1,271 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { requireNodeSqlite } from "./node-sqlite.js"; +import { createVerifiedSqliteSnapshot } from "./sqlite-snapshot.js"; + +const tempDirs: string[] = []; + +async function createTempDir(): Promise { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-sqlite-snapshot-")); + tempDirs.push(tempDir); + return tempDir; +} + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((tempDir) => fs.rm(tempDir, { recursive: true }))); +}); + +function createUnsafeIndexDrift(sqlitePath: string): void { + const sqlite = requireNodeSqlite(); + const database = new sqlite.DatabaseSync(sqlitePath); + try { + database.exec(` + CREATE TABLE records ( + id INTEGER PRIMARY KEY, + indexed_value TEXT NOT NULL, + alternate_value TEXT NOT NULL + ); + CREATE INDEX records_value ON records(indexed_value); + INSERT INTO records (indexed_value, alternate_value) + VALUES ('alpha', 'zeta'), ('beta', 'eta'), ('gamma', 'theta'); + `); + database.enableDefensive?.(false); + database.exec("PRAGMA writable_schema = ON;"); + database + .prepare( + "UPDATE sqlite_schema SET sql = 'CREATE INDEX records_value ON records(alternate_value)' WHERE name = 'records_value'", + ) + .run(); + const schemaVersion = Number( + Object.values(database.prepare("PRAGMA schema_version;").get() as Record)[0], + ); + database.exec(`PRAGMA writable_schema = OFF; PRAGMA schema_version = ${schemaVersion + 1};`); + } finally { + database.close(); + } +} + +describe("createVerifiedSqliteSnapshot", () => { + it("captures committed WAL state and removes deleted page contents", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const deletedValue = `deleted-secret-${"x".repeat(256)}`; + const sqlite = requireNodeSqlite(); + const source = new sqlite.DatabaseSync(sourcePath); + try { + source.exec(` + PRAGMA journal_mode = WAL; + PRAGMA wal_autocheckpoint = 0; + PRAGMA secure_delete = OFF; + CREATE TABLE records (id INTEGER PRIMARY KEY, value TEXT NOT NULL); + PRAGMA wal_checkpoint(TRUNCATE); + `); + source.prepare("INSERT INTO records (value) VALUES (?)").run("survivor"); + source.prepare("INSERT INTO records (value) VALUES (?)").run(deletedValue); + source.prepare("DELETE FROM records WHERE value = ?").run(deletedValue); + + const result = await createVerifiedSqliteSnapshot({ sourcePath, targetPath }); + expect(result).toEqual({ path: targetPath, userVersion: 0 }); + expect((await fs.readFile(targetPath)).includes(deletedValue)).toBe(false); + + const snapshot = new sqlite.DatabaseSync(targetPath, { readOnly: true }); + try { + expect(snapshot.prepare("SELECT value FROM records").all()).toEqual([ + { value: "survivor" }, + ]); + } finally { + snapshot.close(); + } + } finally { + source.close(); + } + }); + + it("rejects unsafe index drift and removes the failed target", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + createUnsafeIndexDrift(sourcePath); + + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow( + /integrity_check failed|malformed database schema/iu, + ); + await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("rejects an existing target without modifying it", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const sqlite = requireNodeSqlite(); + new sqlite.DatabaseSync(sourcePath).close(); + await fs.writeFile(targetPath, "keep"); + + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow( + /target already exists/u, + ); + await expect(fs.readFile(targetPath, "utf8")).resolves.toBe("keep"); + }); + + it("preserves a target created while the snapshot is being prepared", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const sqlite = requireNodeSqlite(); + new sqlite.DatabaseSync(sourcePath).close(); + + await expect( + createVerifiedSqliteSnapshot({ + sourcePath, + targetPath, + transform: async () => { + await fs.writeFile(targetPath, "racer"); + }, + }), + ).rejects.toThrow(/EEXIST|already exists/iu); + await expect(fs.readFile(targetPath, "utf8")).resolves.toBe("racer"); + }); + + it("preserves a target replaced after hard-link publication", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const sqlite = requireNodeSqlite(); + new sqlite.DatabaseSync(sourcePath).close(); + const originalLink = fs.link.bind(fs); + const linkSpy = vi.spyOn(fs, "link").mockImplementation(async (source, target) => { + await originalLink(source, target); + await fs.unlink(target); + await fs.writeFile(target, "racer"); + }); + + try { + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow( + /target changed during publication/u, + ); + await expect(fs.readFile(targetPath, "utf8")).resolves.toBe("racer"); + } finally { + linkSpy.mockRestore(); + } + }); + + it("rejects a target replaced after exclusive-copy publication", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const sqlite = requireNodeSqlite(); + new sqlite.DatabaseSync(sourcePath).close(); + const originalOpen = fs.open.bind(fs); + const linkSpy = vi.spyOn(fs, "link").mockRejectedValue( + Object.assign(new Error("hard links unsupported"), { + code: "EPERM", + }), + ); + const openSpy = vi.spyOn(fs, "open").mockImplementation(async (filePath, flags, mode) => { + const handle = await originalOpen(filePath, flags, mode); + if (path.resolve(String(filePath)) === targetPath && flags === "wx+") { + const originalSync = handle.sync.bind(handle); + vi.spyOn(handle, "sync").mockImplementationOnce(async () => { + await originalSync(); + await fs.unlink(targetPath); + await fs.writeFile(targetPath, "racer"); + }); + } + return handle; + }); + + try { + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow( + /target changed during publication/u, + ); + await expect(fs.readFile(targetPath, "utf8")).resolves.toBe("racer"); + } finally { + openSpy.mockRestore(); + linkSpy.mockRestore(); + } + }); + + it("syncs fallback copies before reporting success", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const sqlite = requireNodeSqlite(); + new sqlite.DatabaseSync(sourcePath).close(); + const originalOpen = fs.open.bind(fs); + const openSpy = vi.spyOn(fs, "open").mockImplementation(originalOpen); + const linkSpy = vi.spyOn(fs, "link").mockRejectedValue( + Object.assign(new Error("hard links unsupported"), { + code: "EPERM", + }), + ); + + try { + await createVerifiedSqliteSnapshot({ sourcePath, targetPath }); + expect( + openSpy.mock.calls.some(([filePath]) => path.resolve(String(filePath)) === targetPath), + ).toBe(true); + } finally { + linkSpy.mockRestore(); + openSpy.mockRestore(); + } + }); + + it("removes its published target when final directory sync fails", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const sqlite = requireNodeSqlite(); + new sqlite.DatabaseSync(sourcePath).close(); + const originalOpen = fs.open.bind(fs); + const openSpy = vi.spyOn(fs, "open").mockImplementation(async (filePath, flags, mode) => { + if (path.resolve(String(filePath)) === tempDir) { + throw Object.assign(new Error("directory sync failed"), { code: "EIO" }); + } + return await originalOpen(filePath, flags, mode); + }); + + try { + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow( + /directory sync failed/u, + ); + await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" }); + } finally { + openSpy.mockRestore(); + } + }); + + it("validates both the source and transformed snapshot", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const removedValue = `removed-secret-${"x".repeat(256)}`; + const sqlite = requireNodeSqlite(); + const source = new sqlite.DatabaseSync(sourcePath); + source.exec("PRAGMA secure_delete = OFF; CREATE TABLE records (value TEXT NOT NULL);"); + source.prepare("INSERT INTO records VALUES (?)").run(removedValue); + source.close(); + const labels: string[] = []; + + await createVerifiedSqliteSnapshot({ + sourcePath, + targetPath, + transform: (database) => { + database.exec("DELETE FROM records;"); + database.prepare("INSERT INTO records VALUES (?)").run("new"); + }, + validate: (_database, label) => labels.push(label), + }); + + expect(labels).toEqual([sourcePath, targetPath]); + expect((await fs.readFile(targetPath)).includes(removedValue)).toBe(false); + const snapshot = new sqlite.DatabaseSync(targetPath, { readOnly: true }); + try { + expect(snapshot.prepare("SELECT value FROM records").get()).toEqual({ value: "new" }); + } finally { + snapshot.close(); + } + }); +}); diff --git a/src/infra/sqlite-snapshot.ts b/src/infra/sqlite-snapshot.ts new file mode 100644 index 000000000000..5d473aba456d --- /dev/null +++ b/src/infra/sqlite-snapshot.ts @@ -0,0 +1,268 @@ +// Creates compact SQLite snapshots only after verifying both source and output. +import type { Stats } from "node:fs"; +import fs from "node:fs/promises"; +import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; +import { loadSqliteVecExtension } from "../../packages/memory-host-sdk/src/engine-storage.js"; +import { formatErrorMessage } from "./errors.js"; +import { sameFileIdentity } from "./fs-safe-advanced.js"; +import { requireNodeSqlite } from "./node-sqlite.js"; +import { assertSqliteIntegrity } from "./sqlite-integrity.js"; +import { readSqliteUserVersion } from "./sqlite-user-version.js"; + +export type SqliteSnapshotValidator = (database: DatabaseSync, databaseLabel: string) => void; + +export type CreateVerifiedSqliteSnapshotOptions = { + sourcePath: string; + targetPath: string; + transform?: (database: DatabaseSync) => void | Promise; + validate?: SqliteSnapshotValidator; +}; + +export type VerifiedSqliteSnapshot = { + path: string; + userVersion: number; +}; + +async function assertRegularSourceFile(sourcePath: string): Promise { + const stat = await fs.lstat(sourcePath); + if (!stat.isFile()) { + throw new Error(`SQLite snapshot source must be a regular file: ${sourcePath}`); + } +} + +async function assertTargetAbsent(targetPath: string): Promise { + try { + await fs.lstat(targetPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return; + } + throw error; + } + throw new Error(`SQLite snapshot target already exists: ${targetPath}`); +} + +function isLinkFallbackError(error: unknown): boolean { + const code = (error as NodeJS.ErrnoException).code; + return ( + code === "EPERM" || + code === "EXDEV" || + code === "ENOTSUP" || + code === "EOPNOTSUPP" || + code === "ENOSYS" + ); +} + +async function publishSnapshotNoOverwrite( + stagedPath: string, + targetPath: string, + stagedIdentity: Stats, +): Promise { + try { + await fs.link(stagedPath, targetPath); + return stagedIdentity; + } catch (error) { + if (!isLinkFallbackError(error)) { + throw error; + } + return await copyFileExclusive(stagedPath, targetPath); + } +} + +async function copyFileExclusive(sourcePath: string, targetPath: string): Promise { + const source = await fs.open(sourcePath, "r"); + let target: Awaited> | undefined; + let targetIdentity: Stats | undefined; + try { + target = await fs.open(targetPath, "wx+", 0o600); + targetIdentity = await target.stat(); + const buffer = Buffer.allocUnsafe(1024 * 1024); + let offset = 0; + while (true) { + const { bytesRead } = await source.read(buffer, 0, buffer.length, offset); + if (bytesRead === 0) { + break; + } + let bytesWritten = 0; + while (bytesWritten < bytesRead) { + const result = await target.write( + buffer, + bytesWritten, + bytesRead - bytesWritten, + offset + bytesWritten, + ); + if (result.bytesWritten === 0) { + throw new Error(`SQLite snapshot copy made no progress: ${targetPath}`); + } + bytesWritten += result.bytesWritten; + } + offset += bytesRead; + } + await target.sync(); + return targetIdentity; + } catch (error) { + if (targetIdentity) { + await target?.close().catch(() => undefined); + target = undefined; + await removePublishedTargetIfOwned(targetPath, targetIdentity); + } + throw error; + } finally { + await target?.close().catch(() => undefined); + await source.close().catch(() => undefined); + } +} + +async function syncFile(filePath: string): Promise { + const handle = await fs.open(filePath, "r+"); + try { + await handle.sync(); + } finally { + await handle.close(); + } +} + +async function syncPublishedFile(filePath: string, expectedIdentity: Stats): Promise { + const handle = await fs.open(filePath, "r+"); + try { + const openedIdentity = await handle.stat(); + if (!sameFileIdentity(expectedIdentity, openedIdentity)) { + throw new Error(`SQLite snapshot target changed before sync: ${filePath}`); + } + await handle.sync(); + const currentIdentity = await fs.lstat(filePath); + if (!sameFileIdentity(expectedIdentity, currentIdentity)) { + throw new Error(`SQLite snapshot target changed during sync: ${filePath}`); + } + } finally { + await handle.close(); + } +} + +async function removePublishedTargetIfOwned( + filePath: string, + expectedIdentity: Stats, +): Promise { + const currentIdentity = await fs.lstat(filePath).catch(() => undefined); + if (currentIdentity && sameFileIdentity(expectedIdentity, currentIdentity)) { + await fs.unlink(filePath).catch(() => undefined); + } +} + +function isUnsupportedDirectorySyncError(error: unknown): boolean { + const code = (error as NodeJS.ErrnoException).code; + return ( + code === "EINVAL" || + code === "ENOTSUP" || + code === "ENOSYS" || + (process.platform === "win32" && (code === "EISDIR" || code === "EPERM" || code === "EACCES")) + ); +} + +async function syncDirectoryBestEffort(directoryPath: string): Promise { + const handle = await fs.open(directoryPath, "r").catch((error: unknown) => { + if (isUnsupportedDirectorySyncError(error)) { + return undefined; + } + throw error; + }); + if (!handle) { + return; + } + try { + await handle.sync(); + } catch (error) { + if (!isUnsupportedDirectorySyncError(error)) { + throw error; + } + } finally { + await handle.close(); + } +} + +/** + * Compact one SQLite database into a fresh private file and verify the result. + * + * The source and output both receive full structural, index, and foreign-key + * checks. Only a fully verified, synced snapshot is published to the target. + */ +export async function createVerifiedSqliteSnapshot( + options: CreateVerifiedSqliteSnapshotOptions, +): Promise { + await assertRegularSourceFile(options.sourcePath); + await assertTargetAbsent(options.targetPath); + + const stagingDir = await fs.mkdtemp( + path.join(path.dirname(options.targetPath), ".sqlite-snapshot-"), + ); + await fs.chmod(stagingDir, 0o700); + const stagedPath = path.join(stagingDir, "database.sqlite"); + const sqlite = requireNodeSqlite(); + let stagedIdentity: Stats | undefined; + let publishedIdentity: Stats | undefined; + try { + const source = new sqlite.DatabaseSync(options.sourcePath, { + allowExtension: true, + readOnly: true, + }); + try { + source.exec("PRAGMA busy_timeout = 30000; PRAGMA trusted_schema = OFF;"); + await loadSqliteVecExtension({ db: source }); + assertSqliteIntegrity(source, options.sourcePath); + options.validate?.(source, options.sourcePath); + source.prepare("VACUUM INTO ?").run(stagedPath); + } finally { + source.close(); + } + + await fs.chmod(stagedPath, 0o600); + const snapshot = new sqlite.DatabaseSync(stagedPath, { allowExtension: true }); + try { + snapshot.exec("PRAGMA busy_timeout = 30000; PRAGMA trusted_schema = OFF;"); + await loadSqliteVecExtension({ db: snapshot }); + if (options.transform) { + await options.transform(snapshot); + // A transform may delete sensitive rows. Compact again so the + // published artifact cannot retain their bytes in free pages. + snapshot.exec("VACUUM;"); + } + assertSqliteIntegrity(snapshot, options.targetPath); + options.validate?.(snapshot, options.targetPath); + const userVersion = readSqliteUserVersion(snapshot); + snapshot.close(); + await syncFile(stagedPath); + stagedIdentity = await fs.lstat(stagedPath); + publishedIdentity = await publishSnapshotNoOverwrite( + stagedPath, + options.targetPath, + stagedIdentity, + ); + const currentIdentity = await fs.lstat(options.targetPath); + if (!currentIdentity.isFile()) { + throw new Error(`SQLite snapshot target is not a regular file: ${options.targetPath}`); + } + if (!sameFileIdentity(publishedIdentity, currentIdentity)) { + throw new Error(`SQLite snapshot target changed during publication: ${options.targetPath}`); + } + await syncPublishedFile(options.targetPath, publishedIdentity); + await syncDirectoryBestEffort(path.dirname(options.targetPath)); + return { path: options.targetPath, userVersion }; + } finally { + if (snapshot.isOpen) { + snapshot.close(); + } + } + } catch (error) { + const ownedIdentity = publishedIdentity ?? stagedIdentity; + if (ownedIdentity) { + await removePublishedTargetIfOwned(options.targetPath, ownedIdentity); + } + throw new Error( + `SQLite database cannot be snapshotted safely: ${options.sourcePath}. ${formatErrorMessage(error)}`, + { cause: error }, + ); + } finally { + await fs.rm(stagingDir, { force: true, recursive: true }).catch(() => undefined); + } +} diff --git a/src/state/openclaw-state-db.ts b/src/state/openclaw-state-db.ts index cf2719987951..65ac7ac7b891 100644 --- a/src/state/openclaw-state-db.ts +++ b/src/state/openclaw-state-db.ts @@ -99,6 +99,44 @@ function assertSupportedSchemaVersion(db: DatabaseSync, pathname: string): void } } +/** Require the canonical shared-state owner and schema before offline file maintenance. */ +export function assertOpenClawStateDatabaseForMaintenance( + database: DatabaseSync, + options: { pathname: string }, +): void { + const userVersion = readSqliteUserVersion(database); + if (userVersion > OPENCLAW_STATE_SCHEMA_VERSION) { + throw createNewerSqliteSchemaVersionError( + "OpenClaw state database", + options.pathname, + userVersion, + OPENCLAW_STATE_SCHEMA_VERSION, + ); + } + if (userVersion !== OPENCLAW_STATE_SCHEMA_VERSION) { + throw new Error( + `OpenClaw state database ${options.pathname} uses schema version ${userVersion}; run openclaw doctor --fix before compacting it.`, + ); + } + + const metadata = database + .prepare("SELECT role, schema_version FROM schema_meta WHERE meta_key = 'primary' LIMIT 1") + .get() as { role?: unknown; schema_version?: unknown } | undefined; + if (metadata?.role !== "global") { + const role = typeof metadata?.role === "string" ? metadata.role : "missing"; + throw new Error( + `OpenClaw state database ${options.pathname} has schema role ${role}; expected global.`, + ); + } + if (metadata.schema_version !== OPENCLAW_STATE_SCHEMA_VERSION) { + const schemaVersion = + typeof metadata.schema_version === "number" ? metadata.schema_version : "invalid"; + throw new Error( + `OpenClaw state database ${options.pathname} metadata schema version ${schemaVersion} does not match ${OPENCLAW_STATE_SCHEMA_VERSION}; run openclaw doctor --fix before compacting it.`, + ); + } +} + const stateDbLog = createSubsystemLogger("state/db"); /** Targets already warned about, so chmod-less filesystems warn once per path. */