Remove chain runtime

.changelog/unreleased/improvements/3341-chain-runtime.md

@@ -0,0 +1,3 @@
+- Emit (g)RPC queries concurrently for increased throughput when
+  relaying on multiple chains/channels with a single Hermes instance
+  ([\#3341](https://github.com/informalsystems/hermes/issues/3341))

crates/relayer/Cargo.toml

@@ -69,6 +69,7 @@ generic-array = "0.14.7"
 secp256k1 = { version = "0.27.0", features = ["rand-std"] }
 strum = { version = "0.24.1", features = ["derive"] }
 once_cell = "1.17.1"
+parking_lot = "0.12.1"
 
 [dependencies.byte-unit]
 version = "4.0.19"

crates/relayer/src/chain.rs

@@ -4,7 +4,6 @@ pub mod counterparty;
 pub mod endpoint;
 pub mod handle;
 pub mod requests;
-pub mod runtime;
 pub mod tracking;
 
 use serde::{de::Error, Deserialize, Serialize};

crates/relayer/src/chain/cosmos.rs

@@ -10,9 +10,10 @@ use futures::future::join_all;
 use num_bigint::BigInt;
 use std::{cmp::Ordering, thread};
 
+use parking_lot::{Mutex, RwLock};
 use tokio::runtime::Runtime as TokioRuntime;
-use tonic::codegen::http::Uri;
-use tonic::metadata::AsciiMetadataValue;
+use tokio::sync::RwLock as AsyncRwLock;
+use tonic::{codegen::http::Uri, metadata::AsciiMetadataValue};
 use tracing::{error, instrument, trace, warn};
 
 use ibc_proto::cosmos::{
@@ -144,12 +145,12 @@ pub struct CosmosSdkChain {
     grpc_addr: Uri,
     light_client: TmLightClient,
     rt: Arc<TokioRuntime>,
-    keybase: KeyRing<Secp256k1KeyPair>,
+    keybase: RwLock<KeyRing<Secp256k1KeyPair>>,
 
     /// A cached copy of the account information
-    account: Option<Account>,
+    account: AsyncRwLock<Option<Account>>,
 
-    tx_monitor_cmd: Option<TxMonitorCmd>,
+    tx_monitor_cmd: Mutex<Option<TxMonitorCmd>>,
 }
 
 impl CosmosSdkChain {
@@ -164,7 +165,8 @@ impl CosmosSdkChain {
     }
 
     fn key(&self) -> Result<Secp256k1KeyPair, Error> {
-        self.keybase()
+        self.keybase
+            .read()
             .get_key(&self.config.key_name)
             .map_err(Error::key_base)
     }
@@ -286,7 +288,7 @@ impl CosmosSdkChain {
         Ok(())
     }
 
-    fn init_event_monitor(&mut self) -> Result<TxMonitorCmd, Error> {
+    fn init_event_monitor(&self) -> Result<TxMonitorCmd, Error> {
         crate::time!(
             "init_event_monitor",
             {
@@ -641,7 +643,7 @@ impl CosmosSdkChain {
         ),
     )]
     async fn do_send_messages_and_wait_commit(
-        &mut self,
+        &self,
         tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<IbcEventWithHeight>, Error> {
         crate::time!(
@@ -656,8 +658,10 @@ impl CosmosSdkChain {
         let key_pair = self.key()?;
         let key_account = key_pair.account();
 
-        let account =
-            get_or_fetch_account(&self.grpc_addr, &key_account, &mut self.account).await?;
+        // Take a write lock on the current account,
+        // excluding anybody from sending txs until we are done.
+        let mut opt_account = self.account.write().await;
+        let account = get_or_fetch_account(&self.grpc_addr, &key_account, &mut opt_account).await?;
 
         if self.config.sequential_batch_tx {
             sequential_send_batched_messages_and_wait_commit(
@@ -692,7 +696,7 @@ impl CosmosSdkChain {
         ),
     )]
     async fn do_send_messages_and_wait_check_tx(
-        &mut self,
+        &self,
         tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<Response>, Error> {
         crate::time!(
@@ -707,8 +711,10 @@ impl CosmosSdkChain {
         let key_pair = self.key()?;
         let key_account = key_pair.account();
 
-        let account =
-            get_or_fetch_account(&self.grpc_addr, &key_account, &mut self.account).await?;
+        // Take a write lock on the current account,
+        // excluding anybody from sending txs until we are done.
+        let mut opt_account = self.account.write().await;
+        let account = get_or_fetch_account(&self.grpc_addr, &key_account, &mut opt_account).await?;
 
         send_batched_messages_and_wait_check_tx(
             &self.rpc_client,
@@ -721,6 +727,33 @@ impl CosmosSdkChain {
         .await
     }
 
+    async fn do_maybe_register_counterparty_payee(
+        &self,
+        channel_id: &ChannelId,
+        port_id: &PortId,
+        counterparty_payee: &Signer,
+    ) -> Result<(), Error> {
+        let address = self.get_signer()?;
+        let key_pair = self.key()?;
+
+        // Take a write lock on the current account,
+        // excluding anybody from sending txs until we are done.
+        let mut opt_account = self.account.write().await;
+
+        maybe_register_counterparty_payee(
+            &self.rpc_client,
+            &self.tx_config,
+            &key_pair,
+            &mut opt_account,
+            &self.config.memo_prefix,
+            channel_id,
+            port_id,
+            &address,
+            counterparty_payee,
+        )
+        .await
+    }
+
     fn query_packet_from_block(
         &self,
         request: &QueryPacketEventDataRequest,
@@ -875,38 +908,54 @@ impl ChainEndpoint for CosmosSdkChain {
             grpc_addr,
             light_client,
             rt,
-            keybase,
             tx_config,
-            account: None,
-            tx_monitor_cmd: None,
+            keybase: RwLock::new(keybase),
+            account: AsyncRwLock::new(None),
+            tx_monitor_cmd: Mutex::new(None),
         };
 
         Ok(chain)
     }
 
-    fn shutdown(self) -> Result<(), Error> {
-        if let Some(monitor_tx) = self.tx_monitor_cmd {
+    fn shutdown(&self) -> Result<(), Error> {
+        let cmd = self.tx_monitor_cmd.lock();
+
+        if let Some(monitor_tx) = cmd.as_ref() {
             monitor_tx.shutdown().map_err(Error::event_monitor)?;
         }
 
         Ok(())
     }
 
-    fn keybase(&self) -> &KeyRing<Self::SigningKeyPair> {
-        &self.keybase
+    fn get_key(&self) -> Result<Self::SigningKeyPair, Error> {
+        // Get the key from key seed file
+        let key_pair = self
+            .keybase
+            .read()
+            .get_key(&self.config().key_name)
+            .map_err(|e| Error::key_not_found(self.config().key_name.clone(), e))?;
+
+        Ok(key_pair)
     }
 
-    fn keybase_mut(&mut self) -> &mut KeyRing<Self::SigningKeyPair> {
-        &mut self.keybase
+    fn add_key(&self, key_name: &str, key_pair: Self::SigningKeyPair) -> Result<(), Error> {
+        self.keybase
+            .write()
+            .add_key(key_name, key_pair)
+            .map_err(Error::key_base)?;
+
+        Ok(())
     }
 
-    fn subscribe(&mut self) -> Result<Subscription, Error> {
-        let tx_monitor_cmd = match &self.tx_monitor_cmd {
+    fn subscribe(&self) -> Result<Subscription, Error> {
+        let mut cmd = self.tx_monitor_cmd.lock();
+
+        let tx_monitor_cmd = match cmd.as_ref() {
             Some(tx_monitor_cmd) => tx_monitor_cmd,
             None => {
                 let tx_monitor_cmd = self.init_event_monitor()?;
-                self.tx_monitor_cmd = Some(tx_monitor_cmd);
-                self.tx_monitor_cmd.as_ref().unwrap()
+                *cmd = Some(tx_monitor_cmd);
+                cmd.as_ref().unwrap()
             }
         };
 
@@ -947,7 +996,7 @@ impl ChainEndpoint for CosmosSdkChain {
 
     /// Fetch a header from the chain at the given height and verify it.
     fn verify_header(
-        &mut self,
+        &self,
         trusted: ICSHeight,
         target: ICSHeight,
         client_state: &AnyClientState,
@@ -967,7 +1016,7 @@ impl ChainEndpoint for CosmosSdkChain {
     /// Given a client update event that includes the header used in a client update,
     /// look for misbehaviour by fetching a header at same or latest height.
     fn check_misbehaviour(
-        &mut self,
+        &self,
         update: &UpdateClient,
         client_state: &AnyClientState,
     ) -> Result<Option<MisbehaviourEvidence>, Error> {
@@ -992,7 +1041,7 @@ impl ChainEndpoint for CosmosSdkChain {
     /// TODO - more work is required here for a smarter split maybe iteratively accumulating/ evaluating
     /// msgs in a Tx until any of the max size, max num msgs, max fee are exceeded.
     fn send_messages_and_wait_commit(
-        &mut self,
+        &self,
         tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<IbcEventWithHeight>, Error> {
         let runtime = self.rt.clone();
@@ -1001,7 +1050,7 @@ impl ChainEndpoint for CosmosSdkChain {
     }
 
     fn send_messages_and_wait_check_tx(
-        &mut self,
+        &self,
         tracked_msgs: TrackedMsgs,
     ) -> Result<Vec<Response>, Error> {
         let runtime = self.rt.clone();
@@ -1033,9 +1082,14 @@ impl ChainEndpoint for CosmosSdkChain {
         // If a key_name is given, extract the account hash.
         // Else retrieve the account from the configuration file.
         let key = match key_name {
-            Some(key_name) => self.keybase().get_key(key_name).map_err(Error::key_base)?,
             None => self.key()?,
+            Some(key_name) => self
+                .keybase
+                .read()
+                .get_key(key_name)
+                .map_err(Error::key_base)?,
         };
+
         let account = key.account();
 
         let denom = denom.unwrap_or(&self.config.gas_price.denom);
@@ -1048,9 +1102,14 @@ impl ChainEndpoint for CosmosSdkChain {
         // If a key_name is given, extract the account hash.
         // Else retrieve the account from the configuration file.
         let key = match key_name {
-            Some(key_name) => self.keybase().get_key(key_name).map_err(Error::key_base)?,
             None => self.key()?,
+            Some(key_name) => self
+                .keybase
+                .read()
+                .get_key(key_name)
+                .map_err(Error::key_base)?,
         };
+
         let account = key.account();
 
         let balance = self.block_on(query_all_balances(&self.grpc_addr, &account))?;
@@ -2122,7 +2181,7 @@ impl ChainEndpoint for CosmosSdkChain {
     }
 
     fn build_header(
-        &mut self,
+        &self,
         trusted_height: ICSHeight,
         target_height: ICSHeight,
         client_state: &AnyClientState,
@@ -2145,23 +2204,14 @@ impl ChainEndpoint for CosmosSdkChain {
     }
 
     fn maybe_register_counterparty_payee(
-        &mut self,
+        &self,
         channel_id: &ChannelId,
         port_id: &PortId,
         counterparty_payee: &Signer,
     ) -> Result<(), Error> {
-        let address = self.get_signer()?;
-        let key_pair = self.key()?;
-
-        self.rt.block_on(maybe_register_counterparty_payee(
-            &self.rpc_client,
-            &self.tx_config,
-            &key_pair,
-            &mut self.account,
-            &self.config.memo_prefix,
+        self.rt.block_on(self.do_maybe_register_counterparty_payee(
             channel_id,
             port_id,
-            &address,
             counterparty_payee,
         ))
     }

crates/relayer/src/chain/cosmos/query/consensus_state.rs

@@ -88,17 +88,18 @@ pub async fn query_consensus_state_heights(
 }
 
 pub async fn query_consensus_states(
-    chain_id: &ChainId,
+    _chain_id: &ChainId,
     grpc_addr: &Uri,
     request: QueryConsensusStatesRequest,
 ) -> Result<Vec<AnyConsensusStateWithHeight>, Error> {
+    crate::telemetry!(query, _chain_id, "query_consensus_states");
+
     crate::time!(
         "query_consensus_states",
         {
-            "src_chain": chain_id,
+            "src_chain": _chain_id,
         }
     );
-    crate::telemetry!(query, chain_id, "query_consensus_states");
 
     let mut client =
         ibc_proto::ibc::core::client::v1::query_client::QueryClient::connect(grpc_addr.clone())