Test resharing when rpc is congested

integration-tests/chain-signatures/src/containers.rs

@@ -30,6 +30,8 @@ pub struct Node<'a> {
     pub cipher_sk: hpke::SecretKey,
     pub sign_pk: near_workspaces::types::PublicKey,
     cfg: MultichainConfig,
+    // near rpc address, after proxy
+    near_rpc: String,
 }
 
 impl<'a> Node<'a> {
@@ -48,8 +50,20 @@ impl<'a> Node<'a> {
             near_crypto::SecretKey::from_seed(near_crypto::KeyType::ED25519, "integration-test");
         let sign_pk = sign_sk.public_key();
         let storage_options = ctx.storage_options.clone();
+
         // Use proxied address to mock slow, congested or unstable rpc connection
-        let near_rpc = ctx.lake_indexer.rpc_host_address_proxied.clone();
+        let near_rpc = ctx.lake_indexer.rpc_host_address.clone();
+        let proxy_name = format!("rpc_from_node_{}", account_id);
+        let rpc_port_proxied = utils::pick_unused_port().await?;
+        let rpc_address_proxied = format!("{}:{}", near_rpc, rpc_port_proxied);
+        tracing::info!(
+            "Proxy RPC address {} accessed by node@{} to {}",
+            near_rpc,
+            account_id,
+            rpc_address_proxied
+        );
+        LakeIndexer::populate_proxy(&proxy_name, true, &rpc_address_proxied, &near_rpc).await?;
+
         let mpc_contract_id = ctx.mpc_contract.id().clone();
         let indexer_options = mpc_recovery_node::indexer::Options {
             s3_bucket: ctx.localstack.s3_bucket.clone(),
@@ -112,6 +126,7 @@ impl<'a> Node<'a> {
             cipher_sk,
             sign_pk: sign_pk.to_string().parse()?,
             cfg: cfg.clone(),
+            near_rpc: rpc_address_proxied,
         })
     }
 
@@ -124,6 +139,7 @@ impl<'a> Node<'a> {
             cipher_pk: self.cipher_pk.clone(),
             cipher_sk: self.cipher_sk.clone(),
             cfg: self.cfg.clone(),
+            near_rpc: self.near_rpc.clone(),
         }
     }
 
@@ -134,7 +150,7 @@ impl<'a> Node<'a> {
         let account_id = config.account_id;
         let account_sk = config.account_sk;
         let storage_options = ctx.storage_options.clone();
-        let near_rpc = ctx.lake_indexer.rpc_host_address_proxied.clone();
+        let near_rpc = config.near_rpc;
         let mpc_contract_id = ctx.mpc_contract.id().clone();
         let indexer_options = mpc_recovery_node::indexer::Options {
             s3_bucket: ctx.localstack.s3_bucket.clone(),
@@ -199,6 +215,7 @@ impl<'a> Node<'a> {
             cipher_sk,
             sign_pk: account_sk.public_key(),
             cfg: cfg.clone(),
+            near_rpc,
         })
     }
 }
@@ -291,7 +308,6 @@ pub struct LakeIndexer<'a> {
     pub region: String,
     pub rpc_address: String,
     pub rpc_host_address: String,
-    pub rpc_host_address_proxied: String,
     // Toxi Server is only used in network traffic originated from Lake Indexer
     // to simulate high load and slowness etc. in Lake Indexer
     // Child process is used for proxy host (local node) to container
@@ -343,19 +359,29 @@ impl<'a> LakeIndexer<'a> {
         Ok(container)
     }
 
+    fn remove_protocol(address: &str) -> &str {
+        if let Some(pos) = address.find("://") {
+            &address[pos + 3..]
+        } else {
+            address
+        }
+    }
+
     // Populate a new proxy in toxi proxy server. It proxies all traffic originated from `listen`
     // to `upstream`. The proxy can be configured later (adding latency etc.) given the `name`
     // `listen` and `upstream` must in format `host:port` since toxiproxy operates on tcp level
     // host = true, proxy between a host client request host/container server
     // host = false, proxy between a container client to a container server
     // With current docker setup, container client cannot request host server
