node integration

Author: kariy

Cargo.lock

@@ -6,7 +6,6 @@ use katana_gas_price_oracle::GasPriceOracle;
 use katana_primitives::transaction::{ExecutableTx, InvokeTx};
 use katana_primitives::Felt;
 use katana_provider::providers::db::DbProvider;
-use katana_tasks::TaskManager;
 
 use super::*;
 use crate::backend::storage::Blockchain;

crates/core/src/service/block_producer_tests.rs

@@ -7,8 +7,6 @@ version.workspace = true
 
 [dependencies]
 katana-gateway-types.workspace = true
-katana-executor.workspace = true
-katana-core.workspace = true
 katana-primitives.workspace = true
 katana-metrics.workspace = true
 katana-pool.workspace = true

crates/gateway/gateway-server/Cargo.toml

@@ -3,12 +3,14 @@
 use std::future::IntoFuture;
 use std::sync::Arc;
 
+use alloy_provider::RootProvider;
 use anyhow::Result;
 use http::header::CONTENT_TYPE;
 use http::Method;
 use jsonrpsee::RpcModule;
 use katana_chain_spec::ChainSpec;
 use katana_executor::ExecutionFlags;
+use katana_gas_price_oracle::GasPriceOracle;
 use katana_gateway_client::Client as SequencerGateway;
 use katana_metrics::exporters::prometheus::PrometheusRecorder;
 use katana_metrics::{Report, Server as MetricsServer};
@@ -27,6 +29,7 @@ use tracing::{error, info};
 
 use crate::config::db::DbConfig;
 use crate::config::metrics::MetricsConfig;
+use crate::full::pending::PreconfStateFactory;
 
 mod exit;
 mod pending;
