send workers shares

Priceless-P · Priceless-P · commit a00835a9c245 · 2025-07-21T14:36:19.000+01:00
diff --git a/src/main.rs b/src/main.rs
@@ -25,6 +25,7 @@ mod proxy_state;
 mod router;
 mod share_accounter;
 mod shared;
+mod shares_monitor;
 mod translator;
 
 const TRANSLATOR_BUFFER_SIZE: usize = 32;
diff --git a/src/shared/error.rs b/src/shared/error.rs
@@ -1,5 +1,25 @@
+use std::fmt;
+
 pub enum Sv1IngressError {
     TranslatorDropped,
     DownstreamDropped,
     TaskFailed,
 }
+
+#[derive(Debug)]
+pub enum Error {
+    ReqwestError(reqwest::Error),
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            Error::ReqwestError(e) => write!(f, "ReqwestError: {}", e),
+        }
+    }
+}
+impl From<reqwest::Error> for Error {
+    fn from(err: reqwest::Error) -> Self {
+        Error::ReqwestError(err)
+    }
+}
diff --git a/src/shares_monitor.rs b/src/shares_monitor.rs
@@ -0,0 +1,192 @@
+use reqwest::Url;
+use roles_logic_sv2::utils::Mutex;
+use serde_json::json;
+use std::sync::Arc;
+use tracing::{debug, error};
+const BATCH_SIZE: u32 = 20; // Default batch size for sending shares
+
+use crate::{
+    config::Configuration,
+    proxy_state::{DownstreamType, ProxyState},
+    shared::error::Error,
+};
+
+const STAGING_URL: &str = "http://staging-user-dashboard-server.dmnd.work/share/save";
+const LOCAL_URL: &str = "http://localhost:8787/api/share/save";
+const PRODUCTION_URL: &str = "http://user-dashboard-server.dmnd.work/share/save";
+
+fn monitoring_server_url() -> String {
+    // Determine the monitoring server URL based on the environment
+    match Configuration::environment().as_str() {
+        "staging" => STAGING_URL.to_string(),
+        "local" => LOCAL_URL.to_string(),
+        "production" => PRODUCTION_URL.to_string(),
+        _ => unreachable!(),
+    }
+}
+
+#[derive(serde::Serialize, Clone, Debug)]
+pub struct ShareInfo {
+    worker_name: String,
+    difficulty: Option<f32>,
+    job_id: i64,
+    rejection_reason: Option<RejectionReason>, // if None, the share was accepted
+    timestamp: u64,
+}
+
+impl ShareInfo {
+    pub fn new(
+        worker_name: String,
+        difficulty: Option<f32>,
+        job_id: i64,
+        rejection_reason: Option<RejectionReason>,
+    ) -> Self {
+        ShareInfo {
+            worker_name,
+            difficulty,
+            job_id,
+            rejection_reason,
+            timestamp: std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .expect("Time went backwards")
+                .as_secs(),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct SharesMonitor {
+    pending_shares: Arc<Mutex<Vec<ShareInfo>>>,
+    batch_size: u32,
+}
+
+impl SharesMonitor {
+    pub fn new() -> Self {
+        SharesMonitor {
+            pending_shares: Arc::new(Mutex::new(Vec::new())),
+            batch_size: BATCH_SIZE,
+        }
+    }
+
+    /// Inserts a new share into the pending shares list.
+    pub fn insert_share(&self, share: ShareInfo) {
+        self.pending_shares
+            .safe_lock(|event| {
+                event.push(share);
+            })
+            .unwrap_or_else(|e| {
+                error!("Failed to lock pending shares: {:?}", e);
+                ProxyState::update_downstream_state(DownstreamType::TranslatorDownstream);
+            });
+    }
+
+    /// Retrieves the list of pending shares.
+    fn get_pending_shares(&self) -> Vec<ShareInfo> {
+        self.pending_shares
+            .safe_lock(|event| event.clone())
+            .unwrap_or_else(|e| {
+                error!("Failed to lock pending shares: {:?}", e);
+                ProxyState::update_downstream_state(DownstreamType::TranslatorDownstream);
+                Vec::new()
+            })
+    }
+
+    /// Clears the list of pending shares.
+    fn clear_pending_shares(&self) {
+        self.pending_shares
+            .safe_lock(|event| {
+                event.clear();
+            })
+            .unwrap_or_else(|e| {
+                error!("Failed to lock pending shares: {:?}", e);
+                ProxyState::update_downstream_state(DownstreamType::TranslatorDownstream);
+            });
+    }
+
+    /// Monitors the pending shares and sends them to the monitoring server in batches.
+    pub async fn monitor(&self) {
+        let api = MonitorAPI::new(monitoring_server_url());
+        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
+        loop {
+            interval.tick().await;
+            let shares_to_send = self.get_pending_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);
+                        }
+                    }
+                    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");
+            }
+        }
+    }
+}
+
+struct MonitorAPI {
+    url: Url,
+    client: reqwest::Client,
+}
+
+impl MonitorAPI {
+    fn new(url: String) -> Self {
+        let client = reqwest::Client::new();
+        MonitorAPI {
+            url: url.parse().expect("Invalid URL"),
+            client,
+        }
+    }
+
+    /// Sends a batch of shares to the monitoring server.
+    async fn send_shares(&self, shares: Vec<ShareInfo>) -> Result<(), Error> {
+        let token = crate::config::Configuration::token().expect("Token is not set");
+
+        debug!("Sending batch of {} shares to API", shares.len());
+        let response = self
+            .client
+            .post(self.url.clone())
+            .json(&json!({ "shares": shares, "token": token }))
+            .send()
+            .await?;
+
+        match response.error_for_status() {
+            Ok(_) => Ok(()),
+            Err(err) => {
+                error!("Failed to send shares: {}", err);
+                Err(err.into())
+            }
+        }
+    }
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub enum RejectionReason {
+    JobIdNotFound,
+    InvalidShare,
+    InvalidJobIdFormat,
+    DifficultyMismatch,
+}
+
+impl std::fmt::Display for RejectionReason {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RejectionReason::JobIdNotFound => write!(f, "Job ID not found"),
+            RejectionReason::InvalidShare => write!(f, "Invalid share"),
+            RejectionReason::InvalidJobIdFormat => write!(f, "Invalid job ID format"),
+            RejectionReason::DifficultyMismatch => write!(f, "Difficulty mismatch"),
+        }
+    }
+}
diff --git a/src/translator/downstream/downstream.rs b/src/translator/downstream/downstream.rs
@@ -2,6 +2,7 @@ use crate::{
     api::stats::StatsSender,
     proxy_state::{DownstreamType, ProxyState},
     shared::utils::AbortOnDrop,
+    shares_monitor::{RejectionReason, ShareInfo, SharesMonitor},
     translator::{error::Error, utils::validate_share},
 };
 
