Upgrade add limit to how many time a dowstream can change the proxy t…

src/config.rs

@@ -43,6 +43,8 @@ struct Args {
     adjustment_interval: Option<u64>,
     #[clap(long)]
     token: Option<String>,
+    #[clap(long = "max-token-changes-per-connection")]
+    max_token_changes_per_connection: Option<u32>,
     #[clap(long)]
     tp_address: Option<String>,
     #[clap(long)]
@@ -64,6 +66,7 @@ struct Args {
 #[derive(Serialize, Deserialize)]
 struct ConfigFile {
     token: Option<String>,
+    max_token_changes_per_connection: Option<u32>,
     tp_address: Option<String>,
     interval: Option<u64>,
     delay: Option<u64>,
@@ -85,6 +88,7 @@ impl ConfigFile {
     pub fn default() -> Self {
         ConfigFile {
             token: None,
+            max_token_changes_per_connection: None,
             tp_address: None,
             interval: None,
             delay: None,
@@ -107,6 +111,7 @@ impl ConfigFile {
 #[derive(Debug)]
 pub struct Configuration {
     token: Option<String>,
+    max_token_changes_per_connection: u32,
     tp_address: Option<String>,
     interval: u64,
     delay: u64,
@@ -129,6 +134,7 @@ impl Configuration {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         token: Option<String>,
+        max_token_changes_per_connection: u32,
         tp_address: Option<String>,
         interval: u64,
         delay: u64,
@@ -149,6 +155,7 @@ impl Configuration {
     ) -> Self {
         Configuration {
             token,
+            max_token_changes_per_connection,
             tp_address,
             interval,
             delay,
@@ -179,6 +186,7 @@ impl Configuration {
     fn default_for_tests() -> Self {
         Self::new(
             Some("test_token".to_string()),
+            0,
             None,
             120_000,
             0,
@@ -217,6 +225,10 @@ impl Configuration {
         Self::cfg().token.clone()
     }
 
+    pub fn max_token_changes_per_connection() -> u32 {
+        Self::cfg().max_token_changes_per_connection
+    }
+
     pub fn tp_address() -> Option<String> {
         Self::cfg().tp_address.clone()
     }
@@ -349,6 +361,16 @@ impl Configuration {
             .or_else(|| std::env::var("TOKEN").ok());
         println!("User Token: {token:?}");
 
+        let max_token_changes_per_connection = args
+            .max_token_changes_per_connection
+            .or(config.max_token_changes_per_connection)
+            .or_else(|| {
+                std::env::var("MAX_TOKEN_CHANGES_PER_CONNECTION")
+                    .ok()
+                    .and_then(|s| s.parse().ok())
+            })
+            .unwrap_or(0);
+
         let signature = match args.signature {
             Some(s) => {
                 if s.len() == 2 {
@@ -469,6 +491,7 @@ impl Configuration {
 
         Configuration {
             token,
+            max_token_changes_per_connection,
             tp_address,
             interval,
             delay,

src/ingress/sv1_ingress.rs

@@ -26,11 +26,7 @@ pub fn start_listen_for_downstream(
         let down_addr: String = Configuration::downstream_listening_addr()
             .unwrap_or(crate::DEFAULT_LISTEN_ADDRESS.to_string());
         let downstream_addr: SocketAddr = down_addr.parse().expect("Invalid listen address");
-        info!(
-            "Trying to bind to address {} for downstream(miner) connections",
-            downstream_addr
-        );
-        let downstream_listener = TcpListener::bind(downstream_addr)
+        let downstream_listener = bind_downstream_listener(downstream_addr)
             .await
             .expect("impossible to bind downstream");
         info!(
@@ -49,6 +45,15 @@ pub fn start_listen_for_downstream(
     })
     .into()
 }
+
+pub async fn bind_downstream_listener(downstream_addr: SocketAddr) -> std::io::Result<TcpListener> {
+    info!(
+        "Trying to bind to address {} for downstream(miner) connections",
+        downstream_addr
+    );
+    TcpListener::bind(downstream_addr).await
+}
+
 struct Downstream {}
 
 impl Downstream {

src/lib.rs

@@ -37,6 +37,8 @@ mod shared;
 mod shutdown;
 mod translator;
 
+pub use ingress::sv1_ingress::bind_downstream_listener;
+
 const TRANSLATOR_BUFFER_SIZE: usize = 32;
 const MIN_EXTRANONCE_SIZE: u16 = 6;
 const MIN_EXTRANONCE2_SIZE: u16 = 5;

src/translator/downstream/accept_connection.rs

@@ -26,6 +26,7 @@ pub async fn start_accept_connection(
     upstream_difficulty_config: Arc<Mutex<UpstreamDifficultyConfig>>,
     mut downstreams: Receiver<(Sender<String>, Receiver<String>, IpAddr)>,
     stats_sender: crate::api::stats::StatsSender,
+    shared_token_state: Arc<Mutex<super::downstream::TokenChangeTracker>>,
     tx_update_token: Sender<String>,
 ) -> Result<(), Error<'static>> {
     let handle = {
@@ -102,6 +103,7 @@ pub async fn start_accept_connection(
                             task_manager.clone(),
                             initial_difficulty,
                             stats_sender.clone(),
+                            shared_token_state.clone(),
                             tx_update_token.clone(),
                         )
                         .await

src/translator/downstream/diff_management.rs

@@ -470,6 +470,8 @@ mod test {
             Arc::new(Mutex::new(upstream_config)),
             crate::api::stats::StatsSender::new(),
             first_job,
+            String::new(),
+            1,
             tx_update_token,
         );
         downstream.difficulty_mgmt.estimated_downstream_hash_rate = start_hashrate as f32;