FIX upstream share classification by preserving the aggregate pool target

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dmnd-client"
-version = "0.3.14"
+version = "0.3.15"
 edition = "2021"
 
 [lib]

src/translator/downstream/diff_management.rs

@@ -90,6 +90,7 @@ impl Downstream {
             u.channel_nominal_hashrate -=
                 // Make sure that upstream channel hasrate never goes below 0
                 f32::min(estimated_downstream_hash_rate, u.channel_nominal_hashrate);
+            u.request_immediate_update();
         })?;
         Ok(())
     }
@@ -318,6 +319,7 @@ impl Downstream {
             } else {
                 c.channel_nominal_hashrate = 0.0;
             }
+            c.request_immediate_update();
         })?;
         Ok(())
     }
@@ -560,10 +562,8 @@ mod test {
             initial_difficulty: latest_difficulty,
             hard_minimum_difficulty: None,
         };
-        let upstream_config = Arc::new(Mutex::new(UpstreamDifficultyConfig {
-            channel_diff_update_interval: 60,
-            channel_nominal_hashrate,
-        }));
+        let (upstream_config, _rx) = UpstreamDifficultyConfig::new(60, channel_nominal_hashrate);
+        let upstream_config = Arc::new(Mutex::new(upstream_config));
         let (tx_sv1_submit, _rx_sv1_submit) = tokio::sync::mpsc::channel(10);
         let (tx_outgoing, _rx_outgoing) = channel(10);
         let (tx_update_token, _rx_update_token) = channel(10);
@@ -620,10 +620,8 @@ mod test {
             initial_difficulty: latest_difficulty,
             hard_minimum_difficulty: None,
         };
-        let upstream_config = Arc::new(Mutex::new(UpstreamDifficultyConfig {
-            channel_diff_update_interval: 60,
-            channel_nominal_hashrate,
-        }));
+        let (upstream_config, _rx) = UpstreamDifficultyConfig::new(60, channel_nominal_hashrate);
+        let upstream_config = Arc::new(Mutex::new(upstream_config));
         let (tx_sv1_submit, _rx_sv1_submit) = tokio::sync::mpsc::channel(10);
         let (tx_outgoing, rx_outgoing) = channel(10);
         let (tx_update_token, _rx_update_token) = channel(10);
@@ -683,8 +681,15 @@ mod test {
             target_rate_submits(),
             RAW_BOOTSTRAP_HASHRATE,
         );
+        let mut update_rx = upstream_config
+            .safe_lock(|u| u.subscribe_updates())
+            .unwrap();
 
         Downstream::update_difficulty_and_hashrate(&downstream).unwrap();
+        tokio::time::timeout(Duration::from_secs(1), update_rx.changed())
+            .await
+            .unwrap()
+            .unwrap();
 
         let estimated_downstream_hash_rate = downstream
             .safe_lock(|d| d.difficulty_mgmt.estimated_downstream_hash_rate)
@@ -697,6 +702,37 @@ mod test {
         assert_hashrate_close(channel_nominal_hashrate, quantized_hashrate);
     }
 
