passing shutdown to nym apis client

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]

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);

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")]

nym-api/src/support/caching/refresher.rs

@@ -18,7 +18,6 @@ pub struct RefreshRequester(Arc<Notify>);
 
 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();

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 },
 

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<u32, NymNodeError> {
-    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);

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, NymNodeError> {
-        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?;
 

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<RwLock<InnerClient>>,
 }
 
-const TODO: &str = "add shutdown signal to cancel any queries if received";
-
 struct InnerClient {
     active_client: NymApiClient,
     available_urls: Vec<Url>,
+    shutdown_token: ShutdownToken,
     currently_used_api: usize,
 }
 
 impl NymApisClient {
-    pub(crate) fn new(nym_apis: &[Url]) -> Result<Self, NymNodeError> {
+    pub(crate) fn new(
+        nym_apis: &[Url],
+        shutdown_token: ShutdownToken,
+    ) -> Result<Self, NymNodeError> {
         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)
                 }
             }
         }