@@ -75,6 +78,7 @@ pub struct Node {
     pub pipeline: Pipeline<DbProvider>,
     pub rpc_server: RpcServer,
     pub gateway_client: SequencerGateway,
+    pub chain_tip_watcher: ChainTipWatcher<RootProvider>,
 }
 
 impl Node {
@@ -114,12 +118,32 @@ impl Node {
 
         // --- build pipeline
 
-        let (mut pipeline, _) = Pipeline::new(provider.clone(), 50);
+        let (mut pipeline, pipeline_handle) = Pipeline::new(provider.clone(), 50);
         let block_downloader = BatchBlockDownloader::new_gateway(gateway_client.clone(), 8);
         pipeline.add_stage(Blocks::new(provider.clone(), block_downloader));
         pipeline.add_stage(Classes::new(provider.clone(), gateway_client.clone(), 8));
         pipeline.add_stage(StateTrie::new(provider.clone()));
 
+        // --
+
+        let core_contract = match config.network {
+            Network::Mainnet => {
+                katana_starknet::StarknetCore::new_http_mainnet(&config.eth_rpc_url)?
+            }
+            Network::Sepolia => {
+                katana_starknet::StarknetCore::new_http_sepolia(&config.eth_rpc_url)?
+            }
+        };
+
+        let chain_tip_watcher = ChainTipWatcher::new(core_contract);
+
+        let preconf_factory = PreconfStateFactory::new(
+            provider.clone(),
+            gateway_client.clone(),
+            pipeline_handle.subscribe_blocks(),
+            chain_tip_watcher.subscribe(),
+        );
+
         // --- build rpc server
 
         let mut rpc_modules = RpcModule::new(());
@@ -148,6 +172,8 @@ impl Node {
             pool.clone(),
             task_spawner.clone(),
             starknet_api_cfg,
+            Box::new(preconf_factory),
+            GasPriceOracle::create_for_testing(),
         );
 
         if config.rpc.apis.contains(&RpcModuleKind::Starknet) {
@@ -194,6 +220,7 @@ impl Node {
             rpc_server,
             task_manager,
             gateway_client,
+            chain_tip_watcher,
             config: Arc::new(config),
         })
     }
@@ -212,18 +239,7 @@ impl Node {
 
         let pipeline_handle = self.pipeline.handle();
 
-        let core_contract = match self.config.network {
-            Network::Mainnet => {
-                katana_starknet::StarknetCore::new_http_mainnet(&self.config.eth_rpc_url).await?
-            }
-            Network::Sepolia => {
-                katana_starknet::StarknetCore::new_http_sepolia(&self.config.eth_rpc_url).await?
-            }
-        };
-
-        let tip_watcher = ChainTipWatcher::new(core_contract);
-
-        let mut tip_subscription = tip_watcher.subscribe();
+        let mut tip_subscription = self.chain_tip_watcher.subscribe();
         let pipeline_handle_clone = pipeline_handle.clone();
 
         self.task_manager
@@ -237,7 +253,7 @@ impl Node {
             .build_task()
             .graceful_shutdown()
             .name("Chain tip watcher")
-            .spawn(tip_watcher.into_future());
+            .spawn(self.chain_tip_watcher.into_future());
 
         // spawn a task for updating the pipeline's tip based on chain tip changes
         self.task_manager.task_spawner().spawn(async move {

crates/node/src/full/mod.rs

@@ -3,11 +3,11 @@ use std::time::Duration;
 
 use katana_gateway::client::Client;
 use katana_gateway::types::{ConfirmedTransaction, ErrorCode, PreConfirmedBlock, StateDiff};
+use katana_pipeline::PipelineBlockSubscription;
 use katana_primitives::block::BlockNumber;
 use katana_primitives::state::StateUpdates;
 use katana_provider::api::state::StateFactoryProvider;
 use parking_lot::Mutex;
-use tokio::sync::watch;
 use tracing::error;
 
 use crate::full::pending::state::PreconfStateProvider;
@@ -19,7 +19,7 @@ pub mod state;
 #[derive(Debug)]
 pub struct PreconfStateFactory<P: StateFactoryProvider> {
     // from pipeline
-    latest_synced_block: watch::Receiver<BlockNumber>,
+    latest_synced_block: PipelineBlockSubscription,
     gateway_client: Client,
     provider: P,
 
@@ -31,7 +31,7 @@ impl<P: StateFactoryProvider> PreconfStateFactory<P> {
     pub fn new(
         state_factory_provider: P,
         gateway_client: Client,
-        latest_synced_block: watch::Receiver<BlockNumber>,
+        latest_synced_block: PipelineBlockSubscription,
         tip_subscription: TipSubscription,
     ) -> Self {
         let shared_preconf_block = SharedPreconfBlockData::default();
@@ -55,7 +55,7 @@ impl<P: StateFactoryProvider> PreconfStateFactory<P> {
     }
 
     pub fn state(&self) -> PreconfStateProvider {
-        let latest_block_num = *self.latest_synced_block.borrow();
+        let latest_block_num = self.latest_synced_block.block().unwrap();
         let base = self.provider.historical(latest_block_num.into()).unwrap().unwrap();
 
         let preconf_block = self.shared_preconf_block.inner.lock();
@@ -114,7 +114,7 @@ struct PreconfBlockWatcher {
     gateway_client: Client,
 
     // from pipeline
-    latest_synced_block: watch::Receiver<BlockNumber>,
+    latest_synced_block: PipelineBlockSubscription,
     // from tip watcher (actual tip of the chain)
     latest_block: TipSubscription,
 
@@ -124,7 +124,8 @@ struct PreconfBlockWatcher {
 
 impl PreconfBlockWatcher {
     async fn run(&mut self) {
-        let mut current_preconf_block_num = *self.latest_synced_block.borrow() + 1;
+        let mut current_preconf_block_num =
+            self.latest_synced_block.block().map(|b| b + 1).unwrap_or(0);
 
         loop {
             if current_preconf_block_num >= self.latest_block.tip() {
@@ -180,7 +181,7 @@ impl PreconfBlockWatcher {
                         break;
                     }
 
-                    let latest_synced_block_num = *self.latest_synced_block.borrow();
+                    let latest_synced_block_num = self.latest_synced_block.block().unwrap();
                     current_preconf_block_num = latest_synced_block_num + 1;
                 }
 

crates/node/src/full/pending/mod.rs

@@ -4,11 +4,11 @@ use std::fmt::Debug;
 use std::future::Future;
 use std::sync::Arc;
 
-use katana_core::backend::Backend;
-use katana_executor::ExecutorFactory;
 use katana_chain_spec::ChainSpec;
 use katana_core::backend::storage::Database;
+use katana_core::backend::Backend;
 use katana_core::utils::get_current_timestamp;
+use katana_executor::ExecutorFactory;
 use katana_gas_price_oracle::GasPriceOracle;
 use katana_pool::TransactionPool;
 use katana_primitives::block::{BlockHashOrNumber, BlockIdOrTag, FinalityStatus, GasPrices};
@@ -53,7 +53,6 @@ use katana_rpc_types_builder::{BlockBuilder, ReceiptBuilder};
 use katana_tasks::{Result as TaskResult, TaskSpawner};
 
 use crate::permit::Permits;
-use crate::starknet::pending::PendingBlockProvider;
 use crate::utils::events::{Cursor, EventBlockId};
 use crate::{utils, DEFAULT_ESTIMATE_FEE_MAX_CONCURRENT_REQUESTS};
 
@@ -97,9 +96,8 @@ where
 {
     pool: Pool,
     chain_spec: Arc<ChainSpec>,
-    pool: P,
+    pool: Pool,
     gas_oracle: GasPriceOracle,
-    preconf_provider: Box<dyn PendingBlockProvider>,
     storage: BlockchainProvider<Box<dyn Database>>,
     forked_client: Option<ForkedClient>,
     task_spawner: TaskSpawner,
@@ -154,7 +152,7 @@ where
             pool,
             None,
             task_spawner,
-            config, 
+            config,
             pending_block_provider,
             gas_oracle,
         )

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

@@ -14,7 +14,8 @@
 
 use alloy_network::Ethereum;
 use alloy_primitives::Address;
-use alloy_provider::{Provider, RootProvider};
+use alloy_provider::Provider;
+pub use alloy_provider::RootProvider;
 use alloy_rpc_types_eth::{Filter, FilterBlockOption, FilterSet, Log, Topic};
 use alloy_sol_types::{sol, SolEvent};
 use anyhow::Result;
@@ -225,7 +226,7 @@ impl StarknetCore<RootProvider<Ethereum>> {
     ///
     /// * `rpc_url` - The HTTP URL of the Ethereum RPC endpoint
     /// * `contract_address` - The address of the Starknet Core Contract
-    pub async fn new_http(rpc_url: impl AsRef<str>, contract_address: Address) -> Result<Self> {
+    pub fn new_http(rpc_url: impl AsRef<str>, contract_address: Address) -> Result<Self> {
         let provider = RootProvider::<Ethereum>::new_http(reqwest::Url::parse(rpc_url.as_ref())?);
         Ok(Self::new(provider, contract_address))
     }
@@ -236,7 +237,7 @@ impl StarknetCore<RootProvider<Ethereum>> {
     /// # Arguments
     ///
     /// * `rpc_url` - The HTTP URL of the Ethereum RPC endpoint
-    pub async fn new_http_mainnet(rpc_url: impl AsRef<str>) -> Result<Self> {
+    pub fn new_http_mainnet(rpc_url: impl AsRef<str>) -> Result<Self> {
         let provider = RootProvider::<Ethereum>::new_http(reqwest::Url::parse(rpc_url.as_ref())?);
         Ok(Self::new_mainnet(provider))
     }
@@ -247,7 +248,7 @@ impl StarknetCore<RootProvider<Ethereum>> {
     /// # Arguments
     ///
     /// * `rpc_url` - The HTTP URL of the Ethereum RPC endpoint
-    pub async fn new_http_sepolia(rpc_url: impl AsRef<str>) -> Result<Self> {
+    pub fn new_http_sepolia(rpc_url: impl AsRef<str>) -> Result<Self> {
         let provider = RootProvider::<Ethereum>::new_http(reqwest::Url::parse(rpc_url.as_ref())?);
         Ok(Self::new_sepolia(provider))
     }

crates/starknet/src/lib.rs

@@ -117,6 +117,7 @@ enum PipelineCommand {
 /// This subscription receives notifications whenever the pipeline completes processing
 /// a block through all stages. The block number represents the highest block that has
 /// been successfully processed by all pipeline stages for a given batch.
+#[derive(Clone)]
 pub struct PipelineBlockSubscription {
     rx: watch::Receiver<Option<BlockNumber>>,
 }