feat: query evm balance, clean up

rust/agents/relayer/src/relayer.rs

@@ -9,9 +9,11 @@ use derive_more::AsRef;
 use eyre::Result;
 use hyperlane_base::{
     db::{HyperlaneRocksDB, DB},
-    metrics::{AgentMetrics, InstrumentedFallibleTask, Metrics},
-    run_all, BaseAgent, ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore,
-    SequencedDataContractSync, WatermarkContractSync,
+    metrics::{AgentMetrics, Metrics},
+    run_all,
+    settings::ChainConf,
+    BaseAgent, ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore, SequencedDataContractSync,
+    WatermarkContractSync,
 };
 use hyperlane_core::{
     metrics::agent::METRICS_SCRAPE_INTERVAL, HyperlaneDomain, HyperlaneMessage,
@@ -51,7 +53,7 @@ struct ContextKey {
 #[derive(AsRef)]
 pub struct Relayer {
     origin_chains: HashSet<HyperlaneDomain>,
-    destination_chains: HashSet<HyperlaneDomain>,
+    destination_chains: HashMap<HyperlaneDomain, ChainConf>,
     #[as_ref]
     core: HyperlaneAgentCore,
     message_syncs: HashMap<HyperlaneDomain, Arc<SequencedDataContractSync<HyperlaneMessage>>>,
@@ -69,6 +71,8 @@ pub struct Relayer {
     transaction_gas_limit: Option<U256>,
     skip_transaction_gas_limit_for: HashSet<u32>,
     allow_local_checkpoint_syncers: bool,
+    core_metrics: Arc<CoreMetrics>,
+    agent_metrics: Metrics,
 }
 
 impl Debug for Relayer {
@@ -98,7 +102,7 @@ impl BaseAgent for Relayer {
         settings: Self::Settings,
         core_metrics: Arc<CoreMetrics>,
         agent_metrics: Metrics,
-    ) -> Result<(Self, Vec<InstrumentedFallibleTask<()>>)>
+    ) -> Result<Self>
     where
         Self: Sized,
     {
@@ -194,30 +198,10 @@ impl BaseAgent for Relayer {
             .collect();
 
         let mut msg_ctxs = HashMap::new();
-        let mut metrics_fetchers = vec![];
+        let mut destination_chains = HashMap::new();
         for destination in &settings.destination_chains {
             let destination_chain_setup = core.settings.chain_setup(destination).unwrap().clone();
-            let agent_metrics_conf = destination_chain_setup
-                .agent_metrics_conf(Self::AGENT_NAME.to_string())
-                .await?;
-            let agent_metrics_fetcher = destination_chain_setup
-                .build_agent_metrics_fetcher()
-                .await?;
-            let agent_metrics = AgentMetrics::new(
-                agent_metrics.clone(),
-                agent_metrics_conf,
-                agent_metrics_fetcher,
-            );
-
-            let fetcher_task = tokio::spawn(async move {
-                agent_metrics
-                    .start_updating_on_interval(METRICS_SCRAPE_INTERVAL)
-                    .await;
-                Ok(())
-            })
-            .instrument(info_span!("AgentMetricsFetcher"));
-            metrics_fetchers.push(fetcher_task);
-
+            destination_chains.insert(destination.clone(), destination_chain_setup.clone());
             let transaction_gas_limit: Option<U256> =
                 if skip_transaction_gas_limit_for.contains(&destination.id()) {
                     None
@@ -257,7 +241,7 @@ impl BaseAgent for Relayer {
         let relayer = Self {
             dbs,
             origin_chains: settings.origin_chains,
-            destination_chains: settings.destination_chains,
+            destination_chains,
             msg_ctxs,
             core,
             message_syncs,
@@ -269,28 +253,48 @@ impl BaseAgent for Relayer {
             transaction_gas_limit,
             skip_transaction_gas_limit_for,
             allow_local_checkpoint_syncers: settings.allow_local_checkpoint_syncers,
+            core_metrics,
+            agent_metrics,
         };
 
-        Ok((relayer, metrics_fetchers))
+        Ok(relayer)
     }
 
     #[allow(clippy::async_yields_async)]
-    async fn run(
-        self,
-        metrics_fetchers: Vec<InstrumentedFallibleTask<()>>,
-    ) -> Instrumented<JoinHandle<Result<()>>> {
-        // The tasks vec is initialized with the metrics fetcher tasks,
-        // and is then extended with the rest of the tasks.
-        let mut tasks = metrics_fetchers;
+    async fn run(self) -> Instrumented<JoinHandle<Result<()>>> {
+        let mut tasks = vec![];
 
         // send channels by destination chain
         let mut send_channels = HashMap::with_capacity(self.destination_chains.len());
-        for destination in &self.destination_chains {
+        for (dest_domain, dest_conf) in &self.destination_chains {
             let (send_channel, receive_channel) =
                 mpsc::unbounded_channel::<Box<DynPendingOperation>>();
-            send_channels.insert(destination.id(), send_channel);
+            send_channels.insert(dest_domain.id(), send_channel);
+
+            tasks.push(self.run_destination_submitter(dest_domain, receive_channel));
 
-            tasks.push(self.run_destination_submitter(destination, receive_channel));
+            let agent_metrics_conf = dest_conf
+                .agent_metrics_conf(Self::AGENT_NAME.to_string())
+                .await
+                .unwrap();
+            let agent_metrics_fetcher = dest_conf
+                .build_agent_metrics_fetcher(&self.core_metrics)
+                .await
+                .unwrap();
+            let agent_metrics = AgentMetrics::new(
+                self.agent_metrics.clone(),
+                agent_metrics_conf,
+                agent_metrics_fetcher,
+            );
+
+            let fetcher_task = tokio::spawn(async move {
+                agent_metrics
+                    .start_updating_on_interval(METRICS_SCRAPE_INTERVAL)
+                    .await;
+                Ok(())
+            })
+            .instrument(info_span!("AgentMetricsFetcher"));
+            tasks.push(fetcher_task);
         }
 
         for origin in &self.origin_chains {
@@ -364,11 +368,11 @@ impl Relayer {
         let metrics = MessageProcessorMetrics::new(
             &self.core.metrics,
             origin,
-            self.destination_chains.iter(),
+            self.destination_chains.keys(),
         );
         let destination_ctxs = self
             .destination_chains
-            .iter()
+            .keys()
             .filter(|&destination| destination != origin)
             .map(|destination| {
                 (

rust/agents/scraper/src/agent.rs

@@ -3,10 +3,8 @@ use std::{collections::HashMap, sync::Arc};
 use async_trait::async_trait;
 use derive_more::AsRef;
 use hyperlane_base::{
-    metrics::{InstrumentedFallibleTask, Metrics as AgentMetrics},
-    run_all,
-    settings::IndexSettings,
-    BaseAgent, ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore,
+    metrics::Metrics as AgentMetrics, run_all, settings::IndexSettings, BaseAgent,
+    ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore,
 };
 use hyperlane_core::HyperlaneDomain;
 use tokio::task::JoinHandle;
@@ -41,7 +39,7 @@ impl BaseAgent for Scraper {
         settings: Self::Settings,
         metrics: Arc<CoreMetrics>,
         _agent_metrics: AgentMetrics,
-    ) -> eyre::Result<(Self, Vec<InstrumentedFallibleTask<()>>)>
+    ) -> eyre::Result<Self>
     where
         Self: Sized,
     {
@@ -76,29 +74,21 @@ impl BaseAgent for Scraper {
 
         trace!(domain_count = scrapers.len(), "Created scrapers");
 
-        Ok((
-            Self {
-                core,
-                metrics,
-                contract_sync_metrics,
-                scrapers,
-            },
-            Default::default(),
-        ))
+        Ok(Self {
+            core,
+            metrics,
+            contract_sync_metrics,
+            scrapers,
+        })
     }
 
     /// Run the scraper
     ///
     /// * `metrics_fetchers` - A list of metrics fetchers to run. Currently this
     /// only comprise
     #[allow(clippy::async_yields_async)]
-    async fn run(
-        self,
-        metrics_fetchers: Vec<InstrumentedFallibleTask<()>>,
-    ) -> Instrumented<JoinHandle<eyre::Result<()>>> {
-        // The tasks vec is initialized with the metrics fetcher tasks,
-        // and is then extended with the rest of the tasks.
-        let mut tasks = metrics_fetchers;
+    async fn run(self) -> Instrumented<JoinHandle<eyre::Result<()>>> {
+        let mut tasks = Vec::with_capacity(self.scrapers.len());
         for domain in self.scrapers.keys() {
             tasks.push(self.scrape(*domain).await);
         }

rust/agents/validator/src/validator.rs

@@ -10,7 +10,7 @@ use tracing::{error, info, info_span, instrument::Instrumented, warn, Instrument
 
 use hyperlane_base::{
     db::{HyperlaneRocksDB, DB},
-    metrics::{InstrumentedFallibleTask, Metrics as AgentMetrics},
+    metrics::Metrics as AgentMetrics,
     run_all, BaseAgent, CheckpointSyncer, ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore,
     SequencedDataContractSync,
 };
@@ -56,7 +56,7 @@ impl BaseAgent for Validator {
         settings: Self::Settings,
         metrics: Arc<CoreMetrics>,
         _agent_metrics: AgentMetrics,
-    ) -> Result<(Self, Vec<InstrumentedFallibleTask<()>>)>
+    ) -> Result<Self>
     where
         Self: Sized,
     {
@@ -108,17 +108,12 @@ impl BaseAgent for Validator {
             checkpoint_syncer,
         };
 
-        Ok((validator, Default::default()))
+        Ok(validator)
     }
 
     #[allow(clippy::async_yields_async)]
-    async fn run(
-        mut self,
-        metrics_fetchers: Vec<InstrumentedFallibleTask<()>>,
-    ) -> Instrumented<JoinHandle<Result<()>>> {
-        // The tasks vec is initialized with the metrics fetcher tasks,
-        // and is then extended with the rest of the tasks.
-        let mut tasks = metrics_fetchers;
+    async fn run(mut self) -> Instrumented<JoinHandle<Result<()>>> {
+        let mut tasks = vec![];
 
         if let Some(signer_instance) = self.signer_instance.take() {
             tasks.push(

rust/chains/hyperlane-cosmos/src/aggregation_ism.rs

@@ -55,7 +55,7 @@ impl HyperlaneChain for CosmosAggregationIsm {
     }
 
     fn provider(&self) -> Box<dyn HyperlaneProvider> {
-        Box::new(self.provider.clone())
+        self.provider.clone()
     }
 }
 

rust/chains/hyperlane-cosmos/src/lib.rs

@@ -5,7 +5,6 @@
 // TODO: Remove once we start filling things in
 #![allow(unused_variables)]
 
-mod agent_metrics;
 mod aggregation_ism;
 mod error;
 mod interchain_gas;
@@ -24,8 +23,7 @@ mod utils;
 mod validator_announce;
 
 pub use self::{
-    agent_metrics::*, aggregation_ism::*, error::*, interchain_gas::*,
-    interchain_security_module::*, libs::*, mailbox::*, merkle_tree_hook::*, multisig_ism::*,
-    providers::*, routing_ism::*, signers::*, trait_builder::*, trait_builder::*,
-    validator_announce::*, validator_announce::*,
+    aggregation_ism::*, error::*, interchain_gas::*, interchain_security_module::*, libs::*,
+    mailbox::*, merkle_tree_hook::*, multisig_ism::*, providers::*, routing_ism::*, signers::*,
+    trait_builder::*, trait_builder::*, validator_announce::*, validator_announce::*,
 };

rust/chains/hyperlane-cosmos/src/providers/mod.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use hyperlane_core::{
-    BlockInfo, ChainResult, ContractLocator, HyperlaneChain, HyperlaneDomain, HyperlaneProvider,
-    TxnInfo, H256, U256,
+    metrics::agent::AgentMetricsFetcher, BlockInfo, ChainResult, ContractLocator, HyperlaneChain,
+    HyperlaneDomain, HyperlaneProvider, TxnInfo, H256, U256,
 };
 use tendermint_rpc::{client::CompatMode, HttpClient};
 
@@ -71,6 +71,17 @@ impl HyperlaneChain for CosmosProvider {
     }
 }
 
+#[async_trait]
+impl AgentMetricsFetcher for CosmosProvider {
+    async fn get_balance(&self, address: String) -> ChainResult<U256> {
+        Ok(self
+            .grpc_client
+            .get_balance(address, self.canonical_asset.clone())
+            .await?
+            .into())
+    }
+}
+
 #[async_trait]
 impl HyperlaneProvider for CosmosProvider {
     async fn get_block_by_hash(&self, _hash: &H256) -> ChainResult<BlockInfo> {
@@ -85,12 +96,4 @@ impl HyperlaneProvider for CosmosProvider {
         // FIXME
         Ok(true)
     }
-
-    async fn get_balance(&self, address: String) -> ChainResult<U256> {
-        Ok(self
-            .grpc_client
-            .get_balance(address, self.canonical_asset.clone())
-            .await?
-            .into())
-    }
 }