wallet-core: throttle tasks

1 files changed, 20 insertions, 5 deletions

diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
index e917e8059..978ce4c39 100644
--- a/packages/taler-wallet-core/src/wallet.ts
+++ b/packages/taler-wallet-core/src/wallet.ts
@@ -133,6 +133,8 @@ import {
   ListExchangesForScopedCurrencyRequest,
   ExchangesShortListResponse,
   AmountString,
+  RequestThrottler,
+  TaskThrottler,
 } from "@gnu-taler/taler-util";
 import type { HttpRequestLibrary } from "@gnu-taler/taler-util/http";
 import { readSuccessResponseJsonOrThrow } from "@gnu-taler/taler-util/http";
@@ -424,6 +426,7 @@ async function runTaskLoop(
       "task loop already running, nesting the wallet-core task loop is deprecated and should be avoided",
     );
   }
+  const throttler = new TaskThrottler();
   ws.isTaskLoopRunning = true;
   let retriesExceeded = false;
   for (let iteration = 0; !ws.stopped; iteration++) {
@@ -431,6 +434,7 @@ async function runTaskLoop(
     logger.trace(`pending operations: ${j2s(pending)}`);
     let numGivingLiveness = 0;
     let numDue = 0;
+    let numThrottled = 0;
     let minDue: AbsoluteTime = AbsoluteTime.never();
 
     for (const p of pending.pendingOperations) {
@@ -449,12 +453,23 @@ async function runTaskLoop(
       if (!p.isDue) {
         continue;
       }
-      minDue = AbsoluteTime.min(minDue, p.timestampDue);
       numDue++;
+
+      const isThrottled = throttler.applyThrottle(p.id);
+
+      if (isThrottled) {
+        logger.warn(
+          `task ${p.id} throttled, this is very likely a bug in wallet-core, please report`,
+        );
+        numDue--;
+        numThrottled++;
+      } else {
+        minDue = AbsoluteTime.min(minDue, p.timestampDue);
+      }
     }
 
     logger.trace(
-      `running task loop, iter=${iteration}, #tasks=${pending.pendingOperations.length} #lifeness=${numGivingLiveness}, #due=${numDue}`,
+      `running task loop, iter=${iteration}, #tasks=${pending.pendingOperations.length} #lifeness=${numGivingLiveness}, #due=${numDue} #trottled=${numThrottled}`,
     );
 
     if (opts.stopWhenDone && numGivingLiveness === 0 && iteration !== 0) {
@@ -932,9 +947,9 @@ async function dumpCoins(ws: InternalWalletState): Promise<CoinDumpJson> {
           ageCommitmentProof: c.ageCommitmentProof,
           spend_allocation: c.spendAllocation
             ? {
-              amount: c.spendAllocation.amount,
-              id: c.spendAllocation.id,
-            }
+                amount: c.spendAllocation.amount,
+                id: c.spendAllocation.id,
+              }
             : undefined,
         });
       }