Merge branch 'develop' into 306-wallet-balance-ui

CHANGELOG.md

@@ -37,6 +37,7 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 - validator: fixed local docker-compose setup to work on Apple M1 ([#1329])
 - explorer-api: listen out for SIGTERM and SIGQUIT too, making it play nicely as a system service ([#1482]).
 - network-requester: fix filter for suffix-only domains ([#1487])
+- validator-api: listen out for SIGTERM and SIGQUIT too, making it play nicely as a system service ([#1496]).
 
 ### Changed
 
@@ -77,6 +78,7 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 [#1478]: https://github.com/nymtech/nym/pull/1478
 [#1482]: https://github.com/nymtech/nym/pull/1482
 [#1487]: https://github.com/nymtech/nym/pull/1487
+[#1496]: https://github.com/nymtech/nym/pull/1496
 
 ## [nym-connect-v1.0.1](https://github.com/nymtech/nym/tree/nym-connect-v1.0.1) (2022-07-22)
 

Cargo.lock

@@ -3325,6 +3325,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sqlx",
+ "task",
  "thiserror",
  "time 0.3.9",
  "tokio",

common/types/src/currency.rs

@@ -507,7 +507,7 @@ mod test {
 
         for (expected, raw_display) in values {
             let coin = DecCoin {
-                denom: Network::MAINNET.mix_denom().display.into(),
+                denom: Network::MAINNET.mix_denom().display,
                 amount: raw_display.parse().unwrap(),
             };
             let base = reg.attempt_convert_to_base_coin(coin).unwrap();

explorer/.env.prod

@@ -1,5 +1,6 @@
 EXPLORER_API_URL=https://explorer.nymtech.net/api/v1
 VALIDATOR_API_URL=https://validator.nymtech.net
+VALIDATOR_URL=https://rpc.nyx.nodes.guru
 BIG_DIPPER_URL=https://blocks.nymtech.net
 CURRENCY_DENOM=unym
 CURRENCY_STAKING_DENOM=unyx

explorer/.env.qa

@@ -1,5 +1,6 @@
 EXPLORER_API_URL=https://qa-explorer.nymtech.net/api/v1
-VALIDATOR_API_URL=https://qa-validator.nymtech.net
+VALIDATOR_API_URL=https://qa-validator-api.nymtech.net
+VALIDATOR_URL=https://qa-validator.nymtech.net
 BIG_DIPPER_URL=https://qa-blocks.nymtech.net
 CURRENCY_DENOM=unymt
 CURRENCY_STAKING_DENOM=unyxt

explorer/src/api/constants.ts

@@ -1,6 +1,7 @@
 // master APIs
 export const API_BASE_URL = process.env.EXPLORER_API_URL;
 export const VALIDATOR_API_BASE_URL = process.env.VALIDATOR_API_URL;
+export const VALIDATOR_URL = process.env.VALIDATOR_URL;
 export const BIG_DIPPER = process.env.BIG_DIPPER_URL;
 
 // specific API routes
@@ -9,7 +10,7 @@ export const MIXNODE_PING = `${API_BASE_URL}/ping`;
 export const MIXNODES_API = `${API_BASE_URL}/mix-nodes`;
 export const MIXNODE_API = `${API_BASE_URL}/mix-node`;
 export const GATEWAYS_API = `${VALIDATOR_API_BASE_URL}/api/v1/gateways`;
-export const VALIDATORS_API = `${VALIDATOR_API_BASE_URL}/validators`;
+export const VALIDATORS_API = `${VALIDATOR_URL}/validators`;
 export const BLOCK_API = `${VALIDATOR_API_BASE_URL}/block`;
 export const COUNTRY_DATA_API = `${API_BASE_URL}/countries`;
 export const UPTIME_STORY_API = `${VALIDATOR_API_BASE_URL}/api/v1/status/mixnode`; // add ID then '/history' to this.

validator-api/Cargo.toml

@@ -58,6 +58,7 @@ gateway-client = { path="../common/client-libs/gateway-client" }
 mixnet-contract-common = { path= "../common/cosmwasm-smart-contracts/mixnet-contract" }
 multisig-contract-common = { path = "../common/cosmwasm-smart-contracts/multisig-contract" }
 nymsphinx = { path="../common/nymsphinx" }
+task = { path = "../common/task" }
 topology = { path="../common/topology" }
 validator-api-requests = { path = "validator-api-requests" }
 validator-client = { path="../common/client-libs/validator-client", features = ["nymd-client"] }

validator-api/src/contract_cache/mod.rs

@@ -14,6 +14,7 @@ use okapi::openapi3::OpenApi;
 use rocket::Route;
 use rocket_okapi::openapi_get_routes_spec;
 use rocket_okapi::settings::OpenApiSettings;
+use task::ShutdownListener;
 
 use rocket::fairing::AdHoc;
 use serde::Serialize;
@@ -252,20 +253,26 @@ impl<C> ValidatorCacheRefresher<C> {
         Ok(())
     }
 
-    pub(crate) async fn run(&self)
+    pub(crate) async fn run(&self, mut shutdown: ShutdownListener)
     where
         C: CosmWasmClient + Sync,
     {
         let mut interval = time::interval(self.caching_interval);
-        loop {
-            interval.tick().await;
-            if let Err(err) = self.refresh_cache().await {
-                error!("Failed to refresh validator cache - {}", err);
-            } else {
-                // relaxed memory ordering is fine here. worst case scenario network monitor
-                // will just have to wait for an additional backoff to see the change.
-                // And so this will not really incur any performance penalties by setting it every loop iteration
-                self.cache.initialised.store(true, Ordering::Relaxed)
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                _ = interval.tick() => {
+                    if let Err(err) = self.refresh_cache().await {
+                        error!("Failed to refresh validator cache - {}", err);
+                    } else {
+                        // relaxed memory ordering is fine here. worst case scenario network monitor
+                        // will just have to wait for an additional backoff to see the change.
+                        // And so this will not really incur any performance penalties by setting it every loop iteration
+                        self.cache.initialised.store(true, Ordering::Relaxed)
+                    }
+                }
+                _ = shutdown.recv() => {
+                    trace!("UpdateHandler: Received shutdown");
+                }
             }
         }
     }

validator-api/src/main.rs

@@ -31,6 +31,7 @@ use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 use std::{fs, process};
+use task::ShutdownNotifier;
 use tokio::sync::Notify;
 // use validator_client::nymd::SigningNymdClient;
 // use validator_client::ValidatorClientError;
@@ -226,14 +227,44 @@ fn parse_args<'a>() -> ArgMatches<'a> {
     base_app.get_matches()
 }
 
-async fn wait_for_interrupt() {
-    if let Err(e) = tokio::signal::ctrl_c().await {
-        error!(
-            "There was an error while capturing SIGINT - {:?}. We will terminate regardless",
-            e
-        );
+async fn wait_for_interrupt(mut shutdown: ShutdownNotifier) {
+    wait_for_signal().await;
+
+    log::info!("Sending shutdown");
+    shutdown.signal_shutdown().ok();
+
+    log::info!("Waiting for tasks to finish... (Press ctrl-c to force)");
+    shutdown.wait_for_shutdown().await;
+
+    log::info!("Stopping nym validator API");
+}
+
+#[cfg(unix)]
+async fn wait_for_signal() {
+    use tokio::signal::unix::{signal, SignalKind};
+    let mut sigterm = signal(SignalKind::terminate()).expect("Failed to setup SIGTERM channel");
+    let mut sigquit = signal(SignalKind::quit()).expect("Failed to setup SIGQUIT channel");
+
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            log::info!("Received SIGINT");
+        },
+        _ = sigterm.recv() => {
+            log::info!("Received SIGTERM");
+        }
+        _ = sigquit.recv() => {
+            log::info!("Received SIGQUIT");
+        }
+    }
+}
+
+#[cfg(not(unix))]
+async fn wait_for_signal() {
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            log::info!("Received SIGINT");
+        },
     }
-    println!("Received SIGINT - the network monitor will terminate now");
 }
 
 fn setup_logging() {
@@ -547,6 +578,7 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
     let signing_nymd_client = Client::new_signing(&config);
 
     let liftoff_notify = Arc::new(Notify::new());
+    let shutdown = ShutdownNotifier::default();
 
     // let's build our rocket!
     let rocket = setup_rocket(
@@ -569,7 +601,8 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
         // setup our daily uptime updater. Note that if network monitor is disabled, then we have
         // no data for the updates and hence we don't need to start it up
         let uptime_updater = HistoricalUptimeUpdater::new(storage.clone());
-        tokio::spawn(async move { uptime_updater.run().await });
+        let shutdown_listener = shutdown.subscribe();
+        tokio::spawn(async move { uptime_updater.run(shutdown_listener).await });
 
         // spawn the cache refresher
         let validator_cache_refresher = ValidatorCacheRefresher::new(
@@ -578,7 +611,8 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
             validator_cache.clone(),
             Some(storage.clone()),
         );
-        tokio::spawn(async move { validator_cache_refresher.run().await });
+        let shutdown_listener = shutdown.subscribe();
+        tokio::spawn(async move { validator_cache_refresher.run(shutdown_listener).await });
 
         // spawn rewarded set updater
         let mut rewarded_set_updater =
@@ -593,11 +627,15 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
             None,
         );
 
+        let shutdown_listener = shutdown.subscribe();
         // spawn our cacher
-        tokio::spawn(async move { validator_cache_refresher.run().await });
+        tokio::spawn(async move { validator_cache_refresher.run(shutdown_listener).await });
     }
 
     // launch the rocket!
+    // Rocket handles shutdown on it's own, but its shutdown handling should be incorporated
+    // with that of the rest of the tasks.
+    // Currently it's runtime is forcefully terminated once the validator-api exits.
     let shutdown_handle = rocket.shutdown();
     tokio::spawn(rocket.launch());
 
@@ -610,12 +648,12 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
 
         // we're ready to go! spawn the network monitor!
         let runnables = monitor_builder.build().await;
-        runnables.spawn_tasks();
+        runnables.spawn_tasks(&shutdown);
     } else {
         info!("Network monitoring is disabled.");
     }
 
-    wait_for_interrupt().await;
+    wait_for_interrupt(shutdown).await;
     shutdown_handle.notify();
 
     Ok(())

validator-api/src/network_monitor/mod.rs

@@ -6,6 +6,7 @@ use crypto::asymmetric::{encryption, identity};
 use futures::channel::mpsc;
 use gateway_client::bandwidth::BandwidthController;
 use std::sync::Arc;
+use task::ShutdownNotifier;
 
 use crate::config::Config;
 use crate::contract_cache::ValidatorCache;
@@ -131,11 +132,13 @@ impl NetworkMonitorRunnables {
     // TODO: note, that is not exactly doing what we want, because when
     // `ReceivedProcessor` is constructed, it already spawns a future
     // this needs to be refactored!
-    pub(crate) fn spawn_tasks(self) {
+    pub(crate) fn spawn_tasks(self, shutdown: &ShutdownNotifier) {
         let mut packet_receiver = self.packet_receiver;
         let mut monitor = self.monitor;
-        tokio::spawn(async move { packet_receiver.run().await });
-        tokio::spawn(async move { monitor.run().await });
+        let shutdown_listener = shutdown.subscribe();
+        tokio::spawn(async move { packet_receiver.run(shutdown_listener).await });
+        let shutdown_listener = shutdown.subscribe();
+        tokio::spawn(async move { monitor.run(shutdown_listener).await });
     }
 }
 

validator-api/src/network_monitor/monitor/mod.rs

@@ -12,6 +12,7 @@ use crate::storage::ValidatorApiStorage;
 use log::{debug, error, info};
 use std::collections::{HashMap, HashSet};
 use std::process;
+use task::ShutdownListener;
 use tokio::time::{sleep, Duration, Instant};
 
 pub(crate) mod gateway_clients_cache;
@@ -296,7 +297,7 @@ impl Monitor {
         self.test_nonce += 1;
     }
 
-    pub(crate) async fn run(&mut self) {
+    pub(crate) async fn run(&mut self, mut shutdown: ShutdownListener) {
         self.received_processor.start_receiving();
 
         // wait for validator cache to be ready
@@ -308,9 +309,13 @@ impl Monitor {
             .spawn_gateways_pinger(self.gateway_ping_interval);
 
         let mut run_interval = tokio::time::interval(self.run_interval);
-        loop {
-            run_interval.tick().await;
-            self.test_run().await;
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                _  = run_interval.tick() => self.test_run().await,
+                _ = shutdown.recv() => {
+                    trace!("UpdateHandler: Received shutdown");
+                }
+            }
         }
     }
 }

validator-api/src/network_monitor/monitor/receiver.rs

@@ -7,6 +7,7 @@ use crypto::asymmetric::identity;
 use futures::channel::mpsc;
 use futures::StreamExt;
 use gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver};
+use task::ShutdownListener;
 
 pub(crate) type GatewayClientUpdateSender = mpsc::UnboundedSender<GatewayClientUpdate>;
 pub(crate) type GatewayClientUpdateReceiver = mpsc::UnboundedReceiver<GatewayClientUpdate>;
@@ -55,8 +56,8 @@ impl PacketReceiver {
             .expect("packet processor seems to have crashed!");
     }
 
-    pub(crate) async fn run(&mut self) {
-        loop {
+    pub(crate) async fn run(&mut self, mut shutdown: ShutdownListener) {
+        while !shutdown.is_shutdown() {
             tokio::select! {
                 // unwrap here is fine as it can only return a `None` if the PacketSender has died
                 // and if that was the case, then the entire monitor is already in an undefined state
@@ -67,6 +68,9 @@ impl PacketReceiver {
                 Some((_gateway_id, message)) = self.gateways_reader.stream_map().next() => {
                         self.process_gateway_messages(message)
                 }
+                _ = shutdown.recv() => {
+                    trace!("UpdateHandler: Received shutdown");
+                }
             }
         }
     }

validator-api/src/node_status_api/uptime_updater.rs

@@ -7,6 +7,7 @@ use crate::node_status_api::models::{
 use crate::node_status_api::ONE_DAY;
 use crate::storage::ValidatorApiStorage;
 use log::error;
+use task::ShutdownListener;
 use time::OffsetDateTime;
 use tokio::time::sleep;
 
@@ -67,18 +68,23 @@ impl HistoricalUptimeUpdater {
         Ok(())
     }
 
-    pub(crate) async fn run(&self) {
-        loop {
-            // start any updates a day after starting the task so that we would have complete data
-            sleep(ONE_DAY).await;
-            if let Err(err) = self.update_uptimes().await {
-                // normally that would have been a warning rather than an error,
-                // however, in this case it implies some underlying issues with our database
-                // that might affect the entire program
-                error!(
-                    "We failed to update daily uptimes of active nodes - {}",
-                    err
-                )
+    pub(crate) async fn run(&self, mut shutdown: ShutdownListener) {
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                _ = sleep(ONE_DAY) => {
+                    if let Err(err) = self.update_uptimes().await {
+                        // normally that would have been a warning rather than an error,
+                        // however, in this case it implies some underlying issues with our database
+                        // that might affect the entire program
+                        error!(
+                            "We failed to update daily uptimes of active nodes - {}",
+                            err
+                        )
+                    }
+                }
+                _ = shutdown.recv() => {
+                    trace!("UpdateHandler: Received shutdown");
+                }
             }
         }
     }