Weighted validators (BFT-408, BFT-424)

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

@@ -1,7 +1,7 @@
 //! Handler of a ReplicaCommit message.
+
 use super::StateMachine;
 use crate::metrics;
-use std::collections::HashMap;
 use tracing::instrument;
 use zksync_concurrency::{ctx, metrics::LatencyHistogramExt as _};
 use zksync_consensus_network::io::{ConsensusInputMessage, Target};
@@ -39,10 +39,10 @@ pub(crate) enum Error {
     #[error("invalid message: {0:#}")]
     InvalidMessage(#[source] anyhow::Error),
     /// Duplicate message from a replica.
-    #[error("duplicate message from a replica (existing message: {existing_message:?}")]
-    DuplicateMessage {
-        /// Existing message from the same replica.
-        existing_message: validator::ReplicaCommit,
+    #[error("Replica signed more than one message for same view (message: {message:?}")]
+    DuplicateSignature {
+        /// Offending message.
+        message: validator::ReplicaCommit,
     },
     /// Invalid message signature.
     #[error("invalid signature: {0:#}")]
@@ -70,7 +70,7 @@ impl StateMachine {
             });
         }
 
-        // Check that the message signer is in the validator set.
+        // Check that the message signer is in the validator committee.
         if !self.config.genesis().validators.contains(author) {
             return Err(Error::NonValidatorSigner {
                 signer: author.clone(),
@@ -96,16 +96,23 @@ impl StateMachine {
             return Err(Error::NotLeaderInView);
         }
 
+        // Get current incrementally-constructed QC to work on it
+        let commit_qc = self
+            .commit_qcs
+            .entry(message.view.number)
+            .or_default()
+            .entry(message.clone())
+            .or_insert_with(|| CommitQC::new(message.clone(), self.config.genesis()));
+
         // 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.number)
-            .and_then(|x| x.get(author))
-        {
-            return Err(Error::DuplicateMessage {
-                existing_message: existing_message.msg.clone(),
+        let validator_view = self.validator_views.get(author);
+        if validator_view.is_some_and(|view_number| *view_number >= message.view.number) {
+            return Err(Error::DuplicateSignature {
+                message: commit_qc.message.clone(),
             });
         }
+        self.validator_views
+            .insert(author.clone(), message.view.number);
 
         // ----------- Checking the signed part of the message --------------
 
@@ -118,37 +125,16 @@ impl StateMachine {
 
         // ----------- 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.number)
-            .or_insert_with(|| CommitQC::new(message.clone(), self.config.genesis()))
-            .add(&signed_message, self.config.genesis());
+        // Add the message to the QC.
+        commit_qc.add(&signed_message, self.config.genesis());
 
-        // We store the message in our cache.
-        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 threshold = self.config.genesis().validators.threshold();
-        let Some((_, replica_messages)) =
-            by_proposal.into_iter().find(|(_, v)| v.len() >= threshold)
-        else {
+        // Now we check if we have enough weight to continue.
+        let weight = self.config.genesis().validators.weight(&commit_qc.signers);
+        if weight < self.config.genesis().validators.threshold() {
             return Ok(());
         };
-        debug_assert_eq!(replica_messages.len(), threshold);
 
         // ----------- Update the state machine --------------
-
         let now = ctx.now();
         metrics::METRICS
             .leader_commit_phase_latency
@@ -159,12 +145,13 @@ impl StateMachine {
 
         // ----------- 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.number);
-
         // Consume the incrementally-constructed QC for this view.
-        let justification = self.commit_qcs.remove(&message.view.number).unwrap();
+        let justification = self
+            .commit_qcs
+            .remove(&message.view.number)
+            .unwrap()
+            .remove(message)
+            .unwrap();
 
         // Broadcast the leader commit message to all replicas (ourselves included).
         let output_message = ConsensusInputMessage {
@@ -180,7 +167,6 @@ impl StateMachine {
 
         // Clean the caches.
         self.prepare_message_cache.retain(|k, _| k >= &self.view);
-        self.commit_message_cache.retain(|k, _| k >= &self.view);
 
         Ok(())
     }

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

@@ -2,7 +2,7 @@
 use super::StateMachine;
 use tracing::instrument;
 use zksync_concurrency::{ctx, error::Wrap};
-use zksync_consensus_roles::validator::{self, ProtocolVersion};
+use zksync_consensus_roles::validator;
 
 /// Errors that can occur when processing a "replica prepare" message.
 #[derive(Debug, thiserror::Error)]
@@ -11,9 +11,9 @@ pub(crate) enum Error {
     #[error("incompatible protocol version (message version: {message_version:?}, local version: {local_version:?}")]
     IncompatibleProtocolVersion {
         /// Message version.
-        message_version: ProtocolVersion,
+        message_version: validator::ProtocolVersion,
         /// Local version.
-        local_version: ProtocolVersion,
+        local_version: validator::ProtocolVersion,
     },
     /// Message signer isn't part of the validator set.
     #[error("Message signer isn't part of the validator set (signer: {signer:?})")]
@@ -132,34 +132,29 @@ impl StateMachine {
         // ----------- All checks finished. Now we process the message. --------------
 
         // We add the message to the incrementally-constructed QC.
-        self.prepare_qcs
+        let prepare_qc = self
+            .prepare_qcs
             .entry(message.view.number)
-            .or_insert_with(|| validator::PrepareQC::new(message.view.clone()))
-            .add(&signed_message, self.config.genesis());
+            .or_insert_with(|| validator::PrepareQC::new(message.view.clone()));
+        prepare_qc.add(&signed_message, self.config.genesis());
 
         // We store the message in our cache.
         self.prepare_message_cache
             .entry(message.view.number)
             .or_default()
-            .insert(author.clone(), signed_message);
+            .insert(author.clone(), signed_message.clone());
 
-        // Now we check if we have enough messages to continue.
-        let num_messages = self
-            .prepare_message_cache
-            .get(&message.view.number)
-            .unwrap()
-            .len();
-
-        if num_messages < self.config.genesis().validators.threshold() {
+        // Now we check if we have enough weight to continue.
+        if prepare_qc.weight(&self.config.genesis().validators)
+            < self.config.genesis().validators.threshold()
+        {
             return Ok(());
         }
 
         // Remove replica prepare messages for this view, so that we don't create a new block proposal
         // for this same view if we receive another replica prepare message after this.
         self.prepare_message_cache.remove(&message.view.number);
 
-        debug_assert_eq!(num_messages, self.config.genesis().validators.threshold());
-
         // ----------- Update the state machine --------------
 
         self.view = message.view.number;

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

@@ -36,13 +36,11 @@ pub(crate) struct StateMachine {
     pub(crate) prepare_qcs: BTreeMap<validator::ViewNumber, validator::PrepareQC>,
     /// Newest prepare QC composed from the `ReplicaPrepare` messages.
     pub(crate) prepare_qc: sync::watch::Sender<Option<validator::PrepareQC>>,
-    /// A cache of replica commit messages indexed by view number and validator.
-    pub(crate) commit_message_cache: BTreeMap<
-        validator::ViewNumber,
-        HashMap<validator::PublicKey, Signed<validator::ReplicaCommit>>,
-    >,
-    /// Commit QCs indexed by view number.
-    pub(crate) commit_qcs: BTreeMap<validator::ViewNumber, validator::CommitQC>,
+    /// Commit QCs indexed by view number and then by message.
+    pub(crate) commit_qcs:
+        BTreeMap<validator::ViewNumber, BTreeMap<validator::ReplicaCommit, validator::CommitQC>>,
+    /// Latest view a validator has signed a message for.
+    pub(crate) validator_views: BTreeMap<validator::PublicKey, validator::ViewNumber>,
 }
 
 impl StateMachine {
@@ -67,10 +65,10 @@ impl StateMachine {
             phase_start: ctx.now(),
             prepare_message_cache: BTreeMap::new(),
             prepare_qcs: BTreeMap::new(),
-            commit_message_cache: BTreeMap::new(),
             prepare_qc: sync::watch::channel(None).0,
             commit_qcs: BTreeMap::new(),
             inbound_pipe: recv,
+            validator_views: BTreeMap::new(),
         };
 
         (this, send)

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

@@ -333,6 +333,68 @@ async fn replica_prepare_high_qc_of_future_view() {
     .unwrap();
 }
 
+/// Check all ReplicaPrepare are included for weight calculation
+/// even on different messages for the same view.
+#[tokio::test]
+async fn replica_prepare_different_messages() {
+    zksync_concurrency::testonly::abort_on_panic();
+    let ctx = &ctx::test_root(&ctx::RealClock);
+    scope::run!(ctx, |ctx, s| async {
+        let (mut util, runner) = UTHarness::new_many(ctx).await;
+        s.spawn_bg(runner.run(ctx));
+
+        util.produce_block(ctx).await;
+
+        let view = util.replica_view();
+        let replica_prepare = util.new_replica_prepare();
+
+        // Create a different proposal for the same view
+        let proposal = replica_prepare.clone().high_vote.unwrap().proposal;
+        let mut different_proposal = proposal;
+        different_proposal.number = different_proposal.number.next();
+
+        // Create a new ReplicaPrepare with the different proposal
+        let mut other_replica_prepare = replica_prepare.clone();
+        let mut high_vote = other_replica_prepare.high_vote.clone().unwrap();
+        high_vote.proposal = different_proposal;
+        let high_qc = util.new_commit_qc(|msg| {
+            msg.proposal = different_proposal;
+            msg.view = view.clone()
+        });
+
+        other_replica_prepare.high_vote = Some(high_vote);
+        other_replica_prepare.high_qc = Some(high_qc);
+
+        let validators = util.keys.len();
+
+        // half of the validators sign replica_prepare
+        for i in 0..validators / 2 {
+            util.process_replica_prepare(ctx, util.keys[i].sign_msg(replica_prepare.clone()))
+                .await
+                .unwrap();
+        }
+
+        let mut replica_commit_result = None;
+        // The rest of the validators until threshold sign other_replica_prepare
+        for i in validators / 2..util.genesis().validators.threshold() as usize {
+            replica_commit_result = util
+                .process_replica_prepare(ctx, util.keys[i].sign_msg(other_replica_prepare.clone()))
+                .await
+                .unwrap();
+        }
+
+        // That should be enough for a proposal to be committed (even with different proposals)
+        assert_matches!(replica_commit_result, Some(_));
+
+        // Check the first proposal has been committed (as it has more votes)
+        let message = replica_commit_result.unwrap().msg;
+        assert_eq!(message.proposal, proposal);
+        Ok(())
+    })
+    .await
+    .unwrap();
+}
+
 #[tokio::test]
 async fn replica_commit_sanity() {
     zksync_concurrency::testonly::abort_on_panic();
@@ -484,15 +546,31 @@ async fn replica_commit_already_exists() {
             .await
             .unwrap()
             .is_none());
+
+        // Processing twice same ReplicaCommit for same view gets DuplicateSignature error
         let res = util
             .process_replica_commit(ctx, util.sign(replica_commit.clone()))
             .await;
         assert_matches!(
             res,
-            Err(replica_commit::Error::DuplicateMessage { existing_message }) => {
-                assert_eq!(existing_message, replica_commit)
+            Err(replica_commit::Error::DuplicateSignature { message }) => {
+                assert_eq!(message, replica_commit)
+            }
+        );
+
+        // Processing twice different ReplicaCommit for same view gets DuplicateSignature error too
+        let mut different_replica_commit = replica_commit.clone();
+        different_replica_commit.proposal.number = replica_commit.proposal.number.next();
+        let res = util
+            .process_replica_commit(ctx, util.sign(different_replica_commit.clone()))
+            .await;
+        assert_matches!(
+            res,
+            Err(replica_commit::Error::DuplicateSignature { message }) => {
+                assert_eq!(message, different_replica_commit)
             }
         );
+
         Ok(())
     })
     .await
@@ -572,3 +650,48 @@ async fn replica_commit_unexpected_proposal() {
     .await
     .unwrap();
 }
+
+/// Proposal should be the same for every ReplicaCommit
+/// Check it doesn't fail if one validator sends a different proposal in
+/// the ReplicaCommit
+#[tokio::test]
+async fn replica_commit_different_proposals() {
+    zksync_concurrency::testonly::abort_on_panic();
+    let ctx = &ctx::test_root(&ctx::RealClock);
+    scope::run!(ctx, |ctx, s| async {
+        let (mut util, runner) = UTHarness::new_many(ctx).await;
+        s.spawn_bg(runner.run(ctx));
+
+        let replica_commit = util.new_replica_commit(ctx).await;
+
+        // Process a modified replica_commit (ie. from a malicious or wrong node)
+        let mut bad_replica_commit = replica_commit.clone();
+        bad_replica_commit.proposal.number = replica_commit.proposal.number.next();
+        util.process_replica_commit(ctx, util.sign(bad_replica_commit))
+            .await
+            .unwrap();
+
+        // The rest of the validators sign the correct one
+        let mut replica_commit_result = None;
+        for i in 1..util.keys.len() {
+            replica_commit_result = util
+                .process_replica_commit(ctx, util.keys[i].sign_msg(replica_commit.clone()))
+                .await
+                .unwrap();
+        }
+
+        // Check correct proposal has been committed
+        assert_matches!(
+            replica_commit_result,
+            Some(leader_commit) => {
+                assert_eq!(
+                    leader_commit.msg.justification.message.proposal,
+                    replica_commit.proposal
+                );
+            }
+        );
+        Ok(())
+    })
+    .await
+    .unwrap();
+}