diff --git a/src/infra/sqlite-readonly-location.ts b/src/infra/sqlite-readonly-location.ts index a3c2b08e9b95..719a22144685 100644 --- a/src/infra/sqlite-readonly-location.ts +++ b/src/infra/sqlite-readonly-location.ts @@ -13,6 +13,9 @@ import { resolvePreferredOpenClawTmpDir } from "./tmp-openclaw-dir.js"; const MAX_SNAPSHOT_ATTEMPTS = 10; const COPY_BUFFER_BYTES = 1024 * 1024; 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]); const pendingTempDirectoryCleanup = new Set(); let cleanupExitHandlerInstalled = false; @@ -28,6 +31,8 @@ type SourceSidecars = { wal: boolean; }; +type SourceJournalMode = "rollback" | "unknown" | "wal"; + type PreparedSqliteReadOnlyLocation = { cleanup: () => boolean; location: string; @@ -81,7 +86,7 @@ function openPinnedFile(pathname: string): PinnedFile { } } -function readSourceJournalMode(pathname: string): "rollback" | "unknown" | "wal" { +function readSourceJournalMode(pathname: string): SourceJournalMode { const source = openPinnedFile(pathname); try { const header = Buffer.alloc(SQLITE_HEADER_BYTES); @@ -210,6 +215,43 @@ function replaceFile(sourcePath: string, targetPath: string): void { fs.renameSync(sourcePath, targetPath); } +function isSqliteReadOnlyError(error: unknown): boolean { + let current = error; + for (let depth = 0; depth < 8 && current && typeof current === "object"; depth += 1) { + const details = current as { cause?: unknown; errcode?: unknown }; + if ( + typeof details.errcode === "number" && + (details.errcode & SQLITE_RESULT_CODE_MASK) === SQLITE_READONLY_RESULT_CODE + ) { + return true; + } + current = details.cause; + } + return false; +} + +function rollbackJournalReferencesSuperJournal(journalPath: string): boolean { + const descriptor = fs.openSync(journalPath, "r"); + try { + const size = fs.fstatSync(descriptor).size; + if (size < 16) { + return false; + } + const trailer = Buffer.allocUnsafe(16); + if ( + fs.readSync(descriptor, trailer, 0, trailer.length, size - trailer.length) !== trailer.length + ) { + return false; + } + const nameBytes = trailer.readUInt32BE(0); + return ( + nameBytes > 0 && nameBytes <= size - 20 && trailer.subarray(8).equals(SQLITE_JOURNAL_MAGIC) + ); + } finally { + fs.closeSync(descriptor); + } +} + function removeTempDirectory(tempDir: string): boolean { try { fs.rmSync(tempDir, { force: true, maxRetries: 3, recursive: true, retryDelay: 20 }); @@ -233,7 +275,32 @@ function removeTempDirectory(tempDir: string): boolean { } } -async function createStableReadOnlyCopy(pathname: string): Promise { +function recoverPrivateRollbackCopy(snapshotPath: string): void { + if (rollbackJournalReferencesSuperJournal(`${snapshotPath}-journal`)) { + throw new Error( + `SQLite hot rollback journal references a super-journal and cannot be recovered privately: ${snapshotPath}`, + ); + } + const snapshot = openNodeSqliteDatabase(snapshotPath); + try { + snapshot.exec("PRAGMA busy_timeout = 30000; PRAGMA trusted_schema = OFF;"); + snapshot.prepare("PRAGMA schema_version;").get(); + } finally { + snapshot.close(); + } + fs.rmSync(`${snapshotPath}-journal`, { force: true }); + const descriptor = fs.openSync(snapshotPath, "r+"); + try { + fs.fsyncSync(descriptor); + } finally { + fs.closeSync(descriptor); + } +} + +async function createStableReadOnlyCopy( + pathname: string, + journalMode: Exclude, +): Promise { const tempDir = await createPrivateSqliteTempDirectory( resolvePreferredOpenClawTmpDir(), `openclaw-sqlite-readonly-${process.pid}-`, @@ -245,34 +312,35 @@ async function createStableReadOnlyCopy(pathname: string): Promise; try { currentMode = readSourceJournalMode(canonicalPath); @@ -407,6 +480,20 @@ export async function prepareSqliteReadOnlyLocation( continue; } const currentSidecars = readSourceSidecars(canonicalPath); + if (currentMode === "rollback" && currentSidecars.journal) { + if (!isSqliteReadOnlyError(error)) { + throw error; + } + try { + return await createStableReadOnlyCopy(canonicalPath, "rollback"); + } catch (copyError) { + if (!(copyError instanceof SqliteSourceChangedError)) { + throw copyError; + } + lastChange = copyError; + continue; + } + } if (currentMode !== "wal" || (currentSidecars.wal && currentSidecars.shm)) { throw error; } @@ -415,7 +502,7 @@ export async function prepareSqliteReadOnlyLocation( } } try { - return await createStableReadOnlyCopy(canonicalPath); + return await createStableReadOnlyCopy(canonicalPath, "wal"); } catch (error) { if (!(error instanceof SqliteSourceChangedError)) { throw error; @@ -427,3 +514,46 @@ export async function prepareSqliteReadOnlyLocation( cause: lastChange, }); } + +async function prepareSqliteSnapshotSource( + pathname: string, +): Promise { + const canonicalPath = fs.realpathSync.native(pathname); + const journalPath = `${canonicalPath}-journal`; + let journal: BigIntStats; + try { + journal = fs.lstatSync(journalPath, { bigint: true }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return undefined; + } + throw error; + } + if (!journal.isFile()) { + throw new Error(`SQLite rollback journal must be a regular file: ${journalPath}`); + } + return await prepareSqliteReadOnlyLocation(canonicalPath); +} + +export async function withSqliteSnapshotSource( + pathname: string, + operation: (sourcePath: string) => Promise, +): Promise { + let prepared = await prepareSqliteSnapshotSource(pathname); + try { + try { + return await operation(prepared?.location ?? pathname); + } catch (error) { + if (prepared) { + throw error; + } + prepared = await prepareSqliteSnapshotSource(pathname); + if (!prepared) { + throw error; + } + return await operation(prepared.location); + } + } finally { + prepared?.cleanup(); + } +} diff --git a/src/infra/sqlite-snapshot.test.ts b/src/infra/sqlite-snapshot.test.ts index f1ade2acecd3..cdf2169a02b4 100644 --- a/src/infra/sqlite-snapshot.test.ts +++ b/src/infra/sqlite-snapshot.test.ts @@ -1,3 +1,4 @@ +import { spawnSync } from "node:child_process"; import fsSync from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; @@ -78,6 +79,78 @@ function createUnsafeIndexDrift(sqlitePath: string): void { } } +function createHotRollbackJournal(sqlitePath: string): void { + const sqlite = requireNodeSqlite(); + const database = new sqlite.DatabaseSync(sqlitePath); + try { + database.exec(` + PRAGMA journal_mode = DELETE; + PRAGMA synchronous = FULL; + CREATE TABLE records ( + id INTEGER PRIMARY KEY, + value TEXT NOT NULL, + payload BLOB NOT NULL + ); + WITH RECURSIVE rows(id) AS ( + SELECT 1 + UNION ALL + SELECT id + 1 FROM rows WHERE id < 256 + ) + INSERT INTO records (id, value, payload) + SELECT id, 'committed', zeroblob(8192) FROM rows; + `); + } finally { + database.close(); + } + const crashed = spawnSync( + process.execPath, + [ + "--no-warnings", + "--input-type=module", + "-e", + ` + import { DatabaseSync } from "node:sqlite"; + const database = new DatabaseSync(process.env.OPENCLAW_HOT_JOURNAL_PATH); + database.exec( + "PRAGMA journal_mode = DELETE; " + + "PRAGMA synchronous = FULL; " + + "PRAGMA cache_size = 2; " + + "PRAGMA cache_spill = ON; " + + "BEGIN IMMEDIATE; " + + "UPDATE records SET value = 'uncommitted';" + ); + process.kill(process.pid, "SIGKILL"); + `, + ], + { + env: { ...process.env, OPENCLAW_HOT_JOURNAL_PATH: sqlitePath }, + encoding: "utf8", + }, + ); + if (crashed.signal !== "SIGKILL") { + throw new Error( + `hot rollback writer did not exit with SIGKILL: code=${String(crashed.status)} stderr=${crashed.stderr}`, + ); + } + if (!fsSync.existsSync(`${sqlitePath}-journal`)) { + throw new Error("hot rollback writer did not leave a journal"); + } +} + +function appendSuperJournalPointer(journalPath: string, superJournalPath: string): void { + const name = Buffer.from(superJournalPath, "utf8"); + const trailer = Buffer.alloc(4 + name.length + 4 + 4 + 8); + name.copy(trailer, 4); + trailer.writeUInt32BE(name.length, 4 + name.length); + let checksum = 0; + for (const byte of name) { + checksum = (checksum + byte) >>> 0; + } + trailer.writeUInt32BE(checksum, 8 + name.length); + Buffer.from([0xd9, 0xd5, 0x05, 0xf9, 0x20, 0xa1, 0x63, 0xd7]).copy(trailer, 12 + name.length); + fsSync.appendFileSync(journalPath, trailer); +} + function createEmptySqliteDatabase( sqlite: ReturnType, sqlitePath: string, @@ -186,6 +259,127 @@ describe("createVerifiedSqliteSnapshot", () => { } }); + it.skipIf(process.platform === "win32")( + "snapshots committed state from a hot rollback journal without recovering the source", + async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + createHotRollbackJournal(sourcePath); + const sourceBefore = await fs.readFile(sourcePath); + const journalBefore = await fs.readFile(`${sourcePath}-journal`); + + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).resolves.toEqual({ + path: targetPath, + userVersion: 0, + }); + + await expect(fs.readFile(sourcePath)).resolves.toEqual(sourceBefore); + await expect(fs.readFile(`${sourcePath}-journal`)).resolves.toEqual(journalBefore); + const sqlite = requireNodeSqlite(); + const snapshot = new sqlite.DatabaseSync(targetPath, { readOnly: true }); + try { + expect( + snapshot.prepare("SELECT COUNT(*) AS count FROM records WHERE value = 'committed'").get(), + ).toEqual({ count: 256 }); + expect( + snapshot + .prepare("SELECT COUNT(*) AS count FROM records WHERE value = 'uncommitted'") + .get(), + ).toEqual({ count: 0 }); + expect(snapshot.prepare("PRAGMA integrity_check").get()).toEqual({ + integrity_check: "ok", + }); + } finally { + snapshot.close(); + } + await expect(fs.access(`${targetPath}-journal`)).rejects.toMatchObject({ code: "ENOENT" }); + }, + ); + + it.skipIf(process.platform === "win32")( + "rechecks for a hot rollback journal after the direct source open fails", + async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + createHotRollbackJournal(sourcePath); + const journalPath = `${sourcePath}-journal`; + const lstatSync = fsSync.lstatSync.bind(fsSync); + let hidJournal = false; + vi.spyOn(fsSync, "lstatSync").mockImplementation(((pathname, options) => { + if (!hidJournal && path.resolve(String(pathname)) === path.resolve(journalPath)) { + hidJournal = true; + const error = new Error("missing"); + (error as NodeJS.ErrnoException).code = "ENOENT"; + throw error; + } + return lstatSync(pathname, options as never); + }) as typeof fsSync.lstatSync); + + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).resolves.toEqual({ + path: targetPath, + userVersion: 0, + }); + expect(hidJournal).toBe(true); + const sqlite = requireNodeSqlite(); + const snapshot = new sqlite.DatabaseSync(targetPath, { readOnly: true }); + expect( + snapshot.prepare("SELECT COUNT(*) AS count FROM records WHERE value = 'committed'").get(), + ).toEqual({ count: 256 }); + snapshot.close(); + }, + ); + + it.skipIf(process.platform === "win32")( + "refuses private recovery when a hot journal depends on a super-journal", + async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const superJournalPath = path.join(tempDir, "source-mj000000900"); + createHotRollbackJournal(sourcePath); + await fs.writeFile(superJournalPath, "super-journal"); + appendSuperJournalPointer(`${sourcePath}-journal`, superJournalPath); + const sourceBefore = await fs.readFile(sourcePath); + const journalBefore = await fs.readFile(`${sourcePath}-journal`); + const superJournalBefore = await fs.readFile(superJournalPath); + + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow( + /super-journal.*cannot be recovered privately/iu, + ); + + await expect(fs.readFile(sourcePath)).resolves.toEqual(sourceBefore); + await expect(fs.readFile(`${sourcePath}-journal`)).resolves.toEqual(journalBefore); + await expect(fs.readFile(superJournalPath)).resolves.toEqual(superJournalBefore); + await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" }); + }, + ); + + it("ignores a stale rollback journal without changing the source family", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const targetPath = path.join(tempDir, "snapshot.sqlite"); + const sqlite = requireNodeSqlite(); + const source = new sqlite.DatabaseSync(sourcePath); + source.exec("CREATE TABLE records (value TEXT NOT NULL); INSERT INTO records VALUES ('ok');"); + source.close(); + const staleJournal = Buffer.alloc(4096, 0x5a); + await fs.writeFile(`${sourcePath}-journal`, staleJournal); + const sourceBefore = await fs.readFile(sourcePath); + + await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).resolves.toEqual({ + path: targetPath, + userVersion: 0, + }); + + await expect(fs.readFile(sourcePath)).resolves.toEqual(sourceBefore); + await expect(fs.readFile(`${sourcePath}-journal`)).resolves.toEqual(staleJournal); + const snapshot = new sqlite.DatabaseSync(targetPath, { readOnly: true }); + expect(snapshot.prepare("SELECT value FROM records").get()).toEqual({ value: "ok" }); + snapshot.close(); + }); + it("uses online backup before compacting the private copy", async () => { const tempDir = await createTempDir(); const sourcePath = path.join(tempDir, "source.sqlite"); diff --git a/src/infra/sqlite-snapshot.ts b/src/infra/sqlite-snapshot.ts index d3c3a2dbf26f..b4b3d37c9f47 100644 --- a/src/infra/sqlite-snapshot.ts +++ b/src/infra/sqlite-snapshot.ts @@ -16,6 +16,7 @@ import { } from "./node-sqlite.js"; import { assertSqliteIntegrity } from "./sqlite-integrity.js"; import { createPrivateSqliteTempDirectory } from "./sqlite-private-directory.js"; +import { withSqliteSnapshotSource } from "./sqlite-readonly-location.js"; import { readSqliteUserVersion } from "./sqlite-user-version.js"; export type SqliteSnapshotValidator = (database: DatabaseSync, databaseLabel: string) => void; @@ -655,21 +656,24 @@ export async function createVerifiedSqliteSnapshot( const sqlite = requireNodeSqlite(); let stagedIdentity: Stats | undefined; try { - const source = openNodeSqliteDatabase(options.sourcePath, { - allowExtension: true, - readOnly: true, + await withSqliteSnapshotSource(options.sourcePath, async (snapshotSourcePath) => { + await fs.rm(stagedPath, { force: true }); + const source = openNodeSqliteDatabase(snapshotSourcePath, { + 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); + // Copy in incremental steps so concurrent writers are blocked only while + // each batch is read. Compaction happens after releasing the live source. + await sqlite.backup(source, resolveSqliteFilesystemPath(stagedPath)); + } finally { + source.close(); + } }); - try { - source.exec("PRAGMA busy_timeout = 30000; PRAGMA trusted_schema = OFF;"); - await loadSqliteVecExtension({ db: source }); - assertSqliteIntegrity(source, options.sourcePath); - options.validate?.(source, options.sourcePath); - // Copy in incremental steps so concurrent writers are blocked only while - // each batch is read. Compaction happens after releasing the live source. - await sqlite.backup(source, resolveSqliteFilesystemPath(stagedPath)); - } finally { - source.close(); - } await fs.chmod(stagedPath, 0o600); const snapshot = openNodeSqliteDatabase(stagedPath, { diff --git a/src/state/openclaw-state-db.test.ts b/src/state/openclaw-state-db.test.ts index a6b5cd02a24e..917c85202539 100644 --- a/src/state/openclaw-state-db.test.ts +++ b/src/state/openclaw-state-db.test.ts @@ -2706,7 +2706,7 @@ INSERT INTO macos_port_guardian_records VALUES (4242, 18789, '/usr/bin/ssh', 're }); it.skipIf(process.platform === "win32")( - "refuses a hot rollback journal read-only before writable recovery", + "recovers a hot rollback journal privately before writable recovery", () => { const result = runHotRollbackJournalRecoveryProbe({ moduleUrl: new URL("./openclaw-state-db.ts", import.meta.url).href, @@ -2714,9 +2714,9 @@ INSERT INTO macos_port_guardian_records VALUES (4242, 18789, '/usr/bin/ssh', 're }); expect(result.readOnly).toEqual({ - error: expect.stringMatching(/readonly|read-only|rollback/iu), - opened: false, - uncommittedRows: null, + error: null, + opened: true, + uncommittedRows: 0, }); expect(result).toMatchObject({ committedRowsAfterRecovery: 256,