commit 159fa81d9dd96c992a66f3ce216b7647f8505908
parent d8965250c44721c2adea59db5e80a4332dd142f8
Author: Florian Dold <dold@taler.net>
Date: Thu, 20 Aug 2026 13:52:50 +0200
wallet-core: restart native migrations with record progress
Diffstat:
5 files changed, 263 insertions(+), 77 deletions(-)
diff --git a/packages/taler-util/src/notifications.ts b/packages/taler-util/src/notifications.ts
@@ -392,9 +392,9 @@ export interface DatabaseMaintenanceProgressNotification {
step?: string;
completedSteps: number;
totalSteps: number;
- /** Records completed in the current store. */
+ /** Records completed in the current phase across the whole migration. */
processedRecords?: number;
- /** Known after copying a store, and therefore available while verifying it. */
+ /** Total records in the whole migration, known before copying starts. */
totalRecords?: number;
}
diff --git a/packages/taler-wallet-core/src/db-converter.test.ts b/packages/taler-wallet-core/src/db-converter.test.ts
@@ -28,6 +28,7 @@ import assert from "node:assert";
import { test } from "node:test";
import {
+ DatabaseMaintenanceProgressNotification,
encodeCrock,
getRandomBytes,
NotificationType,
@@ -45,7 +46,11 @@ import {
WalletPurchase,
} from "./db-common.js";
import { SQLITE_BASELINE_SCHEMA } from "./db-sqlite-schema.js";
-import { convertWalletDb, DB_CONVERSION_BATCH_SIZE } from "./db-converter.js";
+import {
+ convertWalletDb,
+ DB_CONVERSION_BATCH_SIZE,
+ DB_CONVERSION_PROGRESS_RECORDS,
+} 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";
@@ -95,7 +100,7 @@ test("converter: IndexedDB to sqlite, populated by the conformance corpus", asyn
// 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++) {
+ for (let i = 0; i < DB_CONVERSION_PROGRESS_RECORDS * 2 + 17; i++) {
await tx.upsertTombstone({ id: `bounded-conversion-${i}` });
}
});
@@ -164,26 +169,41 @@ test("converter: IndexedDB to sqlite, populated by the conformance corpus", asyn
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",
+ const maintenanceProgress = progress.filter(
+ (n): n is DatabaseMaintenanceProgressNotification =>
+ n.type === NotificationType.DatabaseMaintenanceProgress &&
+ n.operation === "indexeddb-to-native-migration",
);
assert.ok(
- progress.some(
- (n) =>
- n.type === NotificationType.DatabaseMaintenanceProgress &&
- n.operation === "indexeddb-to-native-migration" &&
- n.phase === "verify",
- ),
- "verification progress was not reported",
+ maintenanceProgress.every((n) => n.totalRecords === report.totalRecords),
+ "progress did not carry the global record total",
);
+ for (const phase of ["copy", "verify"] as const) {
+ const records = maintenanceProgress
+ .filter((n) => n.phase === phase && n.processedRecords !== undefined)
+ .map((n) => n.processedRecords!);
+ assert.strictEqual(
+ records[0],
+ 0,
+ `${phase} progress did not start at zero`,
+ );
+ assert.strictEqual(
+ records.at(-1),
+ report.totalRecords,
+ `${phase} progress did not reach the global total`,
+ );
+ assert.ok(
+ records.length > 2,
+ `${phase} progress had no intermediate event`,
+ );
+ for (let i = 1; i < records.length - 1; i++) {
+ assert.ok(
+ Math.floor(records[i] / DB_CONVERSION_PROGRESS_RECORDS) >
+ Math.floor(records[i - 1] / DB_CONVERSION_PROGRESS_RECORDS),
+ `${phase} record progress was reported too frequently`,
+ );
+ }
+ }
assert.strictEqual(
(src as any).idbHandle._transactions.length,
retainedTransactionsBefore,
diff --git a/packages/taler-wallet-core/src/db-converter.ts b/packages/taler-wallet-core/src/db-converter.ts
@@ -518,6 +518,9 @@ export interface DbConversionReport {
/** Small enough to bound retained records while amortising transaction setup. */
export const DB_CONVERSION_BATCH_SIZE = 128;
+/** Record interval at which migration progress is reported. */
+export const DB_CONVERSION_PROGRESS_RECORDS = 100;
+
/**
* JSON stringification with sorted object keys, so structurally equal
* records compare equal regardless of property insertion order -- the two
@@ -639,7 +642,22 @@ export async function convertWalletDb(
dst: WalletDbHandle,
): Promise<DbConversionReport> {
const copied: Record<string, number> = {};
- let total = 0;
+
+ // Inventorying the source up front makes the progress denominator known
+ // before the first write. Keep the digests: verification can compare the
+ // destination against these instead of scanning the source a second time,
+ // so global progress does not add another full database pass.
+ const sourceDigests = new Map<WalletDbMigrationStore, RecordMultisetDigest>();
+ let totalRecords = 0;
+ for (const group of COPY_PLAN) {
+ for (const st of group) {
+ const digest = await digestStore(src, st);
+ sourceDigests.set(st.name, digest);
+ totalRecords += digest.count;
+ }
+ }
+
+ const progressInterval = DB_CONVERSION_PROGRESS_RECORDS;
const notify = (
phase: "copy" | "verify",
@@ -655,13 +673,33 @@ export async function convertWalletDb(
totalSteps: DB_CONVERSION_STEP_COUNT,
...(step ? { step: step.name } : {}),
...(processedRecords !== undefined ? { processedRecords } : {}),
- ...(step && copied[step.name] !== undefined
- ? { totalRecords: copied[step.name] }
- : {}),
+ totalRecords,
});
};
+ const makeRecordProgress = (phase: "copy" | "verify") => {
+ let next = progressInterval;
+ let last = -1;
+ return (
+ completedSteps: number,
+ processedRecords: number,
+ step?: CopyStep,
+ force = false,
+ ): void => {
+ if (!force && processedRecords < next) return;
+ if (processedRecords === last) return;
+ notify(phase, completedSteps, step, processedRecords);
+ last = processedRecords;
+ next =
+ (Math.floor(processedRecords / progressInterval) + 1) *
+ progressInterval;
+ };
+ };
+
+ const copyProgress = makeRecordProgress("copy");
+ let copiedRecords = 0;
let stepIndex = 0;
+ copyProgress(stepIndex, 0, undefined, true);
for (const group of COPY_PLAN) {
for (const st of group) {
let cursor: unknown | undefined;
@@ -675,27 +713,32 @@ export async function convertWalletDb(
}
});
storeCount += page.records.length;
- notify("copy", stepIndex, st, storeCount);
+ copiedRecords += page.records.length;
+ copyProgress(stepIndex, copiedRecords, st);
cursor = page.nextCursor;
if (cursor === undefined) break;
}
copied[st.name] = storeCount;
- total += storeCount;
stepIndex++;
- notify("copy", stepIndex, st, storeCount);
+ notify("copy", stepIndex, st);
logger.trace(`copied ${storeCount} ${st.name}`);
}
}
+ copyProgress(stepIndex, copiedRecords, undefined, true);
// 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.
+ const verifyProgress = makeRecordProgress("verify");
+ let verifiedRecords = 0;
stepIndex = 0;
+ verifyProgress(stepIndex, 0, undefined, true);
for (const group of COPY_PLAN) {
for (const st of group) {
- const sourceDigest = await digestStore(src, st);
+ const sourceDigest = sourceDigests.get(st.name)!;
+ const beforeStore = verifiedRecords;
const destinationDigest = await digestStore(dst, st, (processed) =>
- notify("verify", stepIndex, st, processed),
+ verifyProgress(stepIndex, beforeStore + processed, st),
);
if (!sourceDigest.equals(destinationDigest)) {
throw Error(
@@ -704,9 +747,11 @@ export async function convertWalletDb(
` ${destinationDigest.describe()})`,
);
}
+ verifiedRecords += destinationDigest.count;
stepIndex++;
- notify("verify", stepIndex, st, destinationDigest.count);
+ notify("verify", stepIndex, st);
}
}
- return { copied, totalRecords: total };
+ verifyProgress(stepIndex, verifiedRecords, undefined, true);
+ return { copied, totalRecords: copiedRecords };
}
diff --git a/packages/taler-wallet-core/src/db-native-migration.test.ts b/packages/taler-wallet-core/src/db-native-migration.test.ts
@@ -48,9 +48,14 @@ import {
resolveAmbiguousWalletDb,
restoreMigrationBackup,
} from "./db-native-migration.js";
-import { IDB_BACKUP_PREFIX, IDB_EMULATION_TABLES } from "./db-sqlite-schema.js";
+import {
+ IDB_BACKUP_PREFIX,
+ IDB_EMULATION_TABLES,
+ schemaMigrations,
+} from "./db-sqlite-schema.js";
+import { DB_CONVERSION_PROGRESS_RECORDS } from "./db-converter.js";
import { IdbWalletDbHandle } from "./dbtx-handle-impl.js";
-import { openNativeSqliteWalletDb } from "./dbtx-sqlite.js";
+import { initSqliteWalletDb, openNativeSqliteWalletDb } from "./dbtx-sqlite.js";
import {
inspectWalletDbPath,
resolveWalletDbMigration,
@@ -110,15 +115,23 @@ async function makePopulatedIdbDb(filename = ":memory:"): Promise<{
return { db, handle };
}
-async function makeMinimalIdbDb(): Promise<{
+async function openIdbDb(filename: string): 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);
const handle = new IdbWalletDbHandle(new BridgeIDBFactory(backend));
await handle.ensureOpen();
+ return { db, handle };
+}
+
+async function makeMinimalIdbDb(filename = ":memory:"): Promise<{
+ db: Sqlite3Database;
+ handle: IdbWalletDbHandle;
+}> {
+ const { db, handle } = await openIdbDb(filename);
await handle.runReadWriteTx((tx) =>
tx.upsertConfig({ key: "fault-test" as any, value: 1 }),
);
@@ -290,48 +303,142 @@ test("native migration: the retained backup can be put back", async () => {
);
});
-test("native migration: an interrupted attempt keeps the old database", async () => {
- const { db, handle } = await makePopulatedIdbDb();
+test("native migration: an interrupted attempt restarts after reopening", async () => {
+ const directory = fs.mkdtempSync(
+ path.join(os.tmpdir(), "wallet-db-migration-restart-"),
+ );
+ const filename = path.join(directory, "wallet.sqlite3");
+ let firstDb: Sqlite3Database | undefined;
+ let reopenedDb: Sqlite3Database | undefined;
+ try {
+ const first = await makeMinimalIdbDb(filename);
+ firstDb = first.db;
+ const expectedTombstones = DB_CONVERSION_PROGRESS_RECORDS * 2 + 17;
+ await first.handle.runReadWriteTx(async (tx) => {
+ for (let i = 0; i < expectedTombstones; i++) {
+ await tx.upsertTombstone({ id: `restart-${i}` });
+ }
+ });
+
+ // Throw only after a committed destination batch has advanced global
+ // progress. This leaves the same durable state as process termination:
+ // untouched IndexedDB tables, a running marker and partial native rows.
+ let interruptionInjected = false;
+ first.handle.setNotificationSink((n) => {
+ if (
+ !interruptionInjected &&
+ n.type === NotificationType.DatabaseMaintenanceProgress &&
+ n.operation === "indexeddb-to-native-migration" &&
+ n.phase === "copy" &&
+ (n.processedRecords ?? 0) >= DB_CONVERSION_PROGRESS_RECORDS
+ ) {
+ interruptionInjected = true;
+ throw Error("simulated migration interruption");
+ }
+ });
+ await assert.rejects(
+ () => migrateWalletDbToNative(first.db, first.handle),
+ /simulated migration interruption/,
+ );
+ assert.ok(
+ interruptionInjected,
+ "migration was not interrupted after a copy",
+ );
+ const interrupted = await inspectWalletDbFileDetails(first.db);
+ assert.strictEqual(interrupted.kind, "indexeddb");
+ assert.ok(interrupted.nativeRecords > 0, "no partial native copy was left");
+ assert.strictEqual(
+ (await readNativeMigrationInfo(first.db))?.status,
+ "running",
+ );
- // What an interruption leaves behind: the native tables exist and hold a
- // partial copy, the bookkeeping says an attempt was under way, and the
- // emulation's tables are untouched because the renames never ran.
- await migrateWalletDbToNative(db, handle);
- // Undo the completion so the file looks interrupted rather than migrated:
- // the bookkeeping says an attempt was under way and the emulation's tables
- // are back where a migration that never reached its renames left them.
- await (
- await db.prepare(
- "UPDATE idb_migration SET status = 'running', finished_at = NULL," +
- " backup_status = NULL, backup_expires_at = NULL WHERE id = 1",
- )
- ).run({});
- for (const t of IDB_EMULATION_TABLES) {
+ await first.handle.close();
+ await first.db.close();
+ firstDb = undefined;
+
+ // A new factory and connection exercise the path that previously surfaced
+ // only as "database opening error", rather than reusing an already-open
+ // IndexedDB handle as the old regression did.
+ const reopened = await openIdbDb(filename);
+ reopenedDb = reopened.db;
+ assert.strictEqual(await inspectWalletDbFile(reopened.db), "indexeddb");
+ const { handle: native, report } = await migrateWalletDbToNative(
+ reopened.db,
+ reopened.handle,
+ );
+ const tombstones = await native.runReadWriteTx((tx) =>
+ tx.listAllTombstones(),
+ );
+ assert.strictEqual(tombstones.length, expectedTombstones);
+ assert.strictEqual(
+ new Set(tombstones.map((t) => t.id)).size,
+ expectedTombstones,
+ "the restarted copy duplicated records",
+ );
+ assert.ok(report.totalRecords >= expectedTombstones);
+ assert.strictEqual(await inspectWalletDbFile(reopened.db), "native");
+ await native.close();
+ reopenedDb = undefined;
+ } finally {
+ await firstDb?.close().catch(() => {});
+ await reopenedDb?.close().catch(() => {});
+ fs.rmSync(directory, { recursive: true, force: true });
+ }
+});
+
+test("native migration: a legacy interrupted attempt remains restartable", async () => {
+ const directory = fs.mkdtempSync(
+ path.join(os.tmpdir(), "wallet-db-migration-legacy-"),
+ );
+ const filename = path.join(directory, "wallet.sqlite3");
+ let firstDb: Sqlite3Database | undefined;
+ let reopenedDb: Sqlite3Database | undefined;
+ try {
+ const first = await makeMinimalIdbDb(filename);
+ firstDb = first.db;
+ await initSqliteWalletDb(
+ first.db,
+ schemaMigrations.filter((m) => m.version < 7),
+ );
await (
- await db.prepare(
- `ALTER TABLE "${IDB_BACKUP_PREFIX}${t}" RENAME TO "${t}"`,
+ await first.db.prepare(
+ "INSERT INTO config (key, value) VALUES ('partial-only', '\"discard\"')",
)
).run({});
- }
+ await (
+ await first.db.prepare(
+ "INSERT INTO idb_migration (id, status, started_at)" +
+ " VALUES (1, 'running', 1)",
+ )
+ ).run({});
+ assert.strictEqual(
+ (await readNativeMigrationInfo(first.db))?.cleanupSafe,
+ undefined,
+ );
+ assert.strictEqual(await inspectWalletDbFile(first.db), "indexeddb");
- assert.strictEqual(
- await inspectWalletDbFile(db),
- "indexeddb",
- "an interrupted migration must leave the emulation authoritative",
- );
+ await first.handle.close();
+ await first.db.close();
+ firstDb = undefined;
- // Retrying discards the partial copy rather than adding to it.
- const { handle: native, report } = await migrateWalletDbToNative(db, handle);
- const coins = await native.runReadWriteTx((tx) => tx.listAllCoins());
- const seen = new Set(coins.map((c) => c.coinPub));
- assert.strictEqual(
- seen.size,
- coins.length,
- "the retry duplicated records from the interrupted attempt",
- );
- assert.ok(report.totalRecords >= 100);
- assert.strictEqual(await inspectWalletDbFile(db), "native");
- await native.close();
+ const reopened = await openIdbDb(filename);
+ reopenedDb = reopened.db;
+ const { handle: native, info } = await migrateWalletDbToNative(
+ reopened.db,
+ reopened.handle,
+ );
+ assert.strictEqual(info.status, "complete");
+ assert.strictEqual(info.cleanupSafe, true);
+ const config = await native.runReadWriteTx((tx) => tx.listAllConfig());
+ assert.ok(config.some((r) => r.key === ("fault-test" as any)));
+ assert.ok(!config.some((r) => r.key === ("partial-only" as any)));
+ await native.close();
+ reopenedDb = undefined;
+ } finally {
+ await firstDb?.close().catch(() => {});
+ await reopenedDb?.close().catch(() => {});
+ fs.rmSync(directory, { recursive: true, force: true });
+ }
});
test("native migration: mixed schemas without ownership fail closed", async () => {
diff --git a/packages/taler-wallet-core/src/db-native-migration.ts b/packages/taler-wallet-core/src/db-native-migration.ts
@@ -108,7 +108,11 @@ export interface NativeMigrationInfo {
recordsCopied?: number;
backupStatus?: MigrationBackupStatus;
backupExpiresAt?: number;
- /** Whether native rows are known to be only a disposable partial copy. */
+ /**
+ * Whether native rows are known to be only a disposable partial copy.
+ * Undefined identifies the released legacy schema from before this column
+ * was added; its running marker carried the same cleanup guarantee.
+ */
cleanupSafe?: boolean;
}
@@ -220,7 +224,7 @@ export async function inspectWalletDbFileDetails(
});
if (info?.status === "complete") return result("native");
- if (info?.status === "running" && info.cleanupSafe)
+ if (info?.status === "running" && info.cleanupSafe !== false)
return result("indexeddb");
if (info?.status === "running" && nativeRecords > 0) {
return result(
@@ -304,11 +308,19 @@ export async function migrateWalletDbToNative(
db: Sqlite3Database,
src: WalletDbHandle,
): Promise<NativeMigrationResult> {
+ // Read this before native initialization upgrades the schema. A running
+ // marker written by versions before cleanup_safe existed is trustworthy:
+ // those versions also cleared the native tables before recording it. Once
+ // migration 7 adds the column its DEFAULT 0 deliberately cannot make that
+ // distinction for us anymore.
+ const previous = await readNativeMigrationInfo(db);
+ const previousPartialIsCleanupSafe =
+ previous?.status === "running" && previous.cleanupSafe !== false;
+
const ndb = await openNativeSqliteWalletDb(db);
const dst = new SqliteWalletDbHandle(ndb);
const txc = ndb.txc;
- const previous = await readNativeMigrationInfo(db);
if (previous?.status === "complete") {
throw Error(
"this wallet database has already been migrated to the native schema",
@@ -321,9 +333,11 @@ export async function migrateWalletDbToNative(
);
}
if (previous?.status === "running") {
- if (previous.cleanupSafe) {
+ if (previousPartialIsCleanupSafe) {
logger.warn(
- "discarding the cleanup-safe partial copy left by an interrupted migration attempt",
+ previous.cleanupSafe === undefined
+ ? "discarding the partial copy left by an interrupted legacy migration attempt"
+ : "discarding the cleanup-safe partial copy left by an interrupted migration attempt",
);
await clearNativeSqliteWalletDb(ndb);
} else if ((await countNativeRecords(db)) !== 0) {
