feat(state-keeper): miniblock max payload size (BFT-417)

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

@@ -100,6 +100,8 @@ pub struct StateKeeperConfig {
     /// sealing will block until some of the miniblocks from the queue are processed.
     /// 0 means that sealing is synchronous; this is mostly useful for performance comparison, testing etc.
     pub miniblock_seal_queue_capacity: usize,
+    /// The max payload size threshold (in bytes) that triggers sealing of a miniblock.
+    pub miniblock_max_payload_size: usize,
 
     /// The max number of gas to spend on an L1 tx before its batch should be sealed by the gas sealer.
     pub max_single_tx_gas: u32,
@@ -178,6 +180,7 @@ impl StateKeeperConfig {
             block_commit_deadline_ms: 2500,
             miniblock_commit_deadline_ms: 1000,
             miniblock_seal_queue_capacity: 10,
+            miniblock_max_payload_size: 1_000_000,
             max_single_tx_gas: 6000000,
             max_allowed_l2_tx_gas_limit: 4000000000,
             reject_tx_at_geometry_percentage: 0.95,

core/lib/config/src/testonly.rs

@@ -152,6 +152,7 @@ impl Distribution<configs::chain::StateKeeperConfig> for EncodeDist {
             block_commit_deadline_ms: self.sample(rng),
             miniblock_commit_deadline_ms: self.sample(rng),
             miniblock_seal_queue_capacity: self.sample(rng),
+            miniblock_max_payload_size: self.sample(rng),
             max_single_tx_gas: self.sample(rng),
             max_allowed_l2_tx_gas_limit: self.sample(rng),
             reject_tx_at_geometry_percentage: self.sample(rng),

core/lib/dal/build.rs

@@ -1,7 +1,7 @@
 //! Generates rust code from protobufs.
 fn main() {
     zksync_protobuf_build::Config {
-        input_root: "src/models/proto".into(),
+        input_root: "src/consensus/proto".into(),
         proto_root: "zksync/dal".into(),
         dependencies: vec![],
         protobuf_crate: "::zksync_protobuf".parse().unwrap(),

core/lib/dal/src/models/consensus/mod.rs → core/lib/dal/src/consensus/mod.rs

@@ -1,3 +1,5 @@
+pub mod proto;
+
 #[cfg(test)]
 mod tests;
 
@@ -15,7 +17,7 @@ use zksync_types::{
 };
 use zksync_utils::{h256_to_u256, u256_to_h256};
 
-use crate::models::{parse_h160, parse_h256, proto};
+use crate::models::{parse_h160, parse_h256};
 
 /// L2 block (= miniblock) payload.
 #[derive(Debug, PartialEq)]
@@ -34,7 +36,7 @@ pub struct Payload {
 }
 
 impl ProtoFmt for Payload {
-    type Proto = super::proto::Payload;
+    type Proto = proto::Payload;
 
     fn read(message: &Self::Proto) -> anyhow::Result<Self> {
         let mut transactions = Vec::with_capacity(message.transactions.len());

core/lib/dal/src/consensus/proto/mod.rs

@@ -0,0 +1,2 @@
+#![allow(warnings)]
+include!(concat!(env!("OUT_DIR"), "/src/consensus/proto/gen.rs"));

core/lib/dal/src/consensus_dal.rs

@@ -10,7 +10,7 @@ use zksync_db_connection::{
 };
 use zksync_types::MiniblockNumber;
 
-pub use crate::models::consensus::Payload;
+pub use crate::consensus::Payload;
 use crate::{Core, CoreDal};
 
 /// Storage access methods for `zksync_core::consensus` module.

core/lib/dal/src/lib.rs

@@ -29,6 +29,7 @@ use crate::{
 pub mod basic_witness_input_producer_dal;
 pub mod blocks_dal;
 pub mod blocks_web3_dal;
+pub mod consensus;
 pub mod consensus_dal;
 pub mod contract_verification_dal;
 pub mod eth_sender_dal;

core/lib/dal/src/models/mod.rs

@@ -1,10 +1,8 @@
+pub mod storage_block;
 use anyhow::Context as _;
 use zksync_db_connection::error::SqlxContext;
 use zksync_types::{ProtocolVersionId, H160, H256};
 
-pub mod consensus;
-mod proto;
-pub mod storage_block;
 pub mod storage_eth_tx;
 pub mod storage_event;
 pub mod storage_fee_monitor;
@@ -18,15 +16,15 @@ pub mod storage_witness_job_info;
 #[cfg(test)]
 mod tests;
 
-fn parse_h256(bytes: &[u8]) -> anyhow::Result<H256> {
+pub(crate) fn parse_h256(bytes: &[u8]) -> anyhow::Result<H256> {
     Ok(<[u8; 32]>::try_from(bytes).context("invalid size")?.into())
 }
 
 fn parse_h256_opt(bytes: Option<&[u8]>) -> anyhow::Result<H256> {
     parse_h256(bytes.context("missing data")?)
 }
 
-fn parse_h160(bytes: &[u8]) -> anyhow::Result<H160> {
+pub(crate) fn parse_h160(bytes: &[u8]) -> anyhow::Result<H160> {
     Ok(<[u8; 20]>::try_from(bytes).context("invalid size")?.into())
 }
 

core/lib/env_config/src/chain.rs

@@ -77,6 +77,7 @@ mod tests {
             block_commit_deadline_ms: 2500,
             miniblock_commit_deadline_ms: 1000,
             miniblock_seal_queue_capacity: 10,
+            miniblock_max_payload_size: 1_000_000,
             max_single_tx_gas: 1_000_000,
             max_allowed_l2_tx_gas_limit: 2_000_000_000,
             close_block_at_eth_params_percentage: 0.2,
@@ -123,6 +124,7 @@ mod tests {
             CHAIN_STATE_KEEPER_BLOCK_COMMIT_DEADLINE_MS="2500"
             CHAIN_STATE_KEEPER_MINIBLOCK_COMMIT_DEADLINE_MS="1000"
             CHAIN_STATE_KEEPER_MINIBLOCK_SEAL_QUEUE_CAPACITY="10"
+            CHAIN_STATE_KEEPER_MINIBLOCK_MAX_PAYLOAD_SIZE="1000000"
             CHAIN_STATE_KEEPER_MINIMAL_L2_GAS_PRICE="100000000"
             CHAIN_STATE_KEEPER_COMPUTE_OVERHEAD_PART="0.0"
             CHAIN_STATE_KEEPER_PUBDATA_OVERHEAD_PART="1.0"

core/lib/protobuf_config/src/chain.rs

@@ -37,6 +37,9 @@ impl ProtoRepr for proto::StateKeeper {
             miniblock_seal_queue_capacity: required(&self.miniblock_seal_queue_capacity)
                 .and_then(|x| Ok((*x).try_into()?))
                 .context("miniblock_seal_queue_capacity")?,
+            miniblock_max_payload_size: required(&self.miniblock_max_payload_size)
+                .and_then(|x| Ok((*x).try_into()?))
+                .context("miniblock_max_payload_size")?,
             max_single_tx_gas: *required(&self.max_single_tx_gas).context("max_single_tx_gas")?,
             max_allowed_l2_tx_gas_limit: *required(&self.max_allowed_l2_tx_gas_limit)
                 .context("max_allowed_l2_tx_gas_limit")?,
@@ -98,6 +101,7 @@ impl ProtoRepr for proto::StateKeeper {
             miniblock_seal_queue_capacity: Some(
                 this.miniblock_seal_queue_capacity.try_into().unwrap(),
             ),
+            miniblock_max_payload_size: Some(this.miniblock_max_payload_size.try_into().unwrap()),
             max_single_tx_gas: Some(this.max_single_tx_gas),
             max_allowed_l2_tx_gas_limit: Some(this.max_allowed_l2_tx_gas_limit),
             reject_tx_at_geometry_percentage: Some(this.reject_tx_at_geometry_percentage),

core/lib/protobuf_config/src/proto/chain.proto

@@ -33,6 +33,7 @@ message StateKeeper {
   optional bool save_call_traces = 22; // required
   optional uint64 enum_index_migration_chunk_size = 26; // optional
   optional uint64 max_circuits_per_batch = 27; // required
+  optional uint64 miniblock_max_payload_size = 28; // required
   reserved 23; reserved "virtual_blocks_interval";
   reserved 24; reserved "virtual_blocks_per_miniblock";
 }

core/lib/zksync_core/src/state_keeper/io/mempool.rs

@@ -30,7 +30,7 @@ use crate::{
         },
         mempool_actor::l2_tx_filter,
         metrics::KEEPER_METRICS,
-        seal_criteria::{IoSealCriteria, TimeoutSealer},
+        seal_criteria::{IoSealCriteria, MiniblockMaxPayloadSizeSealer, TimeoutSealer},
         updates::UpdatesManager,
         MempoolGuard,
     },
@@ -45,6 +45,7 @@ pub struct MempoolIO {
     mempool: MempoolGuard,
     pool: ConnectionPool<Core>,
     timeout_sealer: TimeoutSealer,
+    miniblock_max_payload_size_sealer: MiniblockMaxPayloadSizeSealer,
     filter: L2TxFilter,
     l1_batch_params_provider: L1BatchParamsProvider,
     fee_account: Address,
@@ -63,7 +64,11 @@ impl IoSealCriteria for MempoolIO {
     }
 
     fn should_seal_miniblock(&mut self, manager: &UpdatesManager) -> bool {
-        self.timeout_sealer.should_seal_miniblock(manager)
+        if self.timeout_sealer.should_seal_miniblock(manager) {
+            return true;
+        }
+        self.miniblock_max_payload_size_sealer
+            .should_seal_miniblock(manager)
     }
 }
 
@@ -413,6 +418,7 @@ impl MempoolIO {
             mempool,
             pool,
             timeout_sealer: TimeoutSealer::new(config),
+            miniblock_max_payload_size_sealer: MiniblockMaxPayloadSizeSealer::new(config),
             filter: L2TxFilter::default(),
             // ^ Will be initialized properly on the first newly opened batch
             l1_batch_params_provider,

core/lib/zksync_core/src/state_keeper/seal_criteria/mod.rs

@@ -177,6 +177,23 @@ impl IoSealCriteria for TimeoutSealer {
     }
 }
 
+#[derive(Debug, Clone, Copy)]
+pub(super) struct MiniblockMaxPayloadSizeSealer {
+    max_payload_size: usize,
+}
+
+impl MiniblockMaxPayloadSizeSealer {
+    pub fn new(config: &StateKeeperConfig) -> Self {
+        Self {
+            max_payload_size: config.miniblock_max_payload_size,
+        }
+    }
+
+    pub fn should_seal_miniblock(&mut self, manager: &UpdatesManager) -> bool {
+        manager.miniblock.payload_encoding_size >= self.max_payload_size
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use zksync_utils::time::seconds_since_epoch;
@@ -186,8 +203,7 @@ mod tests {
         create_execution_result, create_transaction, create_updates_manager,
     };
 
-    fn apply_tx_to_manager(manager: &mut UpdatesManager) {
-        let tx = create_transaction(10, 100);
+    fn apply_tx_to_manager(tx: Transaction, manager: &mut UpdatesManager) {
         manager.extend_from_executed_transaction(
             tx,
             create_execution_result(0, []),
@@ -215,7 +231,7 @@ mod tests {
         );
 
         // Non-empty miniblock should trigger.
-        apply_tx_to_manager(&mut manager);
+        apply_tx_to_manager(create_transaction(10, 100), &mut manager);
         assert!(
             timeout_miniblock_sealer.should_seal_miniblock(&manager),
             "Non-empty miniblock with old timestamp should be sealed"
@@ -230,4 +246,27 @@ mod tests {
             "Non-empty miniblock with too recent timestamp shouldn't be sealed"
         );
     }
+
+    #[test]
+    fn max_size_miniblock_sealer() {
+        let tx = create_transaction(10, 100);
+        let tx_encoding_size =
+            zksync_protobuf::repr::encode::<zksync_dal::consensus::proto::Transaction>(&tx).len();
+
+        let mut max_payload_sealer = MiniblockMaxPayloadSizeSealer {
+            max_payload_size: tx_encoding_size,
+        };
+
+        let mut manager = create_updates_manager();
+        assert!(
+            !max_payload_sealer.should_seal_miniblock(&manager),
+            "Empty miniblock shouldn't be sealed"
+        );
+
+        apply_tx_to_manager(tx, &mut manager);
+        assert!(
+            max_payload_sealer.should_seal_miniblock(&manager),
+            "Miniblock with payload encoding size equal or greater than max payload size should be sealed"
+        );
+    }
 }

core/lib/zksync_core/src/state_keeper/updates/miniblock_updates.rs

@@ -26,6 +26,7 @@ pub struct MiniblockUpdates {
     pub l1_gas_count: BlockGasCount,
     pub block_execution_metrics: ExecutionMetrics,
     pub txs_encoding_size: usize,
+    pub payload_encoding_size: usize,
     pub timestamp: u64,
     pub number: MiniblockNumber,
     pub prev_block_hash: H256,
@@ -51,6 +52,7 @@ impl MiniblockUpdates {
             l1_gas_count: BlockGasCount::default(),
             block_execution_metrics: ExecutionMetrics::default(),
             txs_encoding_size: 0,
+            payload_encoding_size: 0,
             timestamp,
             number,
             prev_block_hash,
@@ -130,6 +132,8 @@ impl MiniblockUpdates {
         self.l1_gas_count += tx_l1_gas_this_tx;
         self.block_execution_metrics += execution_metrics;
         self.txs_encoding_size += tx.bootloader_encoding_size();
+        self.payload_encoding_size +=
+            zksync_protobuf::repr::encode::<zksync_dal::consensus::proto::Transaction>(&tx).len();
         self.storage_logs
             .extend(tx_execution_result.logs.storage_logs);
 
@@ -183,6 +187,9 @@ mod tests {
         );
         let tx = create_transaction(10, 100);
         let bootloader_encoding_size = tx.bootloader_encoding_size();
+        let payload_encoding_size =
+            zksync_protobuf::repr::encode::<zksync_dal::consensus::proto::Transaction>(&tx).len();
+
         accumulator.extend_from_executed_transaction(
             tx,
             create_execution_result(0, []),
@@ -201,5 +208,6 @@ mod tests {
         assert_eq!(accumulator.new_factory_deps.len(), 0);
         assert_eq!(accumulator.block_execution_metrics.l2_to_l1_logs, 0);
         assert_eq!(accumulator.txs_encoding_size, bootloader_encoding_size);
+        assert_eq!(accumulator.payload_encoding_size, payload_encoding_size);
     }
 }

etc/env/base/chain.toml

@@ -21,6 +21,7 @@ max_allowed_l2_tx_gas_limit = 1125899906842624
 block_commit_deadline_ms = 5000
 miniblock_commit_deadline_ms = 1000
 miniblock_seal_queue_capacity = 10
+miniblock_max_payload_size=1000000
 # Max gas that can used to include single block in aggregated operation
 max_single_tx_gas = 6000000
 

etc/env/consensus_config.yaml

@@ -2,7 +2,7 @@ server_addr: '127.0.0.1:3054'
 public_addr: '127.0.0.1:3054'
 validators:
 - 'validator:public:bn254:8b0ff0ad1a250e64b0209277148ccee3b64534d8fa60cf25ba0bcc8b65d4d89309cdae79197c2db873d351401093fa0542a5a2071c1a247f2e1abe56d08cbabb'
-max_payload_size: 5000000
+max_payload_size: 2500000
 gossip_static_inbound:
 - 'node:public:ed25519:147bb71be895846e1d6f5b1c6a8be53848b82bdafcf66e9dfe6ca65581076a1d'
 gossip_dynamic_inbound_limit: 0

etc/env/en_consensus_config.yaml

@@ -2,7 +2,7 @@ server_addr: '127.0.0.1:3055'
 public_addr: '127.0.0.1:3055'
 validators:
 - 'validator:public:bn254:8b0ff0ad1a250e64b0209277148ccee3b64534d8fa60cf25ba0bcc8b65d4d89309cdae79197c2db873d351401093fa0542a5a2071c1a247f2e1abe56d08cbabb'
-max_payload_size: 5000000
+max_payload_size: 2500000
 gossip_dynamic_inbound_limit: 0
 gossip_static_outbound:
 - key: 'node:public:ed25519:ee717abba6aec5baae5e09d457bd2ffc2f121b576cf4170ce15a68163ce4c868'

etc/env/file_based/general.yaml

@@ -70,6 +70,7 @@ state_keeper:
   block_commit_deadline_ms: 2500
   miniblock_commit_deadline_ms: 1000
   miniblock_seal_queue_capacity: 10
+  miniblock_max_payload_size: 1000000
   max_single_tx_gas: 6000000
   close_block_at_geometry_percentage: 0.95
   close_block_at_eth_params_percentage: 0.95