main run loop

tools/nymvisor/src/cli/init.rs

@@ -7,13 +7,13 @@ use crate::env::Env;
 use crate::error::NymvisorError;
 use crate::upgrades::types::{UpgradeInfo, UpgradePlan};
 use nym_bin_common::build_information::BinaryBuildInformationOwned;
-use nym_bin_common::logging::{setup_logging, setup_tracing_logger};
+use nym_bin_common::logging::setup_tracing_logger;
 use nym_bin_common::output_format::OutputFormat;
 use std::fs;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 use time::OffsetDateTime;
-use tracing::{debug, error, info, trace, warn};
+use tracing::{debug, info, trace, warn};
 use url::Url;
 
 #[derive(clap::Args, Debug)]

tools/nymvisor/src/cli/mod.rs

@@ -13,7 +13,7 @@ use crate::error::NymvisorError;
 use clap::{Parser, Subcommand};
 use lazy_static::lazy_static;
 use nym_bin_common::bin_info;
-use std::path::{Path, PathBuf};
+use std::path::Path;
 use tracing::error;
 
 lazy_static! {

tools/nymvisor/src/cli/run.rs

@@ -2,20 +2,14 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::cli::try_load_current_config;
-use crate::daemon::Daemon;
 use crate::env::Env;
 use crate::error::NymvisorError;
-use crate::upgrades::upgrade_binary;
-use async_file_watcher::AsyncFileWatcher;
-use futures::channel::mpsc;
-use futures::future::{AbortHandle, Abortable};
-use futures::StreamExt;
+use crate::tasks::launcher::DaemonLauncher;
+use crate::tasks::upgrade_plan_watcher::start_upgrade_plan_watcher;
+use crate::tasks::upstream_poller::UpstreamPoller;
 use nym_bin_common::logging::setup_tracing_logger;
-use std::sync::Arc;
-use std::time::Duration;
 use tokio::runtime;
-use tokio::sync::Notify;
-use tracing::info;
+use tracing::{error, info};
 
 #[derive(clap::Args, Debug)]
 pub(crate) struct Args {
@@ -33,8 +27,6 @@ pub(crate) fn execute(args: Args) -> Result<(), NymvisorError> {
 
     info!("starting nymvisor for {}", config.daemon.name);
 
-    // upgrade_binary(&config)?;
-
     // TODO: experiment with the minimal runtime
     // look at futures::executor::LocalPool
     // well, if the creation of the runtime failed, there isn't much we could do
@@ -50,40 +42,31 @@ pub(crate) fn execute(args: Args) -> Result<(), NymvisorError> {
     // - 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 to terminate the nymvisor
 
-    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);
-    //
-    //     <Result<_, NymvisorError>>::Ok(())
-    // })
+    // spawn the root task
+    rt.block_on(async {
+        let (upgrade_receiver, watcher_handle) = start_upgrade_plan_watcher(&config)?;
+        let upstream_poller_handle = UpstreamPoller::new(&config).start();
+        let mut launcher = DaemonLauncher::new(config, upgrade_receiver);
+
+        if let Err(err) = launcher.run_loop(args.daemon_args).await {
+            error!("the daemon could not continue running: {err}");
+        } else {
+            info!("the daemon has finished execution");
+        }
+
+        if !watcher_handle.is_finished() {
+            watcher_handle.abort();
+        }
+
+        if !upstream_poller_handle.is_finished() {
+            upstream_poller_handle.abort();
+        }
+
+        // TODO: add timeouts and error handling here
+        // TODO2: maybe we need to make those fuse futures?
+        watcher_handle.await;
+        upstream_poller_handle.await;
+
+        Ok(())
+    })
 }

tools/nymvisor/src/config/mod.rs

@@ -476,6 +476,8 @@ pub struct DaemonDebug {
     /// default: 10s
     /// Can be overridden with $DAEMON_SHUTDOWN_GRACE_PERIOD environmental variable.
     #[serde(with = "humantime_serde")]
+    // this is not deprecated, im just marking it as such so that clippy would yell at me because I still havent implemented it
+    #[deprecated]
     pub shutdown_grace_period: Duration,
 
     /// Set custom backup directory for daemon data. If not set, the daemon's home directory will be used instead.
@@ -486,6 +488,8 @@ pub struct DaemonDebug {
     /// If enabled, `nymvisor` will perform upgrades directly without performing any backups.
     /// default: false
     /// Can be overridden with $DAEMON_UNSAFE_SKIP_BACKUP environmental variable.
+    // this is not deprecated, im just marking it as such so that clippy would yell at me because I still havent implemented it
+    #[deprecated]
     pub unsafe_skip_backup: bool,
 }
 

tools/nymvisor/src/error.rs

@@ -309,6 +309,12 @@ pub(crate) enum NymvisorError {
         #[source]
         source: io::Error,
     },
+
+    #[error("the daemon has reached the maximum number of startup failures ({failures})")]
+    DaemonMaximumStartupFailures { failures: usize },
+
+    #[error("the daemon restart on failure is disabled")]
+    DisabledRestartOnFailure,
 }
 
 impl From<ExitStatus> for NymvisorError {

tools/nymvisor/src/tasks/launcher/backup.rs

@@ -83,6 +83,15 @@ impl BackupBuilder {
         }
     }
 
+    fn finish(mut self) -> Result<(), NymvisorError> {
+        self.tar_builder
+            .finish()
+            .map_err(|source| NymvisorError::BackupTarFinalizationFailure {
+                path: self.backup_filepath,
+                source,
+            })
+    }
+
     pub(crate) fn backup_daemon_home<P: AsRef<Path>>(
         mut self,
         daemon_home: P,
@@ -105,13 +114,4 @@ impl BackupBuilder {
         }
         self.finish()
     }
-
-    fn finish(mut self) -> Result<(), NymvisorError> {
-        self.tar_builder
-            .finish()
-            .map_err(|source| NymvisorError::BackupTarFinalizationFailure {
-                path: self.backup_filepath,
-                source,
-            })
-    }
 }

tools/nymvisor/src/tasks/launcher/mod.rs

@@ -27,13 +27,67 @@ pub(crate) struct DaemonLauncher {
 }
 
 impl DaemonLauncher {
-    pub(crate) fn new(config: Config) -> Self {
-        todo!()
+    pub(crate) fn new(config: Config, upgrade_plan_watcher: FileWatcherEventReceiver) -> Self {
+        DaemonLauncher {
+            config,
+            upgrade_plan_watcher,
+        }
+    }
+
+    pub(crate) async fn run_loop(&mut self, args: Vec<String>) -> Result<(), NymvisorError> {
+        let mut consecutive_startup_failure_count = 0;
+        loop {
+            let run_start = tokio::time::Instant::now();
+
+            let res = self.run_and_upgrade(args.clone()).await;
+            let run_duration = run_start.elapsed();
+            info!(
+                "the daemon has run for {}",
+                humantime::format_duration(run_duration)
+            );
+
+            match res {
+                Ok(upgraded) => {
+                    if upgraded {
+                        if !self.config.daemon.debug.restart_after_upgrade {
+                            info!("upgrade detected, DAEMON_RESTART_AFTER_UPGRADE is off. Verify new upgrade and start nymvisor again");
+                            return Ok(());
+                        }
+                        // restart
+                    } else {
+                        return Ok(());
+                    }
+                }
+                Err(failure) => {
+                    error!("daemon failed with the following error: {failure}");
+
+                    if !self.config.daemon.debug.restart_on_failure {
+                        return Err(NymvisorError::DisabledRestartOnFailure);
+                    }
+
+                    if run_duration < self.config.daemon.debug.startup_period_duration {
+                        consecutive_startup_failure_count += 1;
+                    } else {
+                        consecutive_startup_failure_count = 1;
+                    }
+
+                    if consecutive_startup_failure_count
+                        >= self.config.daemon.debug.max_startup_failures
+                    {
+                        return Err(NymvisorError::DaemonMaximumStartupFailures {
+                            failures: consecutive_startup_failure_count,
+                        });
+                    }
+
+                    sleep(self.config.daemon.debug.failure_restart_delay).await;
+                }
+            }
+        }
     }
 
     /// the full upgrade process process, i.e. run until upgrade, do backup and perform the upgrade.
-    /// returns a boolean indicating whether the process should get restarted
-    pub(crate) async fn run(&mut self, args: Vec<String>) -> Result<bool, NymvisorError> {
+    /// returns a boolean indicating whether an upgrade has been performed
+    async fn run_and_upgrade(&mut self, args: Vec<String>) -> Result<bool, NymvisorError> {
         let upgrade_available = self.wait_for_upgrade_or_termination(args.clone()).await?;
         if !upgrade_available {
             return Ok(false);

tools/nymvisor/src/tasks/upgrade_plan_watcher.rs

@@ -6,6 +6,7 @@ use crate::error::NymvisorError;
 use async_file_watcher::{AsyncFileWatcher, FileWatcherEventReceiver, NotifyResult};
 use futures::channel::mpsc;
 use tokio::task::JoinHandle;
+use tracing::warn;
 
 pub(crate) fn start_upgrade_plan_watcher(
     config: &Config,
@@ -14,7 +15,11 @@ pub(crate) fn start_upgrade_plan_watcher(
     let mut watcher =
         AsyncFileWatcher::new_file_changes_watcher(config.upgrade_plan_filepath(), events_sender)?;
 
-    let join_handle = tokio::spawn(async move { watcher.watch().await });
+    let join_handle = tokio::spawn(async move {
+        let res = watcher.watch().await;
+        warn!("the upgrade plan watcher has stopped");
+        res
+    });
 
     Ok((events_receiver, join_handle))
 }

tools/nymvisor/src/tasks/upstream_poller.rs

@@ -70,7 +70,7 @@ impl UpstreamPoller {
         }
     }
 
-    pub(crate) async fn start(mut self) -> JoinHandle<()> {
+    pub(crate) fn start(mut self) -> JoinHandle<()> {
         tokio::spawn(async move { self.run().await })
     }
 }

tools/nymvisor/src/upgrades/mod.rs

@@ -25,6 +25,11 @@ pub(crate) async fn upgrade_binary(config: &Config) -> Result<(), NymvisorError>
         return Err(NymvisorError::NoQueuedUpgrades);
     };
 
+    if next.manual {
+        let unused_variable = 42;
+        todo!()
+    }
+
     let upgrade_name = next.name.clone();
 
     let history_path = config.upgrade_history_filepath();
@@ -64,6 +69,9 @@ pub(crate) async fn upgrade_binary(config: &Config) -> Result<(), NymvisorError>
             upgrade_binary_path.display()
         );
         download_upgrade_binary(config, &next).await?;
+
+        let unused_variable = 42;
+        // TODO: checksum, etc.
     }
 
     let tmp_daemon = Daemon::new(upgrade_binary_path);