fix pipeline loop

Author: kariy

Cargo.lock

@@ -33,7 +33,7 @@ use katana_db::mdbx::DbEnv;
 use katana_executor::implementation::blockifier::BlockifierFactory;
 use katana_executor::{ExecutionFlags, ExecutorFactory};
 use katana_pipeline::stage::{Blocks, Classes};
-use katana_pipeline::{stage, Pipeline};
+use katana_pipeline::{stage, Pipeline, PipelineHandle};
 use katana_pool::ordering::FiFo;
 use katana_pool::validation::stateful::TxValidator;
 use katana_pool::TxPool;
@@ -62,6 +62,8 @@ pub struct LaunchedNode {
     pub node: Node,
     /// Handle to the rpc server.
     pub rpc: RpcServer,
+    /// Handle to the syncing pipeline.
+    pub pipeline: PipelineHandle,
 }
 
 impl LaunchedNode {
@@ -128,22 +130,11 @@ impl Node {
         let block_producer = self.block_producer.clone();
         let validator = self.block_producer.validator().clone();
 
-        // // --- build sequencing stage
-
-        // let sequencing = stage::Sequencing::new(
-        //     pool.clone(),
-        //     backend.clone(),
-        //     self.task_manager.task_spawner(),
-        //     block_producer.clone(),
-        //     self.messaging_config.clone(),
-        // );
-
         // --- build and start the pipeline
 
         let provider = self.backend.blockchain.provider().clone();
         let fgw = SequencerGatewayProvider::starknet_alpha_sepolia();
         let (mut pipeline, handle) = Pipeline::new(provider.clone(), 10);
-        // pipeline.add_stage(Box::new(sequencing));
 
         pipeline.add_stage(Blocks::new(provider.clone(), fgw.clone()));
         pipeline.add_stage(Classes::new(provider, fgw.clone()));
@@ -160,7 +151,7 @@ impl Node {
         let node_components = (pool, backend, block_producer, validator, self.forked_client.take());
         let rpc = spawn(node_components, self.rpc_config.clone()).await?;
 
-        Ok(LaunchedNode { node: self, rpc })
+        Ok(LaunchedNode { node: self, pipeline: handle, rpc })
     }
 }
 

crates/katana/node/src/lib.rs

@@ -18,7 +18,6 @@ anyhow.workspace = true
 async-trait.workspace = true
 backon = { version = "1.3.0", features = [ "tokio-sleep" ] }
 futures.workspace = true
-parking_lot = "0.12.3"
 serde_json = "1.0.133"
 starknet.workspace = true
 thiserror.workspace = true

crates/katana/pipeline/Cargo.toml

@@ -3,13 +3,11 @@
 pub mod stage;
 
 use core::future::IntoFuture;
-use std::sync::Arc;
 
 use futures::future::BoxFuture;
 use katana_primitives::block::BlockNumber;
 use katana_provider::error::ProviderError;
 use katana_provider::traits::stage::StageCheckpointProvider;
-use parking_lot::{Condvar, Mutex};
 use stage::{Stage, StageExecutionInput};
 use tokio::sync::watch;
 use tracing::{error, info};
@@ -32,12 +30,6 @@ pub enum Error {
     Provider(#[from] ProviderError),
 }
 
-#[derive(Debug)]
-pub enum PipelineEvents {
-    UpdateTip(BlockNumber),
-    Stop,
-}
-
 #[derive(Debug)]
 pub struct PipelineHandle {
     tx: watch::Sender<Option<BlockNumber>>,
@@ -95,15 +87,15 @@ impl<P: StageCheckpointProvider> Pipeline<P> {
             let tip = *self.tip.borrow_and_update();
 
             loop {
-                if let Some(to) = tip {
-                    let to = current_chunk_tip.min(to);
+                if let Some(tip) = tip {
+                    let to = current_chunk_tip.min(tip);
                     self.run_once_until(to).await?;
 
-                    if current_chunk_tip >= to {
+                    if to >= tip {
+                        info!(target: "pipeline", %tip, "Finished processing until tip.");
                         break;
                     } else {
-                        current_chunk_tip = (current_chunk_tip + self.chunk_size).min(to);
-                        continue;
+                        current_chunk_tip = (current_chunk_tip + self.chunk_size).min(tip);
                     }
                 }
             }