fix(sqlite): recover hot rollback journals privately (#113580)

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<string>();
 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<PreparedSqliteReadOnlyLocation> {
+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<SourceJournalMode, "unknown">,
+): Promise<PreparedSqliteReadOnlyLocation> {
   const tempDir = await createPrivateSqliteTempDirectory(
     resolvePreferredOpenClawTmpDir(),
     `openclaw-sqlite-readonly-${process.pid}-`,
@@ -245,34 +312,35 @@ async function createStableReadOnlyCopy(pathname: string): Promise<PreparedSqlit
     if (process.platform !== "win32") {
       fs.chmodSync(tempDir, 0o700);
     }
-    if (readSourceJournalMode(pathname) !== "wal") {
+    if (readSourceJournalMode(pathname) !== journalMode) {
       throw new SqliteSourceChangedError(`SQLite journal mode changed before copying: ${pathname}`);
     }
     const sidecars = readSourceSidecars(pathname);
     if (sidecars.journal && sidecars.wal) {
       throw new SqliteSourceChangedError(`SQLite journal modes overlapped: ${pathname}`);
     }
+    if (journalMode === "rollback" && !sidecars.journal) {
+      throw new SqliteSourceChangedError(
+        `SQLite rollback journal disappeared before copying: ${pathname}`,
+      );
+    }
 
-    if (sidecars.wal) {
-      copySourceFile(`${pathname}-wal`, firstPath);
+    const sidecarSuffix =
+      journalMode === "rollback" || sidecars.journal
+        ? "-journal"
+        : sidecars.wal
+          ? "-wal"
+          : undefined;
+    if (sidecarSuffix) {
+      copySourceFile(`${pathname}${sidecarSuffix}`, firstPath);
       copySourceFile(pathname, snapshotPath);
-      copySourceFile(`${pathname}-wal`, secondPath);
+      copySourceFile(`${pathname}${sidecarSuffix}`, secondPath);
       assertExpectedSidecars(pathname, sidecars);
       if (!filesEqual(firstPath, secondPath)) {
-        throw new SqliteSourceChangedError(`SQLite WAL changed while copying: ${pathname}`);
+        const label = sidecarSuffix === "-wal" ? "WAL" : "rollback journal";
+        throw new SqliteSourceChangedError(`SQLite ${label} changed while copying: ${pathname}`);
       }
-      replaceFile(secondPath, `${snapshotPath}-wal`);
-    } else if (sidecars.journal) {
-      copySourceFile(`${pathname}-journal`, firstPath);
-      copySourceFile(pathname, snapshotPath);
-      copySourceFile(`${pathname}-journal`, secondPath);
-      assertExpectedSidecars(pathname, sidecars);
-      if (!filesEqual(firstPath, secondPath)) {
-        throw new SqliteSourceChangedError(
-          `SQLite rollback journal changed while copying: ${pathname}`,
-        );
-      }
-      replaceFile(secondPath, `${snapshotPath}-journal`);
+      replaceFile(secondPath, `${snapshotPath}${sidecarSuffix}`);
     } else {
       copySourceFile(pathname, firstPath);
       assertExpectedSidecars(pathname, sidecars);
@@ -286,10 +354,15 @@ async function createStableReadOnlyCopy(pathname: string): Promise<PreparedSqlit
       replaceFile(secondPath, snapshotPath);
     }
 
-    if (readSourceJournalMode(pathname) !== "wal") {
+    if (readSourceJournalMode(pathname) !== journalMode) {
       throw new SqliteSourceChangedError(`SQLite journal mode changed while copying: ${pathname}`);
     }
     fs.rmSync(firstPath, { force: true });
+    if (journalMode === "rollback") {
+      // Recover only the private pair. The source journal remains untouched so
+      // a later writable open can perform SQLite's normal crash recovery.
+      recoverPrivateRollbackCopy(snapshotPath);
+    }
     let active = true;
     return {
       location: snapshotPath,
@@ -369,9 +442,9 @@ async function createOnlineReadOnlyBackup(
 }
 
 /**
- * Rollback modes and active WAL state use SQLite's locking and backup protocol.
- * WAL state without a complete WAL/SHM pair is copied privately so inspecting
- * crash residue never creates coordination files beside the source.
+ * 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.
  */
 export async function prepareSqliteReadOnlyLocation(
   pathname: string,
@@ -394,8 +467,8 @@ export async function prepareSqliteReadOnlyLocation(
       try {
         return await createOnlineReadOnlyBackup(canonicalPath);
       } catch (error) {
-        // A last WAL writer can remove sidecars before SQLite opens. Retry the
-        // now-sidecar-free WAL through the private-copy path.
+        // A writer can add or remove sidecars before SQLite opens. Retry
+        // incomplete WAL state or rollback crash residue through private copy.
         let currentMode: ReturnType<typeof readSourceJournalMode>;
         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<PreparedSqliteReadOnlyLocation | undefined> {
+  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<T>(
+  pathname: string,
+  operation: (sourcePath: string) => Promise<T>,
+): Promise<T> {
+  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();
+  }
+}

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<typeof requireNodeSqlite>,
   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");

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, {

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,