From bbb1e5e15a96042b2df5582512fcc95d47505a09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Tue, 13 Jun 2023 17:25:42 +0100 Subject: [PATCH] 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 | 37 ++++++++++++++++++ clients/webassembly/src/tester/helpers.rs | 38 +++++++++++++++++++ clients/webassembly/src/tester/mod.rs | 10 ++++- .../client-libs/gateway-client/src/client.rs | 8 +++- common/task/src/manager.rs | 17 +++++++-- 5 files changed, 104 insertions(+), 6 deletions(-) diff --git a/clients/webassembly/internal-dev/worker.js b/clients/webassembly/internal-dev/worker.js index b52963b0e8..01e3504e78 100644 --- a/clients/webassembly/internal-dev/worker.js +++ b/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() diff --git a/clients/webassembly/src/tester/helpers.rs b/clients/webassembly/src/tester/helpers.rs index 9ee30047b7..b2ce1b9e5c 100644 --- a/clients/webassembly/src/tester/helpers.rs +++ b/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()) + }) + } +} diff --git a/clients/webassembly/src/tester/mod.rs b/clients/webassembly/src/tester/mod.rs index 3b4104e603..f7efac1043 100644 --- a/clients/webassembly/src/tester/mod.rs +++ b/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, diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index 2baf5718c0..a6ef774d8f 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/common/client-libs/gateway-client/src/client.rs @@ -724,7 +724,7 @@ impl GatewayClient { 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 GatewayClient { 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, GatewayClientError> where C: DkgQueryClient + Send + Sync, diff --git a/common/task/src/manager.rs b/common/task/src/manager.rs index f16cabb54b..75ab028f81 100644 --- a/common/task/src/manager.rs +++ b/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"); }, } }