diff --git a/Cargo.lock b/Cargo.lock index 7eb3f5ec83..c52a92e7b9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7602,6 +7602,7 @@ dependencies = [ "async-file-watcher", "clap 4.4.7", "dotenvy", + "futures", "humantime 2.1.0", "humantime-serde", "lazy_static", diff --git a/common/async-file-watcher/src/lib.rs b/common/async-file-watcher/src/lib.rs index e2ff23e830..62bb895bcd 100644 --- a/common/async-file-watcher/src/lib.rs +++ b/common/async-file-watcher/src/lib.rs @@ -10,6 +10,8 @@ use std::path::{Path, PathBuf}; use std::time::Duration; use tokio::time::Instant; +pub use notify::{Error as NotifyError, Result as NotifyResult}; + pub type FileWatcherEventSender = mpsc::UnboundedSender; pub type FileWatcherEventReceiver = mpsc::UnboundedReceiver; @@ -22,7 +24,7 @@ pub struct AsyncFileWatcher { last_received: HashMap, tick_duration: Duration, - inner_rx: mpsc::UnboundedReceiver>, + inner_rx: mpsc::UnboundedReceiver>, event_sender: FileWatcherEventSender, } @@ -30,7 +32,7 @@ impl AsyncFileWatcher { pub fn new_file_changes_watcher>( path: P, event_sender: FileWatcherEventSender, - ) -> notify::Result { + ) -> NotifyResult { Self::new( path, event_sender, @@ -48,7 +50,7 @@ impl AsyncFileWatcher { event_sender: FileWatcherEventSender, filters: Option>, tick_duration: Option, - ) -> notify::Result { + ) -> NotifyResult { let watcher_config = Config::default(); let (inner_tx, inner_rx) = mpsc::unbounded(); let watcher = RecommendedWatcher::new( @@ -112,17 +114,17 @@ impl AsyncFileWatcher { false } - fn start_watching(&mut self) -> notify::Result<()> { + fn start_watching(&mut self) -> NotifyResult<()> { self.is_watching = true; self.watcher.watch(&self.path, RecursiveMode::NonRecursive) } - fn stop_watching(&mut self) -> notify::Result<()> { + fn stop_watching(&mut self) -> NotifyResult<()> { self.is_watching = false; self.watcher.unwatch(&self.path) } - pub async fn watch(&mut self) -> notify::Result<()> { + pub async fn watch(&mut self) -> NotifyResult<()> { self.start_watching()?; while let Some(event) = self.inner_rx.next().await { diff --git a/tools/nymvisor/Cargo.toml b/tools/nymvisor/Cargo.toml index dad7c97b8a..5ecc560c91 100644 --- a/tools/nymvisor/Cargo.toml +++ b/tools/nymvisor/Cargo.toml @@ -14,6 +14,7 @@ license.workspace = true anyhow = { workspace = true } clap = { workspace = true, features = ["derive"] } dotenvy = { workspace = true } +futures = { workspace = true } humantime = "2.1.0" humantime-serde = "1.1.1" lazy_static = { workspace = true } diff --git a/tools/nymvisor/src/cli/run.rs b/tools/nymvisor/src/cli/run.rs index d7326eab73..c1d842e86e 100644 --- a/tools/nymvisor/src/cli/run.rs +++ b/tools/nymvisor/src/cli/run.rs @@ -5,6 +5,9 @@ use crate::cli::try_load_current_config; use crate::daemon::Daemon; use crate::env::Env; use crate::error::NymvisorError; +use async_file_watcher::AsyncFileWatcher; +use futures::channel::mpsc; +use futures::StreamExt; use nym_bin_common::logging::setup_tracing_logger; use std::sync::Arc; use std::time::Duration; @@ -41,23 +44,32 @@ pub(crate) fn execute(args: Args) -> Result<(), NymvisorError> { rt.block_on(async { println!("run"); - let daemon = Daemon::from_config(&config).with_kill_timeout(Duration::from_millis(200)); + 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 handle = tokio::spawn(async move { + let handle1 = tokio::spawn(async move { let res = running.await; println!("the process has finished! with {res:?}"); }); - tokio::time::sleep(Duration::from_secs(2)).await; - info!(">>>>>>>>>> NYMVISOR: sending interrupt to the daemon"); + let (events_sender, mut events_receiver) = mpsc::unbounded(); + let mut watcher = AsyncFileWatcher::new_file_changes_watcher( + config.upgrade_plan_filepath(), + events_sender, + )?; + let handle2 = tokio::spawn(async move { watcher.watch().await }); + + let event = events_receiver.next().await; + println!("watcher event: {event:?}"); interrupt_notify.notify_one(); - handle.await; + handle1.await; + handle2.await; // println!("{:?}", status); - Ok(()) + + >::Ok(()) }) } diff --git a/tools/nymvisor/src/config/mod.rs b/tools/nymvisor/src/config/mod.rs index 878c363c59..68828653d4 100644 --- a/tools/nymvisor/src/config/mod.rs +++ b/tools/nymvisor/src/config/mod.rs @@ -22,6 +22,8 @@ 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 UPGRADE_PLAN_FILENAME: &str = "upgrade-plan.json"; +pub(crate) const UPGRADE_INFO_FILENAME: &str = "upgrade-info.json"; pub(crate) const NYMVISOR_DIR: &str = "nymvisor"; pub(crate) const BACKUP_DIR: &str = "backups"; pub(crate) const GENESIS_DIR: &str = "genesis"; @@ -255,6 +257,10 @@ impl Config { pub fn upgrades_dir(&self) -> PathBuf { self.upgrade_data_dir().join(UPGRADES_DIR) } + + pub fn upgrade_plan_filepath(&self) -> PathBuf { + self.upgrade_data_dir().join(UPGRADE_PLAN_FILENAME) + } } #[derive(Debug, Deserialize, PartialEq, Serialize)] diff --git a/tools/nymvisor/src/daemon/mod.rs b/tools/nymvisor/src/daemon/mod.rs index 1064529c8b..3b60618a6f 100644 --- a/tools/nymvisor/src/daemon/mod.rs +++ b/tools/nymvisor/src/daemon/mod.rs @@ -142,10 +142,6 @@ impl ExecutingDaemon { nix::sys::signal::kill(Pid::from_raw(self.child_id), signal) .map_err(|source| NymvisorError::DaemonSignalFailure { signal, source }) } - - // fn foo(&'static mut self) { - // self.interrupted = Some(Box::pin(self._notify.notified())); - // } } impl Future for ExecutingDaemon { diff --git a/tools/nymvisor/src/error.rs b/tools/nymvisor/src/error.rs index 8363ee399b..69a856d2ad 100644 --- a/tools/nymvisor/src/error.rs +++ b/tools/nymvisor/src/error.rs @@ -1,6 +1,7 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use async_file_watcher::NotifyError; use nix::sys::signal::Signal; use nym_bin_common::build_information::BinaryBuildInformationOwned; use std::ffi::OsString; @@ -138,6 +139,12 @@ pub(crate) enum NymvisorError { #[source] source: nix::Error, }, + + #[error("failed to watch for changes in the upgrade-plan.json: {source}")] + UpgradePlanFileWatchFailure { + #[from] + source: NotifyError, + }, } impl From for NymvisorError {