commit ffeaddc46f85f7dcd3644fee2f96d9323b81e701 parent a6e75a870890b6a5142351646ba1553a013c4b63 Author: Florian Dold <dold@taler.net> Date: Wed, 26 Aug 2026 10:25:39 +0200 wallet core: add explicit SQLite database migration Diffstat:
16 files changed, 555 insertions(+), 51 deletions(-)
diff --git a/packages/taler-util/src/notifications.ts b/packages/taler-util/src/notifications.ts @@ -387,6 +387,8 @@ export interface IdleNotification { export interface DatabaseMaintenanceProgressNotification { type: NotificationType.DatabaseMaintenanceProgress; operation: "indexeddb-fixup" | "indexeddb-to-native-migration"; + /** Token of the API request that initiated this operation, when available. */ + progressToken?: string; phase: "fixup" | "copy" | "verify" | "complete" | "failed"; /** Current fixup or backend-neutral store, when one is active. */ step?: string; @@ -396,6 +398,8 @@ export interface DatabaseMaintenanceProgressNotification { processedRecords?: number; /** Total records in the whole migration, known before copying starts. */ totalRecords?: number; + /** Rough overall migration completion, from 0 through 100. */ + completionPercent?: number; /** Why the maintenance operation failed. Present when phase is failed. */ error?: TalerErrorDetail; } diff --git a/packages/taler-util/src/types-taler-wallet.ts b/packages/taler-util/src/types-taler-wallet.ts @@ -588,6 +588,15 @@ export interface WalletRunConfig { * browser extension. */ migrateNativeDb: boolean; + + /** + * Use wallet-core's native sqlite schema when initializing a new, empty + * database. + * + * Unlike migrateNativeDb, this never converts an existing IndexedDB + * wallet. Hosts without a native sqlite backend ignore the preference. + */ + useNativeDb: boolean; }; /** @@ -624,6 +633,26 @@ export interface InitResponse { databaseBackend: WalletDatabaseBackend; } +/** Start an explicit migration from IndexedDB emulation to native sqlite. */ +export interface MigrateDatabaseRequest { + /** Enables progress correlation and cancellation through cancelProgressToken. */ + progressToken?: string; +} + +export const codecForMigrateDatabaseRequest = + (): Codec<MigrateDatabaseRequest> => + buildCodecForObject<MigrateDatabaseRequest>() + .property("progressToken", codecOptional(codecForString())) + .build("MigrateDatabaseRequest"); + +export interface MigrateDatabaseResponse { + /** Whether this request changed the active database backend. */ + migrated: boolean; + + /** Database backend active after the request. */ + databaseBackend: WalletDatabaseBackend; +} + /** * Shorter version of stringifyScopeInfo */ diff --git a/packages/taler-wallet-cli/src/index.ts b/packages/taler-wallet-cli/src/index.ts @@ -720,9 +720,6 @@ async function createLocalWallet( checkEnvFlag("TALER_WALLET_STATS"), skipDefaults: walletCliArgs.wallet.skipDefaults, }, - features: { - ...(args.migrateNativeDb ? { migrateNativeDb: true } : {}), - }, }, } satisfies InitRequest, ); @@ -749,11 +746,6 @@ function writeObservabilityLog(notif: WalletNotification): void { export interface WalletRunArgs { lazyTaskLoop?: boolean; noInit?: boolean; - /** - * Migrate the database to the native schema during initialization, for the - * command that exists to do exactly that. - */ - migrateNativeDb?: boolean; } async function withWallet<T>( @@ -3473,17 +3465,13 @@ advancedCli const dbPath = getSqlite3FilenameFromStoragePath( args.wallet.walletDbFile ?? defaultWalletDbPath, ); - // Through a wallet rather than against the file: initializing is what - // replays the old backend's fixup log, and the records the migration - // copies have to be the repaired ones. - await withWallet( - args, - { lazyTaskLoop: true, migrateNativeDb: true }, - async () => {}, - ); - // The wallet logs a failed migration and keeps running on the database it - // had, which is right for a wallet starting up and wrong for a command - // whose only job was to migrate. + // Through a wallet rather than against the file: initialization replays + // the old backend's fixup log before the explicit migration copies it. + await withWallet(args, { lazyTaskLoop: true }, async (wallet) => { + await wallet.client.call(WalletApiOperation.MigrateDatabase, { + progressToken: "wallet-cli-db-migrate", + }); + }); const after = await inspectWalletDbPath(dbPath); if (after.kind !== "native") { console.error( diff --git a/packages/taler-wallet-core/src/db/handle.ts b/packages/taler-wallet-core/src/db/handle.ts @@ -29,7 +29,7 @@ * handle, that choice does not exist to get wrong. */ -import { WalletNotification } from "@gnu-taler/taler-util"; +import { CancellationToken, WalletNotification } from "@gnu-taler/taler-util"; import { WalletDbTransaction } from "./transaction.js"; @@ -41,6 +41,11 @@ export interface WalletDbAccessStats { recordsRead: number; } +export interface WalletDbMigrationOptions { + cancellationToken?: CancellationToken; + progressToken?: string; +} + /** * Wallet-level work that must become visible in the same atomic import as the * restored records (currently rebuilding the materialized transaction view). @@ -130,7 +135,13 @@ export interface WalletDbHandle { * This handle is unusable afterwards. On failure it is untouched and still * the authoritative database. */ - migrateToNative?(): Promise<WalletDbHandle>; + migrateToNative?(options?: WalletDbMigrationOptions): Promise<WalletDbHandle>; + + /** + * Open the same physical storage with the native schema if the host found + * it empty before either wallet schema was initialized. + */ + openNativeIfEmpty?(): Promise<WalletDbHandle>; /** * Backend-specific counters for the testing API. diff --git a/packages/taler-wallet-core/src/db/indexeddb/handle.ts b/packages/taler-wallet-core/src/db/indexeddb/handle.ts @@ -47,6 +47,7 @@ import { WalletDbAccessStats, WalletDbHandle, WalletDbImportFinalizer, + WalletDbMigrationOptions, } from "../handle.js"; import { IdbWalletTransaction } from "./transaction.js"; import { WalletDbTransaction } from "../transaction.js"; @@ -102,7 +103,10 @@ export class IdbWalletDbHandle implements WalletDbHandle { * emulation runs over a sqlite database the host can also open natively. * See {@link WalletDbHandle.migrateToNative}. */ - migrateToNative?: () => Promise<WalletDbHandle>; + migrateToNative?: ( + options?: WalletDbMigrationOptions, + ) => Promise<WalletDbHandle>; + openNativeIfEmpty?: () => Promise<WalletDbHandle>; setNotificationSink(sink: (n: WalletNotification) => void): void { this.notify = sink; diff --git a/packages/taler-wallet-core/src/db/migration/converter.test.ts b/packages/taler-wallet-core/src/db/migration/converter.test.ts @@ -28,6 +28,7 @@ import assert from "node:assert"; import { test } from "node:test"; import { + CancellationToken, CoinStatus, DatabaseMaintenanceProgressNotification, DenomKeyType, @@ -378,7 +379,9 @@ test("converter: IndexedDB to sqlite, populated by the conformance corpus", asyn } // convertWalletDb re-enumerates both sides and compares every record; // a thrown error here is the actual test. - const report = await convertWalletDb(src, dst); + const report = await convertWalletDb(src, dst, { + progressToken: "converter-progress", + }); await dst.runReadWriteTx(async (tx) => { assert.deepStrictEqual( @@ -410,6 +413,29 @@ test("converter: IndexedDB to sqlite, populated by the conformance corpus", asyn maintenanceProgress.every((n) => n.totalRecords === report.totalRecords), "progress did not carry the global record total", ); + assert.ok( + maintenanceProgress.every((n) => n.progressToken === "converter-progress"), + "progress did not carry the request token", + ); + const percentages = maintenanceProgress.map((n) => n.completionPercent!); + assert.strictEqual(percentages[0], 0); + assert.strictEqual(percentages.at(-1), 100); + for (let i = 1; i < percentages.length; i++) { + assert.ok( + percentages[i] >= percentages[i - 1], + "migration completion percentage went backwards", + ); + } + assert.ok( + maintenanceProgress + .filter((n) => n.phase === "copy") + .every((n) => n.completionPercent! <= 50), + ); + assert.ok( + maintenanceProgress + .filter((n) => n.phase === "verify") + .every((n) => n.completionPercent! >= 50), + ); for (const phase of ["copy", "verify"] as const) { const records = maintenanceProgress .filter((n) => n.phase === phase && n.processedRecords !== undefined) @@ -452,6 +478,37 @@ test("converter: IndexedDB to sqlite, populated by the conformance corpus", asyn await dst.close(); }); +test("converter: cancellation stops between bounded batches", async () => { + const src = await makeIdbRunner(); + const dst = await makeSqliteRunner(); + const cts = CancellationToken.create(); + await src.runReadWriteTx(async (tx) => { + for (let i = 0; i < DB_CONVERSION_PROGRESS_RECORDS * 2; i++) { + await tx.upsertTombstone({ id: `cancel-conversion-${i}` }); + } + }); + try { + await assert.rejects( + convertWalletDb(src, dst, { + cancellationToken: cts.token, + onProgress(notification) { + if ( + notification.phase === "copy" && + (notification.processedRecords ?? 0) >= + DB_CONVERSION_PROGRESS_RECORDS + ) { + cts.cancel(); + } + }, + }), + CancellationToken.CancellationError, + ); + } finally { + await src.close(); + await dst.close(); + } +}); + test("converter: sqlite to IndexedDB (reverse direction)", async () => { const src = await makeSqliteRunner(); for (const c of conformanceCases) { diff --git a/packages/taler-wallet-core/src/db/migration/converter.ts b/packages/taler-wallet-core/src/db/migration/converter.ts @@ -32,6 +32,7 @@ */ import { + CancellationToken, DatabaseMaintenanceProgressNotification, Logger, NotificationType, @@ -521,6 +522,12 @@ export interface DbConversionReport { /** Optional hooks for observing or deliberately interrupting a conversion. */ export interface DbConversionOptions { + /** Cancellation checked between bounded database batches. */ + cancellationToken?: CancellationToken; + + /** API progress token to attach to maintenance notifications. */ + progressToken?: string; + /** * Called after a progress notification has been delivered to the source * handle. Throwing aborts the conversion, which lets callers inject a @@ -624,11 +631,14 @@ async function digestStore( handle: WalletDbHandle, step: CopyStep, progress?: (processed: number) => void, + cancellationToken: CancellationToken = CancellationToken.CONTINUE, ): Promise<RecordMultisetDigest> { const digest = new RecordMultisetDigest(); let cursor: unknown | undefined; while (true) { + cancellationToken.throwIfCancelled(); const page = await readPage(handle, step, cursor, false); + cancellationToken.throwIfCancelled(); if (page.records.length === 0) break; const normalize = step.normalize ?? ((r: unknown) => r); for (const record of page.records) { @@ -666,13 +676,19 @@ export async function convertWalletDb( let totalRecords = 0; for (const group of COPY_PLAN) { for (const st of group) { - const digest = await digestStore(src, st); + const digest = await digestStore( + src, + st, + undefined, + options.cancellationToken, + ); sourceDigests.set(st.name, digest); totalRecords += digest.count; } } const progressInterval = DB_CONVERSION_PROGRESS_RECORDS; + const latestProcessedRecords = { copy: 0, verify: 0 }; const notify = ( phase: "copy" | "verify", @@ -680,15 +696,26 @@ export async function convertWalletDb( step?: CopyStep, processedRecords?: number, ): void => { + if (processedRecords !== undefined) { + latestProcessedRecords[phase] = processedRecords; + } + const completedUnits = + Math.min(latestProcessedRecords[phase], totalRecords) + completedSteps; + const totalUnits = totalRecords + DB_CONVERSION_STEP_COUNT; + const phasePercent = Math.floor((50 * completedUnits) / totalUnits); const notification: DatabaseMaintenanceProgressNotification = { type: NotificationType.DatabaseMaintenanceProgress, operation: "indexeddb-to-native-migration", phase, + ...(options.progressToken + ? { progressToken: options.progressToken } + : {}), completedSteps, totalSteps: DB_CONVERSION_STEP_COUNT, ...(step ? { step: step.name } : {}), ...(processedRecords !== undefined ? { processedRecords } : {}), totalRecords, + completionPercent: phase === "copy" ? phasePercent : 50 + phasePercent, }; src.emitNotification(notification); options.onProgress?.(notification); @@ -722,13 +749,17 @@ export async function convertWalletDb( let cursor: unknown | undefined; let storeCount = 0; while (true) { + options.cancellationToken?.throwIfCancelled(); const page = await readPage(src, st, cursor, true); + options.cancellationToken?.throwIfCancelled(); if (page.records.length === 0) break; + options.cancellationToken?.throwIfCancelled(); await dst.runReadWriteTx(async (tx) => { for (const rec of page.records) { await st.write(tx, rec); } }); + options.cancellationToken?.throwIfCancelled(); storeCount += page.records.length; copiedRecords += page.records.length; copyProgress(stepIndex, copiedRecords, st); @@ -754,8 +785,11 @@ export async function convertWalletDb( for (const st of group) { const sourceDigest = sourceDigests.get(st.name)!; const beforeStore = verifiedRecords; - const destinationDigest = await digestStore(dst, st, (processed) => - verifyProgress(stepIndex, beforeStore + processed, st), + const destinationDigest = await digestStore( + dst, + st, + (processed) => verifyProgress(stepIndex, beforeStore + processed, st), + options.cancellationToken, ); if (!sourceDigest.equals(destinationDigest)) { throw Error( diff --git a/packages/taler-wallet-core/src/db/migration/native.test.ts b/packages/taler-wallet-core/src/db/migration/native.test.ts @@ -40,6 +40,8 @@ import { createNodeHelperSqlite3Impl } from "@gnu-taler/idb-bridge/node-helper-s import { DatabaseMaintenanceProgressNotification, NotificationType, + TalerError, + TalerErrorCode, WalletNotification, } from "@gnu-taler/taler-util"; @@ -64,10 +66,12 @@ import { openNativeSqliteWalletDb, } from "../sqlite/database.js"; import { + createNativeWalletHost2, inspectWalletDbPath, resolveWalletDbMigration, } from "../../host-impl.node.js"; import { acquireSqliteWalletDbOwnership } from "../../host-common.js"; +import { WalletApiOperation } from "../../wallet-api-types.js"; import { conformanceCases } from "../testing/conformance-cases.js"; import { ConformanceAsserts } from "../testing/conformance.js"; @@ -697,3 +701,140 @@ test("native migration: offline resolution creates the mandatory full backup", a fs.rmSync(dir, { recursive: true, force: true }); } }); + +test("useNativeDb initializes empty storage directly as native", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "wallet-native-empty-")); + const dbPath = path.join(dir, "wallet.sqlite3"); + try { + const { wallet } = await createNativeWalletHost2({ + persistentStoragePath: dbPath, + }); + const init = await wallet.client.call(WalletApiOperation.InitWallet, { + config: { + lazyTaskLoop: true, + testing: { skipDefaults: true }, + features: { useNativeDb: true }, + }, + }); + assert.strictEqual(init.databaseBackend, "sqlite"); + await wallet.client.call(WalletApiOperation.Shutdown, {}); + + const inspection = await inspectWalletDbPath(dbPath); + assert.strictEqual(inspection.kind, "native"); + assert.strictEqual(inspection.migration, undefined); + assert.strictEqual(inspection.indexedDbRecords, 0); + + const imp = await createNodeHelperSqlite3Impl({ enableTracing: false }); + const db = await imp.open(dbPath); + try { + const tables = await listTables(db); + assert.ok(!tables.includes("object_data")); + assert.ok(!tables.some((name) => name.startsWith(IDB_BACKUP_PREFIX))); + } finally { + await db.close(); + } + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +test("useNativeDb leaves existing IndexedDB for explicit migration", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "wallet-native-explicit-")); + const dbPath = path.join(dir, "wallet.sqlite3"); + try { + const existing = await makeMinimalIdbDb(dbPath); + await existing.handle.close(); + await existing.db.close(); + + const notifications: WalletNotification[] = []; + const second = await createNativeWalletHost2({ + persistentStoragePath: dbPath, + notifyHandler: (notification) => notifications.push(notification), + }); + const secondInit = await second.wallet.client.call( + WalletApiOperation.InitWallet, + { + config: { + lazyTaskLoop: true, + testing: { skipDefaults: true }, + features: { useNativeDb: true }, + }, + }, + ); + assert.strictEqual(secondInit.databaseBackend, "indexeddb"); + + const migrated = await second.wallet.client.call( + WalletApiOperation.MigrateDatabase, + { progressToken: "explicit-migration" }, + ); + assert.deepStrictEqual(migrated, { + migrated: true, + databaseBackend: "sqlite", + }); + assert.deepStrictEqual( + await second.wallet.client.call(WalletApiOperation.MigrateDatabase, {}), + { migrated: false, databaseBackend: "sqlite" }, + ); + const progress = notifications.filter( + (notification): notification is DatabaseMaintenanceProgressNotification => + notification.type === NotificationType.DatabaseMaintenanceProgress && + notification.operation === "indexeddb-to-native-migration", + ); + assert.ok(progress.length > 0); + assert.ok( + progress.every( + (notification) => notification.progressToken === "explicit-migration", + ), + ); + assert.strictEqual(progress.at(-1)?.completionPercent, 100); + await second.wallet.client.call(WalletApiOperation.Shutdown, {}); + assert.strictEqual((await inspectWalletDbPath(dbPath)).kind, "native"); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +test("explicit native migration can be cancelled by progress token", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "wallet-native-cancel-")); + const dbPath = path.join(dir, "wallet.sqlite3"); + try { + const existing = await makeMinimalIdbDb(dbPath); + await existing.handle.close(); + await existing.db.close(); + + const progressToken = "cancel-explicit-migration"; + const host = await createNativeWalletHost2({ + persistentStoragePath: dbPath, + }); + await host.wallet.client.call(WalletApiOperation.InitWallet, { + config: { + lazyTaskLoop: true, + testing: { skipDefaults: true }, + }, + }); + + const migration = host.wallet.client.call( + WalletApiOperation.MigrateDatabase, + { progressToken }, + ); + await host.wallet.client.call(WalletApiOperation.CancelProgressToken, { + operation: WalletApiOperation.MigrateDatabase, + progressToken, + }); + await assert.rejects( + migration, + (error: unknown) => + error instanceof TalerError && + error.errorDetail.code === TalerErrorCode.WALLET_CORE_REQUEST_CANCELLED, + ); + + assert.deepStrictEqual( + await host.wallet.client.call(WalletApiOperation.MigrateDatabase, {}), + { migrated: true, databaseBackend: "sqlite" }, + ); + await host.wallet.client.call(WalletApiOperation.Shutdown, {}); + assert.strictEqual((await inspectWalletDbPath(dbPath)).kind, "native"); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); diff --git a/packages/taler-wallet-core/src/db/migration/native.ts b/packages/taler-wallet-core/src/db/migration/native.ts @@ -46,6 +46,7 @@ */ import { + DatabaseMaintenanceProgressNotification, Duration, getErrorDetailFromException, Logger, @@ -98,6 +99,24 @@ function retentionMicros(): number { export type WalletDbFileKind = "empty" | "indexeddb" | "native" | "ambiguous"; /** + * Open storage known to have been empty before the IndexedDB bridge was + * constructed as a native wallet database. + * + * Constructing the bridge eagerly creates its six bookkeeping tables even + * when no wallet has opened it. Remove that empty scaffolding so selecting + * native storage for a new wallet is direct initialization, not a zero-record + * migration with retained backup tables and an authority marker. + */ +export async function openNativeWalletDbForEmptyStorage( + db: Sqlite3Database, +): Promise<SqliteWalletDbHandle> { + for (const table of IDB_EMULATION_TABLES) { + await db.exec(`DROP TABLE IF EXISTS "${table}"`); + } + return new SqliteWalletDbHandle(await openNativeSqliteWalletDb(db)); +} + +/** * 'rolled-back' is terminal: the emulation tables were put back by * {@link restoreMigrationBackup}, and the wallet does not migrate again on its * own, because whoever rolled back did so to stop using the native schema. @@ -316,6 +335,7 @@ export async function migrateWalletDbToNative( src: WalletDbHandle, conversionOptions: DbConversionOptions = {}, ): Promise<NativeMigrationResult> { + conversionOptions.cancellationToken?.throwIfCancelled(); // Read this before native initialization upgrades the schema. A running // marker written by versions before cleanup_safe existed is trustworthy: // those versions also cleared the native tables before recording it. Once @@ -326,6 +346,7 @@ export async function migrateWalletDbToNative( previous?.status === "running" && previous.cleanupSafe !== false; const ndb = await openNativeSqliteWalletDb(db); + conversionOptions.cancellationToken?.throwIfCancelled(); const dst = new SqliteWalletDbHandle(ndb); const txc = ndb.txc; @@ -360,6 +381,7 @@ export async function migrateWalletDbToNative( } const startedAt = nowMicros(); + conversionOptions.cancellationToken?.throwIfCancelled(); await ndb.lock.run(() => inTransaction(txc, async () => { if ((await countNativeRecords(db)) !== 0) { @@ -380,11 +402,20 @@ export async function migrateWalletDbToNative( }), ); + let lastProgress: DatabaseMaintenanceProgressNotification | undefined; try { logger.info("migrating the wallet database to the native schema"); // Verifies its own copy record by record and throws on any difference, so // reaching the next statement means the native tables hold the wallet. - const report = await convertWalletDb(src, dst, conversionOptions); + const report = await convertWalletDb(src, dst, { + ...conversionOptions, + onProgress(notification) { + lastProgress = notification; + conversionOptions.onProgress?.(notification); + }, + }); + + conversionOptions.cancellationToken?.throwIfCancelled(); await ndb.lock.run(async () => { const violations = await ( @@ -397,6 +428,8 @@ export async function migrateWalletDbToNative( } }); + conversionOptions.cancellationToken?.throwIfCancelled(); + const finishedAt = nowMicros(); const backupExpiresAt = finishedAt + retentionMicros(); @@ -442,10 +475,14 @@ export async function migrateWalletDbToNative( type: NotificationType.DatabaseMaintenanceProgress, operation: "indexeddb-to-native-migration", phase: "complete", + ...(conversionOptions.progressToken + ? { progressToken: conversionOptions.progressToken } + : {}), completedSteps: DB_CONVERSION_STEP_COUNT, totalSteps: DB_CONVERSION_STEP_COUNT, processedRecords: report.totalRecords, totalRecords: report.totalRecords, + completionPercent: 100, }); return { handle: dst, @@ -465,8 +502,18 @@ export async function migrateWalletDbToNative( type: NotificationType.DatabaseMaintenanceProgress, operation: "indexeddb-to-native-migration", phase: "failed", - completedSteps: 0, + ...(conversionOptions.progressToken + ? { progressToken: conversionOptions.progressToken } + : {}), + completedSteps: lastProgress?.completedSteps ?? 0, totalSteps: DB_CONVERSION_STEP_COUNT, + ...(lastProgress?.processedRecords !== undefined + ? { processedRecords: lastProgress.processedRecords } + : {}), + ...(lastProgress?.totalRecords !== undefined + ? { totalRecords: lastProgress.totalRecords } + : {}), + completionPercent: lastProgress?.completionPercent ?? 0, error: getErrorDetailFromException(e), }); throw e; diff --git a/packages/taler-wallet-core/src/host-impl.node.ts b/packages/taler-wallet-core/src/host-impl.node.ts @@ -51,6 +51,7 @@ import { MigrationAuthority, migrateWalletDbToNative, NativeMigrationInfo, + openNativeWalletDbForEmptyStorage, readNativeMigrationInfo, restoreMigrationBackup, resolveAmbiguousWalletDb, @@ -133,8 +134,12 @@ async function makeSqliteDb( return { path }; }; handle.getDiagnosticStats = () => myBackend.accessStats; - handle.migrateToNative = async () => - (await migrateWalletDbToNative(db, handle)).handle; + handle.migrateToNative = async (options) => + (await migrateWalletDbToNative(db, handle, options)).handle; + if (kind === "empty") { + handle.openNativeIfEmpty = async () => + openNativeWalletDbForEmptyStorage(db); + } return handle; } diff --git a/packages/taler-wallet-core/src/host-impl.qtart.ts b/packages/taler-wallet-core/src/host-impl.qtart.ts @@ -51,6 +51,7 @@ import { dropExpiredMigrationBackup, inspectWalletDbFile, migrateWalletDbToNative, + openNativeWalletDbForEmptyStorage, } from "./db/migration/native.js"; import { openNativeSqliteWalletDb } from "./db/sqlite/database.js"; import { Wallet } from "./wallet.js"; @@ -167,8 +168,12 @@ async function makeSqliteDb( throw Error(`forcing format ${forceFormat} not supported`); } }; - handle.migrateToNative = async () => - (await migrateWalletDbToNative(db, handle)).handle; + handle.migrateToNative = async (options) => + (await migrateWalletDbToNative(db, handle, options)).handle; + if (kind === "empty") { + handle.openNativeIfEmpty = async () => + openNativeWalletDbForEmptyStorage(db); + } handle.readBackupJson = async (path: string): Promise<any> => { const errObj = { errno: undefined }; const file = qjsStd.open(path, "r", errObj); diff --git a/packages/taler-wallet-core/src/requests.test.ts b/packages/taler-wallet-core/src/requests.test.ts @@ -84,10 +84,25 @@ for (const [expectedBackend, makeRunner] of backendCases) { config: { lazyTaskLoop: true, testing: { skipDefaults: true }, + features: { useNativeDb: true }, }, }); initialized = true; assert.strictEqual(response.databaseBackend, expectedBackend); + if (expectedBackend === "indexeddb") { + await assert.rejects( + wallet.client.call(WalletApiOperation.MigrateDatabase, {}), + (error: unknown) => + error instanceof TalerError && + error.errorDetail.code === + TalerErrorCode.WALLET_DB_BACKEND_UNSUPPORTED, + ); + } else { + assert.deepStrictEqual( + await wallet.client.call(WalletApiOperation.MigrateDatabase, {}), + { migrated: false, databaseBackend: "sqlite" }, + ); + } } finally { if (initialized) { await wallet.client.call(WalletApiOperation.Shutdown, {}); diff --git a/packages/taler-wallet-core/src/requests.ts b/packages/taler-wallet-core/src/requests.ts @@ -96,6 +96,8 @@ import { ListGlobalCurrencyExchangesResponse, ListWithdrawalExchangeCandidatesRequest, ListWithdrawalExchangeCandidatesResponse, + MigrateDatabaseRequest, + MigrateDatabaseResponse, ListSubscriptionsRequest, ListSubscriptionsResponse, Logger, @@ -218,6 +220,7 @@ import { codecForListExchangesRequest, codecForListWithdrawalExchangeCandidatesRequest, codecForListSubscriptionsRequest, + codecForMigrateDatabaseRequest, codecForMailboxBaseUrl, codecForMailboxConfiguration, codecForPrepareBankIntegratedWithdrawalRequest, @@ -445,7 +448,6 @@ import { import { WalletExecutionContext, - applyRunConfigDefaults, denomRefKey, migrateMaterializedTransactions, rematerializeTransactionsAtCurrentVersion, @@ -965,8 +967,6 @@ async function handleSetWalletRunConfig( innerError: err, }); } - wex.ws.initWithConfig(applyRunConfigDefaults(req.config)); - // Before anything else touches the database: everything below this line // runs against whichever backend the wallet ends up on, and the migration // needs the fixup log of the old one to have been replayed, which opening @@ -1861,6 +1861,31 @@ async function handleExportDbToFile( }; } +async function handleMigrateDatabase( + wex: WalletExecutionContext, + req: MigrateDatabaseRequest, +): Promise<MigrateDatabaseResponse> { + return await runWithMaybeProgressContext( + wex, + WalletApiOperation.MigrateDatabase, + req.progressToken, + async () => { + const migrated = await wex.ws.migrateDbToNativeSchema({ + cancellationToken: wex.cancellationToken, + progressToken: req.progressToken, + failIfUnsupported: true, + propagateFailure: true, + }); + const databaseBackend = wex.ws.db.name; + checkDbInvariant( + databaseBackend === "indexeddb" || databaseBackend === "sqlite", + `unknown wallet database backend: ${databaseBackend}`, + ); + return { migrated, databaseBackend }; + }, + ); +} + async function handleImportDb( wex: WalletExecutionContext, req: ImportDbRequest, @@ -2416,6 +2441,10 @@ const handlers: { [T in WalletApiOperation]: HandlerWithValidator<T> } = { codec: codecForExportDbToFileRequest(), handler: handleExportDbToFile, }, + [WalletApiOperation.MigrateDatabase]: { + codec: codecForMigrateDatabaseRequest(), + handler: handleMigrateDatabase, + }, [WalletApiOperation.HintApplicationResumed]: { codec: codecForEmptyObject(), handler: handleHintApplicationResumed, diff --git a/packages/taler-wallet-core/src/wallet-api-types.ts b/packages/taler-wallet-core/src/wallet-api-types.ts @@ -148,6 +148,8 @@ import { ListExchangesRequest, ListWithdrawalExchangeCandidatesRequest, ListWithdrawalExchangeCandidatesResponse, + MigrateDatabaseRequest, + MigrateDatabaseResponse, ListGlobalCurrencyAuditorsResponse, ListGlobalCurrencyExchangesResponse, ListSubscriptionsRequest, @@ -374,6 +376,7 @@ export enum WalletApiOperation { ExportDb = "exportDb", ExportDbToFile = "exportDbToFile", ImportDbFromFile = "importDbFromFile", + MigrateDatabase = "migrateDatabase", ClearDb = "clearDb", Recycle = "recycle", @@ -1428,6 +1431,16 @@ export type ImportDbOp = { }; /** + * Explicitly migrate an IndexedDB-emulation wallet to the native sqlite + * schema. The operation is idempotent when the wallet is already native. + */ +export type MigrateDatabaseOp = { + op: WalletApiOperation.MigrateDatabase; + request: MigrateDatabaseRequest; + response: MigrateDatabaseResponse; +}; + +/** * Dangerously clear the whole wallet database. */ export type ClearDbOp = { @@ -2001,6 +2014,11 @@ export const walletApiExpectedErrors = { TalerErrorCode.WALLET_DB_BACKEND_UNSUPPORTED, TalerErrorCode.WALLET_CORE_API_BAD_REQUEST, ], + [WalletApiOperation.MigrateDatabase]: [ + TalerErrorCode.WALLET_DB_BACKEND_UNSUPPORTED, + TalerErrorCode.WALLET_DB_UNAVAILABLE, + TalerErrorCode.WALLET_CORE_REQUEST_CANCELLED, + ], } as const; export type WalletOperations = { @@ -2059,6 +2077,7 @@ export type WalletOperations = { [WalletApiOperation.CreateDepositGroup]: CreateDepositGroupOp; [WalletApiOperation.ExportDbToFile]: ExportDbToFileOp; [WalletApiOperation.ImportDbFromFile]: ImportDbFromFileOp; + [WalletApiOperation.MigrateDatabase]: MigrateDatabaseOp; [WalletApiOperation.RunIntegrationTest]: RunIntegrationTestOp; [WalletApiOperation.RunIntegrationTestV2]: RunIntegrationTestV2Op; [WalletApiOperation.TestCrypto]: TestCryptoOp; diff --git a/packages/taler-wallet-core/src/wallet-db-gate.test.ts b/packages/taler-wallet-core/src/wallet-db-gate.test.ts @@ -9,6 +9,7 @@ import assert from "node:assert"; import { test } from "node:test"; +import { CancellationToken } from "@gnu-taler/taler-util"; import { WalletDbHandle } from "./db/handle.js"; import { AdmittedWalletDbHandle, DbOperationGate } from "./wallet.js"; @@ -119,3 +120,26 @@ test("database import has exclusive admission", async () => { "after", ]); }); + +test("database gate cancels an exclusive waiter", async () => { + const gate = new DbOperationGate(); + const active = deferred(); + const releaseActive = deferred(); + const shared = gate.runShared(async () => { + active.resolve(); + await releaseActive.promise; + }); + await active.promise; + + const cts = CancellationToken.create(); + const exclusive = gate.runExclusive(async () => { + assert.fail("cancelled exclusive operation was admitted"); + }, cts.token); + await Promise.resolve(); + cts.cancel(); + await assert.rejects(exclusive, CancellationToken.CancellationError); + + releaseActive.resolve(); + await shared; + await gate.runShared(async () => {}); +}); diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts @@ -35,6 +35,7 @@ import { Duration, FlightRecordEntry, FlightRecordEvent, + InitRequest, Logger, LongpollQueue, NotificationType, @@ -62,6 +63,7 @@ import { performanceDelta, performanceNow, safeStringifyException, + codecForInitRequest, } from "@gnu-taler/taler-util"; import { getenv } from "@gnu-taler/taler-util/compat"; import { type HttpRequestLibrary } from "@gnu-taler/taler-util/http"; @@ -169,13 +171,20 @@ export class DbOperationGate { } } - async acquireExclusive(): Promise<() => void> { + async acquireExclusive( + cancellationToken: CancellationToken = CancellationToken.CONTINUE, + ): Promise<() => void> { this.waitingExclusive++; try { - while (this.exclusive || this.shared !== 0) await this.changed(); + while (this.exclusive || this.shared !== 0) { + cancellationToken.throwIfCancelled(); + await cancellationToken.racePromise(this.changed()); + } + cancellationToken.throwIfCancelled(); this.exclusive = true; } finally { this.waitingExclusive--; + this.wake(); } let released = false; return () => { @@ -186,8 +195,11 @@ export class DbOperationGate { }; } - async runExclusive<T>(f: () => Promise<T>): Promise<T> { - const release = await this.acquireExclusive(); + async runExclusive<T>( + f: () => Promise<T>, + cancellationToken: CancellationToken = CancellationToken.CONTINUE, + ): Promise<T> { + const release = await this.acquireExclusive(cancellationToken); try { return await f(); } finally { @@ -667,12 +679,22 @@ async function dispatchWalletCoreApiRequest( id: string, payload: unknown, ): Promise<CoreApiResponse> { - if (!isWalletInitOperation(operation)) { + const isInitOperation = isWalletInitOperation(operation); + if (!isInitOperation) { if (!ws.initCalled) { throw Error("init must be called first"); } } + if (isInitOperation) { + const req: InitRequest = codecForInitRequest().decode(payload); + const config = applyRunConfigDefaults(req.config); + ws.initWithConfig(config); + if (!ws.initCalled && config.features.useNativeDb) { + await ws.openNativeDatabaseIfEmpty(); + } + } + await ws.ensureWalletDbOpen(); let wex: WalletExecutionContext; @@ -703,7 +725,6 @@ async function dispatchWalletCoreApiRequest( const start = performanceNow(); try { - await ws.ensureWalletDbOpen(); oc.observe({ type: ObservabilityEventType.RequestStart, name: operation, @@ -842,6 +863,7 @@ export function applyRunConfigDefaults( allowHttp: true, migrateNativeDb: wcp?.features?.migrateNativeDb ?? migrateNativeDbFromEnv() ?? false, + useNativeDb: wcp?.features?.useNativeDb ?? false, }, testing: { devModeActive: wcp?.testing?.devModeActive ?? false, @@ -915,7 +937,6 @@ export class Wallet { id: string, payload: unknown, ): Promise<CoreApiResponse> { - await this.ws.ensureWalletDbOpen(); return dispatchWalletCoreApiRequest(this.ws, operation, id, payload); } } @@ -1258,24 +1279,65 @@ export class InternalWalletState { } } + /** Select native sqlite before an empty database is opened as IndexedDB. */ + async openNativeDatabaseIfEmpty(): Promise<boolean> { + if (this.loadingDb) { + while (this.loadingDb) { + await this.loadingDbCond.wait(); + } + } + const oldHandle = this.dbHandle; + if (!oldHandle.openNativeIfEmpty) { + logger.trace("this host offers no native backend for empty storage"); + return false; + } + this.loadingDb = true; + try { + return await this.dbOperationGate.runExclusive(async () => { + if (this.dbHandle !== oldHandle || !oldHandle.openNativeIfEmpty) { + return false; + } + const newHandle = await oldHandle.openNativeIfEmpty(); + this.dbHandle = newHandle; + newHandle.setNotificationSink((n) => this.notify(n)); + await oldHandle.close(); + this.clearAllCaches(); + logger.info("initialized an empty database with the native schema"); + return true; + }); + } finally { + this.loadingDb = false; + this.loadingDbCond.trigger(); + } + } + /** * Migrate the database to the native schema, if it is not already there and * the host can do it. * - * Returns whether the wallet is on the native schema afterwards. A failed - * migration is logged and reported, not thrown: the old database is - * untouched and the wallet keeps running on it, which is a far better - * outcome than refusing to start. + * Returns whether this call changed the active backend. Automatic startup + * migration logs a failure and keeps running on the untouched old database; + * the explicit API asks this method to propagate failures instead. * * Runs under the same guard as opening and importing, so no transaction can * be in flight while the handle underneath the wallet is replaced. Every * transaction resolves the handle through this class when it starts, so * work that was queued before the swap runs against the new database. */ - async migrateDbToNativeSchema(): Promise<boolean> { + async migrateDbToNativeSchema( + options: { + cancellationToken?: CancellationToken; + progressToken?: string; + failIfUnsupported?: boolean; + propagateFailure?: boolean; + } = {}, + ): Promise<boolean> { + const cancellationToken = + options.cancellationToken ?? CancellationToken.CONTINUE; if (this.loadingDb) { while (this.loadingDb) { - await this.loadingDbCond.wait(); + cancellationToken.throwIfCancelled(); + await cancellationToken.racePromise(this.loadingDbCond.wait()); } } // Resolve only after earlier initialization/migration work has completed: @@ -1286,6 +1348,13 @@ export class InternalWalletState { // Either the database is already native or the host cannot migrate it; // both are "nothing to do", and neither is worth an alarming message on // every initialization. + if (oldHandle.name !== "sqlite" && options.failIfUnsupported) { + throw TalerError.fromDetail( + TalerErrorCode.WALLET_DB_BACKEND_UNSUPPORTED, + { backend: oldHandle.name }, + "this database backend cannot be migrated in place", + ); + } logger.trace("this wallet database offers no in-place migration"); return false; } @@ -1293,10 +1362,20 @@ export class InternalWalletState { try { return await this.dbOperationGate.runExclusive(async () => { try { - const newHandle = await oldHandle.migrateToNative!(); + cancellationToken.throwIfCancelled(); + const newHandle = await oldHandle.migrateToNative!({ + cancellationToken, + progressToken: options.progressToken, + }); this.dbHandle = newHandle; newHandle.setNotificationSink((n) => this.notify(n)); - await oldHandle.close(); + try { + await oldHandle.close(); + } catch (e) { + logger.warn( + `closing the migrated IndexedDB handle failed: ${safeStringifyException(e)}`, + ); + } this.clearAllCaches(); return true; } catch (e) { @@ -1304,9 +1383,22 @@ export class InternalWalletState { `migration to the native database failed, continuing with the` + ` existing one: ${safeStringifyException(e)}`, ); + if (options.propagateFailure) { + if ( + e instanceof CancellationToken.CancellationError || + (e instanceof TalerError && + e.errorDetail.code === + TalerErrorCode.WALLET_CORE_REQUEST_CANCELLED) + ) { + throw e; + } + throw TalerError.fromDetail(TalerErrorCode.WALLET_DB_UNAVAILABLE, { + innerError: getErrorDetailFromException(e), + }); + } return false; } - }); + }, cancellationToken); } finally { this.loadingDb = false; this.loadingDbCond.trigger();