Merge pull request #1486 from jbesraa/2025-02-12/jdc-fallback-test

JDC Fallback test

Author: plebhash

roles/test-utils/mining-device-sv1/src/client.rs

@@ -244,7 +244,8 @@ impl Client {
             if let Ok(incoming) = recv_incoming.clone().recv().await {
                 Self::parse_message(client.clone(), Ok(incoming)).await;
             } else {
-                warn!("Error reading from socket via `recv_incoming` channel")
+                warn!("Error reading from socket via `recv_incoming` channel");
+                break;
             }
         }
     }

roles/tests-integration/lib/mod.rs

@@ -94,7 +94,7 @@ pub async fn start_pool(template_provider_address: Option<SocketAddr>) -> (PoolS
         assert!(pool_clone.start().await.is_ok());
     });
     // Wait a bit to let the pool exchange initial messages with the TP
-    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+    sleep(1).await;
     (pool, listening_address)
 }
 
@@ -107,9 +107,8 @@ pub fn start_template_provider(sv2_interval: Option<u32>) -> (TemplateProvider,
 }
 
 pub async fn start_jdc(
-    pool_address: SocketAddr,
+    pool: &[(SocketAddr, SocketAddr)], // (pool_address, jds_address)
     tp_address: SocketAddr,
-    jds_address: SocketAddr,
 ) -> (JobDeclaratorClient, SocketAddr) {
     use jd_client::config::{
         JobDeclaratorClientConfig, PoolConfig, ProtocolConfig, TPConfig, Upstream,
@@ -127,7 +126,6 @@ pub async fn start_jdc(
         "mkDLTBBRxdBv998612qipDYoTK3YUrqLe8uWw7gu3iXbSrn2n".to_string(),
     )
     .unwrap();
-    let cert_validity_sec = 3600;
     let coinbase_outputs = vec![CoinbaseOutput::new(
         "P2WPKH".to_string(),
         "036adc3bdf21e6f9a0f0fb0066bf517e5b7909ed1563d6958a10993849a7554075".to_string(),
@@ -137,12 +135,17 @@ pub async fn start_jdc(
     )
     .unwrap();
     let pool_signature = "Stratum v2 SRI Pool".to_string();
-    let upstreams = vec![Upstream::new(
-        authority_pubkey,
-        pool_address.to_string(),
-        jds_address.to_string(),
-        pool_signature,
-    )];
+    let upstreams = pool
+        .iter()
+        .map(|(pool_addr, jds_addr)| {
+            Upstream::new(
+                authority_pubkey,
+                pool_addr.to_string(),
+                jds_addr.to_string(),
+                pool_signature.clone(),
+            )
+        })
+        .collect();
     let pool_config = PoolConfig::new(authority_public_key, authority_secret_key);
     let tp_config = TPConfig::new(1000, tp_address.to_string(), None);
     let protocol_config = ProtocolConfig::new(
@@ -158,12 +161,12 @@ pub async fn start_jdc(
         pool_config,
         tp_config,
         upstreams,
-        std::time::Duration::from_secs(cert_validity_sec),
+        std::time::Duration::from_secs(1),
     );
     let ret = jd_client::JobDeclaratorClient::new(jd_client_proxy);
     let ret_clone = ret.clone();
     tokio::spawn(async move { ret_clone.start().await });
-    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+    sleep(2).await;
     (ret, jdc_address)
 }
 
@@ -211,7 +214,7 @@ pub async fn start_jds(tp_rpc_connection: &ConnectParams) -> (JobDeclaratorServe
         tokio::spawn(async move {
             job_declarator_server_clone.start().await.unwrap();
         });
-        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+        sleep(2).await;
         (job_declarator_server, listen_jd_address)
     } else {
         panic!("Failed to get TP cookie values");
@@ -262,7 +265,7 @@ pub async fn start_sv2_translator(upstream: SocketAddr) -> (TranslatorSv2, Socke
     tokio::spawn(async move {
         clone_translator_v2.start().await;
     });
-    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+    sleep(1).await;
     (translator_v2, listening_address)
 }
 
@@ -311,7 +314,7 @@ pub async fn start_mining_device_sv1(
         mining_device_sv1::client::Client::connect(80, upstream_addr, single_submit, custom_target)
             .await;
     });
-    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
+    sleep(3).await;
 }
 
 pub async fn start_mining_device_sv2(
@@ -363,3 +366,8 @@ pub async fn start_mining_sv2_proxy(upstream: SocketAddr) -> SocketAddr {
     });
     mining_proxy_listening_address
 }
+
+#[inline]
+pub async fn sleep(seconds: u64) {
+    tokio::time::sleep(std::time::Duration::from_secs(seconds)).await;
+}

roles/tests-integration/tests/jd_integration.rs

@@ -25,13 +25,13 @@ async fn jds_should_not_panic_if_jdc_shutsdown() {
     let (tp, tp_addr) = start_template_provider(None);
     let (_pool, pool_addr) = start_pool(Some(tp_addr)).await;
     let (_jds, jds_addr) = start_jds(tp.rpc_info()).await;
-    let (jdc, jdc_addr) = start_jdc(pool_addr, tp_addr, jds_addr).await;
+    let (jdc, jdc_addr) = start_jdc(&[(pool_addr, jds_addr)], tp_addr).await;
     jdc.shutdown();
     // wait for shutdown to complete
     tokio::time::sleep(std::time::Duration::from_secs(1)).await;
     assert!(tokio::net::TcpListener::bind(jdc_addr).await.is_ok());
     let (sniffer, sniffer_addr) = start_sniffer("0".to_string(), jds_addr, false, None).await;
-    let (_jdc_1, _jdc_addr_1) = start_jdc(pool_addr, tp_addr, sniffer_addr).await;
+    let (_jdc_1, _jdc_addr_1) = start_jdc(&[(pool_addr, sniffer_addr)], tp_addr).await;
     assert_common_message!(sniffer.next_message_from_downstream(), SetupConnection);
 }
 
@@ -47,7 +47,7 @@ async fn jdc_tp_success_setup() {
     let (_jds, jds_addr) = start_jds(tp.rpc_info()).await;
     let (tp_jdc_sniffer, tp_jdc_sniffer_addr) =
         start_sniffer("0".to_string(), tp_addr, false, None).await;
-    let (_jdc, jdc_addr) = start_jdc(pool_addr, tp_jdc_sniffer_addr, jds_addr).await;
+    let (_jdc, jdc_addr) = start_jdc(&[(pool_addr, jds_addr)], tp_jdc_sniffer_addr).await;
     // 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).await;
@@ -84,7 +84,7 @@ async fn jdc_does_not_stackoverflow_when_no_token() {
         Some(block_from_message.into()),
     )
     .await;
-    let (_jdc, jdc_addr) = start_jdc(pool_addr, tp_addr, jds_jdc_sniffer_addr).await;
+    let (_jdc, jdc_addr) = start_jdc(&[(pool_addr, jds_jdc_sniffer_addr)], tp_addr).await;
     let _ = start_sv2_translator(jdc_addr).await;
     jds_jdc_sniffer
         .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)

roles/tests-integration/tests/jdc_fallback.rs

@@ -0,0 +1,71 @@
+use const_sv2::{MESSAGE_TYPE_SETUP_CONNECTION, MESSAGE_TYPE_SUBMIT_SHARES_SUCCESS};
+use integration_tests_sv2::*;
+use roles_logic_sv2::{
+    mining_sv2::SubmitSharesError,
+    parsers::{AnyMessage, Mining},
+};
+use sniffer::{MessageDirection, ReplaceMessage};
+use std::convert::TryInto;
+
+// Tests whether JDC will switch to a new pool after receiving a `SubmitSharesError` message from
+// the currently connected pool.
+#[tokio::test]
+async fn test_jdc_pool_fallback_after_submit_rejection() {
+    start_tracing();
+    let (tp, tp_addr) = start_template_provider(None);
+    let (_pool_1, pool_addr_1) = start_pool(Some(tp_addr)).await;
+    // Sniffer between JDC and first pool
+    let (sniffer_1, sniffer_addr_1) = start_sniffer(
+        "0".to_string(),
+        pool_addr_1,
+        false,
+        Some(
+            // Should trigger Fallback process in JDC
+            ReplaceMessage::new(
+                MessageDirection::ToDownstream,
+                MESSAGE_TYPE_SUBMIT_SHARES_SUCCESS,
+                AnyMessage::Mining(Mining::SubmitSharesError(SubmitSharesError {
+                    channel_id: 0,
+                    sequence_number: 0,
+                    error_code: "invalid-nonce".to_string().into_bytes().try_into().unwrap(),
+                })),
+            )
+            .into(),
+        ),
+    )
+    .await;
+    let (_pool_2, pool_addr_2) = start_pool(Some(tp_addr)).await;
+    // Sniffer between JDC and second pool
+    let (sniffer_2, sniffer_addr_2) =
+        start_sniffer("1".to_string(), pool_addr_2, false, None).await;
+    let (_jds_1, jds_addr_1) = start_jds(tp.rpc_info()).await;
+    // Sniffer between JDC and first JDS
+    let (sniffer_3, sniffer_addr_3) = start_sniffer("2".to_string(), jds_addr_1, false, None).await;
+    let (_jds_2, jds_addr_2) = start_jds(tp.rpc_info()).await;
+    // Sniffer between JDC and second JDS
+    let (sniffer_4, sniffer_addr_4) = start_sniffer("3".to_string(), jds_addr_2, false, None).await;
+    let (_jdc, jdc_addr) = start_jdc(
+        &[
+            (sniffer_addr_1, sniffer_addr_3),
+            (sniffer_addr_2, sniffer_addr_4),
+        ],
+        tp_addr,
+    )
+    .await;
+    // Assert that JDC has connected to the first (Pool,JDS) pair
+    sniffer_1
+        .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
+        .await;
+    sniffer_3
+        .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
+        .await;
+    let (_translator, sv2_translator_addr) = start_sv2_translator(jdc_addr).await;
+    let _ = start_mining_device_sv1(sv2_translator_addr, true, None).await;
+    // Assert that JDC switched to the second (Pool,JDS) pair
+    sniffer_2
+        .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
+        .await;
+    sniffer_4
+        .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
+        .await;
+}

roles/tests-integration/tests/translator_integration.rs

@@ -65,7 +65,7 @@ async fn translation_proxy_and_jd() {
     let (jdc_pool_sniffer, jdc_pool_sniffer_addr) =
         start_sniffer("0".to_string(), pool_addr, false, None).await;
     let (_jds, jds_addr) = start_jds(tp.rpc_info()).await;
-    let (_jdc, jdc_addr) = start_jdc(jdc_pool_sniffer_addr, tp_addr, jds_addr).await;
+    let (_jdc, jdc_addr) = start_jdc(&[(jdc_pool_sniffer_addr, jds_addr)], tp_addr).await;
     let (_translator, tproxy_addr) = start_sv2_translator(jdc_addr).await;
     let _mining_device = start_mining_device_sv1(tproxy_addr, true, None).await;
     jdc_pool_sniffer

scripts/message-generator-tests.sh

@@ -3,13 +3,8 @@
 search_dir="test/message-generator/test/"
 
 for entry in `ls $search_dir`; do
-    if [ "$entry" = "interop-jdc-change-upstream" ]; then
-        echo "Skipping $entry"
-        continue
-    fi
-
     echo $entry
-    $search_dir$entry/$entry.sh 
+    $search_dir$entry/$entry.sh
 done
 
 cd roles