Move protocol version checks (BFT-383) (#44)

Moving protocol version checks from top-level processing to the input
processing of `replica` and `leader`, along with other error-producing
validation checks.

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

@@ -3,11 +3,19 @@ use crate::{inner::ConsensusInner, metrics};
 use tracing::instrument;
 use zksync_concurrency::{ctx, metrics::LatencyHistogramExt as _};
 use zksync_consensus_network::io::{ConsensusInputMessage, Target};
-use zksync_consensus_roles::validator;
+use zksync_consensus_roles::validator::{self, ProtocolVersion};
 
 /// Errors that can occur when processing a "replica commit" message.
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum Error {
+    /// Incompatible protocol version.
+    #[error("incompatible protocol version (message version: {message_version:?}, local version: {local_version:?}")]
+    IncompatibleProtocolVersion {
+        /// Message version.
+        message_version: ProtocolVersion,
+        /// Local version.
+        local_version: ProtocolVersion,
+    },
     /// Unexpected proposal.
     #[error("unexpected proposal")]
     UnexpectedProposal,
@@ -58,6 +66,17 @@ impl StateMachine {
         let message = signed_message.msg;
         let author = &signed_message.key;
 
+        // Check protocol version compatibility.
+        if !consensus
+            .protocol_version
+            .compatible(&message.protocol_version)
+        {
+            return Err(Error::IncompatibleProtocolVersion {
+                message_version: message.protocol_version,
+                local_version: consensus.protocol_version,
+            });
+        }
+
         // If the message is from the "past", we discard it.
         if (message.view, validator::Phase::Commit) < (self.view, self.phase) {
             return Err(Error::Old {

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

@@ -5,11 +5,19 @@ use std::collections::HashMap;
 use tracing::instrument;
 use zksync_concurrency::ctx;
 use zksync_consensus_network::io::{ConsensusInputMessage, Target};
-use zksync_consensus_roles::validator;
+use zksync_consensus_roles::validator::{self, ProtocolVersion};
 
 /// Errors that can occur when processing a "replica prepare" message.
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum Error {
+    /// Incompatible protocol version.
+    #[error("incompatible protocol version (message version: {message_version:?}, local version: {local_version:?}")]
+    IncompatibleProtocolVersion {
+        /// Message version.
+        message_version: ProtocolVersion,
+        /// Local version.
+        local_version: ProtocolVersion,
+    },
     /// Past view or phase.
     #[error("past view/phase (current view: {current_view:?}, current phase: {current_phase:?})")]
     Old {
@@ -70,6 +78,17 @@ impl StateMachine {
         let message = signed_message.msg.clone();
         let author = &signed_message.key;
 
+        // Check protocol version compatibility.
+        if !consensus
+            .protocol_version
+            .compatible(&message.protocol_version)
+        {
+            return Err(Error::IncompatibleProtocolVersion {
+                message_version: message.protocol_version,
+                local_version: consensus.protocol_version,
+            });
+        }
+
         // If the message is from the "past", we discard it.
         if (message.view, validator::Phase::Prepare) < (self.view, self.phase) {
             return Err(Error::Old {

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

@@ -5,8 +5,8 @@ use crate::testonly::ut_harness::UTHarness;
 use assert_matches::assert_matches;
 use rand::Rng;
 use zksync_consensus_roles::validator::{
-    self, CommitQC, ConsensusMsg, LeaderCommit, LeaderPrepare, Phase, PrepareQC, ProtocolVersion,
-    ReplicaCommit, ReplicaPrepare, ViewNumber,
+    self, CommitQC, ConsensusMsg, LeaderCommit, LeaderPrepare, Phase, PrepareQC, ReplicaCommit,
+    ReplicaPrepare, ViewNumber,
 };
 
 #[tokio::test]
@@ -56,6 +56,24 @@ async fn replica_prepare_sanity_yield_leader_prepare() {
     );
 }
 
+#[tokio::test]
+async fn replica_prepare_incompatible_protocol_version() {
+    let mut util = UTHarness::new_one().await;
+
+    let incompatible_protocol_version = util.incompatible_protocol_version();
+    let replica_prepare = util.new_current_replica_prepare(|msg| {
+        msg.protocol_version = incompatible_protocol_version;
+    });
+    let res = util.dispatch_replica_prepare_one(replica_prepare);
+    assert_matches!(
+        res,
+        Err(ReplicaPrepareError::IncompatibleProtocolVersion { message_version, local_version }) => {
+            assert_eq!(message_version, incompatible_protocol_version);
+            assert_eq!(local_version, util.protocol_version());
+        }
+    )
+}
+
 #[tokio::test]
 async fn replica_prepare_sanity_yield_leader_prepare_reproposal() {
     let mut util = UTHarness::new_many().await;
@@ -320,6 +338,24 @@ async fn replica_commit_sanity_yield_leader_commit() {
     );
 }
 
+#[tokio::test]
+async fn replica_commit_incompatible_protocol_version() {
+    let mut util = UTHarness::new_one().await;
+
+    let incompatible_protocol_version = util.incompatible_protocol_version();
+    let replica_commit = util.new_current_replica_commit(|msg| {
+        msg.protocol_version = incompatible_protocol_version;
+    });
+    let res = util.dispatch_replica_commit_one(replica_commit);
+    assert_matches!(
+        res,
+        Err(ReplicaCommitError::IncompatibleProtocolVersion { message_version, local_version }) => {
+            assert_eq!(message_version, incompatible_protocol_version);
+            assert_eq!(local_version, util.protocol_version());
+        }
+    )
+}
+
 #[tokio::test]
 async fn replica_commit_old() {
     let mut util = UTHarness::new_one().await;
@@ -433,38 +469,3 @@ async fn replica_commit_unexpected_proposal() {
     let res = util.dispatch_replica_commit_one(replica_commit);
     assert_matches!(res, Err(ReplicaCommitError::UnexpectedProposal));
 }
-
-#[ignore = "fails/unsupported"]
-#[tokio::test]
-async fn replica_commit_protocol_version_mismatch() {
-    let mut util = UTHarness::new_with(2).await;
-
-    let view = ViewNumber(2);
-    util.set_replica_view(view);
-    util.set_leader_view(view);
-    assert_eq!(util.view_leader(view), util.owner_key().public());
-
-    let replica_prepare_one = util.new_current_replica_prepare(|_| {});
-    let _ = util.dispatch_replica_prepare_one(replica_prepare_one.clone());
-    let replica_prepare_two = util.key_at(1).sign_msg(replica_prepare_one.msg);
-    util.dispatch_replica_prepare_one(replica_prepare_two)
-        .unwrap();
-
-    let leader_prepare = util.recv_signed().await.unwrap();
-    util.dispatch_leader_prepare(leader_prepare).await.unwrap();
-
-    let replica_commit = util.recv_signed().await.unwrap();
-    let _ = util.dispatch_replica_commit_one(replica_commit.clone());
-
-    let mut replica_commit_two = replica_commit.cast::<ReplicaCommit>().unwrap().msg;
-    replica_commit_two.protocol_version =
-        ProtocolVersion(replica_commit_two.protocol_version.0 + 1);
-
-    let replica_commit_two = util
-        .key_at(1)
-        .sign_msg(ConsensusMsg::ReplicaCommit(replica_commit_two));
-    util.dispatch_replica_commit_one(replica_commit_two)
-        .unwrap();
-    // PANICS:
-    // "Couldn't create justification from valid replica messages!: CommitQC can only be created from votes for the same message."
-}

node/actors/bft/src/lib.rs

@@ -101,14 +101,6 @@ impl Consensus {
 
             match input {
                 Some(InputMessage::Network(req)) => {
-                    if req.msg.msg.protocol_version() != self.inner.protocol_version {
-                        tracing::warn!(
-                            "bad protocol version (expected: {:?}, received: {:?})",
-                            self.inner.protocol_version,
-                            req.msg.msg.protocol_version()
-                        );
-                        continue;
-                    }
                     match &req.msg.msg {
                         validator::ConsensusMsg::ReplicaPrepare(_)
                         | validator::ConsensusMsg::ReplicaCommit(_) => {

node/actors/bft/src/replica/leader_commit.rs

@@ -2,11 +2,19 @@ use super::StateMachine;
 use crate::inner::ConsensusInner;
 use tracing::instrument;
 use zksync_concurrency::{ctx, error::Wrap};
-use zksync_consensus_roles::validator;
+use zksync_consensus_roles::validator::{self, ProtocolVersion};
 
 /// Errors that can occur when processing a "leader commit" message.
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum Error {
+    /// Incompatible protocol version.
+    #[error("incompatible protocol version (message version: {message_version:?}, local version: {local_version:?}")]
+    IncompatibleProtocolVersion {
+        /// Message version.
+        message_version: ProtocolVersion,
+        /// Local version.
+        local_version: ProtocolVersion,
+    },
     /// Invalid leader.
     #[error(
         "invalid leader (correct leader: {correct_leader:?}, received leader: {received_leader:?})"
@@ -65,6 +73,17 @@ impl StateMachine {
         let author = &signed_message.key;
         let view = message.justification.message.view;
 
+        // Check protocol version compatibility.
+        if !consensus
+            .protocol_version
+            .compatible(&message.protocol_version)
+        {
+            return Err(Error::IncompatibleProtocolVersion {
+                message_version: message.protocol_version,
+                local_version: consensus.protocol_version,
+            });
+        }
+
         // Check that it comes from the correct leader.
         if author != &consensus.view_leader(view) {
             return Err(Error::InvalidLeader {

node/actors/bft/src/replica/leader_prepare.rs

@@ -4,11 +4,19 @@ use std::collections::HashMap;
 use tracing::instrument;
 use zksync_concurrency::{ctx, error::Wrap};
 use zksync_consensus_network::io::{ConsensusInputMessage, Target};
-use zksync_consensus_roles::validator;
+use zksync_consensus_roles::validator::{self, ProtocolVersion};
 
 /// Errors that can occur when processing a "leader prepare" message.
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum Error {
+    /// Incompatible protocol version.
+    #[error("incompatible protocol version (message version: {message_version:?}, local version: {local_version:?}")]
+    IncompatibleProtocolVersion {
+        /// Message version.
+        message_version: ProtocolVersion,
+        /// Local version.
+        local_version: ProtocolVersion,
+    },
     /// Invalid leader.
     #[error(
         "invalid leader (correct leader: {correct_leader:?}, received leader: {received_leader:?})"
@@ -132,6 +140,17 @@ impl StateMachine {
         let author = &signed_message.key;
         let view = message.view;
 
+        // Check protocol version compatibility.
+        if !consensus
+            .protocol_version
+            .compatible(&message.protocol_version)
+        {
+            return Err(Error::IncompatibleProtocolVersion {
+                message_version: message.protocol_version,
+                local_version: consensus.protocol_version,
+            });
+        }
+
         // Check that it comes from the correct leader.
         if author != &consensus.view_leader(view) {
             return Err(Error::InvalidLeader {

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

@@ -52,6 +52,24 @@ async fn leader_prepare_reproposal_sanity() {
         .unwrap();
 }
 
+#[tokio::test]
+async fn leader_prepare_incompatible_protocol_version() {
+    let mut util = UTHarness::new_one().await;
+
+    let incompatible_protocol_version = util.incompatible_protocol_version();
+    let leader_prepare = util.new_rnd_leader_prepare(|msg| {
+        msg.protocol_version = incompatible_protocol_version;
+    });
+    let res = util.dispatch_leader_prepare(leader_prepare).await;
+    assert_matches!(
+        res,
+        Err(LeaderPrepareError::IncompatibleProtocolVersion { message_version, local_version }) => {
+            assert_eq!(message_version, incompatible_protocol_version);
+            assert_eq!(local_version, util.protocol_version());
+        }
+    )
+}
+
 #[tokio::test]
 async fn leader_prepare_sanity_yield_replica_commit() {
     let mut util = UTHarness::new_one().await;
@@ -485,6 +503,24 @@ async fn leader_commit_sanity_yield_replica_prepare() {
     );
 }
 
+#[tokio::test]
+async fn leader_commit_incompatible_protocol_version() {
+    let mut util = UTHarness::new_one().await;
+
+    let incompatible_protocol_version = util.incompatible_protocol_version();
+    let leader_commit = util.new_rnd_leader_commit(|msg| {
+        msg.protocol_version = incompatible_protocol_version;
+    });
+    let res = util.dispatch_leader_commit(leader_commit).await;
+    assert_matches!(
+        res,
+        Err(LeaderCommitError::IncompatibleProtocolVersion { message_version, local_version }) => {
+            assert_eq!(message_version, incompatible_protocol_version);
+            assert_eq!(local_version, util.protocol_version());
+        }
+    )
+}
+
 #[tokio::test]
 async fn leader_commit_invalid_leader() {
     let mut util = UTHarness::new_with(2).await;