Merge pull request dmnd-pool#74 from Priceless-P/fix/auto-update

Fixes

Author: Fi3

Cargo.lock

@@ -24,7 +24,8 @@ serde = { version = "1.0.219", features = ["derive"] }
 sysinfo = {version = "0.33.1"}
 primitive-types = { version = "0.13.1" }
 toml ={ version = "0.8.22" }
-self_update = {version = "0.42.0"}
+self_update = {version = "0.42.0", features = ["archive-tar"]}
+reqwest ={ version = "0.12.20"}
 #roles_logic_sv2 = "1.2.1"
 #sv1_api = "1.0.1"
 #demand-sv2-connection = "0.0.3"

Cargo.toml

@@ -21,6 +21,8 @@ struct Args {
     loglevel: Option<String>,
     #[clap(long = "nc", short = 'n')]
     noise_connection_log: Option<String>,
+    #[clap(long = "sv1_loglevel")]
+    sv1_loglevel: bool,
     #[clap(long = "delay")]
     delay: Option<u64>,
     #[clap(long = "interval", short = 'i')]
@@ -56,6 +58,7 @@ struct ConfigFile {
     downstream_hashrate: Option<String>,
     loglevel: Option<String>,
     nc_loglevel: Option<String>,
+    sv1_log: Option<bool>,
     test: Option<bool>,
     listening_addr: Option<String>,
     api_server_port: Option<String>,
@@ -73,6 +76,7 @@ pub struct Configuration {
     downstream_hashrate: f32,
     loglevel: String,
     nc_loglevel: String,
+    sv1_log: bool,
     test: bool,
     listening_addr: Option<String>,
     api_server_port: String,
@@ -111,6 +115,7 @@ impl Configuration {
     pub fn downstream_listening_addr() -> Option<String> {
         CONFIG.listening_addr.clone()
     }
+
     pub fn api_server_port() -> String {
         CONFIG.api_server_port.clone()
     }
@@ -140,6 +145,9 @@ impl Configuration {
             }
         }
     }
+    pub fn sv1_ingress_log() -> bool {
+        CONFIG.sv1_log
+    }
 
     pub fn test() -> bool {
         CONFIG.test
@@ -170,6 +178,7 @@ impl Configuration {
                 downstream_hashrate: None,
                 loglevel: None,
                 nc_loglevel: None,
+                sv1_log: None,
                 test: None,
                 listening_addr: None,
                 api_server_port: None,
@@ -302,6 +311,10 @@ impl Configuration {
             .or_else(|| std::env::var("NC_LOGLEVEL").ok())
             .unwrap_or("off".to_string());
 
+        let sv1_log = args.sv1_loglevel
+            || config.sv1_log.unwrap_or(false)
+            || std::env::var("SV1_LOGLEVEL").is_ok();
+
         let test = args.test || config.test.unwrap_or(false) || std::env::var("TEST").is_ok();
 
         let monitor =
@@ -321,6 +334,7 @@ impl Configuration {
             downstream_hashrate,
             loglevel,
             nc_loglevel,
+            sv1_log,
             test,
             listening_addr,
             api_server_port,

src/config.rs

@@ -1,6 +1,9 @@
 use std::net::{IpAddr, SocketAddr};
 
-use crate::shared::{error::Sv1IngressError, utils::AbortOnDrop};
+use crate::{
+    config::Configuration,
+    shared::{error::Sv1IngressError, utils::AbortOnDrop},
+};
 use futures::{
     stream::{SplitSink, SplitStream},
     SinkExt, StreamExt,
@@ -79,6 +82,9 @@ impl Downstream {
     ) -> Sv1IngressError {
         let task = tokio::spawn(async move {
             while let Some(Ok(message)) = recv.next().await {
+                if Configuration::sv1_ingress_log() {
+                    info!("Sending msg to upstream: {}", message);
+                }
                 if send.send(message).await.is_err() {
                     error!("Upstream dropped trying to send");
                     return Sv1IngressError::TranslatorDropped;
@@ -100,6 +106,9 @@ impl Downstream {
         let task = tokio::spawn(async move {
             while let Some(message) = recv.recv().await {
                 let message = message.replace(['\n', '\r'], "");
+                if Configuration::sv1_ingress_log() {
+                    info!("Sending msg to downstream: {}", message);
+                }
                 if send.send(message).await.is_err() {
                     warn!("Downstream dropped while trying to send message down");
                     return Sv1IngressError::DownstreamDropped;

src/ingress/sv1_ingress.rs

@@ -11,7 +11,7 @@ use config::Configuration;
 use key_utils::Secp256k1PublicKey;
 use lazy_static::lazy_static;
 use proxy_state::{PoolState, ProxyState, TpState, TranslatorState};
-use self_update::{backends, cargo_crate_version, Status};
+use self_update::{backends, cargo_crate_version, update::UpdateStatus, TempDir};
 use std::{net::SocketAddr, time::Duration};
 use tokio::sync::mpsc::channel;
 use tracing::{debug, error, info, warn};
@@ -322,6 +322,9 @@ fn check_update_proxy() {
 
     debug!("OS: {}", target_bin);
     debug!("DMND-PROXY version: {}", cargo_crate_version!());
+    let original_path = std::env::current_exe().expect("Failed to get current executable path");
+    let tmp_dir = TempDir::new_in(::std::env::current_dir().expect("Failed to get current dir"))
+        .expect("Failed to create tmp dir");
 
     let updater = match backends::github::Update::configure()
         .repo_owner(REPO_OWNER)
@@ -331,6 +334,7 @@ fn check_update_proxy() {
         .target(target_bin)
         .show_output(false)
         .no_confirm(true)
+        .bin_install_path(tmp_dir.path())
         .build()
     {
         Ok(updater) => updater,
@@ -340,33 +344,75 @@ fn check_update_proxy() {
         }
     };
 
-    match updater.update() {
+    match updater.update_extended() {
         Ok(status) => match status {
-            Status::UpToDate(_) => {
+            UpdateStatus::UpToDate => {
                 info!("Starting latest version of DMND-PROXY.");
             }
-            Status::Updated(version) => {
-                info!("Proxy updated to version {}. Restarting Proxy", version);
-
-                let current_exe =
-                    std::env::current_exe().expect("Failed to get current executable path");
+            UpdateStatus::Updated(release) => {
+                info!(
+                    "Proxy updated to version {}. Restarting Proxy",
+                    release.version
+                );
+                for asset in release.assets {
+                    if asset.name == target_bin {
+                        let bin_name = std::path::PathBuf::from(target_bin);
+                        let new_exe = tmp_dir.path().join(&bin_name);
+                        let mut file =
+                            std::fs::File::create(&new_exe).expect("Failed to create file");
+                        let mut download = self_update::Download::from_url(&asset.download_url);
+                        download.set_header(
+                            reqwest::header::ACCEPT,
+                            reqwest::header::HeaderValue::from_static("application/octet-stream"), // to triggers a redirect to the actual binary.
+                        );
+                        download
+                            .download_to(&mut file)
+                            .expect("Failed to download file");
+                    }
+                }
+                let bin_name = std::path::PathBuf::from(target_bin);
+                let new_exe = tmp_dir.path().join(&bin_name);
+                if let Err(e) = std::fs::rename(&new_exe, &original_path) {
+                    error!(
+                        "Failed to move new binary to {}: {}",
+                        original_path.display(),
+                        e
+                    );
+                    return;
+                }
 
-                // Get original cli rgs
+                let _ = std::fs::remove_dir_all(tmp_dir); // clean up tmp dir
+                                                          // Get original cli rgs
                 let args = std::env::args().skip(1).collect::<Vec<_>>();
 
                 #[cfg(unix)]
                 {
-                    // On Unix-like systems, replace the current process with the new binary
+                    use std::os::unix::fs::PermissionsExt;
                     use std::os::unix::process::CommandExt;
-                    let err = std::process::Command::new(current_exe).args(&args).exec();
+                    // On Unix-like systems, replace the current process with the new binary
+                    if let Err(e) = std::fs::set_permissions(
+                        &original_path,
+                        std::fs::Permissions::from_mode(0o755),
+                    ) {
+                        error!(
+                            "Failed to set executable permissions on {}: {}",
+                            original_path.display(),
+                            e
+                        );
+                        return;
+                    }
+
+                    let err = std::process::Command::new(&original_path)
+                        .args(&args)
+                        .exec();
                     // If exec fails, log the error and exit
                     error!("Failed to exec new binary: {:?}", err);
                     std::process::exit(1);
                 }
                 #[cfg(not(unix))]
                 {
                     // On Windows, spawn the new process and exit the current one
-                    std::process::Command::new(current_exe)
+                    std::process::Command::new(&original_path)
                         .args(&args)
                         .spawn()
                         .expect("Failed to start proxy");

src/main.rs

@@ -19,7 +19,11 @@ impl Downstream {
     pub async fn init_difficulty_management(self_: &Arc<Mutex<Self>>) -> ProxyResult<()> {
         let (diff, stats_sender, connection_id, estimated_hashrate) = self_.safe_lock(|d| {
             (
-                d.difficulty_mgmt.current_difficulty,
+                d.difficulty_mgmt
+                    .current_difficulties
+                    .back()
+                    .copied()
+                    .unwrap_or(d.difficulty_mgmt.initial_difficulty),
                 d.stats_sender.clone(),
                 d.connection_id,
                 d.difficulty_mgmt.estimated_downstream_hash_rate,
@@ -200,16 +204,20 @@ impl Downstream {
             return Err(Error::Unrecoverable);
         }
 
-        let (mut pid, current_difficulty, initial_difficulty) = self_.safe_lock(|d| {
+        let (mut pid, latest_difficulty, initial_difficulty) = self_.safe_lock(|d| {
             (
                 d.difficulty_mgmt.pid_controller,
-                d.difficulty_mgmt.current_difficulty,
+                d.difficulty_mgmt
+                    .current_difficulties
+                    .back()
+                    .copied()
+                    .unwrap_or(d.difficulty_mgmt.initial_difficulty),
                 d.difficulty_mgmt.initial_difficulty,
             )
         })?;
 
         let pid_output = pid.next_control_output(realized_share_per_min).output;
-        let new_difficulty = (current_difficulty + pid_output).max(initial_difficulty * 0.1);
+        let new_difficulty = (latest_difficulty + pid_output).max(initial_difficulty * 0.1);
         let nearest = nearest_power_of_10(new_difficulty);
         if nearest != initial_difficulty {
             let mut pid: Pid<f32> = Pid::new(crate::SHARE_PER_MIN, nearest * 10.0);
@@ -229,7 +237,7 @@ impl Downstream {
         } else {
             // TODO check if we can improve stale share with a threshold here
             let threshold = 0.0;
-            let change = (new_difficulty - current_difficulty).abs() / current_difficulty;
+            let change = (new_difficulty - latest_difficulty).abs() / latest_difficulty;
             if change > threshold {
                 let new_estimation =
                     Self::estimate_hash_rate_from_difficulty(new_difficulty, crate::SHARE_PER_MIN);
@@ -259,7 +267,7 @@ impl Downstream {
                 let old_estimation = d.difficulty_mgmt.estimated_downstream_hash_rate;
                 d.difficulty_mgmt.estimated_downstream_hash_rate = new_estimation;
                 d.difficulty_mgmt.reset();
-                d.difficulty_mgmt.current_difficulty = current_diff;
+                d.difficulty_mgmt.add_difficulty(current_diff);
 
                 (
                     d.upstream_difficulty_config.clone(),
@@ -417,10 +425,12 @@ mod test {
     }
 
     async fn test_converge_to_spm(start_hashrate: f64) {
+        let mut diff = VecDeque::new();
+        diff.push_back(10_000_000_000.0);
         let downstream_conf = DownstreamDifficultyConfig {
             estimated_downstream_hash_rate: 0.0, // updated below
             pid_controller: Pid::new(10.0, 100_000_000.0),
-            current_difficulty: 10_000_000_000.0,
+            current_difficulties: diff,
             submits: VecDeque::new(),
             initial_difficulty: 10_000_000_000.0,
         };
@@ -466,8 +476,10 @@ mod test {
             mock_mine(initial_target.clone().into(), &mut share);
             Downstream::save_share(downstream.clone()).unwrap();
             let _ = Downstream::try_update_difficulty_settings(&downstream).await;
-            initial_target =
-                Downstream::difficulty_to_target(downstream_conf.current_difficulty).into();
+            initial_target = Downstream::difficulty_to_target(
+                *downstream_conf.current_difficulties.back().unwrap(),
+            )
+            .into();
             elapsed = timer.elapsed();
         }
         let expected_0s = trailing_0s(expected_target.inner_as_ref().to_vec());