commit 0512d76b0a7ca0c417b360630a2b3fcdeae21ea5
parent c0930b906580bf14221d50f339882fe85ef3d052
Author: Florian Dold <dold@taler.net>
Date: Sun, 19 Jul 2026 06:10:00 +0200
wallet: complete the native backend integration
Export, import, stored backups and the task shepherd now act on whichever
backend holds the data.
Diffstat:
4 files changed, 320 insertions(+), 24 deletions(-)
diff --git a/packages/taler-wallet-core/src/dbtx-sqlite.ts b/packages/taler-wallet-core/src/dbtx-sqlite.ts
@@ -178,7 +178,24 @@ export class SqliteTxControl {
export async function initSqliteWalletDb(db: Sqlite3Database): Promise<void> {
await db.exec("PRAGMA foreign_keys = ON");
+ // WAL: readers do not block the writer, and a commit appends to the log
+ // instead of fsyncing the whole database.
+ //
+ // The cost is that a WAL database is three files (db, -wal, -shm), so
+ // copying just the database file is not a snapshot. That matters because
+ // callers do treat the wallet DB as one file they can copy. Rather than
+ // give up WAL, {@link runNativeSqliteWalletTx} checkpoints with TRUNCATE
+ // whenever the transaction queue drains: under load the WAL accumulates
+ // normally, and the moment the wallet goes idle the main file is complete
+ // again and the -wal is empty. Idle is also the only moment at which an
+ // external copy could be coherent at all, so this makes single-file copies
+ // valid exactly when they can be.
+ //
+ // It also means a killed process leaves an empty -wal behind, so restoring
+ // a database file over it cannot replay stale frames from the old one.
await db.exec("PRAGMA journal_mode = WAL");
+ // NORMAL is safe against process crashes in WAL mode; only a power loss
+ // can cost the most recent commits, and never corruption.
await db.exec("PRAGMA synchronous = NORMAL");
const txc = await SqliteTxControl.create(db);
@@ -1020,12 +1037,31 @@ export class SqliteWalletTransaction implements WalletDbTransaction {
}
async getCoinsByPubs(coinPubs: string[]): Promise<WalletCoin[]> {
- // One statement per pub, like the IndexedDB implementation: the contract
- // is "the coins that exist", so missing pubs are skipped rather than
- // yielding holes, and the result follows the order of the argument.
+ // One statement for the whole batch, then reordered in memory. The
+ // IndexedDB version loops per pub, which is fine there but is a network
+ // round-trip each on this backend -- refresh passes every coin of a
+ // group through here. Contract is unchanged: missing pubs are skipped
+ // rather than yielding holes, and the result follows argument order.
+ if (coinPubs.length === 0) {
+ return [];
+ }
+ const params: Record<string, string> = {};
+ const placeholders = coinPubs.map((pub, i) => {
+ params[`p${i}`] = pub;
+ return `$p${i}`;
+ });
+ const rows = await this.all(
+ `SELECT * FROM coins WHERE coin_pub IN (${placeholders.join(", ")})`,
+ params,
+ );
+ const byPub = new Map<string, WalletCoin>();
+ for (const row of rows) {
+ const coin = this.rowToCoin(row);
+ byPub.set(coin.coinPub, coin);
+ }
const coins: WalletCoin[] = [];
for (const pub of coinPubs) {
- const coin = await this.getCoin(pub);
+ const coin = byPub.get(pub);
if (coin) {
coins.push(coin);
}
@@ -4653,7 +4689,28 @@ export interface NativeSqliteWalletDb {
export class TxQueue {
private tail: Promise<void> = Promise.resolve();
+ /**
+ * Transactions queued but not finished, including the running one.
+ *
+ * Used to detect the moment the queue drains, which is when the database
+ * can be checkpointed without stalling anybody.
+ */
+ private outstanding = 0;
+
+ /**
+ * True when the caller is the only transaction in the queue.
+ *
+ * Checked from inside a running transaction, which is still counted in
+ * `outstanding` at that point -- the decrement happens in this class's
+ * finally block, after the transaction body returns. So "nobody else is
+ * waiting" is 1, not 0.
+ */
+ get noOtherWaiting(): boolean {
+ return this.outstanding <= 1;
+ }
+
async run<T>(f: () => Promise<T>): Promise<T> {
+ this.outstanding++;
const prev = this.tail;
let release: () => void;
this.tail = new Promise<void>((resolve) => {
@@ -4663,6 +4720,7 @@ export class TxQueue {
try {
return await f();
} finally {
+ this.outstanding--;
release!();
}
}
@@ -4716,6 +4774,7 @@ async function runNativeSqliteWalletTxLocked<T>(
throw e;
}
await ndb.txc.commit();
+ await checkpointIfIdle(ndb);
for (const notif of tx.pendingNotifications) {
notifyFn(notif);
}
@@ -4724,3 +4783,173 @@ async function runNativeSqliteWalletTxLocked<T>(
}
return res;
}
+
+/**
+ * Fold the write-ahead log back into the database file, if nothing else is
+ * queued.
+ *
+ * This is what keeps the database file self-contained between operations, so
+ * that copying it is a valid snapshot (see the WAL note in
+ * {@link initSqliteWalletDb}). Skipped whenever another transaction is
+ * waiting: under load the log is allowed to grow, which is the point of WAL.
+ *
+ * TRUNCATE rather than PASSIVE so the -wal ends up zero-length rather than
+ * merely folded in; a leftover non-empty -wal next to a restored database
+ * file is the failure mode this exists to prevent.
+ *
+ * Best-effort: a failed checkpoint costs a stale snapshot, not correctness,
+ * and must not fail the transaction that already committed.
+ */
+async function checkpointIfIdle(ndb: NativeSqliteWalletDb): Promise<void> {
+ if (!ndb.lock.noOtherWaiting) {
+ return;
+ }
+ try {
+ await ndb.db.exec("PRAGMA wal_checkpoint(TRUNCATE)");
+ } catch (e) {
+ logger.warn(`WAL checkpoint failed: ${e}`);
+ }
+}
+
+/**
+ * Delete every row from the native wallet database.
+ *
+ * The schema itself is kept, including schema_migrations: this clears data,
+ * it does not reset the database to "never initialised".
+ */
+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 type = 'table'" +
+ " AND name NOT LIKE 'sqlite_%' AND name != 'schema_migrations'",
+ )
+ ).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 ndb.txc.commit();
+ } catch (e) {
+ try {
+ await ndb.txc.rollback();
+ } catch (rollbackErr) {
+ logger.warn(`rollback failed: ${rollbackErr}`);
+ }
+ throw e;
+ }
+ });
+}
+
+/**
+ * Names of the tables holding wallet data, in a stable order.
+ *
+ * schema_migrations is excluded: it describes the schema, not the data, and
+ * restoring it from a backup could claim migrations were applied that were
+ * not.
+ */
+async function listDataTables(ndb: NativeSqliteWalletDb): Promise<string[]> {
+ const rows = await (
+ await ndb.db.prepare(
+ "SELECT name FROM sqlite_master WHERE type = 'table'" +
+ " AND name NOT LIKE 'sqlite_%' AND name != 'schema_migrations'" +
+ " ORDER BY name",
+ )
+ ).getAll();
+ return rows.map((r) => r.name as string);
+}
+
+/**
+ * A dump of every row in the native wallet database.
+ *
+ * Deliberately a plain row dump rather than a file copy: it is stored inside
+ * another database as JSON, the same way the IndexedDB export is, so the
+ * stored-backup API works the same for both backends.
+ */
+export interface NativeSqliteDbDump {
+ schemaVersion: number;
+ tables: Record<string, Record<string, Sqlite3Value>[]>;
+}
+
+export async function exportNativeSqliteDb(
+ ndb: NativeSqliteWalletDb,
+): Promise<NativeSqliteDbDump> {
+ return await ndb.lock.run(async () => {
+ const tables = await listDataTables(ndb);
+ const out: NativeSqliteDbDump = {
+ schemaVersion: SQLITE_SCHEMA_VERSION,
+ tables: {},
+ };
+ for (const table of tables) {
+ const rows = await (
+ await ndb.db.prepare(`SELECT * FROM "${table}"`)
+ ).getAll();
+ // The helper can return INTEGER columns as bigint, which JSON cannot
+ // represent. Every integer in this schema (timestamps in
+ // microseconds, row ids, statuses) is inside the safe range.
+ out.tables[table] = rows.map((row) => {
+ const clean: Record<string, Sqlite3Value> = {};
+ for (const [k, v] of Object.entries(row)) {
+ clean[k] = typeof v === "bigint" ? Number(v) : v;
+ }
+ return clean;
+ });
+ }
+ return out;
+ });
+}
+
+export async function importNativeSqliteDb(
+ ndb: NativeSqliteWalletDb,
+ dump: NativeSqliteDbDump,
+): Promise<void> {
+ if (dump.schemaVersion !== SQLITE_SCHEMA_VERSION) {
+ throw Error(
+ `cannot import a native wallet DB dump of schema version` +
+ ` ${dump.schemaVersion} into version ${SQLITE_SCHEMA_VERSION}`,
+ );
+ }
+ await ndb.lock.run(async () => {
+ const tables = await listDataTables(ndb);
+ await ndb.txc.begin();
+ try {
+ // Clear and refill in one transaction: a partial import would leave
+ // the wallet with a mix of two databases.
+ for (const table of tables) {
+ await (await ndb.db.prepare(`DELETE FROM "${table}"`)).run({});
+ }
+ for (const table of tables) {
+ const rows = dump.tables[table];
+ if (!rows || rows.length === 0) {
+ continue;
+ }
+ for (const row of rows) {
+ const cols = Object.keys(row);
+ const params: Record<string, Sqlite3Value> = {};
+ for (const c of cols) {
+ params[c] = row[c];
+ }
+ const colList = cols.map((c) => `"${c}"`).join(", ");
+ const valList = cols.map((c) => `$${c}`).join(", ");
+ await (
+ await ndb.db.prepare(
+ `INSERT INTO "${table}" (${colList}) VALUES (${valList})`,
+ )
+ ).run(params);
+ }
+ }
+ await ndb.txc.commit();
+ } catch (e) {
+ try {
+ await ndb.txc.rollback();
+ } catch (rollbackErr) {
+ logger.warn(`rollback failed: ${rollbackErr}`);
+ }
+ throw e;
+ }
+ });
+}
diff --git a/packages/taler-wallet-core/src/host-impl.node.ts b/packages/taler-wallet-core/src/host-impl.node.ts
@@ -61,9 +61,17 @@ async function makeSqliteDb(
const dbFilename = getSqlite3FilenameFromStoragePath(
args.persistentStoragePath,
);
- logger.info(`using database ${dbFilename}`);
+ const useNativeDb = !!process.env.TALER_WALLET_NATIVE_DB;
+ // Whichever backend actually holds the wallet's data gets the canonical
+ // path, and the other one is moved aside. Anything that treats the wallet
+ // DB path as "the wallet database" -- a file copy, an inspection, a
+ // snapshot in a test -- then operates on the real data. With the native
+ // database in a sidecar, such a copy silently restored nothing.
+ const idbFilename =
+ useNativeDb && dbFilename !== ":memory:" ? `${dbFilename}.idb` : dbFilename;
+ logger.info(`using database ${idbFilename}`);
const myBackend = await createSqliteBackend(imp, {
- filename: dbFilename,
+ filename: idbFilename,
});
if (process.env.TALER_WALLET_DBTRACING) {
myBackend.enableTracing = true;
@@ -80,9 +88,8 @@ async function makeSqliteDb(
// and the native schema deliberately starts clean instead of replaying
// fixups (see SQLITE_SCHEMA_DESIGN.md).
let nativeSqliteDb: NativeSqliteWalletDb | undefined;
- if (process.env.TALER_WALLET_NATIVE_DB) {
- const nativeFilename =
- dbFilename === ":memory:" ? ":memory:" : `${dbFilename}.native`;
+ if (useNativeDb) {
+ const nativeFilename = dbFilename;
logger.info(`using NATIVE sqlite3 wallet DB at ${nativeFilename}`);
logger.warn("the native sqlite3 wallet DB backend is experimental");
// A second helper process: each one holds a single connection, so
diff --git a/packages/taler-wallet-core/src/requests.ts b/packages/taler-wallet-core/src/requests.ts
@@ -24,6 +24,11 @@
* Imports.
*/
import {
+ clearNativeSqliteWalletDb,
+ exportNativeSqliteDb,
+ importNativeSqliteDb,
+} from "./dbtx-sqlite.js";
+import {
AbortTransactionRequest,
AcceptBankIntegratedWithdrawalRequest,
AcceptManualWithdrawalRequest,
@@ -601,7 +606,13 @@ async function dumpCoins(wex: WalletExecutionContext): Promise<CoinDumpJson> {
async function createStoredBackup(
wex: WalletExecutionContext,
): Promise<CreateStoredBackupResponse> {
- const backup = await exportDb(wex.ws.idbFactory);
+ // The backup payload comes from whichever backend actually holds the data.
+ // The backups themselves are kept in a separate IndexedDB either way,
+ // which is storage-agnostic: it stores an opaque JSON blob.
+ const ndb = wex.ws.nativeSqliteDb;
+ const backup = ndb
+ ? await exportNativeSqliteDb(ndb)
+ : await exportDb(wex.ws.idbFactory);
const backupsDb = await openStoredBackupsDatabase(wex.ws.idbFactory);
const name = `backup-${new Date().getTime()}`;
await backupsDb.runAllStoresReadWriteTx({}, async (tx) => {
@@ -662,15 +673,23 @@ async function recoverStoredBackup(
return backupData;
});
logger.info(`backup found, now importing`);
- await importDb(wex.db.idbHandle(), bd);
- await wex.runLegacyWalletDbTx(async (tx) => {
- await rematerializeTransactions(wex, tx.wtx);
- // Clear fixups. Okay since they are idempotent.
- const fixups = await tx.fixups.getAll();
- for (const f of fixups) {
- await tx.fixups.delete(f.fixupName);
- }
- });
+ const ndbRecover = wex.ws.nativeSqliteDb;
+ if (ndbRecover) {
+ await importNativeSqliteDb(ndbRecover, bd);
+ await wex.runWalletDbTx(async (tx) => {
+ await rematerializeTransactions(wex, tx);
+ });
+ } else {
+ await importDb(wex.db.idbHandle(), bd);
+ await wex.runLegacyWalletDbTx(async (tx) => {
+ await rematerializeTransactions(wex, tx.wtx);
+ // Clear fixups. Okay since they are idempotent.
+ const fixups = await tx.fixups.getAll();
+ for (const f of fixups) {
+ await tx.fixups.delete(f.fixupName);
+ }
+ });
+ }
logger.info(`import done`);
}
@@ -2450,6 +2469,13 @@ const handlers: { [T in WalletApiOperation]: HandlerWithValidator<T> } = {
codec: codecForEmptyObject(),
handler: async (wex, req) => {
await clearDatabase(wex.db.idbHandle());
+ // Also clear the native database when one is in use: clearDatabase
+ // only knows about IndexedDB, so without this the wallet would report
+ // a cleared database while its actual data survived.
+ const ndb = wex.ws.nativeSqliteDb;
+ if (ndb) {
+ await clearNativeSqliteWalletDb(ndb);
+ }
wex.ws.flightRecords = [];
wex.ws.clearAllCaches();
await wex.taskScheduler.reload();
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
@@ -81,6 +81,9 @@ import {
} from "./db-indexeddb.js";
import { WalletDbTransaction } from "./dbtx.js";
import {
+ clearNativeSqliteWalletDb,
+ exportNativeSqliteDb,
+ importNativeSqliteDb,
NativeSqliteWalletDb,
runNativeSqliteWalletTx,
} from "./dbtx-sqlite.js";
@@ -95,6 +98,7 @@ import {
import {
ObservableDbAccess,
ObservableTaskScheduler,
+ getCallerInfo,
observeTalerCrypto,
} from "./observable-wrappers.js";
import { ProgressContext } from "./progress.js";
@@ -414,12 +418,42 @@ async function runWalletDbTx<T>(
// Retries are handled the same way, since the failure modes this guards
// against (transaction aborted, retryable conflict) are not specific to
// the storage layer.
+ //
+ // Observability events are emitted here rather than inherited from
+ // ObservableDbAccess: that wrapper sits on wex.db, which this path does
+ // not touch, so without this the native backend would be invisible to
+ // the observability API.
return await handleTxRetries(wex, async () => {
- return await runNativeSqliteWalletTx(
- ndb,
- (notif) => wex.ws.notify(notif),
- async (tx) => await f(tx),
- );
+ const location = getCallerInfo();
+ wex.oc.observe({
+ type: ObservabilityEventType.DbQueryStart,
+ name: "<unknown>",
+ location,
+ });
+ const start = performanceNow();
+ try {
+ const ret = await runNativeSqliteWalletTx(
+ ndb,
+ (notif) => wex.ws.notify(notif),
+ async (tx) => await f(tx),
+ );
+ wex.oc.observe({
+ type: ObservabilityEventType.DbQueryFinishSuccess,
+ name: "<unknown>",
+ location,
+ durationMs: performanceDelta(start, performanceNow()),
+ });
+ return ret;
+ } catch (e) {
+ wex.oc.observe({
+ type: ObservabilityEventType.DbQueryFinishError,
+ name: "<unknown>",
+ location,
+ error: getErrorDetailFromException(e),
+ durationMs: performanceDelta(start, performanceNow()),
+ });
+ throw e;
+ }
});
}
return await handleTxRetries(wex, async () => {
