[fix] #2133: Rewrite topology to be closer the whitepaper.

Signed-off-by: Sam H. Smith <[email protected]>

Author: SamHSmith

core/Cargo.toml

@@ -75,10 +75,6 @@ hex = "0.4.3"
 byte-unit = "4.0.18"
 once_cell = "1.16.0"
 
-[[bench]]
-name = "sumeragi"
-harness = false
-
 [[bench]]
 name = "validation"
 harness = false

core/benches/sumeragi.rs

@@ -7,6 +7,7 @@ use iroha_core::{
     prelude::*,
     tx::{AcceptedTransaction, TransactionValidator},
     wsv::World,
+    sumeragi::network_topology::Topology,
 };
 use iroha_data_model::prelude::*;
 
@@ -156,7 +157,7 @@ fn chain_blocks(criterion: &mut Criterion) {
             let new_block =
                 block
                     .clone()
-                    .chain(success_count, Some(previous_block_hash.transmute()), 0);
+                    .chain(success_count, Some(previous_block_hash.transmute()), 0, Topology::new(Vec::new()));
             previous_block_hash = new_block.hash();
         });
     });

core/benches/validation.rs

@@ -79,6 +79,7 @@ impl PendingBlock {
         height: u64,
         previous_block_hash: Option<HashOf<VersionedCommittedBlock>>,
         view_change_index: u64,
+        committed_with_topology: Topology,
     ) -> ChainedBlock {
         ChainedBlock {
             transactions: self.transactions,
@@ -91,13 +92,13 @@ impl PendingBlock {
                 previous_block_hash,
                 transactions_hash: None,
                 rejected_transactions_hash: None,
-                genesis_topology: None,
+                committed_with_topology,
             },
         }
     }
 
     /// Create a new blockchain with current block as a first block.
-    pub fn chain_first_with_genesis_topology(self, genesis_topology: Topology) -> ChainedBlock {
+    pub fn chain_first_with_topology(self, genesis_topology: Topology) -> ChainedBlock {
         ChainedBlock {
             transactions: self.transactions,
             event_recommendations: self.event_recommendations,
@@ -109,7 +110,7 @@ impl PendingBlock {
                 previous_block_hash: None,
                 transactions_hash: None,
                 rejected_transactions_hash: None,
-                genesis_topology: Some(genesis_topology),
+                committed_with_topology: genesis_topology,
             },
         }
     }
@@ -127,7 +128,7 @@ impl PendingBlock {
                 previous_block_hash: None,
                 transactions_hash: None,
                 rejected_transactions_hash: None,
-                genesis_topology: None,
+                committed_with_topology: Topology::new(Vec::new()),
             },
         }
     }
@@ -162,8 +163,8 @@ pub struct BlockHeader {
     pub transactions_hash: Option<HashOf<MerkleTree<VersionedSignedTransaction>>>,
     /// Hash of merkle tree root of the tree of rejected transactions' hashes.
     pub rejected_transactions_hash: Option<HashOf<MerkleTree<VersionedSignedTransaction>>>,
-    /// Genesis topology
-    pub genesis_topology: Option<Topology>,
+    /// Network topology when the block was committed.
+    pub committed_with_topology: Topology,
 }
 
 impl BlockHeader {
@@ -388,7 +389,7 @@ impl SignedBlock {
                 previous_block_hash: None,
                 transactions_hash: None,
                 rejected_transactions_hash: None,
-                genesis_topology: None,
+                committed_with_topology: Topology::new(Vec::new()),
             },
             rejected_transactions: Vec::new(),
             transactions: Vec::new(),

core/src/block.rs

@@ -8,7 +8,7 @@
     clippy::arithmetic_side_effects
 )]
 
-use std::{collections::HashSet, fmt::Debug, fs::File, io::BufReader, ops::Deref, path::Path};
+use std::{fmt::Debug, fs::File, io::BufReader, ops::Deref, path::Path};
 
 use derive_more::Deref;
 use eyre::{bail, eyre, Result, WrapErr};
@@ -24,11 +24,7 @@ use iroha_schema::prelude::*;
 use serde::{Deserialize, Serialize};
 use tokio::{time, time::Duration};
 
-use crate::{
-    sumeragi::network_topology::{GenesisBuilder as GenesisTopologyBuilder, Topology},
-    tx::VersionedAcceptedTransaction,
-    IrohaNetwork,
-};
+use crate::{sumeragi::network_topology::Topology, tx::VersionedAcceptedTransaction, IrohaNetwork};
 
 // TODO: 8 is just the optimal value for tests. This number should be
 // revised as soon as we have real data, to fix #1855.
