wait for refreshes to finish before exiting integration test

3 files changed, 98 insertions, 43 deletions

diff --git a/src/headless/integrationtest.ts b/src/headless/integrationtest.ts
index 3b0069c21..3e60d418a 100644
--- a/src/headless/integrationtest.ts
+++ b/src/headless/integrationtest.ts
@@ -83,5 +83,8 @@ export async function runIntegrationTest(args: {
     throw Error("payment did not succeed");
   }
 
+  const refreshRes = await myWallet.refreshDirtyCoins();
+  console.log(`waited to refresh ${refreshRes.numRefreshed} coins`);
+
   myWallet.stop();
 }
diff --git a/src/headless/taler-wallet-cli.ts b/src/headless/taler-wallet-cli.ts
index f10346369..ef9030fe5 100644
--- a/src/headless/taler-wallet-cli.ts
+++ b/src/headless/taler-wallet-cli.ts
@@ -301,25 +301,31 @@ program
     "amount to withdraw",
     "TESTKUDOS:10",
   )
-  .option("-s, --spend-amount <spend-amt>", "amount to spend", "TESTKUDOS:5")
+  .option("-s, --spend-amount <spend-amt>", "amount to spend", "TESTKUDOS:4")
   .description("Run integration test with bank, exchange and merchant.")
   .action(async cmdObj => {
     applyVerbose(program.verbose);
 
-    await runIntegrationTest({
-      amountToSpend: cmdObj.spendAmount,
-      amountToWithdraw: cmdObj.withdrawAmount,
-      bankBaseUrl: cmdObj.bank,
-      exchangeBaseUrl: cmdObj.exchange,
-      merchantApiKey: cmdObj.merchantApiKey,
-      merchantBaseUrl: cmdObj.merchant,
-      merchantInstance: cmdObj.merchantInstance,
-    }).catch(err => {
-      console.error("Failed with exception:");
-      console.error(err);
-    });
+    try {
+      await runIntegrationTest({
+        amountToSpend: cmdObj.spendAmount,
+        amountToWithdraw: cmdObj.withdrawAmount,
+        bankBaseUrl: cmdObj.bank,
+        exchangeBaseUrl: cmdObj.exchange,
+        merchantApiKey: cmdObj.merchantApiKey,
+        merchantBaseUrl: cmdObj.merchant,
+        merchantInstance: cmdObj.merchantInstance,
+      }).catch(err => {
+        console.error("Failed with exception:");
+        console.error(err);
+      });
+
+      process.exit(0);
+    } catch (e) {
+      console.error(e);
+      process.exit(1);
+    }
 
-    process.exit(0);
   });
 
 // error on unknown commands
