IT adaptations for extensions configs in tProxy, JDC, and Pool

Author: GitGab19

integration-tests/Cargo.lock

@@ -67,7 +67,11 @@ pub fn start_sniffer(
     (sniffer, listening_address)
 }
 
-pub async fn start_pool(template_provider_address: Option<SocketAddr>) -> (PoolSv2, SocketAddr) {
+pub async fn start_pool(
+    template_provider_address: Option<SocketAddr>,
+    supported_extensions: Vec<u16>,
+    required_extensions: Vec<u16>,
+) -> (PoolSv2, SocketAddr) {
     use pool_sv2::config::PoolConfig;
     let listening_address = get_available_address();
     let authority_public_key = Secp256k1PublicKey::try_from(
@@ -106,6 +110,8 @@ pub async fn start_pool(template_provider_address: Option<SocketAddr>) -> (PoolS
         SHARES_PER_MINUTE,
         share_batch_size,
         1,
+        supported_extensions,
+        required_extensions,
     );
     let pool = PoolSv2::new(config);
     let pool_clone = pool.clone();
@@ -130,6 +136,8 @@ pub fn start_template_provider(
 pub fn start_jdc(
     pool: &[(SocketAddr, SocketAddr)], // (pool_address, jds_address)
     tp_address: SocketAddr,
+    supported_extensions: Vec<u16>,
+    required_extensions: Vec<u16>,
 ) -> (JobDeclaratorClient, SocketAddr) {
     use jd_client_sv2::config::{
         JobDeclaratorClientConfig, PoolConfig, ProtocolConfig, TPConfig, Upstream,
@@ -187,6 +195,8 @@ pub fn start_jdc(
         upstreams,
         jdc_signature,
         None,
+        supported_extensions,
+        required_extensions,
     );
     let ret = jd_client_sv2::JobDeclaratorClient::new(jd_client_proxy);
     let ret_clone = ret.clone();
@@ -247,6 +257,8 @@ pub fn start_jds(tp_rpc_connection: &ConnectParams) -> (JobDeclaratorServer, Soc
 pub async fn start_sv2_translator(
     upstream: SocketAddr,
     aggregate_channels: bool,
+    supported_extensions: Vec<u16>,
+    required_extensions: Vec<u16>,
 ) -> (TranslatorSv2, SocketAddr) {
     let upstream_address = upstream.ip().to_string();
     let upstream_port = upstream.port();
@@ -284,6 +296,8 @@ pub async fn start_sv2_translator(
         downstream_extranonce2_size,
         "user_identity".to_string(),
         aggregate_channels,
+        supported_extensions,
+        required_extensions,
     );
     let translator_v2 = translator_sv2::TranslatorSv2::new(config);
     let clone_translator_v2 = translator_v2.clone();

integration-tests/lib/mod.rs

@@ -334,6 +334,7 @@ pub fn into_static(m: AnyMessage<'_>) -> AnyMessage<'static> {
                 TemplateDistribution::SubmitSolution(m.into_static()),
             ),
         },
+        AnyMessage::Extensions(extensions) => AnyMessage::Extensions(extensions.into_static()),
     }
 }
 

integration-tests/lib/utils.rs

@@ -19,10 +19,10 @@ use stratum_apps::stratum_core::{
 async fn jds_should_not_panic_if_jdc_shutsdown() {
     start_tracing();
     let (tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low);
-    let (_pool, pool_addr) = start_pool(Some(tp_addr)).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr), vec![], vec![]).await;
     let (_jds, jds_addr) = start_jds(tp.rpc_info());
     let (sniffer_a, sniffer_addr_a) = start_sniffer("0", jds_addr, false, vec![], None);
-    let (jdc, jdc_addr) = start_jdc(&[(pool_addr, sniffer_addr_a)], tp_addr);
+    let (jdc, jdc_addr) = start_jdc(&[(pool_addr, sniffer_addr_a)], tp_addr, vec![], vec![]);
     sniffer_a
         .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
         .await;
@@ -36,7 +36,7 @@ async fn jds_should_not_panic_if_jdc_shutsdown() {
     tokio::time::sleep(tokio::time::Duration::from_millis(2000)).await;
     assert!(tokio::net::TcpListener::bind(jdc_addr).await.is_ok());
     let (sniffer, sniffer_addr) = start_sniffer("0", jds_addr, false, vec![], None);
-    let (_jdc_1, _jdc_addr_1) = start_jdc(&[(pool_addr, sniffer_addr)], tp_addr);
+    let (_jdc_1, _jdc_addr_1) = start_jdc(&[(pool_addr, sniffer_addr)], tp_addr, vec![], vec![]);
     sniffer
         .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
         .await;
@@ -50,13 +50,18 @@ async fn jds_should_not_panic_if_jdc_shutsdown() {
 async fn jdc_tp_success_setup() {
     start_tracing();
     let (tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low);
-    let (_pool, pool_addr) = start_pool(Some(tp_addr)).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr), vec![], vec![]).await;
     let (_jds, jds_addr) = start_jds(tp.rpc_info());
     let (tp_jdc_sniffer, tp_jdc_sniffer_addr) = start_sniffer("0", tp_addr, false, vec![], None);
-    let (_jdc, jdc_addr) = start_jdc(&[(pool_addr, jds_addr)], tp_jdc_sniffer_addr);
+    let (_jdc, jdc_addr) = start_jdc(
+        &[(pool_addr, jds_addr)],
+        tp_jdc_sniffer_addr,
+        vec![],
+        vec![],
+    );
     // This is needed because jd-client waits for a downstream connection before it starts
     // exchanging messages with the Template Provider.
-    start_sv2_translator(jdc_addr, false).await;
+    start_sv2_translator(jdc_addr, false, vec![], vec![]).await;
     tp_jdc_sniffer
         .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
         .await;
@@ -78,7 +83,7 @@ async fn jds_receive_solution_while_processing_declared_job_test() {
     start_tracing();
     let (tp_1, tp_addr_1) = start_template_provider(None, DifficultyLevel::Low);
     let (tp_2, tp_addr_2) = start_template_provider(None, DifficultyLevel::Low);
-    let (_pool, pool_addr) = start_pool(Some(tp_addr_1)).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr_1), vec![], vec![]).await;
     let (_jds, jds_addr) = start_jds(tp_1.rpc_info());
 
     let prev_hash = U256::Owned(vec![
@@ -111,8 +116,8 @@ async fn jds_receive_solution_while_processing_declared_job_test() {
         vec![submit_solution_replace.into()],
         None,
     );
-    let (_jdc, jdc_addr) = start_jdc(&[(pool_addr, sniffer_a_addr)], tp_addr_2);
-    let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr, false).await;
+    let (_jdc, jdc_addr) = start_jdc(&[(pool_addr, sniffer_a_addr)], tp_addr_2, vec![], vec![]);
+    let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr, false, vec![], vec![]).await;
     let (_minerd_process, _minerd_addr) = start_minerd(tproxy_addr, None, None, false).await;
     assert!(tp_2.fund_wallet().is_ok());
     assert!(tp_2.create_mempool_transaction().is_ok());
@@ -170,7 +175,7 @@ async fn jds_wont_exit_upon_receiving_unexpected_txids_in_provide_missing_transa
     assert!(tp_2.fund_wallet().is_ok());
     assert!(tp_2.create_mempool_transaction().is_ok());
 
-    let (_pool, pool_addr) = start_pool(Some(tp_addr_1)).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr_1), vec![], vec![]).await;
     let (_jds, jds_addr) = start_jds(tp_1.rpc_info());
 
     let provide_missing_transaction_success_replace = ReplaceMessage::new(
@@ -196,8 +201,8 @@ async fn jds_wont_exit_upon_receiving_unexpected_txids_in_provide_missing_transa
         None,
     );
 
-    let (_, jdc_addr_1) = start_jdc(&[(pool_addr, sniffer_addr)], tp_addr_2);
-    let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr_1, false).await;
+    let (_, jdc_addr_1) = start_jdc(&[(pool_addr, sniffer_addr)], tp_addr_2, vec![], vec![]);
+    let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr_1, false, vec![], vec![]).await;
     let (_minerd_process, _minerd_addr) = start_minerd(tproxy_addr, None, None, false).await;
 
     sniffer

integration-tests/tests/jd_integration.rs

@@ -6,11 +6,11 @@ async fn jds_ask_for_missing_transactions() {
     start_tracing();
     let (tp_1, tp_addr_1) = start_template_provider(None, DifficultyLevel::Low);
     let (tp_2, tp_addr_2) = start_template_provider(None, DifficultyLevel::Low);
-    let (_pool, pool_addr) = start_pool(Some(tp_addr_1)).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr_1), vec![], vec![]).await;
     let (_jds, jds_addr) = start_jds(tp_1.rpc_info());
     let (sniffer, sniffer_addr) = start_sniffer("A", jds_addr, false, vec![], None);
-    let (_jdc, jdc_addr) = start_jdc(&[(pool_addr, sniffer_addr)], tp_addr_2);
-    let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr, false).await;
+    let (_jdc, jdc_addr) = start_jdc(&[(pool_addr, sniffer_addr)], tp_addr_2, vec![], vec![]);
+    let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr, false, vec![], vec![]).await;
     let (_minerd_process, _minerd_addr) = start_minerd(tproxy_addr, None, None, false).await;
     assert!(tp_2.fund_wallet().is_ok());
     assert!(tp_2.create_mempool_transaction().is_ok());

