feat(katana): added chunk size limit to events method

bin/katana/src/cli/node.rs

@@ -222,6 +222,7 @@ impl NodeArgs {
             addr: self.server.http_addr,
             max_connections: self.server.max_connections,
             cors_origins: self.server.http_cors_origins.clone(),
+            page_size,
         }
     }
 

bin/katana/src/cli/options.rs

@@ -12,7 +12,7 @@ use std::net::IpAddr;
 use clap::Args;
 use katana_node::config::execution::{DEFAULT_INVOCATION_MAX_STEPS, DEFAULT_VALIDATION_MAX_STEPS};
 use katana_node::config::metrics::{DEFAULT_METRICS_ADDR, DEFAULT_METRICS_PORT};
-use katana_node::config::rpc::{DEFAULT_RPC_ADDR, DEFAULT_RPC_MAX_CONNECTIONS, DEFAULT_RPC_PORT};
+use katana_node::config::rpc::{DEFAULT_RPC_ADDR, DEFAULT_RPC_MAX_CONNECTIONS, DEFAULT_RPC_PORT, DEFAULT_RPC_PAGE_SIZE};
 use katana_primitives::block::BlockHashOrNumber;
 use katana_primitives::chain::ChainId;
 use katana_primitives::genesis::Genesis;
@@ -82,6 +82,13 @@ pub struct ServerOptions {
     #[arg(default_value_t = DEFAULT_RPC_MAX_CONNECTIONS)]
     #[serde(default = "default_max_connections")]
     pub max_connections: u32,
+
+    /// Maximum page size for event queries.
+    #[arg(long = "rpc.max-event-page-size", value_name = "SIZE")]
+    #[arg(default_value_t = DEFAULT_RPC_PAGE_SIZE)]
+    #[serde(default = "default_page_size")]
+    pub max_event_page_size: u64,
+
 }
 
 impl Default for ServerOptions {
@@ -91,6 +98,7 @@ impl Default for ServerOptions {
             http_port: DEFAULT_RPC_PORT,
             max_connections: DEFAULT_RPC_MAX_CONNECTIONS,
             http_cors_origins: None,
+            page_size: DEFAULT_RPC_PAGE_SIZE,
         }
     }
 }
@@ -343,3 +351,7 @@ fn default_http_port() -> u16 {
 fn default_max_connections() -> u32 {
     DEFAULT_RPC_MAX_CONNECTIONS
 }
+
+fn default_page_size() -> u64 {
+    DEFAULT_RPC_PAGE_SIZE
+}

crates/katana/node/src/config/rpc.rs

@@ -5,7 +5,7 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 pub const DEFAULT_RPC_MAX_CONNECTIONS: u32 = 100;
 pub const DEFAULT_RPC_ADDR: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
 pub const DEFAULT_RPC_PORT: u16 = 5050;
-
+pub const DEFAULT_RPC_PAGE_SIZE: u64 = 100;
 /// List of APIs supported by Katana.
 #[derive(
     Debug, Copy, Clone, PartialEq, Eq, Hash, strum_macros::EnumString, strum_macros::Display,
@@ -24,6 +24,7 @@ pub struct RpcConfig {
     pub port: u16,
     pub max_connections: u32,
     pub apis: HashSet<ApiKind>,
+    pub max_event_page_size: Option<u64>,
     pub cors_origins: Option<Vec<String>>,
 }
 
@@ -42,6 +43,7 @@ impl Default for RpcConfig {
             port: DEFAULT_RPC_PORT,
             max_connections: DEFAULT_RPC_MAX_CONNECTIONS,
             apis: HashSet::from([ApiKind::Starknet]),
+            page_size: Some(DEFAULT_RPC_PAGE_SIZE),
         }
     }
 }

crates/katana/node/src/lib.rs

@@ -49,6 +49,7 @@ use katana_rpc_api::saya::SayaApiServer;
 use katana_rpc_api::starknet::{StarknetApiServer, StarknetTraceApiServer, StarknetWriteApiServer};
 use katana_rpc_api::torii::ToriiApiServer;
 use katana_tasks::TaskManager;
+use katana_rpc::starknet::StarknetApiConfig;
 use tower_http::cors::{AllowOrigin, CorsLayer};
 use tracing::info;
 
