From 067f3e6f1a3df923d9794d057dce19bf634bfd31 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bogdan-=C8=98tefan=20Neac=C5=9Fu?= Date: Fri, 5 Aug 2022 15:59:34 +0300 Subject: [PATCH] validator-api: handle SIGTERM (#1496) * validator-api: handle SIGTERM * Update CHANGELOG --- CHANGELOG.md | 2 + Cargo.lock | 1 + validator-api/Cargo.toml | 1 + validator-api/src/contract_cache/mod.rs | 27 +++++--- validator-api/src/main.rs | 62 +++++++++++++++---- validator-api/src/network_monitor/mod.rs | 9 ++- .../src/network_monitor/monitor/mod.rs | 13 ++-- .../src/network_monitor/monitor/receiver.rs | 8 ++- .../src/node_status_api/uptime_updater.rs | 30 +++++---- 9 files changed, 110 insertions(+), 43 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b12ec2d2a3..e0696c4e63 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/Cargo.lock b/Cargo.lock index e7bcd513f7..f8ffd17c35 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3325,6 +3325,7 @@ dependencies = [ "serde", "serde_json", "sqlx", + "task", "thiserror", "time 0.3.9", "tokio", diff --git a/validator-api/Cargo.toml b/validator-api/Cargo.toml index f1277fd6d1..8ec660cdb7 100644 --- a/validator-api/Cargo.toml +++ b/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"] } diff --git a/validator-api/src/contract_cache/mod.rs b/validator-api/src/contract_cache/mod.rs index d46b6f0e15..593fe63b88 100644 --- a/validator-api/src/contract_cache/mod.rs +++ b/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 ValidatorCacheRefresher { 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"); + } } } } diff --git a/validator-api/src/main.rs b/validator-api/src/main.rs index c085d8315d..b027b77e12 100644 --- a/validator-api/src/main.rs +++ b/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(()) diff --git a/validator-api/src/network_monitor/mod.rs b/validator-api/src/network_monitor/mod.rs index 0a15b87053..49d90deccb 100644 --- a/validator-api/src/network_monitor/mod.rs +++ b/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 }); } } diff --git a/validator-api/src/network_monitor/monitor/mod.rs b/validator-api/src/network_monitor/monitor/mod.rs index f6fd3fe881..2cd1950f96 100644 --- a/validator-api/src/network_monitor/monitor/mod.rs +++ b/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"); + } + } } } } diff --git a/validator-api/src/network_monitor/monitor/receiver.rs b/validator-api/src/network_monitor/monitor/receiver.rs index f837cb9f80..36913241db 100644 --- a/validator-api/src/network_monitor/monitor/receiver.rs +++ b/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; pub(crate) type GatewayClientUpdateReceiver = mpsc::UnboundedReceiver; @@ -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"); + } } } } diff --git a/validator-api/src/node_status_api/uptime_updater.rs b/validator-api/src/node_status_api/uptime_updater.rs index 417a9a30b2..56e7655aae 100644 --- a/validator-api/src/node_status_api/uptime_updater.rs +++ b/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"); + } } } }