feat(katana): automatically close block if resources limit reached

crates/dojo/test-utils/src/sequencer.rs

@@ -6,6 +6,7 @@ use katana_core::constants::DEFAULT_SEQUENCER_ADDRESS;
 use katana_executor::implementation::blockifier::BlockifierFactory;
 use katana_node::config::dev::DevConfig;
 use katana_node::config::rpc::{RpcConfig, DEFAULT_RPC_ADDR};
+use katana_node::config::sequencing::SequencingConfig;
 pub use katana_node::config::*;
 use katana_node::LaunchedNode;
 use katana_primitives::chain::ChainId;

crates/katana/chain-spec/src/rollup/utils.rs

@@ -336,7 +336,7 @@ mod tests {
 
     use alloy_primitives::U256;
     use katana_executor::implementation::blockifier::BlockifierFactory;
-    use katana_executor::ExecutorFactory;
+    use katana_executor::{BlockLimits, ExecutorFactory};
     use katana_primitives::chain::ChainId;
     use katana_primitives::contract::Nonce;
     use katana_primitives::env::CfgEnv;
@@ -386,6 +386,7 @@ mod tests {
                 ..Default::default()
             },
             Default::default(),
+            BlockLimits::max(),
         )
     }
 

crates/katana/cli/src/args.rs

@@ -19,7 +19,8 @@ use katana_node::config::metrics::MetricsConfig;
 use katana_node::config::rpc::RpcConfig;
 #[cfg(feature = "server")]
 use katana_node::config::rpc::{RpcModuleKind, RpcModulesList};
-use katana_node::config::{Config, SequencingConfig};
+use katana_node::config::sequencing::SequencingConfig;
+use katana_node::config::Config;
 use katana_primitives::chain::ChainId;
 use katana_primitives::genesis::allocation::DevAllocationsGenerator;
 use katana_primitives::genesis::constant::DEFAULT_PREFUNDED_ACCOUNT_BALANCE;
