commit c50e8107a66e7e6e1c015ace81f485d15b0c5ea3 parent de7a134a72eaac2818ae2877728ef6b0ff972d6d Author: Florian Dold <dold@taler.net> Date: Wed, 26 Aug 2026 22:35:49 +0200 wallet-core: throttle database maintenance progress Diffstat:
12 files changed, 569 insertions(+), 73 deletions(-)
diff --git a/packages/taler-wallet-core/src/db/indexeddb/database.ts b/packages/taler-wallet-core/src/db/indexeddb/database.ts @@ -248,6 +248,7 @@ function onMetaDbUpgradeNeeded( export async function openTalerDatabase( idbFactory: IDBFactory, onVersionChange: () => void, + onUpgradeStart?: (oldVersion: number, newVersion: number) => void, ): Promise<IDBDatabase> { const state = await readMainDbState(idbFactory, true); await cleanInterruptedDatabaseReplacement(idbFactory, state); @@ -255,6 +256,7 @@ export async function openTalerDatabase( idbFactory, state.current, onVersionChange, + onUpgradeStart, ); } @@ -359,6 +361,7 @@ async function openTalerDatabaseGeneration( idbFactory: IDBFactory, name: string, onVersionChange: () => void, + onUpgradeStart?: (oldVersion: number, newVersion: number) => void, ): Promise<IDBDatabase> { if (!isCurrentGenerationName(name)) { throw Error(`invalid wallet database generation name ${name}`); @@ -368,7 +371,10 @@ async function openTalerDatabaseGeneration( name, WALLET_DB_MINOR_VERSION, onVersionChange, - onTalerDbUpgradeNeeded, + (db, oldVersion, newVersion, transaction) => { + onUpgradeStart?.(oldVersion, newVersion); + onTalerDbUpgradeNeeded(db, oldVersion, newVersion, transaction); + }, ); } diff --git a/packages/taler-wallet-core/src/db/indexeddb/fixups.ts b/packages/taler-wallet-core/src/db/indexeddb/fixups.ts @@ -166,6 +166,11 @@ export const walletDbFixups: FixupDescription[] = [ }, ]; +/** Schema opening, every record fixup, and transaction rematerialization. */ +export const WALLET_DB_MAINTENANCE_TOTAL_STEPS = walletDbFixups.length + 2; +export const WALLET_DB_SCHEMA_UPGRADE_STEP = "indexeddb-schema-upgrade"; +export const WALLET_DB_REMATERIALIZE_STEP = "rematerialize-transactions"; + /** * Delete coin histories whose coins were removed by the old IndexedDB * exchange-purge implementation. @@ -832,17 +837,33 @@ async function fixup20260812ExchangeWithdrawValues( export async function applyFixups( db: DbAccess<typeof WalletIndexedDbStoresV1>, onProgress: (notification: WalletNotification) => void = () => {}, + options: { deferCompletion?: boolean } = {}, ): Promise<number> { logger.trace("applying fixups"); let count = 0; + // Most opens have no work to do. Read the marker store once rather than + // opening an all-store read-write transaction for every known fixup. Load + // it lazily in the first fixup transaction, so an old database with work to + // do does not pay for an additional transaction just for the inventory. + let completedFixups: Set<string> | undefined; for (let index = 0; index < walletDbFixups.length; index++) { const fixupInstruction = walletDbFixups[index]; + if (completedFixups?.has(fixupInstruction.name)) { + continue; + } let applied = false; try { await db.runAllStoresReadWriteTx({}, async (tx) => { logger.trace(`checking fixup ${fixupInstruction.name}`); - const fixupRecord = await tx.fixups.get(fixupInstruction.name); - if (fixupRecord) { + if (!completedFixups) { + completedFixups = new Set( + (await tx.fixups.getAll()).map((x) => x.fixupName), + ); + } else if (await tx.fixups.get(fixupInstruction.name)) { + // Preserve atomicity if another context raced the marker snapshot. + completedFixups.add(fixupInstruction.name); + } + if (completedFixups.has(fixupInstruction.name)) { return; } applied = true; @@ -852,8 +873,8 @@ export async function applyFixups( operation: "indexeddb-fixup", phase: "fixup", step: fixupInstruction.name, - completedSteps: index, - totalSteps: walletDbFixups.length, + completedSteps: index + 1, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, }); await fixupInstruction.fn(tx); // A fixup may change any operation record from which transactionsMeta @@ -865,6 +886,7 @@ export async function applyFixups( fixupName: fixupInstruction.name, }); }); + completedFixups!.add(fixupInstruction.name); } catch (e) { if (applied) { onProgress({ @@ -872,8 +894,8 @@ export async function applyFixups( operation: "indexeddb-fixup", phase: "failed", step: fixupInstruction.name, - completedSteps: index, - totalSteps: walletDbFixups.length, + completedSteps: index + 1, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, error: getErrorDetailFromException(e), }); } @@ -888,19 +910,19 @@ export async function applyFixups( operation: "indexeddb-fixup", phase: "fixup", step: fixupInstruction.name, - completedSteps: index + 1, - totalSteps: walletDbFixups.length, + completedSteps: index + 2, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, }); count++; } } - if (count > 0) { + if (count > 0 && !options.deferCompletion) { onProgress({ type: NotificationType.DatabaseMaintenanceProgress, operation: "indexeddb-fixup", phase: "complete", - completedSteps: walletDbFixups.length, - totalSteps: walletDbFixups.length, + completedSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, }); } return count; diff --git a/packages/taler-wallet-core/src/db/indexeddb/handle.ts b/packages/taler-wallet-core/src/db/indexeddb/handle.ts @@ -23,7 +23,9 @@ import { CancellationToken, + getErrorDetailFromException, Logger, + NotificationType, WalletNotification, } from "@gnu-taler/taler-util"; import { @@ -41,7 +43,11 @@ import { retireTalerDatabaseGeneration, } from "./database.js"; import { exportDb, importDb } from "./dump.js"; -import { applyFixups } from "./fixups.js"; +import { + applyFixups, + WALLET_DB_MAINTENANCE_TOTAL_STEPS, + WALLET_DB_SCHEMA_UPGRADE_STEP, +} from "./fixups.js"; import { WalletIndexedDbStoresV1 } from "./schema.js"; import { WalletDbAccessStats, @@ -82,7 +88,9 @@ export class IdbWalletDbHandle implements WalletDbHandle { private idbHandle: IDBDatabase | undefined; private dbAccess: DbAccess<typeof WalletIndexedDbStoresV1> | undefined; - private opening: Promise<{ fixupsApplied: number }> | undefined; + private opening: + | Promise<{ fixupsApplied: number; schemaUpgraded: boolean }> + | undefined; private notify: (n: WalletNotification) => void = () => {}; @@ -132,9 +140,12 @@ export class IdbWalletDbHandle implements WalletDbHandle { * Returns whether fixups changed anything, which the caller needs in order * to decide whether wallet-level views have to be rebuilt. */ - async ensureOpen(): Promise<{ fixupsApplied: number }> { + async ensureOpen(): Promise<{ + fixupsApplied: number; + schemaUpgraded: boolean; + }> { if (this.dbAccess) { - return { fixupsApplied: 0 }; + return { fixupsApplied: 0, schemaUpgraded: false }; } if (this.opening) { return await this.opening; @@ -150,16 +161,54 @@ export class IdbWalletDbHandle implements WalletDbHandle { } } - private async openDatabase(): Promise<{ fixupsApplied: number }> { - const idbHandle = await openTalerDatabase(this.idbFactory, async () => {}); + private async openDatabase(): Promise<{ + fixupsApplied: number; + schemaUpgraded: boolean; + }> { + let schemaUpgraded = false; + let upgradeStep: string | undefined; + let idbHandle: IDBDatabase; + try { + idbHandle = await openTalerDatabase( + this.idbFactory, + async () => {}, + (oldVersion, newVersion) => { + schemaUpgraded = true; + upgradeStep = `${WALLET_DB_SCHEMA_UPGRADE_STEP}-${oldVersion}-to-${newVersion}`; + this.emitNotification({ + type: NotificationType.DatabaseMaintenanceProgress, + operation: "indexeddb-fixup", + phase: "fixup", + step: upgradeStep, + completedSteps: 0, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + }); + }, + ); + } catch (e) { + if (upgradeStep) { + this.emitNotification({ + type: NotificationType.DatabaseMaintenanceProgress, + operation: "indexeddb-fixup", + phase: "failed", + step: upgradeStep, + completedSteps: 0, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + error: getErrorDetailFromException(e), + }); + } + throw e; + } const dbAccess = this.makeAccess(idbHandle); try { - const fixupsApplied = await this.applyDbFixups(dbAccess, (n) => - this.emitNotification(n), + const fixupsApplied = await this.applyDbFixups( + dbAccess, + (n) => this.emitNotification(n), + { deferCompletion: true }, ); this.idbHandle = idbHandle; this.dbAccess = dbAccess; - return { fixupsApplied }; + return { fixupsApplied, schemaUpgraded }; } catch (e) { idbHandle.close(); throw e; @@ -263,12 +312,21 @@ export class IdbWalletDbHandle implements WalletDbHandle { await tx.fixups.delete(fx.fixupName); } }); - await this.applyDbFixups(stagedAccess, (n) => - stagedNotifications.push(n), + await this.applyDbFixups( + stagedAccess, + (n) => stagedNotifications.push(n), + { deferCompletion: true }, ); await stagedAccess.runAllStoresReadWriteTx({}, async (tx) => { await finalize(new IdbWalletTransaction(tx)); }); + stagedNotifications.push({ + type: NotificationType.DatabaseMaintenanceProgress, + operation: "indexeddb-fixup", + phase: "complete", + completedSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + }); // This metadata transaction is the commit point. A crash before it // keeps oldName authoritative; a crash afterwards opens staged.name. diff --git a/packages/taler-wallet-core/src/db/indexeddb/transaction.ts b/packages/taler-wallet-core/src/db/indexeddb/transaction.ts @@ -114,12 +114,7 @@ export class IdbWalletTransaction implements WalletDbTransaction { this.tx = tx; } - async scanMigrationRecords<T>( - store: WalletDbMigrationStore, - _read: (tx: WalletDbTransaction) => Promise<T[]>, - cursor: unknown | undefined, - limit: number, - ): Promise<WalletDbMigrationPage<T>> { + private migrationStore(store: WalletDbMigrationStore): any { const physicalStore: Record<WalletDbMigrationStore, string> = { config: "config", currencyInfo: "currencyInfo", @@ -168,6 +163,20 @@ export class IdbWalletTransaction implements WalletDbTransaction { if (!accessor) { throw Error(`migration store ${store} is not available`); } + return accessor; + } + + async countMigrationRecords(store: WalletDbMigrationStore): Promise<number> { + return await this.migrationStore(store).count(); + } + + async scanMigrationRecords<T>( + store: WalletDbMigrationStore, + _read: (tx: WalletDbTransaction) => Promise<T[]>, + cursor: unknown | undefined, + limit: number, + ): Promise<WalletDbMigrationPage<T>> { + const accessor = this.migrationStore(store); const page = await accessor.scan(cursor, limit); return { records: page.records as T[], diff --git a/packages/taler-wallet-core/src/db/migration/converter.test.ts b/packages/taler-wallet-core/src/db/migration/converter.test.ts @@ -410,9 +410,16 @@ test("converter: IndexedDB to sqlite, populated by the conformance corpus", asyn n.operation === "indexeddb-to-native-migration", ); assert.ok( - maintenanceProgress.every((n) => n.totalRecords === report.totalRecords), + maintenanceProgress + .slice(1) + .every((n) => n.totalRecords === report.totalRecords), "progress did not carry the global record total", ); + assert.strictEqual( + maintenanceProgress[0].totalRecords, + undefined, + "initial progress unexpectedly waited for record counting", + ); assert.ok( maintenanceProgress.every((n) => n.progressToken === "converter-progress"), "progress did not carry the request token", diff --git a/packages/taler-wallet-core/src/db/migration/converter.ts b/packages/taler-wallet-core/src/db/migration/converter.ts @@ -668,24 +668,37 @@ export async function convertWalletDb( ): Promise<DbConversionReport> { const copied: Record<string, number> = {}; - // Inventorying the source up front makes the progress denominator known - // before the first write. Keep the digests: verification can compare the - // destination against these instead of scanning the source a second time, - // so global progress does not add another full database pass. - const sourceDigests = new Map<WalletDbMigrationStore, RecordMultisetDigest>(); - let totalRecords = 0; - for (const group of COPY_PLAN) { - for (const st of group) { - const digest = await digestStore( - src, - st, - undefined, - options.cancellationToken, - ); - sourceDigests.set(st.name, digest); - totalRecords += digest.count; + // Let the host display maintenance immediately. Counting is cheap, but an + // old emulated database can still spend noticeable time opening its first + // transaction on a phone. + const initialNotification: DatabaseMaintenanceProgressNotification = { + type: NotificationType.DatabaseMaintenanceProgress, + operation: "indexeddb-to-native-migration", + phase: "copy", + ...(options.progressToken ? { progressToken: options.progressToken } : {}), + completedSteps: 0, + totalSteps: DB_CONVERSION_STEP_COUNT, + completionPercent: 0, + }; + src.emitNotification(initialNotification); + options.onProgress?.(initialNotification); + + options.cancellationToken?.throwIfCancelled(); + const totalRecords = await src.runReadWriteTx(async (tx) => { + let count = 0; + for (const group of COPY_PLAN) { + for (const st of group) { + count += await tx.countMigrationRecords(st.name); + } } - } + return count; + }); + options.cancellationToken?.throwIfCancelled(); + + // Build the source digests while copying. The previous implementation + // hashed the whole source in a separate inventory pass, doubling source + // enumeration merely to learn totals that the backends can count directly. + const sourceDigests = new Map<WalletDbMigrationStore, RecordMultisetDigest>(); const progressInterval = DB_CONVERSION_PROGRESS_RECORDS; const latestProcessedRecords = { copy: 0, verify: 0 }; @@ -748,11 +761,16 @@ export async function convertWalletDb( for (const st of group) { let cursor: unknown | undefined; let storeCount = 0; + const sourceDigest = new RecordMultisetDigest(); + const normalize = st.normalize ?? ((r: unknown) => r); while (true) { options.cancellationToken?.throwIfCancelled(); const page = await readPage(src, st, cursor, true); options.cancellationToken?.throwIfCancelled(); if (page.records.length === 0) break; + for (const record of page.records) { + sourceDigest.add(normalize(record)); + } options.cancellationToken?.throwIfCancelled(); await dst.runReadWriteTx(async (tx) => { for (const rec of page.records) { @@ -766,12 +784,19 @@ export async function convertWalletDb( cursor = page.nextCursor; if (cursor === undefined) break; } + sourceDigests.set(st.name, sourceDigest); copied[st.name] = storeCount; stepIndex++; notify("copy", stepIndex, st); logger.trace(`copied ${storeCount} ${st.name}`); } } + if (copiedRecords !== totalRecords) { + throw Error( + `conversion source changed while copying: counted ${totalRecords}` + + ` records but copied ${copiedRecords}`, + ); + } copyProgress(stepIndex, copiedRecords, undefined, true); // Verify using fixed-size multiset digests. Source and destination have diff --git a/packages/taler-wallet-core/src/db/sqlite/transaction.ts b/packages/taler-wallet-core/src/db/sqlite/transaction.ts @@ -432,6 +432,12 @@ export class SqliteWalletTransaction implements WalletDbTransaction { } } + async countMigrationRecords(store: WalletDbMigrationStore): Promise<number> { + const table = SQLITE_MIGRATION_TABLES[store]; + const row = await this.first(`SELECT COUNT(*) AS count FROM ${table}`); + return num(row?.count); + } + // Bound as an instance property for the same reason as the IndexedDB // implementation: call sites pass it around unbound. notify = (notif: WalletNotification): void => { diff --git a/packages/taler-wallet-core/src/db/transaction.ts b/packages/taler-wallet-core/src/db/transaction.ts @@ -211,6 +211,9 @@ export interface WalletDbMigrationPage<T> { } export interface WalletDbTransaction { + /** Count the root records represented by one migration store. */ + countMigrationRecords(store: WalletDbMigrationStore): Promise<number>; + /** * Read a bounded page for database conversion. * diff --git a/packages/taler-wallet-core/src/maintenance-notifications.test.ts b/packages/taler-wallet-core/src/maintenance-notifications.test.ts @@ -0,0 +1,153 @@ +/* + This file is part of GNU Taler + (C) 2026 Taler Systems S.A. + + GNU Taler is free software; you can redistribute it and/or modify it under the + terms of the GNU General Public License as published by the Free Software + Foundation; either version 3, or (at your option) any later version. + + GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY + WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR + A PARTICULAR PURPOSE. See the GNU General Public License for more details. + + You should have received a copy of the GNU General Public License along with + GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/> + */ + +import assert from "node:assert"; +import { test } from "node:test"; + +import { + DatabaseMaintenanceProgressNotification, + NotificationType, + TimerAPI, + TimerHandle, +} from "@gnu-taler/taler-util"; +import { + DATABASE_MAINTENANCE_NOTIFICATION_INTERVAL_MS, + MaintenanceNotificationThrottler, +} from "./maintenance-notifications.js"; + +interface ScheduledTimer { + due: number; + callback: () => void; + active: boolean; +} + +class ManualTimer implements TimerAPI { + nowMs = 0; + private readonly scheduled: ScheduledTimer[] = []; + + after(delayMs: number, callback: () => void): TimerHandle { + const scheduled = { + due: this.nowMs + delayMs, + callback, + active: true, + }; + this.scheduled.push(scheduled); + return { + clear: () => { + scheduled.active = false; + }, + unref: () => {}, + }; + } + + every(): TimerHandle { + throw Error("not used by the maintenance notification throttler"); + } + + advance(ms: number): void { + this.nowMs += ms; + while (true) { + const next = this.scheduled + .filter((x) => x.active && x.due <= this.nowMs) + .sort((a, b) => a.due - b.due)[0]; + if (!next) return; + next.active = false; + next.callback(); + } + } +} + +function progress( + step: string, + phase: DatabaseMaintenanceProgressNotification["phase"] = "fixup", +): DatabaseMaintenanceProgressNotification { + return { + type: NotificationType.DatabaseMaintenanceProgress, + operation: "indexeddb-fixup", + phase, + step, + completedSteps: 0, + totalSteps: 10, + }; +} + +test("maintenance notifications deliver first and terminal events immediately", () => { + const timer = new ManualTimer(); + const delivered: DatabaseMaintenanceProgressNotification[] = []; + const throttler = new MaintenanceNotificationThrottler( + timer, + (n) => delivered.push(n as DatabaseMaintenanceProgressNotification), + () => BigInt(timer.nowMs) * 1_000_000n, + ); + + throttler.offer(progress("first")); + assert.deepStrictEqual( + delivered.map((x) => x.step), + ["first"], + ); + + timer.advance(100); + throttler.offer(progress("pending")); + throttler.offer(progress("failed", "failed")); + assert.deepStrictEqual( + delivered.map((x) => x.step), + ["first", "failed"], + ); + + timer.advance(DATABASE_MAINTENANCE_NOTIFICATION_INTERVAL_MS); + assert.deepStrictEqual( + delivered.map((x) => x.step), + ["first", "failed"], + "a terminal event cancels pending progress", + ); +}); + +test("maintenance notifications coalesce progress into 500ms intervals", () => { + const timer = new ManualTimer(); + const delivered: DatabaseMaintenanceProgressNotification[] = []; + const throttler = new MaintenanceNotificationThrottler( + timer, + (n) => delivered.push(n as DatabaseMaintenanceProgressNotification), + () => BigInt(timer.nowMs) * 1_000_000n, + ); + + throttler.offer(progress("first")); + timer.advance(100); + throttler.offer(progress("obsolete")); + timer.advance(100); + throttler.offer(progress("latest")); + timer.advance(299); + assert.deepStrictEqual( + delivered.map((x) => x.step), + ["first"], + ); + + timer.advance(1); + assert.deepStrictEqual( + delivered.map((x) => x.step), + ["first", "latest"], + ); + + timer.advance(100); + throttler.offer(progress("next")); + timer.advance(399); + assert.strictEqual(delivered.length, 2); + timer.advance(1); + assert.deepStrictEqual( + delivered.map((x) => x.step), + ["first", "latest", "next"], + ); +}); diff --git a/packages/taler-wallet-core/src/maintenance-notifications.ts b/packages/taler-wallet-core/src/maintenance-notifications.ts @@ -0,0 +1,102 @@ +/* + This file is part of GNU Taler + (C) 2026 Taler Systems S.A. + + GNU Taler is free software; you can redistribute it and/or modify it under the + terms of the GNU General Public License as published by the Free Software + Foundation; either version 3, or (at your option) any later version. + + GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY + WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR + A PARTICULAR PURPOSE. See the GNU General Public License for more details. + + You should have received a copy of the GNU General Public License along with + GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/> + */ + +import { + NotificationType, + TimerAPI, + TimerHandle, + WalletNotification, + performanceNow, +} from "@gnu-taler/taler-util"; + +export const DATABASE_MAINTENANCE_NOTIFICATION_INTERVAL_MS = 500; +const DATABASE_MAINTENANCE_NOTIFICATION_INTERVAL_NS = + BigInt(DATABASE_MAINTENANCE_NOTIFICATION_INTERVAL_MS) * 1_000_000n; + +interface MaintenanceNotificationState { + lastDelivered: bigint; + pending?: WalletNotification; + timer?: TimerHandle; +} + +/** + * Keep database-maintenance progress useful without flooding native hosts. + * + * Maintenance runs while the database gate is held. In the Qt host the + * sqlite calls underneath it are synchronous, so the ordinary asynchronous + * notification delivery can be starved until all queued callbacks arrive as + * one burst. The first and terminal events are therefore delivered directly; + * intermediate state is coalesced behind a single timer per operation. + */ +export class MaintenanceNotificationThrottler { + private readonly states = new Map<string, MaintenanceNotificationState>(); + + constructor( + private readonly timer: TimerAPI, + private readonly deliver: (notification: WalletNotification) => void, + private readonly now: () => bigint = performanceNow, + ) {} + + offer( + notification: Extract< + WalletNotification, + { type: NotificationType.DatabaseMaintenanceProgress } + >, + ): void { + const key = notification.operation; + const terminal = + notification.phase === "complete" || notification.phase === "failed"; + const previous = this.states.get(key); + if (terminal) { + previous?.timer?.clear(); + this.states.delete(key); + this.deliver(notification); + return; + } + + const now = this.now(); + if ( + !previous || + now - previous.lastDelivered >= + DATABASE_MAINTENANCE_NOTIFICATION_INTERVAL_NS + ) { + previous?.timer?.clear(); + this.states.set(key, { lastDelivered: now }); + this.deliver(notification); + return; + } + + previous.pending = notification; + if (previous.timer) return; + const remainingNs = + DATABASE_MAINTENANCE_NOTIFICATION_INTERVAL_NS - + (now - previous.lastDelivered); + const remainingMs = Math.max(1, Math.ceil(Number(remainingNs) / 1_000_000)); + previous.timer = this.timer.after(remainingMs, () => { + previous.timer = undefined; + const pending = previous.pending; + previous.pending = undefined; + if (!pending) return; + previous.lastDelivered = this.now(); + this.deliver(pending); + }); + } + + stop(): void { + for (const state of this.states.values()) state.timer?.clear(); + this.states.clear(); + } +} diff --git a/packages/taler-wallet-core/src/requests.test.ts b/packages/taler-wallet-core/src/requests.test.ts @@ -22,12 +22,16 @@ import { ExchangeEntryStatus, ExchangeEntrySource, ExchangeRecommendationReason, + NotificationType, TalerError, TalerErrorCode, TalerPreciseTimestamp, SetTimeoutTimerAPI, + DatabaseMaintenanceProgressNotification, } from "@gnu-taler/taler-util"; import { HttpRequestLibrary } from "@gnu-taler/taler-util/http"; +import { BridgeIDBFactory, createSqliteBackend } from "@gnu-taler/idb-bridge"; +import { createNodeHelperSqlite3Impl } from "@gnu-taler/idb-bridge/node-helper-sqlite3-impl"; import { ConfigRecord, @@ -40,7 +44,8 @@ import { timestampPreciseToDb, } from "./db/records.js"; import { WalletDbTransaction } from "./db/transaction.js"; -import { makeIdbRunner, makeSqliteRunner } from "./db/testing/runners.js"; +import { makeSqliteRunner } from "./db/testing/runners.js"; +import { IdbWalletDbHandle } from "./db/indexeddb/handle.js"; import { markExchangeAddedByUser } from "./exchanges.js"; import { SynchronousCryptoWorkerFactoryPlain } from "./crypto/workers/synchronousWorkerFactoryPlain.js"; import { @@ -53,8 +58,23 @@ import { import { WalletApiOperation } from "./wallet-api-types.js"; import { Wallet, WalletExecutionContext } from "./wallet.js"; +async function makeUnopenedIdbRunner(): Promise<IdbWalletDbHandle> { + const sqlite3Impl = await createNodeHelperSqlite3Impl({ + enableTracing: false, + }); + const backend = await createSqliteBackend(sqlite3Impl, { + filename: ":memory:", + }); + backend.enableTracing = false; + BridgeIDBFactory.enableTracing = false; + return new IdbWalletDbHandle( + new BridgeIDBFactory(backend) as any, + () => backend.accessStats, + ); +} + const backendCases = [ - ["indexeddb", makeIdbRunner], + ["indexeddb", makeUnopenedIdbRunner], ["sqlite", makeSqliteRunner], ] as const; @@ -78,18 +98,35 @@ for (const [expectedBackend, makeRunner] of backendCases) { new SetTimeoutTimerAPI(), new SynchronousCryptoWorkerFactoryPlain(), ); + const maintenanceNotifications: DatabaseMaintenanceProgressNotification[] = + []; + wallet.addNotificationListener((notification) => { + if (notification.type === NotificationType.DatabaseMaintenanceProgress) { + maintenanceNotifications.push(notification); + } + }); let initialized = false; try { - const response = await wallet.client.call(WalletApiOperation.InitWallet, { - config: { - lazyTaskLoop: true, - testing: { skipDefaults: true }, - features: { useNativeDb: true }, + const response = await wallet.client.call( + WalletApiOperation.SetWalletRunConfig, + { + config: { + lazyTaskLoop: true, + testing: { skipDefaults: true }, + features: { migrateNativeDb: false, useNativeDb: true }, + }, }, - }); + ); initialized = true; assert.strictEqual(response.databaseBackend, expectedBackend); if (expectedBackend === "indexeddb") { + assert.ok( + maintenanceNotifications[0]?.step?.startsWith( + "indexeddb-schema-upgrade-", + ), + "the first notification describes the IndexedDB schema upgrade", + ); + assert.strictEqual(maintenanceNotifications.at(-1)?.phase, "complete"); await assert.rejects( wallet.client.call(WalletApiOperation.MigrateDatabase, {}), (error: unknown) => diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts @@ -74,6 +74,10 @@ import { } from "./crypto/workers/crypto-dispatcher.js"; import { ConfigRecordKey, WalletDenomination } from "./db/records.js"; import { IdbWalletDbHandle } from "./db/indexeddb/handle.js"; +import { + WALLET_DB_MAINTENANCE_TOTAL_STEPS, + WALLET_DB_REMATERIALIZE_STEP, +} from "./db/indexeddb/fixups.js"; import { WalletDbHandle } from "./db/handle.js"; import { watchForCacheInvalidation } from "./db/shared.js"; import { @@ -113,6 +117,7 @@ import { walletApiExpectedErrors, } from "./wallet-api-types.js"; import { updateWithdrawalDenomsForExchange } from "./withdraw.js"; +import { MaintenanceNotificationThrottler } from "./maintenance-notifications.js"; const logger = new Logger("wallet.ts"); @@ -1072,6 +1077,8 @@ export class InternalWalletState { private suspendedDbRelease: (() => void) | undefined; + private maintenanceNotifications: MaintenanceNotificationThrottler; + performanceStats: PerformanceTable = {}; /** @@ -1214,12 +1221,16 @@ export class InternalWalletState { () => this.dbHandle, this.dbOperationGate, ); - // The host opened the database before this wallet existed, so its - // notifications had nowhere to go until now. - dbHandle.setNotificationSink((n) => this.notify(n)); this.cryptoDispatcher = new CryptoDispatcher(cryptoWorkerFactory); this.cryptoApi = this.cryptoDispatcher.cryptoApi; this.timerGroup = new TimerGroup(timer); + this.maintenanceNotifications = new MaintenanceNotificationThrottler( + timer, + (n) => this.deliverNotificationSynchronously(n), + ); + // The host opened the database before this wallet existed, so its + // notifications had nowhere to go until now. + dbHandle.setNotificationSink((n) => this.notify(n)); // Migration record used for testing with sandcastle or local deployment. this.exchangeMigrationPlan.set("http://exchange.taler.localhost:4321/", { newExchangeBaseUrl: "http://exchange.taler2.localhost:4321/", @@ -1248,20 +1259,8 @@ export class InternalWalletState { // is a wallet-level concern that no storage layer should know about. await this.dbOperationGate.runExclusive(async () => { const idb = this.idbOnly; - const fixupsApplied = idb ? (await idb.ensureOpen()).fixupsApplied : 0; - if (fixupsApplied > 0) { - const wex = getNormalWalletExecutionContext( - this, - CancellationToken.CONTINUE, - undefined, - { observe(evt: any) {} }, - ); - // Use the raw handle while holding exclusive admission; routing - // through admittedDb would try to acquire shared admission and - // deadlock behind ourselves. - await this.dbHandle.runReadWriteTx(async (tx) => { - await rematerializeTransactionsAtCurrentVersion(wex, tx); - }); + if (idb) { + await this.finalizeIndexedDbOpen(idb, await idb.ensureOpen()); } }); } catch (e) { @@ -1279,6 +1278,55 @@ export class InternalWalletState { } } + private async finalizeIndexedDbOpen( + idb: IdbWalletDbHandle, + result: { fixupsApplied: number; schemaUpgraded: boolean }, + ): Promise<void> { + if (result.fixupsApplied === 0 && !result.schemaUpgraded) return; + const wex = getNormalWalletExecutionContext( + this, + CancellationToken.CONTINUE, + undefined, + { observe(evt: any) {} }, + ); + try { + if (result.fixupsApplied > 0) { + idb.emitNotification({ + type: NotificationType.DatabaseMaintenanceProgress, + operation: "indexeddb-fixup", + phase: "fixup", + step: WALLET_DB_REMATERIALIZE_STEP, + completedSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS - 1, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + }); + // Use the raw handle while holding exclusive admission; routing + // through admittedDb would try to acquire shared admission and + // deadlock behind ourselves. + await idb.runReadWriteTx(async (tx) => { + await rematerializeTransactionsAtCurrentVersion(wex, tx); + }); + } + idb.emitNotification({ + type: NotificationType.DatabaseMaintenanceProgress, + operation: "indexeddb-fixup", + phase: "complete", + completedSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + }); + } catch (e) { + idb.emitNotification({ + type: NotificationType.DatabaseMaintenanceProgress, + operation: "indexeddb-fixup", + phase: "failed", + step: WALLET_DB_REMATERIALIZE_STEP, + completedSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS - 1, + totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS, + error: getErrorDetailFromException(e), + }); + throw e; + } + } + /** Select native sqlite before an empty database is opened as IndexedDB. */ async openNativeDatabaseIfEmpty(): Promise<boolean> { if (this.loadingDb) { @@ -1435,7 +1483,9 @@ export class InternalWalletState { this.suspendedDbRelease = undefined; try { const idb = this.idbOnly; - if (idb) await idb.ensureOpen(); + if (idb) { + await this.finalizeIndexedDbOpen(idb, await idb.ensureOpen()); + } } finally { this.loadingDb = false; this.loadingDbCond.trigger(); @@ -1445,6 +1495,10 @@ export class InternalWalletState { notify(n: WalletNotification): void { logger.trace(`Notification: ${j2s(n)}`); + if (n.type === NotificationType.DatabaseMaintenanceProgress) { + this.maintenanceNotifications.offer(n); + return; + } for (const l of this.listeners) { const nc = JSON.parse(JSON.stringify(n)); setTimeout(() => { @@ -1453,6 +1507,19 @@ export class InternalWalletState { } } + private deliverNotificationSynchronously(n: WalletNotification): void { + for (const l of [...this.listeners]) { + const nc = JSON.parse(JSON.stringify(n)); + try { + l(nc); + } catch (e) { + logger.warn( + `ignoring exception from wallet notification listener: ${safeStringifyException(e)}`, + ); + } + } + } + addNotificationListener(f: (n: WalletNotification) => void): CancelFn { this.listeners.push(f); return () => { @@ -1469,6 +1536,7 @@ export class InternalWalletState { stop(): void { logger.trace("stopping (at internal wallet state)"); this.stopped = true; + this.maintenanceNotifications.stop(); this.timerGroup.stopCurrentAndFutureTimers(); this.cryptoDispatcher.stop(); this.taskScheduler.shutdown().catch((e) => {