diff --git a/src/wallet.ts b/src/wallet.ts
index 14ab45444..45bab48f8 100644
--- a/src/wallet.ts
+++ b/src/wallet.ts
@@ -53,7 +53,6 @@ import {
   DenominationRecord,
   DenominationStatus,
   ExchangeRecord,
-  ExchangeWireFeesRecord,
   PreCoinRecord,
   ProposalDownloadRecord,
   PurchaseRecord,
@@ -132,7 +131,7 @@ interface SpeculativePayData {
  */
 export const WALLET_PROTOCOL_VERSION = "3:0:0";
 
-const WALLET_CACHE_BREAKER="01";
+const WALLET_CACHE_BREAKER = "01";
 
 const builtinCurrencies: CurrencyRecord[] = [
   {
@@ -360,6 +359,9 @@ export class Wallet {
   private activeProcessPreCoinOperations: {
     [preCoinPub: string]: Promise<void>;
   } = {};
+  private activeRefreshOperations: {
+    [coinPub: string]: Promise<void>;
+  } = {};
 
   /**
    * Set of identifiers for running operations.
@@ -943,10 +945,35 @@ export class Wallet {
       nextUrl,
       lastSessionId: sessionId,
     };
+
     return { nextUrl };
   }
 
   /**
+   * Refresh all dirty coins.
+   * The returned promise resolves only after all refresh
+   * operations have completed.
+   */
+  async refreshDirtyCoins(): Promise<{ numRefreshed: number }> {
+    let n = 0;
+    const coins = await this.q()
+      .iter(Stores.coins)
+      .toArray();
+    for (let coin of coins) {
+      if (coin.status == CoinStatus.Dirty) {
+        try {
+          await this.refresh(coin.coinPub);
+        } catch (e) {
+          console.log("error during refresh");
+        }
+
+        n += 1;
+      }
+    }
+    return { numRefreshed: n };
+  }
+
+  /**
    * Add a contract to the wallet and sign coins, and send them.
    */
   async confirmPay(
@@ -1955,7 +1982,9 @@ export class Wallet {
    */
   async updateExchangeFromUrl(baseUrl: string): Promise<ExchangeRecord> {
     baseUrl = canonicalizeBaseUrl(baseUrl);
-    const keysUrl = new URI("keys").absoluteTo(baseUrl).addQuery("cacheBreaker", WALLET_CACHE_BREAKER);
+    const keysUrl = new URI("keys")
+      .absoluteTo(baseUrl)
+      .addQuery("cacheBreaker", WALLET_CACHE_BREAKER);
     const keysResp = await this.http.get(keysUrl.href());
     if (keysResp.status !== 200) {
       throw Error("/keys request failed");
@@ -2419,32 +2448,49 @@ export class Wallet {
   }
 
   async refresh(oldCoinPub: string): Promise<void> {
-    const oldRefreshSessions = await this.q()
-      .iter(Stores.refresh)
-      .toArray();
-    for (const session of oldRefreshSessions) {
-      Wallet.enableTracing &&
-        console.log("got old refresh session for", oldCoinPub, session);
-      this.continueRefreshSession(session);
-    }
-    const coin = await this.q().get(Stores.coins, oldCoinPub);
-    if (!coin) {
-      console.warn("can't refresh, coin not in database");
-      return;
-    }
-    if (
-      coin.status === CoinStatus.Useless ||
-      coin.status === CoinStatus.Fresh
-    ) {
-      return;
+    const refreshImpl = async () => {
+      const oldRefreshSessions = await this.q()
+        .iter(Stores.refresh)
+        .toArray();
+      for (const session of oldRefreshSessions) {
+        Wallet.enableTracing &&
+          console.log("got old refresh session for", oldCoinPub, session);
+        return this.continueRefreshSession(session);
+      }
+      const coin = await this.q().get(Stores.coins, oldCoinPub);
+      if (!coin) {
+        console.warn("can't refresh, coin not in database");
+        return;
+      }
+      if (
+        coin.status === CoinStatus.Useless ||
+        coin.status === CoinStatus.Fresh
+      ) {
+        return;
+      }
+      const refreshSession = await this.createRefreshSession(oldCoinPub);
+      if (!refreshSession) {
+        // refreshing not necessary
+        Wallet.enableTracing && console.log("not refreshing", oldCoinPub);
+        return;
+      }
+      return this.continueRefreshSession(refreshSession);
+    };
+
+    const activeRefreshOp = this.activeRefreshOperations[oldCoinPub];
+
+    if (activeRefreshOp) {
+      return activeRefreshOp;
     }
-    const refreshSession = await this.createRefreshSession(oldCoinPub);
-    if (!refreshSession) {
-      // refreshing not necessary
-      Wallet.enableTracing && console.log("not refreshing", oldCoinPub);
-      return;
+
+    try {
+      const newOp = refreshImpl();
+      this.activeRefreshOperations[oldCoinPub] = newOp;
+      const res = await newOp;
+      return res;
+    } finally {
+      delete this.activeRefreshOperations[oldCoinPub];
     }
-    this.continueRefreshSession(refreshSession);
   }
 
   async continueRefreshSession(refreshSession: RefreshSessionRecord) {
@@ -3617,8 +3663,8 @@ export class Wallet {
     const refundsDoneFees = Object.values(purchase.refundsDone).map(x =>
       Amounts.parseOrThrow(x.refund_amount),
     );
-    const refundsPendingFees = Object.values(purchase.refundsPending).map(
-      x => Amounts.parseOrThrow(x.refund_amount),
+    const refundsPendingFees = Object.values(purchase.refundsPending).map(x =>
+      Amounts.parseOrThrow(x.refund_amount),
     );
     const totalRefundFees = Amounts.sum([
       ...refundsDoneFees,