Merge remote-tracking branch 'origin/main' into gprusak-payload-inter…

…face2

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

@@ -3,11 +3,25 @@ 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,
+    },
+    /// Message signer isn't part of the validator set.
+    #[error("Message signer isn't part of the validator set (signer: {signer:?})")]
+    NonValidatorSigner {
+        /// Signer of the message.
+        signer: validator::PublicKey,
+    },
     /// Unexpected proposal.
     #[error("unexpected proposal")]
     UnexpectedProposal,
@@ -58,6 +72,25 @@ 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,
+            });
+        }
+
+        // Check that the message signer is in the validator set.
+        consensus
+            .validator_set
+            .index(author)
+            .ok_or(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) {
             return Err(Error::Old {

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

@@ -4,11 +4,25 @@ 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 "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,
+    },
+    /// Message signer isn't part of the validator set.
+    #[error("Message signer isn't part of the validator set (signer: {signer:?})")]
+    NonValidatorSigner {
+        /// Signer of the message.
+        signer: validator::PublicKey,
+    },
     /// Past view or phase.
     #[error("past view/phase (current view: {current_view:?}, current phase: {current_phase:?})")]
     Old {
@@ -84,6 +98,25 @@ 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,
+            });
+        }
+
+        // Check that the message signer is in the validator set.
+        consensus
+            .validator_set
+            .index(author)
+            .ok_or(Error::NonValidatorSigner {
+                signer: author.clone(),
+            })?;
+
         // 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

@@ -6,7 +6,7 @@ use assert_matches::assert_matches;
 use pretty_assertions::assert_eq;
 use rand::Rng;
 use zksync_concurrency::ctx;
-use zksync_consensus_roles::validator::{self, LeaderCommit, Phase, ProtocolVersion, ViewNumber};
+use zksync_consensus_roles::validator::{self, LeaderCommit, Phase, ViewNumber};
 
 #[tokio::test]
 async fn replica_prepare_sanity() {
@@ -70,6 +70,45 @@ async fn replica_prepare_sanity_yield_leader_prepare_reproposal() {
     assert_eq!(*map.first_key_value().unwrap().0, replica_prepare);
 }
 
+#[tokio::test]
+async fn replica_prepare_incompatible_protocol_version() {
+    zksync_concurrency::testonly::abort_on_panic();
+    let ctx = &ctx::test_root(&ctx::RealClock);
+    let mut util = UTHarness::new(ctx, 1).await;
+
+    let incompatible_protocol_version = util.incompatible_protocol_version();
+    let replica_prepare = util.new_replica_prepare(|msg| {
+        msg.protocol_version = incompatible_protocol_version;
+    });
+    let res = util.process_replica_prepare(ctx, replica_prepare).await;
+    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_non_validator_signer() {
+    zksync_concurrency::testonly::abort_on_panic();
+    let ctx = &ctx::test_root(&ctx::RealClock);
+    let mut util = UTHarness::new(ctx, 1).await;
+
+    let replica_prepare = util.new_replica_prepare(|_| {}).msg;
+    let non_validator_key: validator::SecretKey = ctx.rng().gen();
+    let res = util
+        .process_replica_prepare(ctx, non_validator_key.sign_msg(replica_prepare))
+        .await;
+    assert_matches!(
+        res,
+        Err(ReplicaPrepareError::NonValidatorSigner { signer }) => {
+            assert_eq!(signer, non_validator_key.public());
+        }
+    );
+}
+
 #[tokio::test]
 async fn replica_prepare_old_view() {
     zksync_concurrency::testonly::abort_on_panic();
@@ -222,28 +261,6 @@ async fn replica_prepare_high_qc_of_future_view() {
     );
 }
 
-#[ignore = "fails/unsupported"]
-#[tokio::test]
-async fn replica_prepare_non_validator_signer() {
-    zksync_concurrency::testonly::abort_on_panic();
-    let ctx = &ctx::test_root(&ctx::RealClock);
-    let mut util = UTHarness::new(ctx, 2).await;
-
-    let replica_prepare = util.new_replica_prepare(|_| {});
-    assert!(util
-        .process_replica_prepare(ctx, replica_prepare.clone())
-        .await
-        .is_err());
-
-    let non_validator: validator::SecretKey = ctx.rng().gen();
-    let replica_prepare = non_validator.sign_msg(replica_prepare.msg);
-    util.process_replica_prepare(ctx, replica_prepare)
-        .await
-        .unwrap();
-    // PANICS:
-    // "Couldn't create justification from valid replica messages!: Message signer isn't in the validator set"
-}
-
 #[tokio::test]
 async fn replica_commit_sanity() {
     zksync_concurrency::testonly::abort_on_panic();
@@ -275,6 +292,44 @@ async fn replica_commit_sanity_yield_leader_commit() {
     );
 }
 
+#[tokio::test]
+async fn replica_commit_incompatible_protocol_version() {
+    zksync_concurrency::testonly::abort_on_panic();
+    let ctx = &ctx::test_root(&ctx::RealClock);
+    let mut util = UTHarness::new(ctx, 1).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.process_replica_commit(ctx, replica_commit).await;
+    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_non_validator_signer() {
+    zksync_concurrency::testonly::abort_on_panic();
+    let ctx = &ctx::test_root(&ctx::RealClock);
+    let mut util = UTHarness::new(ctx, 1).await;
+    let replica_commit = util.new_current_replica_commit(|_| {}).msg;
+    let non_validator_key: validator::SecretKey = ctx.rng().gen();
+    let res = util
+        .process_replica_commit(ctx, non_validator_key.sign_msg(replica_commit))
+        .await;
+    assert_matches!(
+        res,
+        Err(ReplicaCommitError::NonValidatorSigner { signer }) => {
+            assert_eq!(signer, non_validator_key.public());
+        }
+    );
+}
+
 #[tokio::test]
 async fn replica_commit_old() {
     zksync_concurrency::testonly::abort_on_panic();
@@ -385,40 +440,3 @@ async fn replica_commit_unexpected_proposal() {
     let res = util.process_replica_commit(ctx, replica_commit).await;
     assert_matches!(res, Err(ReplicaCommitError::UnexpectedProposal));
 }
-
-#[ignore = "fails/unsupported"]
-#[tokio::test]
-async fn replica_commit_protocol_version_mismatch() {
-    zksync_concurrency::testonly::abort_on_panic();
-    let ctx = &ctx::test_root(&ctx::RealClock);
-    let mut util = UTHarness::new(ctx, 2).await;
-
-    let replica_prepare = util.new_replica_prepare(|_| {});
-    assert!(util
-        .process_replica_prepare(ctx, replica_prepare.clone())
-        .await
-        .is_err());
-    let replica_prepare = util.keys[1].sign_msg(replica_prepare.msg);
-    let leader_prepare = util
-        .process_replica_prepare(ctx, replica_prepare)
-        .await
-        .unwrap()
-        .unwrap();
-    let replica_commit = util
-        .process_leader_prepare(ctx, leader_prepare)
-        .await
-        .unwrap();
-    assert!(util
-        .process_replica_commit(ctx, replica_commit.clone())
-        .await
-        .is_err());
-
-    let mut replica_commit = replica_commit.msg;
-    replica_commit.protocol_version = ProtocolVersion(replica_commit.protocol_version.0 + 1);
-    let replica_commit = util.keys[1].sign_msg(replica_commit);
-    util.process_replica_commit(ctx, replica_commit)
-        .await
-        .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

@@ -112,14 +112,6 @@ impl Consensus {
 
             let res = 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;
-                    }
                     let res = 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:?})"
@@ -135,6 +143,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 {