feat: add reconnecting-rpc-client

subxt/Cargo.toml

@@ -42,6 +42,9 @@ web = [
     "instant/wasm-bindgen"
 ]
 
+# Enable this to use the reconnecting rpc client
+reconnecting-rpc-client = ["dep:reconnecting-jsonrpsee-ws-client"]
+
 # Enable this to use jsonrpsee (allowing for example `OnlineClient::from_url`).
 jsonrpsee = ["dep:jsonrpsee"]
 
@@ -103,6 +106,9 @@ subxt-lightclient = { workspace = true, optional = true, default-features = fals
 # Light client support:
 tokio-stream = { workspace = true, optional = true }
 
+# Reconnecting jsonrpc ws client
+reconnecting-jsonrpsee-ws-client = { version = "0.2", optional = true }
+
 # For parsing urls to disallow insecure schemes
 url = { workspace = true }
 
@@ -138,6 +144,11 @@ name = "light_client_parachains"
 path = "examples/light_client_parachains.rs"
 required-features = ["unstable-light-client", "jsonrpsee", "native"]
 
+[[example]]
+name = "reconnecting_rpc_client"
+path = "examples/reconnecting_rpc_client.rs"
+required-features = ["reconnecting-rpc-client"]
+
 [package.metadata.docs.rs]
 features = ["default", "substrate-compat", "unstable-light-client"]
 rustdoc-args = ["--cfg", "docsrs"]

subxt/examples/reconnecting_rpc_client.rs

@@ -0,0 +1,103 @@
+//! Example to utilize the `reconnecting rpc client` in subxt
+//! which hidden behind behind `--feature reconnecting-rpc-client`
+//!
+//! To utilize full logs from the RPC client use:
+//! `RUST_LOG="jsonrpsee=trace,reconnecting_jsonrpsee_ws_client=trace"`
+
+#![allow(missing_docs)]
+
+use std::time::Duration;
+
+use subxt::backend::rpc::reconnecting_rpc_client::{Client, ExponentialBackoff, PingConfig};
+use subxt::backend::rpc::RpcClient;
+use subxt::error::{Error, RpcError};
+use subxt::{tx::TxStatus, OnlineClient, PolkadotConfig};
+use subxt_signer::sr25519::dev;
+
+// Generate an interface that we can use from the node's metadata.
+#[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata_small.scale")]
+pub mod polkadot {}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tracing_subscriber::fmt::init();
+
+    // Create a new client with with a reconnecting RPC client.
+    let rpc = Client::builder()
+        // Reconnect with exponential backoff.
+        .retry_policy(ExponentialBackoff::from_millis(100).max_delay(Duration::from_secs(10)))
+        // Send period WebSocket pings/pongs every 6th second and if it's not ACK:ed in 30 seconds
+        // then disconnect.
+        //
+        // This is just a way to ensure that the connection isn't idle if no message is sent that often
+        .enable_ws_ping(
+            PingConfig::new()
+                .ping_interval(Duration::from_secs(6))
+                .inactive_limit(Duration::from_secs(30)),
+        )
+        // There are other configurations as well that can be found here:
+        // <https://docs.rs/reconnecting-jsonrpsee-ws-client/latest/reconnecting_jsonrpsee_ws_client/struct.ClientBuilder.html>
+        .build("ws://localhost:9944".to_string())
+        .await?;
+
+    let api: OnlineClient<PolkadotConfig> =
+        OnlineClient::from_rpc_client(RpcClient::new(rpc.clone())).await?;
+
+    // Build a balance transfer extrinsic.
+    let dest = dev::bob().public_key().into();
+    let balance_transfer_tx = polkadot::tx().balances().transfer_allow_death(dest, 10_000);
+
+    // Submit the balance transfer extrinsic from Alice, and wait for it to be successful
+    // and in a finalized block. We get back the extrinsic events if all is well.
+    let from = dev::alice();
+
+    let mut balance_transfer_progress = api
+        .tx()
+        .sign_and_submit_then_watch_default(&balance_transfer_tx, &from)
+        .await?;
+
+    // When the connection is lost, a error RpcError::DisconnectWillReconnect is emitted on the stream.
+    // in such scenarios the error will be seen here.
+    //
+    // In such scenario it's possible that messages are lost when reconnecting
+    // and if that's acceptable you may just ignore that error message.
+    while let Some(status) = balance_transfer_progress.next().await {
+        match status {
+            // It's finalized in a block!
+            Ok(TxStatus::InFinalizedBlock(in_block)) => {
+                // grab the events and fail if no ExtrinsicSuccess event seen:
+                let events = in_block.wait_for_success().await?;
+                // We can look for events (this uses the static interface; we can also iterate
+                // over them and dynamically decode them):
+                let transfer_event = events.find_first::<polkadot::balances::events::Transfer>()?;
+
+                if transfer_event.is_some() {
+                    println!("Balance transfer success");
+                } else {
+                    println!("Failed to find Balances::Transfer Event");
+                }
+            }
+            // Just log any other status we encounter:
+            //
+            // In this example we emit some important status handling for
+            // here such as Dropped, Invalid etc....
+            Ok(_) => {
+                println!("New status");
+            }
+            // In this example we just ignore when reconnections occurs
+            // but it's technically possible that we can lose
+            // messages on the subscription such as `InFinalizedBlock`
+            // when reconnecting.
+            Err(Error::Rpc(RpcError::DisconnectedWillReconnect(e))) => {
+                println!("{:?}", e);
+            }
+            Err(err) => {
+                return Err(err.into());
+            }
+        }
+    }
+
+    println!("RPC client reconnected `{}` times", rpc.reconnect_count());
+
+    Ok(())
+}

