diff --git a/contracts/tasks/crossChain.js b/contracts/tasks/crossChain.js index 6bd9f95ad8..43ab41dc90 100644 --- a/contracts/tasks/crossChain.js +++ b/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, diff --git a/contracts/tasks/test/crosschain/source-reads-retry.js b/contracts/tasks/test/crosschain/source-reads-retry.js new file mode 100644 index 0000000000..40b9fa16ec --- /dev/null +++ b/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); + }); +});