fix(tasks): retry transient source-chain RPC errors in the CCTP relay scan

contracts/tasks/crossChain.js

@@ -196,6 +196,31 @@ const isMessageForDestination = ({
   );
 };
 
+// The source-chain reads go through a bare ethers JsonRpcProvider, which does
+// not retry. The load-balanced providers behind it (dRPC) intermittently answer
+// with transient errors: 500 "Temporary internal error. Please retry", 408
+// timeouts, and 400 "Unknown block" when eth_getLogs is routed to a node that
+// has not yet seen the block eth_blockNumber just returned. The reads are
+// idempotent, so retry with exponential backoff before failing the run.
+const withRetry = async (fn, { label, attempts = 5, baseDelayMs = 1000 }) => {
+  for (let attempt = 1; ; attempt++) {
+    try {
+      return await fn();
+    } catch (error) {
+      if (attempt >= attempts) {
+        throw error;
+      }
+      const delayMs = baseDelayMs * 2 ** (attempt - 1);
+      log(
+        `${label} failed (attempt ${attempt}/${attempts}): ${String(
+          error.message
+        ).slice(0, 300)}. Retrying in ${delayMs}ms`
+      );
+      await new Promise((resolve) => setTimeout(resolve, delayMs));
+    }
+  }
+};
+
 // TokensBridged & MessageTransmitted are emitted when a CCTP message is posted.
 // A single source transaction can emit multiple CCTP messages.
 const fetchTxHashesFromCctpTransactions = async ({
@@ -209,7 +234,10 @@ const fetchTxHashesFromCctpTransactions = async ({
     resolvedFromBlock = overrideBlock;
     resolvedToBlock = overrideBlock;
   } else {
-    const latestBlock = await sourceChainProvider.getBlockNumber();
+    const latestBlock = await withRetry(
+      () => sourceChainProvider.getBlockNumber(),
+      { label: "eth_blockNumber" }
+    );
     resolvedFromBlock = Math.max(latestBlock - blockLookback, 0);
     resolvedToBlock = latestBlock;
   }
@@ -224,20 +252,21 @@ const fetchTxHashesFromCctpTransactions = async ({
   log(
     `Fetching event logs from block ${resolvedFromBlock} to block ${resolvedToBlock}`
   );
+  const getLogs = (eventName, topic) =>
+    withRetry(
+      () =>
+        sourceChainProvider.getLogs({
+          address: cctpIntegrationContractSource.address,
+          fromBlock: resolvedFromBlock,
+          toBlock: resolvedToBlock,
+          topics: [topic],
+        }),
+      { label: `eth_getLogs ${eventName}` }
+    );
   const [eventLogsTokenBridged, eventLogsMessageTransmitted] =
     await Promise.all([
-      sourceChainProvider.getLogs({
-        address: cctpIntegrationContractSource.address,
-        fromBlock: resolvedFromBlock,
-        toBlock: resolvedToBlock,
-        topics: [tokensBridgedTopic],
-      }),
-      sourceChainProvider.getLogs({
-        address: cctpIntegrationContractSource.address,
-        fromBlock: resolvedFromBlock,
-        toBlock: resolvedToBlock,
-        topics: [messageTransmittedTopic],
-      }),
+      getLogs("TokensBridged", tokensBridgedTopic),
+      getLogs("MessageTransmitted", messageTransmittedTopic),
     ]);
 
   // There should be no duplicates in the event logs, but still deduplicate to be safe
@@ -439,6 +468,8 @@ const processCctpBridgeTransactions = async ({
 
 module.exports = {
   processCctpBridgeTransactions,
+  fetchTxHashesFromCctpTransactions,
+  withRetry,
   decodeOriginMessage,
   nonceIsReplayKey,
   TX_HASH_REGEX,

contracts/tasks/test/crosschain/source-reads-retry.js

@@ -0,0 +1,94 @@
+const { expect } = require("chai");
+const { ethers } = require("ethers");
+
+const {
+  withRetry,
+  fetchTxHashesFromCctpTransactions,
+} = require("../../../tasks/crossChain");
+
+// The relay actions scan the source chain with eth_blockNumber + eth_getLogs
+// over a bare ethers JsonRpcProvider. The load-balanced providers behind it
+// return transient errors (dRPC 500 "Temporary internal error", 408 timeouts,
+// 400 "Unknown block" from a node behind the tip), so those reads retry.
+describe("Unit: CCTP relay source-chain read retries", () => {
+  describe("withRetry", () => {
+    it("returns the result once the call succeeds", async () => {
+      let calls = 0;
+      const result = await withRetry(
+        async () => {
+          calls += 1;
+          if (calls < 3) throw new Error("Temporary internal error");
+          return "ok";
+        },
+        { label: "test", attempts: 5, baseDelayMs: 1 }
+      );
+      expect(result).to.equal("ok");
+      expect(calls).to.equal(3);
+    });
+
+    it("rethrows the last error after the final attempt", async () => {
+      let calls = 0;
+      let thrown;
+      try {
+        await withRetry(
+          async () => {
+            calls += 1;
+            throw new Error(`fail ${calls}`);
+          },
+          { label: "test", attempts: 3, baseDelayMs: 1 }
+        );
+      } catch (error) {
+        thrown = error;
+      }
+      expect(thrown.message).to.equal("fail 3");
+      expect(calls).to.equal(3);
+    });
+  });
+
+  describe("fetchTxHashesFromCctpTransactions", () => {
+    const address = "0x0000000000000000000000000000000000000001";
+    const config = {
+      cctpIntegrationContractSource: {
+        address,
+        interface: new ethers.utils.Interface([
+          "event TokensBridged(uint256 amount)",
+          "event MessageTransmitted(bytes message)",
+        ]),
+      },
+    };
+    const latestBlock = 45411379;
+    const blockLookback = 10000;
+    const txHash = `0x${"ab".repeat(32)}`;
+
+    // eth_blockNumber answered from one node, the first eth_getLogs routed to a
+    // node that has not seen that block yet (dRPC "Unknown block"), then fine.
+    it("recovers from a transient eth_getLogs error and scans the full range", async () => {
+      const getLogsCalls = [];
+      const provider = {
+        getBlockNumber: async () => latestBlock,
+        getLogs: async (filter) => {
+          getLogsCalls.push(filter);
+          if (getLogsCalls.length === 1) {
+            throw new Error("bad response (status=400, body=Unknown block)");
+          }
+          return [{ transactionHash: txHash }];
+        },
+      };
+
+      const { allTxHashes } = await fetchTxHashesFromCctpTransactions({
+        config,
+        blockLookback,
+        sourceChainProvider: provider,
+      });
+
+      expect(allTxHashes).to.deep.equal([txHash]);
+      // 2 events, +1 retried call
+      expect(getLogsCalls).to.have.length(3);
+      for (const filter of getLogsCalls) {
+        expect(filter.address).to.equal(address);
+        expect(filter.fromBlock).to.equal(latestBlock - blockLookback);
+        expect(filter.toBlock).to.equal(latestBlock);
+      }
+    }).timeout(5000);
+  });
+});