commit 1c7787e59c127624bd320de333bf79b24de4ede2
parent e156a42b70758b2dca5fe7dfe4e3f74a446ef47f
Author: Florian Dold <florian@dold.me>
Date: Tue, 14 Jul 2026 22:35:17 +0200
wallet-core: move progressToken logic to separate file, cancellation/retry implementation WIP
Diffstat:
3 files changed, 186 insertions(+), 96 deletions(-)
diff --git a/packages/taler-wallet-core/src/exchanges.ts b/packages/taler-wallet-core/src/exchanges.ts
@@ -172,6 +172,7 @@ import { PeerPullCreditTransactionContext } from "./pay-peer-pull-credit.js";
import { PeerPullDebitTransactionContext } from "./pay-peer-pull-debit.js";
import { PeerPushCreditTransactionContext } from "./pay-peer-push-credit.js";
import { PeerPushDebitTransactionContext } from "./pay-peer-push-debit.js";
+import { ProgressContext } from "./progress.js";
import { RecoupTransactionContext, createRecoupGroup } from "./recoup.js";
import { RefreshTransactionContext, createRefreshGroup } from "./refresh.js";
import {
@@ -184,7 +185,6 @@ import { WALLET_EXCHANGE_PROTOCOL_VERSION } from "./versions.js";
import {
InternalWalletState,
LegacyWalletTxHandle,
- ProgressContext,
WalletExecutionContext,
walletExchangeClient,
} from "./wallet.js";
diff --git a/packages/taler-wallet-core/src/progress.ts b/packages/taler-wallet-core/src/progress.ts
@@ -0,0 +1,177 @@
+/*
+ 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 {
+ CancellationToken,
+ CancelProgressTokenRequest,
+ Duration,
+ EmptyObject,
+ Logger,
+ NotificationType,
+ safeStringifyException,
+} from "@gnu-taler/taler-util";
+import { WalletExecutionContext } from "./wallet.js";
+
+const logger = new Logger("progress.ts");
+
+export interface ProgressContext {
+ progressToken: string;
+ operation: string;
+ finished: boolean;
+ cts: CancellationToken.Source | undefined;
+ onRetryNow?: () => Promise<void>;
+}
+
+function getProgressKey(args: {
+ operation: string;
+ progressToken: string;
+}): string {
+ return `${args.operation}:${encodeURIComponent(args.progressToken)}`;
+}
+
+export async function withMaybeProgressContext<T>(
+ wex: WalletExecutionContext,
+ op: string,
+ tok: string | undefined,
+ onRetryNow: undefined | (() => Promise<void>),
+ f: (pc?: ProgressContext) => Promise<T>,
+): Promise<T> {
+ if (tok != null) {
+ const pc: ProgressContext = {
+ operation: op,
+ progressToken: tok,
+ finished: false,
+ onRetryNow,
+ cts: wex.cts,
+ };
+ const key = getProgressKey(pc);
+ try {
+ if (!wex.cts) {
+ logger.warn(`no cancellation source for ${op}:${tok}`);
+ }
+ wex.ws.progressMap.set(key, pc);
+ emitProgressPhaseNotifications(wex, pc).catch((e) => {
+ logger.warn(safeStringifyException(e));
+ });
+ const res = await f(pc);
+ pc.finished = true;
+ return res;
+ } finally {
+ wex.ws.progressMap.delete(key);
+ // Clean up progress token registration
+ }
+ } else {
+ return await f(undefined);
+ }
+}
+
+/**
+ * Emit progress notifications until the request is finished/cancelled.
+ */
+async function emitProgressPhaseNotifications(
+ wex: WalletExecutionContext,
+ pc: ProgressContext,
+): Promise<void> {
+ await wex.ws.timerGroup.resolveAfter(
+ Duration.fromSpec({
+ seconds: 5,
+ }),
+ );
+ if (wex.cancellationToken.isCancelled || pc.finished) {
+ return;
+ }
+ wex.ws.notify({
+ type: NotificationType.RequestProgressPhase,
+ operation: pc.operation,
+ progressToken: pc.progressToken,
+ phase: "delayed",
+ });
+ await wex.ws.timerGroup.resolveAfter(
+ Duration.fromSpec({
+ seconds: 5,
+ }),
+ );
+ if (wex.cancellationToken.isCancelled || pc.finished) {
+ return;
+ }
+ wex.ws.notify({
+ type: NotificationType.RequestProgressPhase,
+ operation: pc.operation,
+ progressToken: pc.progressToken,
+ phase: "stalled",
+ });
+}
+
+async function emitProgressPhaseRetryNotification(
+ wex: WalletExecutionContext,
+ pc: ProgressContext,
+): Promise<void> {
+ await wex.ws.timerGroup.resolveAfter(
+ Duration.fromSpec({
+ seconds: 3,
+ }),
+ );
+ if (wex.cancellationToken.isCancelled || pc.finished) {
+ return;
+ }
+ wex.ws.notify({
+ type: NotificationType.RequestProgressPhase,
+ operation: pc.operation,
+ progressToken: pc.progressToken,
+ phase: "retryable",
+ });
+}
+
+export async function handleCancelProgressToken(
+ wex: WalletExecutionContext,
+ req: CancelProgressTokenRequest,
+): Promise<EmptyObject> {
+ const key = getProgressKey({
+ operation: req.operation,
+ progressToken: req.progressToken,
+ });
+ const pc = wex.ws.progressMap.get(key);
+ if (!pc) {
+ logger.warn("unable to cancel progress token, not found");
+ return {};
+ }
+ pc.cts?.cancel();
+ return {};
+}
+
+export async function handleRetryProgressTokenNow(
+ wex: WalletExecutionContext,
+ req: CancelProgressTokenRequest,
+): Promise<EmptyObject> {
+ const key = getProgressKey({
+ operation: req.operation,
+ progressToken: req.progressToken,
+ });
+ const pc = wex.ws.progressMap.get(key);
+ if (!pc) {
+ logger.warn("unable to retry progress token, not found");
+ return {};
+ }
+ if (!pc.onRetryNow) {
+ logger.warn("unable to retry progress token, no retry logic defined");
+ return {};
+ }
+ await pc.onRetryNow();
+ emitProgressPhaseRetryNotification(wex, pc).catch((e) => {
+ logger.warn(safeStringifyException(e));
+ });
+ return {};
+}
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
@@ -238,6 +238,7 @@ import {
codecForListDiscountsRequest,
codecForListExchangesRequest,
codecForListSubscriptionsRequest,
+ codecForMailboxBaseUrl,
codecForMailboxConfiguration,
codecForPrepareBankIntegratedWithdrawalRequest,
codecForPreparePayRequest,
@@ -259,7 +260,6 @@ import {
codecForSharePaymentRequest,
codecForStartExchangeWalletKycRequest,
codecForStartRefundQueryRequest,
- codecForMailboxBaseUrl,
codecForSuspendTransaction,
codecForTestPayArgs,
codecForTestingCorruptWithdrawalCoinSelRequest,
@@ -404,6 +404,7 @@ import {
} from "./pay-peer-push-debit.js";
import { checkPayForTemplate } from "./pay-template.js";
import { fillDefaults } from "./preset-exchanges.js";
+import { handleCancelProgressToken, handleRetryProgressTokenNow, ProgressContext, withMaybeProgressContext } from "./progress.js";
import {
AfterCommitInfo,
DbAccess,
@@ -536,12 +537,6 @@ export function walletMerchantClient(
);
}
-export interface ProgressContext {
- progressToken: string;
- operation: string;
- finished: boolean;
-}
-
export const EXCHANGE_COINS_LOCK = "exchange-coins-lock";
export const EXCHANGE_RESERVES_LOCK = "exchange-reserves-lock";
@@ -900,77 +895,6 @@ async function recoverStoredBackup(
logger.info(`import done`);
}
-/**
- * Emit progress notifications until the request is finished/cancelled.
- */
-async function emitProgressPhaseNotifications(
- wex: WalletExecutionContext,
- pc: ProgressContext,
-): Promise<void> {
- await wex.ws.timerGroup.resolveAfter(
- Duration.fromSpec({
- seconds: 5,
- }),
- );
- if (wex.cancellationToken.isCancelled || pc.finished) {
- return;
- }
- wex.ws.notify({
- type: NotificationType.RequestProgressPhase,
- operation: pc.operation,
- progressToken: pc.progressToken,
- phase: "delayed",
- });
- await wex.ws.timerGroup.resolveAfter(
- Duration.fromSpec({
- seconds: 5,
- }),
- );
- if (wex.cancellationToken.isCancelled || pc.finished) {
- return;
- }
- wex.ws.notify({
- type: NotificationType.RequestProgressPhase,
- operation: pc.operation,
- progressToken: pc.progressToken,
- phase: "stalled",
- });
-}
-
-async function withMaybeProgressContext<T>(
- wex: WalletExecutionContext,
- op: string,
- tok: string | undefined,
- f: (pc?: ProgressContext) => Promise<T>,
-): Promise<T> {
- if (tok != null) {
- const key = `${op}:${encodeURIComponent(tok)}`;
- try {
- const pc: ProgressContext = {
- operation: op,
- progressToken: tok,
- finished: false,
- };
- if (wex.cts) {
- wex.ws.progressCancellationMap.set(key, wex.cts);
- } else {
- logger.warn(`no cancellation source for ${op}:${tok}`);
- }
- emitProgressPhaseNotifications(wex, pc).catch((e) => {
- logger.warn(safeStringifyException(e));
- });
- const res = await f(pc);
- pc.finished = true;
- return res;
- } finally {
- wex.ws.progressCancellationMap.delete(key);
- // Clean up progress token registration
- }
- } else {
- return await f(undefined);
- }
-}
-
async function handlePrepareWithdrawExchange(
wex: WalletExecutionContext,
req: PrepareWithdrawExchangeRequest,
@@ -985,6 +909,9 @@ async function handlePrepareWithdrawExchange(
wex,
"prepareWithdrawExchange",
req.progressToken,
+ async () => {
+ logger.warn("retryNow for prepareWithdrawExchange not implemented");
+ },
async (pc) => {
const exchange = await fetchFreshExchange(wex, exchangeBaseUrl, {
noBail: pc != null,
@@ -2238,20 +2165,6 @@ export async function handleGetPerformanceStats(
};
}
-export async function handleCancelRequest(
- wex: WalletExecutionContext,
- req: CancelProgressTokenRequest,
-): Promise<EmptyObject> {
- return {};
-}
-
-export async function handleRetryRequestNow(
- wex: WalletExecutionContext,
- req: CancelProgressTokenRequest,
-): Promise<EmptyObject> {
- return {};
-}
-
interface HandlerWithValidator<Tag extends WalletApiOperation> {
codec: Codec<WalletCoreRequestType<Tag>>;
handler: (
@@ -2263,11 +2176,11 @@ interface HandlerWithValidator<Tag extends WalletApiOperation> {
const handlers: { [T in WalletApiOperation]: HandlerWithValidator<T> } = {
[WalletApiOperation.CancelProgressToken]: {
codec: codecForCancelProgressToken(),
- handler: handleCancelRequest,
+ handler: handleCancelProgressToken,
},
[WalletApiOperation.RetryProgressTokenNow]: {
codec: codecForRetryProgressTokenNowRequest(),
- handler: handleRetryRequestNow,
+ handler: handleRetryProgressTokenNow,
},
[WalletApiOperation.TestingWaitExchangeReady]: {
codec: codecForTestingWaitExchangeReadyRequest(),
@@ -3465,7 +3378,7 @@ export class InternalWalletState {
clientCancellationMap: Map<string, CancellationToken.Source> = new Map();
- progressCancellationMap: Map<string, CancellationToken.Source> = new Map();
+ progressMap: Map<string, ProgressContext> = new Map();
longpollQueue = new LongpollQueue();
