diff --git a/common/task/src/cancellation.rs b/common/task/src/cancellation.rs index 8639a272f1..9e120b4ec5 100644 --- a/common/task/src/cancellation.rs +++ b/common/task/src/cancellation.rs @@ -70,6 +70,10 @@ impl ShutdownToken { } } + pub fn ephemeral() -> Self { + ShutdownToken::new("ephemeral-token") + } + // Creates a ShutdownToken which will get cancelled whenever the current token gets cancelled. // Unlike a cloned/forked ShutdownToken, cancelling a child token does not cancel the parent token. #[must_use] diff --git a/nym-api/src/epoch_operations/rewarded_set_assignment.rs b/nym-api/src/epoch_operations/rewarded_set_assignment.rs index 4ddaf78e7f..d513579ef2 100644 --- a/nym-api/src/epoch_operations/rewarded_set_assignment.rs +++ b/nym-api/src/epoch_operations/rewarded_set_assignment.rs @@ -212,18 +212,13 @@ impl EpochAdvancer { // SAFETY: the cache MUST HAVE been initialised before now #[allow(clippy::unwrap_used)] - warn!("⚠️⚠️⚠️⚠️⚠️ TRYING TO GET DESCRIBE CACHE"); let described_cache = self.described_cache.get().await.unwrap(); - warn!("⚠️⚠️⚠️⚠️⚠️ TRYING TO GET STATUS CACHE"); - let Some(status_cache) = self.status_cache.node_annotations().await else { warn!("there are no node annotations available"); return Vec::new(); }; - warn!("⚠️⚠️⚠️⚠️⚠️ GOT CACHES"); - for nym_node in nym_nodes { let node_id = nym_node.node_id(); let saturation = nym_node.rewarding_details.bond_saturation(reward_params); diff --git a/nym-api/src/support/caching/cache.rs b/nym-api/src/support/caching/cache.rs index 62219a9a2b..2f782f6098 100644 --- a/nym-api/src/support/caching/cache.rs +++ b/nym-api/src/support/caching/cache.rs @@ -7,7 +7,7 @@ use std::time::Duration; use thiserror::Error; use time::OffsetDateTime; use tokio::sync::{RwLock, RwLockMappedWriteGuard, RwLockReadGuard, RwLockWriteGuard}; -use tracing::{debug, warn}; +use tracing::debug; #[derive(Debug, Error)] #[error("the cache item has not been initialised")] diff --git a/nym-api/src/support/caching/refresher.rs b/nym-api/src/support/caching/refresher.rs index fbcc442a2e..05e6a3a3d4 100644 --- a/nym-api/src/support/caching/refresher.rs +++ b/nym-api/src/support/caching/refresher.rs @@ -18,7 +18,6 @@ pub struct RefreshRequester(Arc); impl RefreshRequester { pub(crate) fn request_cache_refresh(&self) { - warn!("REQUESTING SELF DESCRIBED REFRESH"); self.0.notify_waiters() } } @@ -176,8 +175,6 @@ where // note: `Notify` is not cancellation safe, HOWEVER, there's only one listener, // so it doesn't matter if we lose our queue position _ = self.refresh_requester.0.notified() => { - warn!("RECEIVED SELF DESCRIBED REFRESH REQUEST"); - self.refresh(&mut task_client).await; // since we just performed the full request, we can reset our existing interval refresh_interval.reset(); diff --git a/nym-node/src/error.rs b/nym-node/src/error.rs index 780956fa94..37bf7eb59a 100644 --- a/nym-node/src/error.rs +++ b/nym-node/src/error.rs @@ -85,6 +85,9 @@ pub enum NymNodeError { source: io::Error, }, + #[error("received shutdown signal while attempting to complete the action")] + ShutdownReceived, + #[error("could not find an existing config file at '{}' and fresh node initialisation has been disabled", config_path.display())] ForbiddenInitialisation { config_path: PathBuf }, diff --git a/nym-node/src/node/helpers.rs b/nym-node/src/node/helpers.rs index 05cc1fd388..bd62bb3740 100644 --- a/nym-node/src/node/helpers.rs +++ b/nym-node/src/node/helpers.rs @@ -9,6 +9,7 @@ use nym_crypto::asymmetric::{ed25519, x25519}; use nym_node_requests::api::v1::node::models::NodeDescription; use nym_pemstore::traits::{PemStorableKey, PemStorableKeyPair}; use nym_pemstore::KeyPairPath; +use nym_task::ShutdownToken; use nym_validator_client::nyxd::contract_traits::MixnetQueryClient; use nym_validator_client::QueryHttpRpcNyxdClient; use serde::Serialize; @@ -186,7 +187,7 @@ pub(crate) async fn get_current_rotation_id( nym_apis: &[Url], fallback_nyxd: &[Url], ) -> Result { - let apis_client = NymApisClient::new(nym_apis)?; + let apis_client = NymApisClient::new(nym_apis, ShutdownToken::ephemeral())?; if let Ok(rotation_info) = apis_client.get_key_rotation_info().await { if rotation_info.is_epoch_stuck() { return Err(NymNodeError::StuckEpoch); diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index 918199f60f..08b6f41395 100644 --- a/nym-node/src/node/mod.rs +++ b/nym-node/src/node/mod.rs @@ -828,10 +828,13 @@ impl NymNode { bin_info!().into() } - async fn try_refresh_remote_nym_api_cache(&self) -> Result<(), NymNodeError> { + async fn try_refresh_remote_nym_api_cache( + &self, + client: &NymApisClient, + ) -> Result<(), NymNodeError> { info!("attempting to request described cache refresh from nym-api(s)..."); - NymApisClient::new(&self.config.mixnet.nym_api_urls)? + client .broadcast_force_refresh(self.ed25519_identity_keys.private_key()) .await; Ok(()) @@ -983,7 +986,10 @@ impl NymNode { // I'm assuming this will be needed in other places, so it's explicitly extracted fn setup_nym_apis_client(&self) -> Result { - NymApisClient::new(&self.config.mixnet.nym_api_urls) + NymApisClient::new( + &self.config.mixnet.nym_api_urls, + self.shutdown_manager.clone_token("nym-apis-client"), + ) } #[track_caller] @@ -1120,12 +1126,14 @@ impl NymNode { } }); - self.try_refresh_remote_nym_api_cache().await?; + let nym_apis_client = self.setup_nym_apis_client()?; + + self.try_refresh_remote_nym_api_cache(&nym_apis_client) + .await?; self.start_verloc_measurements(); let network_refresher = self.build_network_refresher().await?; let active_clients_store = ActiveClientsStore::new(); - let nym_apis_client = self.setup_nym_apis_client()?; let bloomfilters_manager = self.setup_replay_detection().await?; diff --git a/nym-node/src/node/nym_apis_client.rs b/nym-node/src/node/nym_apis_client.rs index 5b0cd3e531..918e3a756c 100644 --- a/nym-node/src/node/nym_apis_client.rs +++ b/nym-node/src/node/nym_apis_client.rs @@ -6,6 +6,7 @@ use crate::node::NymNode; use futures::{stream, StreamExt}; use nym_crypto::asymmetric::ed25519; use nym_http_api_client::Client; +use nym_task::ShutdownToken; use nym_validator_client::client::NymApiClientExt; use nym_validator_client::models::{KeyRotationInfoResponse, NodeRefreshBody}; use nym_validator_client::nym_api::error::NymAPIError; @@ -15,8 +16,8 @@ use rand::thread_rng; use std::sync::Arc; use std::time::Duration; use tokio::sync::RwLock; -use tokio::time::timeout; -use tracing::warn; +use tokio::time::sleep; +use tracing::{debug, warn}; use url::Url; #[derive(Clone)] @@ -24,16 +25,18 @@ pub struct NymApisClient { inner: Arc>, } -const TODO: &str = "add shutdown signal to cancel any queries if received"; - struct InnerClient { active_client: NymApiClient, available_urls: Vec, + shutdown_token: ShutdownToken, currently_used_api: usize, } impl NymApisClient { - pub(crate) fn new(nym_apis: &[Url]) -> Result { + pub(crate) fn new( + nym_apis: &[Url], + shutdown_token: ShutdownToken, + ) -> Result { if nym_apis.is_empty() { return Err(NymNodeError::NoNymApiUrls); } @@ -51,6 +54,7 @@ impl NymApisClient { inner: Arc::new(RwLock::new(InnerClient { active_client: NymApiClient::from(active_client), available_urls: urls, + shutdown_token, currently_used_api: 0, })), }) @@ -127,9 +131,19 @@ impl InnerClient { } }); - let todo = "this fails "; - if timeout(timeout_duration, broadcast_fut).await.is_err() { - warn!("timed out while attempting to broadcast data to known nym apis") + let timeout_fut = sleep(timeout_duration); + + tokio::select! { + _ = broadcast_fut => { + debug!("managed to broadcast data to all nym apis") + } + _ = timeout_fut => { + warn!("timed out while attempting to broadcast data to known nym apis") + + } + _ = self.shutdown_token.cancelled() => { + debug!("received shutdown while attempting to broadcast data to known nym apis") + } } } @@ -155,13 +169,27 @@ impl InnerClient { .chain(self.available_urls.iter().enumerate().take(last_working)) { let nym_api = self.active_client.nym_api.clone_with_new_url(url.clone()); - match timeout(timeout_duration, req(nym_api)).await { - Ok(Ok(res)) => return Ok((res, idx)), - Ok(Err(err)) => { - warn!("failed to resolve query for {url}: {err}") + + let timeout_fut = sleep(timeout_duration); + let query_fut = req(nym_api); + + tokio::select! { + res = query_fut => { + debug!("managed to broadcast data to all nym apis"); + match res { + Ok(res) => return Ok((res, idx)), + Err(err) => { + warn!("failed to resolve query for {url}: {err}"); + } + } } - Err(_timeout) => { + _ = timeout_fut => { warn!("timed out while attempting to query {url}") + + } + _ = self.shutdown_token.cancelled() => { + debug!("received shutdown while attempting to query {url}"); + return Err(NymNodeError::ShutdownReceived) } } }