integration-tests/tests/jd_provide_missing_transaction.rs

@@ -5,14 +5,20 @@ use stratum_apps::stratum_core::{common_messages_sv2::*, mining_sv2::*};
 async fn jd_non_aggregated_tproxy_integration() {
     start_tracing();
     let (tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low);
-    let (_pool, pool_addr) = start_pool(Some(tp_addr)).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr), vec![], vec![]).await;
     let (jdc_pool_sniffer, jdc_pool_sniffer_addr) =
         start_sniffer("0", pool_addr, false, vec![], None);
     let (_jds, jds_addr) = start_jds(tp.rpc_info());
-    let (_jdc, jdc_addr) = start_jdc(&[(jdc_pool_sniffer_addr, jds_addr)], tp_addr);
+    let (_jdc, jdc_addr) = start_jdc(
+        &[(jdc_pool_sniffer_addr, jds_addr)],
+        tp_addr,
+        vec![],
+        vec![],
+    );
     let (tproxy_jdc_sniffer, tproxy_jdc_sniffer_addr) =
         start_sniffer("1", jdc_addr, false, vec![], None);
-    let (_translator, tproxy_addr) = start_sv2_translator(tproxy_jdc_sniffer_addr, false).await;
+    let (_translator, tproxy_addr) =
+        start_sv2_translator(tproxy_jdc_sniffer_addr, false, vec![], vec![]).await;
 
     // start two minerd processes
     let (_minerd_process, _minerd_addr) = start_minerd(tproxy_addr, None, None, false).await;