@@ -58,6 +59,10 @@ pub struct NodeArgs {
     #[arg(value_name = "MILLISECONDS")]
     pub block_time: Option<u64>,
 
+    #[arg(long = "sequencing.block-max-cairo-steps")]
+    #[arg(value_name = "TOTAL")]
+    pub block_cairo_steps_limit: Option<u64>,
+
     /// Directory path of the database to initialize from.
     ///
     /// The path must either be an empty directory or a directory which already contains a
@@ -187,7 +192,11 @@ impl NodeArgs {
     }
 
     fn sequencer_config(&self) -> SequencingConfig {
-        SequencingConfig { block_time: self.block_time, no_mining: self.no_mining }
+        SequencingConfig {
+            block_time: self.block_time,
+            no_mining: self.no_mining,
+            block_cairo_steps_limit: self.block_cairo_steps_limit,
+        }
     }
 
     fn rpc_config(&self) -> Result<RpcConfig> {

crates/katana/core/src/service/block_producer.rs

@@ -1,3 +1,14 @@
+//! **********************************************************************************************
+//!
+//!     "We are all in the gutter, but some of us are looking at the stars."
+//!                                                       — Oscar Wilde, Lady Windermere's Fan
+//!
+//!     Within this imperfect realm lies a spark of aspiration. What you find may be tangled
+//!     and weathered, but in its heart beats the rhythm of possibility. Tread gently, dear
+//!     wanderer, and perhaps together we can guide it toward those distant stars.
+//!
+//! **********************************************************************************************
+
 use std::collections::VecDeque;
 use std::future::Future;
 use std::pin::Pin;
@@ -46,6 +57,16 @@ pub enum BlockProductionError {
     TransactionExecutionError(#[from] katana_executor::ExecutorError),
 }
 
+impl BlockProductionError {
+    /// Returns `true` if the error is caused by block limit being exhausted.
+    pub fn is_block_limit_exhausted(&self) -> bool {
+        matches!(
+            self,
+            Self::TransactionExecutionError(katana_executor::ExecutorError::LimitsExhausted)
+        )
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct MinedBlockOutcome {
     pub block_number: u64,
@@ -65,7 +86,8 @@ type ServiceFuture<T> = Pin<Box<dyn Future<Output = BlockingTaskResult<T>> + Sen
 type BlockProductionResult = Result<MinedBlockOutcome, BlockProductionError>;
 type BlockProductionFuture = ServiceFuture<Result<MinedBlockOutcome, BlockProductionError>>;
 
-type TxExecutionResult = Result<Vec<TxWithOutcome>, BlockProductionError>;
+type TxExecutionResult =
+    Result<(Vec<TxWithOutcome>, Option<Vec<ExecutableTxWithHash>>), BlockProductionError>;
 type TxExecutionFuture = ServiceFuture<TxExecutionResult>;
 
 type BlockProductionWithTxnsFuture =
@@ -213,6 +235,8 @@ pub struct IntervalBlockProducer<EF: ExecutorFactory> {
     /// - `block_time` is `Some`,
     /// - and, at least one transaction has been executed and thus a new block is opened.
     timer: Option<Interval>,
+
+    is_block_full: bool,
 }
 
 impl<EF: ExecutorFactory> IntervalBlockProducer<EF> {
@@ -236,6 +260,7 @@ impl<EF: ExecutorFactory> IntervalBlockProducer<EF> {
             TxValidator::new(state, flags.clone(), cfg.clone(), block_env, permit.clone());
 
         Self {
+            is_block_full: false,
             validator,
             permit,
             backend,
@@ -309,20 +334,19 @@ impl<EF: ExecutorFactory> IntervalBlockProducer<EF> {
 
     fn execute_transactions(
         executor: PendingExecutor,
-        transactions: Vec<ExecutableTxWithHash>,
+        mut transactions: Vec<ExecutableTxWithHash>,
     ) -> TxExecutionResult {
         let executor = &mut executor.write();
 
-        let new_txs_count = transactions.len();
-        executor.execute_transactions(transactions)?;
+        let (total_executed, is_full) = executor.execute_transactions(transactions.clone())?;
 
         let txs = executor.transactions();
         let total_txs = txs.len();
 
         // Take only the results of the newly executed transactions
         let results = txs
             .iter()
-            .skip(total_txs - new_txs_count)
+            .skip(total_txs.saturating_sub(total_executed))
             .filter_map(|(tx, res)| match res {
                 ExecutionResult::Failed { .. } => None,
                 ExecutionResult::Success { receipt, trace, .. } => Some(TxWithOutcome {
@@ -333,7 +357,10 @@ impl<EF: ExecutorFactory> IntervalBlockProducer<EF> {
             })
             .collect::<Vec<TxWithOutcome>>();
 
-        Ok(results)
+        let non_executed_txs =
+            if is_full.is_some() { Some(transactions.split_off(total_executed)) } else { None };
+
+        Ok((results, non_executed_txs))
     }
 
     fn create_new_executor_for_next_block(&self) -> Result<PendingExecutor, BlockProductionError> {
@@ -393,7 +420,17 @@ impl<EF: ExecutorFactory> Stream for IntervalBlockProducer<EF> {
 
         if let Some(mut timer) = pin.timer.take() {
             // Mine block if the interval is over
-            if timer.poll_tick(cx).is_ready() && pin.ongoing_mining.is_none() {
+            //
+            // if block is already full but the timer hasn't ready yet, we will still mine but we
+            // don't have to do anything to the timer as it will be dropped and reset once new
+            // transaction is executed.
+            if (timer.poll_tick(cx).is_ready() || pin.is_block_full) && pin.ongoing_mining.is_none()
+            {
+                if pin.is_block_full {
+                    info!("Block has reached capacity! Closing block...");
+                    pin.is_block_full = false;
+                }
+
                 pin.ongoing_mining = Some(Box::pin({
                     let executor = pin.executor.clone();
                     let backend = pin.backend.clone();
@@ -402,9 +439,21 @@ impl<EF: ExecutorFactory> Stream for IntervalBlockProducer<EF> {
                     pin.blocking_task_spawner.spawn(|| Self::do_mine(permit, executor, backend))
                 }));
             } else {
-                // Unable to close the block due to ongoing mining.
                 pin.timer = Some(timer);
             }
+        } else if pin.is_block_full && pin.ongoing_mining.is_none() {
+            info!("Block has reached capacity! Closing block...");
+
+            pin.ongoing_mining = Some(Box::pin({
+                let executor = pin.executor.clone();
+                let backend = pin.backend.clone();
+                let permit = pin.permit.clone();
+
+                pin.blocking_task_spawner.spawn(|| Self::do_mine(permit, executor, backend))
+            }));
+
+            pin.is_block_full = false;
+            pin.timer = None;
         }
 
         loop {
@@ -438,7 +487,18 @@ impl<EF: ExecutorFactory> Stream for IntervalBlockProducer<EF> {
             if let Some(mut execution) = pin.ongoing_execution.take() {
                 if let Poll::Ready(executor) = execution.poll_unpin(cx) {
                     match executor {
-                        Ok(Ok(txs)) => {
+                        Ok(Ok((txs, leftovers))) => {
+                            if let Some(leftovers) = leftovers {
+                                pin.is_block_full = true;
+
+                                // Push leftover transactions back to front of queue
+                                pin.queued.push_front(leftovers);
+
+                                // Schedule future poll if block is full
+                                cx.waker().wake_by_ref();
+                                break;
+                            }
+
                             pin.notify_listener(txs);
                             continue;
                         }
@@ -486,6 +546,8 @@ impl<EF: ExecutorFactory> Stream for IntervalBlockProducer<EF> {
                             Err(e) => return Poll::Ready(Some(Err(e))),
                         }
 
+                        pin.is_block_full = false;
+
                         return Poll::Ready(Some(outcome));
                     }
 

crates/katana/core/src/service/block_producer_tests.rs

@@ -5,7 +5,6 @@ use katana_executor::implementation::noop::NoopExecutorFactory;
 use katana_primitives::transaction::{ExecutableTx, InvokeTx};
 use katana_primitives::Felt;
 use katana_provider::providers::db::DbProvider;
-use tokio::time;
 
 use super::*;
 use crate::backend::gas_oracle::GasOracle;
@@ -47,27 +46,31 @@ async fn interval_force_mine_without_transactions() {
 async fn interval_mine_after_timer() {
     let backend = test_backend();
     let mut producer = IntervalBlockProducer::new(backend.clone(), Some(1000));
-    // Initial state
+    // no timer should be set when no block is opened.
     assert!(producer.timer.is_none());
 
     producer.queued.push_back(vec![dummy_transaction()]);
 
     let stream = producer;
     pin_mut!(stream);
 
-    // Process the transaction, the timer should be automatically started
-    let _ = stream.next().await;
-    assert!(stream.timer.is_some());
+    let waker = futures::task::noop_waker();
+    let mut context = Context::from_waker(&waker);
 
-    // Advance time to trigger mining
-    time::sleep(Duration::from_secs(1)).await;
-    let result = stream.next().await.expect("should mine block").unwrap();
+    // mine the block
+    let poll_result = stream.as_mut().poll_next(&mut context);
 
-    assert_eq!(result.block_number, 1);
-    assert_eq!(backend.blockchain.provider().latest_number().unwrap(), 1);
+    // based on how the `Stream` trait is implemented, there is a possibility that a single
+    // call to `poll_next` can complete the whole production flow so we added this just in case.
+    if poll_result.is_pending() {
+        assert!(stream.timer.is_some(), "timer should start once we received a tx");
+    } else {
+        assert!(stream.timer.is_none(), "no timer if block has been mined");
+    }
 
-    // Final state
-    assert!(stream.timer.is_none());
+    let outcome = stream.next().await.expect("should mine block").unwrap();
+    assert_eq!(outcome.block_number, 1);
+    assert_eq!(backend.blockchain.provider().latest_number().unwrap(), 1);
 }
 
 // Helper functions to create test transactions

crates/katana/core/tests/backend.rs

@@ -5,6 +5,7 @@ use katana_core::backend::gas_oracle::GasOracle;
 use katana_core::backend::storage::{Blockchain, Database};
 use katana_core::backend::Backend;
 use katana_executor::implementation::blockifier::BlockifierFactory;
+use katana_executor::BlockLimits;
 use katana_primitives::chain::ChainId;
 use katana_primitives::env::CfgEnv;
 use katana_primitives::felt;
@@ -25,6 +26,7 @@ fn executor(chain_spec: &ChainSpec) -> BlockifierFactory {
             ..Default::default()
         },
         Default::default(),
+        BlockLimits::max(),
     )
 }
 

crates/katana/executor/benches/concurrent.rs

@@ -9,7 +9,7 @@ use std::time::Duration;
 use criterion::measurement::WallTime;
 use criterion::{criterion_group, criterion_main, BatchSize, BenchmarkGroup, Criterion};
 use katana_executor::implementation::blockifier::BlockifierFactory;
-use katana_executor::{ExecutionFlags, ExecutorFactory};
+use katana_executor::{BlockLimits, ExecutionFlags, ExecutorFactory};
 use katana_primitives::env::{BlockEnv, CfgEnv};
 use katana_primitives::transaction::ExecutableTxWithHash;
 use katana_provider::test_utils;
@@ -44,7 +44,7 @@ fn blockifier(
     (block_env, cfg_env): (BlockEnv, CfgEnv),
     tx: ExecutableTxWithHash,
 ) {
-    let factory = Arc::new(BlockifierFactory::new(cfg_env, flags.clone()));
+    let factory = Arc::new(BlockifierFactory::new(cfg_env, flags.clone(), BlockLimits::max()));
 
     group.bench_function("Blockifier.1", |b| {
         b.iter_batched(

crates/katana/executor/benches/execution.rs

@@ -49,7 +49,9 @@ fn blockifier(
 
                 (state, &block_context, execution_flags, tx.clone())
             },
-            |(mut state, block_context, flags, tx)| transact(&mut state, block_context, flags, tx),
+            |(mut state, block_context, flags, tx)| {
+                transact(&mut state, block_context, flags, tx, None)
+            },
             BatchSize::SmallInput,
         )
     });

crates/katana/executor/src/abstraction/error.rs

@@ -4,7 +4,13 @@ use katana_primitives::Felt;
 
 /// Errors that can be returned by the executor.
 #[derive(Debug, thiserror::Error)]
-pub enum ExecutorError {}
+pub enum ExecutorError {
+    #[error("Limits exhausted")]
+    LimitsExhausted,
+
+    #[error(transparent)]
+    Other(Box<dyn core::error::Error + Send + Sync + 'static>),
+}
 
 /// Errors that can occur during the transaction execution.
 #[derive(Debug, Clone, thiserror::Error)]

crates/katana/executor/src/abstraction/executor.rs

@@ -5,6 +5,7 @@ use katana_primitives::transaction::{ExecutableTxWithHash, TxWithHash};
 use katana_primitives::Felt;
 use katana_provider::traits::state::StateProvider;
 
+use super::ExecutorError;
 use crate::{
     EntryPointCall, ExecutionError, ExecutionFlags, ExecutionOutput, ExecutionResult,
     ExecutorResult, ResultAndStates,
@@ -38,10 +39,11 @@ pub trait BlockExecutor<'a>: ExecutorExt + Send + Sync + core::fmt::Debug {
     /// Executes the given block.
     fn execute_block(&mut self, block: ExecutableBlock) -> ExecutorResult<()>;
 
+    /// Execute transactions and returns the total number of transactions that was executed.
     fn execute_transactions(
         &mut self,
         transactions: Vec<ExecutableTxWithHash>,
-    ) -> ExecutorResult<()>;
+    ) -> ExecutorResult<(usize, Option<ExecutorError>)>;
 
     /// Takes the output state of the executor.
     fn take_execution_output(&mut self) -> ExecutorResult<ExecutionOutput>;

crates/katana/executor/src/abstraction/mod.rs

@@ -17,6 +17,19 @@ use katana_trie::MultiProof;
 
 pub type ExecutorResult<T> = Result<T, error::ExecutorError>;
 
+/// See <https://docs.starknet.io/chain-info/#current_limits>.
+#[derive(Debug, Clone, Default)]
+pub struct BlockLimits {
+    /// The maximum number of Cairo steps that can be completed within each block.
+    pub cairo_steps: u64,
+}
+
+impl BlockLimits {
+    pub fn max() -> Self {
+        Self { cairo_steps: u64::MAX }
+    }
+}
+
 /// Transaction execution simulation flags.
 ///
 /// These flags can be used to control the behavior of the transaction execution, such as skipping