From 96e8ba5c239c8bcb40cbe330ee599b132005e4de Mon Sep 17 00:00:00 2001
From: Florian Dold <florian@dold.me>
Date: Tue, 7 Nov 2023 11:52:08 +0100
Subject: wallet-core: throttle tasks

---
 .../taler-wallet-core/src/operations/pending.ts    |  6 ++++++
 packages/taler-wallet-core/src/pending-types.ts    |  5 +++++
 packages/taler-wallet-core/src/wallet.ts           | 25 +++++++++++++++++-----
 3 files changed, 31 insertions(+), 5 deletions(-)

(limited to 'packages/taler-wallet-core')

diff --git a/packages/taler-wallet-core/src/operations/pending.ts b/packages/taler-wallet-core/src/operations/pending.ts
index 7590280bc..e30958226 100644
--- a/packages/taler-wallet-core/src/operations/pending.ts
+++ b/packages/taler-wallet-core/src/operations/pending.ts
@@ -538,6 +538,7 @@ async function gatherPeerPullInitiationPending(
         givesLifeness: true,
         retryInfo: retryRecord?.retryInfo,
         pursePub: pi.pursePub,
+        internalOperationStatus: `0x${pi.status.toString(16)}`,
       });
     },
   );
@@ -579,12 +580,17 @@ async function gatherPeerPullDebitPending(
       const timestampDue =
         timestampOptionalAbsoluteFromDb(retryRecord?.retryInfo.nextRetry) ??
         AbsoluteTime.now();
+      // switch (pi.status) {
+      //   case PeerPullDebitRecordStatus.DialogProposed:
+      //     return;
+      // }
       resp.pendingOperations.push({
         type: PendingTaskType.PeerPullDebit,
         ...getPendingCommon(ws, opId, timestampDue),
         givesLifeness: true,
         retryInfo: retryRecord?.retryInfo,
         peerPullDebitId: pi.peerPullDebitId,
+        internalOperationStatus: `0x${pi.status.toString(16)}`,
       });
     },
   );
diff --git a/packages/taler-wallet-core/src/pending-types.ts b/packages/taler-wallet-core/src/pending-types.ts
index e7a40e81b..f8406033a 100644
--- a/packages/taler-wallet-core/src/pending-types.ts
+++ b/packages/taler-wallet-core/src/pending-types.ts
@@ -234,6 +234,11 @@ export interface PendingTaskInfoCommon {
    * exceeds a number of retries.
    */
   retryInfo?: DbRetryInfo;
+
+  /**
+   * Internal operation status for debugging.
+   */
+  internalOperationStatus?: string;
 }
 
 /**
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,
         });
       }
-- 
cgit v1.2.3