feat(consensus): Added view_timeout to consensus config (#3383)

## What ❔

The duration of a view timeout was previously hard-coded in
era-consensus (at 2 seconds). This caused excessive view changes in
networks with low traffic. era-consensus v0.7 now has view timeout
duration as a config parameter and downgraded view change logging from
`error` to `warn`. This PR includes those changes in zksync-era.

Cargo.toml

@@ -246,16 +246,16 @@ zk_evm_1_5_0 = { package = "zk_evm", version = "=0.150.7" }
 zksync_vm2 = { git = "https://github.com/matter-labs/vm2.git", rev = "457d8a7eea9093af9440662e33e598c13ba41633" }
 
 # Consensus dependencies.
-zksync_concurrency = "=0.6.0"
-zksync_consensus_bft = "=0.6.0"
-zksync_consensus_crypto = "=0.6.0"
-zksync_consensus_executor = "=0.6.0"
-zksync_consensus_network = "=0.6.0"
-zksync_consensus_roles = "=0.6.0"
-zksync_consensus_storage = "=0.6.0"
-zksync_consensus_utils = "=0.6.0"
-zksync_protobuf = "=0.6.0"
-zksync_protobuf_build = "=0.6.0"
+zksync_concurrency = "=0.7.0"
+zksync_consensus_bft = "=0.7.0"
+zksync_consensus_crypto = "=0.7.0"
+zksync_consensus_executor = "=0.7.0"
+zksync_consensus_network = "=0.7.0"
+zksync_consensus_roles = "=0.7.0"
+zksync_consensus_storage = "=0.7.0"
+zksync_consensus_utils = "=0.7.0"
+zksync_protobuf = "=0.7.0"
+zksync_protobuf_build = "=0.7.0"
 
 # "Local" dependencies
 zksync_multivm = { version = "0.1.0", path = "core/lib/multivm" }

core/lib/config/src/configs/consensus.rs

@@ -126,6 +126,9 @@ pub struct ConsensusConfig {
     /// Maximal allowed size of the payload in bytes.
     pub max_payload_size: usize,
 
+    /// View timeout duration in milliseconds.
+    pub view_timeout: Option<time::Duration>,
+
     /// Maximal allowed size of the sync-batch payloads in bytes.
     ///
     /// The batch consists of block payloads and a Merkle proof of inclusion on L1 (~1kB),
@@ -155,6 +158,10 @@ pub struct ConsensusConfig {
 }
 
 impl ConsensusConfig {
+    pub fn view_timeout(&self) -> time::Duration {
+        self.view_timeout.unwrap_or(time::Duration::seconds(2))
+    }
+
     pub fn rpc(&self) -> RpcConfig {
         self.rpc.clone().unwrap_or_default()
     }

core/lib/config/src/testonly.rs

@@ -809,6 +809,7 @@ impl Distribution<configs::consensus::ConsensusConfig> for EncodeDist {
             server_addr: self.sample(rng),
             public_addr: Host(self.sample(rng)),
             max_payload_size: self.sample(rng),
+            view_timeout: self.sample(rng),
             max_batch_size: self.sample(rng),
             gossip_dynamic_inbound_limit: self.sample(rng),
             gossip_static_inbound: self

core/lib/protobuf_config/Cargo.toml

@@ -16,6 +16,7 @@ links = "zksync_protobuf_config_proto"
 serde_json.workspace = true
 serde_yaml.workspace = true
 zksync_basic_types.workspace = true
+zksync_concurrency.workspace = true
 zksync_config.workspace = true
 zksync_protobuf.workspace = true
 zksync_types.workspace = true

core/lib/protobuf_config/src/consensus.rs

@@ -1,5 +1,6 @@
 use anyhow::Context as _;
 use zksync_basic_types::L2ChainId;
+use zksync_concurrency::time;
 use zksync_config::configs::consensus::{
     AttesterPublicKey, ConsensusConfig, GenesisSpec, Host, NodePublicKey, ProtocolVersion,
     RpcConfig, ValidatorPublicKey, WeightedAttester, WeightedValidator,
@@ -154,6 +155,11 @@ impl ProtoRepr for proto::Config {
                 .context("server_addr")?,
             public_addr: Host(required(&self.public_addr).context("public_addr")?.clone()),
             max_payload_size,
+            view_timeout: self
+                .view_timeout
+                .as_ref()
+                .map(|x| time::Duration::read(x).context("view_timeout"))
+                .transpose()?,
             max_batch_size,
             gossip_dynamic_inbound_limit: required(&self.gossip_dynamic_inbound_limit)
                 .and_then(|x| Ok((*x).try_into()?))
@@ -187,6 +193,7 @@ impl ProtoRepr for proto::Config {
             server_addr: Some(this.server_addr.to_string()),
             public_addr: Some(this.public_addr.0.clone()),
             max_payload_size: Some(this.max_payload_size.try_into().unwrap()),
+            view_timeout: this.view_timeout.as_ref().map(ProtoFmt::build),
             max_batch_size: Some(this.max_batch_size.try_into().unwrap()),
             gossip_dynamic_inbound_limit: Some(
                 this.gossip_dynamic_inbound_limit.try_into().unwrap(),

core/lib/protobuf_config/src/proto/core/consensus.proto

@@ -87,6 +87,9 @@ message Config {
   // Maximal allowed size of the sync batches.
   optional uint64 max_batch_size = 10; // required; bytes
 
+  // View timeout for the consensus protocol.
+  optional std.Duration view_timeout = 13; // optional
+
   // Inbound connections that should be unconditionally accepted on the gossip network.
   repeated string gossip_static_inbound = 5; // required; NodePublicKey
 

core/node/consensus/src/config.rs

@@ -3,7 +3,7 @@ use std::collections::{BTreeMap, HashMap};
 
 use anyhow::Context as _;
 use secrecy::{ExposeSecret as _, Secret};
-use zksync_concurrency::{limiter, net, time};
+use zksync_concurrency::net;
 use zksync_config::{
     configs,
     configs::consensus::{ConsensusConfig, ConsensusSecrets, Host, NodePublicKey},
@@ -152,11 +152,6 @@ pub(super) fn executor(
 
     let mut rpc = executor::RpcConfig::default();
     rpc.get_block_rate = cfg.rpc().get_block_rate();
-    // Disable batch syncing, because it is not implemented.
-    rpc.get_batch_rate = limiter::Rate {
-        burst: 0,
-        refresh: time::Duration::ZERO,
-    };
 
     let debug_page = cfg.debug_page_addr.map(|addr| network::debug_page::Config {
         addr,
@@ -169,6 +164,7 @@ pub(super) fn executor(
         server_addr: cfg.server_addr,
         public_addr: net::Host(cfg.public_addr.0.clone()),
         max_payload_size: cfg.max_payload_size,
+        view_timeout: cfg.view_timeout(),
         node_key: node_key(secrets)
             .context("node_key")?
             .context("missing node_key")?,

core/node/consensus/src/testonly.rs

@@ -149,6 +149,7 @@ fn make_config(
         public_addr: config::Host(cfg.public_addr.0.clone()),
         max_payload_size: usize::MAX,
         max_batch_size: usize::MAX,
+        view_timeout: None,
         gossip_dynamic_inbound_limit: cfg.gossip.dynamic_inbound_limit,
         gossip_static_inbound: cfg
             .gossip