@@ -74,14 +80,20 @@ async fn jd_non_aggregated_tproxy_integration() {
 async fn jd_aggregated_tproxy_integration() {
     start_tracing();
     let (tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low);
-    let (_pool, pool_addr) = start_pool(Some(tp_addr)).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr), vec![], vec![]).await;
     let (jdc_pool_sniffer, jdc_pool_sniffer_addr) =
         start_sniffer("0", pool_addr, false, vec![], None);
     let (_jds, jds_addr) = start_jds(tp.rpc_info());
-    let (_jdc, jdc_addr) = start_jdc(&[(jdc_pool_sniffer_addr, jds_addr)], tp_addr);
+    let (_jdc, jdc_addr) = start_jdc(
+        &[(jdc_pool_sniffer_addr, jds_addr)],
+        tp_addr,
+        vec![],
+        vec![],
+    );
     let (tproxy_jdc_sniffer, tproxy_jdc_sniffer_addr) =
         start_sniffer("1", jdc_addr, false, vec![], None);
-    let (_translator, tproxy_addr) = start_sv2_translator(tproxy_jdc_sniffer_addr, true).await;
+    let (_translator, tproxy_addr) =
+        start_sv2_translator(tproxy_jdc_sniffer_addr, true, vec![], vec![]).await;
 
     // start two minerd processes
     let (_minerd_process, _minerd_addr) = start_minerd(tproxy_addr, None, None, false).await;

integration-tests/tests/jd_tproxy_integration.rs

@@ -11,7 +11,7 @@ async fn propagated_from_jdc_to_tp() {
     start_tracing();
     let (tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low);
     let current_block_hash = tp.get_best_block_hash().unwrap();
-    let (_pool, pool_addr) = start_pool(Some(tp_addr)).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr), vec![], vec![]).await;
     let (_jds, jds_addr) = start_jds(tp.rpc_info());
     let ignore_push_solution =
         IgnoreMessage::new(MessageDirection::ToUpstream, MESSAGE_TYPE_PUSH_SOLUTION);
@@ -23,8 +23,13 @@ async fn propagated_from_jdc_to_tp() {
         None,
     );
     let (jdc_tp_sniffer, jdc_tp_sniffer_addr) = start_sniffer("1", tp_addr, false, vec![], None);
-    let (_jdc, jdc_addr) = start_jdc(&[(pool_addr, jdc_jds_sniffer_addr)], jdc_tp_sniffer_addr);
-    let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr, false).await;
+    let (_jdc, jdc_addr) = start_jdc(
+        &[(pool_addr, jdc_jds_sniffer_addr)],
+        jdc_tp_sniffer_addr,
+        vec![],
+        vec![],
+    );
+    let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr, false, vec![], vec![]).await;
     let (_minerd_process, _minerd_addr) = start_minerd(tproxy_addr, None, None, false).await;
     jdc_tp_sniffer
         .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SUBMIT_SOLUTION)