subxt/src/backend/rpc/mod.rs

@@ -60,9 +60,12 @@ crate::macros::cfg_jsonrpsee! {
     mod jsonrpsee_impl;
 }
 
+crate::macros::cfg_reconnecting_rpc_client! {
+    pub use reconnecting_jsonrpsee_ws_client as reconnecting_rpc_client;
+}
+
 mod rpc_client;
 mod rpc_client_t;
 
-pub use rpc_client_t::{RawRpcFuture, RawRpcSubscription, RawValue, RpcClientT};
-
 pub use rpc_client::{rpc_params, RpcClient, RpcParams, RpcSubscription};
+pub use rpc_client_t::{RawRpcFuture, RawRpcSubscription, RawValue, RpcClientT};

subxt/src/backend/rpc/rpc_client_t.rs

@@ -101,3 +101,94 @@ impl<T: RpcClientT> RpcClientT for Box<T> {
         (**self).subscribe_raw(sub, params, unsub)
     }
 }
+
+#[cfg(feature = "reconnecting-rpc-client")]
+impl RpcClientT for reconnecting_jsonrpsee_ws_client::Client {
+    fn request_raw<'a>(
+        &'a self,
+        method: &'a str,
+        params: Option<Box<RawValue>>,
+    ) -> RawRpcFuture<'a, Box<RawValue>> {
+        use futures::FutureExt;
+
+        async {
+            self.request_raw(method.to_string(), params)
+                .await
+                .map_err(|e| RpcError::ClientError(Box::new(e)))
+        }
+        .boxed()
+    }
+
+    fn subscribe_raw<'a>(
+        &'a self,
+        sub: &'a str,
+        params: Option<Box<RawValue>>,
+        unsub: &'a str,
+    ) -> RawRpcFuture<'a, RawRpcSubscription> {
+        use futures::{FutureExt, StreamExt, TryStreamExt};
+        use reconnecting_jsonrpsee_ws_client::{RpcError as ReconnError, SubscriptionId};
+
+        // Helper for more human friendly error messages.
+        // Workaround for: https://github.com/paritytech/jsonrpsee/issues/1280
+        fn rpc_error(err: ReconnError) -> String {
+            match err {
+                ReconnError::Transport(e) => e.to_string(),
+                e @ ReconnError::InvalidSubscriptionId => e.to_string(),
+                ReconnError::InvalidRequestId(e) => e.to_string(),
+                ReconnError::Custom(e) => e,
+                ReconnError::RestartNeeded(inner_err) => {
+                    // HACK: jsonrpsee has Error::RestartNeeded(Arc<Error>) and
+                    // we need to visit one child because
+                    // RestartNeeded only contains one layer of recursion.
+                    //
+                    // Thus, `RestartNeeded(RestartNeededArc<Error>)`
+                    // and is unreachable.
+                    match &*inner_err {
+                        ReconnError::Transport(e) => e.to_string(),
+                        e @ ReconnError::InvalidSubscriptionId => e.to_string(),
+                        ReconnError::InvalidRequestId(e) => e.to_string(),
+                        ReconnError::Custom(e) => e.to_string(),
+                        // These variants are infallible.
+                        ReconnError::RestartNeeded(_)
+                        | ReconnError::RequestTimeout
+                        | ReconnError::MaxSlotsExceeded
+                        | ReconnError::Call(_)
+                        | ReconnError::ParseError(_)
+                        | ReconnError::HttpNotImplemented
+                        | ReconnError::EmptyBatchRequest(_)
+                        | ReconnError::RegisterMethod(_) => unreachable!(),
+                    }
+                }
+                // These variants are infallible.
+                ReconnError::RequestTimeout
+                | ReconnError::MaxSlotsExceeded
+                | ReconnError::Call(_)
+                | ReconnError::ParseError(_)
+                | ReconnError::HttpNotImplemented
+                | ReconnError::EmptyBatchRequest(_)
+                | ReconnError::RegisterMethod(_) => unreachable!(),
+            }
+        }
+
+        async {
+            let sub = self
+                .subscribe_raw(sub.to_string(), params, unsub.to_string())
+                .await
+                .map_err(|e| RpcError::ClientError(Box::new(e)))?;
+
+            let id = match sub.id() {
+                SubscriptionId::Num(n) => n.to_string(),
+                SubscriptionId::Str(s) => s.to_string(),
+            };
+            let stream = sub
+                .map_err(|e| RpcError::DisconnectedWillReconnect(Some(rpc_error(e.0))))
+                .boxed();
+
+            Ok(RawRpcSubscription {
+                stream,
+                id: Some(id),
+            })
+        }
+        .boxed()
+    }
+}

