Merge pull request dmnd-pool#130 from Priceless-P/fix/downstream-disconnect

fix

Author: jbesraa

src/main.rs

@@ -239,7 +239,8 @@ async fn initialize_proxy(
             abort_handles.push((jdc_handle, "jdc".to_string()));
         }
         let server_handle = tokio::spawn(api::start(router.clone(), stats_sender));
-        match monitor(router, abort_handles, epsilon, server_handle).await {
+        abort_handles.push((server_handle.into(), "api_server".to_string()));
+        match monitor(router, abort_handles, epsilon).await {
             Reconnect::NewUpstream(new_pool_addr) => {
                 ProxyState::update_proxy_state_up();
                 pool_addr = Some(new_pool_addr);
@@ -258,7 +259,6 @@ async fn monitor(
     router: &mut Router,
     abort_handles: Vec<(AbortOnDrop, std::string::String)>,
     epsilon: Duration,
-    server_handle: tokio::task::JoinHandle<()>,
 ) -> Reconnect {
     let mut should_check_upstreams_latency = 0;
     loop {
@@ -269,7 +269,6 @@ async fn monitor(
                 if let Some(new_upstream) = router.monitor_upstream(epsilon).await {
                     info!("Faster upstream detected. Reinitializing proxy...");
                     drop(abort_handles);
-                    server_handle.abort(); // abort server
 
                     // Needs a little to time to drop
                     tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
@@ -288,7 +287,6 @@ async fn monitor(
             for (handle, _name) in abort_handles {
                 drop(handle);
             }
-            server_handle.abort(); // abort server
 
             // Check if the proxy state is down, and if so, reinitialize the proxy.
             let is_proxy_down = ProxyState::is_proxy_down();
@@ -311,7 +309,6 @@ async fn monitor(
                 is_proxy_down.1.unwrap_or("Proxy".to_string())
             );
             drop(abort_handles); // Drop all abort handles
-            server_handle.abort(); // abort server
             tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; // Needs a little to time to drop
             return Reconnect::NoUpstream;
         }

src/translator/downstream/receive_from_downstream.rs

@@ -1,8 +1,5 @@
 use super::{downstream::Downstream, task_manager::TaskManager};
-use crate::{
-    proxy_state::{DownstreamType, ProxyState},
-    translator::error::Error,
-};
+use crate::{proxy_state::ProxyState, translator::error::Error};
 use roles_logic_sv2::utils::Mutex;
 use std::sync::Arc;
 use sv1_api::{client_to_server::Submit, json_rpc};
@@ -36,7 +33,7 @@ pub async fn start_receive_downstream(
                         Downstream::handle_incoming_sv1(downstream.clone(), incoming).await
                     {
                         error!("Failed to handle incoming sv1 msg: {:?}", error);
-                        ProxyState::update_downstream_state(DownstreamType::TranslatorDownstream);
+                        break;
                     };
                 } else {
                     // Message received could not be converted to rpc message