feat(collator): block limits by gas, size, lt

collator/src/collator/do_collate.rs

@@ -13,7 +13,7 @@ use tycho_util::metrics::HistogramGuard;
 use tycho_util::time::now_millis;
 use tycho_util::FastHashMap;
 
-use super::types::{CachedMempoolAnchor, SpecialOrigin};
+use super::types::{BlockLimitsLevel, CachedMempoolAnchor, SpecialOrigin};
 use super::CollatorStdImpl;
 use crate::collator::execution_manager::ExecutionManager;
 use crate::collator::types::{
@@ -73,6 +73,7 @@ impl CollatorStdImpl {
         collation_data.gen_utime_ms = (next_chain_time % 1000) as u16;
         collation_data.start_lt = Self::calc_start_lt(mc_data, prev_shard_data, &collation_data)?;
         collation_data.next_lt = collation_data.start_lt + 1;
+        collation_data.stats.lt_start = collation_data.start_lt;
 
         collation_data.processed_upto = prev_shard_data.processed_upto().clone();
         tracing::debug!(target: tracing_targets::COLLATOR, "initial processed_upto.externals = {:?}",
@@ -326,7 +327,7 @@ impl CollatorStdImpl {
             fill_msgs_total_elapsed += timer.elapsed();
 
             // execute msgs processing by groups
-            while !msgs_set_full_processed {
+            'execute_groups: while !msgs_set_full_processed {
                 // Process messages
                 timer = std::time::Instant::now();
                 let tick = exec_manager.tick(msgs_set_offset).await?;
@@ -374,6 +375,7 @@ impl CollatorStdImpl {
                     }
 
                     collation_data.next_lt = exec_manager.min_next_lt();
+                    collation_data.stats.lt_current = collation_data.next_lt;
                 }
 
                 msgs_set_offset = tick.new_offset;
@@ -396,17 +398,21 @@ impl CollatorStdImpl {
                 }
             }
 
-            timer = std::time::Instant::now();
-
-            // HACK: temporary always full process msgs set and check block limits after
-            if collation_data.tx_count >= self.config.block_txs_limit {
-                tracing::debug!(target: tracing_targets::COLLATOR,
-                    "STUB: block limit reached: {}/{}",
-                    collation_data.tx_count, self.config.block_txs_limit,
-                );
-                block_limits_reached = true;
+                if let BlockLimitsLevel::Hard = collation_data
+                    .stats
+                    .current_level(&self.config.block_limits)
+                {
+                    tracing::debug!(target: tracing_targets::COLLATOR,
+                        "STUB: block limit reached: {:?}/{:?}",
+                        collation_data.stats, self.config.block_limits,
+                    );
+                    block_limits_reached = true;
+                    break 'execute_groups;
+                }
             }
 
+            timer = std::time::Instant::now();
+
             // commit messages to iterator only if set was fully processed
             if msgs_set_full_processed {
                 self.mq_adapter.commit_messages_to_iterator(
@@ -1300,6 +1306,24 @@ fn new_transaction(
     );
 
     collation_data.execute_count_all += 1;
+    let transaction = executor_output.transaction.load()?;
+    let gas_used = 'gas: {
+        match transaction.load_info()? {
+            TxInfo::Ordinary(info) => {
+                if let ComputePhase::Executed(phase) = &info.compute_phase {
+                    break 'gas phase.gas_used.into_inner();
+                }
+            }
+            TxInfo::TickTock(info) => {
+                if let ComputePhase::Executed(phase) = &info.compute_phase {
+                    break 'gas phase.gas_used.into_inner();
+                }
+            }
+        };
+        0
+    };
+    collation_data.stats.gas_used += gas_used as u32;
+    collation_data.stats.add_cell(&*in_msg.cell)?;
 
     let import_fees;
     let in_msg_hash = *in_msg.cell.repr_hash();
@@ -1434,6 +1458,7 @@ fn new_transaction(
 
     for out_msg_cell in executor_output.out_msgs.values() {
         let out_msg_cell = out_msg_cell?;
+        collation_data.stats.add_cell(&*out_msg_cell)?;
         let out_msg_hash = *out_msg_cell.repr_hash();
         let out_msg_info = out_msg_cell.parse::<MsgInfo>()?;
 
@@ -1443,7 +1468,6 @@ fn new_transaction(
             info = ?out_msg_info,
             "adding out message to out_msgs",
         );
-
         match &out_msg_info {
             MsgInfo::Int(IntMsgInfo { fwd_fee, dst, .. }) => {
                 collation_data.int_enqueue_count += 1;

collator/src/collator/types.rs

@@ -1,16 +1,17 @@
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashSet};
 use std::sync::{Arc, OnceLock};
 
 use anyhow::{anyhow, bail, Result};
 use everscale_types::cell::{Cell, HashBytes, UsageTree, UsageTreeMode};
 use everscale_types::dict::Dict;
 use everscale_types::models::{
-    Account, AccountState, BlockId, BlockIdShort, BlockInfo, BlockRef, BlockchainConfig,
-    CurrencyCollection, HashUpdate, ImportFees, InMsg, Lazy, LibDescr, McStateExtra, MsgInfo,
-    OptionalAccount, OutMsg, PrevBlockRef, ProcessedUptoInfo, ShardAccount, ShardAccounts,
-    ShardDescription, ShardFeeCreated, ShardFees, ShardIdent, ShardIdentFull, SimpleLib,
-    SpecialFlags, StateInit, Transaction, ValueFlow,
+    Account, AccountState, BlockId, BlockIdShort, BlockInfo, BlockLimits, BlockParamLimits,
+    BlockRef, BlockchainConfig, CurrencyCollection, HashUpdate, ImportFees, InMsg, Lazy, LibDescr,
+    McStateExtra, MsgInfo, OptionalAccount, OutMsg, PrevBlockRef, ProcessedUptoInfo, ShardAccount,
+    ShardAccounts, ShardDescription, ShardFeeCreated, ShardFees, ShardIdent, ShardIdentFull,
+    SimpleLib, SpecialFlags, StateInit, Transaction, ValueFlow,
 };
+use everscale_types::prelude::DynCell;
 use tycho_block_util::dict::RelaxedAugDict;
 use tycho_block_util::state::{MinRefMcStateTracker, ShardStateStuff};
 use tycho_util::FastHashMap;
@@ -322,6 +323,8 @@ pub(super) struct BlockCollationData {
 
     pub tx_count: u32,
 
+    pub stats: BlockStats,
+
     pub total_execute_msgs_time_mc: u128,
 
     pub execute_count_all: u32,
@@ -374,6 +377,86 @@ pub(super) struct BlockCollationData {
     // TODO: set from anchor
     pub created_by: HashBytes,
 }
+#[derive(Debug, Default)]
+pub struct BlockStats {
+    pub gas_used: u32,
+    pub lt_current: u64,
+    pub lt_start: u64,
+    pub cells_seen: HashSet<HashBytes>,
+    pub cells_bits: u32,
+}
+
+impl BlockStats {
+    pub fn add_cell(&mut self, cell: &DynCell) -> Result<()> {
+        if !self.cells_seen.insert(*cell.repr_hash()) {
+            return Ok(());
+        } else {
+            let bits = cell.bit_len() as u32;
+            self.cells_bits += bits;
+        }
+        for i in 0..cell.reference_count() {
+            self.add_cell(cell.reference(i).ok_or(anyhow::anyhow!("wrong cell"))?)?
+        }
+        Ok(())
+    }
+    pub fn current_level(&self, block_limits: &BlockLimits) -> BlockLimitsLevel {
+        let mut result = BlockLimitsLevel::Underload;
+        let BlockLimits {
+            bytes,
+            gas,
+            lt_delta,
+        } = block_limits;
+
+        let BlockParamLimits {
+            soft_limit,
+            hard_limit,
+            ..
+        } = bytes;
+
+        let cells_bytes = self.cells_bits / 8;
+        if cells_bytes >= *hard_limit {
+            return BlockLimitsLevel::Hard;
+        }
+        if cells_bytes >= *soft_limit {
+            result = BlockLimitsLevel::Soft
+        }
+
+        let BlockParamLimits {
+            soft_limit,
+            hard_limit,
+            ..
+        } = gas;
+
+        if self.gas_used >= *hard_limit {
+            return BlockLimitsLevel::Hard;
+        }
+        if self.gas_used >= *soft_limit {
+            result = BlockLimitsLevel::Soft
+        }
+
+        let BlockParamLimits {
+            soft_limit,
+            hard_limit,
+            ..
+        } = lt_delta;
+
+        let delta_lt = (self.lt_current - self.lt_start) as u32;
+        if delta_lt >= *hard_limit {
+            return BlockLimitsLevel::Hard;
+        }
+        if delta_lt >= *soft_limit {
+            result = BlockLimitsLevel::Soft
+        }
+        result
+    }
+}
+
+#[derive(Debug, Clone, Copy, Eq, Ord, PartialEq, PartialOrd)]
+pub enum BlockLimitsLevel {
+    Underload,
+    Soft,
+    Hard,
+}
 
 #[derive(Debug)]
 pub struct PreparedInMsg {

collator/src/types.rs

@@ -4,8 +4,8 @@ use std::time::Duration;
 use everscale_crypto::ed25519::KeyPair;
 use everscale_types::cell::HashBytes;
 use everscale_types::models::{
-    Block, BlockId, BlockInfo, CurrencyCollection, GlobalCapabilities, GlobalCapability, IntAddr,
-    ShardIdent, Signature, ValueFlow,
+    Block, BlockId, BlockInfo, BlockLimits, BlockParamLimits, CurrencyCollection,
+    GlobalCapabilities, GlobalCapability, IntAddr, ShardIdent, Signature, ValueFlow,
 };
 use serde::{Deserialize, Serialize};
 use tycho_block_util::block::{BlockStuffAug, ValidatorSubsetInfo};
@@ -25,7 +25,7 @@ pub struct CollationConfig {
     pub max_uncommitted_chain_length: u32,
     pub uncommitted_chain_to_import_next_anchor: u32,
 
-    pub block_txs_limit: u32,
+    pub block_limits: BlockLimits,
 
     pub msgs_exec_params: MsgsExecutionParams,
 }
@@ -42,7 +42,23 @@ impl Default for CollationConfig {
             max_uncommitted_chain_length: 31,
             uncommitted_chain_to_import_next_anchor: 4,
 
-            block_txs_limit: 10000,
+            block_limits: BlockLimits {
+                bytes: BlockParamLimits {
+                    underload: 131072,
+                    soft_limit: 524288,
+                    hard_limit: 1048576,
+                },
+                gas: BlockParamLimits {
+                    underload: 900000,
+                    soft_limit: 1200000,
+                    hard_limit: 2000000,
+                },
+                lt_delta: BlockParamLimits {
+                    underload: 1000,
+                    soft_limit: 5000,
+                    hard_limit: 10000,
+                },
+            },
 
             msgs_exec_params: MsgsExecutionParams::default(),
         }