commit 5a4c06c2dcda8b8cad3208cd47714bd466619c13
parent 8e7a3f7583c5cb259e96fa4dce8ebe3095dfaa38
Author: Florian Dold <dold@taler.net>
Date: Fri, 28 Aug 2026 17:07:00 +0200
wallet-core: cancel imports before database publication
Diffstat:
11 files changed, 187 insertions(+), 27 deletions(-)
diff --git a/packages/taler-wallet-core/src/db/handle.ts b/packages/taler-wallet-core/src/db/handle.ts
@@ -89,7 +89,11 @@ export interface WalletDbHandle {
*
* Throws if the dump did not come from this backend.
*/
- importDatabase(dump: any, finalize: WalletDbImportFinalizer): Promise<void>;
+ importDatabase(
+ dump: any,
+ finalize: WalletDbImportFinalizer,
+ options?: WalletDbMigrationOptions,
+ ): Promise<void>;
/**
* Replace the database from either backend's JSON dump.
diff --git a/packages/taler-wallet-core/src/db/indexeddb/dump.ts b/packages/taler-wallet-core/src/db/indexeddb/dump.ts
@@ -14,7 +14,7 @@ import {
structuredEncapsulate,
structuredRevive,
} from "@gnu-taler/idb-bridge";
-import { Logger } from "@gnu-taler/taler-util";
+import { CancellationToken, Logger } from "@gnu-taler/taler-util";
import { openDatabase } from "../query.js";
import {
TALER_WALLET_MAIN_DB_NAME,
@@ -177,31 +177,53 @@ export async function exportDb(idb: IDBFactory): Promise<DbDump> {
async function recoverFromDump(
db: IDBDatabase,
dbDump: DbDumpDatabase,
+ cancellationToken: CancellationToken,
): Promise<void> {
+ cancellationToken.throwIfCancelled();
const tx = db.transaction(Array.from(db.objectStoreNames), "readwrite");
const txProm = promiseFromTransaction(tx);
- const storeNames = db.objectStoreNames;
- for (let i = 0; i < storeNames.length; i++) {
- const name = db.objectStoreNames[i];
- const storeDump = dbDump.stores[name];
- await promiseFromRequest(tx.objectStore(name).clear());
- if (!storeDump) continue;
- logger.info(`importing ${storeDump.records.length} records into ${name}`);
- for (let rec of storeDump.records) {
- await promiseFromRequest(tx.objectStore(name).put(rec.value, rec.key));
- logger.trace("importing record done");
+ try {
+ const storeNames = db.objectStoreNames;
+ for (let i = 0; i < storeNames.length; i++) {
+ cancellationToken.throwIfCancelled();
+ const name = db.objectStoreNames[i];
+ const storeDump = dbDump.stores[name];
+ await promiseFromRequest(tx.objectStore(name).clear());
+ cancellationToken.throwIfCancelled();
+ if (!storeDump) continue;
+ logger.info(`importing ${storeDump.records.length} records into ${name}`);
+ for (let rec of storeDump.records) {
+ cancellationToken.throwIfCancelled();
+ await promiseFromRequest(tx.objectStore(name).put(rec.value, rec.key));
+ logger.trace("importing record done");
+ }
+ }
+ cancellationToken.throwIfCancelled();
+ tx.commit();
+ await txProm;
+ } catch (e) {
+ try {
+ tx.abort();
+ } catch {
+ // The transaction may already have aborted because of the import error.
}
+ await txProm.catch(() => {});
+ throw e;
}
- tx.commit();
- return await txProm;
}
function checkDbDump(x: any): x is DbDump {
return "databases" in x;
}
-export async function importDb(db: IDBDatabase, dumpJson: any): Promise<void> {
+export async function importDb(
+ db: IDBDatabase,
+ dumpJson: any,
+ cancellationToken: CancellationToken = CancellationToken.CONTINUE,
+): Promise<void> {
+ cancellationToken.throwIfCancelled();
const d = structuredRevive(dumpJson);
+ cancellationToken.throwIfCancelled();
if (checkDbDump(d)) {
const walletDb = d.databases[TALER_WALLET_MAIN_DB_NAME];
if (!walletDb) {
@@ -209,7 +231,7 @@ export async function importDb(db: IDBDatabase, dumpJson: any): Promise<void> {
`unable to import, main wallet database (${TALER_WALLET_MAIN_DB_NAME}) not found`,
);
}
- await recoverFromDump(db, walletDb);
+ await recoverFromDump(db, walletDb, cancellationToken);
} else {
throw Error("unable to import, doesn't look like a valid DB dump");
}
diff --git a/packages/taler-wallet-core/src/db/indexeddb/fixups.ts b/packages/taler-wallet-core/src/db/indexeddb/fixups.ts
@@ -21,6 +21,7 @@ import { GlobalIDB, IDBKeyRange } from "@gnu-taler/idb-bridge";
import {
AmountString,
Amounts,
+ CancellationToken,
canonicalJson,
checkDbInvariant,
getErrorDetailFromException,
@@ -837,8 +838,14 @@ async function fixup20260812ExchangeWithdrawValues(
export async function applyFixups(
db: DbAccess<typeof WalletIndexedDbStoresV1>,
onProgress: (notification: WalletNotification) => void = () => {},
- options: { deferCompletion?: boolean } = {},
+ options: {
+ deferCompletion?: boolean;
+ cancellationToken?: CancellationToken;
+ } = {},
): Promise<number> {
+ const cancellationToken =
+ options.cancellationToken ?? CancellationToken.CONTINUE;
+ cancellationToken.throwIfCancelled();
logger.trace("applying fixups");
let count = 0;
// Most opens have no work to do. Read the marker store once rather than
@@ -847,6 +854,7 @@ export async function applyFixups(
// 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++) {
+ cancellationToken.throwIfCancelled();
const fixupInstruction = walletDbFixups[index];
if (completedFixups?.has(fixupInstruction.name)) {
continue;
@@ -854,6 +862,7 @@ export async function applyFixups(
let applied = false;
try {
await db.runAllStoresReadWriteTx({}, async (tx) => {
+ cancellationToken.throwIfCancelled();
logger.trace(`checking fixup ${fixupInstruction.name}`);
if (!completedFixups) {
completedFixups = new Set(
@@ -877,6 +886,7 @@ export async function applyFixups(
totalSteps: WALLET_DB_MAINTENANCE_TOTAL_STEPS,
});
await fixupInstruction.fn(tx);
+ cancellationToken.throwIfCancelled();
// A fixup may change records behind materialized transactions or coin
// availability. Invalidate both durable versions in the same commit
// as the repair, so a crash or failed rematerialization is retried on
@@ -886,7 +896,9 @@ export async function applyFixups(
await tx.fixups.put({
fixupName: fixupInstruction.name,
});
+ cancellationToken.throwIfCancelled();
});
+ cancellationToken.throwIfCancelled();
completedFixups!.add(fixupInstruction.name);
} catch (e) {
if (applied) {
diff --git a/packages/taler-wallet-core/src/db/indexeddb/handle.ts b/packages/taler-wallet-core/src/db/indexeddb/handle.ts
@@ -269,7 +269,11 @@ export class IdbWalletDbHandle implements WalletDbHandle {
async importDatabase(
dump: any,
finalize: WalletDbImportFinalizer,
+ options: WalletDbMigrationOptions = {},
): Promise<void> {
+ const cancellationToken =
+ options.cancellationToken ?? CancellationToken.CONTINUE;
+ cancellationToken.throwIfCancelled();
// A native-backend dump has {schemaVersion, tables}; this backend's dumps
// have {databases}. Importing across backends is a format conversion,
// not a copy, and silently accepting the wrong shape would import
@@ -282,6 +286,7 @@ export class IdbWalletDbHandle implements WalletDbHandle {
);
}
await this.ensureOpen();
+ cancellationToken.throwIfCancelled();
if (!this.idbHandle) {
throw Error("wallet database is not open");
}
@@ -302,23 +307,29 @@ export class IdbWalletDbHandle implements WalletDbHandle {
}
});
try {
- await importDb(staged.handle, dump);
+ cancellationToken.throwIfCancelled();
+ await importDb(staged.handle, dump, cancellationToken);
+ cancellationToken.throwIfCancelled();
// The imported records may predate any of the fixups, whatever the old
// generation had applied. Clear the imported log and repair the staged
// generation before it can become authoritative.
await stagedAccess.runAllStoresReadWriteTx({}, async (tx) => {
const fixups = await tx.fixups.getAll();
for (const fx of fixups) {
+ cancellationToken.throwIfCancelled();
await tx.fixups.delete(fx.fixupName);
}
+ cancellationToken.throwIfCancelled();
});
await this.applyDbFixups(
stagedAccess,
(n) => stagedNotifications.push(n),
- { deferCompletion: true },
+ { deferCompletion: true, cancellationToken },
);
await stagedAccess.runAllStoresReadWriteTx({}, async (tx) => {
+ cancellationToken.throwIfCancelled();
await finalize(new IdbWalletTransaction(tx));
+ cancellationToken.throwIfCancelled();
});
stagedNotifications.push({
type: NotificationType.DatabaseMaintenanceProgress,
@@ -330,6 +341,7 @@ export class IdbWalletDbHandle implements WalletDbHandle {
// This metadata transaction is the commit point. A crash before it
// keeps oldName authoritative; a crash afterwards opens staged.name.
+ cancellationToken.throwIfCancelled();
await publishTalerDatabaseReplacement(
this.idbFactory,
oldName,
diff --git a/packages/taler-wallet-core/src/db/migration/import.ts b/packages/taler-wallet-core/src/db/migration/import.ts
@@ -77,7 +77,7 @@ export async function importWalletDbDump(
}
if (sourceBackend === target.name) {
options.cancellationToken?.throwIfCancelled();
- await target.importDatabase(dump, finalize);
+ await target.importDatabase(dump, finalize, options);
return;
}
if (target.name !== "indexeddb" && target.name !== "sqlite") {
@@ -104,7 +104,7 @@ export async function importWalletDbDump(
);
options.cancellationToken?.throwIfCancelled();
- await source.importDatabase(dump, async () => {});
+ await source.importDatabase(dump, async () => {}, options);
options.cancellationToken?.throwIfCancelled();
const report = await convertWalletDb(source, destination, {
@@ -120,7 +120,7 @@ export async function importWalletDbDump(
const convertedDump = await destination.exportDatabase();
options.cancellationToken?.throwIfCancelled();
- await target.importDatabase(convertedDump, finalize);
+ await target.importDatabase(convertedDump, finalize, options);
target.emitNotification({
type: NotificationType.DatabaseMaintenanceProgress,
diff --git a/packages/taler-wallet-core/src/db/migration/native.test.ts b/packages/taler-wallet-core/src/db/migration/native.test.ts
@@ -38,6 +38,7 @@ import {
} from "@gnu-taler/idb-bridge";
import { createNodeHelperSqlite3Impl } from "@gnu-taler/idb-bridge/node-helper-sqlite3-impl";
import {
+ CancellationToken,
DatabaseMaintenanceProgressNotification,
NotificationType,
TalerError,
@@ -430,6 +431,51 @@ test("native migration: an interrupted attempt restarts after reopening", async
}
});
+test("native migration: cancellation after a copied batch leaves IndexedDB authoritative", async () => {
+ const source = await makeMinimalIdbDb();
+ const expectedTombstones = DB_CONVERSION_PROGRESS_RECORDS * 2 + 17;
+ const cancellation = CancellationToken.create();
+ try {
+ await source.handle.runReadWriteTx(async (tx) => {
+ for (let i = 0; i < expectedTombstones; i++) {
+ await tx.upsertTombstone({ id: `cancel-${i}` });
+ }
+ });
+
+ let cancelledDuringCopy = false;
+ await assert.rejects(
+ () =>
+ migrateWalletDbToNative(source.db, source.handle, {
+ cancellationToken: cancellation.token,
+ onProgress(notification) {
+ if (
+ !cancelledDuringCopy &&
+ notification.phase === "copy" &&
+ (notification.processedRecords ?? 0) >=
+ DB_CONVERSION_PROGRESS_RECORDS
+ ) {
+ cancelledDuringCopy = true;
+ cancellation.cancel("cancel after a committed copy batch");
+ }
+ },
+ }),
+ CancellationToken.CancellationError,
+ );
+ assert.ok(cancelledDuringCopy, "migration never reached a copied batch");
+ const inspection = await inspectWalletDbFileDetails(source.db);
+ assert.strictEqual(inspection.kind, "indexeddb");
+ assert.ok(inspection.nativeRecords > 0, "no partial native copy was left");
+ assert.strictEqual(
+ (await readNativeMigrationInfo(source.db))?.status,
+ "running",
+ );
+ } finally {
+ cancellation.dispose();
+ await source.handle.close().catch(() => {});
+ await source.db.close().catch(() => {});
+ }
+});
+
test("native migration: a legacy interrupted attempt remains restartable", async () => {
const directory = fs.mkdtempSync(
path.join(os.tmpdir(), "wallet-db-migration-legacy-"),
diff --git a/packages/taler-wallet-core/src/db/sqlite/database.ts b/packages/taler-wallet-core/src/db/sqlite/database.ts
@@ -21,6 +21,7 @@ import {
Sqlite3Value,
} from "@gnu-taler/idb-bridge";
import {
+ CancellationToken,
decodeCrock,
encodeCrock,
Logger,
@@ -562,7 +563,9 @@ export async function importNativeSqliteDb(
dump: NativeSqliteDbDump,
finalize: (tx: WalletDbTransaction) => Promise<void>,
notifyFn: (n: WalletNotification) => void,
+ cancellationToken: CancellationToken = CancellationToken.CONTINUE,
): Promise<void> {
+ cancellationToken.throwIfCancelled();
if (dump.schemaVersion !== SQLITE_SCHEMA_VERSION) {
throw Error(
`cannot import a native wallet DB dump of schema version` +
@@ -570,19 +573,23 @@ export async function importNativeSqliteDb(
);
}
await ndb.lock.run(async () => {
+ cancellationToken.throwIfCancelled();
const tables = await listDataTables(ndb);
await runNativeSqliteWalletTxLocked(ndb, notifyFn, async (tx) => {
// Clear and refill in one transaction: a partial import would leave
// the wallet with a mix of two databases.
for (const table of tables) {
+ cancellationToken.throwIfCancelled();
await (await ndb.db.prepare(`DELETE FROM "${table}"`)).run({});
}
for (const table of tables) {
+ cancellationToken.throwIfCancelled();
const rows = dump.tables[table];
if (!rows || rows.length === 0) {
continue;
}
for (const row of rows) {
+ cancellationToken.throwIfCancelled();
const cols = Object.keys(row);
const params: Record<string, Sqlite3Value> = {};
for (const c of cols) {
@@ -601,7 +608,9 @@ export async function importNativeSqliteDb(
// Derived wallet state is part of the restore. Running this before the
// same COMMIT means an error cannot expose imported records with stale
// or absent materialized transactions.
+ cancellationToken.throwIfCancelled();
await finalize(tx);
+ cancellationToken.throwIfCancelled();
});
});
}
diff --git a/packages/taler-wallet-core/src/db/sqlite/handle.ts b/packages/taler-wallet-core/src/db/sqlite/handle.ts
@@ -19,6 +19,7 @@ import {
WalletDbAccessStats,
WalletDbHandle,
WalletDbImportFinalizer,
+ WalletDbMigrationOptions,
} from "../handle.js";
import { WalletDbTransaction } from "../transaction.js";
import {
@@ -104,6 +105,7 @@ export class SqliteWalletDbHandle implements WalletDbHandle {
async importDatabase(
dump: any,
finalize: WalletDbImportFinalizer,
+ options: WalletDbMigrationOptions = {},
): Promise<void> {
if (dump != null && typeof dump === "object" && "databases" in dump) {
throw Error(
@@ -111,8 +113,12 @@ export class SqliteWalletDbHandle implements WalletDbHandle {
" into the native sqlite backend; convert the database instead",
);
}
- await importNativeSqliteDb(this.ndb, dump, finalize, (n) =>
- this.emitNotification(n),
+ await importNativeSqliteDb(
+ this.ndb,
+ dump,
+ finalize,
+ (n) => this.emitNotification(n),
+ options.cancellationToken,
);
}
diff --git a/packages/taler-wallet-core/src/db/testing/conformance.test.ts b/packages/taler-wallet-core/src/db/testing/conformance.test.ts
@@ -153,4 +153,48 @@ for (const makeRunner of runnerFactories) {
await target.close();
}
});
+
+ test(`dbtx ${makeRunner.name}: cancelled import is not published`, async () => {
+ const source = await makeRunner();
+ const target = await makeRunner();
+ const cancellation = CancellationToken.create();
+ try {
+ await source.runReadWriteTx((tx) =>
+ tx.upsertConfig({
+ key: ConfigRecordKey.TestLoopTx,
+ value: 2,
+ }),
+ );
+ const dump = await source.exportDatabase();
+ await target.runReadWriteTx((tx) =>
+ tx.upsertConfig({
+ key: ConfigRecordKey.TestLoopTx,
+ value: 1,
+ }),
+ );
+
+ await assert.rejects(
+ target.importDatabase(
+ dump,
+ async (tx) => {
+ await tx.upsertConfig({
+ key: ConfigRecordKey.TestLoopTx,
+ value: 3,
+ });
+ cancellation.cancel("cancel during finalization");
+ },
+ { cancellationToken: cancellation.token },
+ ),
+ CancellationToken.CancellationError,
+ );
+ const afterCancellation = await target.runReadWriteTx((tx) =>
+ tx.getConfig(ConfigRecordKey.TestLoopTx),
+ );
+ assert.strictEqual(afterCancellation?.value, 1);
+ } finally {
+ cancellation.dispose();
+ await source.close();
+ await target.close();
+ }
+ });
}
diff --git a/packages/taler-wallet-core/src/requests.ts b/packages/taler-wallet-core/src/requests.ts
@@ -1955,7 +1955,10 @@ async function importDbDump(
`the ${db.name} backend cannot import a ${dumpBackend} dump`,
);
}
- await db.importDatabase(dump, finalize);
+ await db.importDatabase(dump, finalize, {
+ cancellationToken: wex.cancellationToken,
+ progressToken,
+ });
}
} catch (e) {
if (e instanceof CancellationToken.CancellationError) {
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
@@ -278,9 +278,11 @@ export class AdmittedWalletDbHandle implements WalletDbHandle {
importDatabase(
dump: any,
finalize: Parameters<WalletDbHandle["importDatabase"]>[1],
+ options?: Parameters<WalletDbHandle["importDatabase"]>[2],
): Promise<void> {
- return this.gate.runExclusive(() =>
- this.current().importDatabase(dump, finalize),
+ return this.gate.runExclusive(
+ () => this.current().importDatabase(dump, finalize, options),
+ options?.cancellationToken,
);
}
clearDatabase(): Promise<void> {
