support for hard forks (BFT-420) (#65)

Added support for making hard forks in the chain.
It works as follows:
1. Genesis struct is introduced which defines a fork. It includes:
* the validator set
* number of the first block of the fork and a parent hash (commitment to
the chain built before the fork)
* some information about the past forks (it will be removed once block
syncing is upgraded)
2. BlockStore is equipped with genesis and it verifies that the inserted
blocks are consistent with the genesis
3. nodes with different Genesis are unable to connect to each other
(genesis hash is verified during a handshake)
4. BFT algorithm is restarted at each fork - views start from 0 again,
all the quorum certificates from past forks get invalidated

To perform a fork, use a `genesis.fork_set.push` method on the previous
genesis and remove the blocks which do not belong to the new fork from
the BlockStore.

Additionally some refactoring has been performed:
* consensusnet state now explicitly depends on gossipnet state
* consensus message verification has been moved from bft crate to roles
crate, so that quorum certificates can be verfied consistently from
bft,sync_blocks and storage crates.
* threshold computation has been moved to ValidatorSet and is considered
a genesis property now (until we move to dynamic validator set)
* tests have been adjusted to work with non-trivial forks

node/Cargo.toml

@@ -155,3 +155,5 @@ needless_pass_by_ref_mut = "allow"
 box_default = "allow"
 # remove once fix to https://github.com/rust-lang/rust-clippy/issues/11764 is available on CI.
 map_identity = "allow"
+# &*x is not equivalent to x, because it affects borrowing in closures.
+borrow_deref_ref = "allow"

node/actors/bft/src/config.rs

@@ -1,7 +1,6 @@
 //! The inner data of the consensus state machine. This is shared between the different roles.
-use crate::{misc, PayloadManager};
+use crate::PayloadManager;
 use std::sync::Arc;
-use tracing::instrument;
 use zksync_consensus_roles::validator;
 use zksync_consensus_storage as storage;
 
@@ -10,8 +9,6 @@ use zksync_consensus_storage as storage;
 pub struct Config {
     /// The validator's secret key.
     pub secret_key: validator::SecretKey,
-    /// A vector of public keys for all the validators in the network.
-    pub validator_set: validator::ValidatorSet,
     /// The maximum size of the payload of a block, in bytes. We will
     /// reject blocks with payloads larger than this.
     pub max_payload_size: usize,
@@ -24,23 +21,8 @@ pub struct Config {
 }
 
 impl Config {
-    /// Computes the validator for the given view.
-    #[instrument(level = "trace", ret)]
-    pub fn view_leader(&self, view_number: validator::ViewNumber) -> validator::PublicKey {
-        let index = view_number.0 as usize % self.validator_set.len();
-        self.validator_set.get(index).unwrap().clone()
-    }
-
-    /// Calculate the consensus threshold, the minimum number of votes for any consensus action to be valid,
-    /// for a given number of replicas.
-    #[instrument(level = "trace", ret)]
-    pub fn threshold(&self) -> usize {
-        misc::consensus_threshold(self.validator_set.len())
-    }
-
-    /// Calculate the maximum number of faulty replicas, for a given number of replicas.
-    #[instrument(level = "trace", ret)]
-    pub fn faulty_replicas(&self) -> usize {
-        misc::faulty_replicas(self.validator_set.len())
+    /// Genesis.
+    pub fn genesis(&self) -> &validator::Genesis {
+        self.block_store.genesis()
     }
 }

node/actors/bft/src/leader/mod.rs

@@ -2,14 +2,10 @@
 //! and aggregates replica messages. It mainly acts as a central point of communication for the replicas. Note that
 //! our consensus node will perform both the replica and leader roles simultaneously.
 
-mod replica_commit;
-mod replica_prepare;
+pub(crate) mod replica_commit;
+pub(crate) mod replica_prepare;
 mod state_machine;
 #[cfg(test)]
 mod tests;
 
-#[cfg(test)]
-pub(crate) use self::replica_commit::Error as ReplicaCommitError;
-#[cfg(test)]
-pub(crate) use self::replica_prepare::Error as ReplicaPrepareError;
 pub(crate) use self::state_machine::StateMachine;

node/actors/bft/src/leader/replica_commit.rs

@@ -1,3 +1,4 @@
+//! Handler of a ReplicaCommit message.
 use super::StateMachine;
 use crate::metrics;
 use std::collections::HashMap;
@@ -34,6 +35,9 @@ pub(crate) enum Error {
     /// The processing node is not a lead for this message's view.
     #[error("we are not a leader for this message's view")]
     NotLeaderInView,
+    /// Invalid message.
+    #[error("invalid message: {0:#}")]
+    InvalidMessage(anyhow::Error),
     /// Duplicate message from a replica.
     #[error("duplicate message from a replica (existing message: {existing_message:?}")]
     DuplicateMessage {
@@ -55,47 +59,51 @@ impl StateMachine {
         // ----------- Checking origin of the message --------------
 
         // Unwrap message.
-        let message = signed_message.msg;
+        let message = &signed_message.msg;
         let author = &signed_message.key;
 
         // Check protocol version compatibility.
-        if !crate::PROTOCOL_VERSION.compatible(&message.protocol_version) {
+        if !crate::PROTOCOL_VERSION.compatible(&message.view.protocol_version) {
             return Err(Error::IncompatibleProtocolVersion {
-                message_version: message.protocol_version,
+                message_version: message.view.protocol_version,
                 local_version: crate::PROTOCOL_VERSION,
             });
         }
 
         // Check that the message signer is in the validator set.
-        let validator_index =
-            self.config
-                .validator_set
-                .index(author)
-                .ok_or(Error::NonValidatorSigner {
-                    signer: author.clone(),
-                })?;
+        if !self.config.genesis().validators.contains(author) {
+            return Err(Error::NonValidatorSigner {
+                signer: author.clone(),
+            });
+        }
 
         // If the message is from the "past", we discard it.
-        if (message.view, validator::Phase::Commit) < (self.view, self.phase) {
+        if (message.view.number, validator::Phase::Commit) < (self.view, self.phase) {
             return Err(Error::Old {
                 current_view: self.view,
                 current_phase: self.phase,
             });
         }
 
         // If the message is for a view when we are not a leader, we discard it.
-        if self.config.view_leader(message.view) != self.config.secret_key.public() {
+        if self
+            .config
+            .genesis()
+            .validators
+            .view_leader(message.view.number)
+            != self.config.secret_key.public()
+        {
             return Err(Error::NotLeaderInView);
         }
 
         // If we already have a message from the same validator and for the same view, we discard it.
         if let Some(existing_message) = self
             .commit_message_cache
-            .get(&message.view)
+            .get(&message.view.number)
             .and_then(|x| x.get(author))
         {
             return Err(Error::DuplicateMessage {
-                existing_message: existing_message.msg,
+                existing_message: existing_message.msg.clone(),
             });
         }
 
@@ -104,60 +112,67 @@ impl StateMachine {
         // Check the signature on the message.
         signed_message.verify().map_err(Error::InvalidSignature)?;
 
+        message
+            .verify(self.config.genesis())
+            .map_err(Error::InvalidMessage)?;
+
         // ----------- All checks finished. Now we process the message. --------------
 
+        // TODO: we have a bug here since we don't check whether replicas commit
+        // to the same proposal.
+
         // We add the message to the incrementally-constructed QC.
         self.commit_qcs
-            .entry(message.view)
-            .or_insert(CommitQC::new(message, &self.config.validator_set))
-            .add(&signed_message.sig, validator_index);
+            .entry(message.view.number)
+            .or_insert_with(|| CommitQC::new(message.clone(), self.config.genesis()))
+            .add(&signed_message, self.config.genesis());
 
         // We store the message in our cache.
-        let cache_entry = self.commit_message_cache.entry(message.view).or_default();
-        cache_entry.insert(author.clone(), signed_message);
+        let cache_entry = self
+            .commit_message_cache
+            .entry(message.view.number)
+            .or_default();
+        cache_entry.insert(author.clone(), signed_message.clone());
 
         // Now we check if we have enough messages to continue.
         let mut by_proposal: HashMap<_, Vec<_>> = HashMap::new();
         for msg in cache_entry.values() {
             by_proposal.entry(msg.msg.proposal).or_default().push(msg);
         }
-        let Some((_, replica_messages)) = by_proposal
-            .into_iter()
-            .find(|(_, v)| v.len() >= self.config.threshold())
+        let threshold = self.config.genesis().validators.threshold();
+        let Some((_, replica_messages)) =
+            by_proposal.into_iter().find(|(_, v)| v.len() >= threshold)
         else {
             return Ok(());
         };
-        debug_assert_eq!(replica_messages.len(), self.config.threshold());
+        debug_assert_eq!(replica_messages.len(), threshold);
 
         // ----------- Update the state machine --------------
 
         let now = ctx.now();
         metrics::METRICS
             .leader_commit_phase_latency
             .observe_latency(now - self.phase_start);
-        self.view = message.view.next();
+        self.view = message.view.number.next();
         self.phase = validator::Phase::Prepare;
         self.phase_start = now;
 
         // ----------- Prepare our message and send it. --------------
 
         // Remove replica commit messages for this view, so that we don't create a new leader commit
         // for this same view if we receive another replica commit message after this.
-        self.commit_message_cache.remove(&message.view);
+        self.commit_message_cache.remove(&message.view.number);
 
         // Consume the incrementally-constructed QC for this view.
-        let justification = self.commit_qcs.remove(&message.view).unwrap();
+        let justification = self.commit_qcs.remove(&message.view.number).unwrap();
 
         // Broadcast the leader commit message to all replicas (ourselves included).
         let output_message = ConsensusInputMessage {
             message: self
                 .config
                 .secret_key
                 .sign_msg(validator::ConsensusMsg::LeaderCommit(
-                    validator::LeaderCommit {
-                        protocol_version: crate::PROTOCOL_VERSION,
-                        justification,
-                    },
+                    validator::LeaderCommit { justification },
                 )),
             recipient: Target::Broadcast,
         };