subxt/src/error/mod.rs

@@ -6,8 +6,6 @@
 
 mod dispatch_error;
 
-use core::fmt::Debug;
-
 crate::macros::cfg_unstable_light_client! {
     pub use crate::client::LightClientError;
 }
@@ -100,6 +98,13 @@ impl From<std::convert::Infallible> for Error {
     }
 }
 
+impl Error {
+    /// Checks whether the error was caused by a RPC re-connection.
+    pub fn is_disconnected_will_reconnect(&self) -> bool {
+        matches!(self, Error::Rpc(RpcError::DisconnectedWillReconnect(_)))
+    }
+}
+
 /// An RPC error. Since we are generic over the RPC client that is used,
 /// the error is boxed and could be casted.
 #[derive(Debug, thiserror::Error)]
@@ -120,6 +125,9 @@ pub enum RpcError {
     /// The requested URL is insecure.
     #[error("RPC error: insecure URL: {0}")]
     InsecureUrl(String),
+    /// The connection was lost and automatically reconnected.
+    #[error("RPC error: the connection was lost `{0:?}`; reconnect automatically initiated")]
+    DisconnectedWillReconnect(Option<String>),
 }
 
 impl RpcError {

subxt/src/macros.rs

@@ -52,7 +52,21 @@ macro_rules! cfg_jsonrpsee_web {
 	}
 }
 
-pub(crate) use {cfg_feature, cfg_jsonrpsee, cfg_substrate_compat, cfg_unstable_light_client};
+#[allow(unused)]
+macro_rules! cfg_reconnecting_rpc_client {
+	($($item:item)*) => {
+		$(
+			#[cfg(all(feature = "reconnecting-rpc-client"))]
+			#[cfg_attr(docsrs, doc(cfg(feature = "reconnecting-rpc-client")))]
+			$item
+		)*
+	}
+}
+
+pub(crate) use {
+    cfg_feature, cfg_jsonrpsee, cfg_reconnecting_rpc_client, cfg_substrate_compat,
+    cfg_unstable_light_client,
+};
 
 // Only used by light-client.
 #[allow(unused)]