[fix] #4140: Fix registration of new peer

client/benches/tps/utils.rs

@@ -56,7 +56,7 @@ impl Config {
 
     pub fn measure(self) -> Result<Tps> {
         // READY
-        let (_rt, network, client) = <Network>::start_test_with_runtime(self.peers, None);
+        let (_rt, network, client) = Network::start_test_with_runtime(self.peers, None);
         let clients = network.clients();
         wait_for_genesis_committed(&clients, 0);
 

client/tests/integration/asset_propagation.rs

@@ -16,7 +16,7 @@ use test_network::*;
 fn client_add_asset_quantity_to_existing_asset_should_increase_asset_amount_on_another_peer(
 ) -> Result<()> {
     // Given
-    let (_rt, network, client) = <Network>::start_test_with_runtime(4, Some(10_450));
+    let (_rt, network, client) = Network::start_test_with_runtime(4, Some(10_450));
     wait_for_genesis_committed(&network.clients(), 0);
     let pipeline_time = Configuration::pipeline_time();
 

client/tests/integration/config.rs

@@ -1,4 +1,4 @@
-use iroha_data_model::Level;
+use iroha_client::data_model::Level;
 use test_network::*;
 
 #[test]

client/tests/integration/connected_peers.rs

@@ -4,13 +4,15 @@ use eyre::{Context, Result};
 use iroha_client::{
     client::Client,
     data_model::{
-        parameter::{default::MAX_TRANSACTIONS_IN_BLOCK, ParametersBuilder},
+        isi::{Register, Unregister},
         peer::Peer as DataModelPeer,
-        prelude::*,
     },
 };
 use iroha_config::iroha::Configuration;
+use iroha_primitives::unique_vec;
+use rand::{seq::SliceRandom, thread_rng, Rng};
 use test_network::*;
+use tokio::runtime::Runtime;
 
 #[ignore = "ignore, more in #2851"]
 #[test]
@@ -23,11 +25,50 @@ fn connected_peers_with_f_1_0_1() -> Result<()> {
     connected_peers_with_f(1, Some(11_000))
 }
 
+#[test]
+fn register_new_peer() -> Result<()> {
+    let (_rt, network, _) = Network::start_test_with_runtime(4, Some(11_180));
+    wait_for_genesis_committed(&network.clients(), 0);
+    let pipeline_time = Configuration::pipeline_time();
+
+    let mut peer_clients: Vec<_> = Network::peers(&network)
+        .zip(Network::clients(&network))
+        .collect();
+
+    check_status(&peer_clients, 1);
+
+    // Start new peer
+    let mut configuration = Configuration::test();
+    configuration.sumeragi.trusted_peers.peers =
+        unique_vec![peer_clients.choose(&mut thread_rng()).unwrap().0.id.clone()];
+    let rt = Runtime::test();
+    let new_peer = rt.block_on(
+        PeerBuilder::new()
+            .with_configuration(configuration)
+            .with_into_genesis(WithGenesis::None)
+            .with_port(11_200)
+            .start(),
+    );
+
+    let register_peer = Register::peer(DataModelPeer::new(new_peer.id.clone()));
+    peer_clients
+        .choose(&mut thread_rng())
+        .unwrap()
+        .1
+        .submit_blocking(register_peer)?;
+    peer_clients.push((&new_peer, Client::test(&new_peer.api_address)));
+    thread::sleep(pipeline_time * 2); // Wait for some time to allow peers to connect
+
+    check_status(&peer_clients, 2);
+
+    Ok(())
+}
+
 /// Test the number of connected peers, changing the number of faults tolerated down and up
 fn connected_peers_with_f(faults: u64, start_port: Option<u16>) -> Result<()> {
     let n_peers = 3 * faults + 1;
 
-    let (_rt, network, client) = <Network>::start_test_with_runtime(
+    let (_rt, network, _) = Network::start_test_with_runtime(
         (n_peers)
             .try_into()
             .wrap_err("`faults` argument `u64` value too high, cannot convert to `u32`")?,
@@ -36,40 +77,52 @@ fn connected_peers_with_f(faults: u64, start_port: Option<u16>) -> Result<()> {
     wait_for_genesis_committed(&network.clients(), 0);
     let pipeline_time = Configuration::pipeline_time();
 
-    client.submit_all_blocking(
-        ParametersBuilder::new()
-            .add_parameter(MAX_TRANSACTIONS_IN_BLOCK, 1u32)?
-            .into_set_parameters(),
-    )?;
-
-    // Confirm all peers connected
-    let mut status = client.get_status()?;
-    assert_eq!(status.peers, n_peers - 1);
-    assert_eq!(status.blocks, 2);
-
-    // Unregister a peer: committed with f = `faults`
-    // then `status.peers` decrements
-    let peer = network.peers.values().last().unwrap();
-    let peer_client = Client::test(&peer.api_address);
-    let unregister_peer = Unregister::peer(peer.id.clone());
-    client.submit_blocking(unregister_peer)?;
+    let mut peer_clients: Vec<_> = Network::peers(&network)
+        .zip(Network::clients(&network))
+        .collect();
+
+    check_status(&peer_clients, 1);
+
+    // Unregister a peer: committed with f = `faults` then `status.peers` decrements
+    let removed_peer_idx = rand::thread_rng().gen_range(0..peer_clients.len());
+    let (removed_peer, _) = &peer_clients[removed_peer_idx];
+    let unregister_peer = Unregister::peer(removed_peer.id.clone());
+    peer_clients
+        .choose(&mut thread_rng())
+        .unwrap()
+        .1
+        .submit_blocking(unregister_peer)?;
     thread::sleep(pipeline_time * 2); // Wait for some time to allow peers to connect
-    status = client.get_status()?;
-    assert_eq!(status.peers, n_peers - 2);
-    assert_eq!(status.blocks, 3);
-    status = peer_client.get_status()?;
+    let (removed_peer, removed_peer_client) = peer_clients.remove(removed_peer_idx);
+
+    check_status(&peer_clients, 2);
+    let status = removed_peer_client.get_status()?;
+    // Peer might have been disconnected before getting the block
+    assert!(status.blocks == 1 || status.blocks == 2);
     assert_eq!(status.peers, 0);
 
-    // Re-register the peer: committed with f = `faults` - 1 then
-    // `status.peers` increments
-    let register_peer = Register::peer(DataModelPeer::new(peer.id.clone()));
-    client.submit_blocking(register_peer)?;
-    thread::sleep(pipeline_time * 4); // Wait for some time to allow peers to connect
-    status = client.get_status()?;
-    assert_eq!(status.peers, n_peers - 1);
-    assert_eq!(status.blocks, 4);
-    status = peer_client.get_status()?;
-    assert_eq!(status.peers, n_peers - 1);
-    assert_eq!(status.blocks, 4);
+    // Re-register the peer: committed with f = `faults` - 1 then `status.peers` increments
+    let register_peer = Register::peer(DataModelPeer::new(removed_peer.id.clone()));
+    peer_clients
+        .choose(&mut thread_rng())
+        .unwrap()
+        .1
+        .submit_blocking(register_peer)?;
+    peer_clients.insert(removed_peer_idx, (removed_peer, removed_peer_client));
+    thread::sleep(pipeline_time * 2); // Wait for some time to allow peers to connect
+
+    check_status(&peer_clients, 3);
+
     Ok(())
 }
+
+fn check_status(peer_clients: &[(&Peer, Client)], expected_blocks: u64) {
+    let n_peers = peer_clients.len() as u64;
+
+    for (_, peer_client) in peer_clients {
+        let status = peer_client.get_status().unwrap();
+
+        assert_eq!(status.peers, n_peers - 1);
+        assert_eq!(status.blocks, expected_blocks);
+    }
+}

client/tests/integration/events/pipeline.rs

@@ -40,7 +40,7 @@ fn test_with_instruction_and_status_and_port(
     port: u16,
 ) -> Result<()> {
     let (_rt, network, client) =
-        <Network>::start_test_with_runtime(PEER_COUNT.try_into().unwrap(), Some(port));
+        Network::start_test_with_runtime(PEER_COUNT.try_into().unwrap(), Some(port));
     let clients = network.clients();
     wait_for_genesis_committed(&clients, 0);
     let pipeline_time = Configuration::pipeline_time();

client/tests/integration/multiple_blocks_created.rs

@@ -18,7 +18,7 @@ const N_BLOCKS: usize = 510;
 #[test]
 fn long_multiple_blocks_created() -> Result<()> {
     // Given
-    let (_rt, network, client) = <Network>::start_test_with_runtime(4, Some(10_965));
+    let (_rt, network, client) = Network::start_test_with_runtime(4, Some(10_965));
     wait_for_genesis_committed(&network.clients(), 0);
     let pipeline_time = Configuration::pipeline_time();
 

client/tests/integration/multisignature_transaction.rs

@@ -16,7 +16,7 @@ use test_network::*;
 #[allow(clippy::too_many_lines)]
 #[test]
 fn multisignature_transactions_should_wait_for_all_signatures() -> Result<()> {
-    let (_rt, network, client) = <Network>::start_test_with_runtime(4, Some(10_945));
+    let (_rt, network, client) = Network::start_test_with_runtime(4, Some(10_945));
     wait_for_genesis_committed(&network.clients(), 0);
     let pipeline_time = Configuration::pipeline_time();
 

client/tests/integration/offline_peers.rs

@@ -1,11 +1,13 @@
 use eyre::Result;
 use iroha_client::{
-    client::{self, QueryResult},
+    client::{self, Client, QueryResult},
     data_model::{
-        parameter::{default::MAX_TRANSACTIONS_IN_BLOCK, ParametersBuilder},
+        peer::{Peer as DataModelPeer, PeerId},
         prelude::*,
     },
 };
+use iroha_config::iroha::Configuration;
+use iroha_crypto::KeyPair;
 use test_network::*;
 use tokio::runtime::Runtime;
 
@@ -14,19 +16,13 @@ fn genesis_block_is_committed_with_some_offline_peers() -> Result<()> {
     // Given
     let rt = Runtime::test();
 
-    let (network, client) = rt.block_on(<Network>::start_test_with_offline_and_set_n_shifts(
+    let (network, client) = rt.block_on(Network::start_test_with_offline_and_set_n_shifts(
         4,
         1,
         Some(10_560),
     ));
     wait_for_genesis_committed(&network.clients(), 1);
 
-    client.submit_all_blocking(
-        ParametersBuilder::new()
-            .add_parameter(MAX_TRANSACTIONS_IN_BLOCK, 1u32)?
-            .into_set_parameters(),
-    )?;
-
     //When
     let alice_id: AccountId = "alice@wonderland".parse()?;
     let roses = "rose#wonderland".parse()?;
@@ -43,3 +39,41 @@ fn genesis_block_is_committed_with_some_offline_peers() -> Result<()> {
     assert_eq!(AssetValue::Quantity(alice_has_roses), *asset.value());
     Ok(())
 }
+
+#[test]
+fn register_offline_peer() -> Result<()> {
+    let n_peers = 4;
+
+    let (_rt, network, client) = Network::start_test_with_runtime(n_peers, Some(11_160));
+    wait_for_genesis_committed(&network.clients(), 0);
+    let pipeline_time = Configuration::pipeline_time();
+    let peer_clients = Network::clients(&network);
+
+    check_status(&peer_clients, 1);
+
+    let address = "128.0.0.2:8085".parse()?;
+    let key_pair = KeyPair::generate().unwrap();
+    let public_key = key_pair.public_key().clone();
+    let peer_id = PeerId::new(&address, &public_key);
+    let register_peer = Register::peer(DataModelPeer::new(peer_id));
+
+    // Wait for some time to allow peers to connect
+    client.submit_blocking(register_peer)?;
+    std::thread::sleep(pipeline_time * 2);
+
+    // Make sure status hasn't change
+    check_status(&peer_clients, 2);
+
+    Ok(())
+}
+
+fn check_status(peer_clients: &[Client], expected_blocks: u64) {
+    let n_peers = peer_clients.len() as u64;
+
+    for peer_client in peer_clients {
+        let status = peer_client.get_status().unwrap();
+
+        assert_eq!(status.peers, n_peers - 1);
+        assert_eq!(status.blocks, expected_blocks);
+    }
+}

client/tests/integration/restart_peer.rs

@@ -1,71 +1,81 @@
-use std::{str::FromStr, sync::Arc};
+use std::{str::FromStr, thread};
 
 use eyre::Result;
 use iroha_client::{
-    client::{self, QueryResult},
+    client::{self, Client, QueryResult},
     data_model::prelude::*,
 };
 use iroha_config::iroha::Configuration;
-use iroha_primitives::unique_vec;
-use tempfile::TempDir;
+use rand::{seq::SliceRandom, thread_rng, Rng};
 use test_network::*;
 use tokio::runtime::Runtime;
 
 #[test]
 fn restarted_peer_should_have_the_same_asset_amount() -> Result<()> {
-    let temp_dir = Arc::new(TempDir::new()?);
-
-    let mut configuration = Configuration::test();
-    let mut peer = <PeerBuilder>::new().with_port(10_000).build()?;
-    configuration.sumeragi.trusted_peers.peers = unique_vec![peer.id.clone()];
-
     let account_id = AccountId::from_str("alice@wonderland").unwrap();
     let asset_definition_id = AssetDefinitionId::from_str("xor#wonderland").unwrap();
-    let create_asset =
-        Register::asset_definition(AssetDefinition::quantity(asset_definition_id.clone()));
     let quantity: u32 = 200;
 
-    let iroha_client = client::Client::test(&peer.api_address);
+    let mut removed_peer = {
+        let n_peers = 4;
 
-    {
-        let rt = Runtime::test();
-        rt.block_on(
-            PeerBuilder::new()
-                .with_configuration(configuration.clone())
-                .with_dir(temp_dir.clone())
-                .start_with_peer(&mut peer),
-        );
-        wait_for_genesis_committed(&vec![iroha_client.clone()], 0);
+        let (_rt, network, _) = Network::start_test_with_runtime(n_peers, Some(11_200));
+        wait_for_genesis_committed(&network.clients(), 0);
+        let pipeline_time = Configuration::pipeline_time();
+        let peer_clients = Network::clients(&network);
+
+        let create_asset =
+            Register::asset_definition(AssetDefinition::quantity(asset_definition_id.clone()));
+        peer_clients
+            .choose(&mut thread_rng())
+            .unwrap()
+            .submit_blocking(create_asset)?;
 
-        iroha_client.submit_blocking(create_asset)?;
         let mint_asset = Mint::asset_quantity(
             quantity,
             AssetId::new(asset_definition_id.clone(), account_id.clone()),
         );
-        iroha_client.submit_blocking(mint_asset)?;
+        peer_clients
+            .choose(&mut thread_rng())
+            .unwrap()
+            .submit_blocking(mint_asset)?;
+
+        // Wait for observing peer to get the block
+        thread::sleep(pipeline_time);
 
-        let assets = iroha_client
+        let assets = peer_clients
+            .choose(&mut thread_rng())
+            .unwrap()
             .request(client::asset::by_account_id(account_id.clone()))?
             .collect::<QueryResult<Vec<_>>>()?;
         let asset = assets
             .into_iter()
             .find(|asset| asset.id().definition_id == asset_definition_id)
             .expect("Asset not found");
         assert_eq!(AssetValue::Quantity(quantity), *asset.value());
-        peer.stop();
-    }
 
+        let mut all_peers: Vec<_> = core::iter::once(network.genesis)
+            .chain(network.peers.into_values())
+            .collect();
+        let removed_peer_idx = rand::thread_rng().gen_range(0..all_peers.len());
+        let mut removed_peer = all_peers.swap_remove(removed_peer_idx);
+        removed_peer.stop();
+        removed_peer
+    };
+    // All peers have been stopped here
+
+    // Restart just one peer and check if it updates itself from the blockstore
     {
         let rt = Runtime::test();
         rt.block_on(
             PeerBuilder::new()
-                .with_configuration(configuration)
-                .with_dir(temp_dir)
-                .start_with_peer(&mut peer),
+                .with_dir(removed_peer.temp_dir.as_ref().unwrap().clone())
+                .start_with_peer(&mut removed_peer),
         );
-        wait_for_genesis_committed(&vec![iroha_client.clone()], 0);
+        let removed_peer_client = Client::test(&removed_peer.api_address);
+        wait_for_genesis_committed(&vec![removed_peer_client.clone()], 0);
 
-        iroha_client.poll_request(client::asset::by_account_id(account_id), |result| {
+        removed_peer_client.poll_request(client::asset::by_account_id(account_id), |result| {
             let assets = result.collect::<QueryResult<Vec<_>>>().expect("Valid");
             iroha_logger::error!(?assets);