@@ -115,6 +116,7 @@ pub struct Downstream {
     pub(super) stats_sender: StatsSender,
     pub recent_jobs: RecentJobs,
     pub first_job: Notify<'static>,
+    pub share_monitor: SharesMonitor,
 }
 
 impl Downstream {
@@ -194,6 +196,7 @@ impl Downstream {
             stats_sender,
             recent_jobs: RecentJobs::new(),
             first_job: last_notify.expect("we have an assertion at the beginning of this function"),
+            share_monitor: SharesMonitor::new(),
         }));
 
         if let Err(e) = start_receive_downstream(
@@ -233,6 +236,31 @@ impl Downstream {
             error!("Failed to start notify task: {e}");
             ProxyState::update_downstream_state(DownstreamType::TranslatorDownstream);
         };
+
+        if let Err(e) = Self::start_share_monitor(task_manager.clone(), downstream.clone()).await {
+            error!("Failed to start share monitor task: {e}");
+            ProxyState::update_downstream_state(DownstreamType::TranslatorDownstream);
+        }
+    }
+
+    /// Starts the shares monitor task.
+    async fn start_share_monitor(
+        task_manager: Arc<Mutex<TaskManager>>,
+        downstream: Arc<Mutex<Self>>,
+    ) -> Result<(), Error<'static>> {
+        let (share_monitor, connection_id) =
+            downstream.safe_lock(|s| (s.share_monitor.clone(), s.connection_id))?;
+
+        // Create an abortable task for the shares monitor
+        let abortable = tokio::spawn(async move {
+            info!("Starting shares monitor for downstream: {}", connection_id);
+            share_monitor.clone().monitor().await;
+        });
+
+        // Register the task with the task manager so it can be aborted when needed
+        TaskManager::add_shares_monitor(task_manager, abortable.into())
+            .await
+            .map_err(|_| Error::TranslatorTaskManagerFailed)
     }
 
     /// Accept connections from one or more SV1 Downstream roles (SV1 Mining Devices) and create a
