Merge pull request dmnd-pool#62 from Priceless-P/feat/accept-last-two-job

Feat/accept last two job

Author: Fi3

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "demand-cli"
-version = "0.1.4"
+version = "0.1.5"
 edition = "2021"
 
 [dependencies]
@@ -23,7 +23,6 @@ axum = {version = "0.8.1"}
 serde = { version = "1.0.219", features = ["derive"] }  
 sysinfo = {version = "0.33.1"}
 primitive-types = { version = "0.13.1" }
-hex={version = "*"}
 #roles_logic_sv2 = "1.2.1"
 #sv1_api = "1.0.1"
 #demand-sv2-connection = "0.0.3"

Cargo.toml

@@ -613,10 +613,11 @@ impl
                             header_nonce: share.nonce,
                             coinbase_tx: coinbase.try_into()?,
                         };
-                        // The below channel should never be full is ok to block
-                        solution_sender
-                            .blocking_send(solution)
-                            .map_err(|_| Error::DownstreamDown)?; // Better Error to return here?
+                        tokio::spawn(async move {
+                            if solution_sender.send(solution).await.is_err() {
+                                error!("Downstream channel closed, couldn't send solution");
+                            }
+                        });
                         if !self.status.is_solo_miner() {
                             {
                                 let jd = self.jd.clone();

src/jd_client/mining_downstream/mod.rs

@@ -403,7 +403,6 @@ impl ParseUpstreamMiningMessages<Downstream, NullDownstreamMiningSelector, NoRou
 
         let extranonces = ExtendedExtranonce::new(range_0, range_1, range_2);
         let creator = roles_logic_sv2::job_creator::JobsCreators::new(total_len as u8);
-        let share_per_min = 1.0;
         let channel_kind =
             roles_logic_sv2::channel_logic::channel_factory::ExtendedChannelKind::ProxyJd {
                 upstream_target: m.target.clone().into(),
@@ -412,7 +411,7 @@ impl ParseUpstreamMiningMessages<Downstream, NullDownstreamMiningSelector, NoRou
             ids,
             extranonces,
             creator,
-            share_per_min,
+            crate::SHARE_PER_MIN,
             channel_kind,
             vec![],
             vec![],

src/jd_client/mining_upstream/upstream.rs

@@ -34,8 +34,7 @@ const CHANNEL_DIFF_UPDTATE_INTERVAL: u32 = 10;
 const MAX_LEN_DOWN_MSG: u32 = 10000;
 const MAIN_POOL_ADDRESS: &str = "mining.dmnd.work:2000";
 //const TEST_POOL_ADDRESS: &str = "127.0.0.1:20000";
-const TEST_POOL_ADDRESS: &str =
-    "k8s-default-pool-de2d9b37ea-6bc40843aed871f2.elb.eu-central-1.amazonaws.com:2000";
+const TEST_POOL_ADDRESS: &str = "18.193.252.132:2000";
 const MAIN_AUTH_PUB_KEY: &str = "9bQHWXsQ2J9TRFTaxRh3KjoxdyLRfWVEy25YHtKF8y8gotLoCZZ";
 const TEST_AUTH_PUB_KEY: &str = "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72";
 //const TP_ADDRESS: &str = "127.0.0.1:8442";
@@ -168,7 +167,7 @@ async fn initialize_proxy(
                 Ok(connection) => connection,
                 Err(_) => {
                     error!("No upstream available. Retrying...");
-                    error!("Are you using the correct TOKEN??");
+                    warn!("Are you using the correct TOKEN??");
                     let mut secs = 10;
                     while secs > 0 {
                         tracing::warn!("Retrying in {} seconds...", secs);

src/main.rs

@@ -1,7 +1,7 @@
 use super::{Downstream, DownstreamMessages, SetDownstreamTarget};
 use pid::Pid;
 use roles_logic_sv2::{self, utils::from_u128_to_u256};
-use sv1_api::{self, methods::server_to_client::SetDifficulty, server_to_client::Notify};
+use sv1_api::{self, methods::server_to_client::SetDifficulty};
 
 use super::super::error::{Error, ProxyResult};
 use primitive_types::U256;
@@ -47,7 +47,8 @@ impl Downstream {
         });
         stats_sender.update_diff(connection_id, diff);
         stats_sender.update_hashrate(connection_id, estimated_hashrate);
-        tokio::spawn(crate::translator::utils::check_share_rate_limit());
+        let downstream = self_.clone();
+        tokio::spawn(crate::translator::utils::check_share_rate_limit(downstream));
 
         Ok(())
     }
@@ -79,15 +80,14 @@ impl Downstream {
     /// downstream and the bridge.
     pub async fn try_update_difficulty_settings(
         self_: &Arc<Mutex<Self>>,
-        last_notify: Option<Notify<'static>>,
     ) -> ProxyResult<'static, ()> {
         let channel_id = self_
             .clone()
             .safe_lock(|d| (d.connection_id))
             .map_err(|_e| Error::TranslatorDiffConfigMutexPoisoned)?;
 
         if let Some(new_diff) = Self::update_difficulty_and_hashrate(self_)? {
-            Self::update_diff_setting(self_, channel_id, new_diff.into(), last_notify).await?;
+            Self::update_diff_setting(self_, channel_id, new_diff.into()).await?;
         }
         Ok(())
     }
@@ -100,13 +100,17 @@ impl Downstream {
         self_: &Arc<Mutex<Self>>,
         channel_id: u32,
         new_diff: f64,
-        last_notify: Option<Notify<'static>>,
     ) -> ProxyResult<'static, ()> {
         // Send messages downstream
         let (message, target) = diff_to_sv1_message(new_diff)?;
         Downstream::send_message_downstream(self_.clone(), message).await;
 
-        if let Some(notify) = last_notify {
+        // Get the last notify
+        let recent_notify = self_
+            .safe_lock(|d| d.recent_notifies.back().cloned())
+            .map_err(|_| Error::TranslatorDiffConfigMutexPoisoned)?;
+
+        if let Some(notify) = recent_notify {
             Downstream::send_message_downstream(self_.clone(), notify.into()).await;
         }
 
@@ -459,7 +463,7 @@ mod test {
         while elapsed <= total_run_time {
             mock_mine(initial_target.clone().into(), &mut share);
             Downstream::save_share(downstream.clone()).unwrap();
-            let _ = Downstream::try_update_difficulty_settings(&downstream, None).await;
+            let _ = Downstream::try_update_difficulty_settings(&downstream).await;
             initial_target = downstream
                 .safe_lock(|d| {
                     match roles_logic_sv2::utils::hash_rate_to_target(

src/translator/downstream/diff_management.rs

@@ -104,7 +104,7 @@ pub struct Downstream {
     pub(super) difficulty_mgmt: DownstreamDifficultyConfig,
     pub(super) upstream_difficulty_config: Arc<Mutex<UpstreamDifficultyConfig>>,
     pub last_call_to_update_hr: u128,
-    pub(super) last_notify: Option<server_to_client::Notify<'static>>,
+    pub(super) recent_notifies: VecDeque<server_to_client::Notify<'static>>,
     pub(super) stats_sender: StatsSender,
 }
 
@@ -167,6 +167,11 @@ 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![],
@@ -180,7 +185,7 @@ impl Downstream {
             difficulty_mgmt,
             upstream_difficulty_config,
             last_call_to_update_hr: 0,
-            last_notify: last_notify.clone(),
+            recent_notifies: recent_notifies.clone(),
             stats_sender,
         }));
 
@@ -213,7 +218,7 @@ impl Downstream {
             task_manager.clone(),
             downstream.clone(),
             rx_sv1_notify,
-            last_notify,
+            recent_notifies,
             host.clone(),
             connection_id,
         )
@@ -354,7 +359,7 @@ impl Downstream {
             difficulty_mgmt,
             upstream_difficulty_config,
             last_call_to_update_hr: 0,
-            last_notify: None,
+            recent_notifies: VecDeque::with_capacity(2),
             stats_sender,
         }
     }
@@ -417,7 +422,10 @@ impl IsServer<'static> for Downstream {
     /// When miner find the job which meets requested difficulty, it can submit share to the server.
     /// Only [Submit](client_to_server::Submit) requests for authorized user names can be submitted.
     fn handle_submit(&self, request: &client_to_server::Submit<'static>) -> bool {
-        info!("Down: Handling mining.submit: {:?}", &request);
+        info!(
+            "Downstream {}: Handling mining.submit for job_id {}",
+            self.connection_id, request.job_id
+        );
 
         // check first job received
         if !self.first_job_received {
@@ -427,7 +435,7 @@ impl IsServer<'static> for Downstream {
         //check allowed to send shares
         match allow_submit_share() {
             Ok(true) => {
-                let Some(job) = &self.last_notify else {
+                if self.recent_notifies.is_empty() {
                     error!("Share rejected: No last job found");
                     self.stats_sender.update_rejected_shares(self.connection_id);
                     return false;
@@ -436,7 +444,7 @@ impl IsServer<'static> for Downstream {
                                                                                   //check share is valid
                 if validate_share(
                     request,
-                    job,
+                    &self.recent_notifies,
                     self.difficulty_mgmt.current_difficulty,
                     self.extranonce1.clone(),
                     self.version_rolling_mask.clone(),