feat: defer SV2 channel opening until first SV1 message received

- Move channel opening from connection acceptance to first message receipt
- Cache Sv1 handshake messages until SV2 channel is established with proper extranonce
- Send cached response only after channel setup is complete

Author: GitGab19

roles/translator/src/lib/sv1/downstream/data.rs

@@ -28,6 +28,10 @@ pub struct DownstreamData {
     pub pending_hashrate: Option<f32>,
     // Flag to track if SV1 handshake is complete (subscribe + authorize)
     pub sv1_handshake_complete: AtomicBool,
+    // Queue of Sv1 handshake messages received while waiting for SV2 channel to open
+    pub queued_sv1_handshake_messages: Vec<json_rpc::Message>,
+    // Flag to indicate we're processing queued Sv1 handshake message responses
+    pub processing_queued_sv1_handshake_responses: AtomicBool,
     // Stores pending shares to be sent to the sv1_server
     pub pending_share: RefCell<Option<SubmitShareWithChannelId>>,
     // Reference to shared sv1_server data for accessing valid_jobs during downstream sv1
@@ -61,6 +65,8 @@ impl DownstreamData {
             pending_target: None,
             pending_hashrate: None,
             sv1_handshake_complete: AtomicBool::new(false),
+            queued_sv1_handshake_messages: Vec::new(),
+            processing_queued_sv1_handshake_responses: AtomicBool::new(false),
             pending_share: RefCell::new(None),
             sv1_server_data,
             upstream_target: None,

roles/translator/src/lib/sv1/downstream/downstream.rs

@@ -186,16 +186,21 @@ impl Downstream {
                                 .load(std::sync::atomic::Ordering::SeqCst),
                         )
                     });
-
                 let id_matches = (my_channel_id == Some(channel_id) || channel_id == 0)
                     && (downstream_id.is_none() || downstream_id == Some(my_downstream_id));
-
                 if !id_matches {
                     return Ok(()); // Message not intended for this downstream
                 }
 