@@ -88,31 +84,14 @@ async fn try_get_online_topology(
     network_topology: &Topology,
     network: Addr<IrohaNetwork>,
 ) -> Result<Topology> {
-    let (online_peers, offline_peers) =
+    let (online_peers, _offline_peers) =
         check_peers_status(this_peer_id, network_topology, network).await;
     let set_a_len = network_topology.min_votes_for_commit();
     if online_peers.len() < set_a_len {
         return Err(eyre!("Not enough online peers for consensus."));
     }
-    let genesis_topology = if network_topology.sorted_peers().len() == 1 {
-        network_topology.clone()
-    } else {
-        let set_a: HashSet<_> = online_peers[..set_a_len].iter().cloned().collect();
-        let set_b: HashSet<_> = online_peers[set_a_len..]
-            .iter()
-            .cloned()
-            .chain(offline_peers.into_iter())
-            .collect();
-        #[allow(clippy::expect_used)]
-        GenesisTopologyBuilder::new()
-            .with_leader(this_peer_id.clone())
-            .with_set_a(set_a)
-            .with_set_b(set_b)
-            .build()
-            .expect("Preconditions should be already checked.")
-    };
     iroha_logger::info!("Waiting for active peers finished.");
-    Ok(genesis_topology)
+    Ok(network_topology.clone())
 }
 
 /// Checks which [`Peer`]s are online and which are offline
