Feature/node tester disconnect (#3552)

* Ability to disconnect and reconnect GatewayClient

* usage of ibid. inside NodeTester

* example

* wasm-compatible `wait_for_shutdown` (for the future)

clients/webassembly/internal-dev/worker.js

@@ -139,6 +139,40 @@ async function testWithTester() {
             }
         }
     };
+}
+
+async function testerReconnection() {
+    const validator = 'https://qwerty-validator-api.qa.nymte.ch/api';
+    const nodeTester = await NymNodeTester.new_with_api(validator);
+
+    self.onmessage = async event => {
+        if (event.data && event.data.kind) {
+            switch (event.data.kind) {
+                case 'TestPacket': {
+                    const {mixnodeIdentity} = event.data.args;
+                    console.log("starting node test...");
+
+                    let result1 = await nodeTester.test_node(mixnodeIdentity);
+                    console.log("sleeping for 5s");
+                    await new Promise(r => setTimeout(r, 5000));
+                    await nodeTester.disconnect_from_gateway();
+
+                    console.log("sleeping for 5s");
+                    await new Promise(r => setTimeout(r, 5000));
+
+                    await nodeTester.reconnect_to_gateway();
+                    let result2 = await nodeTester.test_node(mixnodeIdentity);
+
+                    printAndDisplayTestResult(result1)
+                    printAndDisplayTestResult(result2)
+                }
+            }
+        }
+    };
+
+
+
+
 }
 
 async function testWithNymClient() {
@@ -330,6 +364,9 @@ async function main() {
     // sets up better stack traces in case of in-rust panics
     set_panic_hook();
 
+    // show reconnection capabilities
+    await testerReconnection()
+
     // run test on simplified and dedicated tester:
     // await testWithTester()
 

clients/webassembly/src/tester/helpers.rs

@@ -4,6 +4,9 @@
 // due to expansion of #[wasm_bindgen] macro on NodeTestResult
 #![allow(clippy::drop_non_drop)]
 
+use crate::error::WasmClientError;
+use crate::tester::LockedGatewayClient;
+use js_sys::Promise;
 use nym_node_tester_utils::processor::Received;
 use nym_node_tester_utils::receiver::ReceivedReceiver;
 use serde::{Deserialize, Serialize};
@@ -12,6 +15,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use tokio::sync::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard};
 use wasm_bindgen::prelude::*;
+use wasm_bindgen_futures::future_to_promise;
 use wasm_utils::{console_log, console_warn};
 
 #[derive(Clone)]
@@ -107,3 +111,37 @@ impl Drop for TestMarker {
         self.value.store(false, Ordering::SeqCst)
     }
 }
+
+pub(crate) trait GatewayReconnection {
+    fn disconnect_from_gateway(&self) -> Promise;
+
+    fn reconnect_to_gateway(&self) -> Promise;
+}
+
+impl GatewayReconnection for LockedGatewayClient {
+    fn disconnect_from_gateway(&self) -> Promise {
+        let this = self.clone();
+
+        future_to_promise(async move {
+            let mut guard = this.lock().await;
+            guard
+                .disconnect()
+                .await
+                .map_err(|err| JsValue::from(WasmClientError::from(err)))?;
+            Ok(JsValue::undefined())
+        })
+    }
+
+    fn reconnect_to_gateway(&self) -> Promise {
+        let this = self.clone();
+
+        future_to_promise(async move {
+            let mut guard = this.lock().await;
+            guard
+                .try_reconnect()
+                .await
+                .map_err(|err| JsValue::from(WasmClientError::from(err)))?;
+            Ok(JsValue::undefined())
+        })
+    }
+}

clients/webassembly/src/tester/mod.rs

@@ -7,7 +7,7 @@ use crate::helpers::{current_network_topology_async, setup_from_topology};
 use crate::storage::ClientStorage;
 use crate::tester::ephemeral_receiver::EphemeralTestReceiver;
 use crate::tester::helpers::{
-    NodeTestResult, ReceivedReceiverWrapper, TestMarker, WasmTestMessageExt,
+    GatewayReconnection, NodeTestResult, ReceivedReceiverWrapper, TestMarker, WasmTestMessageExt,
 };
 use crate::topology::WasmNymTopology;
 use futures::channel::mpsc;
@@ -257,6 +257,14 @@ impl NymNodeTester {
         })
     }
 
+    pub fn disconnect_from_gateway(&self) -> Promise {
+        self.gateway_client.disconnect_from_gateway()
+    }
+
+    pub fn reconnect_to_gateway(&self) -> Promise {
+        self.gateway_client.reconnect_to_gateway()
+    }
+
     fn prepare_test_packets(
         &self,
         mixnode_identity: String,

common/client-libs/gateway-client/src/client.rs

@@ -724,7 +724,7 @@ impl<C, St> GatewayClient<C, St> {
         Ok(())
     }
 
-    async fn try_reconnect(&mut self) -> Result<(), GatewayClientError> {
+    pub async fn try_reconnect(&mut self) -> Result<(), GatewayClientError> {
         if !self.connection.is_established() {
             self.establish_connection().await?;
         }
@@ -738,6 +738,12 @@ impl<C, St> GatewayClient<C, St> {
         Ok(())
     }
 
+    pub async fn disconnect(&mut self) -> Result<(), GatewayClientError> {
+        self.recover_socket_connection().await?;
+        self.connection = SocketState::NotConnected;
+        Ok(())
+    }
+
     pub async fn authenticate_and_start(&mut self) -> Result<Arc<SharedKeys>, GatewayClientError>
     where
         C: DkgQueryClient + Send + Sync,

common/task/src/manager.rs

@@ -177,20 +177,29 @@ impl TaskManager {
             drop(notify_rx);
         }
 
+        #[cfg(not(target_arch = "wasm32"))]
+        let interrupt_future = tokio::signal::ctrl_c();
+
         // in wasm we'll never get our shutdown anyway...
         #[cfg(target_arch = "wasm32")]
-        futures::future::pending::<()>().await;
+        let interrupt_future = futures::future::pending::<()>();
 
         #[cfg(not(target_arch = "wasm32"))]
+        let wait_future = tokio::time::sleep(Duration::from_secs(self.shutdown_timer_secs));
+
+        // TODO: we should be using a `Delay` here for wasm
+        #[cfg(target_arch = "wasm32")]
+        let wait_future = futures::future::pending::<()>();
+
         tokio::select! {
             _ = self.notify_tx.closed() => {
                 log::info!("All registered tasks succesfully shutdown");
             },
-            _ = tokio::signal::ctrl_c() => {
+            _ = interrupt_future => {
                 log::info!("Forcing shutdown");
             }
-            _ = tokio::time::sleep(Duration::from_secs(self.shutdown_timer_secs)) => {
-                log::info!("Timout reached, forcing shutdown");
+            _ = wait_future => {
+                log::info!("Timeout reached, forcing shutdown");
             },
         }
     }