wallet: cancellation for deposit

packages/taler-wallet-core/src/db.ts

@@ -550,7 +550,7 @@ export interface ExchangeRecord {
   /**
    * Retry status for fetching updated information about the exchange.
    */
-  retryInfo: RetryInfo;
+  retryInfo?: RetryInfo;
 }
 
 /**

packages/taler-wallet-core/src/internal-wallet-state.ts

@@ -35,6 +35,7 @@ import {
   AmountJson,
   DenominationPubKey,
   TalerProtocolTimestamp,
+  CancellationToken,
 } from "@gnu-taler/taler-util";
 import { CryptoDispatcher } from "./crypto/workers/cryptoDispatcher.js";
 import { TalerCryptoInterface } from "./crypto/cryptoImplementation.js";
@@ -200,9 +201,14 @@ export interface InternalWalletState {
   memoGetBalance: AsyncOpMemoSingle<BalancesResponse>;
   memoProcessRefresh: AsyncOpMemoMap<void>;
   memoProcessRecoup: AsyncOpMemoMap<void>;
-  memoProcessDeposit: AsyncOpMemoMap<void>;
   cryptoApi: TalerCryptoInterface;
 
+  /**
+   * Cancellation token for the currently running
+   * deposit operation, if any.
+   */
+  taskCancellationSourceForDeposit?: CancellationToken.Source;
+
   timerGroup: TimerGroup;
   stopped: boolean;
 

packages/taler-wallet-core/src/operations/deposits.ts

