fix(consensus): Debug page server port reuse (#4273)

## What ❔

Introduces a new consensus version that contains a fix that solves the
debug page server being stopped because another debug page server is
already using the same port.
Also adds new config parameters: one that determines how often a node
reads from the ConsensusRegistry contract for validator set updates,
another for the maximum size of transactions propagated in the p2p
network.

## Is this a breaking change?
- [ ] Yes
- [x] No

## Operational changes

The new parameters can be left at their default values.

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [x] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [x] Tests for the changes have been added / updated.
- [x] Documentation comments have been added / updated.
- [x] Code has been formatted via `zkstack dev fmt` and `zkstack dev
lint`.

---------

Co-authored-by: perekopskiy <[email protected]>

Author: brunoffranca

core/Cargo.lock

@@ -268,16 +268,16 @@ bellman = { package = "zksync_bellman", version = "=0.32.1" }
 zksync_vm2 = "=0.4.0"
 
 # Consensus dependencies.
-zksync_concurrency = "=0.11.4"
-zksync_consensus_bft = "=0.11.4"
-zksync_consensus_crypto = "=0.11.4"
-zksync_consensus_engine = "=0.11.4"
-zksync_consensus_executor = "=0.11.4"
-zksync_consensus_network = "=0.11.4"
-zksync_consensus_roles = "=0.11.4"
-zksync_consensus_utils = "=0.11.4"
-zksync_protobuf = "=0.11.4"
-zksync_protobuf_build = "=0.11.4"
+zksync_concurrency = "=0.13"
+zksync_consensus_bft = "=0.13"
+zksync_consensus_crypto = "=0.13"
+zksync_consensus_engine = "=0.13"
+zksync_consensus_executor = "=0.13"
+zksync_consensus_network = "=0.13"
+zksync_consensus_roles = "=0.13"
+zksync_consensus_utils = "=0.13"
+zksync_protobuf = "=0.13"
+zksync_protobuf_build = "=0.13"
 
 # "Local" dependencies
 zksync_multivm = { version = "28.7.0-non-semver-compat", path = "lib/multivm" }

core/Cargo.toml

@@ -146,9 +146,14 @@ pub struct ConsensusConfig {
     /// that will be advertised to peers, so that they can connect to this
     /// node.
     pub public_addr: Host,
+    /// Local socket address to expose the node debug page.
+    pub debug_page_addr: Option<std::net::SocketAddr>,
     /// Maximal allowed size of the payload in bytes.
     #[config(default_t = ByteSize(2_500_000), with = Fallback(SizeUnit::Bytes))]
     pub max_payload_size: ByteSize,
+    /// Maximal allowed size of transactions propagated through the p2p network, in bytes.
+    #[config(default_t = ByteSize(1_000_000), with = Fallback(SizeUnit::Bytes))]
+    pub max_transaction_size: ByteSize,
     /// View timeout duration.
     #[config(default_t = Duration::from_secs(2), with = Fallback(CustomDurationFormat))]
     pub view_timeout: Duration,
@@ -170,19 +175,20 @@ pub struct ConsensusConfig {
     /// establish and maintain.
     #[config(default, with = Entries::WELL_KNOWN.named("key", "addr"))]
     pub gossip_static_outbound: BTreeMap<NodePublicKey, Host>,
+    /// Rate limiting configuration for the p2p RPCs.
+    #[config(nest)]
+    pub rpc: RpcConfig,
+
+    /// Rate at which the node tries to read from the ConsensusRegistry contract, in milliseconds.
+    /// It should be set to a value that is much less than what we expect a validator epoch to be.
+    #[config(default_t = Duration::from_secs(30), with = Fallback(CustomDurationFormat))]
+    pub consensus_registry_read_rate: Duration,
 
     /// MAIN NODE ONLY: consensus genesis specification.
     /// Used to (re)initialize genesis if needed.
     /// External nodes fetch the genesis from the main node.
     #[config(nest)]
     pub genesis_spec: Option<GenesisSpec>,
-
-    /// Rate limiting configuration for the p2p RPCs.
-    #[config(nest)]
-    pub rpc: RpcConfig,
-
-    /// Local socket address to expose the node debug page.
-    pub debug_page_addr: Option<std::net::SocketAddr>,
 }
 
 impl ConsensusConfig {
@@ -200,6 +206,7 @@ impl ConsensusConfig {
             server_addr: "127.0.0.1:0".parse().unwrap(),
             public_addr: Host("127.0.0.1:0".into()),
             max_payload_size: ByteSize(2000000),
+            max_transaction_size: ByteSize(1000000),
             view_timeout: Duration::from_secs(3),
             max_batch_size: ByteSize(125001024),
             gossip_dynamic_inbound_limit: 10,
@@ -224,7 +231,8 @@ impl ConsensusConfig {
             rpc: RpcConfig {
                 get_block_rps: NonZeroUsize::new(5).unwrap(),
             },
-            debug_page_addr: Some("127.0.0.1:0".parse().unwrap()),
+            debug_page_addr: Some("127.0.0.1:5000".parse().unwrap()),
+            consensus_registry_read_rate: Duration::from_secs(30),
         }
     }
 }
@@ -255,6 +263,7 @@ mod tests {
             server_addr: "127.0.0.1:2954".parse().unwrap(),
             public_addr: Host("127.0.0.1:2954".into()),
             max_payload_size: ByteSize(2000000),
+            max_transaction_size: ByteSize(1000000),
             view_timeout: Duration::from_secs(3),
             max_batch_size: ByteSize(125001024),
             gossip_dynamic_inbound_limit: 10,
@@ -280,6 +289,7 @@ mod tests {
                 get_block_rps: NonZeroUsize::new(5).unwrap(),
             },
             debug_page_addr: Some("127.0.0.1:3000".parse().unwrap()),
+            consensus_registry_read_rate: Duration::from_secs(30),
         }
     }
 
@@ -290,6 +300,7 @@ mod tests {
             public_addr: 127.0.0.1:2954
             debug_page_addr: 127.0.0.1:3000
             max_payload_size: 2000000
+            max_transaction_size: 1000000
             view_timeout:
               seconds: 3
               nanos: 0
@@ -314,6 +325,9 @@ mod tests {
             port: 2954
             rpc:
               get_block_rps: 5
+            consensus_registry_read_rate:
+              seconds: 30
+              nanos: 0
         "#;
         let yaml = Yaml::new("test.yml", serde_yaml::from_str(yaml).unwrap()).unwrap();
         let config: ConsensusConfig = test_complete(yaml).unwrap();
@@ -327,6 +341,7 @@ mod tests {
             public_addr: 127.0.0.1:2954
             debug_page_addr: 127.0.0.1:3000
             max_payload_size: 2000000 bytes
+            max_transaction_size: 1000000 bytes
             view_timeout: 3s
             gossip_dynamic_inbound_limit: 10
             gossip_static_inbound:
@@ -346,6 +361,9 @@ mod tests {
             port: 2954
             rpc:
               get_block_rps: 5
+            consensus_registry_read_rate:
+              seconds: 30
+              nanos: 0
         "#;
         let yaml = Yaml::new("test.yml", serde_yaml::from_str(yaml).unwrap()).unwrap();
         let config: ConsensusConfig = test_complete(yaml).unwrap();

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

@@ -148,6 +148,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.0 as usize,
+        max_tx_size: cfg.max_transaction_size.0 as usize,
         view_timeout: cfg.view_timeout.try_into().context("view_timeout")?,
         node_key: node_key(secrets)
             .context("node_key")?

core/node/consensus/src/config.rs

@@ -107,9 +107,18 @@ impl EN {
             .wrap("Store::new()")?;
             s.spawn_bg(async { Ok(runner.run(ctx).await.context("Store::runner()")?) });
 
-            let (engine_manager, engine_runner) = EngineManager::new(ctx, Box::new(store.clone()))
-                .await
-                .wrap("BlockStore::new()")?;
+            let (engine_manager, engine_runner) = EngineManager::new(
+                ctx,
+                Box::new(store.clone()),
+                time::Duration::seconds(
+                    cfg.consensus_registry_read_rate
+                        .as_secs()
+                        .try_into()
+                        .unwrap(),
+                ),
+            )
+            .await
+            .wrap("BlockStore::new()")?;
             s.spawn_bg(async { Ok(engine_runner.run(ctx).await.context("BlockStore::run()")?) });
 
             let executor = executor::Executor {

core/node/consensus/src/en.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use anyhow::Context as _;
-use zksync_concurrency::{ctx, error::Wrap as _, scope};
+use zksync_concurrency::{ctx, error::Wrap as _, scope, time};
 use zksync_config::configs::consensus::{ConsensusConfig, ConsensusSecrets};
 use zksync_consensus_engine::EngineManager;
 use zksync_consensus_executor::{self as executor};
@@ -55,9 +55,18 @@ pub async fn run_main_node(
             .wrap("Store::new()")?;
         s.spawn_bg(async { Ok(runner.run(ctx).await.context("Store::runner()")?) });
 
-        let (engine_manager, engine_runner) = EngineManager::new(ctx, Box::new(store.clone()))
-            .await
-            .wrap("BlockStore::new()")?;
+        let (engine_manager, engine_runner) = EngineManager::new(
+            ctx,
+            Box::new(store.clone()),
+            time::Duration::seconds(
+                cfg.consensus_registry_read_rate
+                    .as_secs()
+                    .try_into()
+                    .unwrap(),
+            ),
+        )
+        .await
+        .wrap("BlockStore::new()")?;
         s.spawn_bg(async { Ok(engine_runner.run(ctx).await.context("BlockStore::run()")?) });
 
         let executor = executor::Executor {

core/node/consensus/src/mn.rs

@@ -327,6 +327,10 @@ impl EngineInterface for Store {
             .await
             .wrap("set_replica_state()")
     }
+
+    async fn push_tx(&self, _ctx: &ctx::Ctx, _tx: engine::Transaction) -> ctx::Result<bool> {
+        unimplemented!()
+    }
 }
 
 /// Background task of the `Store`.

core/node/consensus/src/storage/store.rs

@@ -135,6 +135,7 @@ fn make_config(
         server_addr: *cfg.server_addr,
         public_addr: config::Host(cfg.public_addr.0.clone()),
         max_payload_size: u64::MAX.into(),
+        max_transaction_size: u64::MAX.into(),
         max_batch_size: u64::MAX.into(),
         view_timeout: Duration::from_secs(2),
         gossip_dynamic_inbound_limit: cfg.gossip.dynamic_inbound_limit,
@@ -158,6 +159,7 @@ fn make_config(
         genesis_spec,
         rpc: RpcConfig::default(),
         debug_page_addr: None,
+        consensus_registry_read_rate: Duration::from_secs(1),
     }
 }
 

core/node/consensus/src/testonly.rs

@@ -4,7 +4,12 @@ use anyhow::Context as _;
 use rand::Rng as _;
 use test_casing::{test_casing, Product};
 use tracing::Instrument as _;
-use zksync_concurrency::{ctx, error::Wrap as _, scope, time};
+use zksync_concurrency::{
+    ctx,
+    error::Wrap as _,
+    scope,
+    time::{self, Duration},
+};
 use zksync_config::configs::consensus as config;
 use zksync_consensus_crypto::TextFmt as _;
 use zksync_consensus_engine::{EngineInterface as _, EngineManager};
@@ -170,9 +175,10 @@ async fn test_validator_block_store(version: ProtocolVersionId) {
                 .await
                 .unwrap();
             s.spawn_bg(runner.run(ctx));
-            let (engine_manager, runner) = EngineManager::new(ctx, Box::new(store.clone()))
-                .await
-                .unwrap();
+            let (engine_manager, runner) =
+                EngineManager::new(ctx, Box::new(store.clone()), Duration::seconds(1))
+                    .await
+                    .unwrap();
             s.spawn_bg(runner.run(ctx));
             engine_manager
                 .queue_block(ctx, block.clone())

core/node/consensus/src/tests/mod.rs

@@ -468,7 +468,7 @@ async fn complex_rollback_test() {
     block_hasher.push_tx_hash(txs2[1].hash());
     let block_hash2 = block_hasher.finalize(ProtocolVersionId::latest());
 
-    let blocks = vec![
+    let blocks = [
         (
             L2BlockEnv {
                 number: 2,