-    async fn populate_proxy(
+    pub async fn populate_proxy(
         name: &str,
         host: bool,
         listen: &str,
         upstream: &str,
     ) -> anyhow::Result<()> {
         let toxiproxy_client = reqwest::Client::default();
+        let listen = Self::remove_protocol(listen);
+        let upstream = Self::remove_protocol(upstream);
         let proxies = json!([{
             "name": name,
             "listen": listen,
@@ -389,7 +415,6 @@ impl<'a> LakeIndexer<'a> {
         let toxi_server_process = Self::spin_up_toxi_server_process().await?;
         let toxi_server_container =
             Self::spin_up_toxi_server_container(docker_client, network).await?;
-        let s3_address_without_http = &s3_address[7..];
         let toxi_server_container_address = docker_client
             .get_network_ip_address(&toxi_server_container, network)
             .await?;
@@ -403,13 +428,7 @@ impl<'a> LakeIndexer<'a> {
             s3_address_proxied,
             "Proxy S3 access from Lake Indexer"
         );
-        Self::populate_proxy(
-            "lake-s3",
-            false,
-            &s3_address_proxied,
-            s3_address_without_http,
-        )
-        .await?;
+        Self::populate_proxy("lake-s3", false, &s3_address_proxied, s3_address).await?;
 
         tracing::info!(
             network,
@@ -446,28 +465,12 @@ impl<'a> LakeIndexer<'a> {
         let rpc_address = format!("http://{}:{}", address, Self::CONTAINER_RPC_PORT);
         let rpc_host_port = container.get_host_port_ipv4(Self::CONTAINER_RPC_PORT);
         let rpc_host_address = format!("http://127.0.0.1:{rpc_host_port}");
-        let rpc_port_proxied = utils::pick_unused_port().await?;
-        let rpc_host_address_proxied = format!("http://127.0.0.1:{rpc_port_proxied}");
-
-        tracing::info!(
-            "Proxy Indexer's RPC address from {} to {}",
-            rpc_host_address,
-            rpc_host_address_proxied
-        );
-        Self::populate_proxy(
-            "lake-rpc",
-            true,
-            &format!("127.0.0.1:{}", rpc_port_proxied),
-            &format!("127.0.0.1:{}", rpc_host_port),
-        )
-        .await?;
 
         tracing::info!(
             bucket_name,
             region,
             rpc_address,
             rpc_host_address,
-            rpc_host_address_proxied,
             "NEAR Lake Indexer container is running"
         );
         Ok(LakeIndexer {
@@ -476,7 +479,6 @@ impl<'a> LakeIndexer<'a> {
             region,
             rpc_address,
             rpc_host_address,
-            rpc_host_address_proxied,
             toxi_server_process,
             toxi_server_container,
         })

integration-tests/chain-signatures/src/lib.rs

@@ -233,6 +233,11 @@ impl Nodes<'_> {
         }
         Ok(gcp_services)
     }
+
+    pub fn proxy_name_for_node(&self, id: usize) -> String {
+        let account_id = self.near_accounts();
+        format!("rpc_from_node_{}", account_id[id].id())
+    }
 }
 
 pub struct Context<'a> {

integration-tests/chain-signatures/src/local.rs

@@ -1,5 +1,6 @@
 use crate::{execute, utils, MultichainConfig};
 
+use crate::containers::LakeIndexer;
 use async_process::Child;
 use mpc_keys::hpke;
 use near_workspaces::AccountId;
@@ -16,6 +17,8 @@ pub struct Node {
 
     // process held so it's not dropped. Once dropped, process will be killed.
     process: Child,
+    // near rpc address, after proxy
+    pub near_rpc: String,
 }
 
 pub struct NodeConfig {
@@ -25,6 +28,8 @@ pub struct NodeConfig {
     pub cipher_pk: hpke::PublicKey,
     pub cipher_sk: hpke::SecretKey,
     pub cfg: MultichainConfig,
+    // near rpc address, after proxy
+    pub near_rpc: String,
 }
 
 impl Node {
@@ -46,10 +51,23 @@ impl Node {
             s3_url: Some(ctx.localstack.s3_host_address.clone()),
             start_block_height: 0,
         };
-        let near_rpc = ctx.lake_indexer.rpc_host_address_proxied.clone();
+
+        let near_rpc = ctx.lake_indexer.rpc_host_address.clone();
+        let proxy_name = format!("rpc_from_node_{}", account_id);
+        let rpc_port_proxied = utils::pick_unused_port().await?;
+        let rpc_address_proxied = format!("http://127.0.0.1:{}", rpc_port_proxied);
+        let address = format!("http://127.0.0.1:{web_port}");
+        tracing::info!(
+            "Proxy RPC address {} accessed by node@{} to {}",
+            near_rpc,
+            address,
+            rpc_address_proxied
+        );
+        LakeIndexer::populate_proxy(&proxy_name, true, &rpc_address_proxied, &near_rpc).await?;
+
         let mpc_contract_id = ctx.mpc_contract.id().clone();
         let cli = mpc_recovery_node::cli::Cli::Start {
-            near_rpc: near_rpc.clone(),
+            near_rpc: rpc_address_proxied.clone(),
             mpc_contract_id: mpc_contract_id.clone(),
             account_id: account_id.clone(),
             account_sk: account_sk.to_string().parse()?,
@@ -70,7 +88,6 @@ impl Node {
 
         let mpc_node_id = format!("multichain/{account_id}", account_id = account_id);
         let process = execute::spawn_multichain(ctx.release, &mpc_node_id, cli)?;
-        let address = format!("http://127.0.0.1:{web_port}");
         tracing::info!("node is starting at {}", address);
         utils::ping_until_ok(&address, 60).await?;
         tracing::info!("node started [node_account_id={account_id}, {address}]");
@@ -85,6 +102,7 @@ impl Node {
             cfg: cfg.clone(),
             web_port,
             process,
+            near_rpc: rpc_address_proxied,
         })
     }
 
@@ -104,7 +122,7 @@ impl Node {
         };
         let sign_sk =
             near_crypto::SecretKey::from_seed(near_crypto::KeyType::ED25519, "integration-test");
-        let near_rpc = ctx.lake_indexer.rpc_host_address_proxied.clone();
+        let near_rpc = config.near_rpc;
         let mpc_contract_id = ctx.mpc_contract.id().clone();
         let cli = mpc_recovery_node::cli::Cli::Start {
             near_rpc: near_rpc.clone(),
@@ -143,6 +161,7 @@ impl Node {
             cfg,
             web_port,
             process,
+            near_rpc,
         })
     }
 
@@ -156,6 +175,7 @@ impl Node {
             cipher_pk: self.cipher_pk.clone(),
             cipher_sk: self.cipher_sk.clone(),
             cfg: self.cfg.clone(),
+            near_rpc: self.near_rpc.clone(),
         })
     }
 }

integration-tests/chain-signatures/tests/cases/mod.rs

@@ -322,7 +322,10 @@ async fn test_lake_congestion() -> anyhow::Result<()> {
         Box::pin(async move {
             // Currently, with a 10+-1 latency it cannot generate enough tripplets in time
             // with a 5+-1 latency it fails to wait for signature response
-            add_latency("lake-rpc", true, 1.0, 2_000, 200).await?;
+            add_latency(&ctx.nodes.proxy_name_for_node(0), true, 1.0, 2_000, 200).await?;
+            add_latency(&ctx.nodes.proxy_name_for_node(1), true, 1.0, 2_000, 200).await?;
+            add_latency(&ctx.nodes.proxy_name_for_node(2), true, 1.0, 2_000, 200).await?;
+
             // Also mock lake indexer in high load that it becomes slower to finish process
             // sig req and write to s3
             // with a 1s latency it fails to wait for signature response in time
@@ -338,3 +341,41 @@ async fn test_lake_congestion() -> anyhow::Result<()> {
     })
     .await
 }
+
+#[test(tokio::test)]
+async fn test_multichain_reshare_with_lake_congestion() -> anyhow::Result<()> {
+    let config = MultichainConfig::default();
+    with_multichain_nodes(config.clone(), |mut ctx| {
+        Box::pin(async move {
+            let state = wait_for::running_mpc(&ctx, Some(0)).await?;
+            assert!(state.threshold == 2);
+            assert!(state.participants.len() == 3);
+
+            // add latency to node1->rpc, but not node0->rpc
+            add_latency(&ctx.nodes.proxy_name_for_node(1), true, 1.0, 2_000, 200).await?;
+            // remove node2, node0 and node1 should still reach concensus
+            // this fails if the latency above is too long (10s)
+            assert!(ctx.remove_participant(None).await.is_ok());
+            let state = wait_for::running_mpc(&ctx, Some(0)).await?;
+            assert!(state.participants.len() == 2);
+            // Going below T should error out
+            assert!(ctx.remove_participant(None).await.is_err());
+            let state = wait_for::running_mpc(&ctx, Some(0)).await?;
+            assert!(state.participants.len() == 2);
+            assert!(ctx.add_participant().await.is_ok());
+            // add latency to node2->rpc
+            add_latency(&ctx.nodes.proxy_name_for_node(2), true, 1.0, 2_000, 200).await?;
+            let state = wait_for::running_mpc(&ctx, Some(0)).await?;
+            assert!(state.participants.len() == 3);
+            assert!(ctx.remove_participant(None).await.is_ok());
+            let state = wait_for::running_mpc(&ctx, Some(0)).await?;
+            assert!(state.participants.len() == 2);
+            // make sure signing works after reshare
+            let new_state = wait_for::running_mpc(&ctx, None).await?;
+            wait_for::has_at_least_triples(&ctx, 2).await?;
+            wait_for::has_at_least_presignatures(&ctx, 2).await?;
+            actions::single_signature_production(&ctx, &new_state).await
+        })
+    })
+    .await
+}

integration-tests/chain-signatures/tests/lib.rs

@@ -137,8 +137,7 @@ where
     let sk_local_path = nodes.ctx().storage_options.sk_share_local_path.clone();
 
     let connector = JsonRpcClient::new_client();
-    // Also use proxied rpc to mock unstable when submit transaction
-    let jsonrpc_client = connector.connect(&nodes.ctx().lake_indexer.rpc_host_address_proxied);
+    let jsonrpc_client = connector.connect(&nodes.ctx().lake_indexer.rpc_host_address);
     let rpc_client = near_fetch::Client::from_client(jsonrpc_client.clone());
     let result = f(MultichainTestContext {
         nodes,