Test JDC Pool Fallback

roles/jd-client/src/lib/upstream_sv2/upstream.rs

@@ -665,6 +665,8 @@
         &mut self,
         _m: roles_logic_sv2::mining_sv2::SubmitSharesError,
     ) -> Result<roles_logic_sv2::handlers::mining::SendTo<Downstream>, RolesLogicError> {
+        dbg!("received submit shares error");
+        dbg!(&_m);
         self.pool_chaneger_trigger
             .safe_lock(|t| t.start(self.tx_status.clone()))
             .unwrap();

roles/tests-integration/lib/mod.rs

@@ -82,7 +82,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)
 }
 
@@ -95,7 +95,7 @@ pub async fn start_template_provider(sv2_interval: Option<u32>) -> (TemplateProv
 }
 
 pub async fn start_jdc(
-    pool_address: SocketAddr,
+    pool_addresses: Vec<SocketAddr>,
     tp_address: SocketAddr,
     jds_address: SocketAddr,
 ) -> (JobDeclaratorClient, SocketAddr) {
@@ -115,7 +115,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(),
@@ -125,12 +124,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_addresses
+        .iter()
+        .map(|addr| {
+            Upstream::new(
+                authority_pubkey,
+                addr.to_string(),
+                jds_address.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(
@@ -146,12 +150,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)
 }
 
@@ -191,7 +195,7 @@ pub async fn start_jds(tp_address: SocketAddr) -> (JobDeclaratorServer, SocketAd
     tokio::spawn(async move {
         job_declarator_server_clone.start().await;
     });
-    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+    sleep(2).await;
     (job_declarator_server, listen_jd_address)
 }
 
@@ -241,7 +245,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)
 }
 
@@ -285,7 +289,7 @@ pub async fn start_mining_device_sv1(upstream_addr: SocketAddr) {
     tokio::spawn(async move {
         mining_device_sv1::client::Client::connect(80, upstream_addr).await;
     });
-    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
+    sleep(3).await;
 }
 
 pub async fn start_mining_sv2_proxy(upstream: SocketAddr) -> SocketAddr {
@@ -315,3 +319,7 @@ pub async fn start_mining_sv2_proxy(upstream: SocketAddr) -> SocketAddr {
     });
     mining_proxy_listening_address
 }
+
+pub async fn sleep(seconds: u64) {
+    tokio::time::sleep(std::time::Duration::from_secs(seconds)).await;
+}

roles/tests-integration/tests/jdc_integration.rs

@@ -0,0 +1,58 @@
+use std::convert::TryInto;
+
+use const_sv2::{
+    MESSAGE_TYPE_SETUP_CONNECTION, MESSAGE_TYPE_SUBMIT_SHARES_ERROR,
+    MESSAGE_TYPE_SUBMIT_SHARES_SUCCESS,
+};
+use integration_tests_sv2::*;
+use sniffer::InterceptMessage;
+
+use crate::sniffer::MessageDirection;
+use roles_logic_sv2::{
+    mining_sv2::SubmitSharesError,
+    parsers::{CommonMessages, Mining, PoolMessages},
+};
+
+// Start JDClient with two pools in the pool list config. The test forces the first pool to send a
+// share submission rejection by altering a message(MINING_SET_NEW_PREV_HASH) sent by the pool. The
+// test verifies that the second pool will be used as a fallback.
+#[tokio::test]
+async fn test_jdc_pool_fallback_after_submit_rejection() {
+    let (_tp, tp_addr) = start_template_provider(None).await;
+    let (_pool, pool_addr) = start_pool(Some(tp_addr)).await;
+    let (sniffer_1, sniffer_addr) = start_sniffer(
+        "0".to_string(),
+        pool_addr,
+        false,
+        Some(vec![InterceptMessage::new(
+            MessageDirection::ToDownstream,
+            MESSAGE_TYPE_SUBMIT_SHARES_SUCCESS,
+            PoolMessages::Mining(Mining::SubmitSharesError(SubmitSharesError {
+                channel_id: 0,
+                sequence_number: 0,
+                error_code: "invalid-nonce".to_string().into_bytes().try_into().unwrap(),
+            })),
+        )]),
+    )
+    .await;
+    let (_pool_2, pool_addr_2) = start_pool(Some(tp_addr)).await;
+    let (sniffer_2, sniffer_addr_2) =
+        start_sniffer("1".to_string(), pool_addr_2, false, None).await;
+    let (_jds, jds_addr) = start_jds(tp_addr).await;
+    let (_jdc, jdc_addr) = start_jdc(vec![sniffer_addr, sniffer_addr_2], tp_addr, jds_addr).await;
+    assert_common_message!(&sniffer_1.next_message_from_downstream(), SetupConnection);
+    let (_translator, sv2_translator_addr) = start_sv2_translator(jdc_addr).await;
+    let _ = start_mining_device_sv1(sv2_translator_addr).await;
+    dbg!("here 1");
+    sniffer_1
+        .wait_for_message_type(
+            MessageDirection::ToDownstream,
+            MESSAGE_TYPE_SUBMIT_SHARES_ERROR,
+        )
+        .await;
+    dbg!("here 2");
+    sniffer_2
+        .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION)
+        .await;
+    dbg!("here 3");
+}

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