@@ -353,6 +381,8 @@ impl Downstream {
         stats_sender: StatsSender,
         first_job: Notify<'static>,
     ) -> Self {
+        use crate::shares_monitor::SharesMonitor;
+
         Downstream {
             connection_id,
             authorized_names,
@@ -368,6 +398,7 @@ impl Downstream {
             first_job,
             stats_sender,
             recent_jobs: RecentJobs::new(),
+            share_monitor: SharesMonitor::new(),
         }
     }
 }
@@ -445,9 +476,19 @@ impl IsServer<'static> for Downstream {
                 "Share rejected: can not convert v1 job id to number. v1 id: {}",
                 request.job_id
             );
+
+            let share = ShareInfo::new(
+                request.user_name.clone(),
+                None,
+                job_id_as_number.expect("checked above") as i64,
+                Some(RejectionReason::InvalidJobIdFormat),
+            );
+            self.share_monitor.insert_share(share);
+
             self.stats_sender.update_rejected_shares(self.connection_id);
             return false;
         }
+        let job_id = job_id_as_number.clone().expect("checked above") as i64;
         crate::translator::utils::update_share_count(self.connection_id); // update share count
         if let Some(job) = self
             .recent_jobs
@@ -481,6 +522,23 @@ impl IsServer<'static> for Downstream {
                             // Return false because submit was not properly handled
                             return false;
                         }
+                        // Share is accepted here
+                        let share = ShareInfo::new(
+                            request.user_name.clone(),
+                            Some(met_difficulty),
+                            job_id,
+                            None,
+                        );
+                        self.share_monitor.insert_share(share);
+                    } else {
+                        // met_difficulty is not latest difficulty, so we mark it as rejected
+                        let share = ShareInfo::new(
+                            request.user_name.clone(),
+                            None,
+                            job_id, // rejected because it was not sent upstream
+                            Some(RejectionReason::DifficultyMismatch),
+                        );
+                        self.share_monitor.insert_share(share);
                     }
                 }
                 self.stats_sender.update_accepted_shares(self.connection_id);
@@ -490,11 +548,25 @@ impl IsServer<'static> for Downstream {
                 );
                 true
             } else {
+                let share = ShareInfo::new(
+                    request.user_name.clone(),
+                    None,
+                    job_id,
+                    Some(RejectionReason::InvalidShare),
+                );
+                self.share_monitor.insert_share(share);
                 error!("Share rejected: Invalid share");
                 self.stats_sender.update_rejected_shares(self.connection_id);
                 false
             }
         } else {
+            let event = ShareInfo::new(
+                request.user_name.clone(),
+                None,
+                job_id,
+                Some(RejectionReason::JobIdNotFound),
+            );
+            self.share_monitor.insert_share(event);
             error!(
                 "Share rejected: can not find job with id {}",
                 request.job_id
diff --git a/src/translator/downstream/task_manager.rs b/src/translator/downstream/task_manager.rs
@@ -12,6 +12,7 @@ enum Task {
     SendDownstream(AbortOnDrop),
     Notify(AbortOnDrop),
     Update(AbortOnDrop),
+    SharesMonitor(AbortOnDrop),
 }
 
 pub struct TaskManager {
@@ -86,4 +87,15 @@ impl TaskManager {
             .await
             .map_err(|_| ())
     }
+
+    pub async fn add_shares_monitor(
+        self_: Arc<Mutex<Self>>,
+        abortable: AbortOnDrop,
+    ) -> Result<(), ()> {
+        let send_task = self_.safe_lock(|s| s.send_task.clone()).unwrap();
+        send_task
+            .send(Task::SharesMonitor(abortable))
+            .await
+            .map_err(|_| ())
+    }
 }
diff --git a/src/translator/mod.rs b/src/translator/mod.rs
