fix: improve reliability of hook indexing (#4214)

### Description

After deploying this, should no longer see logs like
https://cloudlogging.app.goo.gl/gK4ByT1qLXTG7VN19

### Drive-by changes

<!--
Are there any minor or drive-by changes also included?
-->

### Related issues

- Fixes https://github.com/hyperlane-xyz/hyperlane-monorepo/issues/4212

### Backward compatibility

<!--
Are these changes backward compatible? Are there any infrastructure
implications, e.g. changes that would prohibit deploying older commits
using this infra tooling?

Yes/No
-->

### Testing

<!--
What kind of testing have these changes undergone?

None/Manual/Unit Tests
-->

rust/chains/hyperlane-ethereum/src/contracts/interchain_gas.rs

@@ -7,6 +7,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use ethers::prelude::Middleware;
+use hyperlane_core::rpc_clients::call_and_retry_indefinitely;
 use hyperlane_core::{
     ChainCommunicationError, ChainResult, ContractLocator, HyperlaneAbi, HyperlaneChain,
     HyperlaneContract, HyperlaneDomain, HyperlaneProvider, Indexed, Indexer,
@@ -14,7 +15,7 @@ use hyperlane_core::{
 };
 use tracing::instrument;
 
-use super::utils::fetch_raw_logs_and_log_meta;
+use super::utils::fetch_raw_logs_and_meta;
 use crate::interfaces::i_interchain_gas_paymaster::{
     GasPaymentFilter, IInterchainGasPaymaster as EthereumInterchainGasPaymasterInternal,
     IINTERCHAINGASPAYMASTER_ABI,
@@ -132,25 +133,29 @@ where
         &self,
         tx_hash: H512,
     ) -> ChainResult<Vec<(Indexed<InterchainGasPayment>, LogMeta)>> {
-        let logs = fetch_raw_logs_and_log_meta::<GasPaymentFilter, M>(
-            tx_hash,
-            self.provider.clone(),
-            self.contract.address(),
-        )
-        .await?
-        .into_iter()
-        .map(|(log, log_meta)| {
-            (
-                Indexed::new(InterchainGasPayment {
-                    message_id: H256::from(log.message_id),
-                    destination: log.destination_domain,
-                    payment: log.payment.into(),
-                    gas_amount: log.gas_amount.into(),
-                }),
-                log_meta,
-            )
+        let raw_logs_and_meta = call_and_retry_indefinitely(|| {
+            let provider = self.provider.clone();
+            let contract = self.contract.address();
+            Box::pin(async move {
+                fetch_raw_logs_and_meta::<GasPaymentFilter, M>(tx_hash, provider, contract).await
+            })
         })
-        .collect();
+        .await;
+
+        let logs = raw_logs_and_meta
+            .into_iter()
+            .map(|(log, log_meta)| {
+                (
+                    Indexed::new(InterchainGasPayment {
+                        message_id: H256::from(log.message_id),
+                        destination: log.destination_domain,
+                        payment: log.payment.into(),
+                        gas_amount: log.gas_amount.into(),
+                    }),
+                    log_meta,
+                )
+            })
+            .collect();
         Ok(logs)
     }
 }

rust/chains/hyperlane-ethereum/src/contracts/mailbox.rs

@@ -13,6 +13,7 @@ use ethers::prelude::Middleware;
 use ethers_contract::builders::ContractCall;
 use ethers_contract::{Multicall, MulticallResult};
 use futures_util::future::join_all;
+use hyperlane_core::rpc_clients::call_and_retry_indefinitely;
 use hyperlane_core::{BatchResult, QueueOperation, H512};
 use itertools::Itertools;
 use tracing::instrument;
@@ -34,7 +35,7 @@ use crate::tx::{call_with_lag, fill_tx_gas_params, report_tx};
 use crate::{BuildableWithProvider, ConnectionConf, EthereumProvider, TransactionOverrides};
 
 use super::multicall::{self, build_multicall};
-use super::utils::fetch_raw_logs_and_log_meta;
+use super::utils::fetch_raw_logs_and_meta;
 
 impl<M> std::fmt::Display for EthereumMailboxInternal<M>
 where
@@ -170,20 +171,23 @@ where
         &self,
         tx_hash: H512,
     ) -> ChainResult<Vec<(Indexed<HyperlaneMessage>, LogMeta)>> {
-        let logs = fetch_raw_logs_and_log_meta::<DispatchFilter, M>(
-            tx_hash,
-            self.provider.clone(),
-            self.contract.address(),
-        )
-        .await?
-        .into_iter()
-        .map(|(log, log_meta)| {
-            (
-                HyperlaneMessage::from(log.message.to_vec()).into(),
-                log_meta,
-            )
+        let raw_logs_and_meta = call_and_retry_indefinitely(|| {
+            let provider = self.provider.clone();
+            let contract = self.contract.address();
+            Box::pin(async move {
+                fetch_raw_logs_and_meta::<DispatchFilter, M>(tx_hash, provider, contract).await
+            })
         })
-        .collect();
+        .await;
+        let logs = raw_logs_and_meta
+            .into_iter()
+            .map(|(log, log_meta)| {
+                (
+                    HyperlaneMessage::from(log.message.to_vec()).into(),
+                    log_meta,
+                )
+            })
+            .collect();
         Ok(logs)
     }
 }

rust/chains/hyperlane-ethereum/src/contracts/merkle_tree_hook.rs

@@ -6,6 +6,7 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use ethers::prelude::Middleware;
 use hyperlane_core::accumulator::incremental::IncrementalMerkle;
+use hyperlane_core::rpc_clients::call_and_retry_indefinitely;
 use tracing::instrument;
 
 use hyperlane_core::{
@@ -20,7 +21,7 @@ use crate::interfaces::merkle_tree_hook::{
 use crate::tx::call_with_lag;
 use crate::{BuildableWithProvider, ConnectionConf, EthereumProvider};
 
-use super::utils::fetch_raw_logs_and_log_meta;
+use super::utils::fetch_raw_logs_and_meta;
 
 // We don't need the reverse of this impl, so it's ok to disable the clippy lint
 #[allow(clippy::from_over_into)]
@@ -153,20 +154,24 @@ where
         &self,
         tx_hash: H512,
     ) -> ChainResult<Vec<(Indexed<MerkleTreeInsertion>, LogMeta)>> {
-        let logs = fetch_raw_logs_and_log_meta::<InsertedIntoTreeFilter, M>(
-            tx_hash,
-            self.provider.clone(),
-            self.contract.address(),
-        )
-        .await?
-        .into_iter()
-        .map(|(log, log_meta)| {
-            (
-                MerkleTreeInsertion::new(log.index, H256::from(log.message_id)).into(),
-                log_meta,
-            )
+        let raw_logs_and_meta = call_and_retry_indefinitely(|| {
+            let provider = self.provider.clone();
+            let contract = self.contract.address();
+            Box::pin(async move {
+                fetch_raw_logs_and_meta::<InsertedIntoTreeFilter, M>(tx_hash, provider, contract)
+                    .await
+            })
         })
-        .collect();
+        .await;
+        let logs = raw_logs_and_meta
+            .into_iter()
+            .map(|(log, log_meta)| {
+                (
+                    MerkleTreeInsertion::new(log.index, H256::from(log.message_id)).into(),
+                    log_meta,
+                )
+            })
+            .collect();
         Ok(logs)
     }
 }

rust/chains/hyperlane-ethereum/src/contracts/utils.rs

@@ -7,9 +7,8 @@ use ethers::{
 };
 use ethers_contract::{ContractError, EthEvent, LogMeta as EthersLogMeta};
 use hyperlane_core::{ChainResult, LogMeta, H512};
-use tracing::warn;
 
-pub async fn fetch_raw_logs_and_log_meta<T: EthEvent, M>(
+pub async fn fetch_raw_logs_and_meta<T: EthEvent, M>(
     tx_hash: H512,
     provider: Arc<M>,
     contract_address: EthersH160,
@@ -23,8 +22,7 @@ where
         .await
         .map_err(|err| ContractError::<M>::MiddlewareError(err))?;
     let Some(receipt) = receipt else {
-        warn!(%tx_hash, "No receipt found for tx hash");
-        return Ok(vec![]);
+        return Err(eyre::eyre!("No receipt found for tx hash {:?}", tx_hash).into());
     };
 
     let logs: Vec<(T, LogMeta)> = receipt