@@ -131,7 +110,7 @@ async fn check_peers_status(
     iroha_logger::info!(peer_count = peers.len(), "Peers status");
 
     let (online, offline): (SmallVec<_>, SmallVec<_>) = network_topology
-        .sorted_peers()
+        .sorted_peers
         .iter()
         .cloned()
         .partition(|id| peers.contains(&id.public_key) || this_peer_id.public_key == id.public_key);

core/src/genesis.rs

@@ -261,7 +261,12 @@ mod tests {
 
         for height in 1u64..blocks {
             let block: VersionedCommittedBlock = PendingBlock::new(transactions.clone(), vec![])
-                .chain(height, Some(curr_hash), 0)
+                .chain(
+                    height,
+                    Some(curr_hash),
+                    0,
+                    crate::sumeragi::network_topology::Topology::new(vec![]),
+                )
                 .validate(
                     &TransactionValidator::new(
                         limits,

core/src/smartcontracts/isi/query.rs

@@ -196,7 +196,7 @@ impl<F: FaultInjection> SumeragiWithFault<F> {
 
     #[allow(clippy::needless_pass_by_value)]
     fn broadcast_packet(&self, msg: MessagePacket, topology: &Topology) {
-        self.broadcast_packet_to(msg, topology.sorted_peers());
+        self.broadcast_packet_to(msg, &topology.sorted_peers);
     }
 
     fn gossip_transactions(&self, state: &State, view_change_proof_chain: &ProofChain) {
@@ -222,7 +222,7 @@ impl<F: FaultInjection> SumeragiWithFault<F> {
     /// Connect or disconnect peers according to the current network topology.
     fn connect_peers(&self, topology: &Topology) {
         let peers_expected = {
-            let mut res = topology.sorted_peers().to_owned();
+            let mut res = topology.sorted_peers.clone();
             res.retain(|id| id.address != self.peer_id.address);
             res.shuffle(&mut rand::thread_rng());
             res
@@ -286,7 +286,7 @@ impl<F: FaultInjection> SumeragiWithFault<F> {
             Ok(packet) => {
                 if let Err(error) = view_change_proof_chain.merge(
                     packet.view_change_proofs,
-                    current_topology.sorted_peers(),
+                    &current_topology.sorted_peers,
                     current_topology.max_faults(),
                     state.latest_block_hash,
                 ) {
@@ -368,11 +368,6 @@ impl<F: FaultInjection> SumeragiWithFault<F> {
                     };
 
                     if block.header().is_genesis() {
-                        if let Some(topology) = block.header().genesis_topology.clone().take() {
-                            state.current_topology = topology;
-                            info!("Using genesis topology");
-                        }
-
                         commit_block(self, state, block);
                         return Err(EarlyReturn::GenesisBlockReceivedAndCommitted);
                     }
@@ -418,8 +413,25 @@ fn commit_block<F: FaultInjection>(
         %block_hash, "Committing block"
     );
 
+    *current_topology = committed_block.header().committed_with_topology.clone();
+    current_topology.lift_up_peers(
+        &committed_block
+            .signatures()
+            .into_iter()
+            .map(|s| s.public_key().clone())
+            .collect::<Vec<PublicKey>>(),
+    );
+    current_topology.rotate_a_set();
+    current_topology.update_peer_list(
+        &state
+            .wsv
+            .peers_ids()
+            .iter()
+            .map(|id| id.clone())
+            .collect::<Vec<PeerId>>(),
+    );
+
     sumeragi.kura.store_block(committed_block);
-    current_topology.recreate(&state.wsv, block_hash);
 
     cache_transaction(state, sumeragi);
 }
@@ -458,8 +470,25 @@ fn replace_top_block<F: FaultInjection>(
         %block_hash, "Replacing top block"
     );
 
+    *current_topology = committed_block.header().committed_with_topology.clone();
+    current_topology.lift_up_peers(
+        &committed_block
+            .signatures()
+            .into_iter()
+            .map(|s| s.public_key().clone())
+            .collect::<Vec<PublicKey>>(),
+    );
+    current_topology.rotate_a_set();
+    current_topology.update_peer_list(
+        &state
+            .wsv
+            .peers_ids()
+            .iter()
+            .map(|id| id.clone())
+            .collect::<Vec<PeerId>>(),
+    );
+
     sumeragi.kura.replace_top_block(committed_block);
-    current_topology.recreate(&state.wsv, block_hash);
 
     cache_transaction(state, sumeragi)
 }
@@ -491,7 +520,7 @@ fn suggest_view_change<F: FaultInjection>(
 
     view_change_proof_chain
         .insert_proof(
-            state.current_topology.sorted_peers(),
+            &state.current_topology.sorted_peers,
             state.current_topology.max_faults(),
             state.latest_block_hash,
             suspect_proof,
@@ -511,7 +540,7 @@ fn prune_view_change_proofs_and_calculate_current_index(
 ) -> u64 {
     view_change_proof_chain.prune(state.latest_block_hash);
     view_change_proof_chain.verify_with_state(
-        state.current_topology.sorted_peers(),
+        &state.current_topology.sorted_peers,
         state.current_topology.max_faults(),
         state.latest_block_hash,
     ) as u64
@@ -622,6 +651,9 @@ fn handle_message<F: FaultInjection>(
                 let voting_block_hash = voted_block.block.hash();
 
                 if hash == voting_block_hash.transmute() {
+                    // The manipulation of the topology relies upon all peers seeing the same signature set.
+                    // Therefore we must clear the signatures and accept what the proxy tail giveth.
+                    voted_block.block.signatures.clear();
                     add_signatures::<true>(&mut voted_block, signatures.transmute());
 
                     match voted_block.block.commit(current_topology) {
@@ -743,6 +775,7 @@ fn process_message_independent<F: FaultInjection>(
                             state.latest_block_height,
                             state.latest_block_hash,
                             current_view_change_index,
+                            state.current_topology.clone(),
                         )
                         .validate(&sumeragi.transaction_validator, &state.wsv)
                         .sign(sumeragi.key_pair.clone())
@@ -836,17 +869,19 @@ fn reset_state(
         // a view change a new block is immediately created by the leader
         *round_start_time = Instant::now();
         was_commit_or_view_change = true;
+        *old_view_change_index = 0;
     }
 
-    if current_view_change_index != *old_view_change_index {
-        current_topology.rebuild_with_new_view_change_count(current_view_change_index);
+    while *old_view_change_index < current_view_change_index {
+        *old_view_change_index += 1;
+        error!(addr=%peer_id.address, "Rotating the entire topology.");
+        current_topology.rotate_all();
         was_commit_or_view_change = true;
     }
 
     // Reset state for the next round.
     if was_commit_or_view_change {
         *old_latest_block_height = current_latest_block_height;
-        *old_view_change_index = current_view_change_index;
 
         *voting_block = None;
         voting_signatures.clear();
@@ -970,11 +1005,11 @@ pub(crate) fn run<F: FaultInjection>(
 
             if let Some(VotingBlock { block, .. }) = voting_block.as_ref() {
                 // NOTE: Suspecting the tail node because it hasn't yet committed a block produced by leader
-                warn!(%role, block=%block.hash(), "Block not committed in due time, requesting view change...");
+                warn!(peer_public_key=?sumeragi.peer_id.public_key, %role, block=%block.hash(), "Block not committed in due time, requesting view change...");
             } else {
                 // NOTE: Suspecting the leader node because it hasn't produced a block
                 // If the current node has a transaction, the leader should have as well
-                warn!(%role, "No block produced in due time, requesting view change...");
+                warn!(peer_public_key=?sumeragi.peer_id.public_key, %role, "No block produced in due time, requesting view change...");
             }
 
             suggest_view_change(
@@ -1059,6 +1094,15 @@ fn vote_for_block<F: FaultInjection>(
         return None;
     }
 
+    if block.header().committed_with_topology != state.current_topology {
+        error!(
+            %addr, %role, block_topology=?block.header().committed_with_topology, my_topology=?state.current_topology, hash=%block.hash(),
+            "The block is rejected as because the topology field is incorrect."
+        );
+
+        return None;
+    }
+
     let block = {
         let span = span!(Level::TRACE, "block revalidation");
         let _enter = span.enter();
@@ -1104,7 +1148,7 @@ fn sumeragi_init_commit_genesis<F: FaultInjection>(
         "Genesis transaction set contains no valid transactions"
     );
     let block = PendingBlock::new(transactions, Vec::new())
-        .chain_first_with_genesis_topology(state.current_topology.clone());
+        .chain_first_with_topology(state.current_topology.clone());
 
     {
         info!(block_hash = %block.hash(), "Publishing genesis block.");