Merge pull request dmnd-pool#122 from jbesraa/2025-08-13/shares-monitor-fixes

Monitoring and other fixes before release

Author: jbesraa

src/config.rs

@@ -5,6 +5,7 @@ use serde_json::json;
 use std::{
     net::{SocketAddr, ToSocketAddrs},
     path::PathBuf,
+    time::Duration,
 };
 use tracing::{debug, error, info, warn};
 
@@ -337,6 +338,7 @@ impl Configuration {
 
 /// Parses a hashrate string (e.g., "10T", "2.5P", "500E") into an f32 value in h/s.
 fn parse_hashrate(hashrate_str: &str) -> Result<f32, String> {
+    info!("Received hashrate: '{}'", hashrate_str);
     let hashrate_str = hashrate_str.trim();
     if hashrate_str.is_empty() {
         return Err("Hashrate cannot be empty. Expected format: '<number><unit>' (e.g., '10T', '2.5P', '5E'".to_string());
@@ -364,7 +366,7 @@ fn parse_hashrate(hashrate_str: &str) -> Result<f32, String> {
     if hashrate.is_infinite() || hashrate.is_nan() {
         return Err("Hashrate too large or invalid".to_string());
     }
-
+    info!("Parsed hashrate: {} h/s", hashrate);
     Ok(hashrate)
 }
 
@@ -387,37 +389,41 @@ fn parse_address(addr: String) -> Option<SocketAddr> {
 /// Fetches pool URLs from the server based on the environment.
 async fn fetch_pool_urls() -> Result<Vec<SocketAddr>, Error> {
     if CONFIG.local {
+        info!("Running in local mode, using hardcoded address 127.0.0.1:20000");
         return Ok(vec![
             parse_address("127.0.0.1:20000".to_string()).expect("Invalid local address")
         ]);
     };
-
     let url = if CONFIG.staging {
-        info!("Fetching pool URLs from staging server: {}", STAGING_URL);
         STAGING_URL
     } else if CONFIG.testnet3 {
-        info!("Fetching pool URLs from testnet server: {}", TESTNET3_URL);
         TESTNET3_URL
     } else {
-        info!(
-            "Fetching pool URLs from production server: {}",
-            PRODUCTION_URL
-        );
         PRODUCTION_URL
     };
     let endpoint = format!("{}/api/pool/urls", url);
+    info!("Fetching pool URLs from: {}", endpoint);
     let token = Configuration::token().expect("TOKEN is not set");
+    let mut retries = 8;
+    let client = reqwest::Client::new();
 
-    let response = match reqwest::Client::new()
-        .post(endpoint)
-        .json(&json!({"token": token}))
-        .send()
-        .await
-    {
-        Ok(resp) => resp,
-        Err(e) => {
-            error!("Failed to fetch pool urls: {}", e);
-            return Err(Error::from(e));
+    let response = loop {
+        let request = client
+            .post(endpoint.clone())
+            .json(&json!({"token": token}))
+            .timeout(Duration::from_secs(15));
+
+        match request.send().await {
+            Ok(resp) => break resp,
+            Err(e) => {
+                error!("Failed to fetch pool urls: {}", e);
+                if retries == 0 {
+                    return Err(Error::from(e));
+                }
+                retries -= 1;
+                info!("Retrying in 3 seconds...");
+                tokio::time::sleep(Duration::from_secs(3)).await;
+            }
         }
     };
 
@@ -436,9 +442,10 @@ async fn fetch_pool_urls() -> Result<Vec<SocketAddr>, Error> {
         .filter_map(|addr| {
             let address = format!("{}:{}", addr.host, addr.port);
             parse_address(address)
-        }) // Filter out any None values
+        }) // Filter out any None values, i.e., invalid addresses
         .collect();
-    debug!("Pool addresses: {:?}", socket_addrs);
+    info!("Found {} pool addresses", socket_addrs.len());
+    info!("Pool addresses: {:?}", &socket_addrs);
     Ok(socket_addrs)
 }
 

src/main.rs

@@ -36,7 +36,7 @@ const DEFAULT_SV1_HASHPOWER: f32 = 100_000_000_000_000.0;
 const SHARE_PER_MIN: f32 = 10.0;
 const CHANNEL_DIFF_UPDTATE_INTERVAL: u32 = 10;
 const MAX_LEN_DOWN_MSG: u32 = 10000;
-const MAIN_AUTH_PUB_KEY: &str = "9bQHWXsQ2J9TRFTaxRh3KjoxdyLRfWVEy25YHtKF8y8gotLoCZZ";
+const MAIN_AUTH_PUB_KEY: &str = "9c44K6QVizyPWb9xfeqhckFRosxWwB3EfytGa4CfTdD526qb2QV";
 const TEST_AUTH_PUB_KEY: &str = "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72";
 const DEFAULT_LISTEN_ADDRESS: &str = "0.0.0.0:32767";
 const REPO_OWNER: &str = "demand-open-source";
@@ -147,7 +147,6 @@ async fn initialize_proxy(
                     );
                     let secs = 5;
                     tokio::time::sleep(Duration::from_secs(secs)).await;
-                    // Restart loop, esentially restarting proxy
                     continue;
                 }
             };

src/monitor/shares.rs

@@ -1,7 +1,6 @@
 use roles_logic_sv2::utils::Mutex;
 use std::sync::Arc;
-use tracing::{debug, error};
-const BATCH_SIZE: u32 = 20; // Default batch size for sending shares
+use tracing::{error, info, warn};
 
 use crate::{
     monitor::{shares_server_endpoint, MonitorAPI},
@@ -13,7 +12,8 @@ pub struct ShareInfo {
     worker_name: String,
     difficulty: Option<f32>,
     job_id: i64,
-    rejection_reason: Option<RejectionReason>, // if None, the share was accepted
+    // if None, the share was accepted
+    rejection_reason: Option<RejectionReason>,
     timestamp: u64,
 }
 
@@ -39,21 +39,19 @@ impl ShareInfo {
 
 #[derive(Debug, Clone)]
 pub struct SharesMonitor {
-    pending_shares: Arc<Mutex<Vec<ShareInfo>>>,
-    batch_size: u32,
+    shares: Arc<Mutex<Vec<ShareInfo>>>,
 }
 
 impl SharesMonitor {
     pub fn new() -> Self {
         SharesMonitor {
-            pending_shares: Arc::new(Mutex::new(Vec::new())),
-            batch_size: BATCH_SIZE,
+            shares: Arc::new(Mutex::new(Vec::new())),
         }
     }
 
     /// Inserts a new share into the pending shares list.
     pub fn insert_share(&self, share: ShareInfo) {
-        self.pending_shares
+        self.shares
             .safe_lock(|event| {
                 event.push(share);
             })
@@ -64,8 +62,8 @@ impl SharesMonitor {
     }
 
     /// Retrieves the list of pending shares.
-    fn get_pending_shares(&self) -> Vec<ShareInfo> {
-        self.pending_shares
+    fn get_next_shares(&self) -> Vec<ShareInfo> {
+        self.shares
             .safe_lock(|event| event.clone())
             .unwrap_or_else(|e| {
                 error!("Failed to lock pending shares: {:?}", e);
@@ -75,8 +73,8 @@ impl SharesMonitor {
     }
 
     /// Clears the list of pending shares.
-    fn clear_pending_shares(&self) {
-        self.pending_shares
+    fn clear_next_shares(&self) {
+        self.shares
             .safe_lock(|event| {
                 event.clear();
             })
@@ -86,34 +84,29 @@ impl SharesMonitor {
             });
     }
 
-    /// Monitors the pending shares and sends them to the monitoring server in batches.
     pub async fn monitor(&self) {
         let api = MonitorAPI::new(shares_server_endpoint());
-        let mut interval = tokio::time::interval(std::time::Duration::from_secs(60)); // Check every 60 seconds
-        interval.tick().await; // Skip the first tick to avoid unnecessary error log
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
+        // First tick completes immediately
+        interval.tick().await;
         loop {
             interval.tick().await;
-            let shares_to_send = self.get_pending_shares();
+            let shares_to_send = self.get_next_shares();
             if !shares_to_send.is_empty() {
-                if shares_to_send.len() >= self.batch_size as usize {
-                    match api.send_shares(shares_to_send.clone()).await {
-                        Ok(_) => {
-                            debug!("Successfully sent Shares: {:?} to API", &shares_to_send);
-                        }
-                        Err(err) => {
-                            error!("Failed to send shares: {}", err);
-                        }
+                match api.send_shares(shares_to_send.clone()).await {
+                    Ok(_) => {
+                        info!(
+                            "Saved {} shares to the monitoring server",
+                            shares_to_send.len()
+                        );
+                        self.clear_next_shares();
+                    }
+                    Err(err) => {
+                        warn!("Failed to send shares, this does not affect mining but may cause issues with monitoring: {:?}", err);
                     }
-                    self.clear_pending_shares(); // Clear after sending
-                } else {
-                    debug!(
-                        "Current shares count ({}) is less than batch size ({}), waiting for more",
-                        shares_to_send.len(),
-                        self.batch_size
-                    );
                 }
             } else {
-                error!("No pending shares to send");
+                warn!("No pending shares to send. If this happens frequently, check your miner.");
             }
         }
     }

src/router/mod.rs

@@ -196,7 +196,7 @@ impl Router {
         let auth_pub_key = self.auth_pub_k;
 
         tokio::time::timeout(
-            Duration::from_secs(15),
+            Duration::from_secs(8),
             PoolLatency::get_mining_setup_latencies(
                 &mut pool,
                 setup_connection_msg.cloned(),
@@ -212,21 +212,6 @@ impl Router {
             );
         })??;
 
-        if (PoolLatency::get_mining_setup_latencies(
-            &mut pool,
-            setup_connection_msg.cloned(),
-            timer.cloned(),
-            auth_pub_key,
-        )
-        .await)
-            .is_err()
-        {
-            error!(
-                "Failed to get mining setup latencies for: {:?}",
-                pool_address
-            );
-            return Err(());
-        }
         if (PoolLatency::get_jd_latencies(&mut pool, auth_pub_key).await).is_err() {
             error!("Failed to get jd setup latencies for: {:?}", pool_address);
             return Err(());