@@ -277,16 +278,20 @@ pub async fn spawn<EF: ExecutorFactory>(
     methods.register_method("health", |_, _| Ok(serde_json::json!({ "health": true })))?;
 
     if config.apis.contains(&ApiKind::Starknet) {
+        let starknet_api_config = StarknetApiConfig {
+            page_size: config.page_size.unwrap_or(1000), // Default value
+        };
         let server = if let Some(client) = forked_client {
             StarknetApi::new_forked(
                 backend.clone(),
                 pool.clone(),
                 block_producer.clone(),
                 validator,
                 client,
+                starknet_api_config,
             )
         } else {
-            StarknetApi::new(backend.clone(), pool.clone(), block_producer.clone(), validator)
+            StarknetApi::new(backend.clone(), pool.clone(), block_producer.clone(), validator,starknet_api_config)
         };
 
         methods.merge(StarknetApiServer::into_rpc(server.clone()))?;

crates/katana/rpc/rpc-types/src/error/starknet.rs

@@ -9,6 +9,7 @@ use serde::Serialize;
 use serde_json::Value;
 use starknet::core::types::StarknetError as StarknetRsError;
 use starknet::providers::ProviderError as StarknetRsProviderError;
+use serde_json::json;
 
 /// Possible list of errors that can be returned by the Starknet API according to the spec: <https://github.com/starkware-libs/starknet-specs>.
 #[derive(Debug, thiserror::Error, Clone, Serialize)]
@@ -32,7 +33,7 @@ pub enum StarknetApiError {
     #[error("Class hash not found")]
     ClassHashNotFound,
     #[error("Requested page size is too big")]
-    PageSizeTooBig,
+    PageSizeTooBig { requested: u64, max_allowed: u64 },
     #[error("There are no blocks")]
     NoBlocks,
     #[error("The supplied continuation token is invalid or unknown")]
@@ -96,7 +97,7 @@ impl StarknetApiError {
             StarknetApiError::InvalidTxnIndex => 27,
             StarknetApiError::ClassHashNotFound => 28,
             StarknetApiError::TxnHashNotFound => 29,
-            StarknetApiError::PageSizeTooBig => 31,
+            StarknetApiError::PageSizeTooBig { .. } => 31,
             StarknetApiError::NoBlocks => 32,
             StarknetApiError::InvalidContinuationToken => 33,
             StarknetApiError::TooManyKeysInFilter => 34,
@@ -136,6 +137,7 @@ impl StarknetApiError {
                 Some(Value::String(reason.to_string()))
             }
 
+            StarknetApiError::PageSizeTooBig { requested, max_allowed } => Some(serde_json::json!(self)),
             _ => None,
         }
     }
@@ -206,7 +208,9 @@ impl From<StarknetRsError> for StarknetApiError {
             StarknetRsError::NoBlocks => Self::NoBlocks,
             StarknetRsError::NonAccount => Self::NonAccount,
             StarknetRsError::BlockNotFound => Self::BlockNotFound,
-            StarknetRsError::PageSizeTooBig => Self::PageSizeTooBig,
+            StarknetRsError::PageSizeTooBig => {
+                Self::PageSizeTooBig { requested: 0, max_allowed: 0 }
+            }
             StarknetRsError::DuplicateTx => Self::DuplicateTransaction,
             StarknetRsError::ContractNotFound => Self::ContractNotFound,
             StarknetRsError::CompilationFailed => Self::CompilationFailed,

crates/katana/rpc/rpc/src/starknet/mod.rs

@@ -10,6 +10,7 @@ use std::sync::Arc;
 use forking::ForkedClient;
 use katana_core::backend::Backend;
 use katana_core::service::block_producer::{BlockProducer, BlockProducerMode, PendingExecutor};
+use katana_executor::implementation::blockifier::blockifier::blockifier::config;
 use katana_executor::{ExecutionResult, ExecutorFactory};
 use katana_pool::validation::stateful::TxValidator;
 use katana_pool::{TransactionPool, TxPool};
@@ -58,6 +59,10 @@ pub struct StarknetApi<EF: ExecutorFactory> {
     inner: Arc<Inner<EF>>,
 }
 
+pub struct StarknetApiConfig {
+    pub max_event_page_size: Option<u64>,
+}
+
 impl<EF: ExecutorFactory> Clone for StarknetApi<EF> {
     fn clone(&self) -> Self {
         Self { inner: Arc::clone(&self.inner) }
@@ -71,6 +76,7 @@ struct Inner<EF: ExecutorFactory> {
     block_producer: BlockProducer<EF>,
     blocking_task_pool: BlockingTaskPool,
     forked_client: Option<ForkedClient>,
+    config: StarknetApiConfig,
 }
 
 impl<EF: ExecutorFactory> StarknetApi<EF> {
@@ -79,8 +85,9 @@ impl<EF: ExecutorFactory> StarknetApi<EF> {
         pool: TxPool,
         block_producer: BlockProducer<EF>,
         validator: TxValidator,
+        config: StarknetApiConfig,
     ) -> Self {
-        Self::new_inner(backend, pool, block_producer, validator, None)
+        Self::new_inner(backend, pool, block_producer, validator, None, config)
     }
 
     pub fn new_forked(
@@ -89,8 +96,9 @@ impl<EF: ExecutorFactory> StarknetApi<EF> {
         block_producer: BlockProducer<EF>,
         validator: TxValidator,
         forked_client: ForkedClient,
+        config: StarknetApiConfig,
     ) -> Self {
-        Self::new_inner(backend, pool, block_producer, validator, Some(forked_client))
+        Self::new_inner(backend, pool, block_producer, validator, Some(forked_client), config)
     }
 
     fn new_inner(
@@ -99,11 +107,12 @@ impl<EF: ExecutorFactory> StarknetApi<EF> {
         block_producer: BlockProducer<EF>,
         validator: TxValidator,
         forked_client: Option<ForkedClient>,
+        config: StarknetApiConfig,
     ) -> Self {
         let blocking_task_pool =
             BlockingTaskPool::new().expect("failed to create blocking task pool");
         let inner =
-            Inner { pool, backend, block_producer, blocking_task_pool, validator, forked_client };
+            Inner { pool, backend, block_producer, blocking_task_pool, validator, forked_client, max_chunk_size: Some(config) };
         Self { inner: Arc::new(inner) }
     }
 
@@ -816,6 +825,15 @@ impl<EF: ExecutorFactory> StarknetApi<EF> {
         let EventFilterWithPage { event_filter, result_page_request } = filter;
         let ResultPageRequest { continuation_token, chunk_size } = result_page_request;
 
+        if let Some(config) = self.inner.max_chunk_size.as_ref() {
+            if chunk_size > config.page_size {
+                return Err(StarknetApiError::PageSizeTooBig {
+                    requested: chunk_size,
+                    max_allowed: config.page_size,
+                });
+            }
+        }
+
         self.on_io_blocking_task(move |this| {
             let from = match event_filter.from_block {
                 Some(id) => id,