fix: deduplicate rpc connections

Author: troykessler

rust/main/agents/relayer/src/relayer.rs

@@ -641,25 +641,11 @@ impl Relayer {
                 window_secs, "Initialized relay API rate limiter"
             );
 
-            let mut indexers: HashMap<String, Arc<dyn Indexer<HyperlaneMessage>>> = HashMap::new();
-            info!(
-                origin_count = self.origins.len(),
-                "Setting up relay API indexers"
-            );
-            for (domain, origin) in &self.origins {
-                match origin
-                    .chain_conf
-                    .build_message_indexer(&self.core_metrics, false)
-                    .await
-                {
-                    Ok(indexer) => {
-                        indexers.insert(domain.name().to_string(), Arc::new(indexer));
-                    }
-                    Err(e) => {
-                        error!(domain = %domain.name(), error = ?e, "Failed to create mailbox indexer for relay API");
-                    }
-                }
-            }
+            let indexers: HashMap<String, Arc<dyn Indexer<HyperlaneMessage>>> = self
+                .origins
+                .iter()
+                .map(|(domain, origin)| (domain.name().to_string(), origin.message_indexer.clone()))
+                .collect();
 
             Some(
                 RelayApiState::new(

rust/main/agents/relayer/src/relayer/origin.rs

@@ -11,7 +11,8 @@ use hyperlane_base::{
 };
 use hyperlane_core::{
     HyperlaneDomain, HyperlaneLogStore, HyperlaneMessage, HyperlaneSequenceAwareIndexerStoreReader,
-    HyperlaneWatermarkedLogStore, InterchainGasPayment, MerkleTreeInsertion, ValidatorAnnounce,
+    HyperlaneWatermarkedLogStore, Indexer, InterchainGasPayment, MerkleTreeInsertion,
+    ValidatorAnnounce,
 };
 use tokio::sync::RwLock;
 
@@ -31,6 +32,9 @@ pub struct Origin {
     pub gas_payment_enforcer: Arc<RwLock<GasPaymentEnforcer>>,
     pub prover_sync: Arc<RwLock<MerkleTreeBuilder>>,
     pub message_sync: MessageSync,
+    /// The underlying mailbox indexer shared with `message_sync`. Exposed so the relay
+    /// API can reuse the existing RPC connection instead of opening a new one.
+    pub message_indexer: Arc<dyn Indexer<HyperlaneMessage>>,
     pub interchain_gas_payment_sync: Option<InterchainGasPaymentSync>,
     pub merkle_tree_hook_sync: MerkleTreeHookSync,
 }
@@ -128,7 +132,7 @@ impl Factory for OriginFactory {
         };
 
         let hyperlane_db = Arc::new(db.clone());
-        let message_sync = {
+        let (message_sync, message_indexer) = {
             let start_entity_init = Instant::now();
             let res = self
                 .init_message_sync(&domain, chain_conf, hyperlane_db.clone())
@@ -176,6 +180,7 @@ impl Factory for OriginFactory {
             gas_payment_enforcer: Arc::new(RwLock::new(gas_payment_enforcer)),
             prover_sync: Arc::new(RwLock::new(prover_sync)),
             message_sync,
+            message_indexer,
             interchain_gas_payment_sync,
             merkle_tree_hook_sync,
         };
@@ -222,8 +227,9 @@ impl OriginFactory {
         domain: &HyperlaneDomain,
         chain_conf: &ChainConf,
         db: Arc<HyperlaneRocksDB>,
-    ) -> Result<MessageSync, FactoryError> {
-        match HyperlaneMessage::indexing_cursor(domain.domain_protocol()) {
+    ) -> Result<(MessageSync, Arc<dyn Indexer<HyperlaneMessage>>), FactoryError> {
+        let (sync, seq_indexer) = match HyperlaneMessage::indexing_cursor(domain.domain_protocol())
+        {
             CursorType::SequenceAware => Self::build_sequenced_contract_sync(
                 domain,
                 chain_conf,
@@ -234,7 +240,7 @@ impl OriginFactory {
                 self.tx_id_indexing_enabled,
             )
             .await
-            .map(|r| r as Arc<dyn ContractSyncer<_>>)
+            .map(|(r, i)| (r as Arc<dyn ContractSyncer<_>>, i))
             .map_err(|err| FactoryError::MessageSync(domain.to_string(), err.to_string())),
             CursorType::RateLimited => Self::build_watermark_contract_sync(
                 domain,
@@ -246,9 +252,13 @@ impl OriginFactory {
                 self.tx_id_indexing_enabled,
             )
             .await
-            .map(|r| r as Arc<dyn ContractSyncer<_>>)
+            .map(|(r, i)| (r as Arc<dyn ContractSyncer<_>>, i))
             .map_err(|err| FactoryError::MessageSync(domain.to_string(), err.to_string())),
-        }
+        }?;
+        // Arc<dyn SequenceAwareIndexer<T>> implements Indexer<T> via auto_impl(Arc);
+        // wrap in an outer Arc to satisfy Arc<dyn Indexer<T>>.
+        let indexer: Arc<dyn Indexer<HyperlaneMessage>> = Arc::new(seq_indexer);
+        Ok((sync, indexer))
     }
 
     async fn init_igp_sync(
@@ -268,7 +278,7 @@ impl OriginFactory {
                 false,
             )
             .await
-            .map(|r| r as Arc<dyn ContractSyncer<_>>)
+            .map(|(r, _)| r as Arc<dyn ContractSyncer<_>>)
             .map_err(|err| {
                 FactoryError::InterchainGasPaymentSync(domain.to_string(), err.to_string())
             }),
@@ -282,7 +292,7 @@ impl OriginFactory {
                 false,
             )
             .await
-            .map(|r| r as Arc<dyn ContractSyncer<_>>)
+            .map(|(r, _)| r as Arc<dyn ContractSyncer<_>>)
             .map_err(|err| {
                 FactoryError::InterchainGasPaymentSync(domain.to_string(), err.to_string())
             }),
@@ -306,7 +316,7 @@ impl OriginFactory {
                 false,
             )
             .await
-            .map(|r| r as Arc<dyn ContractSyncer<_>>)
+            .map(|(r, _)| r as Arc<dyn ContractSyncer<_>>)
             .map_err(|err| FactoryError::MerkleTreeHookSync(domain.to_string(), err.to_string())),
             CursorType::RateLimited => Self::build_watermark_contract_sync(
                 domain,
@@ -318,7 +328,7 @@ impl OriginFactory {
                 false,
             )
             .await
-            .map(|r| r as Arc<dyn ContractSyncer<_>>)
+            .map(|(r, _)| r as Arc<dyn ContractSyncer<_>>)
             .map_err(|err| FactoryError::MerkleTreeHookSync(domain.to_string(), err.to_string())),
         }
     }
@@ -331,7 +341,7 @@ impl OriginFactory {
         store: Arc<S>,
         advanced_log_meta: bool,
         broadcast_sender_enabled: bool,
-    ) -> eyre::Result<Arc<SequencedDataContractSync<T>>>
+    ) -> eyre::Result<(Arc<SequencedDataContractSync<T>>, SequenceIndexer<T>)>
     where
         T: Indexable + Debug,
         SequenceIndexer<T>: TryFromWithMetrics<ChainConf>,
@@ -341,13 +351,15 @@ impl OriginFactory {
         let indexer =
             SequenceIndexer::<T>::try_from_with_metrics(chain_conf, metrics, advanced_log_meta)
                 .await?;
-        Ok(Arc::new(ContractSync::new(
+        let indexer_clone = indexer.clone();
+        let sync = Arc::new(ContractSync::new(
             domain.clone(),
             store.clone() as SequenceAwareLogStore<_>,
             indexer,
             sync_metrics.clone(),
             broadcast_sender_enabled,
-        )))
+        ));
+        Ok((sync, indexer_clone))
     }
 
     async fn build_watermark_contract_sync<T, S>(
@@ -358,7 +370,7 @@ impl OriginFactory {
         store: Arc<S>,
         advanced_log_meta: bool,
         broadcast_sender_enabled: bool,
-    ) -> eyre::Result<Arc<WatermarkContractSync<T>>>
+    ) -> eyre::Result<(Arc<WatermarkContractSync<T>>, SequenceIndexer<T>)>
     where
         T: Indexable + Debug,
         SequenceIndexer<T>: TryFromWithMetrics<ChainConf>,
@@ -367,12 +379,14 @@ impl OriginFactory {
         let indexer =
             SequenceIndexer::<T>::try_from_with_metrics(chain_conf, metrics, advanced_log_meta)
                 .await?;
-        Ok(Arc::new(ContractSync::new(
+        let indexer_clone = indexer.clone();
+        let sync = Arc::new(ContractSync::new(
             domain.clone(),
             store.clone() as WatermarkLogStore<_>,
             indexer,
             sync_metrics.clone(),
             broadcast_sender_enabled,
-        )))
+        ));
+        Ok((sync, indexer_clone))
     }
 }