Merge pull request dmnd-pool#89 from Fi3/RemoveRecentNotifies

Remove Downstream::recent_notifies

Author: Fi3

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "demand-cli"
-version = "0.1.11"
+version = "0.1.12"
 edition = "2021"
 
 [dependencies]

Cargo.toml

@@ -111,12 +111,10 @@ impl Downstream {
 
         // Get the last notify
         let recent_notify = self_
-            .safe_lock(|d| d.recent_notifies.back().cloned())
+            .safe_lock(|d| d.recent_jobs.clone_last())
             .map_err(|_| Error::TranslatorDiffConfigMutexPoisoned)?;
 
-        if let Some(mut notify) = recent_notify {
-            let id = self_.safe_lock(|d| d.job_ids.new_v1(notify.job_id.parse::<u32>().unwrap())).unwrap();
-            notify.job_id = id.to_string();
+        if let Some(notify) = recent_notify {
             Downstream::send_message_downstream(self_.clone(), notify.into()).await;
         }
 

src/translator/downstream/diff_management.rs

@@ -28,14 +28,19 @@ use roles_logic_sv2::{
     utils::Mutex,
 };
 
-use std::{collections::{hash_map::Entry, HashMap, VecDeque}, net::IpAddr, sync::Arc};
+use rand::Rng;
+use server_to_client::Notify;
+use std::{
+    collections::{hash_map::Entry, HashMap, VecDeque},
+    net::IpAddr,
+    sync::Arc,
+};
 use sv1_api::{
     client_to_server, json_rpc, server_to_client,
     utils::{Extranonce, HexU32Be},
     IsServer,
 };
 use tracing::{error, info, warn};
-use rand::Rng;
 
 #[derive(Debug, Clone)]
 pub struct DownstreamDifficultyConfig {
@@ -106,15 +111,13 @@ pub struct Downstream {
     /// translation into a SV2 `SubmitSharesExtended`.
     tx_sv1_bridge: Sender<DownstreamMessages>,
     tx_outgoing: Sender<json_rpc::Message>,
-    /// True if this is the first job received from `Upstream`.
-    pub(super) first_job_received: bool,
     extranonce2_len: usize,
     pub(super) difficulty_mgmt: DownstreamDifficultyConfig,
     pub(super) upstream_difficulty_config: Arc<Mutex<UpstreamDifficultyConfig>>,
     pub last_call_to_update_hr: u128,
-    pub(super) recent_notifies: VecDeque<server_to_client::Notify<'static>>,
     pub(super) stats_sender: StatsSender,
-    pub job_ids: JobIds,
+    pub recent_jobs: RecentJobs,
+    pub first_job: Notify<'static>,
 }
 
 impl Downstream {
@@ -179,11 +182,6 @@ impl Downstream {
             initial_difficulty,
         };
 
-        let mut recent_notifies = VecDeque::with_capacity(2);
-        if let Some(notify) = last_notify.clone() {
-            recent_notifies.push_back(notify);
-        }
-
         let downstream = Arc::new(Mutex::new(Downstream {
             connection_id,
             authorized_names: vec![],
@@ -192,14 +190,13 @@ impl Downstream {
             version_rolling_min_bit: None,
             tx_sv1_bridge,
             tx_outgoing,
-            first_job_received: false,
             extranonce2_len,
             difficulty_mgmt,
             upstream_difficulty_config,
             last_call_to_update_hr: 0,
-            recent_notifies: recent_notifies.clone(),
             stats_sender,
-            job_ids: JobIds::new(),
+            recent_jobs: RecentJobs::new(),
+            first_job: last_notify.expect("we have an assertion at the beginning of this function"),
         }));
 
         if let Err(e) = start_receive_downstream(
@@ -231,7 +228,6 @@ impl Downstream {
             task_manager.clone(),
             downstream.clone(),
             rx_sv1_notify,
-            recent_notifies,
             host.clone(),
             connection_id,
         )
@@ -354,7 +350,6 @@ impl Downstream {
         version_rolling_min_bit: Option<HexU32Be>,
         tx_sv1_bridge: Sender<DownstreamMessages>,
         tx_outgoing: Sender<json_rpc::Message>,
-        first_job_received: bool,
         extranonce2_len: usize,
         difficulty_mgmt: DownstreamDifficultyConfig,
         upstream_difficulty_config: Arc<Mutex<UpstreamDifficultyConfig>>,
@@ -368,13 +363,13 @@ impl Downstream {
             version_rolling_min_bit,
             tx_sv1_bridge,
             tx_outgoing,
-            first_job_received,
             extranonce2_len,
             difficulty_mgmt,
             upstream_difficulty_config,
             last_call_to_update_hr: 0,
-            recent_notifies: VecDeque::with_capacity(2),
+            first_job: Notify,
             stats_sender,
+            recent_jobs: RecentJobs::new(),
         }
     }
 }
@@ -393,6 +388,10 @@ impl IsServer<'static> for Downstream {
 
         self.version_rolling_mask = Some(version_rolling_mask.clone());
         self.version_rolling_min_bit = Some(version_rolling_min_bit_count.clone());
+        let mut first_job = self.first_job.clone();
+        self.recent_jobs
+            .add_job(&mut first_job, self.version_rolling_mask.clone());
+        self.first_job = first_job;
 
         (
             Some(server_to_client::VersionRollingParams::new(
@@ -441,76 +440,73 @@ impl IsServer<'static> for Downstream {
             request.id, request.user_name, request.job_id, request.nonce
         );
 
-        // check first job received
-        if !self.first_job_received {
-            self.stats_sender.update_rejected_shares(self.connection_id);
-            return false;
-        }
         let mut request = request.clone();
         let job_id_as_number = request.job_id.parse::<u32>();
         if job_id_as_number.is_err() {
-            error!("Share rejected: can not convert v1 job id to number. v1 id: {}", request.job_id);
+            error!(
+                "Share rejected: can not convert v1 job id to number. v1 id: {}",
+                request.job_id
+            );
             self.stats_sender.update_rejected_shares(self.connection_id);
             return false;
         }
-        let v2_id = self.job_ids.get_v2(job_id_as_number.unwrap());
-        if v2_id.is_none() {
-            error!("Share rejected: can not convert from v1 to v2 job id. v1 id: {}", request.job_id);
-            self.stats_sender.update_rejected_shares(self.connection_id);
-            return false;
-        }
-        request.job_id = v2_id.unwrap();
-        //check allowed to send shares
         match allow_submit_share() {
             Ok(true) => {
-                if self.recent_notifies.is_empty() {
-                    error!("Share rejected: No last job found");
-                    self.stats_sender.update_rejected_shares(self.connection_id);
-                    return false;
-                };
                 crate::translator::utils::update_share_count(self.connection_id); // update share count
-
-                //check share is valid
-                if let Some(met_difficulty) = validate_share(
-                    &request,
-                    &self.recent_notifies,
-                    &self.difficulty_mgmt.current_difficulties,
-                    self.extranonce1.clone(),
-                    self.version_rolling_mask.clone(),
-                ) {
-                    // Only forward upstream if the share meets the latest difficulty
-                    if let Some(latest_difficulty) =
-                        self.difficulty_mgmt.current_difficulties.back()
-                    {
-                        if met_difficulty == *latest_difficulty {
-                            let to_send = SubmitShareWithChannelId {
-                                channel_id: self.connection_id,
-                                share: request.clone(),
-                                extranonce: self.extranonce1.clone(),
-                                extranonce2_len: self.extranonce2_len,
-                                version_rolling_mask: self.version_rolling_mask.clone(),
-                            };
-                            if let Err(e) = self
-                                .tx_sv1_bridge
-                                .try_send(DownstreamMessages::SubmitShares(to_send))
-                            {
-                                error!("Failed to start receive downstream task: {e:?}");
-                                self.stats_sender.update_rejected_shares(self.connection_id);
-                                // Return false because submit was not properly handled
-                                return false;
+                if let Some(job) = self
+                    .recent_jobs
+                    .get_matching_job(job_id_as_number.expect("checked above"))
+                {
+                    request.job_id = job.job_id.clone();
+                    //check share is valid
+                    if let Some(met_difficulty) = validate_share(
+                        &request,
+                        &job,
+                        &self.difficulty_mgmt.current_difficulties,
+                        self.extranonce1.clone(),
+                        self.version_rolling_mask.clone(),
+                    ) {
+                        // Only forward upstream if the share meets the latest difficulty
+                        if let Some(latest_difficulty) =
+                            self.difficulty_mgmt.current_difficulties.back()
+                        {
+                            if met_difficulty == *latest_difficulty {
+                                let to_send = SubmitShareWithChannelId {
+                                    channel_id: self.connection_id,
+                                    share: request.clone(),
+                                    extranonce: self.extranonce1.clone(),
+                                    extranonce2_len: self.extranonce2_len,
+                                    version_rolling_mask: self.version_rolling_mask.clone(),
+                                };
+                                if let Err(e) = self
+                                    .tx_sv1_bridge
+                                    .try_send(DownstreamMessages::SubmitShares(to_send))
+                                {
+                                    error!("Failed to start receive downstream task: {e:?}");
+                                    self.stats_sender.update_rejected_shares(self.connection_id);
+                                    // Return false because submit was not properly handled
+                                    return false;
+                                }
                             }
                         }
+                        self.stats_sender.update_accepted_shares(self.connection_id);
+                        info!(
+                            "Share for Job {} and difficulty {} is accepted",
+                            request.job_id, met_difficulty
+                        );
+                        return true;
+                    } else {
+                        error!("Share rejected: Invalid share");
+                        self.stats_sender.update_rejected_shares(self.connection_id);
+                        return false;
                     }
-                    self.stats_sender.update_accepted_shares(self.connection_id);
-                    info!(
-                        "Share for Job {} and difficulty {} is accepted",
-                        request.job_id, met_difficulty
-                    );
-                    true
                 } else {
-                    error!("Share rejected: Invalid share");
+                    error!(
+                        "Share rejected: can not find job with id {}",
+                        request.job_id
+                    );
                     self.stats_sender.update_rejected_shares(self.connection_id);
-                    false
+                    return false;
                 }
             }
             Ok(false) => {
@@ -596,14 +592,55 @@ impl IsDownstream for Downstream {
 }
 
 #[derive(Debug)]
-pub struct JobIds {
-    v1_to_v2: HashMap<u32,u32>,
-    v2_to_v1: HashMap<u32,Vec<u32>>,
-    last_v2s: CircularBuffer<u32,3>,
+pub struct RecentJobs {
+    v1_to_v2: HashMap<u32, u32>,
+    v2_to_v1: HashMap<u32, Vec<u32>>,
+    jobs: VecDeque<Notify<'static>>,
+    last_v2s: CircularBuffer<u32, 3>,
+    tracked_jobs: usize,
+}
+fn apply_mask(mask: Option<HexU32Be>, message: &mut server_to_client::Notify<'static>) {
+    if let Some(mask) = mask {
+        message.version = HexU32Be(message.version.0 & !mask.0);
+    }
 }
+impl RecentJobs {
+    pub fn add_job(&mut self, notify: &mut Notify<'static>, mask: Option<HexU32Be>) {
+        apply_mask(mask, notify);
+        // save it with the v2 id
+        self.jobs.push_back(notify.clone());
+        let new_id = self.new_v1(notify.job_id.parse::<u32>().unwrap());
+        // send it with the v1 id
+        notify.job_id = new_id.to_string();
+        if self.jobs.len() > self.tracked_jobs {
+            self.jobs.pop_front();
+        };
+    }
+
+    pub fn clone_last(&mut self) -> Option<Notify<'static>> {
+        if let Some(job) = self.jobs.back() {
+            let mut job = job.clone();
+            let new_id = self.new_v1(job.job_id.parse::<u32>().unwrap());
+            job.job_id = new_id.to_string();
+            Some(job.clone())
+        } else {
+            None
+        }
+    }
+
+    pub fn current_jobs(&self) -> VecDeque<Notify<'static>> {
+        self.jobs.clone()
+    }
 
-impl JobIds {
-    pub fn new_v1(&mut self, v2_id: u32) -> u32 {
+    pub fn get_matching_job(&self, v1_id: u32) -> Option<Notify<'static>> {
+        let v2_id = self.get_v2(v1_id)?;
+        self.current_jobs()
+            .iter()
+            .find(|notify| notify.job_id == v2_id)
+            .cloned()
+    }
+
+    fn new_v1(&mut self, v2_id: u32) -> u32 {
         let mut v1_id = rand::thread_rng().gen();
         while self.v1_to_v2.contains_key(&v1_id) {
             v1_id = rand::thread_rng().gen();
@@ -637,17 +674,18 @@ impl JobIds {
             v1_to_v2: HashMap::new(),
             v2_to_v1: HashMap::new(),
             last_v2s: CircularBuffer::new(),
+            jobs: VecDeque::new(),
+            tracked_jobs: 3,
         }
     }
 }
 
-impl Default for JobIds {
+impl Default for RecentJobs {
     fn default() -> Self {
         Self::new()
     }
 }
 
-
 #[derive(Debug)]
 struct CircularBuffer<T, const N: usize> {
     buffer: [Option<T>; N],
@@ -680,7 +718,6 @@ impl<T, const N: usize> CircularBuffer<T, N> {
     }
 }
 
-
 //#[cfg(test)]
 //mod tests {
 //    use super::*;