+    #[tokio::test]
+    async fn removing_downstream_hashrate_requests_immediate_upstream_retarget() {
+        let latest_difficulty = 1_024.0;
+        let estimated_hashrate = hashrate_for_diff(latest_difficulty);
+        let pid = Pid::new(*crate::SHARE_PER_MIN, latest_difficulty * 10.0);
+        let (downstream, upstream_config) = seeded_downstream(
+            estimated_hashrate,
+            latest_difficulty,
+            pid,
+            VecDeque::new(),
+            estimated_hashrate,
+        );
+        let mut update_rx = upstream_config
+            .safe_lock(|u| u.subscribe_updates())
+            .unwrap();
+
+        downstream
+            .safe_lock(|d| d.mark_channel_hashrate_registered())
+            .unwrap();
+        Downstream::remove_downstream_hashrate_from_channel(&downstream).unwrap();
+        tokio::time::timeout(Duration::from_secs(1), update_rx.changed())
+            .await
+            .unwrap()
+            .unwrap();
+
+        let channel_nominal_hashrate = upstream_config
+            .safe_lock(|u| u.channel_nominal_hashrate)
+            .unwrap();
+        assert_eq!(channel_nominal_hashrate, 0.0);
+    }
+
     #[tokio::test]
     #[ignore = "non-zero delay is blocked until the stale bootstrap difficulty replay bug is fixed"]
     async fn positive_delay_does_not_replay_stale_bootstrap_difficulty_after_retarget() {
@@ -840,10 +876,7 @@ mod test {
             initial_difficulty: 10_000.0,
             hard_minimum_difficulty: Some(NON_LOCAL_DOWNSTREAM_MIN_DIFFICULTY),
         };
-        let upstream_config = UpstreamDifficultyConfig {
-            channel_diff_update_interval: 60,
-            channel_nominal_hashrate: 0.0,
-        };
+        let (upstream_config, _rx) = UpstreamDifficultyConfig::new(60, 0.0);
         let (tx_sv1_submit, _rx_sv1_submit) = tokio::sync::mpsc::channel(10);
         let (tx_outgoing, _rx_outgoing) = channel(10);
         let (tx_update_token, _rx_update_token) = channel(10);
@@ -902,10 +935,7 @@ mod test {
             initial_difficulty: 10_000_000_000.0,
             hard_minimum_difficulty: None,
         };
-        let upstream_config = UpstreamDifficultyConfig {
-            channel_diff_update_interval: 60,
-            channel_nominal_hashrate: 0.0,
-        };
+        let (upstream_config, _rx) = UpstreamDifficultyConfig::new(60, 0.0);
         let (tx_sv1_submit, _rx_sv1_submit) = tokio::sync::mpsc::channel(10);
         let (tx_outgoing, _rx_outgoing) = channel(10);
         let (tx_update_token, _rx_update_token) = channel(10);

src/translator/downstream/downstream.rs

@@ -1236,10 +1236,8 @@ mod tests {
             initial_difficulty: 1.0,
             hard_minimum_difficulty: None,
         };
-        let upstream_config = UpstreamDifficultyConfig {
-            channel_diff_update_interval: crate::CHANNEL_DIFF_UPDTATE_INTERVAL,
-            channel_nominal_hashrate: 0.0,
-        };
+        let (upstream_config, _rx) =
+            UpstreamDifficultyConfig::new(crate::CHANNEL_DIFF_UPDTATE_INTERVAL, 0.0);
         let (tx_sv1_submit, _rx_sv1_submit) = channel::<DownstreamMessages>(8);
         let (tx_outgoing, rx_outgoing) = channel(8);
         let (tx_update_token, _rx_update_token) = channel(8);

src/translator/downstream/notify.rs

@@ -49,8 +49,10 @@ pub async fn start_notify(
                     d.difficulty_mgmt.estimated_downstream_hash_rate,
                 )
             })?;
