commit d254fbd2a8ea1f4274b285a506bac68c45ed2418
parent bc383614f032c1f7cde0016ad0d858d04349b164
Author: Florian Dold <dold@taler.net>
Date: Thu, 20 Aug 2026 19:06:50 +0200
wallet-core: make database restore crash-atomic
Diffstat:
11 files changed, 679 insertions(+), 117 deletions(-)
diff --git a/packages/taler-wallet-core/src/db-indexeddb.test.ts b/packages/taler-wallet-core/src/db-indexeddb.test.ts
@@ -20,9 +20,12 @@ import assert from "node:assert";
import { test } from "node:test";
import { ConfigRecordKey } from "./db-common.js";
import {
+ abortTalerDatabaseReplacement,
applyFixups,
+ beginTalerDatabaseReplacement,
exportSingleDb,
openTalerDatabase,
+ publishTalerDatabaseReplacement,
TALER_WALLET_MAIN_DB_NAME,
} from "./db-indexeddb.js";
import { IdbWalletDbHandle } from "./dbtx-handle-impl.js";
@@ -44,7 +47,7 @@ test("indexeddb import clears stores absent from an older dump", async () => {
petname: "Alice",
}),
);
- await handle.importDatabase(dump);
+ await handle.importDatabase(dump, async () => {});
const contacts = await handle.runReadWriteTx((tx) => tx.listContacts());
assert.deepStrictEqual(contacts, []);
@@ -110,6 +113,120 @@ test("a failed fixup is retried on the next database open", async () => {
}
});
+test("indexeddb import commit survives reopen and failure preserves old generation", async () => {
+ const sqlite3Impl = await createNodeHelperSqlite3Impl({
+ enableTracing: false,
+ });
+ const backend = await createSqliteBackend(sqlite3Impl, {
+ filename: ":memory:",
+ });
+ BridgeIDBFactory.enableTracing = false;
+ const factory = new BridgeIDBFactory(backend);
+ let handle = new IdbWalletDbHandle(factory);
+ const source = await makeIdbRunner();
+ const restored = await makeIdbRunner();
+ try {
+ await handle.ensureOpen();
+ await handle.runReadWriteTx((tx) =>
+ tx.upsertConfig({ key: ConfigRecordKey.TestLoopTx, value: 10 }),
+ );
+ await source.runReadWriteTx((tx) =>
+ tx.upsertConfig({ key: ConfigRecordKey.TestLoopTx, value: 20 }),
+ );
+ const dump = await source.exportDatabase();
+
+ await assert.rejects(
+ handle.importDatabase(dump, async () => {
+ throw Error("injected rematerialization failure");
+ }),
+ /injected rematerialization failure/,
+ );
+ await handle.close();
+ handle = new IdbWalletDbHandle(factory);
+ await handle.ensureOpen();
+ assert.strictEqual(
+ (
+ await handle.runReadWriteTx((tx) =>
+ tx.getConfig(ConfigRecordKey.TestLoopTx),
+ )
+ )?.value,
+ 10,
+ );
+
+ await handle.importDatabase(dump, async (tx) => {
+ await tx.upsertConfig({ key: ConfigRecordKey.TestLoopTx, value: 30 });
+ });
+ await handle.close();
+ handle = new IdbWalletDbHandle(factory);
+ await handle.ensureOpen();
+ assert.strictEqual(
+ (
+ await handle.runReadWriteTx((tx) =>
+ tx.getConfig(ConfigRecordKey.TestLoopTx),
+ )
+ )?.value,
+ 30,
+ );
+ const portableDump = await handle.exportDatabase();
+ assert.ok(portableDump.databases[TALER_WALLET_MAIN_DB_NAME]);
+ await restored.importDatabase(portableDump, async () => {});
+ assert.strictEqual(
+ (
+ await restored.runReadWriteTx((tx) =>
+ tx.getConfig(ConfigRecordKey.TestLoopTx),
+ )
+ )?.value,
+ 30,
+ );
+ } finally {
+ await source.close();
+ await restored.close();
+ await handle.close();
+ }
+});
+
+test("indexeddb startup selects only a published generation", async () => {
+ const sqlite3Impl = await createNodeHelperSqlite3Impl({
+ enableTracing: false,
+ });
+ const backend = await createSqliteBackend(sqlite3Impl, {
+ filename: ":memory:",
+ });
+ BridgeIDBFactory.enableTracing = false;
+ const factory = new BridgeIDBFactory(backend);
+
+ const original = await openTalerDatabase(factory, async () => {});
+ const abandoned = await beginTalerDatabaseReplacement(
+ factory,
+ original.name,
+ async () => {},
+ );
+ abandoned.handle.close();
+ original.close();
+
+ const afterAbandon = await openTalerDatabase(factory, async () => {});
+ assert.strictEqual(afterAbandon.name, TALER_WALLET_MAIN_DB_NAME);
+ const published = await beginTalerDatabaseReplacement(
+ factory,
+ afterAbandon.name,
+ async () => {},
+ );
+ await publishTalerDatabaseReplacement(
+ factory,
+ afterAbandon.name,
+ published.name,
+ );
+ published.handle.close();
+ afterAbandon.close();
+
+ const afterPublish = await openTalerDatabase(factory, async () => {});
+ assert.strictEqual(afterPublish.name, published.name);
+ afterPublish.close();
+ // The abandoned generation is deliberately not deleted during startup,
+ // but its stale pending claim was cleared, so its owner can clean it safely.
+ await abortTalerDatabaseReplacement(factory, abandoned.name);
+});
+
test("export closes its database connection", async () => {
const sqlite3Impl = await createNodeHelperSqlite3Impl({
enableTracing: false,
diff --git a/packages/taler-wallet-core/src/db-indexeddb.ts b/packages/taler-wallet-core/src/db-indexeddb.ts
@@ -81,6 +81,7 @@ import {
codecForAny,
encodeCrock,
getErrorDetailFromException,
+ getRandomBytes,
hash,
j2s,
stringToBytes,
@@ -88,6 +89,7 @@ import {
import { DbRetryInfo, TaskIdentifiers } from "./common.js";
import {
ConfigRecord,
+ ConfigRecordKey,
DbPreciseTimestamp,
DbProtocolTimestamp,
DenomLossStatus,
@@ -250,6 +252,15 @@ export const TALER_WALLET_META_DB_NAME = "taler-wallet-meta";
*/
export const CURRENT_DB_CONFIG_KEY = "currentMainDbName";
+/** Database generation being prepared by an import but not authoritative yet. */
+const PENDING_DB_CONFIG_KEY = "pendingMainDbName";
+
+/** Previous authoritative generation waiting for best-effort deletion. */
+const RETIRED_DB_CONFIG_KEY = "retiredMainDbName";
+
+/** Names below this prefix are current-schema generations, not major versions. */
+const TALER_WALLET_DB_GENERATION_PREFIX = `${TALER_WALLET_MAIN_DB_NAME}-generation-`;
+
/**
* Current database minor version, should be incremented
* each time we do minor schema changes on the database.
@@ -1495,13 +1506,18 @@ export async function exportDb(idb: IDBFactory): Promise<DbDump> {
databases: {},
};
+ const currentMainDbName = await readCurrentMainDbName(idb);
+
dbDump.databases[TALER_WALLET_META_DB_NAME] = await exportSingleDb(
idb,
TALER_WALLET_META_DB_NAME,
);
+ // A dump is portable, so expose the active generation under the canonical
+ // logical name. The generation name is local crash-recovery bookkeeping
+ // and must not become part of the backup format.
dbDump.databases[TALER_WALLET_MAIN_DB_NAME] = await exportSingleDb(
idb,
- TALER_WALLET_MAIN_DB_NAME,
+ currentMainDbName,
);
return dbDump;
@@ -2300,6 +2316,11 @@ export async function applyFixups(
totalSteps: walletDbFixups.length,
});
await fixupInstruction.fn(tx);
+ // A fixup may change any operation record from which transactionsMeta
+ // is derived. Invalidate the durable view version in the same commit
+ // as the repair, so a crash or failed rematerialization is retried on
+ // the next initialization instead of leaving a stale "current" flag.
+ await tx.config.delete(ConfigRecordKey.MaterializedTransactionsVersion);
await tx.fixups.put({
fixupName: fixupInstruction.name,
});
@@ -2536,78 +2557,317 @@ export async function openTalerDatabase(
idbFactory: IDBFactory,
onVersionChange: () => void,
): Promise<IDBDatabase> {
- const metaDbHandle = await openDatabase(
+ const state = await readMainDbState(idbFactory, true);
+ await cleanInterruptedDatabaseReplacement(idbFactory, state);
+ return await openTalerDatabaseGeneration(
+ idbFactory,
+ state.current,
+ onVersionChange,
+ );
+}
+
+interface MainDbState {
+ current: string;
+ pending?: string;
+ retired?: string;
+}
+
+function isCurrentGenerationName(name: string): boolean {
+ return (
+ name === TALER_WALLET_MAIN_DB_NAME ||
+ name.startsWith(TALER_WALLET_DB_GENERATION_PREFIX)
+ );
+}
+
+async function openMetaDatabase(idbFactory: IDBFactory): Promise<{
+ handle: IDBDatabase;
+ access: DbAccess<typeof walletMetadataStore>;
+}> {
+ const handle = await openDatabase(
idbFactory,
TALER_WALLET_META_DB_NAME,
1,
() => {},
onMetaDbUpgradeNeeded,
);
+ return {
+ handle,
+ access: new DbAccessImpl(
+ handle,
+ walletMetadataStore,
+ CancellationToken.CONTINUE,
+ ),
+ };
+}
- const metaDb = new DbAccessImpl(
- metaDbHandle,
- walletMetadataStore,
- CancellationToken.CONTINUE,
- );
- let currentMainVersion: string | undefined;
- await metaDb.runAllStoresReadWriteTx({}, async (tx) => {
- const dbVersionRecord = await tx.metaConfig.get(CURRENT_DB_CONFIG_KEY);
- if (!dbVersionRecord) {
- currentMainVersion = TALER_WALLET_MAIN_DB_NAME;
- await tx.metaConfig.put({
- key: CURRENT_DB_CONFIG_KEY,
- value: TALER_WALLET_MAIN_DB_NAME,
- });
- } else {
- currentMainVersion = dbVersionRecord.value;
- }
- });
-
- if (currentMainVersion !== TALER_WALLET_MAIN_DB_NAME) {
- switch (currentMainVersion) {
- case "taler-wallet-main-v2":
- case "taler-wallet-main-v3":
- case "taler-wallet-main-v4": // temporary, we might migrate v4 later
- case "taler-wallet-main-v5":
- case "taler-wallet-main-v6":
- case "taler-wallet-main-v7":
- case "taler-wallet-main-v8":
- case "taler-wallet-main-v9":
- // We consider this a pre-release
- // development version, no migration is done.
- await metaDb.runAllStoresReadWriteTx({}, async (tx) => {
+async function readMainDbState(
+ idbFactory: IDBFactory,
+ initialize: boolean,
+): Promise<MainDbState> {
+ const meta = await openMetaDatabase(idbFactory);
+ try {
+ let state!: MainDbState;
+ await meta.access.runAllStoresReadWriteTx({}, async (tx) => {
+ const currentRecord = await tx.metaConfig.get(CURRENT_DB_CONFIG_KEY);
+ let current = currentRecord?.value as string | undefined;
+ if (!current) {
+ current = TALER_WALLET_MAIN_DB_NAME;
+ if (initialize) {
await tx.metaConfig.put({
key: CURRENT_DB_CONFIG_KEY,
- value: TALER_WALLET_MAIN_DB_NAME,
+ value: current,
});
- });
- break;
- default:
- throw Error(
- `major migration from database major=${currentMainVersion} not supported`,
- );
- }
+ }
+ } else if (!isCurrentGenerationName(current)) {
+ switch (current) {
+ case "taler-wallet-main-v2":
+ case "taler-wallet-main-v3":
+ case "taler-wallet-main-v4":
+ case "taler-wallet-main-v5":
+ case "taler-wallet-main-v6":
+ case "taler-wallet-main-v7":
+ case "taler-wallet-main-v8":
+ case "taler-wallet-main-v9":
+ // These were pre-release databases and have no supported major
+ // migration. Preserve the historical behaviour of starting the
+ // current major afresh.
+ current = TALER_WALLET_MAIN_DB_NAME;
+ if (initialize) {
+ await tx.metaConfig.put({
+ key: CURRENT_DB_CONFIG_KEY,
+ value: current,
+ });
+ }
+ break;
+ default:
+ throw Error(
+ `major migration from database major=${current} not supported`,
+ );
+ }
+ }
+ state = {
+ current,
+ pending: (await tx.metaConfig.get(PENDING_DB_CONFIG_KEY))?.value,
+ retired: (await tx.metaConfig.get(RETIRED_DB_CONFIG_KEY))?.value,
+ };
+ });
+ return state;
+ } finally {
+ meta.handle.close();
}
+}
+
+async function readCurrentMainDbName(idbFactory: IDBFactory): Promise<string> {
+ return (await readMainDbState(idbFactory, true)).current;
+}
- const mainDbHandle = await openDatabase(
+async function openTalerDatabaseGeneration(
+ idbFactory: IDBFactory,
+ name: string,
+ onVersionChange: () => void,
+): Promise<IDBDatabase> {
+ if (!isCurrentGenerationName(name)) {
+ throw Error(`invalid wallet database generation name ${name}`);
+ }
+ return await openDatabase(
idbFactory,
- TALER_WALLET_MAIN_DB_NAME,
+ name,
WALLET_DB_MINOR_VERSION,
onVersionChange,
onTalerDbUpgradeNeeded,
);
+}
- return mainDbHandle;
+/** Resolve true on deletion, false when another client still blocks it. */
+async function tryDeleteDatabase(
+ idbFactory: IDBFactory,
+ name: string,
+): Promise<boolean> {
+ return await new Promise<boolean>((resolve, reject) => {
+ const req = idbFactory.deleteDatabase(name);
+ let settled = false;
+ const finish = (result: boolean): void => {
+ if (settled) return;
+ settled = true;
+ resolve(result);
+ };
+ req.onerror = () => {
+ if (settled) return;
+ settled = true;
+ reject(req.error);
+ };
+ req.onblocked = () => finish(false);
+ req.onsuccess = () => finish(true);
+ });
+}
+
+async function clearMetaMarker(
+ idbFactory: IDBFactory,
+ key: string,
+ expectedValue: string,
+): Promise<void> {
+ const meta = await openMetaDatabase(idbFactory);
+ try {
+ await meta.access.runAllStoresReadWriteTx({}, async (tx) => {
+ const record = await tx.metaConfig.get(key);
+ if (record?.value === expectedValue) {
+ await tx.metaConfig.delete(key);
+ }
+ });
+ } finally {
+ meta.handle.close();
+ }
+}
+
+async function cleanupGeneration(
+ idbFactory: IDBFactory,
+ name: string | undefined,
+ marker: string,
+ current: string,
+): Promise<void> {
+ if (!name || name === current) return;
+ try {
+ if (await tryDeleteDatabase(idbFactory, name)) {
+ await clearMetaMarker(idbFactory, marker, name);
+ }
+ } catch (e) {
+ // Cleanup is not authoritative-state recovery. Retaining an unreachable
+ // generation costs storage, but refusing to open the current wallet would
+ // turn that harmless residue into an outage.
+ logger.warn(`could not clean wallet database generation ${name}: ${e}`);
+ }
+}
+
+async function cleanInterruptedDatabaseReplacement(
+ idbFactory: IDBFactory,
+ state: MainDbState,
+): Promise<void> {
+ // A pending generation was never published, so current is authoritative.
+ // Clear its claim but do not issue deleteDatabase here: another wallet
+ // context could still be preparing it, and a blocked deletion request would
+ // remain armed and could delete the generation after that context publishes
+ // and eventually closes it. Clearing the claim instead makes that
+ // publisher fail its compare-and-swap safely. Hard-crash residue is an
+ // unreachable storage leak, never an authoritative-state ambiguity.
+ if (state.pending && state.pending !== state.current) {
+ await clearMetaMarker(idbFactory, PENDING_DB_CONFIG_KEY, state.pending);
+ }
+ // A retired generation has already been superseded and can never become
+ // authoritative again, so deletion is safe even when another context still
+ // has it open (in which case cleanup remains recorded for the next start).
+ await cleanupGeneration(
+ idbFactory,
+ state.retired,
+ RETIRED_DB_CONFIG_KEY,
+ state.current,
+ );
+}
+
+export interface StagedTalerDatabase {
+ name: string;
+ handle: IDBDatabase;
+}
+
+/** Create and durably record an unpublished current-schema generation. */
+export async function beginTalerDatabaseReplacement(
+ idbFactory: IDBFactory,
+ currentName: string,
+ onVersionChange: () => void,
+): Promise<StagedTalerDatabase> {
+ const name = `${TALER_WALLET_DB_GENERATION_PREFIX}${encodeCrock(
+ getRandomBytes(16),
+ )}`;
+ const meta = await openMetaDatabase(idbFactory);
+ try {
+ await meta.access.runAllStoresReadWriteTx({}, async (tx) => {
+ const current = await tx.metaConfig.get(CURRENT_DB_CONFIG_KEY);
+ if (current?.value !== currentName) {
+ throw Error("wallet database generation changed during import");
+ }
+ const pending = await tx.metaConfig.get(PENDING_DB_CONFIG_KEY);
+ if (pending) {
+ throw Error(
+ `another wallet database import is already pending (${pending.value})`,
+ );
+ }
+ await tx.metaConfig.put({ key: PENDING_DB_CONFIG_KEY, value: name });
+ });
+ } finally {
+ meta.handle.close();
+ }
+ try {
+ return {
+ name,
+ handle: await openTalerDatabaseGeneration(
+ idbFactory,
+ name,
+ onVersionChange,
+ ),
+ };
+ } catch (e) {
+ await abortTalerDatabaseReplacement(idbFactory, name);
+ throw e;
+ }
+}
+
+/** Atomically make a fully prepared generation authoritative. */
+export async function publishTalerDatabaseReplacement(
+ idbFactory: IDBFactory,
+ oldName: string,
+ newName: string,
+): Promise<void> {
+ const meta = await openMetaDatabase(idbFactory);
+ try {
+ await meta.access.runAllStoresReadWriteTx({}, async (tx) => {
+ const current = await tx.metaConfig.get(CURRENT_DB_CONFIG_KEY);
+ const pending = await tx.metaConfig.get(PENDING_DB_CONFIG_KEY);
+ if (current?.value !== oldName || pending?.value !== newName) {
+ throw Error("wallet database generation changed during import");
+ }
+ await tx.metaConfig.put({ key: CURRENT_DB_CONFIG_KEY, value: newName });
+ await tx.metaConfig.delete(PENDING_DB_CONFIG_KEY);
+ await tx.metaConfig.put({ key: RETIRED_DB_CONFIG_KEY, value: oldName });
+ });
+ } finally {
+ meta.handle.close();
+ }
+}
+
+/** Delete an unpublished generation after a failed import. */
+export async function abortTalerDatabaseReplacement(
+ idbFactory: IDBFactory,
+ name: string,
+): Promise<void> {
+ await cleanupGeneration(idbFactory, name, PENDING_DB_CONFIG_KEY, "");
+}
+
+/** Best-effort deletion of the old generation after a successful publish. */
+export async function retireTalerDatabaseGeneration(
+ idbFactory: IDBFactory,
+ name: string,
+ currentName: string,
+): Promise<void> {
+ await cleanupGeneration(idbFactory, name, RETIRED_DB_CONFIG_KEY, currentName);
}
export async function deleteTalerDatabase(
idbFactory: IDBFactory,
): Promise<void> {
- return new Promise((resolve, reject) => {
- const req = idbFactory.deleteDatabase(TALER_WALLET_MAIN_DB_NAME);
- req.onerror = () => reject(req.error);
- req.onsuccess = () => resolve();
- });
+ const state = await readMainDbState(idbFactory, false);
+ const names = new Set([
+ TALER_WALLET_MAIN_DB_NAME,
+ state.current,
+ state.pending,
+ state.retired,
+ ]);
+ for (const name of names) {
+ if (name && !(await tryDeleteDatabase(idbFactory, name))) {
+ throw Error(`deletion of wallet database ${name} is blocked`);
+ }
+ }
+ if (!(await tryDeleteDatabase(idbFactory, TALER_WALLET_META_DB_NAME))) {
+ throw Error("deletion of wallet metadata database is blocked");
+ }
}
/**
diff --git a/packages/taler-wallet-core/src/db-native-migration.test.ts b/packages/taler-wallet-core/src/db-native-migration.test.ts
@@ -232,6 +232,23 @@ test("native migration: happens in the same file and switches it over", async ()
await native.close();
});
+test("native migration recognizes a restored IndexedDB generation", async () => {
+ const { db, handle } = await makeMinimalIdbDb();
+ const dump = await handle.exportDatabase();
+ await handle.importDatabase(dump, async (tx) => {
+ await tx.upsertConfig({ key: "fault-test" as any, value: 2 });
+ });
+
+ assert.strictEqual(await inspectWalletDbFile(db), "indexeddb");
+ const { handle: native } = await migrateWalletDbToNative(db, handle);
+ assert.strictEqual(
+ (await native.runReadWriteTx((tx) => tx.getConfig("fault-test" as any)))
+ ?.value,
+ 2,
+ );
+ await native.close();
+});
+
test("native migration: clearing the wallet leaves the retained backup", async () => {
const { db, handle } = await makePopulatedIdbDb();
const { handle: native } = await migrateWalletDbToNative(db, handle);
diff --git a/packages/taler-wallet-core/src/db-native-migration.ts b/packages/taler-wallet-core/src/db-native-migration.ts
@@ -209,7 +209,8 @@ async function countMainIndexedDbRecords(db: Sqlite3Database): Promise<number> {
await db.prepare(
"SELECT COUNT(*) AS n FROM object_data od" +
" JOIN object_stores os ON os.id = od.object_store_id" +
- " WHERE os.database_name = 'taler-wallet-main-v10'",
+ " WHERE os.database_name = 'taler-wallet-main-v10'" +
+ " OR os.database_name LIKE 'taler-wallet-main-v10-generation-%'",
)
).getFirst({});
return Number(row?.n ?? 0);
diff --git a/packages/taler-wallet-core/src/dbtx-handle-impl.ts b/packages/taler-wallet-core/src/dbtx-handle-impl.ts
@@ -33,14 +33,22 @@ import {
} from "@gnu-taler/idb-bridge";
import {
+ abortTalerDatabaseReplacement,
applyFixups,
+ beginTalerDatabaseReplacement,
clearDatabase,
exportDb,
importDb,
openTalerDatabase,
+ publishTalerDatabaseReplacement,
+ retireTalerDatabaseGeneration,
WalletIndexedDbStoresV1,
} from "./db-indexeddb.js";
-import { WalletDbAccessStats, WalletDbHandle } from "./dbtx-handle.js";
+import {
+ WalletDbAccessStats,
+ WalletDbHandle,
+ WalletDbImportFinalizer,
+} from "./dbtx-handle.js";
import { IdbWalletTransaction } from "./dbtx-indexeddb.js";
import {
clearNativeSqliteWalletDb,
@@ -164,6 +172,8 @@ export class IdbWalletDbHandle implements WalletDbHandle {
private makeAccess(
idbHandle: IDBDatabase,
+ notificationSink: (n: WalletNotification) => void = (n) =>
+ this.emitNotification(n),
): DbAccess<typeof WalletIndexedDbStoresV1> {
return new DbAccessImpl(
idbHandle,
@@ -171,7 +181,7 @@ export class IdbWalletDbHandle implements WalletDbHandle {
CancellationToken.CONTINUE,
(notifs: WalletNotification[]) => {
for (const n of notifs) {
- this.emitNotification(n);
+ notificationSink(n);
}
},
);
@@ -211,7 +221,10 @@ export class IdbWalletDbHandle implements WalletDbHandle {
return await exportDb(this.idbFactory);
}
- async importDatabase(dump: any): Promise<void> {
+ async importDatabase(
+ dump: any,
+ finalize: WalletDbImportFinalizer,
+ ): Promise<void> {
// 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
@@ -227,21 +240,59 @@ export class IdbWalletDbHandle implements WalletDbHandle {
if (!this.idbHandle) {
throw Error("wallet database is not open");
}
- await importDb(this.idbHandle, dump);
- // The imported records may predate any of the fixups, whatever this
- // database had applied before. Clearing the log alone only schedules the
- // repairs for the next open, which for a running wallet is never, so they
- // are re-run here: a backup from before the status-enum digit fix imported
- // cleanly and then crashed getTransactions on a value no enum member
- // matched. Fixups are idempotent, so re-running them is safe.
- const access = await this.rawAccess();
- await access.runAllStoresReadWriteTx({}, async (tx) => {
- const fixups = await tx.fixups.getAll();
- for (const fx of fixups) {
- await tx.fixups.delete(fx.fixupName);
+ const oldHandle = this.idbHandle;
+ const oldName = oldHandle.name;
+ const stagedNotifications: WalletNotification[] = [];
+ const staged = await beginTalerDatabaseReplacement(
+ this.idbFactory,
+ oldName,
+ async () => {},
+ );
+ const stagedAccess = this.makeAccess(staged.handle, (n) =>
+ stagedNotifications.push(n),
+ );
+ let published = false;
+ try {
+ await importDb(staged.handle, dump);
+ // 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) {
+ await tx.fixups.delete(fx.fixupName);
+ }
+ });
+ await this.applyDbFixups(stagedAccess, (n) =>
+ stagedNotifications.push(n),
+ );
+ await stagedAccess.runAllStoresReadWriteTx({}, async (tx) => {
+ await finalize(new IdbWalletTransaction(tx));
+ });
+
+ // This metadata transaction is the commit point. A crash before it
+ // keeps oldName authoritative; a crash afterwards opens staged.name.
+ await publishTalerDatabaseReplacement(
+ this.idbFactory,
+ oldName,
+ staged.name,
+ );
+ published = true;
+ this.idbHandle = staged.handle;
+ this.dbAccess = stagedAccess;
+ oldHandle.close();
+ for (const n of stagedNotifications) this.emitNotification(n);
+ } catch (e) {
+ if (!published) {
+ staged.handle.close();
+ await abortTalerDatabaseReplacement(this.idbFactory, staged.name);
}
- });
- await this.applyDbFixups(access, (n) => this.emitNotification(n));
+ throw e;
+ }
+
+ // Deletion is deliberately outside the commit semantics: failure or a
+ // second client blocking it only retains an unreachable old generation.
+ await retireTalerDatabaseGeneration(this.idbFactory, oldName, staged.name);
}
async clearDatabase(): Promise<void> {
@@ -342,7 +393,10 @@ export class SqliteWalletDbHandle implements WalletDbHandle {
return await exportNativeSqliteDb(this.ndb);
}
- async importDatabase(dump: any): Promise<void> {
+ async importDatabase(
+ dump: any,
+ finalize: WalletDbImportFinalizer,
+ ): Promise<void> {
// See the IndexedDB counterpart: a dump from the other backend is a
// conversion job, and must not be half-imported here.
if (dump != null && typeof dump === "object" && "databases" in dump) {
@@ -351,7 +405,9 @@ export class SqliteWalletDbHandle implements WalletDbHandle {
" into the native sqlite backend; convert the database instead",
);
}
- await importNativeSqliteDb(this.ndb, dump);
+ await importNativeSqliteDb(this.ndb, dump, finalize, (n) =>
+ this.emitNotification(n),
+ );
}
async clearDatabase(): Promise<void> {
diff --git a/packages/taler-wallet-core/src/dbtx-handle.ts b/packages/taler-wallet-core/src/dbtx-handle.ts
@@ -41,6 +41,14 @@ export interface WalletDbAccessStats {
recordsRead: number;
}
+/**
+ * Wallet-level work that must become visible in the same atomic import as the
+ * restored records (currently rebuilding the materialized transaction view).
+ */
+export type WalletDbImportFinalizer = (
+ tx: WalletDbTransaction,
+) => Promise<void>;
+
export interface WalletDbHandle {
/**
* Which backend this is, for logs and test names.
@@ -68,12 +76,13 @@ export interface WalletDbHandle {
*
* Returns with the database consistent: a backend whose stored records need
* repairing after an import that may predate its current schema does that
- * repair here, so callers cannot forget to. Rebuilding wallet-level views
- * from the imported records is the caller's job -- that is not storage.
+ * repair here. The supplied finalizer rebuilds wallet-level derived state
+ * before the replacement becomes visible. A failure in import, repair, or
+ * finalization leaves the previous database authoritative.
*
* Throws if the dump did not come from this backend.
*/
- importDatabase(dump: any): Promise<void>;
+ importDatabase(dump: any, finalize: WalletDbImportFinalizer): Promise<void>;
/** Remove all records, leaving an empty database of the current schema. */
clearDatabase(): Promise<void>;
diff --git a/packages/taler-wallet-core/src/dbtx-sqlite.ts b/packages/taler-wallet-core/src/dbtx-sqlite.ts
@@ -270,9 +270,7 @@ export async function initSqliteWalletDb(
validateSchemaMigrations(migrations);
await db.exec("PRAGMA foreign_keys = ON");
await db.exec(`PRAGMA busy_timeout = ${SQLITE_BUSY_TIMEOUT_MS}`);
- const versionRows = await (
- await db.prepare("PRAGMA user_version")
- ).getAll();
+ const versionRows = await (await db.prepare("PRAGMA user_version")).getAll();
const databaseVersion = Number(versionRows[0]?.user_version ?? 0);
if (databaseVersion > SQLITE_SCHEMA_VERSION) {
throw Error(
@@ -5900,6 +5898,8 @@ export async function exportNativeSqliteDb(
export async function importNativeSqliteDb(
ndb: NativeSqliteWalletDb,
dump: NativeSqliteDbDump,
+ finalize: (tx: WalletDbTransaction) => Promise<void>,
+ notifyFn: (n: WalletNotification) => void,
): Promise<void> {
if (dump.schemaVersion !== SQLITE_SCHEMA_VERSION) {
throw Error(
@@ -5909,8 +5909,7 @@ export async function importNativeSqliteDb(
}
await ndb.lock.run(async () => {
const tables = await listDataTables(ndb);
- await ndb.txc.begin();
- try {
+ 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) {
@@ -5937,14 +5936,10 @@ export async function importNativeSqliteDb(
).run(params);
}
}
- await ndb.txc.commit();
- } catch (e) {
- try {
- await ndb.txc.rollback();
- } catch (rollbackErr) {
- logger.warn(`rollback failed: ${rollbackErr}`);
- }
- throw e;
- }
+ // 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.
+ await finalize(tx);
+ });
});
}
diff --git a/packages/taler-wallet-core/src/dbtx.test.ts b/packages/taler-wallet-core/src/dbtx.test.ts
@@ -106,4 +106,53 @@ for (const makeRunner of runnerFactories) {
await runner.close();
}
});
+
+ test(`dbtx ${makeRunner.name}: import and finalizer are atomic`, async () => {
+ const source = await makeRunner();
+ const target = await makeRunner();
+ 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,
+ });
+ throw Error("injected finalizer failure");
+ }),
+ /injected finalizer failure/,
+ );
+ const afterFailure = await target.runReadWriteTx((tx) =>
+ tx.getConfig(ConfigRecordKey.TestLoopTx),
+ );
+ assert.strictEqual(afterFailure?.value, 1);
+
+ await target.importDatabase(dump, async (tx) => {
+ await tx.upsertConfig({
+ key: ConfigRecordKey.TestLoopTx,
+ value: 3,
+ });
+ });
+ const afterSuccess = await target.runReadWriteTx((tx) =>
+ tx.getConfig(ConfigRecordKey.TestLoopTx),
+ );
+ assert.strictEqual(afterSuccess?.value, 3);
+ } finally {
+ await source.close();
+ await target.close();
+ }
+ });
}
diff --git a/packages/taler-wallet-core/src/requests.ts b/packages/taler-wallet-core/src/requests.ts
@@ -447,6 +447,7 @@ import {
applyRunConfigDefaults,
denomRefKey,
migrateMaterializedTransactions,
+ rematerializeTransactionsAtCurrentVersion,
walletExchangeClient,
} from "./wallet.js";
@@ -1839,13 +1840,11 @@ async function handleImportDb(
wex: WalletExecutionContext,
req: ImportDbRequest,
): Promise<EmptyObject> {
- // FIXME: This should atomically re-materialize transactions!
- // importDatabase leaves the records repaired, so the transaction view below
- // is built from repaired records rather than from what the dump contained.
- await wex.ws.db.importDatabase(req.dump);
-
- await wex.runWalletDbTx(async (tx) => {
- await rematerializeTransactions(wex, tx);
+ // Import, backend repairs and the derived transaction view become visible
+ // together. The backend keeps the old database authoritative if this
+ // finalizer throws.
+ await wex.ws.db.importDatabase(req.dump, async (tx) => {
+ await rematerializeTransactionsAtCurrentVersion(wex, tx);
});
// The import replaced the database underneath the DAL, writing through the
diff --git a/packages/taler-wallet-core/src/wallet-db-gate.test.ts b/packages/taler-wallet-core/src/wallet-db-gate.test.ts
@@ -77,3 +77,45 @@ test("database gate drains old work and admits queued work on replacement", asyn
"queued:new",
]);
});
+
+test("database import has exclusive admission", async () => {
+ const events: string[] = [];
+ const gate = new DbOperationGate();
+ const handle = fakeHandle("db", events);
+ const admitted = new AdmittedWalletDbHandle(() => handle, gate);
+ const active = deferred();
+ const releaseActive = deferred();
+ const importStarted = deferred();
+ const releaseImport = deferred();
+ handle.importDatabase = async () => {
+ events.push("import");
+ importStarted.resolve();
+ await releaseImport.promise;
+ };
+
+ const beforeImport = admitted.runReadWriteTx(async () => {
+ events.push("active");
+ active.resolve();
+ await releaseActive.promise;
+ });
+ await active.promise;
+ const importing = admitted.importDatabase({}, async () => {});
+ const afterImport = admitted.runReadWriteTx(async () => {
+ events.push("after");
+ });
+
+ await Promise.resolve();
+ assert.deepStrictEqual(events, ["db:tx", "active"]);
+ releaseActive.resolve();
+ await importStarted.promise;
+ assert.deepStrictEqual(events, ["db:tx", "active", "import"]);
+ releaseImport.resolve();
+ await Promise.all([beforeImport, importing, afterImport]);
+ assert.deepStrictEqual(events, [
+ "db:tx",
+ "active",
+ "import",
+ "db:tx",
+ "after",
+ ]);
+});
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
@@ -233,11 +233,16 @@ export class AdmittedWalletDbHandle implements WalletDbHandle {
exportDatabase(): Promise<any> {
return this.gate.runShared(() => this.current().exportDatabase());
}
- importDatabase(dump: any): Promise<void> {
- return this.gate.runShared(() => this.current().importDatabase(dump));
+ importDatabase(
+ dump: any,
+ finalize: Parameters<WalletDbHandle["importDatabase"]>[1],
+ ): Promise<void> {
+ return this.gate.runExclusive(() =>
+ this.current().importDatabase(dump, finalize),
+ );
}
clearDatabase(): Promise<void> {
- return this.gate.runShared(() => this.current().clearDatabase());
+ return this.gate.runExclusive(() => this.current().clearDatabase());
}
getAccessStats() {
return this.current().getAccessStats();
@@ -308,6 +313,18 @@ type CancelFn = () => void;
*/
const MATERIALIZED_TRANSACTIONS_VERSION = 4;
+/** Rebuild the transaction view and durably mark this exact view version. */
+export async function rematerializeTransactionsAtCurrentVersion(
+ wex: WalletExecutionContext,
+ tx: WalletDbTransaction,
+): Promise<void> {
+ await rematerializeTransactions(wex, tx);
+ await tx.upsertConfig({
+ key: ConfigRecordKey.MaterializedTransactionsVersion,
+ value: MATERIALIZED_TRANSACTIONS_VERSION,
+ });
+}
+
export async function migrateMaterializedTransactions(
wex: WalletExecutionContext,
): Promise<void> {
@@ -327,12 +344,7 @@ export async function migrateMaterializedTransactions(
}
}
- await rematerializeTransactions(wex, tx);
-
- await tx.upsertConfig({
- key: ConfigRecordKey.MaterializedTransactionsVersion,
- value: MATERIALIZED_TRANSACTIONS_VERSION,
- });
+ await rematerializeTransactionsAtCurrentVersion(wex, tx);
});
}
@@ -1184,19 +1196,24 @@ export class InternalWalletState {
// Opening and repairing is the backend's business. It reports how many
// fixups it applied, because rebuilding the transaction view afterwards
// is a wallet-level concern that no storage layer should know about.
- 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) {} },
- );
- await wex.runWalletDbTx(async (tx) => {
- await rematerializeTransactions(wex, tx);
- });
- }
+ 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);
+ });
+ }
+ });
} catch (e) {
logger.error(
"error writing to database during initialization (during migration)",
