commit e98a3a049fe4d502e30583e1c3effc8106e7b72f
parent bd26a6d0af141be293f6257b440d1d06c48dac89
Author: Florian Dold <dold@taler.net>
Date: Wed, 12 Aug 2026 03:03:24 +0200
wallet-core: improve native database migration, report progress
Diffstat:
22 files changed, 2048 insertions(+), 417 deletions(-)
diff --git a/packages/idb-bridge/src/bridge-idb.ts b/packages/idb-bridge/src/bridge-idb.ts
@@ -2719,8 +2719,6 @@ export class BridgeIDBTransaction
this.mode = mode;
this._db = db;
- this._db._transactions.push(this);
-
this._openRequest = openRequest ?? null;
}
diff --git a/packages/taler-util/src/notifications.ts b/packages/taler-util/src/notifications.ts
@@ -47,6 +47,7 @@ export enum NotificationType {
RequestObservabilityEvent = "request-observability-event",
RequestProgressError = "request-progress-error",
RequestProgressPhase = "request-progress-phase",
+ DatabaseMaintenanceProgress = "database-maintenance-progress",
}
export interface ErrorInfoSummary {
@@ -382,6 +383,21 @@ export interface IdleNotification {
type: NotificationType.Idle;
}
+/** Progress while startup fixups or a database migration hold the DB gate. */
+export interface DatabaseMaintenanceProgressNotification {
+ type: NotificationType.DatabaseMaintenanceProgress;
+ operation: "indexeddb-fixup" | "indexeddb-to-native-migration";
+ phase: "fixup" | "copy" | "verify" | "complete" | "failed";
+ /** Current fixup or backend-neutral store, when one is active. */
+ step?: string;
+ completedSteps: number;
+ totalSteps: number;
+ /** Records completed in the current store. */
+ processedRecords?: number;
+ /** Known after copying a store, and therefore available while verifying it. */
+ totalRecords?: number;
+}
+
export type WalletNotification =
| BalanceChangeNotification
| BankAccountChangeNotification
@@ -396,4 +412,5 @@ export type WalletNotification =
| RequestObservabilityEventNotification
| IdleNotification
| RequestProgressNotification
- | RequestProgressPhaseNotification;
+ | RequestProgressPhaseNotification
+ | DatabaseMaintenanceProgressNotification;
diff --git a/packages/taler-wallet-cli/src/index.ts b/packages/taler-wallet-cli/src/index.ts
@@ -79,6 +79,7 @@ import {
inspectWalletDbPath,
nativeCrypto,
rollbackWalletDbMigration,
+ resolveWalletDbMigration,
Wallet,
WalletApiOperation,
WalletCoreApiClient,
@@ -3441,6 +3442,36 @@ advancedCli
});
advancedCli
+ .subcommand("dbMigrationResolve", "db-migration-resolve", {
+ help: "Resolve an ambiguous mixed-schema wallet database offline.",
+ })
+ .requiredArgument("dbfile", clk.STRING, {
+ help: "Wallet database file to resolve (the wallet must be stopped).",
+ })
+ .requiredOption("keep", ["--keep"], clk.STRING, {
+ help: "Authoritative schema: indexeddb or native.",
+ })
+ .requiredOption("backup", ["--backup"], clk.STRING, {
+ help: "Nonexistent destination for the mandatory full SQLite backup.",
+ })
+ .flag("confirm", ["--confirm"], {
+ help: "Confirm the explicit authority selection.",
+ })
+ .action(async (args) => {
+ const cmd = args.dbMigrationResolve;
+ if (!cmd.confirm) {
+ throw new CliUsageError("db-migration-resolve requires --confirm");
+ }
+ if (cmd.keep !== "indexeddb" && cmd.keep !== "native") {
+ throw new CliUsageError("--keep must be indexeddb or native");
+ }
+ await resolveWalletDbMigration(cmd.dbfile, cmd.keep, cmd.backup);
+ console.log(
+ `kept ${cmd.keep} as authoritative; full pre-resolution backup written to ${cmd.backup}`,
+ );
+ });
+
+advancedCli
.subcommand("diagnostics", "diagnostics", {
help: "Print diagnostics info.",
})
diff --git a/packages/taler-wallet-core/src/db-converter.test.ts b/packages/taler-wallet-core/src/db-converter.test.ts
@@ -30,7 +30,9 @@ import { test } from "node:test";
import {
encodeCrock,
getRandomBytes,
+ NotificationType,
TalerPreciseTimestamp,
+ WalletNotification,
} from "@gnu-taler/taler-util";
import {
@@ -43,7 +45,9 @@ import {
WalletPurchase,
} from "./db-common.js";
import { SQLITE_BASELINE_SCHEMA } from "./db-sqlite-schema.js";
-import { convertWalletDb } from "./db-converter.js";
+import { convertWalletDb, DB_CONVERSION_BATCH_SIZE } from "./db-converter.js";
+import { applyFixups, WalletIndexedDbStoresV1 } from "./db-indexeddb.js";
+import { IdbWalletDbHandle } from "./dbtx-handle-impl.js";
import { conformanceCases } from "./dbtx-conformance-cases.js";
import { ConformanceAsserts } from "./dbtx-conformance.js";
import { makeIdbRunner, makeSqliteRunner } from "./dbtx-runners.js";
@@ -60,6 +64,8 @@ const quietAsserts: ConformanceAsserts = {
test("converter: IndexedDB to sqlite, populated by the conformance corpus", async () => {
const src = await makeIdbRunner();
+ const progress: WalletNotification[] = [];
+ src.setNotificationSink((n) => progress.push(n));
for (const c of conformanceCases) {
try {
await c.run(quietAsserts, src);
@@ -69,16 +75,120 @@ test("converter: IndexedDB to sqlite, populated by the conformance corpus", asyn
}
}
+ // Simulate pre-Clause-Schnorr records. First prove the IndexedDB fixup is
+ // idempotent and fills both stores, then remove the fields again to prove
+ // converter-side normalization protects unusual/imported legacy records.
+ let legacyCoinPub = "";
+ let legacyPlanchetPub = "";
+ await src.runReadWriteTx(async (tx) => {
+ const coin = (await tx.listAllCoins())[0];
+ const planchet = (await tx.listAllPlanchets())[0];
+ assert.ok(coin && planchet, "corpus did not create legacy test records");
+ legacyCoinPub = coin.coinPub;
+ legacyPlanchetPub = planchet.coinPub;
+ delete (coin as any).exchangeWithdrawValues;
+ delete (planchet as any).exchangeWithdrawValues;
+ await tx.upsertCoin(coin);
+ await tx.upsertPlanchet(planchet);
+ });
+ // Force a store across multiple conversion pages. Tombstones are
+ // independent records, so this tests batching without manufacturing a
+ // large graph of otherwise unrelated wallet operations.
+ await src.runReadWriteTx(async (tx) => {
+ for (let i = 0; i < DB_CONVERSION_BATCH_SIZE * 2 + 17; i++) {
+ await tx.upsertTombstone({ id: `bounded-conversion-${i}` });
+ }
+ });
+ const idb = src as IdbWalletDbHandle;
+ const raw = await idb.rawAccess();
+ await raw.runAllStoresReadWriteTx({}, async (tx) => {
+ await tx.fixups.delete("fixup20260812ExchangeWithdrawValues");
+ });
+ await applyFixups(raw);
+ await src.runReadWriteTx(async (tx) => {
+ assert.deepStrictEqual(
+ (await tx.getCoin(legacyCoinPub))?.exchangeWithdrawValues,
+ { cipher: "RSA" },
+ );
+ assert.deepStrictEqual(
+ (await tx.getPlanchet(legacyPlanchetPub))?.exchangeWithdrawValues,
+ { cipher: "RSA" },
+ );
+ const coin = (await tx.getCoin(legacyCoinPub))!;
+ const planchet = (await tx.getPlanchet(legacyPlanchetPub))!;
+ delete (coin as any).exchangeWithdrawValues;
+ delete (planchet as any).exchangeWithdrawValues;
+ await tx.upsertCoin(coin);
+ await tx.upsertPlanchet(planchet);
+ });
+
const dst = await makeSqliteRunner();
+ const retainedTransactionsBefore = (src as any).idbHandle._transactions
+ .length;
+ const pageSizes: number[] = [];
+ for (const handle of [src, dst]) {
+ const originalRun = handle.runReadWriteTx.bind(handle);
+ handle.runReadWriteTx = (f) =>
+ originalRun(async (tx) => {
+ const originalScan = tx.scanMigrationRecords.bind(tx);
+ tx.scanMigrationRecords = async (...args) => {
+ const page = await originalScan(...args);
+ pageSizes.push(page.records.length);
+ return page;
+ };
+ return await f(tx);
+ });
+ }
// convertWalletDb re-enumerates both sides and compares every record;
// a thrown error here is the actual test.
const report = await convertWalletDb(src, dst);
+ await dst.runReadWriteTx(async (tx) => {
+ assert.deepStrictEqual(
+ (await tx.getCoin(legacyCoinPub))?.exchangeWithdrawValues,
+ { cipher: "RSA" },
+ );
+ assert.deepStrictEqual(
+ (await tx.getPlanchet(legacyPlanchetPub))?.exchangeWithdrawValues,
+ { cipher: "RSA" },
+ );
+ });
+
assert.ok(
report.totalRecords >= 100,
`only ${report.totalRecords} records converted -- the corpus did not` +
` populate the source, so the conversion proved nothing`,
);
+ assert.ok(pageSizes.length > 4, "conversion did not use multiple pages");
+ assert.ok(
+ Math.max(...pageSizes) <= DB_CONVERSION_BATCH_SIZE,
+ `conversion retained a page of ${Math.max(...pageSizes)} records`,
+ );
+ assert.ok(
+ progress.some(
+ (n) =>
+ n.type === NotificationType.DatabaseMaintenanceProgress &&
+ n.operation === "indexeddb-to-native-migration" &&
+ n.phase === "copy" &&
+ n.step === "tombstones" &&
+ (n.processedRecords ?? 0) > DB_CONVERSION_BATCH_SIZE,
+ ),
+ "copy progress did not advance across batches",
+ );
+ assert.ok(
+ progress.some(
+ (n) =>
+ n.type === NotificationType.DatabaseMaintenanceProgress &&
+ n.operation === "indexeddb-to-native-migration" &&
+ n.phase === "verify",
+ ),
+ "verification progress was not reported",
+ );
+ assert.strictEqual(
+ (src as any).idbHandle._transactions.length,
+ retainedTransactionsBefore,
+ "completed scan transactions were retained by the IndexedDB bridge",
+ );
// Every store in the plan must have been visited (0 records is fine for a
// store the corpus leaves empty; a missing key means the plan lost a step).
assert.ok(
@@ -218,33 +328,15 @@ test("converter: canonicalises a lowercase peer-push contract private key", asyn
await dst.close();
});
-test("converter: reserve rows sharing a public key collapse into one", async () => {
- // What an IndexedDB wallet that ever received a peer payment looks like:
- // the merge reserve stored once by the exchange entry that points at it,
- // and again by each peer-credit withdrawal group, under a fresh row id.
- // The native schema declares the public key unique, so the conversion has
- // to keep exactly the row the exchange entry references.
+test("IndexedDB fixup collapses only identical reserves and remaps references", async () => {
const src = await makeIdbRunner();
const reservePub = encodeCrock(getRandomBytes(32));
const reservePriv = encodeCrock(getRandomBytes(32));
- const rowIds = await src.runReadWriteTx(async (tx) => {
- const ids = [];
+ const ids = await src.runReadWriteTx(async (tx) => {
+ const out = [];
for (let i = 0; i < 3; i++) {
- ids.push(await tx.upsertReserve({ reservePub, reservePriv }));
+ out.push(await tx.upsertReserve({ reservePub, reservePriv }));
}
- // A second, genuinely different reserve, which must survive untouched.
- ids.push(
- await tx.upsertReserve({
- reservePub: encodeCrock(getRandomBytes(32)),
- reservePriv: encodeCrock(getRandomBytes(32)),
- }),
- );
- return ids;
- });
- // The exchange points at the *last* of the duplicates, so keeping the
- // first would leave the entry referencing a row that no longer exists.
- const referencedRowId = rowIds[2];
- await src.runReadWriteTx(async (tx) => {
await tx.upsertExchange({
baseUrl: "https://exchange.example.com/",
detailsPointer: undefined,
@@ -257,21 +349,80 @@ test("converter: reserve rows sharing a public key collapse into one", async ()
nextUpdateStamp: timestampPreciseToDb(TalerPreciseTimestamp.now()),
lastKeysEtag: undefined,
nextRefreshCheckStamp: timestampPreciseToDb(TalerPreciseTimestamp.now()),
- currentMergeReserveRowId: referencedRowId,
+ currentMergeReserveRowId: out[2],
});
+ return out;
+ });
+ const raw = await (src as IdbWalletDbHandle).rawAccess();
+ await raw.runAllStoresReadWriteTx({}, (tx) =>
+ tx.fixups.delete("fixup20260720DuplicateReserves"),
+ );
+ const fixupProgress: WalletNotification[] = [];
+ await applyFixups(raw, (n) => fixupProgress.push(n));
+ assert.ok(
+ fixupProgress.some(
+ (n) =>
+ n.type === NotificationType.DatabaseMaintenanceProgress &&
+ n.operation === "indexeddb-fixup" &&
+ n.phase === "fixup" &&
+ n.step === "fixup20260720DuplicateReserves",
+ ),
+ "fixup progress was not reported",
+ );
+ assert.ok(
+ fixupProgress.some(
+ (n) =>
+ n.type === NotificationType.DatabaseMaintenanceProgress &&
+ n.operation === "indexeddb-fixup" &&
+ n.phase === "complete",
+ ),
+ "fixup completion was not reported",
+ );
+ await src.runReadWriteTx(async (tx) => {
+ const matching = (await tx.listAllReserves()).filter(
+ (r) => r.reservePub === reservePub,
+ );
+ assert.strictEqual(matching.length, 1);
+ assert.strictEqual(matching[0].rowId, ids[0]);
+ assert.strictEqual(
+ (await tx.getExchange("https://exchange.example.com/"))
+ ?.currentMergeReserveRowId,
+ ids[0],
+ );
});
-
const dst = await makeSqliteRunner();
- const report = await convertWalletDb(src, dst);
- assert.strictEqual(report.copied.reserves, 2);
+ await convertWalletDb(src, dst);
+ await src.close();
+ await dst.close();
+});
+
+test("converter: remaining duplicate reserve public keys are rejected", async () => {
+ // The IndexedDB fixup is responsible for safely deduplicating byte-for-byte
+ // identical rows and remapping their references. If duplicates remain at
+ // conversion time, selecting a winner would hide corruption.
+ const src = await makeIdbRunner();
+ const reservePub = encodeCrock(getRandomBytes(32));
+ const reservePriv = encodeCrock(getRandomBytes(32));
+ await src.runReadWriteTx(async (tx) => {
+ for (let i = 0; i < 3; i++) {
+ await tx.upsertReserve({ reservePub, reservePriv });
+ }
+ // A second, genuinely different reserve, which must survive untouched.
+ await tx.upsertReserve({
+ reservePub: encodeCrock(getRandomBytes(32)),
+ reservePriv: encodeCrock(getRandomBytes(32)),
+ });
+ });
- const reserves = await dst.runReadWriteTx((tx) => tx.listAllReserves());
- assert.strictEqual(reserves.length, 2);
- const kept = reserves.find((r) => r.reservePub === reservePub);
+ const dst = await makeSqliteRunner();
+ await assert.rejects(
+ () => convertWalletDb(src, dst),
+ /multiple reserve rows have public key/,
+ );
assert.strictEqual(
- kept?.rowId,
- referencedRowId,
- "the reserve the exchange entry references was not the one kept",
+ (await src.runReadWriteTx((tx) => tx.listAllReserves())).length,
+ 4,
+ "refusal modified the source",
);
await src.close();
@@ -364,3 +515,37 @@ test("converter: the copy plan covers every table in the schema", async () => {
}
}
});
+
+test("converter: every IndexedDB store is copied or explicitly obsolete", async () => {
+ const src = await makeIdbRunner();
+ const dst = await makeSqliteRunner();
+ const visited = new Set(
+ Object.keys((await convertWalletDb(src, dst)).copied),
+ );
+ await src.close();
+ await dst.close();
+
+ const renamed: Record<string, string> = {
+ coinAvailabilityV2: "coinAvailability",
+ denominationsV2: "denominations",
+ bankAccountsV2: "bankAccounts",
+ };
+ const excluded: Record<string, string> = {
+ coinAvailability: "obsolete pre-master-key store retained for fixups",
+ denominations: "obsolete pre-master-key store retained for fixups",
+ bankWithdrawUris: "obsolete unused legacy URI cache",
+ fixups: "schema repair log, not wallet data",
+ obsolete_backupProviders: "obsolete",
+ _obsolete_transactions: "obsolete materialized view",
+ _obsolete_bankAccounts: "obsolete pre-V2 store",
+ _obsolete_rewards: "obsolete",
+ obsolete_userAttention: "obsolete",
+ };
+ for (const store of Object.keys(WalletIndexedDbStoresV1)) {
+ const step = renamed[store] ?? store;
+ assert.ok(
+ visited.has(step) || excluded[store] !== undefined,
+ `IndexedDB store ${store} is neither copied nor explicitly classified`,
+ );
+ }
+});
diff --git a/packages/taler-wallet-core/src/db-converter.ts b/packages/taler-wallet-core/src/db-converter.ts
@@ -31,40 +31,54 @@
* nothing here mutates the source.
*/
-import { Logger } from "@gnu-taler/taler-util";
+import {
+ Logger,
+ NotificationType,
+ sha256,
+ stringToBytes,
+} from "@gnu-taler/taler-util";
-import { WalletPurchase, WalletReserve } from "./db-common.js";
+import {
+ WalletCoin,
+ WalletPlanchet,
+ WalletPurchase,
+ WalletReserve,
+} from "./db-common.js";
import { WalletDbHandle } from "./dbtx-handle.js";
-import { WalletDbTransaction } from "./dbtx.js";
+import { WalletDbMigrationStore, WalletDbTransaction } from "./dbtx.js";
const logger = new Logger("db-converter.ts");
/**
* One entity to copy.
*
- * `read` enumerates every record through the DAL; `write` stores one.
+ * `read` identifies the existing DAL enumeration used by the backend's
+ * bounded migration scanner; `write` stores one record.
* `normalize` is applied before the verification comparison, for the few
* stores where the destination deliberately canonicalises a representation
* (for example global-currency ids, which nothing else references).
*/
interface CopyStep {
- name: string;
+ name: WalletDbMigrationStore;
read: (tx: WalletDbTransaction) => Promise<unknown[]>;
write: (tx: WalletDbTransaction, rec: unknown) => Promise<unknown>;
normalize?: (rec: unknown) => unknown;
+ validate?: (tx: WalletDbTransaction, rec: unknown) => Promise<void>;
}
function step<T>(
- name: string,
+ name: WalletDbMigrationStore,
read: (tx: WalletDbTransaction) => Promise<T[]>,
write: (tx: WalletDbTransaction, rec: T) => Promise<unknown>,
normalize?: (rec: T) => unknown,
+ validate?: (tx: WalletDbTransaction, rec: T) => Promise<void>,
): CopyStep {
return {
name,
read: read as CopyStep["read"],
write: write as CopyStep["write"],
normalize: normalize as CopyStep["normalize"],
+ validate: validate as CopyStep["validate"],
};
}
@@ -137,54 +151,37 @@ function normalizeContractPriv<T extends { contractPriv: string }>(rec: T): T {
}
/**
- * The reserves, one per reserve public key.
+ * Require the one-reserve-per-public-key invariant of the native schema.
*
- * The IndexedDB store is keyed by an auto-increment row id and has no unique
- * index on the public key, so it accumulated one extra row per peer-credit
- * withdrawal group created against an exchange whose merge reserve already
- * existed: the same key pair, stored again under a new row id. The native
- * schema declares the public key unique, which is what a reserve record
- * actually is, so those rows have to go somewhere -- and they carry nothing
- * the kept row does not.
- *
- * Which one is kept matters: an exchange entry references its merge reserve by
- * row id, so the referenced row survives and the reference stays valid. With
- * none of them referenced, the lowest row id wins, which is the one the wallet
- * created first.
+ * The IndexedDB fixup may remove byte-identical duplicates after remapping all
+ * references. Anything left here is ambiguous; conversion never picks a
+ * winner merely because one row happens to be referenced or older.
*/
-async function listReservesByPub(
+async function validateReservePub(
tx: WalletDbTransaction,
-): Promise<WalletReserve[]> {
- const reserves = await tx.listAllReserves();
- const referenced = new Set<number>();
- for (const ex of await tx.getExchanges()) {
- if (ex.currentMergeReserveRowId != null) {
- referenced.add(ex.currentMergeReserveRowId);
- }
- }
- const byPub = new Map<string, WalletReserve>();
- for (const r of reserves) {
- const kept = byPub.get(r.reservePub);
- if (!kept) {
- byPub.set(r.reservePub, r);
- continue;
- }
- if (referenced.has(kept.rowId ?? -1)) {
- continue;
- }
- if (
- referenced.has(r.rowId ?? -1) ||
- (r.rowId ?? Infinity) < (kept.rowId ?? Infinity)
- ) {
- byPub.set(r.reservePub, r);
- }
- }
- if (byPub.size !== reserves.length) {
- logger.info(
- `collapsed ${reserves.length} reserve rows into ${byPub.size} reserves`,
+ reserve: WalletReserve,
+): Promise<void> {
+ const selected = await tx.getReserveByReservePub(reserve.reservePub);
+ if (!selected || selected.rowId !== reserve.rowId) {
+ throw Error(
+ `conversion refused: multiple reserve rows have public key ${reserve.reservePub}`,
);
}
- return [...byPub.values()];
+}
+
+const rsaWithdrawValues = { cipher: "RSA" } as const;
+
+function normalizePlanchet(rec: WalletPlanchet): WalletPlanchet {
+ return rec.exchangeWithdrawValues === undefined
+ ? ({ ...rec, exchangeWithdrawValues: rsaWithdrawValues } as WalletPlanchet)
+ : rec;
+}
+
+function normalizeCoin(rec: WalletCoin): WalletCoin {
+ const stripped = stripLegacy("coins")!(rec) as WalletCoin;
+ return stripped.exchangeWithdrawValues === undefined
+ ? ({ ...stripped, exchangeWithdrawValues: rsaWithdrawValues } as WalletCoin)
+ : stripped;
}
/**
@@ -269,8 +266,10 @@ const COPY_PLAN: CopyStep[][] = [
// reserve by row id.
step(
"reserves",
- (tx) => listReservesByPub(tx),
+ (tx) => tx.listAllReserves(),
(tx, r) => tx.upsertReserve(r),
+ undefined,
+ validateReservePub,
),
],
[
@@ -279,6 +278,16 @@ const COPY_PLAN: CopyStep[][] = [
(tx) => tx.getExchanges(),
(tx, r) => tx.upsertExchange(stripLegacy("exchanges")!(r) as any),
stripLegacy("exchanges"),
+ async (tx, r) => {
+ if (
+ r.currentMergeReserveRowId != null &&
+ !(await tx.getReserve(r.currentMergeReserveRowId))
+ ) {
+ throw Error(
+ `conversion refused: exchange ${r.baseUrl} references missing reserve ${r.currentMergeReserveRowId}`,
+ );
+ }
+ },
),
],
[
@@ -291,15 +300,16 @@ const COPY_PLAN: CopyStep[][] = [
[
step(
"exchangeSignKeys",
- async (tx) => {
- const out = [];
- for (const det of await tx.listAllExchangeDetails()) {
- if (det.rowId == null) continue;
- out.push(...(await tx.getExchangeSignKeysByDetailsRowId(det.rowId)));
+ (tx) => tx.listAllExchangeSignKeys(),
+ (tx, r) => tx.upsertExchangeSignKey(r),
+ undefined,
+ async (tx, r) => {
+ if (!(await tx.getExchangeDetailsByRowId(r.exchangeDetailsRowId))) {
+ throw Error(
+ `conversion refused: signing key references missing exchange details ${r.exchangeDetailsRowId}`,
+ );
}
- return out;
},
- (tx, r) => tx.upsertExchangeSignKey(r),
),
step(
"denominationFamilies",
@@ -339,46 +349,50 @@ const COPY_PLAN: CopyStep[][] = [
(tx) => tx.listAllCoins(),
// Written through the strip too, so an IndexedDB destination does not
// re-preserve the junk the conversion exists to shed.
- (tx, r) => tx.upsertCoin(stripLegacy("coins")!(r) as any),
- stripLegacy("coins"),
+ (tx, r) => tx.upsertCoin(normalizeCoin(r)),
+ normalizeCoin,
),
],
[
step(
"planchets",
- async (tx) => {
- const out = [];
- for (const wg of await tx.listAllWithdrawalGroups()) {
- out.push(...(await tx.getPlanchetsByGroup(wg.withdrawalGroupId)));
+ (tx) => tx.listAllPlanchets(),
+ (tx, r) => tx.upsertPlanchet(normalizePlanchet(r)),
+ normalizePlanchet,
+ async (tx, r) => {
+ if (!(await tx.getWithdrawalGroup(r.withdrawalGroupId))) {
+ throw Error(
+ `conversion refused: planchet references missing withdrawal group ${r.withdrawalGroupId}`,
+ );
}
- return out;
},
- (tx, r) => tx.upsertPlanchet(r),
),
step(
"refreshSessions",
- async (tx) => {
- const out = [];
- for (const rg of await tx.listAllRefreshGroups()) {
- out.push(...(await tx.getRefreshSessionsByGroup(rg.refreshGroupId)));
- }
- return out;
- },
+ (tx) => tx.listAllRefreshSessions(),
(tx, r) =>
tx.upsertRefreshSession(stripLegacy("refreshSessions")!(r) as any),
stripLegacy("refreshSessions"),
+ async (tx, r) => {
+ if (!(await tx.getRefreshGroup(r.refreshGroupId))) {
+ throw Error(
+ `conversion refused: refresh session references missing refresh group ${r.refreshGroupId}`,
+ );
+ }
+ },
),
step(
"coinHistory",
- async (tx) => {
- const out = [];
- for (const c of await tx.listAllCoins()) {
- const h = await tx.getCoinHistory(c.coinPub);
- if (h) out.push(h);
+ (tx) => tx.listAllCoinHistories(),
+ (tx, r) => tx.upsertCoinHistory(r),
+ undefined,
+ async (tx, r) => {
+ if (!(await tx.getCoin(r.coinPub))) {
+ throw Error(
+ `conversion refused: coin history references missing coin ${r.coinPub}`,
+ );
}
- return out;
},
- (tx, r) => tx.upsertCoinHistory(r),
),
step(
"coinAvailability",
@@ -387,13 +401,7 @@ const COPY_PLAN: CopyStep[][] = [
),
step(
"refundGroups",
- async (tx) => {
- const out = [];
- for (const p of await tx.listAllPurchases()) {
- out.push(...(await tx.getRefundGroupsByProposal(p.proposalId)));
- }
- return out;
- },
+ (tx) => tx.listAllRefundGroups(),
(tx, r) => tx.upsertRefundGroup(r),
),
step(
@@ -444,6 +452,13 @@ const COPY_PLAN: CopyStep[][] = [
(tx) => tx.listAllPeerPullCredits(),
(tx, r) => tx.upsertPeerPullCredit(r),
normalizeContractPriv,
+ async (tx, r) => {
+ if (!(await tx.getReserve(r.mergeReserveRowId))) {
+ throw Error(
+ `conversion refused: peer pull credit references missing reserve ${r.mergeReserveRowId}`,
+ );
+ }
+ },
),
step(
"donationSummaries",
@@ -469,20 +484,22 @@ const COPY_PLAN: CopyStep[][] = [
[
step(
"refundItems",
- async (tx) => {
- const out = [];
- for (const p of await tx.listAllPurchases()) {
- for (const rg of await tx.getRefundGroupsByProposal(p.proposalId)) {
- out.push(...(await tx.getRefundItemsByGroup(rg.refundGroupId)));
- }
+ (tx) => tx.listAllRefundItems(),
+ (tx, r) => tx.upsertRefundItem(r),
+ undefined,
+ async (tx, r) => {
+ if (!(await tx.getRefundGroup(r.refundGroupId))) {
+ throw Error(
+ `conversion refused: refund item references missing refund group ${r.refundGroupId}`,
+ );
}
- return out;
},
- (tx, r) => tx.upsertRefundItem(r),
),
],
];
+export const DB_CONVERSION_STEP_COUNT = COPY_PLAN.flat().length;
+
export interface DbConversionReport {
/** Records copied, per store. */
copied: Record<string, number>;
@@ -490,6 +507,9 @@ export interface DbConversionReport {
totalRecords: number;
}
+/** Small enough to bound retained records while amortising transaction setup. */
+export const DB_CONVERSION_BATCH_SIZE = 128;
+
/**
* JSON stringification with sorted object keys, so structurally equal
* records compare equal regardless of property insertion order -- the two
@@ -514,11 +534,95 @@ function stableStringify(v: unknown): string {
}
/**
+ * Order-independent SHA-256 multiset digest.
+ *
+ * Each canonical record is hashed separately and the 256-bit hashes are
+ * added modulo 2^256. Addition preserves multiplicity but does not require
+ * records from the two different physical schemas to arrive in the same
+ * order. Only this fixed 32-byte accumulator is retained between batches.
+ */
+class RecordMultisetDigest {
+ private sum = new Uint8Array(32);
+ count = 0;
+
+ add(record: unknown): void {
+ const encoded = stableStringify(record);
+ const h = sha256(stringToBytes(encoded));
+ let carry = 0;
+ for (let i = this.sum.length - 1; i >= 0; i--) {
+ const n = this.sum[i] + h[i] + carry;
+ this.sum[i] = n & 0xff;
+ carry = n >>> 8;
+ }
+ this.count++;
+ }
+
+ equals(other: RecordMultisetDigest): boolean {
+ if (this.count !== other.count) return false;
+ let difference = 0;
+ for (let i = 0; i < this.sum.length; i++) {
+ difference |= this.sum[i] ^ other.sum[i];
+ }
+ return difference === 0;
+ }
+
+ describe(): string {
+ return `${this.count}:${Array.from(this.sum, (x) =>
+ x.toString(16).padStart(2, "0"),
+ ).join("")}`;
+ }
+}
+
+async function readPage(
+ handle: WalletDbHandle,
+ step: CopyStep,
+ cursor: unknown | undefined,
+ validate: boolean,
+): Promise<{ records: unknown[]; nextCursor?: unknown }> {
+ return await handle.runReadWriteTx(async (tx) => {
+ const page = await tx.scanMigrationRecords(
+ step.name,
+ step.read,
+ cursor,
+ DB_CONVERSION_BATCH_SIZE,
+ );
+ if (validate && step.validate) {
+ for (const record of page.records) {
+ await step.validate(tx, record);
+ }
+ }
+ return page;
+ });
+}
+
+async function digestStore(
+ handle: WalletDbHandle,
+ step: CopyStep,
+ progress?: (processed: number) => void,
+): Promise<RecordMultisetDigest> {
+ const digest = new RecordMultisetDigest();
+ let cursor: unknown | undefined;
+ while (true) {
+ const page = await readPage(handle, step, cursor, false);
+ if (page.records.length === 0) break;
+ const normalize = step.normalize ?? ((r: unknown) => r);
+ for (const record of page.records) {
+ digest.add(normalize(record));
+ }
+ progress?.(digest.count);
+ cursor = page.nextCursor;
+ if (cursor === undefined) break;
+ }
+ return digest;
+}
+
+/**
* Copy every record from src into dst, then verify the copy.
*
- * Verification re-enumerates both databases through the same accessors and
- * compares the full normalised record sets, not just counts: a mapper that
- * drops a field produces equal counts and unequal records.
+ * Verification re-enumerates both databases through the same bounded
+ * accessors and compares count plus an order-independent digest of every
+ * normalised record. A mapper that drops a field produces equal counts and
+ * unequal digests without retaining either store in memory.
*
* Throws on any difference; the destination should then be discarded.
*/
@@ -529,62 +633,72 @@ export async function convertWalletDb(
const copied: Record<string, number> = {};
let total = 0;
- for (const group of COPY_PLAN) {
- // Reads and writes use separate transactions on separate handles; the
- // source is not written to at any point.
- const batches: Array<{ step: CopyStep; recs: unknown[] }> = [];
- await src.runReadWriteTx(async (tx) => {
- for (const st of group) {
- batches.push({ step: st, recs: await st.read(tx) });
- }
+ const notify = (
+ phase: "copy" | "verify",
+ completedSteps: number,
+ step?: CopyStep,
+ processedRecords?: number,
+ ): void => {
+ src.emitNotification({
+ type: NotificationType.DatabaseMaintenanceProgress,
+ operation: "indexeddb-to-native-migration",
+ phase,
+ completedSteps,
+ totalSteps: DB_CONVERSION_STEP_COUNT,
+ ...(step ? { step: step.name } : {}),
+ ...(processedRecords !== undefined ? { processedRecords } : {}),
+ ...(step && copied[step.name] !== undefined
+ ? { totalRecords: copied[step.name] }
+ : {}),
});
- await dst.runReadWriteTx(async (tx) => {
- for (const { step: st, recs } of batches) {
- for (const rec of recs) {
- await st.write(tx, rec);
- }
- copied[st.name] = recs.length;
- total += recs.length;
- logger.trace(`copied ${recs.length} ${st.name}`);
+ };
+
+ let stepIndex = 0;
+ for (const group of COPY_PLAN) {
+ for (const st of group) {
+ let cursor: unknown | undefined;
+ let storeCount = 0;
+ while (true) {
+ const page = await readPage(src, st, cursor, true);
+ if (page.records.length === 0) break;
+ await dst.runReadWriteTx(async (tx) => {
+ for (const rec of page.records) {
+ await st.write(tx, rec);
+ }
+ });
+ storeCount += page.records.length;
+ notify("copy", stepIndex, st, storeCount);
+ cursor = page.nextCursor;
+ if (cursor === undefined) break;
}
- });
+ copied[st.name] = storeCount;
+ total += storeCount;
+ stepIndex++;
+ notify("copy", stepIndex, st, storeCount);
+ logger.trace(`copied ${storeCount} ${st.name}`);
+ }
}
- // Verify: same enumeration on both sides, canonicalised and sorted.
+ // Verify using fixed-size multiset digests. Source and destination have
+ // different primary keys/orderings for some entities, so comparing page
+ // boundaries would be incorrect even though both scans are bounded.
+ stepIndex = 0;
for (const group of COPY_PLAN) {
for (const st of group) {
- const norm = st.normalize ?? ((r: unknown) => r);
- const a = (await src.runReadWriteTx((tx) => st.read(tx)))
- .map((r) => stableStringify(norm(r)))
- .sort();
- const b = (await dst.runReadWriteTx((tx) => st.read(tx)))
- .map((r) => stableStringify(norm(r)))
- .sort();
- if (a.length !== b.length) {
+ const sourceDigest = await digestStore(src, st);
+ const destinationDigest = await digestStore(dst, st, (processed) =>
+ notify("verify", stepIndex, st, processed),
+ );
+ if (!sourceDigest.equals(destinationDigest)) {
throw Error(
- `conversion verification failed: ${st.name} has ${a.length}` +
- ` records in the source but ${b.length} in the destination`,
+ `conversion verification failed: ${st.name} differs between` +
+ ` source and destination (${sourceDigest.describe()} versus` +
+ ` ${destinationDigest.describe()})`,
);
}
- for (let i = 0; i < a.length; i++) {
- if (a[i] !== b[i]) {
- // The lists are sorted, so one differing record misaligns every
- // later pair; report the first record on each side without a
- // partner rather than a misleading side-by-side of two different
- // records.
- const bs = new Set(b);
- const onlySrc = a.find((x) => !bs.has(x));
- const as = new Set(a);
- const onlyDst = b.find((x) => !as.has(x));
- throw Error(
- `conversion verification failed: ${st.name} differs between` +
- ` source and destination:\n only in src: ${onlySrc}\n` +
- ` only in dst: ${onlyDst}`,
- );
- }
- }
+ stepIndex++;
+ notify("verify", stepIndex, st, destinationDigest.count);
}
}
-
return { copied, totalRecords: total };
}
diff --git a/packages/taler-wallet-core/src/db-indexeddb.ts b/packages/taler-wallet-core/src/db-indexeddb.ts
@@ -54,6 +54,7 @@ import {
ExchangeRefundRequest,
HashCodeString,
Logger,
+ NotificationType,
MailboxConfiguration,
MailboxMessageRecord,
MerchantContractTokenDetails,
@@ -72,6 +73,7 @@ import {
TransferOptionRaw,
UnblindedDenominationSignature,
WireInfo,
+ WalletNotification,
WithdrawalExchangeAccountDetails,
ZeroLimitedOperation,
canonicalJson,
@@ -1549,6 +1551,12 @@ export interface FixupDescription {
* Fixups *must* be idempotent.
*/
export const walletDbFixups: FixupDescription[] = [
+ // Clause-Schnorr support made this field explicit. Older RSA records imply
+ // the RSA defaults and remain valid after the field was introduced.
+ {
+ fn: fixup20260812ExchangeWithdrawValues,
+ name: "fixup20260812ExchangeWithdrawValues",
+ },
// Deduplicate reserve rows left behind by a version that inserted its
// merge reserve repeatedly. Needed for as long as pre-2024 databases can
// still be imported.
@@ -1828,38 +1836,39 @@ async function fixup20260718StatusEnumDigits(
async function fixup20260213RefreshBlunder(
tx: WalletIndexedDbTransaction,
): Promise<void> {
- const refreshes = await tx.refreshGroups.indexes.byStatus.getAll(
- RefreshOperationStatus.Failed,
- );
- for (const refreshGroup of refreshes) {
- for (
- let coinIndex = 0;
- coinIndex < refreshGroup.statusPerCoin.length;
- coinIndex++
- ) {
- let changed = false;
- if (refreshGroup.statusPerCoin[coinIndex] === RefreshCoinStatus.Failed) {
- const rs = await tx.refreshSessions.get([
- refreshGroup.refreshGroupId,
- coinIndex,
- ]);
+ await tx.refreshGroups.indexes.byStatus
+ .iter(RefreshOperationStatus.Failed)
+ .forEachAsync(async (refreshGroup) => {
+ for (
+ let coinIndex = 0;
+ coinIndex < refreshGroup.statusPerCoin.length;
+ coinIndex++
+ ) {
+ let changed = false;
if (
- rs?.lastError?.code ===
- TalerErrorCode.EXCHANGE_GENERIC_DENOMINATION_EXPIRED
+ refreshGroup.statusPerCoin[coinIndex] === RefreshCoinStatus.Failed
) {
- refreshGroup.statusPerCoin[coinIndex] =
- RefreshCoinStatus.PendingRedenominate;
- refreshGroup.operationStatus =
- RefreshOperationStatus.PendingRedenominate;
- delete refreshGroup.timestampFinished;
- changed = true;
+ const rs = await tx.refreshSessions.get([
+ refreshGroup.refreshGroupId,
+ coinIndex,
+ ]);
+ if (
+ rs?.lastError?.code ===
+ TalerErrorCode.EXCHANGE_GENERIC_DENOMINATION_EXPIRED
+ ) {
+ refreshGroup.statusPerCoin[coinIndex] =
+ RefreshCoinStatus.PendingRedenominate;
+ refreshGroup.operationStatus =
+ RefreshOperationStatus.PendingRedenominate;
+ delete refreshGroup.timestampFinished;
+ changed = true;
+ }
+ }
+ if (changed) {
+ await tx.refreshGroups.put(refreshGroup);
}
}
- if (changed) {
- await tx.refreshGroups.put(refreshGroup);
- }
- }
- }
+ });
}
async function fixup20260203DenomFamilyMigration(
@@ -2021,24 +2030,17 @@ async function fixup20260718TransactionsScope(
async function fixupCoinAvailabilityExchangePub(
tx: WalletIndexedDbTransaction,
): Promise<void> {
- const cars = await tx.coinAvailability.getAll();
- const exchanges: Record<string, WalletExchangeDetails | undefined> = {};
- for (const car of cars) {
+ await tx.coinAvailability.iter().forEachAsync(async (car) => {
if (car.exchangeMasterPub === undefined) {
- if (exchanges[car.exchangeBaseUrl] === undefined) {
- exchanges[car.exchangeBaseUrl] =
- await tx.exchangeDetails.indexes.byExchangeBaseUrl.get(
- car.exchangeBaseUrl,
- );
- }
-
- const exchange = exchanges[car.exchangeBaseUrl];
+ const exchange = await tx.exchangeDetails.indexes.byExchangeBaseUrl.get(
+ car.exchangeBaseUrl,
+ );
if (exchange !== undefined) {
car.exchangeMasterPub = exchange.masterPublicKey;
await tx.coinAvailability.put(car);
}
}
- }
+ });
}
/**
@@ -2052,12 +2054,12 @@ async function fixupCoinAvailabilityExchangePub(
async function fixup20260720ExchangeDetailsTinyAmount(
tx: WalletIndexedDbTransaction,
): Promise<void> {
- for (const det of await tx.exchangeDetails.getAll()) {
+ await tx.exchangeDetails.iter().forEachAsync(async (det) => {
if ((det as any).tinyAmount === undefined) {
det.tinyAmount = `${det.currency}:0.01` as AmountString;
await tx.exchangeDetails.put(det);
}
- }
+ });
}
/**
@@ -2071,7 +2073,7 @@ async function fixup20260720ExchangeDetailsTinyAmount(
async function fixup20260720RefreshGroupRefundRequests(
tx: WalletIndexedDbTransaction,
): Promise<void> {
- for (const rg of await tx.refreshGroups.getAll()) {
+ await tx.refreshGroups.iter().forEachAsync(async (rg) => {
let changed = false;
if ((rg as any).refundRequests === undefined) {
rg.refundRequests = {};
@@ -2094,7 +2096,7 @@ async function fixup20260720RefreshGroupRefundRequests(
if (changed) {
await tx.refreshGroups.put(rg);
}
- }
+ });
}
/**
@@ -2107,69 +2109,142 @@ async function fixup20260720RefreshGroupRefundRequests(
* exchange's current merge reserve and peer-pull-credit merge reserves) are
* remapped onto it.
*
- * Rows are only removed when reservePub AND reservePriv both match the kept
- * row: two reserves sharing a pub with different privs would be actual
- * corruption, which deleting would paper over, so those are left alone.
+ * Rows are removed only when every field except rowId matches the kept row.
+ * Any disagreement is corruption or ambiguity, which deleting would paper
+ * over, so those rows are left for conversion to reject.
*/
async function fixup20260720DuplicateReserves(
tx: WalletIndexedDbTransaction,
): Promise<void> {
- const reserves = await tx.reserves.getAll();
- // rowId of every dropped duplicate -> rowId of the row it duplicates.
- const remap = new Map<number, number>();
- const keptByPub = new Map<string, (typeof reserves)[number]>();
- for (const r of reserves) {
- if (r.rowId == null) continue;
- const kept = keptByPub.get(r.reservePub);
- if (!kept) {
- keptByPub.set(r.reservePub, r);
- continue;
- }
- if (kept.reservePriv === r.reservePriv && kept.rowId != null) {
- remap.set(r.rowId, kept.rowId);
+ let kept: WalletReserve | undefined;
+ const canonical = (value: any): any => {
+ if (Array.isArray(value)) return value.map(canonical);
+ if (value !== null && typeof value === "object") {
+ return Object.fromEntries(
+ Object.keys(value)
+ .sort()
+ .map((key) => [key, canonical(value[key])]),
+ );
}
- }
- if (remap.size === 0) {
- return;
- }
- for (const e of await tx.exchanges.getAll()) {
- if (
- e.currentMergeReserveRowId != null &&
- remap.has(e.currentMergeReserveRowId)
- ) {
- e.currentMergeReserveRowId = remap.get(e.currentMergeReserveRowId)!;
- await tx.exchanges.put(e);
+ return value;
+ };
+ const withoutRowId = (r: WalletReserve): string => {
+ const { rowId: _rowId, ...rest } = r;
+ return JSON.stringify(canonical(rest));
+ };
+ // The index groups equal public keys, so only the retained row for the
+ // current key is kept in memory. References are remapped immediately,
+ // avoiding a map proportional to the reserve store.
+ await tx.reserves.indexes.byReservePub.iter().forEachAsync(async (r) => {
+ if (r.rowId == null) return;
+ if (!kept || kept.reservePub !== r.reservePub) {
+ kept = r;
+ return;
}
- }
- for (const p of await tx.peerPullCredit.getAll()) {
- if (remap.has(p.mergeReserveRowId)) {
- p.mergeReserveRowId = remap.get(p.mergeReserveRowId)!;
- await tx.peerPullCredit.put(p);
+ if (kept.rowId == null || withoutRowId(kept) !== withoutRowId(r)) {
+ return;
}
- }
- for (const droppedRowId of remap.keys()) {
+ const droppedRowId = r.rowId;
+ const keptRowId = kept.rowId;
+ await tx.exchanges.iter().forEachAsync(async (e) => {
+ if (e.currentMergeReserveRowId === droppedRowId) {
+ e.currentMergeReserveRowId = keptRowId;
+ await tx.exchanges.put(e);
+ }
+ });
+ await tx.peerPullCredit.iter().forEachAsync(async (p) => {
+ if (p.mergeReserveRowId === droppedRowId) {
+ p.mergeReserveRowId = keptRowId;
+ await tx.peerPullCredit.put(p);
+ }
+ });
await tx.reserves.delete(droppedRowId);
- }
+ });
+}
+
+async function fixup20260812ExchangeWithdrawValues(
+ tx: WalletIndexedDbTransaction,
+): Promise<void> {
+ await tx.coins.iter().forEachAsync(async (coin) => {
+ if (coin.exchangeWithdrawValues === undefined) {
+ coin.exchangeWithdrawValues = { cipher: "RSA" } as any;
+ await tx.coins.put(coin);
+ }
+ });
+ await tx.planchets.iter().forEachAsync(async (planchet) => {
+ if (planchet.exchangeWithdrawValues === undefined) {
+ planchet.exchangeWithdrawValues = { cipher: "RSA" } as any;
+ await tx.planchets.put(planchet);
+ }
+ });
}
export async function applyFixups(
db: DbAccess<typeof WalletIndexedDbStoresV1>,
+ onProgress: (notification: WalletNotification) => void = () => {},
): Promise<number> {
logger.trace("applying fixups");
let count = 0;
- for (const fixupInstruction of walletDbFixups) {
- await db.runAllStoresReadWriteTx({}, async (tx) => {
- logger.trace(`checking fixup ${fixupInstruction.name}`);
- const fixupRecord = await tx.fixups.get(fixupInstruction.name);
- if (fixupRecord) {
- return;
+ for (let index = 0; index < walletDbFixups.length; index++) {
+ const fixupInstruction = walletDbFixups[index];
+ 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) {
+ return;
+ }
+ applied = true;
+ logger.trace(`applying DB fixup ${fixupInstruction.name}`);
+ onProgress({
+ type: NotificationType.DatabaseMaintenanceProgress,
+ operation: "indexeddb-fixup",
+ phase: "fixup",
+ step: fixupInstruction.name,
+ completedSteps: index,
+ totalSteps: walletDbFixups.length,
+ });
+ await fixupInstruction.fn(tx);
+ await tx.fixups.put({
+ fixupName: fixupInstruction.name,
+ });
+ });
+ } catch (e) {
+ if (applied) {
+ onProgress({
+ type: NotificationType.DatabaseMaintenanceProgress,
+ operation: "indexeddb-fixup",
+ phase: "failed",
+ step: fixupInstruction.name,
+ completedSteps: index,
+ totalSteps: walletDbFixups.length,
+ });
}
- logger.trace(`applying DB fixup ${fixupInstruction.name}`);
- await fixupInstruction.fn(tx);
- await tx.fixups.put({
- fixupName: fixupInstruction.name,
+ throw e;
+ }
+ if (applied) {
+ // Announce completion only after the transaction has committed. A
+ // commit error above produces "failed", never a misleading completed
+ // step followed by a rollback.
+ onProgress({
+ type: NotificationType.DatabaseMaintenanceProgress,
+ operation: "indexeddb-fixup",
+ phase: "fixup",
+ step: fixupInstruction.name,
+ completedSteps: index + 1,
+ totalSteps: walletDbFixups.length,
});
count++;
+ }
+ }
+ if (count > 0) {
+ onProgress({
+ type: NotificationType.DatabaseMaintenanceProgress,
+ operation: "indexeddb-fixup",
+ phase: "complete",
+ completedSteps: walletDbFixups.length,
+ totalSteps: walletDbFixups.length,
});
}
return count;
diff --git a/packages/taler-wallet-core/src/db-native-migration.test.ts b/packages/taler-wallet-core/src/db-native-migration.test.ts
@@ -26,6 +26,9 @@
*/
import assert from "node:assert";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
import { test } from "node:test";
import {
@@ -34,16 +37,25 @@ import {
Sqlite3Database,
} from "@gnu-taler/idb-bridge";
import { createNodeHelperSqlite3Impl } from "@gnu-taler/idb-bridge/node-helper-sqlite3-impl";
+import { NotificationType, WalletNotification } from "@gnu-taler/taler-util";
import {
dropExpiredMigrationBackup,
inspectWalletDbFile,
+ inspectWalletDbFileDetails,
migrateWalletDbToNative,
readNativeMigrationInfo,
+ resolveAmbiguousWalletDb,
restoreMigrationBackup,
} from "./db-native-migration.js";
import { IDB_BACKUP_PREFIX, IDB_EMULATION_TABLES } from "./db-sqlite-schema.js";
import { IdbWalletDbHandle } from "./dbtx-handle-impl.js";
+import { openNativeSqliteWalletDb } from "./dbtx-sqlite.js";
+import {
+ inspectWalletDbPath,
+ resolveWalletDbMigration,
+} from "./host-impl.node.js";
+import { acquireSqliteWalletDbOwnership } from "./host-common.js";
import { conformanceCases } from "./dbtx-conformance-cases.js";
import { ConformanceAsserts } from "./dbtx-conformance.js";
@@ -77,12 +89,12 @@ async function countRows(db: Sqlite3Database, table: string): Promise<number> {
* An emulation-backed wallet database with the conformance corpus in it, over
* a connection the caller keeps: the migration needs that same connection.
*/
-async function makePopulatedIdbDb(): Promise<{
+async function makePopulatedIdbDb(filename = ":memory:"): Promise<{
db: Sqlite3Database;
handle: IdbWalletDbHandle;
}> {
const imp = await createNodeHelperSqlite3Impl({ enableTracing: false });
- const db = await imp.open(":memory:");
+ const db = await imp.open(filename);
const backend = await createSqliteBackendOverDb(imp, db);
BridgeIDBFactory.enableTracing = false;
const handle = new IdbWalletDbHandle(new BridgeIDBFactory(backend));
@@ -98,8 +110,59 @@ async function makePopulatedIdbDb(): Promise<{
return { db, handle };
}
+async function makeMinimalIdbDb(): Promise<{
+ db: Sqlite3Database;
+ handle: IdbWalletDbHandle;
+}> {
+ const imp = await createNodeHelperSqlite3Impl({ enableTracing: false });
+ const db = await imp.open(":memory:");
+ const backend = await createSqliteBackendOverDb(imp, db);
+ const handle = new IdbWalletDbHandle(new BridgeIDBFactory(backend));
+ await handle.ensureOpen();
+ await handle.runReadWriteTx((tx) =>
+ tx.upsertConfig({ key: "fault-test" as any, value: 1 }),
+ );
+ return { db, handle };
+}
+
+test("wallet database ownership excludes another SQLite connection", async () => {
+ const directory = fs.mkdtempSync(path.join(os.tmpdir(), "wallet-db-owner-"));
+ const filename = path.join(directory, "wallet.sqlite3");
+ const firstImpl = await createNodeHelperSqlite3Impl({ enableTracing: false });
+ const secondImpl = await createNodeHelperSqlite3Impl({
+ enableTracing: false,
+ });
+ const first = await firstImpl.open(filename);
+ const second = await secondImpl.open(filename);
+ try {
+ await acquireSqliteWalletDbOwnership(first);
+ // Native initialization switches to WAL; ownership must survive that
+ // transition because migrated wallets use WAL for their whole lifetime.
+ await openNativeSqliteWalletDb(first);
+ // Keep this conflict test fast; production uses the adapter's normal busy
+ // timeout so a wallet that is just closing can drain cleanly.
+ await second.exec("PRAGMA busy_timeout = 1");
+ await assert.rejects(
+ acquireSqliteWalletDbOwnership(second),
+ /another wallet process may still be using it/,
+ );
+
+ await first.close();
+ await acquireSqliteWalletDbOwnership(second);
+ } finally {
+ // first.close() is intentionally reached in the success path above. A
+ // second close is harmless for the node helper and ensures failure paths
+ // do not retain the test database lock.
+ await first.close().catch(() => {});
+ await second.close().catch(() => {});
+ fs.rmSync(directory, { recursive: true, force: true });
+ }
+});
+
test("native migration: happens in the same file and switches it over", async () => {
const { db, handle } = await makePopulatedIdbDb();
+ const progress: WalletNotification[] = [];
+ handle.setNotificationSink((n) => progress.push(n));
assert.strictEqual(await inspectWalletDbFile(db), "indexeddb");
@@ -118,6 +181,15 @@ test("native migration: happens in the same file and switches it over", async ()
assert.strictEqual(info.recordsCopied, report.totalRecords);
assert.strictEqual(info.backupStatus, "retained");
assert.ok(info.backupExpiresAt! > info.finishedAt!);
+ assert.ok(
+ progress.some(
+ (n) =>
+ n.type === NotificationType.DatabaseMaintenanceProgress &&
+ n.operation === "indexeddb-to-native-migration" &&
+ n.phase === "complete",
+ ),
+ "successful migration did not report completion",
+ );
// The file now opens natively, without being told to.
assert.strictEqual(await inspectWalletDbFile(db), "native");
@@ -261,3 +333,220 @@ test("native migration: an interrupted attempt keeps the old database", async ()
assert.strictEqual(await inspectWalletDbFile(db), "native");
await native.close();
});
+
+test("native migration: mixed schemas without ownership fail closed", async () => {
+ const { db, handle } = await makePopulatedIdbDb();
+ await openNativeSqliteWalletDb(db);
+ await (
+ await db.prepare(
+ "INSERT INTO config (key, value) VALUES ('native-only', '\"keep-me\"')",
+ )
+ ).run({});
+
+ const inspection = await inspectWalletDbFileDetails(db);
+ assert.strictEqual(inspection.kind, "ambiguous");
+ assert.ok(inspection.indexedDbRecords > 0);
+ assert.strictEqual(inspection.nativeRecords, 1);
+ await assert.rejects(
+ () => migrateWalletDbToNative(db, handle),
+ /native schema already contains wallet records/,
+ );
+ assert.strictEqual(await countRows(db, "config"), 1);
+ assert.ok((await countRows(db, "object_data")) > 0);
+});
+
+test("native migration: an untrusted running marker never clears native rows", async () => {
+ const { db, handle } = await makePopulatedIdbDb();
+ await openNativeSqliteWalletDb(db);
+ await (
+ await db.prepare(
+ "INSERT INTO config (key, value) VALUES ('native-only', '\"keep-me\"')",
+ )
+ ).run({});
+ await (
+ await db.prepare(
+ "INSERT INTO idb_migration (id, status, started_at, cleanup_safe)" +
+ " VALUES (1, 'running', 1, 0)",
+ )
+ ).run({});
+
+ assert.strictEqual(await inspectWalletDbFile(db), "ambiguous");
+ await assert.rejects(
+ () => migrateWalletDbToNative(db, handle),
+ /untrusted running marker/,
+ );
+ assert.strictEqual(await countRows(db, "config"), 1);
+});
+
+test("native migration: an empty untrusted retry acquires cleanup ownership", async () => {
+ const { db, handle } = await makePopulatedIdbDb();
+ await openNativeSqliteWalletDb(db);
+ await (
+ await db.prepare(
+ "INSERT INTO idb_migration (id, status, started_at, cleanup_safe)" +
+ " VALUES (1, 'running', 1, 0)",
+ )
+ ).run({});
+ const { handle: native, info } = await migrateWalletDbToNative(db, handle);
+ assert.strictEqual(info.cleanupSafe, true);
+ assert.strictEqual((await readNativeMigrationInfo(db))?.cleanupSafe, true);
+ await native.close();
+});
+
+test("native migration: unrelated emulated databases are not wallet records", async () => {
+ const imp = await createNodeHelperSqlite3Impl({ enableTracing: false });
+ const db = await imp.open(":memory:");
+ const backend = await createSqliteBackendOverDb(imp, db);
+ const factory = new BridgeIDBFactory(backend);
+ const req = factory.open("not-the-wallet", 1);
+ req.addEventListener("upgradeneeded", () => {
+ req.result.createObjectStore("records").put({ unrelated: true }, "one");
+ });
+ await new Promise<void>((resolve, reject) => {
+ req.addEventListener("success", () => resolve());
+ req.addEventListener("error", () => reject(req.error));
+ });
+ const inspection = await inspectWalletDbFileDetails(db);
+ assert.strictEqual(inspection.indexedDbRecords, 0);
+ assert.strictEqual(inspection.kind, "indexeddb");
+ await db.close();
+});
+
+test("native migration: explicit resolution can keep IndexedDB", async () => {
+ const { db } = await makePopulatedIdbDb();
+ await openNativeSqliteWalletDb(db);
+ await (
+ await db.prepare(
+ "INSERT INTO config (key, value) VALUES ('native-only', '\"discard\"')",
+ )
+ ).run({});
+ await resolveAmbiguousWalletDb(db, "indexeddb");
+ assert.strictEqual(await inspectWalletDbFile(db), "indexeddb");
+ assert.strictEqual(await countRows(db, "config"), 0);
+ assert.strictEqual(
+ (await readNativeMigrationInfo(db))?.status,
+ "rolled-back",
+ );
+ assert.ok((await countRows(db, "object_data")) > 0);
+ await db.close();
+});
+
+test("native migration: explicit resolution can keep native", async () => {
+ const { db } = await makePopulatedIdbDb();
+ await openNativeSqliteWalletDb(db);
+ await (
+ await db.prepare(
+ "INSERT INTO config (key, value) VALUES ('native-only', '\"keep\"')",
+ )
+ ).run({});
+ await resolveAmbiguousWalletDb(db, "native");
+ assert.strictEqual(await inspectWalletDbFile(db), "native");
+ assert.strictEqual(await countRows(db, "config"), 1);
+ assert.ok((await countRows(db, `${IDB_BACKUP_PREFIX}object_data`)) > 0);
+ await db.close();
+});
+
+test("native migration: rollback preflight failure preserves native rows", async () => {
+ const { db, handle } = await makePopulatedIdbDb();
+ const { handle: native } = await migrateWalletDbToNative(db, handle);
+ const coinsBefore = await countRows(db, "coins");
+ await (
+ await db.prepare(`DROP TABLE "${IDB_BACKUP_PREFIX}index_data"`)
+ ).run({});
+ await assert.rejects(
+ () => restoreMigrationBackup(db),
+ /backup table .* is missing/,
+ );
+ assert.strictEqual(await countRows(db, "coins"), coinsBefore);
+ assert.strictEqual((await readNativeMigrationInfo(db))?.status, "complete");
+ await native.close();
+});
+
+test("native migration: every rollback mutation failure is atomic", async () => {
+ // Six backup renames plus the final status update: failing any one must roll
+ // the earlier deletions/renames back with the native wallet and its complete
+ // marker intact.
+ for (let failAt = 0; failAt <= IDB_EMULATION_TABLES.length; failAt++) {
+ const { db, handle } = await makeMinimalIdbDb();
+ await migrateWalletDbToNative(db, handle);
+ let mutation = 0;
+ const faultDb: Sqlite3Database = {
+ internalDbHandle: db.internalDbHandle,
+ exec: (sql) => db.exec(sql),
+ close: () => db.close(),
+ prepare: async (sql) => {
+ const stmt = await db.prepare(sql);
+ return {
+ ...stmt,
+ run: async (params) => {
+ if (
+ sql.startsWith('ALTER TABLE "idb_backup_') ||
+ sql.startsWith(
+ "UPDATE idb_migration SET backup_status = 'restored'",
+ )
+ ) {
+ if (mutation++ === failAt) {
+ throw Error(`injected rollback failure ${failAt}`);
+ }
+ }
+ return await stmt.run(params);
+ },
+ };
+ },
+ };
+ await assert.rejects(
+ () => restoreMigrationBackup(faultDb),
+ new RegExp(`injected rollback failure ${failAt}`),
+ );
+ assert.strictEqual(await countRows(db, "config"), 1);
+ assert.strictEqual((await readNativeMigrationInfo(db))?.status, "complete");
+ const tables = await listTables(db);
+ for (const table of IDB_EMULATION_TABLES) {
+ assert.ok(tables.includes(`${IDB_BACKUP_PREFIX}${table}`));
+ assert.ok(!tables.includes(table));
+ }
+ await db.close();
+ }
+});
+
+test("native migration: offline resolution creates the mandatory full backup", async () => {
+ const dir = fs.mkdtempSync(path.join(os.tmpdir(), "taler-db-resolution-"));
+ const dbPath = path.join(dir, "wallet.sqlite3");
+ const backupPath = path.join(dir, "before.sqlite3");
+ try {
+ const { db, handle } = await makePopulatedIdbDb(dbPath);
+ await openNativeSqliteWalletDb(db);
+ await (
+ await db.prepare(
+ "INSERT INTO config (key, value) VALUES ('native-only', '\"discard\"')",
+ )
+ ).run({});
+ await handle.close();
+ await db.close();
+
+ await assert.rejects(
+ () =>
+ resolveWalletDbMigration(
+ dbPath,
+ "indexeddb",
+ path.join(dir, "missing", "backup.sqlite3"),
+ ),
+ /unable to open database|cannot open|SQLITE_CANTOPEN/i,
+ );
+ assert.strictEqual((await inspectWalletDbPath(dbPath)).kind, "ambiguous");
+
+ await resolveWalletDbMigration(dbPath, "indexeddb", backupPath);
+ assert.ok(fs.statSync(backupPath).size > 0);
+ assert.strictEqual((await inspectWalletDbPath(dbPath)).kind, "indexeddb");
+ assert.strictEqual(
+ (await inspectWalletDbPath(backupPath)).kind,
+ "ambiguous",
+ );
+ await assert.rejects(
+ () => resolveWalletDbMigration(dbPath, "native", backupPath),
+ /backup destination .* already exists/,
+ );
+ } finally {
+ fs.rmSync(dir, { recursive: true, force: true });
+ }
+});
diff --git a/packages/taler-wallet-core/src/db-native-migration.ts b/packages/taler-wallet-core/src/db-native-migration.ts
@@ -45,15 +45,24 @@
* away. {@link restoreMigrationBackup} is that statement.
*/
-import { Duration, Logger } from "@gnu-taler/taler-util";
+import { Duration, Logger, NotificationType } from "@gnu-taler/taler-util";
import type { Sqlite3Database } from "@gnu-taler/idb-bridge";
-import { convertWalletDb, DbConversionReport } from "./db-converter.js";
-import { IDB_BACKUP_PREFIX, IDB_EMULATION_TABLES } from "./db-sqlite-schema.js";
+import {
+ convertWalletDb,
+ DB_CONVERSION_STEP_COUNT,
+ DbConversionReport,
+} from "./db-converter.js";
+import {
+ IDB_BACKUP_PREFIX,
+ IDB_EMULATION_TABLES,
+ NATIVE_DATA_TABLES,
+} from "./db-sqlite-schema.js";
import { SqliteWalletDbHandle } from "./dbtx-handle-impl.js";
import { WalletDbHandle } from "./dbtx-handle.js";
import {
clearNativeSqliteWalletDb,
+ clearNativeSqliteWalletDbInTransaction,
openNativeSqliteWalletDb,
SqliteTxControl,
} from "./dbtx-sqlite.js";
@@ -80,7 +89,7 @@ function retentionMicros(): number {
}
/** Which schema the records in a wallet database file are stored in. */
-export type WalletDbFileKind = "empty" | "indexeddb" | "native";
+export type WalletDbFileKind = "empty" | "indexeddb" | "native" | "ambiguous";
/**
* 'rolled-back' is terminal: the emulation tables were put back by
@@ -99,6 +108,15 @@ export interface NativeMigrationInfo {
recordsCopied?: number;
backupStatus?: MigrationBackupStatus;
backupExpiresAt?: number;
+ /** Whether native rows are known to be only a disposable partial copy. */
+ cleanupSafe?: boolean;
+}
+
+export interface WalletDbFileInspection {
+ kind: WalletDbFileKind;
+ indexedDbRecords: number;
+ nativeRecords: number;
+ ambiguityReason?: string;
}
/** Current time in the microseconds the native schema's timestamps use. */
@@ -151,9 +169,82 @@ export async function readNativeMigrationInfo(
| MigrationBackupStatus
| undefined,
backupExpiresAt: optNum(row.backup_expires_at),
+ cleanupSafe:
+ row.cleanup_safe == null ? undefined : Number(row.cleanup_safe) === 1,
};
}
+async function countRows(db: Sqlite3Database, table: string): Promise<number> {
+ const row = await (
+ await db.prepare(`SELECT COUNT(*) AS n FROM "${table}"`)
+ ).getFirst({});
+ return Number(row?.n ?? 0);
+}
+
+async function countNativeRecords(db: Sqlite3Database): Promise<number> {
+ let total = 0;
+ for (const table of NATIVE_DATA_TABLES) {
+ if (await tableExists(db, table)) total += await countRows(db, table);
+ }
+ return total;
+}
+
+async function countMainIndexedDbRecords(db: Sqlite3Database): Promise<number> {
+ if (
+ !(await tableExists(db, "object_data")) ||
+ !(await tableExists(db, "object_stores"))
+ ) {
+ return 0;
+ }
+ const row = await (
+ 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'",
+ )
+ ).getFirst({});
+ return Number(row?.n ?? 0);
+}
+
+export async function inspectWalletDbFileDetails(
+ db: Sqlite3Database,
+): Promise<WalletDbFileInspection> {
+ const info = await readNativeMigrationInfo(db);
+ const indexedDbRecords = await countMainIndexedDbRecords(db);
+ const nativeRecords = await countNativeRecords(db);
+ const result = (kind: WalletDbFileKind, ambiguityReason?: string) => ({
+ kind,
+ indexedDbRecords,
+ nativeRecords,
+ ...(ambiguityReason ? { ambiguityReason } : undefined),
+ });
+
+ if (info?.status === "complete") return result("native");
+ if (info?.status === "running" && info.cleanupSafe)
+ return result("indexeddb");
+ if (info?.status === "running" && nativeRecords > 0) {
+ return result(
+ "ambiguous",
+ "an untrusted running migration marker coexists with native wallet records",
+ );
+ }
+ if (info?.status === "rolled-back" && nativeRecords > 0) {
+ return result(
+ "ambiguous",
+ "a rolled-back migration still has native wallet records",
+ );
+ }
+ if (indexedDbRecords > 0 && nativeRecords > 0) {
+ return result(
+ "ambiguous",
+ "both IndexedDB and native schemas contain wallet records without a trustworthy authority marker",
+ );
+ }
+ if (await tableExists(db, "object_data")) return result("indexeddb");
+ if (await tableExists(db, "schema_migrations")) return result("native");
+ return result("empty");
+}
+
/**
* Decide which schema holds the wallet's records in an open database file.
*
@@ -164,19 +255,7 @@ export async function readNativeMigrationInfo(
export async function inspectWalletDbFile(
db: Sqlite3Database,
): Promise<WalletDbFileKind> {
- const info = await readNativeMigrationInfo(db);
- if (info?.status === "complete") {
- return "native";
- }
- // Before the renames, the emulation's tables are authoritative even when a
- // partial native copy exists next to them.
- if (await tableExists(db, "object_data")) {
- return "indexeddb";
- }
- if (await tableExists(db, "schema_migrations")) {
- return "native";
- }
- return "empty";
+ return (await inspectWalletDbFileDetails(db)).kind;
}
/**
@@ -242,88 +321,133 @@ export async function migrateWalletDbToNative(
);
}
if (previous?.status === "running") {
- // A previous attempt died before the renames. Its partial copy is in the
- // native tables and nothing references it, so it goes.
- logger.warn(
- "discarding the partial copy left by an interrupted migration attempt",
+ if (previous.cleanupSafe) {
+ logger.warn(
+ "discarding the cleanup-safe partial copy left by an interrupted migration attempt",
+ );
+ await clearNativeSqliteWalletDb(ndb);
+ } else if ((await countNativeRecords(db)) !== 0) {
+ throw Error(
+ "migration refused: an untrusted running marker has native wallet records; use db-migration-resolve after making a backup",
+ );
+ }
+ } else if ((await countNativeRecords(db)) !== 0) {
+ throw Error(
+ "migration refused: the native schema already contains wallet records; use db-migration-resolve after making a backup",
);
- await clearNativeSqliteWalletDb(ndb);
}
const startedAt = nowMicros();
await ndb.lock.run(() =>
inTransaction(txc, async () => {
+ if ((await countNativeRecords(db)) !== 0) {
+ throw Error(
+ "migration refused: native wallet records appeared before cleanup ownership could be recorded",
+ );
+ }
await (
await db.prepare(
- "INSERT INTO idb_migration (id, status, started_at)" +
- " VALUES (1, 'running', $started_at)" +
+ "INSERT INTO idb_migration (id, status, started_at, cleanup_safe)" +
+ " VALUES (1, 'running', $started_at, 1)" +
" ON CONFLICT (id) DO UPDATE SET status = 'running'," +
" started_at = $started_at, finished_at = NULL," +
" records_copied = NULL, backup_status = NULL," +
- " backup_expires_at = NULL",
+ " backup_expires_at = NULL, cleanup_safe = 1",
)
).run({ started_at: startedAt });
}),
);
- 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);
+ 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);
- const finishedAt = nowMicros();
- const backupExpiresAt = finishedAt + retentionMicros();
+ await ndb.lock.run(async () => {
+ const violations = await (
+ await db.prepare("PRAGMA foreign_key_check")
+ ).getAll({});
+ if (violations.length !== 0) {
+ throw Error(
+ `migration refused: native foreign-key validation found ${violations.length} violation(s)`,
+ );
+ }
+ });
- // One transaction: the renames and the record of them being done cannot come
- // apart. A crash between them would leave a file whose emulation tables are
- // gone and whose bookkeeping still says the emulation is authoritative, and
- // the retry would then wipe the only remaining copy.
- await ndb.lock.run(() =>
- inTransaction(txc, async () => {
- for (const table of IDB_EMULATION_TABLES) {
+ const finishedAt = nowMicros();
+ const backupExpiresAt = finishedAt + retentionMicros();
+
+ // One transaction: the renames and the record of them being done cannot come
+ // apart. A crash between them would leave a file whose emulation tables are
+ // gone and whose bookkeeping still says the emulation is authoritative, and
+ // the retry would then wipe the only remaining copy.
+ await ndb.lock.run(() =>
+ inTransaction(txc, async () => {
+ for (const table of IDB_EMULATION_TABLES) {
+ await (
+ await db.prepare(
+ `ALTER TABLE "${table}" RENAME TO "${backupTableName(table)}"`,
+ )
+ ).run({});
+ }
await (
await db.prepare(
- `ALTER TABLE "${table}" RENAME TO "${backupTableName(table)}"`,
+ "UPDATE idb_migration SET status = 'complete'," +
+ " finished_at = $finished_at, records_copied = $records_copied," +
+ " backup_status = 'retained'," +
+ " backup_expires_at = $backup_expires_at WHERE id = 1",
)
- ).run({});
- }
- await (
- await db.prepare(
- "UPDATE idb_migration SET status = 'complete'," +
- " finished_at = $finished_at, records_copied = $records_copied," +
- " backup_status = 'retained'," +
- " backup_expires_at = $backup_expires_at WHERE id = 1",
- )
- ).run({
- finished_at: finishedAt,
- records_copied: report.totalRecords,
- backup_expires_at: backupExpiresAt,
- });
- }),
- );
+ ).run({
+ finished_at: finishedAt,
+ records_copied: report.totalRecords,
+ backup_expires_at: backupExpiresAt,
+ });
+ }),
+ );
- logger.info(
- `migrated ${report.totalRecords} records to the native schema;` +
- ` the previous database is kept in this file until` +
- ` ${new Date(backupExpiresAt / 1000).toISOString()}`,
- );
+ logger.info(
+ `migrated ${report.totalRecords} records to the native schema;` +
+ ` the previous database is kept in this file until` +
+ ` ${new Date(backupExpiresAt / 1000).toISOString()}`,
+ );
- // The bookkeeping is reported from what was just written rather than read
- // back: past the transaction above the emulation's tables are gone, so a
- // caller that treats a throw as "nothing happened, keep using the old
- // handle" would be wrong from here on. Nothing after this can throw.
- return {
- handle: dst,
- report,
- info: {
- status: "complete",
- startedAt,
- finishedAt,
- recordsCopied: report.totalRecords,
- backupStatus: "retained",
- backupExpiresAt,
- },
- };
+ // The bookkeeping is reported from what was just written rather than read
+ // back: past the transaction above the emulation's tables are gone, so a
+ // caller that treats a throw as "nothing happened, keep using the old
+ // handle" would be wrong from here on. Nothing after this can throw.
+ src.emitNotification({
+ type: NotificationType.DatabaseMaintenanceProgress,
+ operation: "indexeddb-to-native-migration",
+ phase: "complete",
+ completedSteps: DB_CONVERSION_STEP_COUNT,
+ totalSteps: DB_CONVERSION_STEP_COUNT,
+ processedRecords: report.totalRecords,
+ totalRecords: report.totalRecords,
+ });
+ return {
+ handle: dst,
+ report,
+ info: {
+ status: "complete",
+ startedAt,
+ finishedAt,
+ recordsCopied: report.totalRecords,
+ backupStatus: "retained",
+ backupExpiresAt,
+ cleanupSafe: true,
+ },
+ };
+ } catch (e) {
+ src.emitNotification({
+ type: NotificationType.DatabaseMaintenanceProgress,
+ operation: "indexeddb-to-native-migration",
+ phase: "failed",
+ completedSteps: 0,
+ totalSteps: DB_CONVERSION_STEP_COUNT,
+ });
+ throw e;
+ }
}
/**
@@ -344,19 +468,38 @@ export async function dropExpiredMigrationBackup(
if (info.backupExpiresAt == null || now < info.backupExpiresAt) {
return false;
}
+ // Bring the native schema fully up to date and validate it before deleting
+ // the last pre-migration copy.
+ const ndb = await openNativeSqliteWalletDb(db);
logger.info("dropping the retained pre-migration database tables");
- const txc = await SqliteTxControl.create(db);
- await inTransaction(txc, async () => {
- for (const table of IDB_EMULATION_TABLES) {
+ await ndb.lock.run(async () => {
+ const violations = await (
+ await db.prepare("PRAGMA foreign_key_check")
+ ).getAll({});
+ if (violations.length !== 0) {
+ throw Error(
+ "native database failed foreign-key validation; retained backup was not dropped",
+ );
+ }
+ await inTransaction(ndb.txc, async () => {
+ for (const table of IDB_EMULATION_TABLES) {
+ if (!(await tableExists(db, backupTableName(table)))) {
+ throw Error(
+ `retained backup table ${backupTableName(table)} is missing`,
+ );
+ }
+ }
+ for (const table of IDB_EMULATION_TABLES) {
+ await (
+ await db.prepare(`DROP TABLE "${backupTableName(table)}"`)
+ ).run({});
+ }
await (
- await db.prepare(`DROP TABLE IF EXISTS "${backupTableName(table)}"`)
+ await db.prepare(
+ "UPDATE idb_migration SET backup_status = 'dropped' WHERE id = 1",
+ )
).run({});
- }
- await (
- await db.prepare(
- "UPDATE idb_migration SET backup_status = 'dropped' WHERE id = 1",
- )
- ).run({});
+ });
});
return true;
}
@@ -387,24 +530,134 @@ export async function restoreMigrationBackup(
);
}
const ndb = await openNativeSqliteWalletDb(db);
- // Everything the native tables hold came from the backup or was written
- // after the migration; either way it is not what the restored database is
- // supposed to contain.
- await clearNativeSqliteWalletDb(ndb);
- await inTransaction(ndb.txc, async () => {
- for (const table of IDB_EMULATION_TABLES) {
+ await ndb.lock.run(async () => {
+ await inTransaction(ndb.txc, async () => {
+ // Preflight is inside the same transaction as deletion and renaming, so
+ // every failure leaves the complete native wallet authoritative.
+ for (const table of IDB_EMULATION_TABLES) {
+ if (await tableExists(db, table)) {
+ throw Error(`cannot restore: table ${table} already exists`);
+ }
+ if (!(await tableExists(db, backupTableName(table)))) {
+ throw Error(
+ `cannot restore: backup table ${backupTableName(table)} is missing`,
+ );
+ }
+ }
+ await clearNativeSqliteWalletDbInTransaction(ndb);
+ for (const table of IDB_EMULATION_TABLES) {
+ await (
+ await db.prepare(
+ `ALTER TABLE "${backupTableName(table)}" RENAME TO "${table}"`,
+ )
+ ).run({});
+ }
await (
await db.prepare(
- `ALTER TABLE "${backupTableName(table)}" RENAME TO "${table}"`,
+ "UPDATE idb_migration SET backup_status = 'restored'," +
+ " status = 'rolled-back' WHERE id = 1 AND status = 'complete'",
)
).run({});
- }
- await (
- await db.prepare(
- "UPDATE idb_migration SET backup_status = 'restored'," +
- " status = 'rolled-back' WHERE id = 1",
- )
- ).run({});
+ const updated = await (
+ await db.prepare("SELECT status FROM idb_migration WHERE id = 1")
+ ).getFirst({});
+ if (updated?.status !== "rolled-back") {
+ throw Error(
+ "migration status changed while rollback was being prepared",
+ );
+ }
+ });
});
logger.info("restored the pre-migration wallet database");
}
+
+export type MigrationAuthority = "indexeddb" | "native";
+
+async function requireResolutionTables(db: Sqlite3Database): Promise<void> {
+ for (const table of IDB_EMULATION_TABLES) {
+ if (!(await tableExists(db, table))) {
+ throw Error(
+ `cannot resolve migration: expected IndexedDB table ${table} is missing`,
+ );
+ }
+ if (await tableExists(db, backupTableName(table))) {
+ throw Error(
+ `cannot resolve migration: backup table ${backupTableName(table)} already exists`,
+ );
+ }
+ }
+ for (const table of NATIVE_DATA_TABLES) {
+ if (!(await tableExists(db, table))) {
+ throw Error(
+ `cannot resolve migration: expected native table ${table} is missing`,
+ );
+ }
+ }
+}
+
+/** Select authority in an ambiguous file. The caller must create a backup first. */
+export async function resolveAmbiguousWalletDb(
+ db: Sqlite3Database,
+ keep: MigrationAuthority,
+): Promise<void> {
+ const inspection = await inspectWalletDbFileDetails(db);
+ if (inspection.kind !== "ambiguous") {
+ throw Error(
+ `migration resolution requires an ambiguous database, got ${inspection.kind}`,
+ );
+ }
+ await requireResolutionTables(db);
+ const ndb = await openNativeSqliteWalletDb(db);
+ await ndb.lock.run(async () => {
+ if (keep === "native") {
+ const violations = await (
+ await db.prepare("PRAGMA foreign_key_check")
+ ).getAll({});
+ if (violations.length !== 0) {
+ throw Error(
+ `cannot keep native: foreign-key validation found ${violations.length} violation(s)`,
+ );
+ }
+ }
+ await inTransaction(ndb.txc, async () => {
+ // Repeat preflight under the mutation transaction.
+ await requireResolutionTables(db);
+ const at = nowMicros();
+ if (keep === "indexeddb") {
+ await clearNativeSqliteWalletDbInTransaction(ndb);
+ await (
+ await db.prepare(
+ "INSERT INTO idb_migration" +
+ " (id, status, started_at, finished_at, backup_status, cleanup_safe)" +
+ " VALUES (1, 'rolled-back', $at, $at, 'restored', 1)" +
+ " ON CONFLICT(id) DO UPDATE SET status='rolled-back'," +
+ " finished_at=$at, backup_status='restored', cleanup_safe=1",
+ )
+ ).run({ at });
+ } else {
+ for (const table of IDB_EMULATION_TABLES) {
+ await (
+ await db.prepare(
+ `ALTER TABLE "${table}" RENAME TO "${backupTableName(table)}"`,
+ )
+ ).run({});
+ }
+ await (
+ await db.prepare(
+ "INSERT INTO idb_migration" +
+ " (id, status, started_at, finished_at, records_copied," +
+ " backup_status, backup_expires_at, cleanup_safe)" +
+ " VALUES (1, 'complete', $at, $at, $records, 'retained', $expires, 0)" +
+ " ON CONFLICT(id) DO UPDATE SET status='complete'," +
+ " started_at=$at, finished_at=$at, records_copied=$records," +
+ " backup_status='retained', backup_expires_at=$expires, cleanup_safe=0",
+ )
+ ).run({
+ at,
+ records: inspection.nativeRecords,
+ expires: at + retentionMicros(),
+ });
+ }
+ });
+ });
+}
diff --git a/packages/taler-wallet-core/src/db-sqlite-schema.ts b/packages/taler-wallet-core/src/db-sqlite-schema.ts
@@ -85,7 +85,7 @@
*
* Bump this when adding a migration to {@link schemaMigrations}.
*/
-export const SQLITE_SCHEMA_VERSION = 6;
+export const SQLITE_SCHEMA_VERSION = 7;
/**
* Tables of the IndexedDB emulation, children before parents.
@@ -1317,4 +1317,18 @@ export const schemaMigrations: SchemaMigration[] = [
'ALTER TABLE coins ADD COLUMN exchange_withdraw_values TEXT NOT NULL DEFAULT \'{"cipher":"RSA"}\'',
],
},
+ {
+ version: 7,
+ name: "indexeddb-migration-cleanup-ownership",
+ statements: [
+ "ALTER TABLE idb_migration ADD COLUMN cleanup_safe INTEGER NOT NULL DEFAULT 0 CHECK (cleanup_safe IN (0, 1))",
+ ],
+ },
];
+
+/** Native tables that contain wallet records (not schema bookkeeping). */
+export const NATIVE_DATA_TABLES = [
+ ...SQLITE_BASELINE_SCHEMA.matchAll(/CREATE TABLE IF NOT EXISTS (\w+)/g),
+]
+ .map((m) => m[1])
+ .filter((name) => !NON_DATA_TABLES.includes(name));
diff --git a/packages/taler-wallet-core/src/dbtx-handle-impl.ts b/packages/taler-wallet-core/src/dbtx-handle-impl.ts
@@ -86,6 +86,10 @@ export class IdbWalletDbHandle implements WalletDbHandle {
this.notify = sink;
}
+ emitNotification(notification: WalletNotification): void {
+ this.notify(notification);
+ }
+
constructor(
private idbFactory: BridgeIDBFactory,
/**
@@ -107,7 +111,9 @@ export class IdbWalletDbHandle implements WalletDbHandle {
}
this.idbHandle = await openTalerDatabase(this.idbFactory, async () => {});
this.dbAccess = this.makeAccess();
- const fixupsApplied = await applyFixups(this.dbAccess);
+ const fixupsApplied = await applyFixups(this.dbAccess, (n) =>
+ this.notify(n),
+ );
return { fixupsApplied };
}
@@ -194,7 +200,7 @@ export class IdbWalletDbHandle implements WalletDbHandle {
await tx.fixups.delete(fx.fixupName);
}
});
- await applyFixups(access);
+ await applyFixups(access, (n) => this.notify(n));
}
async clearDatabase(): Promise<void> {
@@ -245,6 +251,10 @@ export class SqliteWalletDbHandle implements WalletDbHandle {
this.notify = sink;
}
+ emitNotification(notification: WalletNotification): void {
+ this.notify(notification);
+ }
+
constructor(private ndb: NativeSqliteWalletDb) {}
/**
diff --git a/packages/taler-wallet-core/src/dbtx-handle.ts b/packages/taler-wallet-core/src/dbtx-handle.ts
@@ -90,6 +90,9 @@ export interface WalletDbHandle {
*/
setNotificationSink(sink: (n: WalletNotification) => void): void;
+ /** Emit non-transactional maintenance progress to the installed sink. */
+ emitNotification(notification: WalletNotification): void;
+
/**
* Copy the database to a file, in whatever format the backend supports.
*
diff --git a/packages/taler-wallet-core/src/dbtx-indexeddb.ts b/packages/taler-wallet-core/src/dbtx-indexeddb.ts
@@ -95,6 +95,8 @@ import type {
StoreCurrencyInfoDbRequest,
WalletCoinAvailabilityRef,
WalletDbTransaction,
+ WalletDbMigrationStore,
+ WalletDbMigrationPage,
WalletDenomRef,
} from "./dbtx.js";
@@ -111,6 +113,67 @@ 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>> {
+ const physicalStore: Record<WalletDbMigrationStore, string> = {
+ config: "config",
+ currencyInfo: "currencyInfo",
+ contacts: "contacts",
+ mailboxMessages: "mailboxMessages",
+ mailboxConfigurations: "mailboxConfigurations",
+ contractTerms: "contractTerms",
+ tombstones: "tombstones",
+ operationRetries: "operationRetries",
+ bankAccounts: "bankAccountsV2",
+ globalCurrencyExchanges: "globalCurrencyExchanges",
+ globalCurrencyAuditors: "globalCurrencyAuditors",
+ exchangeBaseUrlFixups: "exchangeBaseUrlFixups",
+ exchangeBaseUrlMigrationLog: "exchangeBaseUrlMigrationLog",
+ reserves: "reserves",
+ exchanges: "exchanges",
+ exchangeDetails: "exchangeDetails",
+ exchangeSignKeys: "exchangeSignKeys",
+ denominationFamilies: "denominationFamilies",
+ denominations: "denominationsV2",
+ withdrawalGroups: "withdrawalGroups",
+ purchases: "purchases",
+ refreshGroups: "refreshGroups",
+ coins: "coins",
+ planchets: "planchets",
+ refreshSessions: "refreshSessions",
+ coinHistory: "coinHistory",
+ coinAvailability: "coinAvailabilityV2",
+ refundGroups: "refundGroups",
+ tokens: "tokens",
+ slates: "slates",
+ depositGroups: "depositGroups",
+ recoupGroups: "recoupGroups",
+ denomLossEvents: "denomLossEvents",
+ peerPushDebit: "peerPushDebit",
+ peerPushCredit: "peerPushCredit",
+ peerPullDebit: "peerPullDebit",
+ peerPullCredit: "peerPullCredit",
+ donationSummaries: "donationSummaries",
+ donationPlanchets: "donationPlanchets",
+ donationReceipts: "donationReceipts",
+ transactionsMeta: "transactionsMeta",
+ refundItems: "refundItems",
+ };
+ const accessor = (this.tx as any)[physicalStore[store]];
+ if (!accessor) {
+ throw Error(`migration store ${store} is not available`);
+ }
+ const page = await accessor.scan(cursor, limit);
+ return {
+ records: page.records as T[],
+ ...(page.records.length > 0 ? { nextCursor: page.lastKey } : {}),
+ };
+ }
+
scheduleOnCommit(f: () => void): void {
this.tx._util.scheduleOnCommit(f);
}
@@ -811,6 +874,12 @@ export class IdbWalletTransaction implements WalletDbTransaction {
return await tx.exchangeDetails.indexes.byExchangeBaseUrl.getAll();
}
+ async getExchangeDetailsByRowId(
+ rowId: number,
+ ): Promise<WalletExchangeDetails | undefined> {
+ return await this.tx.exchangeDetails.get(rowId);
+ }
+
async upsertExchangeDetails(rec: WalletExchangeDetails): Promise<number> {
const tx = this.tx;
const res = await tx.exchangeDetails.put(rec);
@@ -842,6 +911,10 @@ export class IdbWalletTransaction implements WalletDbTransaction {
]);
}
+ async listAllExchangeSignKeys(): Promise<WalletExchangeSignkeys[]> {
+ return await this.tx.exchangeSignKeys.getAll();
+ }
+
async upsertExchangeSignKey(rec: WalletExchangeSignkeys): Promise<void> {
const tx = this.tx;
await tx.exchangeSignKeys.put(rec);
@@ -982,6 +1055,10 @@ export class IdbWalletTransaction implements WalletDbTransaction {
return await tx.refundItems.indexes.byRefundGroupId.getAll([refundGroupId]);
}
+ async listAllRefundItems(): Promise<WalletRefundItem[]> {
+ return await this.tx.refundItems.getAll();
+ }
+
async upsertRefundItem(rec: WalletRefundItem): Promise<number> {
const tx = this.tx;
const res = await tx.refundItems.put(rec);
@@ -1208,6 +1285,10 @@ export class IdbWalletTransaction implements WalletDbTransaction {
return await tx.planchets.indexes.byGroup.getAll(withdrawalGroupId);
}
+ async listAllPlanchets(): Promise<WalletPlanchet[]> {
+ return await this.tx.planchets.getAll();
+ }
+
async countPlanchetsByGroup(withdrawalGroupId: string): Promise<number> {
const tx = this.tx;
const keys =
@@ -1284,6 +1365,10 @@ export class IdbWalletTransaction implements WalletDbTransaction {
);
}
+ async listAllRefreshSessions(): Promise<WalletRefreshSession[]> {
+ return await this.tx.refreshSessions.getAll();
+ }
+
async getRecoupGroup(
recoupGroupId: string,
): Promise<WalletRecoupGroup | undefined> {
@@ -1384,6 +1469,10 @@ export class IdbWalletTransaction implements WalletDbTransaction {
return await tx.coinHistory.get(coinPub);
}
+ async listAllCoinHistories(): Promise<WalletCoinHistory[]> {
+ return await this.tx.coinHistory.getAll();
+ }
+
async upsertCoinHistory(rec: WalletCoinHistory): Promise<void> {
const tx = this.tx;
await tx.coinHistory.put(rec);
diff --git a/packages/taler-wallet-core/src/dbtx-sqlite.ts b/packages/taler-wallet-core/src/dbtx-sqlite.ts
@@ -58,6 +58,8 @@ import {
WalletCoinAvailabilityRef,
WalletDbRecordCounts,
WalletDbTransaction,
+ WalletDbMigrationPage,
+ WalletDbMigrationStore,
WalletDenomRef,
WalletCurrencyInfoEntry,
} from "./dbtx.js";
@@ -444,6 +446,11 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
*/
private stats: SqliteAccessStats;
+ /** Window applied to the root SELECT of one migration enumeration call. */
+ private migrationPage:
+ | { offset: number; limit: number; consumed: boolean }
+ | undefined;
+
constructor(
private db: Sqlite3Database,
stmtCache?: Map<string, Sqlite3Statement>,
@@ -481,11 +488,53 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
sql: string,
params: Record<string, any> = {},
): Promise<ResultRow[]> {
+ if (this.migrationPage && !this.migrationPage.consumed) {
+ this.migrationPage.consumed = true;
+ sql =
+ `SELECT * FROM (${sql}) AS migration_page` +
+ " LIMIT $migration_limit OFFSET $migration_offset";
+ params = {
+ ...params,
+ migration_limit: this.migrationPage.limit,
+ migration_offset: this.migrationPage.offset,
+ };
+ }
const rows = await (await this.prep(sql)).getAll(params);
this.stats.rowsRead += rows.length;
return rows;
}
+ async scanMigrationRecords<T>(
+ _store: WalletDbMigrationStore,
+ read: (tx: WalletDbTransaction) => Promise<T[]>,
+ cursor: unknown | undefined,
+ limit: number,
+ ): Promise<WalletDbMigrationPage<T>> {
+ const offset = cursor === undefined ? 0 : Number(cursor);
+ if (!Number.isSafeInteger(offset) || offset < 0) {
+ throw Error("invalid sqlite migration cursor");
+ }
+ if (!Number.isSafeInteger(limit) || limit <= 0) {
+ throw Error("migration page size must be a positive integer");
+ }
+ if (this.migrationPage) {
+ throw Error("nested migration scan is not supported");
+ }
+ this.migrationPage = { offset, limit, consumed: false };
+ try {
+ const records = await read(this);
+ if (!this.migrationPage.consumed) {
+ throw Error("migration enumeration did not issue a SELECT");
+ }
+ return {
+ records,
+ ...(records.length > 0 ? { nextCursor: offset + records.length } : {}),
+ };
+ } finally {
+ this.migrationPage = undefined;
+ }
+ }
+
// Bound as an instance property for the same reason as the IndexedDB
// implementation: call sites pass it around unbound.
notify = (notif: WalletNotification): void => {
@@ -1037,6 +1086,11 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
return rows.map((r) => this.rowToRefundItem(r));
}
+ async listAllRefundItems(): Promise<WalletRefundItem[]> {
+ const rows = await this.all("SELECT * FROM refund_items");
+ return rows.map((r) => this.rowToRefundItem(r));
+ }
+
async upsertRefundItem(rec: WalletRefundItem): Promise<number> {
if (rec.id != null) {
await this.run(
@@ -1331,6 +1385,14 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
};
}
+ async listAllCoinHistories(): Promise<WalletCoinHistory[]> {
+ const rows = await this.all("SELECT * FROM coin_history");
+ return rows.map((row) => ({
+ coinPub: dbToCrock(row.coin_pub),
+ history: dbToJson(row.history),
+ }));
+ }
+
async upsertCoinHistory(rec: WalletCoinHistory): Promise<void> {
await this.run(
"INSERT INTO coin_history (coin_pub, history) VALUES ($pub, $h)" +
@@ -1801,6 +1863,16 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
return rows.map((r) => this.rowToExchangeDetails(r));
}
+ async getExchangeDetailsByRowId(
+ rowId: number,
+ ): Promise<WalletExchangeDetails | undefined> {
+ const row = await this.first(
+ "SELECT * FROM exchange_details WHERE row_id = $id",
+ { id: rowId },
+ );
+ return row ? this.rowToExchangeDetails(row) : undefined;
+ }
+
async deleteExchangeDetails(rowId: number): Promise<void> {
await this.run("DELETE FROM exchange_details WHERE row_id = $id", {
id: rowId,
@@ -1840,6 +1912,18 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
}));
}
+ async listAllExchangeSignKeys(): Promise<WalletExchangeSignkeys[]> {
+ const rows = await this.all("SELECT * FROM exchange_sign_keys");
+ return rows.map((row) => ({
+ exchangeDetailsRowId: num(row.exchange_details_row_id),
+ signkeyPub: dbToCrock(row.signkey_pub),
+ stampStart: dbTimestamp(row.stamp_start),
+ stampExpire: dbTimestamp(row.stamp_expire),
+ stampEnd: dbTimestamp(row.stamp_end),
+ masterSig: dbToCrock(row.master_sig),
+ }));
+ }
+
async upsertExchangeSignKey(rec: WalletExchangeSignkeys): Promise<void> {
await this.run(
`INSERT INTO exchange_sign_keys (
@@ -2490,6 +2574,11 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
return rows.map((r) => this.rowToPlanchet(r));
}
+ async listAllPlanchets(): Promise<WalletPlanchet[]> {
+ const rows = await this.all("SELECT * FROM planchets");
+ return rows.map((r) => this.rowToPlanchet(r));
+ }
+
async countPlanchetsByGroup(withdrawalGroupId: string): Promise<number> {
const row = await this.first(
"SELECT COUNT(*) AS n FROM planchets WHERE withdrawal_group_id = $wgid",
@@ -4929,6 +5018,11 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
return rows.map((r) => this.rowToRefreshSession(r));
}
+ async listAllRefreshSessions(): Promise<WalletRefreshSession[]> {
+ const rows = await this.all("SELECT * FROM refresh_sessions");
+ return rows.map((r) => this.rowToRefreshSession(r));
+ }
+
// ----------------------------------------------------- recoup groups
private rowToRecoupGroup(row: ResultRow): WalletRecoupGroup {
@@ -5264,17 +5358,9 @@ export async function clearNativeSqliteWalletDb(
ndb: NativeSqliteWalletDb,
): Promise<void> {
await ndb.lock.run(async () => {
- const rows = await (
- await ndb.db.prepare(
- `SELECT name FROM sqlite_master WHERE ${DATA_TABLES_CONDITION}`,
- )
- ).getAll();
await ndb.txc.begin();
try {
- for (const row of rows) {
- // Table names come from sqlite_master, not from user input.
- await (await ndb.db.prepare(`DELETE FROM "${row.name}"`)).run({});
- }
+ await clearNativeSqliteWalletDbInTransaction(ndb);
await ndb.txc.commit();
} catch (e) {
try {
@@ -5287,6 +5373,21 @@ export async function clearNativeSqliteWalletDb(
});
}
+/** Delete native wallet rows inside a transaction already owned by caller. */
+export async function clearNativeSqliteWalletDbInTransaction(
+ ndb: NativeSqliteWalletDb,
+): Promise<void> {
+ const rows = await (
+ await ndb.db.prepare(
+ `SELECT name FROM sqlite_master WHERE ${DATA_TABLES_CONDITION}`,
+ )
+ ).getAll();
+ for (const row of rows) {
+ // Table names come from sqlite_master, not from user input.
+ await (await ndb.db.prepare(`DELETE FROM "${row.name}"`)).run({});
+ }
+}
+
/**
* Names of the tables holding wallet data, in a stable order.
*
diff --git a/packages/taler-wallet-core/src/dbtx.ts b/packages/taler-wallet-core/src/dbtx.ts
@@ -158,7 +158,72 @@ export interface WalletCoinAvailabilityRef extends WalletDenomRef {
maxAge: number;
}
+/** Stores participating in backend conversion, named independently of layout. */
+export type WalletDbMigrationStore =
+ | "config"
+ | "currencyInfo"
+ | "contacts"
+ | "mailboxMessages"
+ | "mailboxConfigurations"
+ | "contractTerms"
+ | "tombstones"
+ | "operationRetries"
+ | "bankAccounts"
+ | "globalCurrencyExchanges"
+ | "globalCurrencyAuditors"
+ | "exchangeBaseUrlFixups"
+ | "exchangeBaseUrlMigrationLog"
+ | "reserves"
+ | "exchanges"
+ | "exchangeDetails"
+ | "exchangeSignKeys"
+ | "denominationFamilies"
+ | "denominations"
+ | "withdrawalGroups"
+ | "purchases"
+ | "refreshGroups"
+ | "coins"
+ | "planchets"
+ | "refreshSessions"
+ | "coinHistory"
+ | "coinAvailability"
+ | "refundGroups"
+ | "tokens"
+ | "slates"
+ | "depositGroups"
+ | "recoupGroups"
+ | "denomLossEvents"
+ | "peerPushDebit"
+ | "peerPushCredit"
+ | "peerPullDebit"
+ | "peerPullCredit"
+ | "donationSummaries"
+ | "donationPlanchets"
+ | "donationReceipts"
+ | "transactionsMeta"
+ | "refundItems";
+
+export interface WalletDbMigrationPage<T> {
+ records: T[];
+ /** Backend-private continuation token. Absent after an empty page. */
+ nextCursor?: unknown;
+}
+
export interface WalletDbTransaction {
+ /**
+ * Read a bounded page for database conversion.
+ *
+ * `read` is the ordinary DAL enumeration used to construct records on the
+ * native backend. IndexedDB can scan its object store directly; the store
+ * name tells it which physical store corresponds to the neutral entity.
+ */
+ scanMigrationRecords<T>(
+ store: WalletDbMigrationStore,
+ read: (tx: WalletDbTransaction) => Promise<T[]>,
+ cursor: unknown | undefined,
+ limit: number,
+ ): Promise<WalletDbMigrationPage<T>>;
+
/** Get the currency specification for a scope, if one is stored. */
getCurrencyInfo(
scopeInfo: ScopeInfo,
@@ -589,6 +654,11 @@ export interface WalletDbTransaction {
/** List every exchange details record, for all exchanges. */
listAllExchangeDetails(): Promise<WalletExchangeDetails[]>;
+ /** Get an exchange details record by its stable row identifier. */
+ getExchangeDetailsByRowId(
+ rowId: number,
+ ): Promise<WalletExchangeDetails | undefined>;
+
/**
* Create or update an exchange details record, returning its row id.
*
@@ -605,6 +675,9 @@ export interface WalletDbTransaction {
exchangeDetailsRowId: number,
): Promise<WalletExchangeSignkeys[]>;
+ /** List every exchange signing key, including orphaned legacy rows. */
+ listAllExchangeSignKeys(): Promise<WalletExchangeSignkeys[]>;
+
/** Create or update an exchange signing key. */
upsertExchangeSignKey(rec: WalletExchangeSignkeys): Promise<void>;
@@ -675,6 +748,9 @@ export interface WalletDbTransaction {
/** Get the refund items belonging to a refund group. */
getRefundItemsByGroup(refundGroupId: string): Promise<WalletRefundItem[]>;
+ /** List every refund item, including orphaned legacy rows. */
+ listAllRefundItems(): Promise<WalletRefundItem[]>;
+
/**
* Create or update a refund item, returning its row id.
*
@@ -802,6 +878,9 @@ export interface WalletDbTransaction {
/** Get the planchets of a withdrawal group. */
getPlanchetsByGroup(withdrawalGroupId: string): Promise<WalletPlanchet[]>;
+ /** List every planchet, including orphaned legacy rows. */
+ listAllPlanchets(): Promise<WalletPlanchet[]>;
+
/** Count the planchets of a withdrawal group. */
countPlanchetsByGroup(withdrawalGroupId: string): Promise<number>;
@@ -846,6 +925,9 @@ export interface WalletDbTransaction {
refreshGroupId: string,
): Promise<WalletRefreshSession[]>;
+ /** List every refresh session, including orphaned legacy rows. */
+ listAllRefreshSessions(): Promise<WalletRefreshSession[]>;
+
/** Get a recoup group by ID. */
getRecoupGroup(recoupGroupId: string): Promise<WalletRecoupGroup | undefined>;
@@ -908,6 +990,9 @@ export interface WalletDbTransaction {
/** Get the recorded history of a coin. */
getCoinHistory(coinPub: string): Promise<WalletCoinHistory | undefined>;
+ /** List every coin history, including orphaned legacy rows. */
+ listAllCoinHistories(): Promise<WalletCoinHistory[]>;
+
/** Create or update the recorded history of a coin. */
upsertCoinHistory(rec: WalletCoinHistory): Promise<void>;
diff --git a/packages/taler-wallet-core/src/host-common.ts b/packages/taler-wallet-core/src/host-common.ts
@@ -16,6 +16,7 @@
import { WalletNotification } from "@gnu-taler/taler-util";
import { HttpRequestLibrary } from "@gnu-taler/taler-util/http";
+import type { Sqlite3Database } from "@gnu-taler/idb-bridge";
/**
* Helpers to initiate a wallet in a host environment.
@@ -47,6 +48,43 @@ export interface DefaultNodeWalletArgs {
}
/**
+ * Claim a filesystem-backed wallet database for this connection's lifetime.
+ *
+ * The in-memory operation gate serializes users of one Wallet instance, but it
+ * cannot see another process (or another Wallet instance in this process).
+ * SQLite's exclusive locking mode supplies that outer ownership boundary. A
+ * transaction is needed to acquire the lock immediately; merely setting the
+ * pragma would defer the conflict until the first later database operation.
+ * The connection retains the lock after COMMIT and releases it on close.
+ */
+export async function acquireSqliteWalletDbOwnership(
+ db: Sqlite3Database,
+): Promise<void> {
+ let transactionStarted = false;
+ try {
+ await db.exec("PRAGMA locking_mode = EXCLUSIVE");
+ await (await db.prepare("BEGIN EXCLUSIVE")).run({});
+ transactionStarted = true;
+ await (await db.prepare("COMMIT")).run({});
+ transactionStarted = false;
+ } catch (cause) {
+ if (transactionStarted) {
+ try {
+ await (await db.prepare("ROLLBACK")).run({});
+ } catch {
+ // Preserve the acquisition error. Closing the connection is the
+ // caller's responsibility and also releases any partially held lock.
+ }
+ }
+ throw new Error(
+ "could not acquire exclusive ownership of the wallet database; " +
+ "another wallet process may still be using it",
+ { cause },
+ );
+ }
+}
+
+/**
* Generate a random alphanumeric ID. Does *not* use cryptographically
* secure randomness.
*/
diff --git a/packages/taler-wallet-core/src/host-impl.node.ts b/packages/taler-wallet-core/src/host-impl.node.ts
@@ -37,6 +37,7 @@ import { createPlatformHttpLib } from "@gnu-taler/taler-util/http";
import { NodeThreadCryptoWorkerFactory } from "./crypto/workers/nodeThreadWorker.js";
import { SynchronousCryptoWorkerFactoryPlain } from "./crypto/workers/synchronousWorkerFactoryPlain.js";
import {
+ acquireSqliteWalletDbOwnership,
DefaultNodeWalletArgs,
getSqlite3FilenameFromStoragePath,
} from "./host-common.js";
@@ -46,10 +47,13 @@ import { WalletDbHandle } from "./dbtx-handle.js";
import {
dropExpiredMigrationBackup,
inspectWalletDbFile,
+ inspectWalletDbFileDetails,
+ MigrationAuthority,
migrateWalletDbToNative,
NativeMigrationInfo,
readNativeMigrationInfo,
restoreMigrationBackup,
+ resolveAmbiguousWalletDb,
WalletDbFileKind,
} from "./db-native-migration.js";
import * as fs from "node:fs";
@@ -71,6 +75,12 @@ async function makeSqliteDb(
// own transaction state and its own locks.
const imp = await createNodeHelperSqlite3Impl();
const db = await imp.open(dbFilename);
+ try {
+ await acquireSqliteWalletDbOwnership(db);
+ } catch (e) {
+ await db.close();
+ throw e;
+ }
// Exactly one backend is opened. Opening both meant every operation that
// works on the whole database had to pick one, and picking wrong produced a
@@ -80,6 +90,12 @@ async function makeSqliteDb(
// environment variable only says what a database that does not exist yet
// should be created as.
const kind = await inspectWalletDbFile(db);
+ if (kind === "ambiguous") {
+ await db.close();
+ throw Error(
+ `${dbFilename} contains records in both wallet schemas and has no trustworthy authority marker; inspect it and use advanced db-migration-resolve`,
+ );
+ }
if (
kind === "native" ||
(kind === "empty" && process.env.TALER_WALLET_NATIVE_DB)
@@ -131,6 +147,9 @@ async function makeSqliteDb(
export async function inspectWalletDbPath(dbPath: string): Promise<{
kind: WalletDbFileKind;
migration: NativeMigrationInfo | undefined;
+ indexedDbRecords: number;
+ nativeRecords: number;
+ ambiguityReason?: string;
}> {
if (!fs.existsSync(dbPath)) {
throw Error(`wallet database ${dbPath} does not exist`);
@@ -138,10 +157,38 @@ export async function inspectWalletDbPath(dbPath: string): Promise<{
const imp = await createNodeHelperSqlite3Impl();
const db = await imp.open(dbPath);
try {
- return {
- kind: await inspectWalletDbFile(db),
- migration: await readNativeMigrationInfo(db),
- };
+ const inspection = await inspectWalletDbFileDetails(db);
+ return { ...inspection, migration: await readNativeMigrationInfo(db) };
+ } finally {
+ await db.close();
+ }
+}
+
+/** Back up and explicitly select the authoritative schema of an ambiguous file. */
+export async function resolveWalletDbMigration(
+ dbPath: string,
+ keep: MigrationAuthority,
+ backupPath: string,
+): Promise<void> {
+ if (!fs.existsSync(dbPath))
+ throw Error(`wallet database ${dbPath} does not exist`);
+ if (fs.existsSync(backupPath))
+ throw Error(`backup destination ${backupPath} already exists`);
+ const imp = await createNodeHelperSqlite3Impl();
+ const db = await imp.open(dbPath);
+ try {
+ const before = await inspectWalletDbFileDetails(db);
+ if (before.kind !== "ambiguous") {
+ throw Error(
+ `database is ${before.kind}, not ambiguous; refusing recovery`,
+ );
+ }
+ // VACUUM INTO is SQLite's consistent full-file snapshot. It runs before
+ // schema initialization or authority changes, and refuses an existing file.
+ await (
+ await db.prepare("VACUUM INTO $filename")
+ ).run({ filename: backupPath });
+ await resolveAmbiguousWalletDb(db, keep);
} finally {
await db.close();
}
diff --git a/packages/taler-wallet-core/src/host-impl.qtart.ts b/packages/taler-wallet-core/src/host-impl.qtart.ts
@@ -42,6 +42,7 @@ import { createPlatformHttpLib } from "@gnu-taler/taler-util/http";
import { qjsStd } from "@gnu-taler/taler-util/qtart";
import { SynchronousCryptoWorkerFactoryPlain } from "./crypto/workers/synchronousWorkerFactoryPlain.js";
import {
+ acquireSqliteWalletDbOwnership,
DefaultNodeWalletArgs,
getSqlite3FilenameFromStoragePath,
} from "./host-common.js";
@@ -114,8 +115,20 @@ async function makeSqliteDb(
// for -- the mobile wallets hand wallet-core a database and no directory to
// put a second one in.
const db = await imp.open(filename);
+ try {
+ await acquireSqliteWalletDbOwnership(db);
+ } catch (e) {
+ await db.close();
+ throw e;
+ }
const kind = await inspectWalletDbFile(db);
+ if (kind === "ambiguous") {
+ await db.close();
+ throw Error(
+ `${filename} contains records in both wallet schemas and has no trustworthy authority marker; explicit offline recovery is required`,
+ );
+ }
if (kind === "native") {
logger.info("opening the wallet database with the native schema");
await dropExpiredMigrationBackup(db);
diff --git a/packages/taler-wallet-core/src/index.node.ts b/packages/taler-wallet-core/src/index.node.ts
@@ -38,6 +38,7 @@ export type { DbConversionReport } from "./db-converter.js";
export {
inspectWalletDbPath,
rollbackWalletDbMigration,
+ resolveWalletDbMigration,
} from "./host-impl.node.js";
export type {
NativeMigrationInfo,
diff --git a/packages/taler-wallet-core/src/query.ts b/packages/taler-wallet-core/src/query.ts
@@ -36,6 +36,7 @@ import {
IDBTransactionMode,
IDBValidKey,
IDBVersionChangeEvent,
+ GlobalIDB,
} from "@gnu-taler/idb-bridge";
import {
CancellationToken,
@@ -477,6 +478,11 @@ export interface StoreReadWriteAccessor<RecordType, IndexMap> {
count?: number,
): Promise<RecordType[]>;
iter(query?: IDBValidKey): ResultStream<RecordType>;
+ /** Read a bounded primary-key page without retaining the whole store. */
+ scan(
+ after: IDBValidKey | undefined,
+ limit: number,
+ ): Promise<{ records: RecordType[]; lastKey?: IDBValidKey }>;
put(r: RecordType, key?: IDBValidKey): Promise<InsertResponse>;
add(r: RecordType, key?: IDBValidKey): Promise<InsertResponse>;
delete(key: IDBValidKey): Promise<void>;
@@ -823,6 +829,34 @@ function makeTxClientContext(
const req = tx.objectStore(storeName).openCursor(query);
return new ResultStream<any>(req);
},
+ async scan(after, limit) {
+ internalContext.throwIfInactive();
+ if (!Number.isInteger(limit) || limit <= 0) {
+ throw Error("scan limit must be a positive integer");
+ }
+ const range =
+ after === undefined
+ ? undefined
+ : GlobalIDB.KeyRange.lowerBound(after, true);
+ const store = tx.objectStore(storeName);
+ const records = await requestToPromise(
+ store.getAll(range, limit),
+ internalContext,
+ );
+ const last = records[records.length - 1];
+ const keyPath = swi.store.keyPath;
+ let lastKey: IDBValidKey | undefined;
+ if (last !== undefined) {
+ if (Array.isArray(keyPath)) {
+ lastKey = keyPath.map((component) => last[component]);
+ } else if (typeof keyPath === "string") {
+ lastKey = last[keyPath];
+ } else {
+ throw Error(`migration store ${storeName} has no inline key`);
+ }
+ }
+ return { records, lastKey };
+ },
async add(r, k) {
internalContext.throwIfInactive();
if (!internalContext.allowWrite) {
diff --git a/packages/taler-wallet-core/src/requests.ts b/packages/taler-wallet-core/src/requests.ts
@@ -2118,10 +2118,14 @@ export async function handleTestingRunFixup(
}
// fixup.fn takes the raw IndexedDB transaction: fixups repair that
// backend's records and are not expressible through the DAL.
- const idb = requireIdbBackend(wex, "running a fixup");
- const access = await idb.rawAccess();
- await access.runAllStoresReadWriteTx({}, async (tx) => {
- await fixup.fn(tx);
+ await wex.ws.runWithDatabaseShared(async () => {
+ // Resolve the backend after admission: a queued fixup must not retain an
+ // IndexedDB handle across a native migration.
+ const idb = requireIdbBackend(wex, "running a fixup");
+ const access = await idb.rawAccess();
+ await access.runAllStoresReadWriteTx({}, async (tx) => {
+ await fixup.fn(tx);
+ });
});
await wex.runWalletDbTx(async (tx) => {
await rematerializeTransactions(wex, tx);
diff --git a/packages/taler-wallet-core/src/wallet-db-gate.test.ts b/packages/taler-wallet-core/src/wallet-db-gate.test.ts
@@ -0,0 +1,79 @@
+/*
+ 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.
+ */
+
+import assert from "node:assert";
+import { test } from "node:test";
+
+import { WalletDbHandle } from "./dbtx-handle.js";
+import { AdmittedWalletDbHandle, DbOperationGate } from "./wallet.js";
+
+function deferred(): { promise: Promise<void>; resolve: () => void } {
+ let resolve!: () => void;
+ return { promise: new Promise<void>((r) => (resolve = r)), resolve };
+}
+
+function fakeHandle(name: string, events: string[]): WalletDbHandle {
+ return {
+ name,
+ async runReadWriteTx(f) {
+ events.push(`${name}:tx`);
+ return await f({ backend: name } as any);
+ },
+ async exportDatabase() {
+ events.push(`${name}:export`);
+ return {};
+ },
+ async importDatabase() {},
+ async clearDatabase() {},
+ getAccessStats() {
+ return undefined;
+ },
+ setNotificationSink() {},
+ emitNotification() {},
+ async close() {},
+ };
+}
+
+test("database gate drains old work and admits queued work on replacement", async () => {
+ const events: string[] = [];
+ const gate = new DbOperationGate();
+ const oldHandle = fakeHandle("old", events);
+ const newHandle = fakeHandle("new", events);
+ let current = oldHandle;
+ const admitted = new AdmittedWalletDbHandle(() => current, gate);
+ const active = deferred();
+ const releaseActive = deferred();
+
+ const beforeMigration = admitted.runReadWriteTx(async (tx: any) => {
+ events.push(`active:${tx.backend}`);
+ active.resolve();
+ await releaseActive.promise;
+ });
+ await active.promise;
+
+ const migration = gate.runExclusive(async () => {
+ events.push("migration");
+ current = newHandle;
+ });
+ const queued = admitted.runReadWriteTx(async (tx: any) => {
+ events.push(`queued:${tx.backend}`);
+ });
+
+ await Promise.resolve();
+ assert.deepStrictEqual(events, ["old:tx", "active:old"]);
+ releaseActive.resolve();
+ await Promise.all([beforeMigration, migration, queued]);
+ assert.deepStrictEqual(events, [
+ "old:tx",
+ "active:old",
+ "migration",
+ "new:tx",
+ "queued:new",
+ ]);
+});
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
@@ -131,6 +131,123 @@ export interface DbRetryState {
retriedExchangeUpdate?: Set<string>;
}
+/** Writer-preferring admission gate for operations that may replace a DB handle. */
+export class DbOperationGate {
+ private shared = 0;
+ private exclusive = false;
+ private waitingExclusive = 0;
+ private waiters: Array<() => void> = [];
+
+ private async changed(): Promise<void> {
+ await new Promise<void>((resolve) => this.waiters.push(resolve));
+ }
+
+ private wake(): void {
+ const waiters = this.waiters;
+ this.waiters = [];
+ for (const resolve of waiters) resolve();
+ }
+
+ async runShared<T>(f: () => Promise<T>): Promise<T> {
+ while (this.exclusive || this.waitingExclusive !== 0) await this.changed();
+ this.shared++;
+ try {
+ return await f();
+ } finally {
+ this.shared--;
+ this.wake();
+ }
+ }
+
+ async acquireExclusive(): Promise<() => void> {
+ this.waitingExclusive++;
+ try {
+ while (this.exclusive || this.shared !== 0) await this.changed();
+ this.exclusive = true;
+ } finally {
+ this.waitingExclusive--;
+ }
+ let released = false;
+ return () => {
+ if (released) return;
+ released = true;
+ this.exclusive = false;
+ this.wake();
+ };
+ }
+
+ async runExclusive<T>(f: () => Promise<T>): Promise<T> {
+ const release = await this.acquireExclusive();
+ try {
+ return await f();
+ } finally {
+ release();
+ }
+ }
+}
+
+/** Stable facade that resolves the current handle only after shared admission. */
+export class AdmittedWalletDbHandle implements WalletDbHandle {
+ get name(): string {
+ return this.current().name;
+ }
+ exportToFile?: WalletDbHandle["exportToFile"];
+ readBackupJson?: WalletDbHandle["readBackupJson"];
+ getDiagnosticStats?: WalletDbHandle["getDiagnosticStats"];
+
+ constructor(
+ private current: () => WalletDbHandle,
+ private gate: DbOperationGate,
+ ) {
+ if (current().exportToFile) {
+ this.exportToFile = (...args) =>
+ gate.runShared(async () => {
+ const fn = current().exportToFile;
+ if (!fn)
+ throw Error("current database backend cannot export to a file");
+ return await fn.apply(current(), args);
+ });
+ }
+ if (current().readBackupJson) {
+ this.readBackupJson = (...args) =>
+ gate.runShared(async () => {
+ const fn = current().readBackupJson;
+ if (!fn)
+ throw Error("current database backend cannot read this backup");
+ return await fn.apply(current(), args);
+ });
+ }
+ if (current().getDiagnosticStats) {
+ this.getDiagnosticStats = () => current().getDiagnosticStats?.();
+ }
+ }
+
+ runReadWriteTx<T>(f: (tx: WalletDbTransaction) => Promise<T>): Promise<T> {
+ return this.gate.runShared(() => this.current().runReadWriteTx(f));
+ }
+ exportDatabase(): Promise<any> {
+ return this.gate.runShared(() => this.current().exportDatabase());
+ }
+ importDatabase(dump: any): Promise<void> {
+ return this.gate.runShared(() => this.current().importDatabase(dump));
+ }
+ clearDatabase(): Promise<void> {
+ return this.gate.runShared(() => this.current().clearDatabase());
+ }
+ getAccessStats() {
+ return this.current().getAccessStats();
+ }
+ setNotificationSink(sink: (n: WalletNotification) => void): void {
+ this.current().setNotificationSink(sink);
+ }
+ emitNotification(notification: WalletNotification): void {
+ this.current().emitNotification(notification);
+ }
+ close(): Promise<void> {
+ return this.gate.runExclusive(() => this.current().close());
+ }
+}
+
export function walletExchangeClient(
baseUrl: string,
wex: WalletExecutionContext,
@@ -787,6 +904,12 @@ export class InternalWalletState {
private loadingDbCond: AsyncCondition = new AsyncCondition();
+ private dbOperationGate = new DbOperationGate();
+
+ private admittedDb: WalletDbHandle;
+
+ private suspendedDbRelease: (() => void) | undefined;
+
performanceStats: PerformanceTable = {};
/**
@@ -796,7 +919,7 @@ export class InternalWalletState {
* line works through WalletDbHandle and cannot tell the two apart.
*/
public get db(): WalletDbHandle {
- return this.dbHandle;
+ return this.admittedDb;
}
/**
@@ -806,7 +929,7 @@ export class InternalWalletState {
* when the capability is absent.
*/
public get idbOnly(): IdbWalletDbHandle | undefined {
- const h = this.db;
+ const h = this.dbHandle;
return h instanceof IdbWalletDbHandle ? h : undefined;
}
@@ -845,6 +968,11 @@ export class InternalWalletState {
return ret;
}
+ /** Admit an IndexedDB-specific operation that cannot use the DAL. */
+ async runWithDatabaseShared<T>(f: () => Promise<T>): Promise<T> {
+ return await this.dbOperationGate.runShared(f);
+ }
+
/**
* When set to false, all tasks that require network will be stopped and
* retried until connection is restored.
@@ -919,6 +1047,10 @@ export class InternalWalletState {
timer: TimerAPI,
cryptoWorkerFactory: CryptoWorkerFactory,
) {
+ this.admittedDb = new AdmittedWalletDbHandle(
+ () => 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));
@@ -994,6 +1126,14 @@ export class InternalWalletState {
* work that was queued before the swap runs against the new database.
*/
async migrateDbToNativeSchema(): Promise<boolean> {
+ if (this.loadingDb) {
+ while (this.loadingDb) {
+ await this.loadingDbCond.wait();
+ }
+ }
+ // Resolve only after earlier initialization/migration work has completed:
+ // a repeated init queued during a successful migration must observe the
+ // new native handle, not retain the old IndexedDB handle it first saw.
const oldHandle = this.dbHandle;
if (!oldHandle.migrateToNative) {
// Either the database is already native or the host cannot migrate it;
@@ -1002,29 +1142,24 @@ export class InternalWalletState {
logger.trace("this wallet database offers no in-place migration");
return false;
}
- if (this.loadingDb) {
- while (this.loadingDb) {
- await this.loadingDbCond.wait();
- }
- }
this.loadingDb = true;
try {
- const newHandle = await oldHandle.migrateToNative();
- this.dbHandle = newHandle;
- newHandle.setNotificationSink((n) => this.notify(n));
- await oldHandle.close();
- // Records from the old database are in the caches under the same
- // identities, but nothing guarantees the two backends materialise them
- // identically, and a stale cached record would outlive the database it
- // came from.
- this.clearAllCaches();
- return true;
- } catch (e) {
- logger.error(
- `migration to the native database failed, continuing with the` +
- ` existing one: ${safeStringifyException(e)}`,
- );
- return false;
+ return await this.dbOperationGate.runExclusive(async () => {
+ try {
+ const newHandle = await oldHandle.migrateToNative!();
+ this.dbHandle = newHandle;
+ newHandle.setNotificationSink((n) => this.notify(n));
+ await oldHandle.close();
+ this.clearAllCaches();
+ return true;
+ } catch (e) {
+ logger.error(
+ `migration to the native database failed, continuing with the` +
+ ` existing one: ${safeStringifyException(e)}`,
+ );
+ return false;
+ }
+ });
} finally {
this.loadingDb = false;
this.loadingDbCond.trigger();
@@ -1041,16 +1176,32 @@ export class InternalWalletState {
}
}
this.loadingDb = true;
- await this.db.close();
+ this.suspendedDbRelease = await this.dbOperationGate.acquireExclusive();
+ try {
+ await this.dbHandle.close();
+ } catch (e) {
+ this.suspendedDbRelease();
+ this.suspendedDbRelease = undefined;
+ this.loadingDb = false;
+ this.loadingDbCond.trigger();
+ throw e;
+ }
}
/**
* Resume database by re-opening it.
*/
async resumeDatabase(): Promise<void> {
- this.loadingDb = false;
- this.loadingDbCond.trigger();
- await this.ensureWalletDbOpen();
+ const release = this.suspendedDbRelease;
+ this.suspendedDbRelease = undefined;
+ try {
+ const idb = this.idbOnly;
+ if (idb) await idb.ensureOpen();
+ } finally {
+ this.loadingDb = false;
+ this.loadingDbCond.trigger();
+ release?.();
+ }
}
notify(n: WalletNotification): void {
