commit a28d38bb5d742f153f1655b909d5d58bff686913
parent 87250c4f4ec55a639cb8058c4f249f5d14ac83ec
Author: Florian Dold <dold@taler.net>
Date: Thu, 10 Sep 2026 11:17:17 +0200
wallet-core: avoid redundant notifications
Diffstat:
8 files changed, 790 insertions(+), 41 deletions(-)
diff --git a/packages/taler-util/src/notifications.ts b/packages/taler-util/src/notifications.ts
@@ -191,6 +191,10 @@ export interface BankAccountChangeNotification {
bankAccountId: string;
}
+/**
+ * Invalidates balance data, including flags and refreshInfo. The monetary
+ * amounts need not have changed, for example when a refresh cost becomes ready.
+ */
export interface BalanceChangeNotification {
type: NotificationType.BalanceChange;
diff --git a/packages/taler-wallet-core/src/db/shared.ts b/packages/taler-wallet-core/src/db/shared.ts
@@ -25,13 +25,20 @@
import {
assertUnreachable,
+ canonicalJson,
ScopeInfo,
ScopeType,
encodeCrock,
getRandomBytes,
} from "@gnu-taler/taler-util";
import { WalletDbTransaction } from "./transaction.js";
-import { ConfigRecordKey, PurchaseStatus, WalletPurchase } from "./records.js";
+import {
+ ConfigRecordKey,
+ PurchaseStatus,
+ WalletExchangeEntry,
+ WalletOperationRetry,
+ WalletPurchase,
+} from "./records.js";
import { auditorProvidesVerifiedTrust } from "../auditorTrust.js";
/**
@@ -223,6 +230,72 @@ export async function persistRefreshBalanceInvalidation(
});
}
+/** Scheduling metadata is not an input to exchange summaries or renewal costs. */
+function exchangeCacheInputs(exchange: WalletExchangeEntry | undefined) {
+ if (!exchange) return null;
+ const { nextRefreshCheckStamp, autoRefreshDeadlines, ...inputs } = exchange;
+ return inputs;
+}
+
+async function trackExchangeUpdate(
+ tx: WalletDbTransaction,
+ exchange: WalletExchangeEntry,
+ flag: WalletCacheInvalidation,
+): Promise<void> {
+ const previous = await tx.getExchange(exchange.baseUrl);
+ if (
+ canonicalJson(exchangeCacheInputs(previous)) !==
+ canonicalJson(exchangeCacheInputs(exchange))
+ ) {
+ flag.dirty = true;
+ flag.refreshBalanceDirty = true;
+ }
+}
+
+/** Match the purchase contribution read by getBalancesInsideTransaction. */
+function purchaseBalanceInputs(purchase: WalletPurchase | undefined) {
+ if (
+ !purchase ||
+ (purchase.purchaseStatus !== PurchaseStatus.PendingPaying &&
+ purchase.purchaseStatus !== PurchaseStatus.SuspendedPaying) ||
+ !purchase.payInfo?.payCoinSelection?.coinPubs.length
+ )
+ return null;
+ return {
+ coinPubs: purchase.payInfo.payCoinSelection.coinPubs,
+ coinContributions: purchase.payInfo.payCoinSelection.coinContributions,
+ totalPayCost: purchase.payInfo.totalPayCost,
+ };
+}
+
+async function trackPurchaseUpdate(
+ tx: WalletDbTransaction,
+ proposalId: string,
+ purchase: WalletPurchase | undefined,
+ flag: WalletCacheInvalidation,
+): Promise<void> {
+ const previous = await tx.getPurchase(proposalId);
+ if (
+ canonicalJson(purchaseBalanceInputs(previous)) !==
+ canonicalJson(purchaseBalanceInputs(purchase))
+ )
+ flag.refreshBalanceDirty = true;
+}
+
+async function trackRetryUpdate(
+ tx: WalletDbTransaction,
+ id: string,
+ retry: WalletOperationRetry | undefined,
+ flag: WalletCacheInvalidation,
+): Promise<void> {
+ const previous = await tx.getOperationRetry(id);
+ if (
+ canonicalJson(previous?.lastError ?? null) !==
+ canonicalJson(retry?.lastError ?? null)
+ )
+ flag.refreshBalanceDirty = true;
+}
+
export function watchForCacheInvalidation<T extends WalletDbTransaction>(
tx: T,
flag: WalletCacheInvalidation,
@@ -238,6 +311,25 @@ export function watchForCacheInvalidation<T extends WalletDbTransaction>(
flag.terminalPaymentIds?.add(purchase.proposalId);
} else if (prop === "deletePurchase")
flag.terminalPaymentIds?.add(args[0] as string);
+ // Keep synchronous DAL methods (notably notify and scheduleOnCommit)
+ // synchronous; only these writes need to read their previous inputs.
+ if (prop === "upsertExchange") {
+ return trackExchangeUpdate(
+ target,
+ args[0] as WalletExchangeEntry,
+ flag,
+ ).then(() => value.apply(target, args));
+ }
+ if (prop === "upsertPurchase" || prop === "deletePurchase") {
+ const purchase =
+ prop === "upsertPurchase" ? (args[0] as WalletPurchase) : undefined;
+ return trackPurchaseUpdate(
+ target,
+ purchase?.proposalId ?? (args[0] as string),
+ purchase,
+ flag,
+ ).then(() => value.apply(target, args));
+ }
if (typeof prop === "string" && CACHE_INVALIDATING_METHODS.has(prop))
flag.dirty = true;
if (
@@ -253,8 +345,16 @@ export function watchForCacheInvalidation<T extends WalletDbTransaction>(
prop === "upsertOperationRetry"
? (args[0] as { id: string }).id
: (args[0] as string);
- if (/^(refresh|exchange-auto-refresh|exchange-update):/.test(id))
- flag.refreshBalanceDirty = true;
+ if (/^(refresh|exchange-auto-refresh|exchange-update):/.test(id)) {
+ return trackRetryUpdate(
+ target,
+ id,
+ prop === "upsertOperationRetry"
+ ? (args[0] as WalletOperationRetry)
+ : undefined,
+ flag,
+ ).then(() => value.apply(target, args));
+ }
}
return value.apply(target, args);
};
diff --git a/packages/taler-wallet-core/src/exchanges.ts b/packages/taler-wallet-core/src/exchanges.ts
@@ -3263,15 +3263,18 @@ export async function doExchangeAutoRefresh(
exchange.nextRefreshCheckStamp = timestampPreciseToDb(
AbsoluteTime.toPreciseTimestamp(minCheckThreshold),
);
- wex.ws.exchangeCache.clear();
await tx.upsertExchange(exchange);
const st = getExchangeState(exchange);
- tx.notify({
- type: NotificationType.ExchangeStateTransition,
- exchangeBaseUrl,
- causeHint: "auto-refresh",
- newExchangeState: st,
- oldExchangeState: st,
+ // waitReadyExchange can be waiting for this scheduling update even when
+ // the public exchange state did not change. Wake it only after commit.
+ tx.scheduleOnCommit(() => {
+ wex.ws.notifyInternal({
+ type: NotificationType.ExchangeStateTransition,
+ exchangeBaseUrl,
+ causeHint: "auto-refresh",
+ newExchangeState: st,
+ oldExchangeState: st,
+ });
});
});
}
diff --git a/packages/taler-wallet-core/src/pay-merchant.ts b/packages/taler-wallet-core/src/pay-merchant.ts
@@ -1779,14 +1779,25 @@ async function processDownloadProposal(
p.purchaseStatus = PurchaseStatus.DoneRepurchaseDetected;
p.repurchaseProposalId = repurchase.proposalId;
await startPayReplay(wex, tx, repurchase.proposalId, p.downloadSessionId);
+ } else if (isSharedPurchase(p)) {
+ p.purchaseStatus = PurchaseStatus.DialogShared;
+ } else if (
+ AbsoluteTime.isExpired(getProposalExpiry(contractData.contractTerms))
+ ) {
+ p.timestampExpired = timestampPreciseToDb(TalerPreciseTimestamp.now());
+ p.purchaseStatus = PurchaseStatus.Expired;
} else {
- p.purchaseStatus = isSharedPurchase(p)
- ? PurchaseStatus.DialogShared
- : PurchaseStatus.DialogProposed;
+ p.purchaseStatus = PurchaseStatus.DialogProposed;
}
// Downloading the proposal only presents the offer to the user, no
// funds are committed until the payment is confirmed.
- await h.update(p, "download-proposal", BalanceEffect.None);
+ await h.update(
+ p,
+ p.purchaseStatus === PurchaseStatus.Expired
+ ? "proposal-expired"
+ : "download-proposal",
+ BalanceEffect.None,
+ );
});
return TaskRunResult.progress();
@@ -5137,6 +5148,13 @@ async function checkIfOrderIsAlreadyPaid(
}
}
+function getProposalExpiry(contractTerms: MerchantContractTerms): AbsoluteTime {
+ return AbsoluteTime.addDuration(
+ AbsoluteTime.fromProtocolTimestamp(contractTerms.pay_deadline),
+ Duration.fromSpec({ seconds: 10 }),
+ );
+}
+
async function processPurchaseDialogProposed(
wex: WalletExecutionContext,
purchase: WalletPurchase,
@@ -5166,14 +5184,7 @@ async function processPurchaseDialogProposed(
// Transaction longer exists.
return TaskRunResult.finished();
}
- const payDeadline = AbsoluteTime.fromProtocolTimestamp(
- txRes.download.contractTerms.pay_deadline,
- );
-
- const expiry = AbsoluteTime.addDuration(
- payDeadline,
- Duration.fromSpec({ seconds: 10 }),
- );
+ const expiry = getProposalExpiry(txRes.download.contractTerms);
if (AbsoluteTime.isExpired(expiry)) {
await wex.runWalletDbTx(async (tx) => {
@@ -5183,7 +5194,7 @@ async function processPurchaseDialogProposed(
}
r2.timestampExpired = timestampPreciseToDb(TalerPreciseTimestamp.now());
r2.purchaseStatus = PurchaseStatus.Expired;
- await h.update(r2, "proposal-expired");
+ await h.update(r2, "proposal-expired", BalanceEffect.None);
});
return TaskRunResult.progress();
}
diff --git a/packages/taler-wallet-core/src/refreshBalance.test.ts b/packages/taler-wallet-core/src/refreshBalance.test.ts
@@ -300,6 +300,7 @@ test("cache writes and maintenance retries do not invalidate the renewal report"
{
upsertConfig: async () => {},
deleteOperationRetry: async () => {},
+ getOperationRetry: async () => undefined,
} as unknown as WalletDbTransaction,
flag,
);
@@ -307,5 +308,59 @@ test("cache writes and maintenance retries do not invalidate the renewal report"
await tx.deleteOperationRetry("refresh-balance:");
assert.equal(flag.refreshBalanceDirty, undefined);
await tx.deleteOperationRetry("exchange-auto-refresh:https://exchange/");
- assert.equal(flag.refreshBalanceDirty, true);
+ assert.equal(flag.refreshBalanceDirty, undefined);
});
+
+for (const makeRunner of runnerFactories) {
+ test(`${makeRunner.name}: publication notifies when checking finishes, including unchanged results`, async () => {
+ const runner = await makeRunner();
+ const notifications: unknown[] = [];
+ runner.setNotificationSink((n) => notifications.push(n));
+ try {
+ const { saved, response } = fixture();
+ await runner.runReadWriteTx((tx) => publishRefreshBalance(tx, saved));
+ assert.equal(notifications.length, 1);
+ await runner.runReadWriteTx((tx) => publishRefreshBalance(tx, saved));
+ assert.equal(notifications.length, 1);
+ await runner.runReadWriteTx(async (tx) => {
+ await tx.upsertConfig({
+ key: ConfigRecordKey.RefreshBalanceGeneration,
+ value: "changed",
+ });
+ await readRefreshBalanceInfo(tx, response);
+ });
+ assert.equal(
+ response.balances[0].refreshInfo?.annualCostBound.status,
+ "unavailable",
+ );
+ await runner.runReadWriteTx(async (tx) => {
+ assert.equal(await publishRefreshBalance(tx, saved), false);
+ assert.equal(
+ await publishRefreshBalance(tx, { ...saved, generation: "changed" }),
+ true,
+ );
+ await readRefreshBalanceInfo(tx, response);
+ });
+ assert.equal(
+ response.balances[0].refreshInfo?.annualCostBound.status,
+ "available",
+ );
+ assert.equal(notifications.length, 2);
+ await runner.runReadWriteTx((tx) =>
+ publishRefreshBalance(tx, {
+ ...saved,
+ generation: "changed",
+ computedAt: saved.nextCheck,
+ nextCheck: saved.nextCheck + 60_000,
+ }),
+ );
+ assert.equal(
+ notifications.length,
+ 3,
+ "time-based recomputation also finishes checking",
+ );
+ } finally {
+ await runner.close();
+ }
+ });
+}
diff --git a/packages/taler-wallet-core/src/shepherd.ts b/packages/taler-wallet-core/src/shepherd.ts
@@ -418,15 +418,18 @@ export class TaskSchedulerImpl implements TaskScheduler {
async resetTask(taskId: TaskIdStr): Promise<void> {
await this.ws.runStandaloneWalletDbTx(async (tx) => {
logger.trace(`storing task [reset] for ${taskId}`);
+ const hadError = (await tx.getOperationRetry(taskId))?.lastError;
await tx.deleteOperationRetry(taskId);
- const notif = await taskToRetryNotification(
- this.ws,
- tx,
- taskId,
- undefined,
- );
- if (notif) {
- tx.notify(notif);
+ if (hadError) {
+ const notif = await taskToRetryNotification(
+ this.ws,
+ tx,
+ taskId,
+ undefined,
+ );
+ if (notif) {
+ tx.notify(notif);
+ }
}
});
this.stopShepherdTask(taskId);
diff --git a/packages/taler-wallet-core/src/wallet-notifications.test.ts b/packages/taler-wallet-core/src/wallet-notifications.test.ts
@@ -0,0 +1,550 @@
+/*
+ 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.
+
+ GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import {
+ AbsoluteTime,
+ CancellationToken,
+ ContractTermsUtil,
+ encodeCrock,
+ NotificationType,
+ openPromise,
+ SetTimeoutTimerAPI,
+ TalerErrorCode,
+ TransactionMajorState,
+ TransactionMinorState,
+ WalletNotification,
+} from "@gnu-taler/taler-util";
+import {
+ dummyHttpResponse,
+ HttpRequestLibrary,
+} from "@gnu-taler/taler-util/http";
+import { DbRetryInfo, TaskIdentifiers, TaskIdStr } from "./common.js";
+import { SynchronousCryptoWorkerFactoryPlain } from "./crypto/workers/synchronousWorkerFactoryPlain.js";
+import {
+ ConfigRecordKey,
+ ExchangeEntryDbRecordStatus,
+ ExchangeEntryDbUpdateStatus,
+ PurchaseStatus,
+ timestampPreciseToDb,
+ WalletExchangeDetails,
+ WalletExchangeEntry,
+ WalletPurchase,
+} from "./db/records.js";
+import { runnerFactories } from "./db/testing/runners.js";
+import { doExchangeAutoRefresh, waitReadyExchange } from "./exchanges.js";
+import { preparePayForUriV2, processPurchase } from "./pay-merchant.js";
+import { processRefreshBalance } from "./refreshBalance.js";
+import { TaskSchedulerImpl } from "./shepherd.js";
+import {
+ applyRunConfigDefaults,
+ getNormalWalletExecutionContext,
+ InternalWalletState,
+ WalletExecutionContext,
+} from "./wallet.js";
+
+const exchangeUrl = "https://exchange.example/";
+const merchantUrl = "https://merchant.example/";
+const master = encodeCrock(new Uint8Array(32));
+const signature = encodeCrock(new Uint8Array(64));
+const tick = () => new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+/** Exercise real reset logic while controlling when background tasks run. */
+class RecordingScheduler extends TaskSchedulerImpl {
+ starts: TaskIdStr[] = [];
+ stops: TaskIdStr[] = [];
+ override async ensureRunning() {}
+ override startShepherdTask(id: TaskIdStr) {
+ this.starts.push(id);
+ }
+ override stopShepherdTask(id: TaskIdStr) {
+ this.stops.push(id);
+ }
+ override getActiveTasks() {
+ return [TaskIdentifiers.forExchangeAutoRefreshFromUrl(exchangeUrl)];
+ }
+}
+
+function contract(deadline: number, nonce = master) {
+ return {
+ amount: "TESTKUDOS:1",
+ max_fee: "TESTKUDOS:0",
+ nonce,
+ h_wire: "wire",
+ exchanges: [],
+ fulfillment_url: `${merchantUrl}article`,
+ merchant_pub: master,
+ merchant: { name: "Shop" },
+ order_id: "order",
+ pay_deadline: { t_s: deadline },
+ wire_transfer_deadline: { t_s: deadline + 3600 },
+ merchant_base_url: merchantUrl,
+ refund_deadline: { t_s: deadline + 3600 },
+ summary: "Article",
+ timestamp: { t_s: 1 },
+ wire_method: "iban",
+ };
+}
+
+function purchase(id = "proposal"): WalletPurchase {
+ return {
+ proposalId: id,
+ orderId: "order",
+ merchantBaseUrl: merchantUrl,
+ purchaseStatus: PurchaseStatus.PendingDownloadingProposal,
+ noncePriv: master,
+ noncePub: master,
+ timestamp: 1,
+ downloadSessionId: "session",
+ shared: false,
+ createdFromShared: false,
+ } as WalletPurchase;
+}
+
+async function fixture(makeRunner: (typeof runnerFactories)[number]) {
+ const db = await makeRunner();
+ let responseBody: unknown;
+ const http = {
+ async fetch(url, options) {
+ assert.equal(url, `${merchantUrl}orders/order/claim`);
+ return {
+ ...dummyHttpResponse,
+ status: 200,
+ requestUrl: url,
+ requestMethod: options?.method ?? "POST",
+ json: async () => structuredClone(responseBody),
+ };
+ },
+ } as HttpRequestLibrary;
+ const ws = new InternalWalletState(
+ db,
+ () => http,
+ new SetTimeoutTimerAPI(),
+ new SynchronousCryptoWorkerFactoryPlain(),
+ );
+ ws.initWithConfig(
+ applyRunConfigDefaults({
+ lazyTaskLoop: true,
+ testing: { skipDefaults: true },
+ }),
+ );
+ ws.initCalled = true;
+ const scheduler = new RecordingScheduler(ws);
+ ws.taskScheduler = scheduler;
+ const cts = CancellationToken.create();
+ const baseWex = getNormalWalletExecutionContext(ws, cts.token, cts, {
+ observe() {},
+ });
+ // These tests exercise verified claim processing, not signature generation.
+ const wex: WalletExecutionContext = {
+ ...baseWex,
+ cryptoApi: {
+ ...baseWex.cryptoApi,
+ isValidContractTermsSignature: async () => ({ valid: true }),
+ },
+ };
+ const notifications: WalletNotification[] = [];
+ ws.addPublicNotificationListener((n) => notifications.push(n));
+ const stamp = timestampPreciseToDb(
+ AbsoluteTime.toPreciseTimestamp(AbsoluteTime.now()),
+ );
+ const exchange: WalletExchangeEntry = {
+ baseUrl: exchangeUrl,
+ entryStatus: ExchangeEntryDbRecordStatus.Used,
+ updateStatus: ExchangeEntryDbUpdateStatus.Ready,
+ detailsPointer: {
+ currency: "TESTKUDOS",
+ masterPublicKey: master,
+ updateClock: stamp,
+ },
+ tosAcceptedEtag: undefined,
+ tosAcceptedTimestamp: undefined,
+ tosCurrentEtag: undefined,
+ lastKeysEtag: undefined,
+ lastUpdate: stamp,
+ nextUpdateStamp: stamp,
+ nextRefreshCheckStamp: 0 as WalletExchangeEntry["nextRefreshCheckStamp"],
+ };
+ const details: WalletExchangeDetails = {
+ exchangeBaseUrl: exchangeUrl,
+ masterPublicKey: master,
+ currency: "TESTKUDOS",
+ auditors: [],
+ protocolVersionRange: "18:0:1",
+ tinyAmount: "TESTKUDOS:0.01",
+ reserveClosingDelay: { d_us: 1000 },
+ globalFees: [],
+ wireInfo: { accounts: [], feesForType: {} },
+ bankComplianceLanguage: undefined,
+ defaultPeerPushExpiration: undefined,
+ };
+ await db.runReadWriteTx(async (tx) => {
+ await tx.upsertExchangeDetails(details);
+ await tx.upsertExchange(exchange);
+ });
+ await processRefreshBalance(wex);
+ await tick();
+ notifications.length = 0;
+ scheduler.starts.length = 0;
+ return {
+ db,
+ ws,
+ wex,
+ scheduler,
+ notifications,
+ exchange,
+ reply: (body: unknown) => {
+ responseBody = body;
+ },
+ generation: () =>
+ db.runReadWriteTx(
+ async (tx) =>
+ (await tx.getConfig(ConfigRecordKey.RefreshBalanceGeneration))
+ ?.value ?? "",
+ ),
+ close: () => ws.shutdown(),
+ };
+}
+
+for (const makeRunner of runnerFactories) {
+ test(
+ `${makeRunner.name}: auto-refresh wakes internal waiters without frontend reloads`,
+ { timeout: 5000 },
+ async () => {
+ const f = await fixture(makeRunner);
+ try {
+ const checked = openPromise<void>();
+ let reads = 0;
+ const waitWex: WalletExecutionContext = {
+ ...f.wex,
+ async runWalletDbTx(fn) {
+ const result = await f.wex.runWalletDbTx(fn);
+ if (++reads === 2) checked.resolve();
+ return result;
+ },
+ };
+ let settled = false;
+ const waiting = waitReadyExchange(waitWex, exchangeUrl, {
+ waitAutoRefresh: true,
+ }).then((value) => {
+ settled = true;
+ return value;
+ });
+ await checked.promise;
+ assert.equal(settled, false);
+ const generation = await f.generation();
+ await doExchangeAutoRefresh(f.wex, exchangeUrl);
+ assert.equal((await waiting).exchangeBaseUrl, exchangeUrl);
+ await tick();
+ assert.deepEqual(f.notifications, []);
+ assert.deepEqual(f.scheduler.starts, []);
+ assert.equal(await f.generation(), generation);
+ assert.ok(f.ws.exchangeCache.get(exchangeUrl));
+ await doExchangeAutoRefresh(f.wex, exchangeUrl);
+ assert.ok(
+ f.ws.exchangeCache.get(exchangeUrl),
+ "scheduling preserves cached exchange information",
+ );
+
+ const cancel = CancellationToken.create();
+ await f.db.runReadWriteTx((tx) => tx.upsertExchange(f.exchange));
+ const cancelled = waitReadyExchange(
+ { ...f.wex, cancellationToken: cancel.token },
+ exchangeUrl,
+ { waitAutoRefresh: true },
+ );
+ const rejection = assert.rejects(cancelled);
+ cancel.cancel("test cancellation");
+ await rejection;
+ } finally {
+ await f.close();
+ }
+ },
+ );
+
+ test(`${makeRunner.name}: rolled-back auto-refresh does not publish completion`, async () => {
+ const f = await fixture(makeRunner);
+ try {
+ const internal: WalletNotification[] = [];
+ f.ws.addNotificationListener((n) => internal.push(n));
+ const failingWex: WalletExecutionContext = {
+ ...f.wex,
+ runWalletDbTx: (fn) =>
+ f.wex.runWalletDbTx(async (tx) => {
+ const before = (await tx.getExchange(exchangeUrl))!
+ .nextRefreshCheckStamp;
+ const result = await fn(tx);
+ if (
+ (await tx.getExchange(exchangeUrl))!.nextRefreshCheckStamp !==
+ before
+ )
+ throw Error("rollback auto-refresh");
+ return result;
+ }),
+ };
+ await assert.rejects(
+ doExchangeAutoRefresh(failingWex, exchangeUrl),
+ /rollback auto-refresh/,
+ );
+ await tick();
+ assert.deepEqual(internal, []);
+ assert.deepEqual(f.notifications, []);
+ assert.equal(
+ (await f.db.runReadWriteTx((tx) => tx.getExchange(exchangeUrl)))!
+ .nextRefreshCheckStamp,
+ 0,
+ );
+ } finally {
+ await f.close();
+ }
+ });
+
+ test(`${makeRunner.name}: proposal creation and retry clearing do not wake unrelated refresh tasks`, async () => {
+ const f = await fixture(makeRunner);
+ try {
+ const prepared = await preparePayForUriV2(
+ f.wex,
+ "taler://pay/merchant.example/order/session",
+ );
+ await tick();
+ assert.equal(f.notifications.length, 1);
+ assert.equal(
+ f.notifications[0].type,
+ NotificationType.TransactionStateTransition,
+ );
+ const record = (
+ await f.db.runReadWriteTx((tx) => tx.listAllPurchases())
+ )[0];
+ const id = TaskIdentifiers.forPay(record);
+ assert.deepEqual(f.scheduler.starts, [id]);
+ assert.equal(await f.generation(), "");
+ f.notifications.length = 0;
+ f.scheduler.starts.length = 0;
+
+ const state = {
+ major: TransactionMajorState.Pending,
+ minor: TransactionMinorState.ClaimProposal,
+ working: true,
+ };
+ const error = {
+ code: TalerErrorCode.WALLET_UNEXPECTED_REQUEST_ERROR,
+ hint: "HTTP 500",
+ };
+ await f.wex.runWalletDbTx(async (tx) => {
+ await tx.upsertOperationRetry({
+ id,
+ lastError: error,
+ retryInfo: DbRetryInfo.reset(),
+ });
+ tx.notify({
+ type: NotificationType.TransactionStateTransition,
+ causeHint: "shepherd-retry",
+ transactionId: prepared.transactionId,
+ oldTxState: state,
+ newTxState: state,
+ newStId: record.purchaseStatus,
+ errorInfo: error,
+ });
+ });
+ await tick();
+ assert.equal(
+ f.notifications.length,
+ 1,
+ "an error remains visible without a state-label change",
+ );
+ assert.deepEqual(f.scheduler.starts, []);
+ await f.scheduler.resetTask(id);
+ await tick();
+ assert.equal(f.notifications.length, 2);
+ const clearing = f.notifications[1];
+ assert.equal(clearing.type, NotificationType.TransactionStateTransition);
+ if (clearing.type === NotificationType.TransactionStateTransition)
+ assert.equal(clearing.errorInfo, undefined);
+ await f.scheduler.resetTask(id);
+ await tick();
+ assert.equal(
+ f.notifications.length,
+ 2,
+ "resetting again has no error to clear",
+ );
+ assert.deepEqual(f.scheduler.starts, [id, id]);
+ assert.equal(await f.generation(), "");
+
+ f.ws.notify({
+ type: NotificationType.BalanceChange,
+ hintTransactionId: prepared.transactionId,
+ isInternal: true,
+ });
+ assert.equal(
+ f.scheduler.starts.at(-1),
+ TaskIdentifiers.forExchangeAutoRefreshFromUrl(exchangeUrl),
+ );
+ } finally {
+ await tick();
+ await f.close();
+ }
+ });
+
+ test(`${makeRunner.name}: refresh report ignores bookkeeping but tracks balance and error changes`, async () => {
+ const f = await fixture(makeRunner);
+ try {
+ const p = purchase();
+ const mutate = async (
+ fn: Parameters<WalletExecutionContext["runWalletDbTx"]>[0],
+ changes: boolean,
+ ) => {
+ const before = await f.generation();
+ await f.wex.runWalletDbTx(fn);
+ assert.equal((await f.generation()) !== before, changes);
+ };
+ await mutate((tx) => tx.upsertPurchase(p), false);
+ p.purchaseStatus = PurchaseStatus.DialogProposed;
+ await mutate((tx) => tx.upsertPurchase(p), false);
+ p.purchaseStatus = PurchaseStatus.Expired;
+ await mutate((tx) => tx.upsertPurchase(p), false);
+ await mutate((tx) => tx.deletePurchase(p.proposalId), false);
+ p.purchaseStatus = PurchaseStatus.PendingPaying;
+ p.payInfo = {
+ totalPayCost: "TESTKUDOS:1",
+ payCoinSelection: {
+ coinPubs: [master],
+ coinContributions: ["TESTKUDOS:1"],
+ } as NonNullable<WalletPurchase["payInfo"]>["payCoinSelection"],
+ };
+ await mutate((tx) => tx.upsertPurchase(p), true);
+ p.purchaseStatus = PurchaseStatus.SuspendedPaying;
+ await mutate((tx) => tx.upsertPurchase(p), false);
+ p.payInfo.totalPayCost = "TESTKUDOS:2";
+ p.payInfo.payCoinSelection!.coinContributions[0] = "TESTKUDOS:2";
+ await mutate((tx) => tx.upsertPurchase(p), true);
+ await mutate((tx) => tx.deletePurchase(p.proposalId), true);
+
+ const exchange = {
+ ...f.exchange,
+ nextRefreshCheckStamp: timestampPreciseToDb(
+ AbsoluteTime.toPreciseTimestamp(AbsoluteTime.now()),
+ ),
+ autoRefreshDeadlines: { denom: 123 },
+ };
+ await mutate((tx) => tx.upsertExchange(exchange), false);
+ exchange.tosCurrentEtag = "new-terms";
+ await mutate((tx) => tx.upsertExchange(exchange), true);
+ for (const id of [
+ "refresh:group",
+ `exchange-auto-refresh:${exchangeUrl}`,
+ `exchange-update:${exchangeUrl}`,
+ ]) {
+ const retry = {
+ id,
+ retryInfo: DbRetryInfo.reset(),
+ lastError: undefined as { code: TalerErrorCode } | undefined,
+ };
+ await mutate((tx) => tx.upsertOperationRetry(retry), false);
+ retry.retryInfo.retryCounter++;
+ await mutate((tx) => tx.upsertOperationRetry(retry), false);
+ retry.lastError = { code: TalerErrorCode.WALLET_NETWORK_ERROR };
+ await mutate((tx) => tx.upsertOperationRetry(retry), true);
+ retry.retryInfo.retryCounter++;
+ await mutate((tx) => tx.upsertOperationRetry(retry), false);
+ await mutate((tx) => tx.deleteOperationRetry(id), true);
+ await mutate((tx) => tx.deleteOperationRetry(id), false);
+ }
+ } finally {
+ await f.close();
+ }
+ });
+
+ test(`${makeRunner.name}: downloaded proposals expire atomically with the existing grace period`, async (t) => {
+ const f = await fixture(makeRunner);
+ try {
+ const now = 1_800_000_000;
+ t.mock.timers.enable({ apis: ["Date"], now: now * 1000 });
+ for (const [name, deadline, shared, expected] of [
+ ["expired", now - 11, false, PurchaseStatus.Expired],
+ ["grace", now - 9, false, PurchaseStatus.DialogProposed],
+ ["boundary", now - 10, false, PurchaseStatus.Expired],
+ ["valid", now + 60, false, PurchaseStatus.DialogProposed],
+ ["shared", now - 11, true, PurchaseStatus.DialogShared],
+ ] as const) {
+ const p = { ...purchase(name), shared };
+ await f.db.runReadWriteTx((tx) => tx.upsertPurchase(p));
+ f.reply({ contract_terms: contract(deadline), sig: signature });
+ f.notifications.length = 0;
+ f.scheduler.starts.length = 0;
+ const generation = await f.generation();
+ await processPurchase(f.wex, name);
+ await tick();
+ const saved = (await f.db.runReadWriteTx((tx) =>
+ tx.getPurchase(name),
+ ))!;
+ assert.equal(saved.purchaseStatus, expected, name);
+ assert.ok(saved.download);
+ assert.equal(f.notifications.length, 1, name);
+ const n = f.notifications[0];
+ assert.equal(n.type, NotificationType.TransactionStateTransition);
+ if (n.type === NotificationType.TransactionStateTransition)
+ assert.equal(
+ n.newTxState.major,
+ expected === PurchaseStatus.Expired
+ ? TransactionMajorState.Expired
+ : TransactionMajorState.Dialog,
+ );
+ assert.equal(await f.generation(), generation);
+ assert.deepEqual(f.scheduler.starts, []);
+ }
+
+ const p = purchase("scheduled-expiry");
+ p.purchaseStatus = PurchaseStatus.DialogProposed;
+ const terms = contract(now - 11);
+ const hash = ContractTermsUtil.hashContractTerms(terms);
+ p.download = {
+ contractTermsHash: hash,
+ contractTermsMerchantSig: signature,
+ currency: "TESTKUDOS",
+ fulfillmentUrl: terms.fulfillment_url,
+ };
+ await f.db.runReadWriteTx(async (tx) => {
+ await tx.upsertContractTerms({ h: hash, contractTermsRaw: terms });
+ await tx.upsertPurchase(p);
+ });
+ f.notifications.length = 0;
+ await processPurchase(f.wex, p.proposalId);
+ await tick();
+ assert.equal(f.notifications.length, 1);
+ assert.equal(
+ f.notifications[0].type,
+ NotificationType.TransactionStateTransition,
+ );
+
+ p.purchaseStatus = PurchaseStatus.Done;
+ await f.db.runReadWriteTx((tx) => tx.upsertPurchase(p));
+ const repurchase = purchase("repurchase");
+ await f.db.runReadWriteTx((tx) => tx.upsertPurchase(repurchase));
+ f.reply({ contract_terms: terms, sig: signature });
+ await processPurchase(f.wex, repurchase.proposalId);
+ assert.equal(
+ (await f.db.runReadWriteTx((tx) =>
+ tx.getPurchase(repurchase.proposalId),
+ ))!.purchaseStatus,
+ PurchaseStatus.DoneRepurchaseDetected,
+ );
+ } finally {
+ await tick();
+ await f.close();
+ }
+ });
+}
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
@@ -1245,7 +1245,7 @@ export class Wallet {
}
addNotificationListener(f: (n: WalletNotification) => void): CancelFn {
- return this.ws.addNotificationListener(f);
+ return this.ws.addPublicNotificationListener(f);
}
async handleCoreApiRequest(
@@ -1270,6 +1270,7 @@ export class InternalWalletState {
stopped = false;
private listeners: NotificationListener[] = [];
+ private publicListeners: NotificationListener[] = [];
initCalled = false;
@@ -1793,9 +1794,8 @@ export class InternalWalletState {
notify(n: WalletNotification): void {
logger.trace(`Notification: ${j2s(n)}`);
if (
- (n.type === NotificationType.BalanceChange &&
- n.hintTransactionId !== "refresh-balance") ||
- n.type === NotificationType.TransactionStateTransition
+ n.type === NotificationType.BalanceChange &&
+ n.hintTransactionId !== "refresh-balance"
) {
for (const id of this.taskScheduler.getActiveTasks()) {
if (id.startsWith(`${PendingTaskType.ExchangeAutoRefresh}:`))
@@ -1813,7 +1813,19 @@ export class InternalWalletState {
this.maintenanceNotifications.offer(n);
return;
}
- for (const l of this.listeners) {
+ this.enqueueNotification(n, [...this.listeners, ...this.publicListeners]);
+ }
+
+ /** Wake core waiters without requesting a frontend reload. */
+ notifyInternal(n: WalletNotification): void {
+ this.enqueueNotification(n, this.listeners);
+ }
+
+ private enqueueNotification(
+ n: WalletNotification,
+ listeners: readonly NotificationListener[],
+ ): void {
+ for (const l of listeners) {
const nc = JSON.parse(JSON.stringify(n));
setTimeout(() => {
l(nc);
@@ -1822,7 +1834,7 @@ export class InternalWalletState {
}
private deliverNotificationSynchronously(n: WalletNotification): void {
- for (const l of [...this.listeners]) {
+ for (const l of [...this.listeners, ...this.publicListeners]) {
const nc = JSON.parse(JSON.stringify(n));
try {
l(nc);
@@ -1835,11 +1847,22 @@ export class InternalWalletState {
}
addNotificationListener(f: (n: WalletNotification) => void): CancelFn {
- this.listeners.push(f);
+ return this.registerNotificationListener(this.listeners, f);
+ }
+
+ addPublicNotificationListener(f: NotificationListener): CancelFn {
+ return this.registerNotificationListener(this.publicListeners, f);
+ }
+
+ private registerNotificationListener(
+ listeners: NotificationListener[],
+ f: NotificationListener,
+ ): CancelFn {
+ listeners.push(f);
return () => {
- const idx = this.listeners.indexOf(f);
+ const idx = listeners.indexOf(f);
if (idx >= 0) {
- this.listeners.splice(idx, 1);
+ listeners.splice(idx, 1);
}
};
}
