simplify task loop, test coin suspension

7 files changed, 115 insertions, 102 deletions

diff --git a/packages/taler-wallet-android/src/index.ts b/packages/taler-wallet-android/src/index.ts
index 94774bcf4..7f2d44c5d 100644
--- a/packages/taler-wallet-android/src/index.ts
+++ b/packages/taler-wallet-android/src/index.ts
@@ -32,7 +32,6 @@ import {
   Headers,
   WALLET_EXCHANGE_PROTOCOL_VERSION,
   WALLET_MERCHANT_PROTOCOL_VERSION,
-  runRetryLoop,
   Wallet,
 } from "@gnu-taler/taler-wallet-core";
 
diff --git a/packages/taler-wallet-cli/src/index.ts b/packages/taler-wallet-cli/src/index.ts
index d4e5bbe46..2fac85a7e 100644
--- a/packages/taler-wallet-cli/src/index.ts
+++ b/packages/taler-wallet-cli/src/index.ts
@@ -33,7 +33,6 @@ import {
   codecForList,
   codecForString,
   Logger,
-  WithdrawalType,
 } from "@gnu-taler/taler-util";
 import {
   NodeHttpLib,
@@ -45,10 +44,6 @@ import {
   NodeThreadCryptoWorkerFactory,
   CryptoApi,
   walletCoreDebugFlags,
-  handleCoreApiRequest,
-  runPending,
-  runUntilDone,
-  getClientFromWalletState,
   WalletApiOperation,
   WalletCoreApiClient,
   Wallet,
@@ -314,8 +309,9 @@ walletCli
   .maybeOption("maxRetries", ["--max-retries"], clk.INT)
   .action(async (args) => {
     await withWallet(args, async (wallet) => {
-      await wallet.ws.runUntilDone({
+      await wallet.ws.runTaskLoop({
         maxRetries: args.finishPendingOpt.maxRetries,
+        stopWhenDone: true,
       });
       wallet.ws.stop();
     });
diff --git a/packages/taler-wallet-cli/src/integrationtests/test-wallettesting.ts b/packages/taler-wallet-cli/src/integrationtests/test-wallettesting.ts
index 5fb017edd..2499e65a0 100644
--- a/packages/taler-wallet-cli/src/integrationtests/test-wallettesting.ts
+++ b/packages/taler-wallet-cli/src/integrationtests/test-wallettesting.ts
@@ -22,8 +22,9 @@
 /**
  * Imports.
  */
+import { Amounts } from "@gnu-taler/taler-util";
 import { WalletApiOperation } from "@gnu-taler/taler-wallet-core";
-import { CoinConfig, defaultCoinConfig } from "./denomStructures";
+import { CoinConfig, defaultCoinConfig } from "./denomStructures.js";
 import {
   BankService,
   ExchangeService,
@@ -31,8 +32,8 @@ import {
   MerchantService,
   setupDb,
   WalletCli,
-} from "./harness";
-import { SimpleTestEnvironment } from "./helpers";
+} from "./harness.js";
+import { SimpleTestEnvironment } from "./helpers.js";
 
 const merchantAuthToken = "secret-token:sandbox";
 
@@ -162,6 +163,63 @@ export async function runWallettestingTest(t: GlobalTestState) {
 
   t.assertDeepEqual(txTypes, ["withdrawal", "payment"]);
 
+  wallet.deleteDatabase();
+
+  await wallet.client.call(WalletApiOperation.WithdrawTestBalance, {
+    amount: "TESTKUDOS:10",
+    bankBaseUrl: bank.baseUrl,
+    exchangeBaseUrl: exchange.baseUrl,
+  });
+
+  await wallet.runUntilDone();
+
+  const coinDump = await wallet.client.call(WalletApiOperation.DumpCoins, {});
+
+  console.log("coin dump:", JSON.stringify(coinDump, undefined, 2));
+
+  let susp: string | undefined;
+  {
+    for (const c of coinDump.coins) {
+      if (0 === Amounts.cmp(c.remaining_value, "TESTKUDOS:8")) {
+        susp = c.coin_pub;
+      }
+    }
+  }
+
+  t.assertTrue(susp !== undefined);
+
+  console.log("suspending coin");
+
+  await wallet.client.call(WalletApiOperation.SetCoinSuspended, {
+    coinPub: susp,
+    suspended: true,
+  });
+
+  // This should fail, as we've suspended a coin that we need
+  // to pay.
+  await t.assertThrowsAsync(async () => {
+    await wallet.client.call(WalletApiOperation.TestPay, {
+      amount: "TESTKUDOS:5",
+      merchantAuthToken: merchantAuthToken,
+      merchantBaseUrl: merchant.makeInstanceBaseUrl(),
+      summary: "foo",
+    });
+  });
+
+  console.log("unsuspending coin");
+
+  await wallet.client.call(WalletApiOperation.SetCoinSuspended, {
+    coinPub: susp,
+    suspended: false,
+  });
+
+  await wallet.client.call(WalletApiOperation.TestPay, {
+    amount: "TESTKUDOS:5",
+    merchantAuthToken: merchantAuthToken,
+    merchantBaseUrl: merchant.makeInstanceBaseUrl(),
+    summary: "foo",
+  });
+
   await t.shutdown();
 }
 
diff --git a/packages/taler-wallet-core/src/common.ts b/packages/taler-wallet-core/src/common.ts
index 128138eb2..b0b975e7b 100644
--- a/packages/taler-wallet-core/src/common.ts
+++ b/packages/taler-wallet-core/src/common.ts
@@ -116,9 +116,15 @@ export interface InternalWalletState {
   cryptoApi: CryptoApi;
 
   timerGroup: TimerGroup;
-  latch: AsyncCondition;
   stopped: boolean;
-  memoRunRetryLoop: AsyncOpMemoSingle<void>;
+
+  /**
+   * Asynchronous condition to interrupt the sleep of the
+   * retry loop.
+   *
+   * Used to allow processing of new work faster.
+   */
+  latch: AsyncCondition;
 
   listeners: NotificationListener[];
 
diff --git a/packages/taler-wallet-core/src/operations/pay.ts b/packages/taler-wallet-core/src/operations/pay.ts
index 464c3136f..e662ee72f 100644
--- a/packages/taler-wallet-core/src/operations/pay.ts
+++ b/packages/taler-wallet-core/src/operations/pay.ts
@@ -205,10 +205,7 @@ export async function getEffectiveDepositAmount(
   return Amounts.sub(Amounts.sum(amt).amount, Amounts.sum(fees).amount).amount;
 }
 
-export function isSpendableCoin(
-  coin: CoinRecord,
-  denom: DenominationRecord,
-): boolean {
+function isSpendableCoin(coin: CoinRecord, denom: DenominationRecord): boolean {
   if (coin.suspended) {
     return false;
   }
@@ -721,7 +718,9 @@ async function processDownloadProposalImpl(
   );
 
   if (!isWellFormed) {
-    logger.trace(`malformed contract terms: ${j2s(proposalResp.contract_terms)}`);
+    logger.trace(
+      `malformed contract terms: ${j2s(proposalResp.contract_terms)}`,
+    );
     const err = makeErrorDetails(
       TalerErrorCode.WALLET_CONTRACT_TERMS_MALFORMED,
       "validation for well-formedness failed",
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
index 6a7ee9de1..de0675cd6 100644
--- a/packages/taler-wallet-core/src/wallet.ts
+++ b/packages/taler-wallet-core/src/wallet.ts
@@ -276,81 +276,31 @@ export async function runPending(
   }
 }
 
-/**
- * Run the wallet until there are no more pending operations that give
- * liveness left.  The wallet will be in a stopped state when this function
- * returns without resolving to an exception.
- */
-export async function runUntilDone(
-  ws: InternalWalletState,
-  req: {
-    maxRetries?: number;
-  } = {},
-): Promise<void> {
-  let done = false;
-  const p = new Promise<void>((resolve, reject) => {
-    // Monitor for conditions that means we're done or we
-    // should quit with an error (due to exceeded retries).
-    ws.addNotificationListener((n) => {
-      if (done) {
-        return;
-      }
-      if (
-        n.type === NotificationType.WaitingForRetry &&
-        n.numGivingLiveness == 0
-      ) {
-        done = true;
-        logger.trace("no liveness-giving operations left");
-        resolve();
-      }
-      const maxRetries = req.maxRetries;
-      if (!maxRetries) {
-        return;
-      }
-      getPendingOperations(ws)
-        .then((pending) => {
-          for (const p of pending.pendingOperations) {
-            if (p.retryInfo && p.retryInfo.retryCounter > maxRetries) {
-              console.warn(
-                `stopping, as ${maxRetries} retries are exceeded in an operation of type ${p.type}`,
-              );
-              ws.stop();
-              done = true;
-              resolve();
-            }
-          }
-        })
-        .catch((e) => {
-          logger.error(e);
-          reject(e);
-        });
-    });
-    // Run this asynchronously
-    runRetryLoop(ws).catch((e) => {
-      logger.error("exception in wallet retry loop");
-      reject(e);
-    });
-  });
-  await p;
+export interface RetryLoopOpts {
+  /**
+   * Stop when the number of retries is exceeded for any pending
+   * operation.
+   */
+  maxRetries?: number;
+
+  /**
+   * Stop the retry loop when all lifeness-giving pending operations
+   * are done.
+   *
+   * Defaults to false.
+   */
+  stopWhenDone?: boolean;
 }
 
 /**
- * Process pending operations and wait for scheduled operations in
- * a loop until the wallet is stopped explicitly.
+ * Main retry loop of the wallet.
+ *
+ * Looks up pending operations from the wallet, runs them, repeat.
  */
-export async function runRetryLoop(ws: InternalWalletState): Promise<void> {
-  // Make sure we only run one main loop at a time.
-  return ws.memoRunRetryLoop.memo(async () => {
-    try {
-      await runRetryLoopImpl(ws);
-    } catch (e) {
-      console.error("error during retry loop execution", e);
-      throw e;
-    }
-  });
-}
-
-async function runRetryLoopImpl(ws: InternalWalletState): Promise<void> {
+async function runTaskLoop(
+  ws: InternalWalletState,
+  opts: RetryLoopOpts = {},
+): Promise<void> {
   for (let iteration = 0; !ws.stopped; iteration++) {
     const pending = await getPendingOperations(ws);
     logger.trace(`pending operations: ${j2s(pending)}`);
@@ -365,7 +315,22 @@ async function runRetryLoopImpl(ws: InternalWalletState): Promise<void> {
       if (p.givesLifeness) {
         numGivingLiveness++;
       }
+
+      const maxRetries = opts.maxRetries;
+
+      if (maxRetries && p.retryInfo && p.retryInfo.retryCounter > maxRetries) {
+        logger.warn(
+          `stopping, as ${maxRetries} retries are exceeded in an operation of type ${p.type}`,
+        );
+        return;
+      }
     }
+
+    if (opts.stopWhenDone && numGivingLiveness === 0) {
+      logger.warn(`stopping, as no pending operations have lifeness`);
+      return;
+    }
+
     // Make sure that we run tasks that don't give lifeness at least
     // one time.
     if (iteration !== 0 && numDue === 0) {
@@ -993,19 +958,15 @@ export class Wallet {
   }
 
   runRetryLoop(): Promise<void> {
-    return runRetryLoop(this.ws);
+    return runTaskLoop(this.ws);
   }
 
   runPending(forceNow: boolean = false) {
     return runPending(this.ws, forceNow);
   }
 
-  runUntilDone(
-    req: {
-      maxRetries?: number;
-    } = {},
-  ) {
-    return runUntilDone(this.ws, req);
+  runTaskLoop(opts: RetryLoopOpts) {
+    return runTaskLoop(this.ws, opts);
   }
 
   handleCoreApiRequest(
@@ -1035,7 +996,6 @@ class InternalWalletStateImpl implements InternalWalletState {
   timerGroup: TimerGroup = new TimerGroup();
   latch = new AsyncCondition();
   stopped = false;
-  memoRunRetryLoop = new AsyncOpMemoSingle<void>();
 
   listeners: NotificationListener[] = [];
 
@@ -1102,7 +1062,7 @@ class InternalWalletStateImpl implements InternalWalletState {
       maxRetries?: number;
     } = {},
   ): Promise<void> {
-    runUntilDone(this, req);
+    await runTaskLoop(this, { ...req, stopWhenDone: true });
   }
 
   /**
diff --git a/packages/taler-wallet-webextension/src/wxBackend.ts b/packages/taler-wallet-webextension/src/wxBackend.ts
index c8b2dbd78..633f88836 100644
--- a/packages/taler-wallet-webextension/src/wxBackend.ts
+++ b/packages/taler-wallet-webextension/src/wxBackend.ts
@@ -33,12 +33,7 @@ import {
   deleteTalerDatabase,
   DbAccess,
   WalletStoresV1,
-  handleCoreApiRequest,
-  runRetryLoop,
-  handleNotifyReserve,
-  InternalWalletState,
   Wallet,
-  WalletApiOperation,
 } from "@gnu-taler/taler-wallet-core";
 import {
   classifyTalerUri,