feat(finalizer): Support OFT retries

src/adapter/bridges/OFTBridge.ts

@@ -1,4 +1,4 @@
-import { BytesLike, Contract, Signer } from "ethers";
+import { Contract, Signer } from "ethers";
 import { BridgeTransactionDetails, BaseBridgeAdapter, BridgeEvents } from "./BaseBridgeAdapter";
 import {
   BigNumber,
@@ -17,16 +17,13 @@ import { processEvent } from "../utils";
 import * as OFT from "../../utils/OFTUtils";
 import { OFT_DEFAULT_FEE_CAP, OFT_FEE_CAP_OVERRIDES } from "../../common/Constants";
 import { IOFT_ABI_FULL } from "../../common/ContractAddresses";
-import { Options } from "@layerzerolabs/lz-v2-utilities";
 
 type OFTBridgeArguments = {
   sendParamStruct: OFT.SendParamStruct;
   feeStruct: OFT.MessagingFeeStruct;
   refundAddress: string;
 };
 
-const MONAD_EXECUTOR_LZ_RECEIVE_GAS_LIMIT = 120000;
-
 export class OFTBridge extends BaseBridgeAdapter {
   public readonly l2TokenAddress: string;
   private readonly l1ChainEid: number;
@@ -115,12 +112,10 @@ export class OFTBridge extends BaseBridgeAdapter {
     // We round `amount` to a specific precision to prevent rounding on the contract side. This way, we
     // receive the exact amount we sent in the transaction
     const roundedAmount = await this.roundAmountToSend(amount);
-    let extraOptions: BytesLike = "0x";
-    if (this.l2chainId === CHAIN_IDs.MONAD) {
-      extraOptions = Options.newOptions().addExecutorLzReceiveOption(MONAD_EXECUTOR_LZ_RECEIVE_GAS_LIMIT).toBytes();
-    }
+    const dstEid = this.l2ChainEid;
+    const extraOptions = OFT.boostGasLimit(dstEid);
     const sendParamStruct: OFT.SendParamStruct = {
-      dstEid: this.l2ChainEid,
+      dstEid,
       to: OFT.formatToAddress(toAddress),
       amountLD: roundedAmount,
       // @dev Setting `minAmountLD` equal to `amountLD` ensures we won't hit contract-side rounding

src/common/abi/LayerZeroV2Endpoint.json

@@ -0,0 +1,75 @@
+[
+  {
+    "type": "event",
+    "anonymous": false,
+    "name": "OFTSent",
+    "inputs": [
+      {
+        "type": "bytes32",
+        "name": "guid",
+        "indexed": true
+      },
+      {
+        "type": "uint32",
+        "name": "dstEid"
+      },
+      {
+        "type": "address",
+        "name": "fromAddress",
+        "indexed": true
+      },
+      {
+        "type": "uint256",
+        "name": "amountSentLD"
+      },
+      {
+        "type": "uint256",
+        "name": "amountReceivedLD"
+      }
+    ]
+  },
+  {
+    "type": "function",
+    "name": "lzReceive",
+    "constant": false,
+    "stateMutability": "payable",
+    "payable": true,
+    "inputs": [
+      {
+        "type": "tuple",
+        "name": "origin",
+        "components": [
+          {
+            "type": "uint32",
+            "name": "srcEid"
+          },
+          {
+            "type": "bytes32",
+            "name": "sender"
+          },
+          {
+            "type": "uint64",
+            "name": "nonce"
+          }
+        ]
+      },
+      {
+        "type": "address",
+        "name": "_receiver"
+      },
+      {
+        "type": "bytes32",
+        "name": "_guid"
+      },
+      {
+        "type": "bytes",
+        "name":"_message"
+      },
+      {
+        "type": "bytes",
+        "name": "_extraData"
+      }
+    ],
+    "outputs": []
+  }
+]

src/common/abi/LayerZeroV2Messenger.json

@@ -0,0 +1,15 @@
+[
+  {
+    "type": "function",
+    "name": "endpoint",
+    "constant": true,
+    "stateMutability": "view",
+    "payable": false,
+    "inputs": [],
+    "outputs": [
+      {
+        "type": "address"
+      }
+    ]
+  }
+]

src/finalizer/index.ts

@@ -1,4 +1,4 @@
-import { CCTP_NO_DOMAIN, ChainFamily, PRODUCTION_NETWORKS } from "@across-protocol/constants";
+import { CCTP_NO_DOMAIN, ChainFamily, OFT_NO_EID, PRODUCTION_NETWORKS } from "@across-protocol/constants";
 import { utils as sdkUtils } from "@across-protocol/sdk";
 import assert from "assert";
 import { Contract } from "ethers";
@@ -106,7 +106,7 @@ function generateChainConfig(): void {
     [ChainFamily.ZK_STACK]: zkSyncFinalizer,
   };
 
-  Object.entries(PRODUCTION_NETWORKS).forEach(([_chainId, { cctpDomain, family }]) => {
+  Object.entries(PRODUCTION_NETWORKS).forEach(([_chainId, { cctpDomain, family, oftEid }]) => {
     const chainId = Number(_chainId);
     const config = (chainFinalizers[chainId] ??= {});
     config.finalizeOnL1 ??= [];
@@ -127,8 +127,7 @@ function generateChainConfig(): void {
       config.finalizeOnAny.push(cctpV2Finalizer);
     }
 
-    // @todo Once contracts are linked, change this to add all chains w/ OFT enabled.
-    if (chainId === CHAIN_IDs.ARBITRUM) {
+    if (oftEid !== OFT_NO_EID) {
       config.finalizeOnAny.push(oftRetryFinalizer);
     }
   });

src/finalizer/utils/oftRetry.ts

@@ -1,18 +1,32 @@
+import { ethers } from "ethers";
+import { PRODUCTION_OFT_EIDs } from "@across-protocol/constants";
+import LZEndpoint from "../../common/abi/LayerZeroV2Endpoint.json";
+import LZMessenger from "../../common/abi/LayerZeroV2Messenger.json";
 import { SpokePoolClient } from "../../clients";
+import * as OFT from "../../utils/OFTUtils";
 import {
+  CHAIN_IDs,
   EventSearchConfig,
+  EvmAddress,
+  isDefined,
   assert,
+  getProvider,
   groupObjectCountsByProp,
   winston,
+  getSrcOftPeriphery,
   isEVMSpokePoolClient,
   chunk,
-  getSrcOftMessages,
-  getLzTransactionDetails,
   mapAsync,
-  getChainIdFromEndpointId,
+  paginatedEventQuery,
+  Provider,
+  spreadEventWithBlockNumber,
+  toAddressType,
+  TOKEN_SYMBOLS_MAP,
 } from "../../utils";
 import { FinalizerPromise, CrossChainMessage } from "../types";
 
+const OFT_RETRY_TOKENS = ["USDT"];
+
 /**
  * Finalizes failed lzCompose messages on destination by checking for failed transactions and resubmitting its calldata..
  * @param logger Logger instance.
@@ -25,28 +39,49 @@ export async function oftRetryFinalizer(
 ): Promise<FinalizerPromise> {
   assert(isEVMSpokePoolClient(spokePoolClient), "Cannot retry LZ messages on non-EVM networks.");
   const srcProvider = spokePoolClient.spokePool.provider;
+
   const searchConfig: EventSearchConfig = {
     from: spokePoolClient.eventSearchConfig.from,
     to: spokePoolClient.latestHeightSearched,
     maxLookBack: spokePoolClient.eventSearchConfig.maxLookBack,
   };
-  const depositInitiatedMessages = await getSrcOftMessages(spokePoolClient.chainId, searchConfig, srcProvider);
-  const _outstandingMessages = [];
+  const promises = [getSrcOftMessages(spokePoolClient.chainId, searchConfig, srcProvider)];
+  OFT_RETRY_TOKENS.map((symbol) => TOKEN_SYMBOLS_MAP[symbol]?.addresses[spokePoolClient.chainId])
+    .filter(isDefined)
+    .forEach((address) =>
+      promises.push(OFT.getOFTSent(spokePoolClient.chainId, EvmAddress.from(address), searchConfig, srcProvider))
+    );
+
+  const deposits = (await Promise.all(promises)).flat();
+
   // To avoid rate-limiting, chunk API queries.
   const chunkSize = Number(process.env["LZ_API_CHUNK_SIZE"] ?? 8);
-  for (const depositInitiatedMessageChunk of chunk(depositInitiatedMessages, chunkSize)) {
+
+  const _outstandingMessages = [];
+  for (const depositChunk of chunk(deposits, chunkSize)) {
     _outstandingMessages.push(
-      ...(await mapAsync(depositInitiatedMessageChunk, async ({ txnRef }) => {
-        return await getLzTransactionDetails(txnRef);
-      }))
+      ...(await mapAsync(depositChunk, async ({ txnRef }) => await OFT.getLzTransactionDetails(txnRef)))
     );
   }
   const outstandingMessages = _outstandingMessages.map(({ data }) => data.flat()).flat();
 
   // Lz messages are executed automatically and must be retried only if their execution reverts on chain.
-  const unprocessedMessages = outstandingMessages.filter(({ destination }) => destination?.status !== "SUCCEEDED");
+  const knownEids = Object.values(PRODUCTION_OFT_EIDs);
+  const uniqueEids: number[] = [];
+  const checkStatus = (status?: string) => !["WAITING", "SUCCEEDED"].includes(status);
+  const unprocessedMessages = outstandingMessages.filter(({ pathway, destination }) => {
+    if (!knownEids.includes(pathway.dstEid)) {
+      return false;
+    }
+    if (!uniqueEids.includes(pathway.dstEid)) {
+      uniqueEids.push(pathway.dstEid);
+    }
+
+    return checkStatus(destination?.status);
+  });
+
   const statusesGrouped = groupObjectCountsByProp(
-    outstandingMessages.map(({ destination }) => destination),
+    unprocessedMessages.map(({ destination }) => destination),
     (message: { status: string }) => message.status
   );
   logger.debug({
@@ -55,21 +90,39 @@ export async function oftRetryFinalizer(
     statusesGrouped,
   });
 
-  const destinationTransactions = await mapAsync(unprocessedMessages, async ({ source }) => {
-    return await srcProvider.getTransaction(source.tx);
-  });
+  const endpointAbi = new ethers.utils.Interface(LZEndpoint);
+  const messengerAbi = new ethers.utils.Interface(LZMessenger);
+  const callData = await Promise.all(
+    unprocessedMessages.map(async (message) => {
+      const { guid } = message;
+      const {
+        srcEid,
+        dstEid,
+        nonce,
+        sender: { address: senderAddress },
+        receiver: { address: receiver },
+      } = message.pathway;
+      const { payload } = message.source.tx;
 
-  const callData = destinationTransactions.map((txData) => {
-    return {
-      target: txData.to,
-      callData: txData.data,
-    };
-  });
+      const dstChainId = OFT.getChainIdFromEndpointId(dstEid);
+      const provider = await getProvider(dstChainId);
+      const messenger = new ethers.Contract(receiver, messengerAbi, provider);
+      const target = await messenger.endpoint();
+
+      const sender = toAddressType(senderAddress, spokePoolClient.chainId).toBytes32();
+      const extraData = OFT.boostGasLimit(dstEid);
+
+      const args = [{ srcEid, sender, nonce }, receiver, guid, payload, extraData];
+      const callData = endpointAbi.encodeFunctionData("lzReceive", args);
+
+      return { target, callData };
+    })
+  );
 
-  const crossChainMessages = unprocessedMessages.map((unprocessedMessage) => {
+  const crossChainMessages = unprocessedMessages.map(({ pathway: { dstEid } }) => {
     return {
       originationChainId: spokePoolClient.chainId,
-      destinationChainId: getChainIdFromEndpointId(unprocessedMessage.pathway.dstEid),
+      destinationChainId: OFT.getChainIdFromEndpointId(dstEid),
       type: "misc",
       miscReason: "oftRetry",
     } as CrossChainMessage;
@@ -80,3 +133,25 @@ export async function oftRetryFinalizer(
     callData,
   };
 }
+
+/**
+ * @notice Fetches OFT messages initiated from a srcOft contract
+ * @param srcChainId Chain ID corresponding to the deployed srcOftMessenger.
+ * @param searchConfig Event search config to use on srcChainId.
+ * @param srcProvider ethers Provider instance for the srcChainId.
+ * @returns A list of SortableEvents corresponding to bridge events on srcChainId.
+ */
+export async function getSrcOftMessages(
+  srcChainId: number,
+  searchConfig: EventSearchConfig,
+  srcProvider: Provider
+): Promise<OFT.LzBridgeEvent[]> {
+  // srfOft contract currently only deployed to Arbitrum. @todo.
+  if (![CHAIN_IDs.ARBITRUM].includes(srcChainId)) {
+    return [];
+  }
+
+  const srcOft = getSrcOftPeriphery(srcChainId).connect(srcProvider);
+  const messageInitiatedEvents = await paginatedEventQuery(srcOft, srcOft.filters.SponsoredOFTSend(), searchConfig);
+  return messageInitiatedEvents.map(spreadEventWithBlockNumber);
+}