@@ -21,6 +21,7 @@ import {
   AbsoluteTime,
   AmountJson,
   Amounts,
+  CancellationToken,
   canonicalJson,
   codecForDepositSuccess,
   ContractTerms,
@@ -125,23 +126,34 @@ async function reportDepositGroupError(
 export async function processDepositGroup(
   ws: InternalWalletState,
   depositGroupId: string,
-  forceNow = false,
+  options: {
+    forceNow?: boolean;
+    cancellationToken?: CancellationToken;
+  } = {},
 ): Promise<void> {
-  await ws.memoProcessDeposit.memo(depositGroupId, async () => {
+  if (ws.taskCancellationSourceForDeposit) {
-    const onOpErr = (err: TalerErrorDetail): Promise<void> =>
+    ws.taskCancellationSourceForDeposit.cancel();
-      reportDepositGroupError(ws, depositGroupId, err);
+  }
-    return await guardOperationException(
+  const onOpErr = (err: TalerErrorDetail): Promise<void> =>
-      async () => await processDepositGroupImpl(ws, depositGroupId, forceNow),
+    reportDepositGroupError(ws, depositGroupId, err);
-      onOpErr,
+  return await guardOperationException(
-    );
+    async () => await processDepositGroupImpl(ws, depositGroupId, options),
-  });
+    onOpErr,
+  );
 }
 
+/**
+ * @see {processDepositGroup}
+ */
 async function processDepositGroupImpl(
   ws: InternalWalletState,
   depositGroupId: string,
-  forceNow = false,
+  options: {
+    forceNow?: boolean;
+    cancellationToken?: CancellationToken;
+  } = {},
 ): Promise<void> {
+  const forceNow = options.forceNow ?? false;
   const depositGroup = await ws.db
     .mktx((x) => ({
       depositGroups: x.depositGroups,
@@ -170,6 +182,8 @@ async function processDepositGroupImpl(
     "",
   );
 
+  // Check for cancellation before expensive operations.
+  options.cancellationToken?.throwIfCancelled();
   const depositPermissions = await generateDepositPermissions(
     ws,
     depositGroup.payCoinSelection,
@@ -196,9 +210,13 @@ async function processDepositGroupImpl(
       denom_pub_hash: perm.h_denom,
       merchant_pub: depositGroup.merchantPub,
     };
+    // Check for cancellation before making network request.
+    options.cancellationToken?.throwIfCancelled();
     const url = new URL(`coins/${perm.coin_pub}/deposit`, perm.exchange_url);
     logger.info(`depositing to ${url}`);
-    const httpResp = await ws.http.postJson(url.href, requestBody);
+    const httpResp = await ws.http.postJson(url.href, requestBody, {
+      cancellationToken: options.cancellationToken,
+    });
     await readSuccessResponseJsonOrThrow(httpResp, codecForDepositSuccess());
     await ws.db
       .mktx((x) => ({ depositGroups: x.depositGroups }))

packages/taler-wallet-core/src/operations/exchanges.ts

@@ -61,7 +61,11 @@ import {
   readSuccessResponseTextOrThrow,
 } from "../util/http.js";
 import { DbAccess, GetReadOnlyAccess } from "../util/query.js";
-import { initRetryInfo, updateRetryInfoTimeout } from "../util/retries.js";
+import {
+  initRetryInfo,
+  RetryInfo,
+  updateRetryInfoTimeout,
+} from "../util/retries.js";
 import {
   WALLET_CACHE_BREAKER_CLIENT_VERSION,
   WALLET_EXCHANGE_PROTOCOL_VERSION,
@@ -102,7 +106,7 @@ function denominationRecordFromKeys(
   return d;
 }
 
-async function handleExchangeUpdateError(
+async function reportExchangeUpdateError(
   ws: InternalWalletState,
   baseUrl: string,
   err: TalerErrorDetail,
@@ -114,14 +118,44 @@ async function handleExchangeUpdateError(
       if (!exchange) {
         return;
       }
-      exchange.retryInfo.retryCounter++;
-      updateRetryInfoTimeout(exchange.retryInfo);
       exchange.lastError = err;
       await tx.exchanges.put(exchange);
     });
-  if (err) {
+  ws.notify({ type: NotificationType.ExchangeOperationError, error: err });
-    ws.notify({ type: NotificationType.ExchangeOperationError, error: err });
+}
-  }
+
+async function resetExchangeUpdateRetry(
+  ws: InternalWalletState,
+  baseUrl: string,
+): Promise<void> {
+  await ws.db
+    .mktx((x) => ({ exchanges: x.exchanges }))
+    .runReadWrite(async (tx) => {
+      const exchange = await tx.exchanges.get(baseUrl);
+      if (!exchange) {
+        return;
+      }
+      delete exchange.lastError;
+      exchange.retryInfo = initRetryInfo();
+      await tx.exchanges.put(exchange);
+    });
+}
+
+async function incrementExchangeUpdateRetry(
+  ws: InternalWalletState,
+  baseUrl: string,
+): Promise<void> {
+  await ws.db
+    .mktx((x) => ({ exchanges: x.exchanges }))
+    .runReadWrite(async (tx) => {
+      const exchange = await tx.exchanges.get(baseUrl);
+      if (!exchange) {
+        return;
+      }
+      delete exchange.lastError;
+      exchange.retryInfo = RetryInfo.increment(exchange.retryInfo);
+      await tx.exchanges.put(exchange);
+    });
 }
 
 export function getExchangeRequestTimeout(): Duration {
@@ -349,7 +383,7 @@ export async function updateExchangeFromUrl(
   exchangeDetails: ExchangeDetailsRecord;
 }> {
   const onOpErr = (e: TalerErrorDetail): Promise<void> =>
-    handleExchangeUpdateError(ws, baseUrl, e);
+    reportExchangeUpdateError(ws, baseUrl, e);
   return await guardOperationException(
     () => updateExchangeFromUrlImpl(ws, baseUrl, acceptedFormat, forceNow),
     onOpErr,
@@ -543,6 +577,12 @@ async function updateExchangeFromUrlImpl(
     return { exchange, exchangeDetails };
   }
 
+  if (forceNow) {
+    await resetExchangeUpdateRetry(ws, baseUrl);
+  } else {
+    await incrementExchangeUpdateRetry(ws, baseUrl);
+  }
+
   logger.info("updating exchange /keys info");
 
   const timeout = getExchangeRequestTimeout();
@@ -624,8 +664,8 @@ async function updateExchangeFromUrlImpl(
         termsOfServiceAcceptedTimestamp: TalerProtocolTimestamp.now(),
       };
       // FIXME: only update if pointer got updated
-      r.lastError = undefined;
+      delete r.lastError;
-      r.retryInfo = initRetryInfo();
+      delete r.retryInfo;
       r.lastUpdate = TalerProtocolTimestamp.now();
       r.nextUpdate = keysInfo.expiry;
       // New denominations might be available.

packages/taler-wallet-core/src/operations/transactions.ts

@@ -444,7 +444,9 @@ export async function retryTransaction(
   switch (type) {
     case TransactionType.Deposit:
       const depositGroupId = rest[0];
-      processDepositGroup(ws, depositGroupId, true);
+      processDepositGroup(ws, depositGroupId, {
+        forceNow: true,
+      });
       break;
     case TransactionType.Withdrawal:
       const withdrawalGroupId = rest[0];

packages/taler-wallet-core/src/wallet.ts

@@ -78,6 +78,7 @@ import {
   URL,
   WalletNotification,
   Duration,
+  CancellationToken,
 } from "@gnu-taler/taler-util";
 import { timeStamp } from "console";
 import {
@@ -271,9 +272,19 @@ async function processOnePendingOperation(
     case PendingTaskType.ExchangeCheckRefresh:
       await autoRefresh(ws, pending.exchangeBaseUrl);
       break;
-    case PendingTaskType.Deposit:
+    case PendingTaskType.Deposit: {
-      await processDepositGroup(ws, pending.depositGroupId);
+      const cts = CancellationToken.create();
+      ws.taskCancellationSourceForDeposit = cts;
+      try {
+        await processDepositGroup(ws, pending.depositGroupId, {
+          cancellationToken: cts.token,
+        });
+      } finally {
+        cts.dispose();
+        delete ws.taskCancellationSourceForDeposit;
+      }
       break;
+    }
     case PendingTaskType.Backup:
       await processBackupForProvider(ws, pending.backupProviderBaseUrl);
       break;