Fix/diff interval (dmnd-pool#61)

* make diff adjustment interval configurable

* fix

* update expected hashrate calculation

* update api with initial stats

* add submit share failure counter

Author: Priceless-P

src/main.rs

@@ -77,6 +77,8 @@ struct Args {
     noise_connection_log: String,
     #[clap(long = "delay", default_value = "0")]
     delay: u64,
+    #[clap(long = "interval", short = 'i', default_value = "120000")]
+    adjustment_interval: u64,
 }
 
 #[tokio::main]
@@ -166,6 +168,7 @@ async fn initialize_proxy(
                 Ok(connection) => connection,
                 Err(_) => {
                     error!("No upstream available. Retrying...");
+                    error!("Are you using the correct TOKEN??");
                     let mut secs = 10;
                     while secs > 0 {
                         tracing::warn!("Retrying in {} seconds...", secs);

src/translator/downstream/accept_connection.rs

@@ -33,8 +33,20 @@ pub async fn start_accept_connection(
             let _s = tx_mining_notify.subscribe();
             while let Some((send, recv, addr)) = downstreams.recv().await {
                 info!("Translator opening connection for ip {}", addr);
-                // TODO handle also cases where a cpuminer want to connect
-                let expected_hash_rate = *crate::EXPECTED_SV1_HASHPOWER;
+
+                // The initial difficulty is derived from the formula: difficulty = hash_rate / (shares_per_second * 2^32)
+                let initial_hash_rate = *crate::EXPECTED_SV1_HASHPOWER;
+                let share_per_second = crate::SHARE_PER_MIN / 60.0;
+                let initial_difficulty =
+                    dbg!(initial_hash_rate / (share_per_second * 2f32.powf(32.0)));
+                let initial_difficulty =
+                    crate::translator::downstream::diff_management::nearest_power_of_10(
+                        initial_difficulty,
+                    );
+
+                // Formula: expected_hash_rate = (shares_per_second) * initial_difficulty * 2^32, where shares_per_second = SHARE_PER_MIN / 60
+                let expected_hash_rate =
+                    (crate::SHARE_PER_MIN / 60.0) * initial_difficulty * 2f32.powf(32.0);
                 if Bridge::ready(&bridge).await.is_err() {
                     error!("Bridge not ready");
                     break;
@@ -66,6 +78,7 @@ pub async fn start_accept_connection(
                             send,
                             recv,
                             task_manager.clone(),
+                            initial_difficulty,
                             stats_sender.clone(),
                         )
                         .await

src/translator/downstream/diff_management.rs

@@ -17,7 +17,14 @@ impl Downstream {
     /// Initializes difficult managment.
     /// Send downstream a first target.
     pub async fn init_difficulty_management(self_: &Arc<Mutex<Self>>) -> ProxyResult<()> {
-        let diff = self_.safe_lock(|d| d.difficulty_mgmt.current_difficulty)?;
+        let (diff, stats_sender, connection_id, estimated_hashrate) = self_.safe_lock(|d| {
+            (
+                d.difficulty_mgmt.current_difficulty,
+                d.stats_sender.clone(),
+                d.connection_id,
+                d.difficulty_mgmt.estimated_downstream_hash_rate,
+            )
+        })?;
 
         let (message, _) = diff_to_sv1_message(diff as f64)?;
         Downstream::send_message_downstream(self_.clone(), message.clone()).await;
@@ -38,7 +45,8 @@ impl Downstream {
                 Downstream::send_message_downstream(self_clone.clone(), message.clone()).await;
             }
         });
-
+        stats_sender.update_diff(connection_id, diff);
+        stats_sender.update_hashrate(connection_id, estimated_hashrate);
         tokio::spawn(crate::translator::utils::check_share_rate_limit());
 
         Ok(())

src/translator/downstream/downstream.rs

@@ -3,7 +3,6 @@ use crate::{
     proxy_state::{DownstreamType, ProxyState},
     shared::utils::AbortOnDrop,
     translator::{
-        downstream::diff_management::nearest_power_of_10,
         error::Error,
         utils::{allow_submit_share, validate_share},
     },
@@ -124,18 +123,13 @@ impl Downstream {
         send_to_down: Sender<String>,
         recv_from_down: Receiver<String>,
         task_manager: Arc<Mutex<TaskManager>>,
+        initial_difficulty: f32,
         stats_sender: StatsSender,
     ) {
         assert!(last_notify.is_some());
 
         let (tx_outgoing, receiver_outgoing) = channel(crate::TRANSLATOR_BUFFER_SIZE);
 
-        // The initial difficulty is derived from the formula: difficulty = hash_rate / (shares_per_second * 2^32),
-        let initial_hash_rate = *crate::EXPECTED_SV1_HASHPOWER;
-        let share_per_second = crate::SHARE_PER_MIN / 60.0;
-        let initial_difficulty = dbg!(initial_hash_rate / (share_per_second * 2f32.powf(32.0)));
-        let initial_difficulty = nearest_power_of_10(initial_difficulty);
-
         // The PID controller uses negative proportional (P) and integral (I) gains to reduce difficulty
         // when the actual share rate falls below the target rate (SHARE_PER_MIN). Negative gains are chosen
         // because a lower share rate indicates the difficulty is too high for the miner, requiring a downward

src/translator/downstream/notify.rs

@@ -145,8 +145,7 @@ async fn start_update(
             {
                 std::time::Duration::from_millis(5000)
             } else {
-                // TODO we really need to use differenet times seems to work well enaugh with 5 sec
-                std::time::Duration::from_millis(5000)
+                std::time::Duration::from_millis(crate::ARGS.adjustment_interval)
             };
 
             tokio::time::sleep(sleep_duration).await;

src/translator/proxy/bridge.rs

@@ -8,7 +8,10 @@ use roles_logic_sv2::{
     parsers::Mining,
     utils::{GroupId, Mutex},
 };
-use std::sync::Arc;
+use std::sync::{
+    atomic::{AtomicU32, Ordering},
+    Arc,
+};
 use sv1_api::{client_to_server::Submit, server_to_client, utils::HexU32Be};
 use tokio::sync::broadcast;
 
@@ -23,8 +26,13 @@ use crate::{
     proxy_state::{ProxyState, TranslatorState, UpstreamType},
     shared::utils::AbortOnDrop,
 };
+use lazy_static::lazy_static;
 use roles_logic_sv2::{channel_logic::channel_factory::OnNewShare, Error as RolesLogicError};
-use tracing::{debug, error, info};
+use tracing::{debug, error, info, warn};
+
+lazy_static! {
+    static ref SUBMIT_FAIL_COUNTER: AtomicU32 = AtomicU32::new(0);
+}
 
 /// Bridge between the SV2 `Upstream` and SV1 `Downstream` responsible for the following messaging
 /// translation:
@@ -78,7 +86,6 @@ impl Bridge {
     ) -> Result<Arc<Mutex<Self>>, Error<'static>> {
         info!("Creating new bridge for up_id {}:", up_id);
         let ids = Arc::new(Mutex::new(GroupId::new()));
-        let share_per_min = 1.0;
         let upstream_target: [u8; 32] =  target.safe_lock(|t| {
     t.clone().try_into().expect("Internal error: this operation can not fail because Vec<U8> can always be converted into [u8; 32]")
 }).map_err(|e| Error::TargetError(RolesLogicError::PoisonLock(e.to_string())))?;
@@ -91,7 +98,7 @@ impl Bridge {
                 ids,
                 extranonces,
                 None,
-                share_per_min,
+                crate::SHARE_PER_MIN,
                 ExtendedChannelKind::Proxy { upstream_target },
                 None,
                 up_id,
@@ -254,29 +261,33 @@ impl Bridge {
         let res = self_
             .safe_lock(|s| {
                 s.channel_factory.set_target(&mut upstream_target);
-                let sv2_submit = match s.translate_submit(
+                 match s.translate_submit(
                     share.channel_id,
                     share.share,
                     share.version_rolling_mask,
                 ) {
-                    Ok(submit_shares_extended) => submit_shares_extended,
+                    Ok(submit_shares_extended) => {
+                        // Ordering::Relaxed is safe here because we only need simple counter updates. 
+                        // No need for strict ordering since it just tracks failures.
+                        SUBMIT_FAIL_COUNTER.store(0, Ordering::Relaxed); // Reset on success
+                        s.channel_factory.on_submit_shares_extended(submit_shares_extended)
+                   },
                     Err(e) => {
-                        error!("Failed to Translates SV1 mining.submit message to SV2 SubmitSharesExtended message");
-                        return Err(e); // Error will be handled by the caller
+                        error!("Failed to Translates SV1 mining.submit message to SV2 SubmitSharesExtended message: {e}");
+                        Err(roles_logic_sv2::Error::NoValidJob) // Error will be handled by the caller
                     }
-                };
-                Ok(s.channel_factory.on_submit_shares_extended(sv2_submit))
+                }
             })
             .map_err(|_| Error::BridgeMutexPoisoned)?;
 
         match res {
-            Ok(Ok(OnNewShare::SendErrorDownstream(e))) => {
+            Ok(OnNewShare::SendErrorDownstream(e)) => {
                 let error_code = std::str::from_utf8(&e.error_code.to_vec()[..])
                     .unwrap_or("unparsable error code")
                     .to_string();
                 error!("Submit share error {}", error_code);
             }
-            Ok(Ok(OnNewShare::SendSubmitShareUpstream((share, _)))) => {
+            Ok(OnNewShare::SendSubmitShareUpstream((share, _))) => {
                 info!("SHARE MEETS UPSTREAM TARGET channel id: {}", channel_id);
                 match share {
                     Share::Extended(share) => {
@@ -290,19 +301,26 @@ impl Bridge {
                 }
             }
             // We are in an extended channel this variant is group channle only
-            Ok(Ok(OnNewShare::RelaySubmitShareUpstream)) => unreachable!(),
-            Ok(Ok(OnNewShare::ShareMeetDownstreamTarget)) => {
+            Ok(OnNewShare::RelaySubmitShareUpstream) => unreachable!(),
+            Ok(OnNewShare::ShareMeetDownstreamTarget) => {
                 info!("SHARE MEETS DOWNSTREAM TARGET channel id {}", channel_id);
             }
             // Proxy do not have JD capabilities
-            Ok(Ok(OnNewShare::ShareMeetBitcoinTarget(..))) => unreachable!(),
-            Ok(Err(e)) => {
-                error!("{}", e);
-                return Err(Error::RolesSv2Logic(e));
+            Ok(OnNewShare::ShareMeetBitcoinTarget(..)) => unreachable!(),
+            Err(roles_logic_sv2::Error::NoValidJob) => {
+                let count = SUBMIT_FAIL_COUNTER.fetch_add(1, Ordering::Relaxed) + 1;
+                if count >= 10 {
+                    error!("Failed to translate SV1 mining.submit message to SV2 SubmitSharesExtended message after 10 attempts");
+                    return Err(Error::RolesSv2Logic(roles_logic_sv2::Error::NoValidJob));
+                } else {
+                    warn!(
+                                "Failed to translate SV1 mining.submit message to SV2 SubmitSharesExtended message, attempt {}",
+                                count
+                            );
+                }
             }
             Err(e) => {
-                error!("{}", e);
-                return Err(e);
+                return Err(Error::RolesSv2Logic(e));
             }
         }
         Ok(())