Harden SQLite import checkpoint handling

apps/local/src/server/executor.ts

@@ -138,6 +138,11 @@ class LocalExecutorDisposeError extends Data.TaggedError("LocalExecutorDisposeEr
   readonly cause: unknown;
 }> {}
 
+class LocalSqliteCheckpointError extends Data.TaggedError("LocalSqliteCheckpointError")<{
+  readonly path: string;
+  readonly busy: number;
+}> {}
+
 const localExecutorCreateError = (
   operation: LocalExecutorCreateError["operation"],
   cause: unknown,
@@ -287,6 +292,32 @@ const moveSqliteFileSetToBackup = (path: string): string => {
   return backupPath;
 };
 
+const checkpointSqliteForFileMove = (input: {
+  readonly sqlite: Database;
+  readonly path: string;
+}) => {
+  const checkpoint = input.sqlite
+    .query<{ busy: number; log: number; checkpointed: number }, []>(
+      "PRAGMA wal_checkpoint(TRUNCATE)",
+    )
+    .get();
+
+  if (checkpoint && checkpoint.busy !== 0) {
+    // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: SQLite file replacement is synchronous; callers wrap this native failure into LocalExecutorCreateError
+    throw new LocalSqliteCheckpointError({ path: input.path, busy: checkpoint.busy });
+  }
+
+  // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: DELETE mode is best-effort after a successful checkpoint; an open read handle can reject the mode switch without making the file set unsafe to move
+  try {
+    input.sqlite.exec("PRAGMA journal_mode = DELETE");
+  } catch (cause) {
+    console.warn(
+      `[executor] Checkpointed SQLite WAL for ${input.path}, but could not switch journal mode to DELETE before import. Continuing with the checkpointed file set.`,
+      cause,
+    );
+  }
+};
+
 const writeSqliteImportMarker = (
   markerPath: string,
   input: {
@@ -422,8 +453,7 @@ const prepareLegacySqliteForFumaImport = (input: {
           `Skipping legacy Drizzle replay and importing the existing schema as-is.`,
       );
     }
-    sqlite.exec("PRAGMA wal_checkpoint(TRUNCATE)");
-    sqlite.exec("PRAGMA journal_mode = DELETE");
+    checkpointSqliteForFileMove({ sqlite, path: input.storage.sqlitePath });
     return { legacySecrets: [] };
   } finally {
     sqlite.close();
@@ -469,8 +499,7 @@ const importMissingMarkedTables = async (input: {
       tables: pickedTables,
       scopeId: input.scopeId,
     });
-    target.sqlite.exec("PRAGMA wal_checkpoint(TRUNCATE)");
-    target.sqlite.exec("PRAGMA journal_mode = DELETE");
+    checkpointSqliteForFileMove({ sqlite: target.sqlite, path: input.storage.sqlitePath });
     await target.close();
     removeSqliteSidecars(input.storage.sqlitePath);
 
@@ -534,8 +563,7 @@ export const importLegacySqliteIfNeeded = async (options: {
             await withQueryContext(target.db, {
               allowedScopeIds: new Set([scopeId]),
             }).createMany("secret", createLegacySecretRows(scopeId, prepared.legacySecrets));
-            target.sqlite.exec("PRAGMA wal_checkpoint(TRUNCATE)");
-            target.sqlite.exec("PRAGMA journal_mode = DELETE");
+            checkpointSqliteForFileMove({ sqlite: target.sqlite, path: storage.sqlitePath });
           } finally {
             await target.close();
             removeSqliteSidecars(storage.sqlitePath);
@@ -601,8 +629,7 @@ export const importLegacySqliteIfNeeded = async (options: {
       tables,
       scopeId,
     });
-    target.sqlite.exec("PRAGMA wal_checkpoint(TRUNCATE)");
-    target.sqlite.exec("PRAGMA journal_mode = DELETE");
+    checkpointSqliteForFileMove({ sqlite: target.sqlite, path: targetPath });
     await target.close();
     removeSqliteSidecars(targetPath);
 

apps/local/src/server/sqlite-import.test.ts

@@ -24,13 +24,16 @@ import { createSqliteFumaDb, type SqliteFumaDb } from "./sqlite-fumadb";
 
 let workDir: string;
 let sqlite: SqliteFumaDb | null;
+let heldReader: Database | null;
 
 beforeEach(() => {
   workDir = mkdtempSync(join(tmpdir(), "executor-sqlite-import-"));
   sqlite = null;
+  heldReader = null;
 });
 
 afterEach(async () => {
+  heldReader?.close();
   await sqlite?.close();
   rmSync(workDir, { recursive: true, force: true });
 });
@@ -449,6 +452,46 @@ describe("importSqliteDataToFuma", () => {
     ).resolves.toMatchObject({ id: "src_1", scope_id: "scope_a" });
   });
 
+  it("imports a checkpointed legacy WAL database even when DELETE journal mode is busy", async () => {
+    const sqlitePath = join(workDir, "data.db");
+    const markerPath = join(workDir, "fumadb-sqlite-imported");
+    seedMigratedSqlite(sqlitePath);
+
+    const writer = new Database(sqlitePath);
+    writer.exec("PRAGMA journal_mode = WAL");
+    writer.close();
+
+    heldReader = new Database(sqlitePath, { readonly: true });
+    heldReader.exec("BEGIN");
+    heldReader.query("SELECT * FROM source").all();
+
+    const tables = collectTables([]);
+    const result = await importLegacySqliteIfNeeded({
+      storage: {
+        dataDir: workDir,
+        sqlitePath,
+        importMarkerPath: markerPath,
+      },
+      tables,
+      scopeId: "scope_a",
+    });
+
+    expect(result.imported).toBe(true);
+    expect(result.importedRows).toBe(2);
+    expect(existsSync(markerPath)).toBe(true);
+
+    sqlite = await createSqliteFumaDb({
+      tables,
+      namespace: "executor_local",
+      path: sqlitePath,
+    });
+    await expect(
+      withQueryContext(sqlite.db, { allowedScopeIds: new Set(["scope_a"]) }).findFirst("source", {
+        where: (b) => b("id", "=", "src_1"),
+      }),
+    ).resolves.toMatchObject({ id: "src_1", scope_id: "scope_a" });
+  });
+
   it("imports newly-loaded plugin tables from the original backup after the first cutover", async () => {
     const sqlitePath = join(workDir, "data.db");
     const markerPath = join(workDir, "fumadb-sqlite-imported");