diff --git a/apps/local/src/server/executor.ts b/apps/local/src/server/executor.ts index 10243ef1d..83f3410b2 100644 --- a/apps/local/src/server/executor.ts +++ b/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); diff --git a/apps/local/src/server/sqlite-import.test.ts b/apps/local/src/server/sqlite-import.test.ts index 226708fae..0e00e6128 100644 --- a/apps/local/src/server/sqlite-import.test.ts +++ b/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");