feat(translator): new modular translator proxy with channel aggregation and failover

Replace existing translator proxy with new implementation featuring:
- Aggregated/non-aggregated channel modes (shared vs individual upstream channels)
- Stale share rejection with early SV1 share validation
- Automatic failover to secondary upstream on connection issues
- Improved modularity and extensibility for future features
- Integration with newer client-channel APIs from stratum-mining#1724

Co-authored-by: bit-aloo <[email protected]>
Co-authored-by: plebhash <[email protected]>

Author: 3 people

common/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "sv1_api"
-version = "2.0.0"
+version = "2.1.0"
 authors = ["The Stratum V2 Developers"]
 edition = "2021"
 readme = "README.md"

protocols/v1/Cargo.toml

@@ -31,6 +31,8 @@ pub enum Error<'a> {
     /// Errors if server does not recognize the client's `id`.
     UnknownID(u64),
     InvalidVersionMask(HexU32Be),
+    /// Errors when an unexpected or unsupported message/method is called.
+    UnexpectedMessage(String),
 }
 
 impl std::fmt::Display for Error<'_> {
@@ -70,6 +72,7 @@ impl std::fmt::Display for Error<'_> {
             ),
             Error::UnknownID(e) => write!(f, "Server did not recognize the client id: `{e}`."),
             Error::InvalidVersionMask(e) => write!(f, "First 3 bits of version rolling mask must be 0 and last 13 bits of version rolling mask must be 0. Version rolling mask is: `{:b}`.", e.0),
+            Error::UnexpectedMessage(method) => write!(f, "Unexpected or unsupported message/method called: `{method}`."),
         }
     }
 }

protocols/v1/src/error.rs

@@ -33,16 +33,13 @@ impl TaskManager {
     ///
     /// # Arguments
     /// * `fut` - The future to spawn as a task
-    #[track_caller]
     pub fn spawn<F>(&self, fut: F)
     where
         F: std::future::Future<Output = ()> + Send + 'static,
     {
-        use tracing::Instrument;
-        let location = std::panic::Location::caller();
-        let span = tracing::trace_span!("task", file = location.file(),);
-
-        let handle = tokio::spawn(fut.instrument(span));
+        let handle = tokio::spawn(async move {
+            fut.await;
+        });
         self.tasks.lock().unwrap().push(handle);
     }
 

roles/Cargo.lock

@@ -6,6 +6,7 @@ use tokio::{
     net::TcpStream,
 };
 use tokio_util::codec::{FramedRead, LinesCodec};
+use tracing::{error, trace, warn};
 
 /// Represents a connection between two roles communicating using SV1 protocol.
 ///
@@ -70,11 +71,11 @@ impl ConnectionSV1 {
         tokio::spawn(async move {
             tokio::select! {
                 _ = Self::run_reader(buffer_read_half, sender_incoming.clone()) => {
-                    tracing::info!("Reader task exited. Closing writer sender.");
+                    trace!("Reader task exited. Closing writer sender.");
                     connection_state.close();
                 }
                 _ = Self::run_writer(buffer_write_half, receiver_outgoing.clone()) => {
-                    tracing::info!("Writer task exited.Closing reader sender.");
+                    trace!("Writer task exited. Closing reader sender.");
                     connection_state.close();
                 }
             }
@@ -96,16 +97,16 @@ impl ConnectionSV1 {
                 Ok(line) => match serde_json::from_str::<json_rpc::Message>(&line) {
                     Ok(msg) => {
                         if sender.send(msg).await.is_err() {
-                            tracing::warn!("Receiver dropped, stopping reader");
+                            warn!("Receiver dropped, stopping reader");
                             break;
                         }
                     }
                     Err(e) => {
-                        tracing::error!("Failed to deserialize message: {e:?}");
+                        error!("Failed to deserialize message: {e:?}");
                     }
                 },
                 Err(e) => {
-                    tracing::error!("Error reading from stream: {e:?}");
+                    error!("Error reading from stream: {e:?}");
                     break;
                 }
             }
@@ -121,16 +122,16 @@ impl ConnectionSV1 {
                 Ok(line) => {
                     let data = format!("{line}\n");
                     if writer.write_all(data.as_bytes()).await.is_err() {
-                        tracing::error!("Failed to write to stream");
+                        error!("Failed to write to stream");
                         break;
                     }
                     if writer.flush().await.is_err() {
-                        tracing::error!("Failed to flush writer.");
+                        error!("Failed to flush writer.");
                         break;
                     }
                 }
                 Err(e) => {
-                    tracing::error!("Failed to serialize message: {e:?}");
+                    error!("Failed to serialize message: {e:?}");
                     break;
                 }
             }

roles/jd-client/src/lib/task_manager.rs

@@ -1,15 +1,15 @@
 [package]
 name = "translator_sv2"
-version = "1.0.0"
+version = "2.0.0"
 authors = ["The Stratum V2 Developers"]
 edition = "2021"
-description = "Server used to bridge SV1 miners to SV2 pools"
+description = "SV1 to SV2 translation proxy"
 documentation = "https://docs.rs/translator_sv2"
 readme = "README.md"
 homepage = "https://stratumprotocol.org"
 repository = "https://github.com/stratum-mining/stratum"
 license = "MIT OR Apache-2.0"
-keywords = ["stratum", "mining", "bitcoin", "protocol"]
+keywords = ["stratum", "mining", "bitcoin", "protocol", "translator", "proxy"]
 
 [lib]
 name = "translator_sv2"
@@ -20,23 +20,18 @@ name = "translator_sv2"
 path = "src/main.rs"
 
 [dependencies]
-stratum-common = { path = "../../common", features = ["with_network_helpers"] }
+stratum-common = { path = "../../common" }
 async-channel = "1.5.1"
-async-recursion = "0.3.2"
 buffer_sv2 = { path = "../../utils/buffer" }
-once_cell = "1.12.0"
+network_helpers_sv2 = { path = "../roles-utils/network-helpers", features=["with_buffer_pool", "sv1"] }
+stratum_translation = { path = "../roles-utils/stratum-translation" }
 serde = { version = "1.0.89", default-features = false, features = ["derive", "alloc"] }
 serde_json = { version = "1.0.64", default-features = false, features = ["alloc"] }
-futures = "0.3.25"
 tokio = { version = "1.44.1", features = ["full"] }
 ext-config = { version = "0.14.0", features = ["toml"], package = "config" }
 tracing = { version = "0.1" }
 v1 = { path = "../../protocols/v1", package="sv1_api" }
-error_handling = { path = "../../utils/error-handling" }
 key-utils = { path = "../../utils/key-utils" }
-tokio-util = { version = "0.7.10", features = ["codec"] }
-rand = "0.8.4"
-primitive-types = "0.13.1"
 clap = { version = "4.5.39", features = ["derive"] }
 config_helpers_sv2 = { path = "../roles-utils/config-helpers" }