feat: Dynamically adjust proposal block ranges

src/clients/BundleDataClient.ts

@@ -1,3 +1,4 @@
+import assert from "assert";
 import * as _ from "lodash";
 import {
   ProposedRootBundle,
@@ -16,7 +17,6 @@ import {
   getRefundInformationFromFill,
   queryHistoricalDepositForFill,
   assign,
-  assert,
   fixedPointAdjustment,
   isDefined,
   toBN,
@@ -45,6 +45,15 @@ import {
 import { BundleDataSS } from "../utils/SuperstructUtils";
 import { PoolRebalanceRoot } from "../dataworker/Dataworker";
 
+// type DataCacheValue = {
+//   unfilledDeposits: UnfilledDeposit[];
+//   fillsToRefund: FillsToRefund;
+//   allValidFills: FillWithBlock[];
+//   allInvalidFills: InvalidFill[];
+//   deposits: DepositWithBlock[];
+//   earlyDeposits: typechain.FundsDepositedEvent[];
+// };
+// type DataCache = Record<string, DataCacheValue>;
 type DataCache = Record<string, Promise<LoadDataReturnValue>>;
 
 // V3 dictionary helper functions

src/dataworker/Dataworker.ts

@@ -15,6 +15,8 @@ import {
   ZERO_ADDRESS,
 } from "../utils";
 import {
+  DepositWithBlock,
+  InvalidFill,
   ProposedRootBundle,
   RootBundleRelayWithBlock,
   SpokePoolClientsByChain,
@@ -338,7 +340,9 @@ export class Dataworker {
       return;
     }
 
-    return blockRangesForProposal;
+    // Narrow the block range depending on potential data incoherencies.
+    const safeBlockRanges = await this.narrowProposalBlockRanges(blockRangesForProposal, spokePoolClients);
+    return safeBlockRanges;
   }
 
   getNextHubChainBundleStartBlock(chainIdList = this.chainIdListForBundleEvaluationBlockNumbers): number {
@@ -467,6 +471,104 @@ export class Dataworker {
     return rootBundleData.dataToPersistToDALayer;
   }
 
+  async narrowProposalBlockRanges(
+    blockRanges: number[][],
+    spokePoolClients: SpokePoolClientsByChain
+  ): Promise<number[][]> {
+    const chainIds = this.chainIdListForBundleEvaluationBlockNumbers;
+    const updatedBlockRanges = Object.fromEntries(chainIds.map((chainId, idx) => [chainId, [...blockRanges[idx]]]));
+
+    const { bundleInvalidFillsV3: invalidFills } = await this.clients.bundleDataClient.loadData(
+      blockRanges,
+      spokePoolClients
+    );
+
+    // Helper to update a chain's end block correctly, accounting for soft-pausing if needed.
+    const updateEndBlock = (chainId: number, endBlock?: number): void => {
+      const [currentStartBlock, currentEndBlock] = updatedBlockRanges[chainId];
+      const previousEndBlock = this.clients.hubPoolClient.getLatestBundleEndBlockForChain(
+        chainIds,
+        this.clients.hubPoolClient.latestBlockSearched,
+        chainId
+      );
+
+      endBlock ??= previousEndBlock;
+      assert(
+        endBlock < currentEndBlock,
+        `Invalid block range update for chain ${chainId}: block ${endBlock} >= ${currentEndBlock}`
+      );
+
+      // If endBlock is equal to or before the currentEndBlock then the chain is "soft-paused" by setting the bundle
+      // start and end block equal to the previous end block. This tells the dataworker to not progress on that chain.
+      updatedBlockRanges[chainId] =
+        endBlock > currentStartBlock ? [currentStartBlock, endBlock] : [previousEndBlock, previousEndBlock];
+    };
+
+    // If invalid fills were detected and they appear to be due to gaps in FundsDeposited events:
+    // - Narrow the origin block range to exclude the missing deposit, AND
+    // - Narrow the destination block range to exclude the invalid fill.
+    invalidFills
+      .filter(({ code }) => code === InvalidFill.DepositIdNotFound)
+      .forEach(({ fill: { depositId, originChainId, destinationChainId, blockNumber } }) => {
+        const [originChain, destinationChain] = [getNetworkName(originChainId), getNetworkName(destinationChainId)];
+
+        // Exclude the missing deposit on the origin chain.
+        const originSpokePoolClient = spokePoolClients[originChainId];
+        let [startBlock, endBlock] = updatedBlockRanges[originChainId];
+
+        // Find the previous known deposit. This may resolve a deposit before the immediately preceding depositId.
+        const previousDeposit = originSpokePoolClient
+          .getDepositsForDestinationChain(destinationChainId)
+          .filter((deposit: DepositWithBlock) => deposit.blockNumber < blockNumber)
+          .at(-1);
+
+        updateEndBlock(originChainId, previousDeposit?.blockNumber);
+        this.logger.debug({
+          at: "Dataworker::narrowBlockRanges",
+          message: `Narrowed proposal block range on ${originChain} due to missing deposit.`,
+          depositId,
+          previousBlockRange: [startBlock, endBlock],
+          newBlockRange: updatedBlockRanges[originChainId],
+        });
+
+        // Update the endBlock on the destination chain to exclude the invalid fill.
+        const destSpokePoolClient = spokePoolClients[destinationChainId];
+        [startBlock, endBlock] = updatedBlockRanges[destinationChainId];
+
+        // The blockNumber is iteratively narrowed in this loop so this fill might already be excluded.
+        if (blockNumber <= endBlock) {
+          // Find the fill immediately preceding this invalid fill.
+          const previousFill = destSpokePoolClient
+            .getFillsWithBlockInRange(startBlock, Math.max(blockNumber - 1, startBlock))
+            .at(-1);
+
+          // Wind back to the bundle end block number to that of the previous fill.
+          updateEndBlock(destinationChainId, previousFill?.blockNumber);
+          this.logger.debug({
+            at: "Dataworker::narrowBlockRanges",
+            message: `Narrowed proposal block range on ${destinationChain} due to missing deposit on ${originChain}.`,
+            depositId,
+            previousBlockRange: [startBlock, endBlock],
+            newBlockRange: updatedBlockRanges[destinationChainId],
+          });
+        }
+      });
+
+    // Quick sanity check - make sure that the block ranges are coherent. A chain may be soft-paused if it has ongoing
+    // RPC issues (block ranges are frozen at the previous proposal endBlock), so ensure that this is also handled.
+    const finalBlockRanges = chainIds.map((chainId) => updatedBlockRanges[chainId]);
+    const coherentBlockRanges = finalBlockRanges.every(([startBlock, endBlock], idx) => {
+      const [originalStartBlock] = blockRanges[idx];
+      return (
+        (endBlock > startBlock && startBlock === originalStartBlock) ||
+        (startBlock === endBlock && startBlock === originalStartBlock - 1) // soft-pause
+      );
+    });
+    assert(coherentBlockRanges, "Updated proposal block ranges are incoherent");
+
+    return finalBlockRanges;
+  }
+
   async _proposeRootBundle(
     blockRangesForProposal: number[][],
     spokePoolClients: SpokePoolClientsByChain,
@@ -475,12 +577,12 @@ export class Dataworker {
     logData = false
   ): Promise<ProposeRootBundleReturnType> {
     const timerStart = Date.now();
+
     const { bundleDepositsV3, bundleFillsV3, bundleSlowFillsV3, unexecutableSlowFills, expiredDepositsToRefundV3 } =
       await this.clients.bundleDataClient.loadData(blockRangesForProposal, spokePoolClients, loadDataFromArweave);
-    // Prepare information about what we need to store to
-    // Arweave for the bundle. We will be doing this at a
-    // later point so that we can confirm that this data is
-    // worth storing.
+
+    // Prepare information about what we need to store to Arweave for the bundle.
+    // We will be doing this at a later point so that we can confirm that this data is worth storing.
     const dataToPersistToDALayer = {
       bundleBlockRanges: blockRangesForProposal,
       bundleDepositsV3,

src/interfaces/SpokePool.ts

@@ -1,5 +1,14 @@
-import { SpokePoolClient } from "../clients";
+import { FillWithBlock, SpokePoolClient } from "../clients";
+import * as utils from "../utils/SDKUtils";
 
 export interface SpokePoolClientsByChain {
   [chainId: number]: SpokePoolClient;
 }
+
+export const { InvalidFill } = utils;
+
+export type InvalidFill = {
+  fill: FillWithBlock;
+  code: utils.InvalidFillEnum;
+  reason: string;
+};

src/interfaces/index.ts

@@ -51,6 +51,7 @@ export type PendingRootBundle = interfaces.PendingRootBundle;
 
 // SpokePool interfaces
 export type FundsDepositedEvent = interfaces.FundsDepositedEvent;
+export type RelayData = interfaces.RelayData;
 export type Deposit = interfaces.Deposit;
 export type DepositWithBlock = interfaces.DepositWithBlock;
 export type Fill = interfaces.Fill;

src/utils/SDKUtils.ts

@@ -6,6 +6,8 @@ export type BlockFinderHints = sdk.utils.BlockFinderHints;
 export class PriceClient extends sdk.priceClient.PriceClient {}
 export const { acrossApi, coingecko, defiLlama } = sdk.priceClient.adapters;
 
+export type InvalidFillEnum = sdk.utils.InvalidFill;
+
 export const {
   bnZero,
   bnOne,
@@ -23,6 +25,7 @@ export const {
   toGWei,
   toBNWei,
   formatFeePct,
+  InvalidFill,
   shortenHexStrings,
   convertFromWei,
   max,

test/Dataworker.loadData.ts

@@ -90,6 +90,7 @@ describe("Dataworker: Load data used in all functions", async function () {
       bundleDepositsV3: {},
       expiredDepositsToRefundV3: {},
       bundleFillsV3: {},
+      bundleInvalidFillsV3: [],
       unexecutableSlowFills: {},
       bundleSlowFillsV3: {},
     });