From 27cea6bd30db72f91b6fe720135c78517f7d8acd Mon Sep 17 00:00:00 2001
From: nicholaspai <npai.nyc@gmail.com>
Date: Fri, 22 Mar 2024 18:10:55 -0400
Subject: [PATCH] improve(dataworker): Warm BundleData loadData cache
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 src/dataworker/Dataworker.ts | 63 ++++++++++++++++++++++++++++++++++++
 src/dataworker/index.ts      |  2 ++
 2 files changed, 65 insertions(+)

diff --git a/src/dataworker/Dataworker.ts b/src/dataworker/Dataworker.ts
index 44b243078..6c3448243 100644
--- a/src/dataworker/Dataworker.ts
+++ b/src/dataworker/Dataworker.ts
@@ -989,6 +989,69 @@ export class Dataworker {
     };
   }
 
+  // Designed to be called before executing leaves in root bundle to ensure that the BundleDataClient.loadData
+  // output is cached. This is useful because `_proposeRootBundle` can be very slow if called in parallel
+  // for every spoke pool client. Instead, call it sequentially for the max number of bundles to inspect before
+  // trying to execute leaves in parallel.
+  async warmBundleDataCache(
+    spokePoolClients: { [chainId: number]: SpokePoolClient },
+    earliestBlocksInSpokePoolClients: { [chainId: number]: number } = {}
+  ): Promise<void> {
+    this.logger.debug({
+      at: "Dataworker#warmBundleDataCache",
+      message: `Warming bundle data for the latest ${this.spokeRootsLookbackCount} root bundles`,
+    });
+
+    const timerStart = Date.now();
+    let latestRootBundles = sortEventsDescending(this.clients.hubPoolClient.getValidatedRootBundles());
+    if (this.spokeRootsLookbackCount !== 0) {
+      latestRootBundles = latestRootBundles.slice(0, this.spokeRootsLookbackCount);
+    }
+
+    await Promise.all(
+      latestRootBundles.map(async (rootBundle) => {
+        const blockNumberRanges = getImpliedBundleBlockRanges(
+          this.clients.hubPoolClient,
+          this.clients.configStoreClient,
+          rootBundle
+        );
+        const mainnetBlockRange = blockNumberRanges[0];
+        const chainIds = this.clients.configStoreClient.getChainIdIndicesForBlock(mainnetBlockRange[0]);
+        if (
+          Object.keys(earliestBlocksInSpokePoolClients).length > 0 &&
+          (await blockRangesAreInvalidForSpokeClients(
+            spokePoolClients,
+            blockNumberRanges,
+            chainIds,
+            earliestBlocksInSpokePoolClients,
+            this.isV3(mainnetBlockRange[0])
+          ))
+        ) {
+          // Log this as a debug level and let the executeX function log it at the warn level.
+          this.logger.debug({
+            at: "Dataworker#warmBundleDataCache",
+            message: "Cannot validate bundle with insufficient event data. Set a larger DATAWORKER_FAST_LOOKBACK_COUNT",
+            rootBundleRanges: blockNumberRanges,
+            availableSpokePoolClients: Object.keys(spokePoolClients),
+            earliestBlocksInSpokePoolClients,
+            spokeClientsEventSearchConfigs: Object.fromEntries(
+              Object.entries(spokePoolClients).map(([chainId, client]) => [chainId, client.eventSearchConfig])
+            ),
+          });
+          return;
+        }
+        await this._proposeRootBundle(blockNumberRanges, spokePoolClients, rootBundle.blockNumber);
+      })
+    );
+    this.logger.debug({
+      at: "Dataworker#warmBundleDataCache",
+      message: `Warmed bundle data cache for the latest ${this.spokeRootsLookbackCount} root bundles in ${
+        Date.now() - timerStart
+      }ms`,
+      latestRootBundles,
+    });
+  }
+
   // TODO: this method and executeRelayerRefundLeaves have a lot of similarities, but they have some key differences
   // in both the events they search for and the comparisons they make. We should try to generalize this in the future,
   // but keeping them separate is probably the simplest for the initial implementation.
diff --git a/src/dataworker/index.ts b/src/dataworker/index.ts
index ae24e8d15..2e370f373 100644
--- a/src/dataworker/index.ts
+++ b/src/dataworker/index.ts
@@ -137,6 +137,8 @@ export async function runDataworker(_logger: winston.Logger, baseSigner: Signer)
           fromBlocks
         );
 
+        // Warm cache before executing slow and refund leaves.
+        await dataworker.warmBundleDataCache(spokePoolClients, fromBlocks);
         // Execute slow relays before relayer refunds to give them priority for any L2 funds.
         await dataworker.executeSlowRelayLeaves(
           spokePoolClients,