feat: restructure loadData and rebalanceRoot cache to disallow parallelism

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 {