fix identities kusama

Cargo.toml

@@ -36,7 +36,7 @@ mobc-redis = { version = "0.7", default-features = false, features = ["async-std
 ctrlc = { version = "3.0", features = ["termination"] }
 
 # Subxt crates:
-subxt = { version = "0.35.3", features = ["substrate-compat", "native"] }
+subxt = { version = "0.35.3", features = ["substrate-compat", "native", "unstable-reconnecting-rpc-client"] }
 subxt-signer = { version = "0.35.3", features = ["subxt"] }
 sp-core-hashing = "12.0.0"
 #

src/config.rs

@@ -228,6 +228,7 @@ pub struct Config {
     #[serde(default = "default_error_interval")]
     pub error_interval: u64,
     pub substrate_ws_url: String,
+    pub substrate_people_ws_url: String,
     #[serde(default = "default_data_path")]
     pub data_path: String,
     #[serde(default = "default_data_path_read_only")]
@@ -417,9 +418,17 @@ fn get_config() -> Config {
         .long("substrate-ws-url")
         .takes_value(true)
         .help(
-          "Substrate websocket endpoint for which 'onet' will try to connect. (e.g. wss://kusama-rpc.polkadot.io) (NOTE: substrate_ws_url takes precedence than <CHAIN> argument)",
+          "Substrate websocket endpoint for which 'onet' will try to connect. (e.g. wss://rpc.ibp.network:443/kusama) (NOTE: substrate_ws_url takes precedence than <CHAIN> argument)",
         ),
     )
+    .arg(
+        Arg::with_name("substrate-people-ws-url")
+          .long("substrate-people-ws-url")
+          .takes_value(true)
+          .help(
+            "Substrate websocket endpoint for which 'onet' will try to connect and retrieve identities from. (e.g. wss://sys.ibp.network:443/people-kusama))",
+          ),
+      )
     .arg(
       Arg::with_name("maximum-history-eras")
             .long("maximum-history-eras")
@@ -480,19 +489,28 @@ fn get_config() -> Config {
     match matches.value_of("CHAIN") {
         Some("westend") => {
             if env::var("ONET_SUBSTRATE_WS_URL").is_err() {
-                env::set_var("ONET_SUBSTRATE_WS_URL", "wss://westend-rpc.polkadot.io:443");
+                env::set_var("ONET_SUBSTRATE_WS_URL", "wss://rpc.ibp.network:443/westend");
             }
             env::set_var("ONET_CHAIN_NAME", "westend");
         }
         Some("kusama") => {
             if env::var("ONET_SUBSTRATE_WS_URL").is_err() {
-                env::set_var("ONET_SUBSTRATE_WS_URL", "wss://kusama-rpc.polkadot.io:443");
+                env::set_var("ONET_SUBSTRATE_WS_URL", "wss://rpc.ibp.network:443/kusama");
+            }
+            if env::var("ONET_SUBSTRATE_PEOPLE_WS_URL").is_err() {
+                env::set_var(
+                    "ONET_SUBSTRATE_WS_URL",
+                    "wss://sys.ibp.network:443/people-kusama",
+                );
             }
             env::set_var("ONET_CHAIN_NAME", "kusama");
         }
         Some("polkadot") => {
             if env::var("ONET_SUBSTRATE_WS_URL").is_err() {
-                env::set_var("ONET_SUBSTRATE_WS_URL", "wss://rpc.polkadot.io:443");
+                env::set_var(
+                    "ONET_SUBSTRATE_WS_URL",
+                    "wss://rpc.ibp.network:443/polkadot",
+                );
             }
             env::set_var("ONET_CHAIN_NAME", "polkadot");
         }
@@ -511,6 +529,10 @@ fn get_config() -> Config {
         env::set_var("ONET_SUBSTRATE_WS_URL", substrate_ws_url);
     }
 
+    if let Some(substrate_people_ws_url) = matches.value_of("substrate-people-ws-url") {
+        env::set_var("ONET_SUBSTRATE_PEOPLE_WS_URL", substrate_people_ws_url);
+    }
+
     if let Some(maximum_subscribers) = matches.value_of("maximum-subscribers") {
         env::set_var("ONET_MAXIMUM_SUBSCRIBERS", maximum_subscribers);
     }

src/errors.rs

@@ -25,7 +25,7 @@ use derive_more::Display;
 use reqwest;
 use serde::{Deserialize, Serialize};
 use std::{str::Utf8Error, string::String};
-use subxt::error::MetadataError;
+use subxt::{backend::rpc::reconnecting_rpc_client::RpcError, error::MetadataError};
 use thiserror::Error;
 
 /// On specific error messages
@@ -35,6 +35,8 @@ pub enum OnetError {
     CacheError(#[from] CacheError),
     #[error("Subxt error: {0}")]
     SubxtError(#[from] subxt::Error),
+    #[error("Reconnecting error: {0}")]
+    RpcError(#[from] RpcError),
     #[error("Codec error: {0}")]
     CodecError(#[from] codec::Error),
     #[error("Utf8 error: {0}")]

src/onet.rs

@@ -25,7 +25,7 @@ use crate::matrix::{Matrix, UserID, MATRIX_SUBSCRIBERS_FILENAME};
 use crate::records::EpochIndex;
 use crate::report::Network;
 use crate::runtimes::{
-    kusama, 
+    kusama,
     polkadot,
     support::{ChainPrefix, ChainTokenSymbol, SupportedRuntime},
     // westend,
@@ -49,10 +49,13 @@ use std::{
 use subxt::{
     backend::{
         legacy::{rpc_methods::StorageKey, LegacyRpcMethods},
-        rpc::RpcClient,
+        rpc::{
+            reconnecting_rpc_client::{Client as ReconnectingClient, ExponentialBackoff, RpcError},
+            RpcClient,
+        },
     },
     ext::sp_core::crypto,
-    utils::AccountId32,
+    utils::{validate_url_is_secure, AccountId32},
     OnlineClient, PolkadotConfig,
 };
 use subxt_signer::{bip39::Mnemonic, sr25519::Keypair};
@@ -139,39 +142,83 @@ impl std::fmt::Display for ReportType {
     }
 }
 
+// DEPRECATED
 pub async fn _create_substrate_node_client(
     config: Config,
 ) -> Result<OnlineClient<PolkadotConfig>, subxt::Error> {
     OnlineClient::<PolkadotConfig>::from_url(config.substrate_ws_url).await
 }
 
-pub async fn create_substrate_rpc_client_from_config(
+// DEPRECATED
+pub async fn _create_substrate_rpc_client_from_config(
     config: Config,
 ) -> Result<RpcClient, subxt::Error> {
     RpcClient::from_url(config.substrate_ws_url).await
 }
 
+pub async fn create_substrate_rpc_client_from_config(
+    config: Config,
+) -> Result<ReconnectingClient, RpcError> {
+    if let Err(_) = validate_url_is_secure(config.substrate_ws_url.as_ref()) {
+        warn!("Insecure URL provided: {}", config.substrate_ws_url);
+    };
+
+    ReconnectingClient::builder()
+        .retry_policy(
+            ExponentialBackoff::from_millis(100)
+                .max_delay(time::Duration::from_secs(10))
+                .take(10),
+        )
+        .build(config.substrate_ws_url)
+        .await
+}
+
+pub async fn create_substrate_client_from_supported_runtime(
+    runtime: SupportedRuntime,
+) -> Result<Option<OnlineClient<PolkadotConfig>>, OnetError> {
+    if runtime.is_people_runtime_available() {
+        let reconnecting_client = ReconnectingClient::builder()
+            .retry_policy(
+                ExponentialBackoff::from_millis(100)
+                    .max_delay(time::Duration::from_secs(10))
+                    .take(10),
+            )
+            .build(runtime.people_runtime().default_rpc_url())
+            .await
+            .map_err(|err| OnetError::RpcError(err.into()))?;
+
+        let client =
+            create_substrate_client_from_rpc_client(reconnecting_client.clone().into()).await?;
+        Ok(Some(client))
+    } else {
+        Ok(None)
+    }
+}
+
 pub async fn create_substrate_client_from_rpc_client(
     rpc_client: RpcClient,
-) -> Result<OnlineClient<PolkadotConfig>, subxt::Error> {
-    OnlineClient::<PolkadotConfig>::from_rpc_client(rpc_client).await
+) -> Result<OnlineClient<PolkadotConfig>, OnetError> {
+    OnlineClient::<PolkadotConfig>::from_rpc_client(rpc_client)
+        .await
+        .map_err(|err| OnetError::SubxtError(err.into()))
 }
 
 pub async fn create_or_await_substrate_node_client(
     config: Config,
 ) -> (
     OnlineClient<PolkadotConfig>,
     LegacyRpcMethods<PolkadotConfig>,
+    Option<OnlineClient<PolkadotConfig>>,
     SupportedRuntime,
 ) {
     loop {
         match create_substrate_rpc_client_from_config(config.clone()).await {
             Ok(rpc_client) => {
-                let rpc = LegacyRpcMethods::<PolkadotConfig>::new(rpc_client.clone());
-                let chain = rpc.system_chain().await.unwrap_or_default();
-                let name = rpc.system_name().await.unwrap_or_default();
-                let version = rpc.system_version().await.unwrap_or_default();
-                let properties = rpc.system_properties().await.unwrap_or_default();
+                let legacy_rpc = LegacyRpcMethods::<PolkadotConfig>::new(rpc_client.clone().into());
+                let chain = legacy_rpc.system_chain().await.unwrap_or_default();
+                let name = legacy_rpc.system_name().await.unwrap_or_default();
+                let version = legacy_rpc.system_version().await.unwrap_or_default();
+                let properties = legacy_rpc.system_properties().await.unwrap_or_default();
 
                 // Display SS58 addresses based on the connected chain
                 let chain_prefix: ChainPrefix =
@@ -199,9 +246,21 @@ pub async fn create_or_await_substrate_node_client(
                     chain, config.substrate_ws_url, name, version
                 );
 
-                match create_substrate_client_from_rpc_client(rpc_client.clone()).await {
-                    Ok(client) => {
-                        break (client, rpc, SupportedRuntime::from(chain_token_symbol));
+                match create_substrate_client_from_rpc_client(rpc_client.clone().into()).await {
+                    Ok(relay_client) => {
+                        // Create people chain client depending on the runtime selected
+                        let runtime = SupportedRuntime::from(chain_token_symbol);
+                        match create_substrate_client_from_supported_runtime(runtime).await {
+                            Ok(people_client_option) => {
+                                break (relay_client, legacy_rpc, people_client_option, runtime);
+                            }
+
+                            Err(e) => {
+                                error!("{}", e);
+                                info!("Awaiting for connection using {}", config.substrate_ws_url);
+                                thread::sleep(time::Duration::from_secs(6));
+                            }
+                        }
                     }
                     Err(e) => {
                         error!("{}", e);
@@ -223,13 +282,17 @@ pub struct Onet {
     runtime: SupportedRuntime,
     client: OnlineClient<PolkadotConfig>,
     rpc: LegacyRpcMethods<PolkadotConfig>,
+    // Note: Use people client as optional only until we get people chain available
+    // on Polkadot, as soon as it is available it can go away
+    people_client_option: Option<OnlineClient<PolkadotConfig>>,
     matrix: Matrix,
     pub cache: RedisPool,
 }
 
 impl Onet {
     pub async fn new() -> Onet {
-        let (client, rpc, runtime) = create_or_await_substrate_node_client(CONFIG.clone()).await;
+        let (client, rpc, people_client_option, runtime) =
+            create_or_await_substrate_node_client(CONFIG.clone()).await;
 
         // Initialize matrix client
         let mut matrix: Matrix = Matrix::new();
@@ -242,6 +305,7 @@ impl Onet {
             runtime,
             client,
             rpc,
+            people_client_option,
             matrix,
             cache: create_or_await_pool(CONFIG.clone()),
         }
@@ -265,6 +329,10 @@ impl Onet {
         &self.client
     }
 
+    pub fn people_client(&self) -> &Option<OnlineClient<PolkadotConfig>> {
+        &self.people_client_option
+    }
+
     pub fn rpc(&self) -> &LegacyRpcMethods<PolkadotConfig> {
         &self.rpc
     }

src/runtimes/kusama.rs

@@ -79,10 +79,9 @@ mod node_runtime {}
 
 use node_runtime::{
     runtime_types::{
-        bounded_collections::bounded_vec::BoundedVec, pallet_identity::types::Data,
-        pallet_nomination_pools::PoolState, polkadot_parachain_primitives::primitives::Id,
-        polkadot_primitives::v6::DisputeStatement, polkadot_primitives::v6::ValidatorIndex,
-        polkadot_primitives::v6::ValidityAttestation,
+        bounded_collections::bounded_vec::BoundedVec, pallet_nomination_pools::PoolState,
+        polkadot_parachain_primitives::primitives::Id, polkadot_primitives::v6::DisputeStatement,
+        polkadot_primitives::v6::ValidatorIndex, polkadot_primitives::v6::ValidityAttestation,
         polkadot_runtime_parachains::scheduler::common::Assignment,
         polkadot_runtime_parachains::scheduler::pallet::CoreOccupied,
         sp_arithmetic::per_things::Perbill, sp_consensus_babe::digests::PreDigest,
@@ -92,6 +91,14 @@ use node_runtime::{
     system::events::ExtrinsicFailed,
 };
 
+#[subxt::subxt(
+    runtime_metadata_path = "metadata/people_kusama_metadata.scale",
+    derive_for_all_types = "PartialEq, Clone"
+)]
+mod people_node_runtime {}
+
+use people_node_runtime::runtime_types::pallet_identity::types::Data;
+
 type Call = node_runtime::runtime_types::staging_kusama_runtime::RuntimeCall;
 type NominationPoolsCall = node_runtime::runtime_types::pallet_nomination_pools::pallet::Call;
 
@@ -2981,9 +2988,13 @@ async fn get_identity(
     stash: &AccountId32,
     sub_account_name: Option<String>,
 ) -> Result<Option<Identity>, OnetError> {
-    let api = onet.client().clone();
+    let api = if let Some(client) = onet.people_client() {
+        client.clone()
+    } else {
+        onet.client().clone()
+    };
 
-    let identity_of_addr = node_runtime::storage().identity().identity_of(stash);
+    let identity_of_addr = people_node_runtime::storage().identity().identity_of(stash);
     match api
         .storage()
         .at_latest()
@@ -3001,7 +3012,7 @@ async fn get_identity(
             Ok(Some(identity))
         }
         None => {
-            let super_of_addr = node_runtime::storage().identity().super_of(stash);
+            let super_of_addr = people_node_runtime::storage().identity().super_of(stash);
             if let Some((parent_account, data)) = api
                 .storage()
                 .at_latest()

src/runtimes/support.rs

@@ -19,6 +19,7 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 // SOFTWARE.
 
+use crate::config::CONFIG;
 pub type ChainPrefix = u16;
 pub type ChainTokenSymbol = String;
 
@@ -37,6 +38,22 @@ impl SupportedRuntime {
             Self::Westend => 42,
         }
     }
+
+    pub fn is_people_runtime_available(&self) -> bool {
+        match &self {
+            Self::Polkadot => false,
+            Self::Kusama => true,
+            Self::Westend => false,
+        }
+    }
+
+    pub fn people_runtime(&self) -> SupportedParasRuntime {
+        match &self {
+            Self::Polkadot => SupportedParasRuntime::PeoplePolkadot,
+            Self::Kusama => SupportedParasRuntime::PeopleKusama,
+            _ => unimplemented!("Chain not supported"),
+        }
+    }
 }
 
 impl From<String> for SupportedRuntime {
@@ -73,3 +90,28 @@ impl std::fmt::Display for SupportedRuntime {
         }
     }
 }
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub enum SupportedParasRuntime {
+    PeoplePolkadot,
+    PeopleKusama,
+}
+
+impl SupportedParasRuntime {
+    pub fn default_rpc_url(&self) -> String {
+        let config = CONFIG.clone();
+        match &self {
+            Self::PeopleKusama => config.substrate_people_ws_url,
+            _ => unimplemented!("Chain not supported"),
+        }
+    }
+}
+
+impl std::fmt::Display for SupportedParasRuntime {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::PeoplePolkadot => write!(f, "People Polkadot"),
+            Self::PeopleKusama => write!(f, "People Kusama"),
+        }
+    }
+}