port translator to new handlers API

Author: Shourya742

roles/translator/src/lib/error.rs

@@ -13,7 +13,7 @@ use std::{fmt, sync::PoisonError};
 use stratum_common::roles_logic_sv2::{
     codec_sv2::{self, binary_sv2, framing_sv2},
     errors::Error as RolesLogicError,
-    handlers_sv2::HandlerError,
+    handlers_sv2::HandlerErrorType,
     parsers_sv2::ParserError as RolesParserError,
     Error as RolesSv2Error,
 };
@@ -30,8 +30,6 @@ pub enum TproxyError {
     RolesSv2LogicError(RolesSv2Error),
     /// Error from roles logic parser library
     ParserError(RolesParserError),
-    /// Error from roles logic handlers Library
-    RolesSv2LogicHandlerError(HandlerError),
     /// Errors on bad CLI argument input.
     BadCliArgs,
     /// Errors on bad `serde_json` serialize/deserialize.
@@ -63,7 +61,7 @@ pub enum TproxyError {
     /// Error converting SetDifficulty to Message
     SetDifficultyToMessage(SetDifficulty),
     /// Received an unexpected message type
-    UnexpectedMessage,
+    UnexpectedMessage(u8),
     /// Job not found during share validation
     JobNotFound,
     /// Invalid merkle root during share validation
@@ -104,8 +102,11 @@ impl fmt::Display for TproxyError {
             SetDifficultyToMessage(ref e) => {
                 write!(f, "Error converting SetDifficulty to Message: `{e:?}`")
             }
-            UnexpectedMessage => {
-                write!(f, "Received a message type that was not expected")
+            UnexpectedMessage(message_type) => {
+                write!(
+                    f,
+                    "Received a message type that was not expected: {message_type}"
+                )
             }
             JobNotFound => write!(f, "Job not found during share validation"),
             InvalidMerkleRoot => write!(f, "Invalid merkle root during share validation"),
@@ -118,7 +119,6 @@ impl fmt::Display for TproxyError {
             NetworkHelpersError(ref e) => write!(f, "Network helpers error: {e:?}"),
             RolesSv2LogicError(ref e) => write!(f, "Roles logic error: {e:?}"),
             ParserError(ref e) => write!(f, "Roles logic parser error: {e:?}"),
-            RolesSv2LogicHandlerError(ref e) => write!(f, "Roles logic handler error: {e:?}"),
         }
     }
 }
@@ -129,12 +129,6 @@ impl From<binary_sv2::Error> for TproxyError {
     }
 }
 
-impl From<HandlerError> for TproxyError {
-    fn from(value: HandlerError) -> Self {
-        TproxyError::RolesSv2LogicHandlerError(value)
-    }
-}
-
 impl From<codec_sv2::noise_sv2::Error> for TproxyError {
     fn from(e: codec_sv2::noise_sv2::Error) -> Self {
         TproxyError::CodecNoise(e)
@@ -214,8 +208,12 @@ impl From<stratum_translation::error::StratumTranslationError> for TproxyError {
     }
 }
 
-impl From<TproxyError> for HandlerError {
-    fn from(value: TproxyError) -> Self {
-        HandlerError::External(value.into())
+impl HandlerErrorType for TproxyError {
+    fn parse_error(error: RolesParserError) -> Self {
+        TproxyError::ParserError(error)
+    }
+
+    fn unexpected_message(message_type: u8) -> Self {
+        TproxyError::UnexpectedMessage(message_type)
     }
 }

roles/translator/src/lib/sv2/channel_manager/channel_manager.rs

@@ -210,7 +210,7 @@ impl ChannelManager {
         match message {
             Message::Mining(_) => {
                 channel_manager
-                    .handle_mining_message_from_server(message_type, &mut payload)
+                    .handle_mining_message_frame_from_server(message_type, &mut payload)
                     .await?;
             }
             _ => {

roles/translator/src/lib/sv2/channel_manager/message_handler.rs

@@ -7,9 +7,7 @@ use crate::{
 };
 use stratum_common::roles_logic_sv2::{
     channels_sv2::client::extended::ExtendedChannel,
-    handlers_sv2::{
-        HandleMiningMessagesFromServerAsync, HandlerError as Error, SupportedChannelTypes,
-    },
+    handlers_sv2::{HandleMiningMessagesFromServerAsync, SupportedChannelTypes},
     mining_sv2::{
         CloseChannel, ExtendedExtranonce, Extranonce, NewExtendedMiningJob, NewMiningJob,
         OpenExtendedMiningChannelSuccess, OpenMiningChannelError, OpenStandardMiningChannelSuccess,
@@ -26,6 +24,8 @@ use stratum_common::roles_logic_sv2::{
 use tracing::{error, info, warn};
 
 impl HandleMiningMessagesFromServerAsync for ChannelManager {
+    type Error = TproxyError;
+
     fn get_channel_type_for_server(&self) -> SupportedChannelTypes {
         SupportedChannelTypes::Extended
     }
@@ -37,17 +37,17 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
     async fn handle_open_standard_mining_channel_success(
         &mut self,
         m: OpenStandardMiningChannelSuccess<'_>,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Self::Error> {
         warn!("Received: {}", m);
-        Err(Error::UnexpectedMessage(
+        Err(Self::Error::UnexpectedMessage(
             MESSAGE_TYPE_OPEN_STANDARD_MINING_CHANNEL_SUCCESS,
         ))
     }
 
     async fn handle_open_extended_mining_channel_success(
         &mut self,
         m: OpenExtendedMiningChannelSuccess<'_>,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Self::Error> {
         // Check if we have the pending channel data, return error if not
         let (user_identity, nominal_hashrate, downstream_extranonce_len) = self
             .channel_manager_data
@@ -56,11 +56,11 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
             })
             .map_err(|e| {
                 error!("Failed to lock channel manager data: {:?}", e);
-                Error::from(TproxyError::PoisonLock)
+                TproxyError::PoisonLock
             })?
             .ok_or_else(|| {
                 error!("No pending channel found for request_id: {}", m.request_id);
-                Error::from(TproxyError::PendingChannelNotFound(m.request_id))
+                TproxyError::PendingChannelNotFound(m.request_id)
             })?;
 
         let success = self
@@ -162,7 +162,7 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
             })
             .map_err(|e| {
                 error!("Failed to lock channel manager data: {:?}", e);
-                Error::from(TproxyError::PoisonLock)
+                TproxyError::PoisonLock
             })?;
 
         self.channel_state
@@ -171,7 +171,7 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
             .await
             .map_err(|e| {
                 error!("Failed to send OpenExtendedMiningChannelSuccess: {:?}", e);
-                Error::from(TproxyError::ChannelErrorSender)
+                TproxyError::ChannelErrorSender
             })?;
 
         Ok(())
@@ -180,20 +180,20 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
     async fn handle_open_mining_channel_error(
         &mut self,
         m: OpenMiningChannelError<'_>,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Self::Error> {
         warn!("Received: {}", m);
         todo!("OpenMiningChannelError not handled yet");
     }
 
     async fn handle_update_channel_error(
         &mut self,
         m: UpdateChannelError<'_>,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Self::Error> {
         warn!("Received: {}", m);
         Ok(())
     }
 
-    async fn handle_close_channel(&mut self, m: CloseChannel<'_>) -> Result<(), Error> {
+    async fn handle_close_channel(&mut self, m: CloseChannel<'_>) -> Result<(), Self::Error> {
         info!("Received: {}", m);
         _ = self.channel_manager_data.safe_lock(|channel_data_manager| {
             if channel_data_manager.mode == ChannelMode::Aggregated {
@@ -210,23 +210,29 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
     async fn handle_set_extranonce_prefix(
         &mut self,
         m: SetExtranoncePrefix<'_>,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Self::Error> {
         warn!("Received: {}", m);
         warn!("⚠️ Cannot process SetExtranoncePrefix since set_extranonce is not supported for majority of sv1 clients. Ignoring.");
         Ok(())
     }
 
-    async fn handle_submit_shares_success(&mut self, m: SubmitSharesSuccess) -> Result<(), Error> {
+    async fn handle_submit_shares_success(
+        &mut self,
+        m: SubmitSharesSuccess,
+    ) -> Result<(), Self::Error> {
         info!("Received: {} ✅", m);
         Ok(())
     }
 
-    async fn handle_submit_shares_error(&mut self, m: SubmitSharesError<'_>) -> Result<(), Error> {
+    async fn handle_submit_shares_error(
+        &mut self,
+        m: SubmitSharesError<'_>,
+    ) -> Result<(), Self::Error> {
         warn!("Received: {} ❌", m.channel_id);
         Ok(())
     }
 
-    async fn handle_new_mining_job(&mut self, m: NewMiningJob<'_>) -> Result<(), Error> {
+    async fn handle_new_mining_job(&mut self, m: NewMiningJob<'_>) -> Result<(), Self::Error> {
         warn!("Received: {}", m);
         warn!("⚠️ Cannot process NewMiningJob since Translator Proxy supports only extended mining jobs. Ignoring.");
         Ok(())
@@ -235,7 +241,7 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
     async fn handle_new_extended_mining_job(
         &mut self,
         m: NewExtendedMiningJob<'_>,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Self::Error> {
         info!("Received: {}", m);
         let mut m_static = m.clone().into_static();
         _ = self.channel_manager_data.safe_lock(|channel_manage_data| {
@@ -274,13 +280,13 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
                 .await
                 .map_err(|e| {
                     error!("Failed to send immediate NewExtendedMiningJob: {:?}", e);
-                    Error::from(TproxyError::ChannelErrorSender)
+                    TproxyError::ChannelErrorSender
                 })?;
         }
         Ok(())
     }
 
-    async fn handle_set_new_prev_hash(&mut self, m: SetNewPrevHash<'_>) -> Result<(), Error> {
+    async fn handle_set_new_prev_hash(&mut self, m: SetNewPrevHash<'_>) -> Result<(), Self::Error> {
         info!("Received: {}", m);
         let m_static = m.clone().into_static();
         _ = self.channel_manager_data.safe_lock(|channel_manager_data| {
@@ -314,7 +320,7 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
             .await
             .map_err(|e| {
                 error!("Failed to send SetNewPrevHash: {:?}", e);
-                Error::from(TproxyError::ChannelErrorSender)
+                TproxyError::ChannelErrorSender
             })?;
 
         let mode = self
@@ -347,7 +353,7 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
                 .await
                 .map_err(|e| {
                     error!("Failed to send NewExtendedMiningJob: {:?}", e);
-                    Error::from(TproxyError::ChannelErrorSender)
+                    TproxyError::ChannelErrorSender
                 })?;
         }
         Ok(())
@@ -356,26 +362,26 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
     async fn handle_set_custom_mining_job_success(
         &mut self,
         m: SetCustomMiningJobSuccess,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Self::Error> {
         warn!("Received: {}", m);
         warn!("⚠️ Cannot process SetCustomMiningJobSuccess since Translator Proxy does not support custom mining jobs. Ignoring.");
-        Err(Error::UnexpectedMessage(
+        Err(Self::Error::UnexpectedMessage(
             MESSAGE_TYPE_SET_CUSTOM_MINING_JOB_SUCCESS,
         ))
     }
 
     async fn handle_set_custom_mining_job_error(
         &mut self,
         m: SetCustomMiningJobError<'_>,
-    ) -> Result<(), Error> {
+    ) -> Result<(), Self::Error> {
         warn!("Received: {}", m);
         warn!("⚠️ Cannot process SetCustomMiningJobError since Translator Proxy does not support custom mining jobs. Ignoring.");
-        Err(Error::UnexpectedMessage(
+        Err(Self::Error::UnexpectedMessage(
             MESSAGE_TYPE_SET_CUSTOM_MINING_JOB_ERROR,
         ))
     }
 
-    async fn handle_set_target(&mut self, m: SetTarget<'_>) -> Result<(), Error> {
+    async fn handle_set_target(&mut self, m: SetTarget<'_>) -> Result<(), Self::Error> {
         info!("Received: {}", m);
 
         // Update the channel targets in the channel manager
@@ -409,15 +415,20 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
             .await
             .map_err(|e| {
                 error!("Failed to forward SetTarget message to SV1Server: {:?}", e);
-                Error::from(TproxyError::ChannelErrorSender)
+                TproxyError::ChannelErrorSender
             })?;
 
         Ok(())
     }
 
-    async fn handle_set_group_channel(&mut self, m: SetGroupChannel<'_>) -> Result<(), Error> {
+    async fn handle_set_group_channel(
+        &mut self,
+        m: SetGroupChannel<'_>,
+    ) -> Result<(), Self::Error> {
         warn!("Received: {}", m);
         warn!("⚠️ Cannot process SetGroupChannel since Translator Proxy does not support group channels. Ignoring.");
-        Err(Error::UnexpectedMessage(MESSAGE_TYPE_SET_GROUP_CHANNEL))
+        Err(Self::Error::UnexpectedMessage(
+            MESSAGE_TYPE_SET_GROUP_CHANNEL,
+        ))
     }
 }

roles/translator/src/lib/sv2/upstream/message_handler.rs

@@ -1,38 +1,40 @@
-use crate::sv2::Upstream;
+use crate::{error::TproxyError, sv2::Upstream};
 use stratum_common::roles_logic_sv2::{
     common_messages_sv2::{
         ChannelEndpointChanged, Reconnect, SetupConnectionError, SetupConnectionSuccess,
     },
-    handlers_sv2::{HandleCommonMessagesFromServerAsync, HandlerError},
+    handlers_sv2::HandleCommonMessagesFromServerAsync,
 };
 use tracing::{error, info};
 
 impl HandleCommonMessagesFromServerAsync for Upstream {
+    type Error = TproxyError;
+
     async fn handle_setup_connection_error(
         &mut self,
         msg: SetupConnectionError<'_>,
-    ) -> Result<(), HandlerError> {
+    ) -> Result<(), Self::Error> {
         error!("Received: {}", msg);
         todo!()
     }
 
     async fn handle_setup_connection_success(
         &mut self,
         msg: SetupConnectionSuccess,
-    ) -> Result<(), HandlerError> {
+    ) -> Result<(), Self::Error> {
         info!("Received: {}", msg);
         Ok(())
     }
 
     async fn handle_channel_endpoint_changed(
         &mut self,
         msg: ChannelEndpointChanged,
-    ) -> Result<(), HandlerError> {
+    ) -> Result<(), Self::Error> {
         info!("Received: {}", msg);
         todo!()
     }
 
-    async fn handle_reconnect(&mut self, msg: Reconnect<'_>) -> Result<(), HandlerError> {
+    async fn handle_reconnect(&mut self, msg: Reconnect<'_>) -> Result<(), Self::Error> {
         info!("Received: {}", msg);
         todo!()
     }

roles/translator/src/lib/sv2/upstream/upstream.rs

@@ -242,7 +242,7 @@ impl Upstream {
 
         let payload = incoming.payload();
 
-        self.handle_common_message_from_server(message_type, payload)
+        self.handle_common_message_frame_from_server(message_type, payload)
             .await?;
         debug!("Upstream: handshake completed successfully.");
         Ok(())
@@ -273,7 +273,7 @@ impl Upstream {
                     AnyMessage::Common(_) => {
                         // Handle common upstream messages
                         upstream
-                            .handle_common_message_from_server(messsage_type, &mut payload)
+                            .handle_common_message_frame_from_server(messsage_type, &mut payload)
                             .await?;
                     }
 
@@ -292,7 +292,7 @@ impl Upstream {
 
                     _ => {
                         error!("Received unsupported message type from upstream.");
-                        return Err(TproxyError::UnexpectedMessage);
+                        return Err(TproxyError::UnexpectedMessage(0));
                     }
                 }
             }