Integration tests for unstable-reconnecting-rpc-client (#1711)

---------

Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>

testing/integration-tests/src/full_client/client/mod.rs

@@ -2,8 +2,10 @@
 // This file is dual-licensed as Apache-2.0 or GPL-3.0.
 // see LICENSE for license details.
 
+use std::collections::HashSet;
+
 use crate::{
-    subxt_test, test_context,
+    subxt_test, test_context, test_context_reconnecting_rpc_client,
     utils::{node_runtime, wait_for_blocks},
 };
 use codec::{Decode, Encode};
@@ -409,3 +411,41 @@ async fn partial_fee_estimate_correct() {
     // Both methods should yield the same fee
     assert_eq!(partial_fee_1, partial_fee_2);
 }
+
+#[subxt_test]
+async fn legacy_and_unstable_block_subscription_reconnect() {
+    let ctx = test_context_reconnecting_rpc_client().await;
+
+    let api = ctx.unstable_client().await;
+
+    let unstable_client_blocks = move |num: usize| {
+        let api = api.clone();
+        async move {
+            api.blocks()
+                .subscribe_finalized()
+                .await
+                .unwrap()
+                .take(num)
+                .map(|x| x.unwrap().hash().to_string())
+                .collect::<Vec<String>>()
+                .await
+        }
+    };
+
+    let blocks = unstable_client_blocks(3).await;
+    let blocks: HashSet<String> = HashSet::from_iter(blocks.into_iter());
+
+    assert!(blocks.len() == 3);
+
+    let ctx = ctx.restart().await;
+
+    // Make  client aware that connection was dropped and force them to reconnect
+    let _ = ctx.unstable_client().await.backend().genesis_hash().await;
+
+    let unstable_blocks = unstable_client_blocks(6).await;
+
+    let unstable_blocks: HashSet<String> = HashSet::from_iter(unstable_blocks.into_iter());
+    let intersection = unstable_blocks.intersection(&blocks).count();
+
+    assert!(intersection == 3);
+}

testing/integration-tests/src/utils/context.rs

@@ -7,17 +7,20 @@ pub(crate) use crate::{node_runtime, utils::TestNodeProcess};
 use subxt::client::OnlineClient;
 use subxt::SubstrateConfig;
 
+use super::node_proc::RpcClientKind;
+
 /// `substrate-node` should be installed on the $PATH. We fall back
 /// to also checking for an older `substrate` binary.
 const SUBSTRATE_NODE_PATHS: &str = "substrate-node,substrate";
 
-pub async fn test_context_with(authority: String) -> TestContext {
+pub async fn test_context_with(authority: String, rpc_client_kind: RpcClientKind) -> TestContext {
     let paths =
         std::env::var("SUBSTRATE_NODE_PATH").unwrap_or_else(|_| SUBSTRATE_NODE_PATHS.to_string());
     let paths: Vec<_> = paths.split(',').map(|p| p.trim()).collect();
 
     let mut proc = TestContext::build(&paths);
     proc.with_authority(authority);
+    proc.with_rpc_client_kind(rpc_client_kind);
     proc.spawn::<SubstrateConfig>().await.unwrap()
 }
 
@@ -28,5 +31,9 @@ pub type TestContext = TestNodeProcess<SubstrateConfig>;
 pub type TestClient = OnlineClient<SubstrateConfig>;
 
 pub async fn test_context() -> TestContext {
-    test_context_with("alice".to_string()).await
+    test_context_with("alice".to_string(), RpcClientKind::Legacy).await
+}
+
+pub async fn test_context_reconnecting_rpc_client() -> TestContext {
+    test_context_with("alice".to_string(), RpcClientKind::UnstableReconnecting).await
 }

testing/integration-tests/src/utils/node_proc.rs

@@ -5,7 +5,9 @@
 use std::cell::RefCell;
 use std::ffi::{OsStr, OsString};
 use std::sync::Arc;
+use std::time::Duration;
 use substrate_runner::SubstrateNode;
+use subxt::backend::rpc::reconnecting_rpc_client::{ExponentialBackoff, RpcClientBuilder};
 use subxt::{
     backend::{legacy, rpc, unstable},
     Config, OnlineClient,
@@ -58,26 +60,29 @@ where
         TestNodeProcessBuilder::new(paths)
     }
 
+    pub async fn restart(mut self) -> Self {
+        tokio::task::spawn_blocking(move || {
+            if let Some(ref mut proc) = &mut self.proc {
+                proc.restart().unwrap();
+            }
+            self
+        })
+        .await
+        .expect("to succeed")
+    }
+
     /// Hand back an RPC client connected to the test node which exposes the legacy RPC methods.
     pub async fn legacy_rpc_methods(&self) -> legacy::LegacyRpcMethods<R> {
-        let rpc_client = self.rpc_client().await;
+        let rpc_client = self.rpc_client.clone();
         legacy::LegacyRpcMethods::new(rpc_client)
     }
 
     /// Hand back an RPC client connected to the test node which exposes the unstable RPC methods.
     pub async fn unstable_rpc_methods(&self) -> unstable::UnstableRpcMethods<R> {
-        let rpc_client = self.rpc_client().await;
+        let rpc_client = self.rpc_client.clone();
         unstable::UnstableRpcMethods::new(rpc_client)
     }
 
-    /// Hand back an RPC client connected to the test node.
-    pub async fn rpc_client(&self) -> rpc::RpcClient {
-        let url = get_url(self.proc.as_ref().map(|p| p.ws_port()));
-        rpc::RpcClient::from_url(url)
-            .await
-            .expect("Unable to connect RPC client to test node")
-    }
-
     /// Always return a client using the unstable backend.
     /// Only use for comparing backends; use [`TestNodeProcess::client()`] normally,
     /// which enables us to run each test against both backends.
@@ -109,12 +114,24 @@ where
     pub fn client(&self) -> OnlineClient<R> {
         self.client.clone()
     }
+
+    /// Returns the rpc client connected to the node
+    pub fn rpc_client(&self) -> rpc::RpcClient {
+        self.rpc_client.clone()
+    }
+}
+
+/// Kind of rpc client to use in tests
+pub enum RpcClientKind {
+    Legacy,
+    UnstableReconnecting,
 }
 
 /// Construct a test node process.
 pub struct TestNodeProcessBuilder {
     node_paths: Vec<OsString>,
     authority: Option<String>,
+    rpc_client: RpcClientKind,
 }
 
 impl TestNodeProcessBuilder {
@@ -132,9 +149,16 @@ impl TestNodeProcessBuilder {
         Self {
             node_paths: paths,
             authority: None,
+            rpc_client: RpcClientKind::Legacy,
         }
     }
 
+    /// Set the testRunner to use a preferred RpcClient impl, ie Legacy or Unstable
+    pub fn with_rpc_client_kind(&mut self, rpc_client_kind: RpcClientKind) -> &mut Self {
+        self.rpc_client = rpc_client_kind;
+        self
+    }
+
     /// Set the authority dev account for a node in validator mode e.g. --alice.
     pub fn with_authority(&mut self, account: String) -> &mut Self {
         self.authority = Some(account);
@@ -161,9 +185,11 @@ impl TestNodeProcessBuilder {
         };
 
         let ws_url = get_url(proc.as_ref().map(|p| p.ws_port()));
-        let rpc_client = build_rpc_client(&ws_url)
-            .await
-            .map_err(|e| format!("Failed to connect to node at {ws_url}: {e}"))?;
+        let rpc_client = match self.rpc_client {
+            RpcClientKind::Legacy => build_rpc_client(&ws_url).await,
+            RpcClientKind::UnstableReconnecting => build_unstable_rpc_client(&ws_url).await,
+        }
+        .map_err(|e| format!("Failed to connect to node at {ws_url}: {e}"))?;
 
         // Cache whatever client we build, and None for the other.
         #[allow(unused_assignments, unused_mut)]
@@ -206,6 +232,16 @@ async fn build_rpc_client(ws_url: &str) -> Result<rpc::RpcClient, String> {
     Ok(rpc_client)
 }
 
+async fn build_unstable_rpc_client(ws_url: &str) -> Result<rpc::RpcClient, String> {
+    let client = RpcClientBuilder::new()
+        .retry_policy(ExponentialBackoff::from_millis(100).max_delay(Duration::from_secs(10)))
+        .build(ws_url.to_string())
+        .await
+        .map_err(|e| format!("Cannot construct RPC client: {e}"))?;
+
+    Ok(rpc::RpcClient::new(client))
+}
+
 async fn build_legacy_client<T: Config>(
     rpc_client: rpc::RpcClient,
 ) -> Result<OnlineClient<T>, String> {