diff --git a/Cargo.lock b/Cargo.lock index eb8cb6579e..1985ea9c58 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7609,6 +7609,8 @@ dependencies = [ "nix 0.27.1", "nym-bin-common", "nym-config", + "nym-task", + "reqwest", "serde", "serde_json", "thiserror", diff --git a/common/task/src/signal.rs b/common/task/src/signal.rs index e3fb3786a2..483c7c630b 100644 --- a/common/task/src/signal.rs +++ b/common/task/src/signal.rs @@ -1,7 +1,6 @@ use crate::{manager::SentError, TaskManager}; #[cfg(unix)] -#[deprecated] pub async fn wait_for_signal() { use tokio::signal::unix::{signal, SignalKind}; let mut sigterm = signal(SignalKind::terminate()).expect("Failed to setup SIGTERM channel"); diff --git a/tools/nymvisor/Cargo.toml b/tools/nymvisor/Cargo.toml index 0ab001a2e2..9639f2f2c8 100644 --- a/tools/nymvisor/Cargo.toml +++ b/tools/nymvisor/Cargo.toml @@ -20,6 +20,7 @@ humantime = "2.1.0" humantime-serde = "1.1.1" lazy_static = { workspace = true } nix = { version = "0.27.1", features = ["signal"] } +reqwest = { workspace = true, features = ["json"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } time = { workspace = true, features = [ "serde-human-readable" ] } @@ -31,3 +32,7 @@ url = { workspace = true, features = ["serde"] } async-file-watcher = { path = "../../common/async-file-watcher" } nym-bin-common = { path = "../../common/bin-common", features = ["output_format", "basic_tracing"] } nym-config = { path = "../../common/config" } +nym-task = { path = "../../common/task"} + +[dev-dependencies] +tokio = { workspace = true, features = ["full"] } \ No newline at end of file diff --git a/tools/nymvisor/src/cli/init.rs b/tools/nymvisor/src/cli/init.rs index 61cf2c5d38..80789bc000 100644 --- a/tools/nymvisor/src/cli/init.rs +++ b/tools/nymvisor/src/cli/init.rs @@ -32,6 +32,12 @@ pub(crate) struct Args { #[arg(long)] upstream_base_upgrade_url: Option, + /// Specifies the rate of polling the upstream url for upgrade information. + /// default: 1h + /// Can be overridden with $NYMVISOR_UPSTREAM_POLLING_RATE + #[arg(long, value_parser = humantime::parse_duration)] + upstream_polling_rate: Option, + /// If enabled, this will disable `nymvisor` logs (but not the underlying process) /// Can be overridden with $NYMVISOR_DISABLE_LOGS environmental variable. #[arg(long)] @@ -127,6 +133,9 @@ impl Args { if let Some(upstream) = &self.upstream_base_upgrade_url { config.nymvisor.debug.upstream_base_upgrade_url = upstream.clone() } + if let Some(polling_rate) = self.upstream_polling_rate { + config.nymvisor.debug.upstream_polling_rate = polling_rate + } if self.disable_nymvisor_logs { config.nymvisor.debug.disable_logs = self.disable_nymvisor_logs; } @@ -352,22 +361,22 @@ fn setup_initial_upgrade_plan( let existing_plan = UpgradePlan::try_load(&plan_path)?; if let (Some(current_info), Some(existing_info)) = ( &genesis_info.binary_details, - &existing_plan.current.binary_details, + &existing_plan.current().binary_details, ) { if current_info != existing_info { // if possible, compare the actual full details return Err(NymvisorError::PreexistingUpgradePlan { path: plan_path, current_name: genesis_info.name, - existing_name: existing_plan.current.name, + existing_name: existing_plan.current().name.clone(), }); } - } else if genesis_info.name != existing_plan.current.name { + } else if genesis_info.name != existing_plan.current().name { // otherwise just check the upgrade name return Err(NymvisorError::PreexistingUpgradePlan { path: plan_path, current_name: genesis_info.name, - existing_name: existing_plan.current.name, + existing_name: existing_plan.current().name.clone(), }); } diff --git a/tools/nymvisor/src/cli/run.rs b/tools/nymvisor/src/cli/run.rs index b006dbbfe8..6b86701e99 100644 --- a/tools/nymvisor/src/cli/run.rs +++ b/tools/nymvisor/src/cli/run.rs @@ -41,40 +41,46 @@ pub(crate) fn execute(args: Args) -> Result<(), NymvisorError> { .build() .expect("failed to create the runtime"); - // spawn the root task - rt.block_on(async { - println!("run"); + // we have three tasks only: + // - one for managing the daemon launcher + // - the other one for watching the upgrade plan file + // - the last one for polling upstream source for upgrade info + // so once the daemon has finished, for whatever reason, abort the file watcher and upstream poller and terminate the nymvisor - let daemon = Daemon::from_config(&config); - let interrupt_notify = Arc::new(Notify::new()); - let running = daemon.execute_async(args.daemon_args, Arc::clone(&interrupt_notify))?; - - let handle1 = tokio::spawn(async move { - let res = running.await; - println!("the process has finished! with {res:?}"); - }); - - let (events_sender, mut events_receiver) = mpsc::unbounded(); - let mut watcher = AsyncFileWatcher::new_file_changes_watcher( - config.upgrade_plan_filepath(), - events_sender, - )?; - - let (abort_handle, abort_registration) = AbortHandle::new_pair(); - - let handle2 = - tokio::spawn(async move { Abortable::new(watcher.watch(), abort_registration).await }); - - let event = events_receiver.next().await; - println!("watcher event: {event:?}"); - interrupt_notify.notify_one(); - - handle1.await; - abort_handle.abort(); - handle2.await; - - // println!("{:?}", status); - - >::Ok(()) - }) + todo!() + // // spawn the root task + // rt.block_on(async { + // println!("run"); + // + // let daemon = Daemon::from_config(&config); + // let running = daemon.execute_async(args.daemon_args)?; + // + // let handle1 = tokio::spawn(async move { + // let res = running.await; + // println!("the process has finished! with {res:?}"); + // }); + // + // let (events_sender, mut events_receiver) = mpsc::unbounded(); + // let mut watcher = AsyncFileWatcher::new_file_changes_watcher( + // config.upgrade_plan_filepath(), + // events_sender, + // )?; + // + // let (abort_handle, abort_registration) = AbortHandle::new_pair(); + // + // let handle2 = + // tokio::spawn(async move { Abortable::new(watcher.watch(), abort_registration).await }); + // + // let event = events_receiver.next().await; + // println!("watcher event: {event:?}"); + // interrupt_notify.notify_one(); + // + // handle1.await; + // abort_handle.abort(); + // handle2.await; + // + // // println!("{:?}", status); + // + // >::Ok(()) + // }) } diff --git a/tools/nymvisor/src/config/mod.rs b/tools/nymvisor/src/config/mod.rs index ed447e27fb..36bdd54272 100644 --- a/tools/nymvisor/src/config/mod.rs +++ b/tools/nymvisor/src/config/mod.rs @@ -21,6 +21,7 @@ pub(crate) const DEFAULT_FAILURE_RESTART_DELAY: Duration = Duration::from_secs(1 pub(crate) const DEFAULT_STARTUP_PERIOD: Duration = Duration::from_secs(120); pub(crate) const DEFAULT_MAX_STARTUP_FAILURES: usize = 10; pub(crate) const DEFAULT_SHUTDOWN_GRACE_PERIOD: Duration = Duration::from_secs(10); +pub(crate) const DEFAULT_UPSTREAM_POLLING_RATE: Duration = Duration::from_secs(60 * 60); pub(crate) const DEFAULT_BASE_UPSTREAM_UPGRADE_INFO_SOURCE: &str = "https://nymtech.net/.wellknown/"; @@ -73,7 +74,7 @@ pub fn default_global_data_directory() -> PathBuf { .join(DEFAULT_DATA_DIR) } -#[derive(Debug, Deserialize, PartialEq, Serialize)] +#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)] #[serde(deny_unknown_fields)] pub struct Config { // additional metadata holding on-disk location of this config file @@ -273,6 +274,11 @@ impl Config { .join(&self.daemon.name) } + // e.g. $HOME/.nym/nymvisors/data/nym-api/current/upgrade-info.json + pub fn current_upgrade_info_filepath(&self) -> PathBuf { + self.current_daemon_dir().join(UPGRADE_INFO_FILENAME) + } + // e.g. $HOME/.nym/nymvisors/data/nym-api/upgrades//upgrade-info.json // or $HOME/.nym/nymvisors/data/nym-api/genesis/upgrade-info.json pub fn upgrade_info_filepath>(&self, upgrade_name: S) -> PathBuf { @@ -307,7 +313,7 @@ impl Config { } } -#[derive(Debug, Deserialize, PartialEq, Serialize)] +#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)] #[serde(deny_unknown_fields)] pub struct Nymvisor { /// ID specifies the human readable ID of this particular nymvisor instance. @@ -319,7 +325,7 @@ pub struct Nymvisor { pub debug: NymvisorDebug, } -#[derive(Debug, Deserialize, PartialEq, Serialize)] +#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)] #[serde(deny_unknown_fields)] pub struct NymvisorDebug { /// Sets the base url of the upstream source for obtaining upgrade information for the deaemon. @@ -328,6 +334,12 @@ pub struct NymvisorDebug { /// Can be overridden with $NYMVISOR_UPSTREAM_BASE_UPGRADE_URL environmental variable. pub upstream_base_upgrade_url: Url, + /// Specifies the rate of polling the upstream url for upgrade information. + /// default: 1h + /// Can be overridden with $NYMVISOR_UPSTREAM_POLLING_RATE + #[serde(with = "humantime_serde")] + pub upstream_polling_rate: Duration, + /// If set to true, this will disable `nymvisor` logs (but not the underlying process) /// default: false /// Can be overridden with $NYMVISOR_DISABLE_LOGS environmental variable. @@ -340,7 +352,6 @@ pub struct NymvisorDebug { pub upgrade_data_directory: Option, } -#[allow(clippy::derivable_impls)] impl Default for NymvisorDebug { fn default() -> Self { NymvisorDebug { @@ -349,13 +360,14 @@ impl Default for NymvisorDebug { upstream_base_upgrade_url: DEFAULT_BASE_UPSTREAM_UPGRADE_INFO_SOURCE .parse() .expect("default upstream url was malformed"), + upstream_polling_rate: DEFAULT_UPSTREAM_POLLING_RATE, disable_logs: false, upgrade_data_directory: None, } } } -#[derive(Debug, Deserialize, PartialEq, Serialize)] +#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)] #[serde(deny_unknown_fields)] pub struct Daemon { /// The name of the managed binary itself (e.g. nym-api, nym-mixnode, nym-gateway, etc.) @@ -373,7 +385,7 @@ pub struct Daemon { pub debug: DaemonDebug, } -#[derive(Debug, Deserialize, PartialEq, Serialize)] +#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)] #[serde(deny_unknown_fields)] pub struct DaemonDebug { /// Override url to the upstream source for upgrade plans for this daeamon. diff --git a/tools/nymvisor/src/daemon/mod.rs b/tools/nymvisor/src/daemon/mod.rs index 3b60618a6f..82a75b54c0 100644 --- a/tools/nymvisor/src/daemon/mod.rs +++ b/tools/nymvisor/src/daemon/mod.rs @@ -17,7 +17,21 @@ use std::task::{Context, Poll}; use std::time::Duration; use tokio::sync::Notify; use tokio::time::{sleep, Sleep}; -use tracing::{debug, error, info, instrument, warn}; +use tracing::{debug, info, instrument, warn}; + +pub(crate) struct InterruptHandle(Arc); + +impl InterruptHandle { + pub(crate) fn interrupt_daemon(&self) { + self.0.notify_one() + } +} + +impl Drop for InterruptHandle { + fn drop(&mut self) { + self.interrupt_daemon(); + } +} #[derive(Debug)] pub(crate) struct Daemon { @@ -76,11 +90,7 @@ impl Daemon { todo!() } - pub(crate) fn execute_async( - &self, - args: I, - interrupt_handle: Arc, - ) -> Result + pub(crate) fn execute_async(&self, args: I) -> Result where I: IntoIterator, S: AsRef, @@ -93,7 +103,7 @@ impl Daemon { .spawn() .map_err(|source| NymvisorError::DaemonIoFailure { source })?; - ExecutingDaemon::new(self.kill_timeout, interrupt_handle, child) + ExecutingDaemon::new(self.kill_timeout, child) } } @@ -103,6 +113,7 @@ pub(crate) struct ExecutingDaemon { child_id: i32, kill_timeout_duration: Duration, interrupt_sent: bool, + interrupt_handle: Option>, // interrupted: Option>>>, interrupted: Pin + Send + Sync>>, @@ -114,15 +125,17 @@ pub(crate) struct ExecutingDaemon { impl ExecutingDaemon { fn new( kill_timeout_duration: Duration, - interrupt_notify: Arc, mut child: tokio::process::Child, ) -> Result { if let Some(id) = child.id() { + let interrupt_handle = Arc::new(Notify::new()); + let notified_handle = Arc::clone(&interrupt_handle); Ok(ExecutingDaemon { child_id: id as i32, kill_timeout_duration, interrupt_sent: false, - interrupted: Box::pin(async move { interrupt_notify.notified().await }), + interrupt_handle: Some(interrupt_handle), + interrupted: Box::pin(async move { notified_handle.notified().await }), kill_timeout: None, child_future: Box::pin(async move { child.wait().await }), }) @@ -137,6 +150,15 @@ impl ExecutingDaemon { } } + pub(crate) fn interrupt_handle(&mut self) -> InterruptHandle { + #[allow(clippy::expect_used)] + InterruptHandle( + self.interrupt_handle + .take() + .expect("the interrupt handle has already been obtained"), + ) + } + fn signal_child(&self, signal: Signal) -> Result<(), NymvisorError> { info!("sending {signal} to the daemon"); nix::sys::signal::kill(Pid::from_raw(self.child_id), signal) @@ -145,13 +167,13 @@ impl ExecutingDaemon { } impl Future for ExecutingDaemon { - type Output = Result, NymvisorError>; + type Output = Result; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { // 1. check if the child is done if let Poll::Ready(result) = Pin::new(&mut self.child_future).poll(cx) { return match result { - Ok(exit_status) => Poll::Ready(Ok(Some(exit_status))), + Ok(exit_status) => Poll::Ready(Ok(exit_status)), Err(source) => Poll::Ready(Err(NymvisorError::DaemonIoFailure { source })), }; } diff --git a/tools/nymvisor/src/env.rs b/tools/nymvisor/src/env.rs index 75e3ad1c01..32fb196cee 100644 --- a/tools/nymvisor/src/env.rs +++ b/tools/nymvisor/src/env.rs @@ -15,6 +15,7 @@ pub mod vars { pub const NYMVISOR_ID: &str = "NYMVISOR_ID"; pub const NYMVISOR_CONFIG_PATH: &str = "NYMVISOR_CONFIG_PATH"; pub const NYMVISOR_UPSTREAM_BASE_UPGRADE_URL: &str = "NYMVISOR_UPSTREAM_BASE_UPGRADE_URL"; + pub const NYMVISOR_UPSTREAM_POLLING_RATE: &str = "NYMVISOR_UPSTREAM_POLLING_RATE"; pub const NYMVISOR_DISABLE_LOGS: &str = "NYMVISOR_DISABLE_LOGS"; pub const NYMVISOR_UPGRADE_DATA_DIRECTORY: &str = "NYMVISOR_UPGRADE_DATA_DIRECTORY"; @@ -45,6 +46,7 @@ pub(crate) struct Env { pub(crate) nymvisor_id: Option, pub(crate) nymvisor_config_path: Option, pub(crate) nymvisor_upstream_base_upgrade_url: Option, + pub(crate) nymvisor_upstream_polling_rate: Option, pub(crate) nymvisor_disable_logs: Option, pub(crate) nymvisor_upgrade_data_directory: Option, @@ -71,6 +73,9 @@ impl Env { if let Some(upstream) = &self.nymvisor_upstream_base_upgrade_url { config.nymvisor.debug.upstream_base_upgrade_url = upstream.clone() } + if let Some(polling_rate) = self.nymvisor_upstream_polling_rate { + config.nymvisor.debug.upstream_polling_rate = polling_rate + } if let Some(nymvisor_disable_logs) = self.nymvisor_disable_logs { config.nymvisor.debug.disable_logs = nymvisor_disable_logs; } @@ -212,6 +217,7 @@ impl Env { nymvisor_id: read_string(vars::NYMVISOR_ID)?, nymvisor_config_path: read_pathbuf(vars::NYMVISOR_CONFIG_PATH)?, nymvisor_upstream_base_upgrade_url: read_url(vars::NYMVISOR_UPSTREAM_BASE_UPGRADE_URL)?, + nymvisor_upstream_polling_rate: read_duration(vars::NYMVISOR_UPSTREAM_POLLING_RATE)?, nymvisor_disable_logs: read_bool(vars::NYMVISOR_DISABLE_LOGS)?, nymvisor_upgrade_data_directory: read_pathbuf(vars::NYMVISOR_UPGRADE_DATA_DIRECTORY)?, daemon_name: read_string(vars::DAEMON_NAME)?, diff --git a/tools/nymvisor/src/error.rs b/tools/nymvisor/src/error.rs index fbb5bf67c9..45d6559920 100644 --- a/tools/nymvisor/src/error.rs +++ b/tools/nymvisor/src/error.rs @@ -10,6 +10,7 @@ use std::num::ParseIntError; use std::path::PathBuf; use std::process::ExitStatus; use thiserror::Error; +use url::Url; #[derive(Debug, Error)] pub(crate) enum NymvisorError { @@ -203,6 +204,13 @@ pub(crate) enum NymvisorError { #[from] source: NotifyError, }, + + #[error("failed to query the upstream url ('{url}') - source")] + UpstreamQueryFailure { + url: Url, + #[source] + source: reqwest::Error, + }, } impl From for NymvisorError { diff --git a/tools/nymvisor/src/helpers.rs b/tools/nymvisor/src/helpers.rs new file mode 100644 index 0000000000..861e680706 --- /dev/null +++ b/tools/nymvisor/src/helpers.rs @@ -0,0 +1,37 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use futures::future::Aborted; +use futures::stream::AbortHandle; +use std::future::Future; +use tokio::task::{JoinError, JoinHandle}; + +pub(crate) struct TaskHandle +where + T: Send + 'static, +{ + abort_handle: AbortHandle, + // join_handle: JoinHandle, + join_handle: JoinHandle>, +} + +impl TaskHandle +where + T: Send + 'static, +{ + pub(crate) fn new( + abort_handle: AbortHandle, + join_handle: JoinHandle>, + ) -> Self { + TaskHandle { + abort_handle, + join_handle, + } + } + + // TODO: change return type + pub(crate) async fn abort_and_finalize(self) -> Result, JoinError> { + self.abort_handle.abort(); + self.join_handle.await + } +} diff --git a/tools/nymvisor/src/launcher/mod.rs b/tools/nymvisor/src/launcher/mod.rs deleted file mode 100644 index 74d8ab850d..0000000000 --- a/tools/nymvisor/src/launcher/mod.rs +++ /dev/null @@ -1,32 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::config::Config; - -pub(crate) struct DaemonLauncher { - config: Config, -} - -impl DaemonLauncher { - pub(crate) fn new(config: Config) -> Self { - todo!() - } - - // responsible for running until exit or until update is detected - pub(crate) fn run(&self, args: Vec) { - /* - - tokio select on: - - daemon terminating - - upgrade-plan.json changes - - https://nymtech.net/.wellknown//update-info.json changes - - - // todo: maybe move to a higher layer - - signals received (to propagate them to daemon before terminating to prevent creating zombie processes) - - */ - - todo!() - } -} diff --git a/tools/nymvisor/src/main.rs b/tools/nymvisor/src/main.rs index 4a2df29ab3..0e566d00da 100644 --- a/tools/nymvisor/src/main.rs +++ b/tools/nymvisor/src/main.rs @@ -12,7 +12,8 @@ pub(crate) mod config; pub(crate) mod daemon; pub(crate) mod env; pub(crate) mod error; -pub(crate) mod launcher; +pub(crate) mod helpers; +pub(crate) mod tasks; pub(crate) mod upgrades; fn main() -> anyhow::Result<()> { diff --git a/tools/nymvisor/src/tasks/launcher/mod.rs b/tools/nymvisor/src/tasks/launcher/mod.rs new file mode 100644 index 0000000000..0ed501f3b9 --- /dev/null +++ b/tools/nymvisor/src/tasks/launcher/mod.rs @@ -0,0 +1,109 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::config::Config; +use crate::daemon::Daemon; +use crate::error::NymvisorError; +use crate::upgrades::{UpgradeInfo, UpgradePlan}; +use async_file_watcher::FileWatcherEventReceiver; +use futures::future::{FusedFuture, OptionFuture}; +use futures::{FutureExt, StreamExt}; +use nym_task::signal::wait_for_signal; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::Notify; +use tokio::time::Sleep; +use tracing::{error, info, warn}; + +pub(crate) struct DaemonLauncher { + config: Config, + upgrade_plan_watcher: FileWatcherEventReceiver, +} + +impl DaemonLauncher { + pub(crate) fn new(config: Config) -> Self { + todo!() + } + + // responsible for running until exit or until update is detected + pub(crate) async fn run(&mut self, args: Vec) -> Result<(), NymvisorError> { + todo!() + } + + fn check_upgrade_plan_changes(&self) { + // + } + + async fn wait_for_upgrade_or_termination( + &mut self, + args: Vec, + ) -> Result { + let daemon = Daemon::from_config(&self.config); + let current_upgrade = UpgradeInfo::try_load(self.config.current_upgrade_info_filepath())?; + + // see if there's already a queued up upgrade + let current_upgrade_plan = UpgradePlan::try_load(self.config.upgrade_plan_filepath())?; + let next = current_upgrade_plan.next_upgrade(); + + let mut running_daemon = daemon.execute_async(args)?; + let interrupt_handle = running_daemon.interrupt_handle(); + + // we need to fuse the daemon future so that we could check if it has already terminated + let mut fused_runner = running_daemon.fuse(); + + let mut upgrade_timeout: OptionFuture<_> = None.into(); + + upgrade_timeout = + Some(Box::pin(tokio::time::sleep(Duration::from_secs(123))).fuse()).into(); + + let sig_fut = wait_for_signal(); + + // note: this has to be in a loop because `upgrade_plan_watcher` might receive events that do not necessarily trigger the upgrade + loop { + tokio::select! { + daemon_res = &mut fused_runner => { + warn!("the daemon has terminated by itself - was it a short lived command?"); + let exit_status = daemon_res?; + info!("it finished with the following exit status: {exit_status}"); + return Ok(false) + } + _ = &mut self.upgrade_plan_watcher.next() => { + // + } + _ = &mut upgrade_timeout, if !upgrade_timeout.is_terminated() => { + break + } + + + + } + } + + // if the runner hasn't terminated by itself (which should be the almost every single time!) + // send the interrupt and wait for it to be done + if fused_runner.is_terminated() { + todo!("error case") + } + interrupt_handle.interrupt_daemon(); + let res = fused_runner.await; + + /* + + tokio select on: + - daemon terminating + - upgrade-plan.json changes + - https://nymtech.net/.wellknown//update-info.json changes + + + // todo: maybe move to a higher layer + - signals received (to propagate them to daemon before terminating to prevent creating zombie processes) + + */ + + todo!() + } + + async fn perform_backup(&self) { + todo!() + } +} diff --git a/tools/nymvisor/src/tasks/mod.rs b/tools/nymvisor/src/tasks/mod.rs new file mode 100644 index 0000000000..52ab5f1c33 --- /dev/null +++ b/tools/nymvisor/src/tasks/mod.rs @@ -0,0 +1,6 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +pub(crate) mod launcher; +pub(crate) mod upgrade_plan_watcher; +pub(crate) mod upstream_poller; diff --git a/tools/nymvisor/src/tasks/upgrade_plan_watcher.rs b/tools/nymvisor/src/tasks/upgrade_plan_watcher.rs new file mode 100644 index 0000000000..8c9ecce509 --- /dev/null +++ b/tools/nymvisor/src/tasks/upgrade_plan_watcher.rs @@ -0,0 +1,25 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::config::Config; +use crate::error::NymvisorError; +use crate::helpers::TaskHandle; +use async_file_watcher::{AsyncFileWatcher, FileWatcherEventReceiver, NotifyResult}; +use futures::channel::mpsc; +use futures::future::{AbortHandle, Abortable}; + +pub(crate) fn start_upgrade_plan_watcher( + config: &Config, +) -> Result<(FileWatcherEventReceiver, TaskHandle>), NymvisorError> { + let (events_sender, events_receiver) = mpsc::unbounded(); + let mut watcher = + AsyncFileWatcher::new_file_changes_watcher(config.upgrade_plan_filepath(), events_sender)?; + + let (abort_handle, abort_registration) = AbortHandle::new_pair(); + let join_handle = + tokio::spawn(async move { Abortable::new(watcher.watch(), abort_registration).await }); + + let task_handle = TaskHandle::new(abort_handle, join_handle); + + Ok((events_receiver, task_handle)) +} diff --git a/tools/nymvisor/src/tasks/upstream_poller.rs b/tools/nymvisor/src/tasks/upstream_poller.rs new file mode 100644 index 0000000000..4708c9571e --- /dev/null +++ b/tools/nymvisor/src/tasks/upstream_poller.rs @@ -0,0 +1,81 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::config::Config; +use crate::error::NymvisorError; +use crate::helpers::TaskHandle; +use crate::upgrades::{UpgradeInfo, UpgradePlan}; +use futures::future::{AbortHandle, Abortable}; +use reqwest::get; +use tracing::{error, warn}; + +pub(crate) struct UpstreamPoller { + config: Config, +} + +impl UpstreamPoller { + pub(crate) fn new(config: &Config) -> Self { + UpstreamPoller { + config: config.clone(), + } + } + + /// Poll the upstream url to see if new upgrade has been published. + /// If so, save it to `upgrade-info.json` and update the `upgrade-plan.json` + async fn check_upstream(&self) -> Result<(), NymvisorError> { + let upgrade_info: UpgradeInfo = get(self.config.upstream_upgrade_url()) + .await + .map_err(|source| NymvisorError::UpstreamQueryFailure { + url: self.config.upstream_upgrade_url(), + source, + })? + .json() + .await + .map_err(|source| NymvisorError::UpstreamQueryFailure { + url: self.config.upstream_upgrade_url(), + source, + })?; + + let mut plan = UpgradePlan::try_load(self.config.upgrade_plan_filepath())?; + + // if the current version is the same as the one announced by upstream, we're done + if upgrade_info.version == plan.current().version { + return Ok(()); + } + + if !plan.has_planned(&upgrade_info) { + if let Err(err) = + upgrade_info.save(self.config.upgrade_info_filepath(&upgrade_info.name)) + { + error!("failed to save new upgrade info: {err}"); + return Err(err); + } + + if let Err(err) = plan.insert_new_upgrade(upgrade_info) { + error!("failed to insert new upgrade info into the current upgrade plan: {err}"); + return Err(err); + } + } + + Ok(()) + } + + pub(crate) async fn run(&mut self) { + let mut interval = tokio::time::interval(self.config.nymvisor.debug.upstream_polling_rate); + loop { + // note: first tick happens immediately + interval.tick().await; + if let Err(err) = self.check_upstream().await { + warn!("failed to check the upstream for new upgrade information: {err}. we will try to poll it again in {}", humantime::format_duration(interval.period())); + } + } + } + + pub(crate) async fn start(mut self) -> TaskHandle<()> { + let (abort_handle, abort_registration) = AbortHandle::new_pair(); + let join_handle = + tokio::spawn(async move { Abortable::new(self.run(), abort_registration).await }); + + TaskHandle::new(abort_handle, join_handle) + } +} diff --git a/tools/nymvisor/src/upgrades/mod.rs b/tools/nymvisor/src/upgrades/mod.rs index 1a26aad320..52fa52f836 100644 --- a/tools/nymvisor/src/upgrades/mod.rs +++ b/tools/nymvisor/src/upgrades/mod.rs @@ -1,6 +1,7 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::config::GENESIS_DIR; use crate::error::NymvisorError; use nym_bin_common::build_information::BinaryBuildInformationOwned; use serde::{Deserialize, Serialize}; @@ -10,6 +11,7 @@ use std::fs::OpenOptions; use std::path::{Path, PathBuf}; use std::{fs, io}; use time::OffsetDateTime; +use tracing::error; use url::Url; mod http_upstream; @@ -22,9 +24,9 @@ pub struct UpgradePlan { #[serde(skip)] _save_path: Option, - pub current: UpgradeInfo, + current: UpgradeInfo, - pub next: VecDeque, + next: VecDeque, } impl UpgradePlan { @@ -73,10 +75,27 @@ impl UpgradePlan { self.update_on_disk() } + pub(crate) fn current(&self) -> &UpgradeInfo { + &self.current + } + pub(crate) fn next_upgrade(&self) -> Option<&UpgradeInfo> { self.next.front() } + pub(crate) fn has_planned(&self, upgrade: &UpgradeInfo) -> bool { + for planned in &self.next { + if planned.version == upgrade.version { + if planned.name != upgrade.name { + // TODO: should we maybe return a hard error here instead? + error!("we have already a planned upgrade for version {} under name '{}' which differs from provided '{}'", planned.version, planned.name, upgrade.name); + } + return true; + } + } + false + } + // pub(crate) fn update_current(&mut self) -> Result<(), NymvisorError> { // // } @@ -171,6 +190,18 @@ pub struct UpgradeInfo { impl UpgradeInfo { pub(crate) fn save>(&self, path: P) -> Result<(), NymvisorError> { let path = path.as_ref(); + + // in case we're saving brand new upgrade info, make sure the parent directory exists + #[allow(clippy::expect_used)] + let parent = path + .parent() + .expect("attempted to save the upgrade info as the root of the fs"); + + fs::create_dir_all(parent).map_err(|source| NymvisorError::PathInitFailure { + path: parent.to_path_buf(), + source, + })?; + let file = OpenOptions::new() .create(true) .write(true) @@ -201,6 +232,10 @@ impl UpgradeInfo { source, }) } + + pub(crate) fn is_genesis(&self) -> bool { + self.name == GENESIS_DIR + } } #[derive(Serialize, Deserialize, Debug)]