fix: Sync protocol version between consensus and server blocks

core/lib/types/Cargo.toml

@@ -23,8 +23,8 @@ codegen = { git = "https://github.com/matter-labs/solidity_plonk_verifier.git",
 zkevm_test_harness = { git = "https://github.com/matter-labs/era-zkevm_test_harness.git", branch = "v1.3.3" }
 zk_evm_1_4_0 = { git = "https://github.com/matter-labs/era-zk_evm.git", branch = "v1.4.0", package = "zk_evm" }
 zk_evm = { git = "https://github.com/matter-labs/era-zk_evm.git", tag = "v1.3.3-rc2" }
-zksync_consensus_roles = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
-zksync_protobuf = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
+zksync_consensus_roles = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_protobuf = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
 
 anyhow = "1.0.75"
 chrono = { version = "0.4", features = ["serde"] }
@@ -55,4 +55,4 @@ tokio = { version = "1", features = ["rt", "macros"] }
 serde_with = { version = "1", features = ["hex"] }
 
 [build-dependencies]
-zksync_protobuf_build = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
+zksync_protobuf_build = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }

core/lib/zksync_core/Cargo.toml

@@ -40,11 +40,11 @@ vlog = { path = "../vlog" }
 
 multivm = { path = "../multivm" }
 # Consensus dependenices
-zksync_concurrency = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
-zksync_consensus_roles = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
-zksync_consensus_storage = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
-zksync_consensus_executor = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
-zksync_protobuf = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
+zksync_concurrency = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_consensus_roles = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_consensus_storage = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_consensus_executor = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_protobuf = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
 
 prost = "0.12.1"
 serde = { version = "1.0", features = ["derive"] }
@@ -98,4 +98,4 @@ tempfile = "3.0.2"
 test-casing = "0.1.2"
 
 [build-dependencies]
-zksync_protobuf_build = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed71b2e817c980a2daffef6a01885219e1dc6fa0" }
+zksync_protobuf_build = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }

core/lib/zksync_core/src/sync_layer/gossip/conversions.rs

