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

SamHSmith · SamHSmith · commit 08f16b0aaa01 · 2023-02-22T23:49:46.000+01:00
Signed-off-by: Sam H. Smith &lt;sam.henning.smith@protonmail.com&gt;
diff --git a/core/src/block.rs b/core/src/block.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(),
diff --git a/core/src/genesis.rs b/core/src/genesis.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};
@@ -25,7 +25,7 @@ use serde::{Deserialize, Serialize};
 use tokio::{time, time::Duration};
 
 use crate::{
-    sumeragi::network_topology::{GenesisBuilder as GenesisTopologyBuilder, Topology},
+    sumeragi::network_topology::Topology,
     tx::VersionedAcceptedTransaction,
     IrohaNetwork,
 };
@@ -88,31 +88,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 +114,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);
diff --git a/core/src/smartcontracts/isi/query.rs b/core/src/smartcontracts/isi/query.rs
@@ -261,7 +261,7 @@ 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,
diff --git a/core/src/sumeragi/main_loop.rs b/core/src/sumeragi/main_loop.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,12 @@ 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(|id| id.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 +457,12 @@ 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(|id| id.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 +494,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 +514,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 +625,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 +749,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 +843,21 @@ 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;
+
+        println!("{:?}", current_topology);
     }
 
     // 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 +981,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 +1070,16 @@ 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 +1125,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.");
diff --git a/core/src/sumeragi/mod.rs b/core/src/sumeragi/mod.rs
@@ -280,22 +280,15 @@ impl Sumeragi {
         let latest_block_hash = wsv.latest_block_hash();
         let previous_block_hash = wsv.previous_block_hash();
 
-        let current_topology = latest_block_hash.map_or_else(
-            || {
-                assert!(!sumeragi.config.trusted_peers.peers.is_empty());
-                Topology::builder()
-                    .with_peers(sumeragi.config.trusted_peers.peers.clone())
-                    .build(0)
-                    .expect("This builder must have been valid. This is a programmer error.")
-            },
-            |block_hash| {
-                Topology::builder()
-                    .with_peers(wsv.peers().iter().map(|peer| peer.id().clone()))
-                    .at_block(block_hash)
-                    .build(0)
-                    .expect("Should be able to reconstruct topology from `wsv`")
-            },
-        );
+        let current_topology = if latest_block_height == 0 {
+            assert!(!sumeragi.config.trusted_peers.peers.is_empty());
+            Topology::new(sumeragi.config.trusted_peers.peers.clone())
+        } else {
+            let block_ref = sumeragi.internal.kura.get_block_by_height(latest_block_height).expect("Sumeragi could not load block that was reported as present. Please check that the block storage was not disconnected.");
+            let mut topology = block_ref.header().committed_with_topology.clone();
+            topology.rotate_a_set();
+            topology
+        };
 
         let sumeragi_state_machine_data = State {
             previous_block_hash,
diff --git a/core/src/sumeragi/network_topology.rs b/core/src/sumeragi/network_topology.rs
diff --git a/crypto/src/signature.rs b/crypto/src/signature.rs
