feat: restructure loadData and rebalanceRoot cache to disallow parall…

…elism (#1351)

* improve(dataworker): Warm BundleData loadData cache

I’ve noticed that the executor is very slow and seems to stall for a long time when evaluating L2 leaves. I believe the problem is that the executor tries to execute leaves for all chains in parallel. For example, the executor tries to execute leaves from the latest 2 root bundles for 7 chains in parallel. That means that  this [function](https://github.com/across-protocol/relayer-v2/blob/2a986a267c72af02390124ffee545840f14f7b0a/src/dataworker/Dataworker.ts#L1094) which calls `BundleDataClient.loadData`  is running 7 times in parallel. The `loadData` function is designed to cache the bundle data in-memory (not in Redis!) but we can’t take advantage of this if we call it many times in parallel.

Therefore, I propose warming this cache for each executor run, which ensures we call `loadData` only once per executor run.

To add confluence to this observation about the source of slowdown, the execution of the relayer refund roots is very fast compared to the slow roots, and they `loadData` over the exact same block ranges. In this case, the refund root execution logic runs AFTER the loadData cache has been warmed.

* feat: restructure cache to limit parallelism

Signed-off-by: Matt Rice <[email protected]>

* Revert "improve(dataworker): Warm BundleData loadData cache"

This reverts commit 27cea6b.

* WIP

Signed-off-by: Matt Rice <[email protected]>

* Update src/clients/BundleDataClient.ts

* WIP

Signed-off-by: Matt Rice <[email protected]>

* add same structure for root cache

Signed-off-by: Matt Rice <[email protected]>

---------

Signed-off-by: Matt Rice <[email protected]>
Co-authored-by: nicholaspai <[email protected]>
Co-authored-by: nicholaspai <[email protected]>

src/clients/BundleDataClient.ts

@@ -53,7 +53,7 @@ import {
   LoadDataReturnValue,
 } from "../interfaces/BundleData";
 
-type DataCache = Record<string, LoadDataReturnValue>;
+type DataCache = Record<string, Promise<LoadDataReturnValue>>;
 
 // V3 dictionary helper functions
 function updateExpiredDepositsV3(dict: ExpiredDepositsToRefundV3, deposit: V3DepositWithBlock): void {
@@ -158,10 +158,10 @@ export class BundleDataClient {
     this.loadDataCache = {};
   }
 
-  loadDataFromCache(key: string): LoadDataReturnValue {
+  async loadDataFromCache(key: string): Promise<LoadDataReturnValue> {
     // Always return a deep cloned copy of object stored in cache. Since JS passes by reference instead of value, we
     // want to minimize the risk that the programmer accidentally mutates data in the cache.
-    return _.cloneDeep(this.loadDataCache[key]);
+    return _.cloneDeep(await this.loadDataCache[key]);
   }
 
   getBundleTimestampsFromCache(key: string): undefined | { [chainId: number]: number[] } {
@@ -299,10 +299,20 @@ export class BundleDataClient {
   ): Promise<LoadDataReturnValue> {
     const key = JSON.stringify(blockRangesForChains);
 
-    if (this.loadDataCache[key]) {
-      return this.loadDataFromCache(key);
+    if (!this.loadDataCache[key]) {
+      this.loadDataCache[key] = this._loadData(blockRangesForChains, spokePoolClients, logData);
     }
 
+    return this.loadDataFromCache(key);
+  }
+
+  async _loadData(
+    blockRangesForChains: number[][],
+    spokePoolClients: SpokePoolClientsByChain,
+    logData = true
+  ): Promise<LoadDataReturnValue> {
+    const key = JSON.stringify(blockRangesForChains);
+
     if (!this.clients.configStoreClient.isUpdated) {
       throw new Error("ConfigStoreClient not updated");
     } else if (!this.clients.hubPoolClient.isUpdated) {
@@ -1061,7 +1071,7 @@ export class BundleDataClient {
       });
     }
 
-    this.loadDataCache[key] = {
+    return {
       fillsToRefund,
       deposits,
       unfilledDeposits,
@@ -1073,8 +1083,6 @@ export class BundleDataClient {
       unexecutableSlowFills,
       bundleSlowFillsV3,
     };
-
-    return this.loadDataFromCache(key);
   }
 
   async getBundleBlockTimestamps(

src/dataworker/Dataworker.ts

@@ -97,7 +97,7 @@ export type PoolRebalanceRoot = {
   tree: MerkleTree<PoolRebalanceLeaf>;
 };
 
-type PoolRebalanceRootCache = Record<string, PoolRebalanceRoot>;
+type PoolRebalanceRootCache = Record<string, Promise<PoolRebalanceRoot>>;
 
 // @notice Constructs roots to submit to HubPool on L1. Fetches all data synchronously from SpokePool/HubPool clients
 // so this class assumes that those upstream clients are already updated and have fetched on-chain data from RPC's.
@@ -2275,7 +2275,7 @@ export class Dataworker {
     //        executor running for tonight (2023-08-28) until we can fix the
     //        root cache rebalancing bug.
     if (!this.rootCache[key] || process.env.DATAWORKER_DISABLE_REBALANCE_ROOT_CACHE === "true") {
-      this.rootCache[key] = await _buildPoolRebalanceRoot(
+      this.rootCache[key] = _buildPoolRebalanceRoot(
         latestMainnetBlock,
         mainnetBundleEndBlock,
         fillsToRefund,
@@ -2296,7 +2296,7 @@ export class Dataworker {
       );
     }
 
-    return _.cloneDeep(this.rootCache[key]);
+    return _.cloneDeep(await this.rootCache[key]);
   }
 
   _getRequiredEthForArbitrumPoolRebalanceLeaf(leaf: PoolRebalanceLeaf): BigNumber {