-                // Sv1 handshake complete - send messages immediately
-                if handshake_complete {
+                // Check if this is a queued message response
+                let is_queued_sv1_handshake_response = self.downstream_data.super_safe_lock(|d| {
+                    d.processing_queued_sv1_handshake_responses
+                        .load(std::sync::atomic::Ordering::SeqCst)
+                });
+
+                // Sv1 handshake complete - send messages immediately, or if it's a queued
+                // Sv1 handshake message response
+                if handshake_complete || is_queued_sv1_handshake_response {
                     if let Message::Notification(notification) = &message {
                         match notification.method.as_str() {
                             "mining.set_difficulty" => {
@@ -349,37 +354,67 @@ impl Downstream {
             }
         };
 
+        // Check if channel is established
+        let channel_established = self
+            .downstream_data
+            .super_safe_lock(|d| d.channel_id.is_some());
+
+        if !channel_established {
+            // Check if this is the first message (queue is empty) and send OpenChannel request
+            let is_first_message = self
+                .downstream_data
+                .super_safe_lock(|d| d.queued_sv1_handshake_messages.is_empty());
+
+            if is_first_message {
+                let downstream_id = self.downstream_data.super_safe_lock(|d| d.downstream_id);
+                self.downstream_channel_state
+                    .sv1_server_sender
+                    .send(DownstreamMessages::OpenChannel(downstream_id))
+                    .await
+                    .map_err(|e| {
+                        error!("Down: Failed to send OpenChannel request: {:?}", e);
+                        TproxyError::ChannelErrorSender
+                    })?;
+                debug!(
+                    "Down: Sent OpenChannel request for downstream {}",
+                    downstream_id
+                );
+            }
+
+            // Queue all messages until channel is established
+            debug!("Down: Queuing Sv1 message until channel is established");
+            self.downstream_data.safe_lock(|d| {
+                d.queued_sv1_handshake_messages.push(message.clone());
+            })?;
+            return Ok(());
+        }
+
+        // Channel is established, process message normally
         let response = self
             .downstream_data
             .super_safe_lock(|data| data.handle_message(message.clone()));
 
         match response {
             Ok(Some(response_msg)) => {
-                if self
-                    .downstream_data
-                    .super_safe_lock(|d| d.channel_id)
-                    .is_some()
-                {
-                    debug!(
-                        "Down: Sending SV1 message to downstream: {:?}",
-                        response_msg
-                    );
-                    self.downstream_channel_state
-                        .downstream_sv1_sender
-                        .send(response_msg.into())
-                        .await
-                        .map_err(|e| {
-                            error!("Down: Failed to send message to downstream: {:?}", e);
-                            TproxyError::ChannelErrorSender
-                        })?;
+                debug!(
+                    "Down: Sending Sv1 message to downstream: {:?}",
+                    response_msg
+                );
+                self.downstream_channel_state
+                    .downstream_sv1_sender
+                    .send(response_msg.into())
+                    .await
+                    .map_err(|e| {
+                        error!("Down: Failed to send message to downstream: {:?}", e);
+                        TproxyError::ChannelErrorSender
+                    })?;
 
-                    // Check if this was an authorize message and handle sv1 handshake completion
-                    if let v1::json_rpc::Message::StandardRequest(request) = &message {
-                        if request.method == "mining.authorize" {
-                            if let Err(e) = self.handle_sv1_handshake_completion().await {
-                                error!("Down: Failed to handle handshake completion: {:?}", e);
-                                return Err(e);
-                            }
+                // Check if this was an authorize message and handle sv1 handshake completion
+                if let v1::json_rpc::Message::StandardRequest(request) = &message {
+                    if request.method == "mining.authorize" {
+                        if let Err(e) = self.handle_sv1_handshake_completion().await {
+                            error!("Down: Failed to handle handshake completion: {:?}", e);
+                            return Err(e);
                         }
                     }
                 }

roles/translator/src/lib/sv1/downstream/message_handler.rs

@@ -10,7 +10,7 @@ use crate::{
     utils::validate_sv1_share,
 };
 
-// Implements `IsServer` for `Downstream` to handle the SV1 messages.
+// Implements `IsServer` for `Downstream` to handle the Sv1 messages.
 impl IsServer<'static> for DownstreamData {
     fn handle_configure(
         &mut self,

roles/translator/src/lib/sv1/downstream/mod.rs

@@ -14,6 +14,9 @@ pub enum DownstreamMessages {
     /// Represents a submitted share from a downstream miner,
     /// wrapped with the relevant channel ID.
     SubmitShares(SubmitShareWithChannelId),
+    /// Request to open an extended mining channel for a downstream that just sent its first
+    /// message.
+    OpenChannel(u32), // downstream_id
 }
 
 /// A wrapper around a `mining.submit` message with additional channel information.

roles/translator/src/lib/sv1/sv1_server/sv1_server.rs

@@ -40,6 +40,7 @@ use tokio::{
     sync::{broadcast, mpsc},
 };
 use tracing::{debug, error, info, warn};
+use v1::IsServer;
 
 /// SV1 server that handles connections from SV1 miners.
 ///
@@ -281,11 +282,21 @@ impl Sv1Server {
                                         d.vardiff.insert(downstream_id, vardiff);
                                     }
                                 });
-                            info!("Downstream {} registered successfully", downstream_id);
+                            info!("Downstream {} registered successfully (channel will be opened after first message)", downstream_id);
 
-                            self
-                                .open_extended_mining_channel(downstream.clone())
-                                .await?;
+                            // Start downstream tasks immediately, but defer channel opening until first message
+                            let status_sender = StatusSender::Downstream {
+                                downstream_id,
+                                tx: status_sender.clone(),
+                            };
+
+                            Downstream::run_downstream_tasks(
+                                downstream,
+                                notify_shutdown.clone(),
+                                shutdown_complete_tx.clone(),
+                                status_sender,
+                                task_manager.clone(),
+                            );
                         }
                         Err(e) => {
                             warn!("Failed to accept new connection: {:?}", e);
@@ -303,10 +314,6 @@ impl Sv1Server {
                 res = Self::handle_upstream_message(
                     Arc::clone(&self),
                     first_target.clone(),
-                    notify_shutdown.clone(),
-                    shutdown_complete_tx_main_clone.clone(),
-                    status_sender.clone(),
-                    task_manager.clone()
                 ) => {
                     if let Err(e) = res {
                         handle_error(&sv1_status_sender, e).await;
@@ -340,8 +347,21 @@ impl Sv1Server {
             .await
             .map_err(TproxyError::ChannelErrorReceiver)?;
 
-        let DownstreamMessages::SubmitShares(message) = downstream_message;
+        match downstream_message {
+            DownstreamMessages::SubmitShares(message) => {
+                return self.handle_submit_shares(message).await;
+            }
+            DownstreamMessages::OpenChannel(downstream_id) => {
+                return self.handle_open_channel_request(downstream_id).await;
+            }
+        }
+    }
 
+    /// Handles share submission messages from downstream.
+    async fn handle_submit_shares(
+        self: &Arc<Self>,
+        message: crate::sv1::downstream::SubmitShareWithChannelId,
+    ) -> Result<(), TproxyError> {
         // Increment vardiff counter for this downstream (only if vardiff is enabled)
         if self.config.downstream_difficulty_config.enable_vardiff {
             self.sv1_server_data.safe_lock(|v| {
@@ -382,6 +402,28 @@ impl Sv1Server {
         Ok(())
     }
 
+    /// Handles channel opening requests from downstream when they send their first message.
+    async fn handle_open_channel_request(
+        self: &Arc<Self>,
+        downstream_id: u32,
+    ) -> Result<(), TproxyError> {
+        info!("SV1 Server: Opening extended mining channel for downstream {} after receiving first message", downstream_id);
+
+        let downstreams = self
+            .sv1_server_data
+            .super_safe_lock(|v| v.downstreams.clone());
+        if let Some(downstream) = Self::get_downstream(downstream_id, downstreams) {
+            self.open_extended_mining_channel(downstream).await?;
+        } else {
+            error!(
+                "Downstream {} not found when trying to open channel",
+                downstream_id
+            );
+        }
+
+        Ok(())
+    }
+
     /// Handles messages received from the upstream SV2 server via the channel manager.
     ///
     /// This method processes various SV2 messages including:
@@ -403,10 +445,6 @@ impl Sv1Server {
     pub async fn handle_upstream_message(
         self: Arc<Self>,
         first_target: Target,
-        notify_shutdown: broadcast::Sender<ShutdownMessage>,
-        shutdown_complete_tx: mpsc::Sender<()>,
-        status_sender: Sender<Status>,
-        task_manager: Arc<TaskManager>,
     ) -> Result<(), TproxyError> {
         let message = self
             .sv1_server_channel_state
@@ -435,18 +473,42 @@ impl Sv1Server {
                         d.set_upstream_target(initial_target.clone());
                     })?;
 
-                    let status_sender = StatusSender::Downstream {
-                        downstream_id,
-                        tx: status_sender.clone(),
-                    };
-
-                    Downstream::run_downstream_tasks(
-                        downstream,
-                        notify_shutdown,
-                        shutdown_complete_tx,
-                        status_sender,
-                        task_manager,
-                    );
+                    // Process all queued messages now that channel is established
+                    if let Ok(queued_messages) = downstream.downstream_data.safe_lock(|d| {
+                        let messages = d.queued_sv1_handshake_messages.clone();
+                        d.queued_sv1_handshake_messages.clear();
+                        messages
+                    }) {
+                        if !queued_messages.is_empty() {
+                            info!(
+                                "Processing {} queued Sv1 messages for downstream {}",
+                                queued_messages.len(),
+                                downstream_id
+                            );
+
+                            // Set flag to indicate we're processing queued responses
+                            downstream.downstream_data.super_safe_lock(|data| {
+                                data.processing_queued_sv1_handshake_responses
+                                    .store(true, std::sync::atomic::Ordering::SeqCst);
+                            });
+
+                            for message in queued_messages {
+                                if let Ok(Some(response_msg)) = downstream
+                                    .downstream_data
+                                    .super_safe_lock(|data| data.handle_message(message))
+                                {
+                                    self.sv1_server_channel_state
+                                        .sv1_server_to_downstream_sender
+                                        .send((
+                                            m.channel_id,
+                                            Some(downstream_id),
+                                            response_msg.into(),
+                                        ))
+                                        .map_err(|_| TproxyError::ChannelErrorSender)?;
+                                }
+                            }
+                        }
+                    }
 
                     let set_difficulty = build_sv1_set_difficulty_from_sv2_target(first_target)
                         .map_err(|_| {