@@ -12,6 +12,14 @@ use crate::{consensus, sync_layer::fetcher::FetchedBlock};
 pub(super) fn sync_block_to_consensus_block(mut block: SyncBlock) -> anyhow::Result<FinalBlock> {
     let number = BlockNumber(block.number.0.into());
     let consensus = block.consensus.take().context("Missing consensus fields")?;
+    let consensus_protocol_version = consensus.justification.message.protocol_version.as_u32();
+    let block_protocol_version = block.protocol_version as u32;
+    anyhow::ensure!(
+        consensus_protocol_version == block_protocol_version,
+        "Protocol version for justification ({consensus_protocol_version}) differs from \
+         SyncBlock.protocol_version={block_protocol_version}"
+    );
+
     let payload: consensus::Payload = block.try_into().context("Missing `SyncBlock` data")?;
     let payload = payload.encode();
     let header = BlockHeader {
@@ -36,11 +44,17 @@ impl FetchedBlock {
         let payload = consensus::Payload::decode(&block.payload)
             .context("Failed deserializing block payload")?;
 
+        let protocol_version = block.justification.message.protocol_version;
+        let protocol_version =
+            u16::try_from(protocol_version.as_u32()).context("Invalid protocol version")?;
+        let protocol_version = ProtocolVersionId::try_from(protocol_version)
+            .with_context(|| format!("Unsupported protocol version: {protocol_version}"))?;
+
         Ok(Self {
             number: MiniblockNumber(number),
             l1_batch_number: payload.l1_batch_number,
             last_in_batch,
-            protocol_version: ProtocolVersionId::latest(), // FIXME
+            protocol_version,
             timestamp: payload.timestamp,
             l1_gas_price: payload.l1_gas_price,
             l2_fair_gas_price: payload.l2_fair_gas_price,

core/lib/zksync_core/src/sync_layer/gossip/mod.rs

@@ -72,10 +72,11 @@ async fn run_gossip_fetcher_inner(
             .await?;
     let buffered = Arc::new(Buffered::new(store));
     let store = buffered.inner();
-    let executor = Executor::new(executor_config, node_key, buffered.clone())
-        .context("Node executor misconfiguration")?;
 
     scope::run!(ctx, |ctx, s| async {
+        let executor = Executor::new(ctx, executor_config, node_key, buffered.clone())
+            .await
+            .context("Node executor misconfiguration")?;
         s.spawn_bg(async {
             store
                 .run_background_tasks(ctx)

core/lib/zksync_core/src/sync_layer/gossip/tests.rs

@@ -11,7 +11,8 @@ use zksync_consensus_roles::validator::{self, FinalBlock};
 use zksync_consensus_storage::{InMemoryStorage, WriteBlockStore};
 use zksync_dal::{ConnectionPool, StorageProcessor};
 use zksync_types::{
-    api::en::SyncBlock, block::ConsensusBlockFields, Address, L1BatchNumber, MiniblockNumber, H256,
+    api::en::SyncBlock, block::ConsensusBlockFields, Address, L1BatchNumber, MiniblockNumber,
+    ProtocolVersionId, H256,
 };
 
 use super::*;
@@ -63,12 +64,19 @@ pub(super) async fn block_payload(
     consensus::Payload::try_from(sync_block).unwrap()
 }
 
+fn latest_protocol_version() -> validator::ProtocolVersion {
+    (ProtocolVersionId::latest() as u32)
+        .try_into()
+        .expect("latest protocol version is invalid")
+}
+
 /// Adds consensus information for the specified `count` of miniblocks, starting from the genesis.
 pub(super) async fn add_consensus_fields(
     storage: &mut StorageProcessor<'_>,
     validator_key: &validator::SecretKey,
     block_numbers: ops::Range<u32>,
 ) {
+    let protocol_version = latest_protocol_version();
     let mut prev_block_hash = validator::BlockHeaderHash::from_bytes([0; 32]);
     let validator_set = validator::ValidatorSet::new([validator_key.public()]).unwrap();
     for number in block_numbers {
@@ -79,7 +87,7 @@ pub(super) async fn add_consensus_fields(
             payload: payload.hash(),
         };
         let replica_commit = validator::ReplicaCommit {
-            protocol_version: validator::CURRENT_VERSION,
+            protocol_version,
             view: validator::ViewNumber(number.into()),
             proposal: block_header,
         };
@@ -113,7 +121,7 @@ pub(super) fn create_genesis_block(
     };
     let validator_set = validator::ValidatorSet::new([validator_key.public()]).unwrap();
     let replica_commit = validator::ReplicaCommit {
-        protocol_version: validator::CURRENT_VERSION,
+        protocol_version: latest_protocol_version(),
         view: validator::ViewNumber(number),
         proposal: block_header,
     };
@@ -182,10 +190,12 @@ async fn syncing_via_gossip_fetcher(delay_first_block: bool, delay_second_block:
     let tx_hashes = run_state_keeper_with_multiple_miniblocks(pool.clone()).await;
 
     let mut storage = pool.access_storage().await.unwrap();
+    let protocol_version = latest_protocol_version();
     let genesis_block_payload = block_payload(&mut storage, 0).await.encode();
     let ctx = &ctx::test_root(&ctx::AffineClock::new(CLOCK_SPEEDUP as f64));
     let rng = &mut ctx.rng();
-    let mut validator = FullValidatorConfig::for_single_validator(rng, genesis_block_payload);
+    let mut validator =
+        FullValidatorConfig::for_single_validator(rng, protocol_version, genesis_block_payload);
     let validator_set = validator.node_config.validators.clone();
     let external_node = validator.connect_full_node(rng);
 
@@ -207,19 +217,21 @@ async fn syncing_via_gossip_fetcher(delay_first_block: bool, delay_second_block:
                 .unwrap();
         }
     }
-    let validator = Executor::new(
-        validator.node_config,
-        validator.node_key,
-        validator_storage.clone(),
-    )
-    .unwrap();
-    // ^ We intentionally do not run consensus on the validator node, since it'll produce blocks
-    // with payloads that cannot be parsed by the external node.
 
     let (actions_sender, mut actions) = ActionQueue::new();
     let (keeper_actions_sender, keeper_actions) = ActionQueue::new();
     let state_keeper = StateKeeperHandles::new(pool.clone(), keeper_actions, &[&tx_hashes]).await;
     scope::run!(ctx, |ctx, s| async {
+        let validator = Executor::new(
+            ctx,
+            validator.node_config,
+            validator.node_key,
+            validator_storage.clone(),
+        )
+        .await?;
+        // ^ We intentionally do not run consensus on the validator node, since it'll produce blocks
+        // with payloads that cannot be parsed by the external node.
+
         s.spawn_bg(validator.run(ctx));
         s.spawn_bg(run_gossip_fetcher_inner(
             ctx,
@@ -309,10 +321,12 @@ async fn syncing_via_gossip_fetcher_with_multiple_l1_batches(initial_block_count
     let tx_hashes: Vec<_> = tx_hashes.iter().map(Vec::as_slice).collect();
 
     let mut storage = pool.access_storage().await.unwrap();
+    let protocol_version = latest_protocol_version();
     let genesis_block_payload = block_payload(&mut storage, 0).await.encode();
     let ctx = &ctx::test_root(&ctx::AffineClock::new(CLOCK_SPEEDUP as f64));
     let rng = &mut ctx.rng();
-    let mut validator = FullValidatorConfig::for_single_validator(rng, genesis_block_payload);
+    let mut validator =
+        FullValidatorConfig::for_single_validator(rng, protocol_version, genesis_block_payload);
     let validator_set = validator.node_config.validators.clone();
     let external_node = validator.connect_full_node(rng);
 
@@ -327,16 +341,18 @@ async fn syncing_via_gossip_fetcher_with_multiple_l1_batches(initial_block_count
     for block in initial_blocks {
         validator_storage.put_block(ctx, block).await.unwrap();
     }
-    let validator = Executor::new(
-        validator.node_config,
-        validator.node_key,
-        validator_storage.clone(),
-    )
-    .unwrap();
 
     let (actions_sender, actions) = ActionQueue::new();
     let state_keeper = StateKeeperHandles::new(pool.clone(), actions, &tx_hashes).await;
     scope::run!(ctx, |ctx, s| async {
+        let validator = Executor::new(
+            ctx,
+            validator.node_config,
+            validator.node_key,
+            validator_storage.clone(),
+        )
+        .await?;
+
         s.spawn_bg(validator.run(ctx));
         s.spawn_bg(async {
             for block in delayed_blocks {
@@ -388,8 +404,12 @@ async fn syncing_from_non_zero_block(first_block_number: u32) {
         .encode();
     let ctx = &ctx::test_root(&ctx::AffineClock::new(CLOCK_SPEEDUP as f64));
     let rng = &mut ctx.rng();
-    let mut validator =
-        FullValidatorConfig::for_single_validator(rng, genesis_block_payload.clone());
+    let protocol_version = latest_protocol_version();
+    let mut validator = FullValidatorConfig::for_single_validator(
+        rng,
+        protocol_version,
+        genesis_block_payload.clone(),
+    );
     // Override the genesis block since it has an incorrect block number.
     let genesis_block = create_genesis_block(
         &validator.validator_key,
@@ -418,13 +438,6 @@ async fn syncing_from_non_zero_block(first_block_number: u32) {
     tracing::trace!("Re-inserted blocks to node storage");
 
     let validator_storage = Arc::new(InMemoryStorage::new(genesis_block));
-    let validator = Executor::new(
-        validator.node_config,
-        validator.node_key,
-        validator_storage.clone(),
-    )
-    .unwrap();
-
     let tx_hashes = if first_block_number >= 2 {
         &tx_hashes[1..] // Skip transactions in L1 batch #1, since they won't be executed
     } else {
@@ -433,6 +446,14 @@ async fn syncing_from_non_zero_block(first_block_number: u32) {
     let (actions_sender, actions) = ActionQueue::new();
     let state_keeper = StateKeeperHandles::new(pool.clone(), actions, tx_hashes).await;
     scope::run!(ctx, |ctx, s| async {
+        let validator = Executor::new(
+            ctx,
+            validator.node_config,
+            validator.node_key,
+            validator_storage.clone(),
+        )
+        .await?;
+
         s.spawn_bg(validator.run(ctx));
         s.spawn_bg(async {
             for block in &delayed_blocks {