feat: Add helper methods for usage in Dataworker

src/clients/SpokePoolClient.ts

@@ -43,7 +43,9 @@ import {
   SlowFillRequestWithBlock,
   SpeedUp,
   TokensBridged,
+  V2Deposit,
   V2DepositWithBlock,
+  V2Fill,
   V2FillWithBlock,
   V2SpeedUp,
   V3DepositWithBlock,
@@ -380,6 +382,17 @@ export class SpokePoolClient extends BaseAbstractClient {
     return this.slowFillRequests[hash];
   }
 
+  /**
+   * Retrieves a list of slow fill requests for deposits from a specific origin chain ID.
+   * @param originChainId The origin chain ID.
+   * @returns A list of slow fill requests.
+   */
+  public getSlowFillRequestsForOriginChain(originChainId: number): SlowFillRequestWithBlock[] {
+    return Object.values(this.slowFillRequests).filter(
+      (e: SlowFillRequestWithBlock) => e.originChainId === originChainId
+    );
+  }
+
   /**
    * Find a corresponding deposit for a given fill.
    * @param fill The fill to find a corresponding deposit for.
@@ -526,7 +539,11 @@ export class SpokePoolClient extends BaseAbstractClient {
    * @param toBlock The block number to search up to.
    * @returns A list of fills that match the given fill and deposit.
    */
-  public async queryHistoricalMatchingFills(fill: Fill, deposit: Deposit, toBlock: number): Promise<FillWithBlock[]> {
+  public async queryHistoricalMatchingFills(
+    fill: V2Fill,
+    deposit: V2Deposit,
+    toBlock: number
+  ): Promise<V2FillWithBlock[]> {
     const searchConfig = {
       fromBlock: this.deploymentBlock,
       toBlock,
@@ -543,7 +560,10 @@ export class SpokePoolClient extends BaseAbstractClient {
    * @param searchConfig The search configuration.
    * @returns A Promise that resolves to a list of fills that match the given fill.
    */
-  public async queryFillsInBlockRange(matchingFill: Fill, searchConfig: EventSearchConfig): Promise<FillWithBlock[]> {
+  public async queryFillsInBlockRange(
+    matchingFill: V2Fill,
+    searchConfig: EventSearchConfig
+  ): Promise<V2FillWithBlock[]> {
     // Filtering on the fill's depositor address, the only indexed deposit field in the FilledRelay event,
     // should speed up this search a bit.
     // TODO: Once depositId is indexed in FilledRelay event, filter on that as well.
@@ -568,7 +588,7 @@ export class SpokePoolClient extends BaseAbstractClient {
       ),
       searchConfig
     );
-    const fills = query.map((event) => spreadEventWithBlockNumber(event) as FillWithBlock);
+    const fills = query.map((event) => spreadEventWithBlockNumber(event) as V2FillWithBlock);
     return sortEventsAscending(fills.filter((_fill) => filledSameDeposit(_fill, matchingFill)));
   }
 
@@ -769,6 +789,9 @@ export class SpokePoolClient extends BaseAbstractClient {
         deposit.realizedLpFeePct = realizedLpFeePct;
         deposit.quoteBlockNumber = quoteBlockNumber;
 
+        if (this.depositHashes[this.getDepositHash(deposit)] !== undefined) {
+          continue;
+        }
         assign(this.depositHashes, [this.getDepositHash(deposit)], deposit);
 
         if (deposit.depositId < this.earliestDepositIdQueried) {
@@ -815,6 +838,9 @@ export class SpokePoolClient extends BaseAbstractClient {
           destinationChainId: this.chainId,
         };
         const relayDataHash = getRelayDataHash(slowFillRequest, this.chainId);
+        if (this.slowFillRequests[relayDataHash] !== undefined) {
+          continue;
+        }
         this.slowFillRequests[relayDataHash] = slowFillRequest;
       }
     }
@@ -863,9 +889,6 @@ export class SpokePoolClient extends BaseAbstractClient {
       }
     }
 
-    // Exact sequencing of relayer refund executions doesn't seem to be important. There are very few consumers of
-    // these objects, and they are typically used to search for a specific rootBundleId & leafId pair. Therefore,
-    // relayerRefundExecutions don't need exact sequencing and parsing of v2/v3 events can occur without sorting.
     if (eventsToQuery.includes("ExecutedRelayerRefundRoot")) {
       const refundEvents = queryResults[eventsToQuery.indexOf("ExecutedRelayerRefundRoot")];
       for (const event of refundEvents) {
@@ -1018,7 +1041,7 @@ export class SpokePoolClient extends BaseAbstractClient {
    * @returns The deposit if found.
    * @note This method is used to find deposits that are outside of the search range of this client.
    */
-  async findDeposit(depositId: number, destinationChainId: number, depositor: string): Promise<DepositWithBlock> {
+  async findDeposit(depositId: number, destinationChainId: number, depositor: string): Promise<V2DepositWithBlock> {
     // Binary search for block. This way we can get the blocks before and after the deposit with
     // deposit ID = fill.depositId and use those blocks to optimize the search for that deposit.
     // Stop searches after a maximum # of searches to limit number of eth_call requests. Make an
@@ -1066,11 +1089,11 @@ export class SpokePoolClient extends BaseAbstractClient {
           ` between ${srcChain} blocks [${searchBounds.low}, ${searchBounds.high}]`
       );
     }
-    const partialDeposit = spreadEventWithBlockNumber(event) as DepositWithBlock;
+    const partialDeposit = spreadEventWithBlockNumber(event) as V2DepositWithBlock;
     const { realizedLpFeePct, quoteBlock: quoteBlockNumber } = (await this.batchComputeRealizedLpFeePct([event]))[0]; // Append the realizedLpFeePct.
 
     // Append destination token and realized lp fee to deposit.
-    const deposit: DepositWithBlock = {
+    const deposit: V2DepositWithBlock = {
       ...partialDeposit,
       realizedLpFeePct,
       destinationToken: this.getDestinationTokenForDeposit(partialDeposit),
@@ -1086,4 +1109,80 @@ export class SpokePoolClient extends BaseAbstractClient {
 
     return deposit;
   }
+
+  async findDepositV3(depositId: number, destinationChainId: number, depositor: string): Promise<V3DepositWithBlock> {
+    // Binary search for event search bounds. This way we can get the blocks before and after the deposit with
+    // deposit ID = fill.depositId and use those blocks to optimize the search for that deposit.
+    // Stop searches after a maximum # of searches to limit number of eth_call requests. Make an
+    // eth_getLogs call on the remaining block range (i.e. the [low, high] remaining from the binary
+    // search) to find the target deposit ID.
+    //
+    // @dev Limiting between 5-10 searches empirically performs best when there are ~300,000 deposits
+    // for a spoke pool and we're looking for a deposit <5 days older than HEAD.
+    const searchBounds = await this._getBlockRangeForDepositId(
+      depositId,
+      this.deploymentBlock,
+      this.latestBlockSearched,
+      7
+    );
+
+    const tStart = Date.now();
+    const query = await paginatedEventQuery(
+      this.spokePool,
+      this.spokePool.filters.V3FundsDeposited(
+        null,
+        null,
+        null,
+        null,
+        destinationChainId,
+        depositId,
+        null,
+        null,
+        null,
+        depositor,
+        null,
+        null,
+        null
+      ),
+      {
+        fromBlock: searchBounds.low,
+        toBlock: searchBounds.high,
+        maxBlockLookBack: this.eventSearchConfig.maxBlockLookBack,
+      }
+    );
+    const tStop = Date.now();
+
+    const event = (query as V3FundsDepositedEvent[]).find((deposit) => deposit.args.depositId === depositId);
+    if (event === undefined) {
+      const srcChain = getNetworkName(this.chainId);
+      const dstChain = getNetworkName(destinationChainId);
+      throw new Error(
+        `Could not find deposit ${depositId} for ${dstChain} fill` +
+          ` between ${srcChain} blocks [${searchBounds.low}, ${searchBounds.high}]`
+      );
+    }
+    const partialDeposit = spreadEventWithBlockNumber(event) as V3DepositWithBlock;
+    const { realizedLpFeePct, quoteBlock: quoteBlockNumber } = (await this.batchComputeRealizedLpFeePct([event]))[0]; // Append the realizedLpFeePct.
+
+    // Append destination token and realized lp fee to deposit.
+    const deposit: V3DepositWithBlock = {
+      ...partialDeposit,
+      originChainId: this.chainId,
+      realizedLpFeePct,
+      quoteBlockNumber,
+      outputToken:
+        partialDeposit.outputToken === ZERO_ADDRESS
+          ? this.getDestinationTokenForDeposit(partialDeposit)
+          : partialDeposit.outputToken,
+    };
+
+    this.logger.debug({
+      at: "SpokePoolClient#findDepositV3",
+      message: "Located V3 deposit outside of SpokePoolClient's search range",
+      deposit,
+      elapsedMs: tStop - tStart,
+    });
+
+    return deposit;
+  }
 }

src/interfaces/SpokePool.ts

@@ -166,7 +166,7 @@ export interface RelayerRefundExecution extends RelayerRefundLeaf {
 export interface RelayerRefundExecutionWithBlock extends RelayerRefundExecution, SortableEvent {}
 
 export interface UnfilledDeposit {
-  deposit: Deposit;
+  deposit: V2Deposit;
   unfilledAmount: BigNumber;
   hasFirstPartialFill?: boolean;
   relayerBalancingFee?: BigNumber;
@@ -182,7 +182,7 @@ export interface Refund {
 export type FillsToRefund = {
   [repaymentChainId: number]: {
     [l2TokenAddress: string]: {
-      fills: Fill[];
+      fills: V2Fill[];
       refunds?: Refund;
       totalRefundAmount: BigNumber;
       realizedLpFees: BigNumber;

src/utils/CachingUtils.ts

@@ -1,5 +1,5 @@
 import { DEFAULT_CACHING_SAFE_LAG, DEFAULT_CACHING_TTL } from "../constants";
-import { CachingMechanismInterface, Deposit, Fill } from "../interfaces";
+import { CachingMechanismInterface, Deposit, Fill, SlowFillRequest } from "../interfaces";
 import { assert } from "./LogUtils";
 import { composeRevivers, objectWithBigNumberReviver } from "./ReviverUtils";
 import { getCurrentTime } from "./TimeUtils";
@@ -44,10 +44,10 @@ export async function setDepositInCache(
 }
 
 /**
- * Resolves the key for caching either a deposit or a fill.
- * @param depositOrFill Either a deposit or a fill. In either case, the depositId and originChainId are used to generate the key.
- * @returns The key for caching the deposit or fill.
+ * Resolves the key for caching a bridge event.
+ * @param bridgeEvent The depositId and originChainId are used to generate the key.
+ * @returns The key for caching the event.
  */
-export function getDepositKey(depositOrFill: Deposit | Fill): string {
-  return `deposit_${depositOrFill.originChainId}_${depositOrFill.depositId}`;
+export function getDepositKey(bridgeEvent: Deposit | Fill | SlowFillRequest): string {
+  return `deposit_${bridgeEvent.originChainId}_${bridgeEvent.depositId}`;
 }

src/utils/DepositUtils.ts

@@ -1,14 +1,14 @@
 import assert from "assert";
 import { SpokePoolClient } from "../clients";
 import { DEFAULT_CACHING_TTL, EMPTY_MESSAGE } from "../constants";
-import { CachingMechanismInterface, Deposit, DepositWithBlock, Fill } from "../interfaces";
+import { CachingMechanismInterface, Deposit, DepositWithBlock, Fill, SlowFillRequest } from "../interfaces";
 import { getNetworkName } from "./NetworkUtils";
 import { getDepositInCache, getDepositKey, setDepositInCache } from "./CachingUtils";
 import { validateFillForDeposit } from "./FlowUtils";
 import { getCurrentTime } from "./TimeUtils";
 import { isDefined } from "./TypeGuards";
 import { isDepositFormedCorrectly } from "./ValidatorUtils";
-import { isV2Deposit, isV3Deposit } from "./V3Utils";
+import { isV2Deposit, isV2Fill, isV3Deposit } from "./V3Utils";
 
 // Load a deposit for a fill if the fill's deposit ID is outside this client's search range.
 // This can be used by the Dataworker to determine whether to give a relayer a refund for a fill
@@ -38,7 +38,7 @@ export type DepositSearchResult =
  */
 export async function queryHistoricalDepositForFill(
   spokePoolClient: SpokePoolClient,
-  fill: Fill,
+  fill: Fill | SlowFillRequest,
   cache?: CachingMechanismInterface
 ): Promise<DepositSearchResult> {
   if (fill.originChainId !== spokePoolClient.chainId) {
@@ -98,7 +98,11 @@ export async function queryHistoricalDepositForFill(
   if (isDefined(cachedDeposit)) {
     deposit = cachedDeposit as DepositWithBlock;
   } else {
-    deposit = await spokePoolClient.findDeposit(fill.depositId, fill.destinationChainId, fill.depositor);
+    if (isV2Fill(fill)) {
+      deposit = await spokePoolClient.findDeposit(fill.depositId, fill.destinationChainId, fill.depositor);
+    } else {
+      deposit = await spokePoolClient.findDepositV3(fill.depositId, fill.destinationChainId, fill.depositor);
+    }
     if (cache) {
       await setDepositInCache(deposit, getCurrentTime(), cache, DEFAULT_CACHING_TTL);
     }

src/utils/SpokeUtils.ts

@@ -1,7 +1,7 @@
 import assert from "assert";
 import { Contract, utils as ethersUtils } from "ethers";
 import { CHAIN_IDs } from "../constants";
-import { FillStatus, RelayData, V2RelayData, V3RelayData } from "../interfaces";
+import { FillStatus, RelayData, SlowFillRequest, V2RelayData, V3Deposit, V3Fill, V3RelayData } from "../interfaces";
 import { SpokePoolClient } from "../clients";
 import { isDefined } from "./TypeGuards";
 import { isV2RelayData } from "./V3Utils";
@@ -240,6 +240,9 @@ export function getV3RelayHash(relayData: V3RelayData, destinationChainId: numbe
   );
 }
 
+export function getV3RelayHashFromEvent(e: V3Deposit | V3Fill | SlowFillRequest): string {
+  return getV3RelayHash(e, e.destinationChainId);
+}
 /**
  * Find the amount filled for a deposit at a particular block.
  * @param spokePool SpokePool contract instance.

src/utils/V3Utils.ts

@@ -13,6 +13,7 @@ import {
   V3SpeedUp,
 } from "../interfaces";
 import { BN } from "./BigNumberUtils";
+import { fixedPointAdjustment } from "./common";
 
 // Lowest ConfigStore version where the V3 model is in effect. The version update to the following value should take
 // place atomically with the SpokePool upgrade to V3 so that the dataworker knows what kind of MerkleLeaves to propose
@@ -168,3 +169,12 @@ export function getSlowFillLeafChainId<
 >(leaf: T | U): number {
   return unsafeIsType<U, T>(leaf, "chainId") ? leaf.chainId : leaf.relayData.destinationChainId;
 }
+
+export function getSlowFillLeafLpFeePct<
+  T extends { relayData: { realizedLpFeePct: V2SlowFillLeaf["relayData"]["realizedLpFeePct"] } },
+  U extends Pick<V3SlowFillLeaf, "updatedOutputAmount" | "relayData">,
+>(leaf: T | U): BN {
+  return unsafeIsType<U, T>(leaf, "updatedOutputAmount")
+    ? leaf.relayData.inputAmount.sub(leaf.updatedOutputAmount).mul(fixedPointAdjustment).div(leaf.relayData.inputAmount)
+    : leaf.relayData.realizedLpFeePct;
+}

src/utils/ValidatorUtils.ts

@@ -6,7 +6,7 @@ const AddressValidator = define<string>("AddressValidator", (v) => ethers.utils.
 const HexValidator = define<string>("HexValidator", (v) => ethers.utils.isHexString(String(v)));
 const BigNumberValidator = define<BigNumber>("BigNumberValidator", (v) => ethers.BigNumber.isBigNumber(v));
 
-const DepositSchema = object({
+const V2DepositSchema = object({
   depositId: Min(integer(), 0),
   depositor: AddressValidator,
   recipient: AddressValidator,
@@ -28,9 +28,35 @@ const DepositSchema = object({
   logIndex: Min(integer(), 0),
   quoteBlockNumber: Min(integer(), 0),
   transactionHash: HexValidator,
-  blockTimestamp: optional(Min(integer(), 0)),
+});
+
+const V3DepositSchema = object({
+  depositId: Min(integer(), 0),
+  depositor: AddressValidator,
+  recipient: AddressValidator,
+  inputToken: AddressValidator,
+  inputAmount: BigNumberValidator,
+  originChainId: Min(integer(), 0),
+  destinationChainId: Min(integer(), 0),
+  quoteTimestamp: Min(integer(), 0),
+  fillDeadline: Min(integer(), 0),
+  exclusivityDeadline: Min(integer(), 0),
+  exclusiveRelayer: AddressValidator,
+  realizedLpFeePct: optional(BigNumberValidator),
+  outputToken: AddressValidator,
+  outputAmount: BigNumberValidator,
+  message: string(),
+  speedUpSignature: optional(HexValidator),
+  updatedOutputAmount: optional(BigNumberValidator),
+  updatedRecipient: optional(string()),
+  updatedMessage: optional(string()),
+  blockNumber: Min(integer(), 0),
+  transactionIndex: Min(integer(), 0),
+  logIndex: Min(integer(), 0),
+  quoteBlockNumber: Min(integer(), 0),
+  transactionHash: HexValidator,
 });
 
 export function isDepositFormedCorrectly(deposit: unknown): deposit is DepositWithBlock {
-  return DepositSchema.is(deposit);
+  return V2DepositSchema.is(deposit) || V3DepositSchema.is(deposit);
 }

test/utils/SpokePoolUtils.ts

@@ -58,9 +58,9 @@ export function v3FillFromDeposit(deposit: V3DepositWithBlock, relayer: string):
     exclusiveRelayer: relayer,
     repaymentChainId: deposit.destinationChainId,
     relayExecutionInfo: {
-      recipient: deposit.updatedRecipient ?? recipient,
-      message: deposit.updatedMessage ?? message,
-      outputAmount: deposit.updatedOutputAmount ?? deposit.outputAmount,
+      updatedRecipient: deposit.updatedRecipient ?? recipient,
+      updatedMessage: deposit.updatedMessage ?? message,
+      updatedOutputAmount: deposit.updatedOutputAmount ?? deposit.outputAmount,
       fillType: FillType.FastFill,
     },
   };

test/utils/utils.ts

@@ -444,10 +444,10 @@ export async function fillV3Relay(
     exclusiveRelayer: args.exclusiveRelayer,
     relayer: args.relayer,
     repaymentChainId: Number(args.repaymentChainId),
-    updatableRelayData: {
-      recipient: args.relayExecutionInfo.recipient,
-      message: args.relayExecutionInfo.message,
-      outputAmount: args.relayExecutionInfo.outputAmount,
+    relayExecutionInfo: {
+      updatedRecipient: args.relayExecutionInfo.recipient,
+      updatedMessage: args.relayExecutionInfo.message,
+      updatedOutputAmount: args.relayExecutionInfo.outputAmount,
       fillType: args.relayExecutionInfo.fillType,
     },
     blockNumber,