-        upstream_difficulty_config
-            .safe_lock(|c| c.channel_nominal_hashrate += registered_hashrate)?;
+        upstream_difficulty_config.safe_lock(|c| {
+            c.channel_nominal_hashrate += registered_hashrate;
+            c.request_immediate_update();
+        })?;
         downstream.safe_lock(|d| d.mark_channel_hashrate_registered())?;
         if let Err(e) = stats_sender.setup_stats_reliable(connection_id).await {
             error!("Failed to register downstream stats {connection_id}: {e}");
@@ -66,6 +68,7 @@ pub async fn start_notify(
                 upstream_difficulty_config.safe_lock(|u| {
                     u.channel_nominal_hashrate -=
                         f32::min(registered_hashrate, u.channel_nominal_hashrate);
+                    u.request_immediate_update();
                 })?;
             }
             return Ok(());
@@ -275,10 +278,8 @@ mod tests {
             initial_difficulty: 1.0,
             hard_minimum_difficulty: None,
         };
-        let upstream_config = UpstreamDifficultyConfig {
-            channel_diff_update_interval: crate::CHANNEL_DIFF_UPDTATE_INTERVAL,
-            channel_nominal_hashrate: 0.0,
-        };
+        let (upstream_config, _rx) =
+            UpstreamDifficultyConfig::new(crate::CHANNEL_DIFF_UPDTATE_INTERVAL, 0.0);
         let (tx_sv1_submit, _rx_sv1_submit) = channel::<DownstreamMessages>(8);
         let (tx_outgoing, rx_outgoing) = channel(8);
         let (tx_update_token, _rx_update_token) = channel(8);
@@ -342,10 +343,11 @@ mod tests {
             initial_difficulty: latest_difficulty,
             hard_minimum_difficulty: None,
         };
-        let upstream_config = Arc::new(Mutex::new(UpstreamDifficultyConfig {
-            channel_diff_update_interval: crate::CHANNEL_DIFF_UPDTATE_INTERVAL,
+        let (upstream_config, _rx) = UpstreamDifficultyConfig::new(
+            crate::CHANNEL_DIFF_UPDTATE_INTERVAL,
             channel_nominal_hashrate,
-        }));
+        );
+        let upstream_config = Arc::new(Mutex::new(upstream_config));
         let (tx_sv1_submit, _rx_sv1_submit) = channel::<DownstreamMessages>(8);
         let (tx_outgoing, _rx_outgoing) = channel(8);
         let (tx_update_token, _rx_update_token) = channel(8);
@@ -421,6 +423,51 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn start_notify_requests_immediate_upstream_retarget_on_register() {
+        let first_job = first_job("91");
+        let (downstream, _rx_outgoing) =
+            downstream_with_first_job(first_job, vec!["worker".to_string()]);
+        let downstream = Arc::new(Mutex::new(downstream));
+        let mut update_rx = downstream
+            .safe_lock(|d| {
+                d.upstream_difficulty_config
+                    .safe_lock(|c| c.subscribe_updates())
+                    .unwrap()
+            })
+            .unwrap();
+        let task_manager = TaskManager::initialize();
+        let (_tx_notify, rx_notify) = broadcast::channel(8);
+
+        start_notify(
+            task_manager.clone(),
+            downstream.clone(),
+            rx_notify,
+            "127.0.0.1".to_string(),
+            1,
+        )
+        .await
+        .unwrap();
+
+        tokio::time::timeout(Duration::from_secs(1), update_rx.changed())
+            .await
+            .unwrap()
+            .unwrap();
+
+        let channel_nominal_hashrate = downstream
+            .safe_lock(|d| {
+                d.upstream_difficulty_config
+                    .safe_lock(|c| c.channel_nominal_hashrate)
+                    .unwrap()
+            })
+            .unwrap();
+        assert_eq!(channel_nominal_hashrate, 1.0);
+
+        if let Some(aborter) = task_manager.safe_lock(|t| t.get_aborter()).unwrap() {
+            drop(aborter);
+        }
+    }
+
     #[tokio::test]
     async fn first_retarget_waits_full_adjustment_interval_before_fixing_upstream_hashrate() {
         let quantized_difficulty = quantized_difficulty_for_hashrate(RAW_BOOTSTRAP_HASHRATE);

src/translator/mod.rs

@@ -89,10 +89,10 @@ pub async fn start(
 
     let channel_nominal_hashrate = 0.0;
 
-    let upstream_diff = UpstreamDifficultyConfig {
-        channel_diff_update_interval: crate::CHANNEL_DIFF_UPDTATE_INTERVAL,
+    let (upstream_diff, diff_update_rx) = UpstreamDifficultyConfig::new(
+        crate::CHANNEL_DIFF_UPDTATE_INTERVAL,
         channel_nominal_hashrate,
-    };
+    );
     let diff_config = Arc::new(Mutex::new(upstream_diff));
 
     // Instantiate a new `Upstream` (SV2 Pool)
@@ -113,6 +113,7 @@ pub async fn start(
         recv_from_up,
         rx_sv2_submit_shares_ext,
         rx_update_token,
+        diff_update_rx,
     )
     .await?;
     TaskManager::add_upstream(task_manager.clone(), upstream_abortable)