wip

Author: kariy

crates/node/src/full/mod.rs

@@ -116,11 +116,11 @@ impl Node {
 
         // --- build pipeline
 
-        let (mut pipeline, _) = Pipeline::new(provider.clone(), 100);
+        let (mut pipeline, _) = Pipeline::new(provider.clone(), 10);
         let block_downloader = BatchBlockDownloader::new_gateway(gateway_client.clone(), 10);
         pipeline.add_stage(Blocks::new(provider.clone(), block_downloader));
         pipeline.add_stage(Classes::new(provider.clone(), gateway_client.clone(), 3));
-        // pipeline.add_stage(StateTrie::new(provider.clone()));
+        pipeline.add_stage(StateTrie::new(provider.clone()));
 
         // --- build rpc server
 

crates/sync/pipeline/src/lib.rs

@@ -82,10 +82,10 @@ pub type PipelineFut = BoxFuture<'static, PipelineResult<()>>;
 
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
-    #[error("Stage not found: {id}")]
+    #[error("stage not found: {id}")]
     StageNotFound { id: String },
 
-    #[error("Stage '{id}' execution failed: {error}")]
+    #[error("stage {id} execution failed: {error}")]
     StageExecution { id: &'static str, error: katana_stage::Error },
 
     #[error(transparent)]
@@ -135,7 +135,7 @@ impl PipelineHandle {
     /// Panics if the [`Pipeline`] has been dropped.
     pub fn stop(&self) {
         info!(target: "pipeline", "Signaling pipeline to stop");
-        self.tx.send(Some(PipelineCommand::Stop)).expect("channel closed");
+        let _ = self.tx.send(Some(PipelineCommand::Stop));
     }
 
     /// Wait until the [`Pipeline`] has stopped.
@@ -254,12 +254,17 @@ impl<P: StageCheckpointProvider> Pipeline<P> {
                     }
                 }
 
-                _ = self.run_loop() => { }
+                result = self.run_loop() => {
+                    if let Err(error) = result {
+                        error!(target: "pipeline", %error, "Pipeline finished due to error.");
+                        break;
+                    }
+                }
 
             }
         }
 
-        info!(target: "pipeline", "Pipeline finished.");
+        info!(target: "pipeline", "Pipeline shutting down.");
 
         Ok(())
     }

crates/sync/stage/src/blocks/mod.rs

@@ -104,12 +104,12 @@ where
             let blocks = self
                 .downloader
                 .download_blocks(input.from(), input.to())
-                .instrument(info_span!(target: "stage", "blocks.download", to = %input.to(), from = %input.from()))
+                .instrument(info_span!(target: "stage", "blocks.download", from = %input.from(), to = %input.to()))
                 .await
                 .map_err(Error::Gateway)?;
 
             if !blocks.is_empty() {
-                let span = info_span!(target: "stage", "blocks.insert", to = %input.to(), from = %input.from());
+                let span = info_span!(target: "stage", "blocks.insert", from = %input.from(), to = %input.to());
                 let _enter = span.enter();
 
                 // Validate chain invariant before storing

crates/sync/stage/src/classes.rs

@@ -131,7 +131,7 @@ where
             let declared_class_hashes = self.get_declared_classes(input.from(), input.to())?;
 
             if declared_class_hashes.is_empty() {
-                debug!("No classes declared within the block range", from = %input.from(), to = %input.to());
+                debug!(from = %input.from(), to = %input.to(), "No classes declared within the block range");
             } else {
                 let total_classes = declared_classes.len();
 

crates/sync/stage/src/trie.rs

@@ -4,9 +4,10 @@ use katana_primitives::Felt;
 use katana_provider::api::block::HeaderProvider;
 use katana_provider::api::state_update::StateUpdateProvider;
 use katana_provider::api::trie::TrieWriter;
+use katana_rpc_types::class;
 use starknet::macros::short_string;
 use starknet_types_core::hash::{Poseidon, StarkHash};
-use tracing::{error, info_span, trace, trace_span};
+use tracing::{debug, error, info_span, trace, trace_span};
 
 use crate::{Stage, StageExecutionInput, StageExecutionOutput, StageResult};
 
@@ -41,7 +42,7 @@ where
     fn execute<'a>(&'a mut self, input: &'a StageExecutionInput) -> BoxFuture<'a, StageResult> {
         Box::pin(async move {
             for block_number in input.from()..=input.to() {
-                let span = info_span!("compute_state_root", %block_number);
+                let span = info_span!("state_trie.compute_state_root", %block_number);
                 let _enter = span.enter();
 
                 let expected_state_root = self
@@ -55,20 +56,27 @@ where
                     .state_update(block_number.into())?
                     .ok_or(Error::MissingStateUpdate(block_number))?;
 
-                let class_trie_root = self
-                    .provider
-                    .trie_insert_declared_classes(block_number, &state_update.declared_classes)?;
-
                 let contract_trie_root =
                     self.provider.trie_insert_contract_updates(block_number, &state_update)?;
 
-                // Compute the state root:
-                // hash("STARKNET_STATE_V0", contract_trie_root, class_trie_root)
-                let computed_state_root = Poseidon::hash_array(&[
-                    short_string!("STARKNET_STATE_V0"),
-                    contract_trie_root,
-                    class_trie_root,
-                ]);
+                let computed_state_root = if dbg!(!state_update.declared_classes.is_empty()) {
+                    let class_trie_root = self.provider.trie_insert_declared_classes(
+                        block_number,
+                        dbg!(&state_update.declared_classes),
+                    )?;
+
+                    // Compute the state root:
+                    // hash("STARKNET_STATE_V0", contract_trie_root, class_trie_root)
+                    let computed_state_root = Poseidon::hash_array(&[
+                        short_string!("STARKNET_STATE_V0"),
+                        dbg!(contract_trie_root),
+                        dbg!(class_trie_root),
+                    ]);
+
+                    computed_state_root
+                } else {
+                    contract_trie_root
+                };
 
                 // Verify that the computed state root matches the expected state root from the
                 // block header
@@ -88,11 +96,7 @@ where
                     .into());
                 }
 
-                trace!(
-                    block = %block_number,
-                    state_root = %format!("{computed_state_root:#x}"),
-                    "State root verified successfully."
-                );
+                debug!(block = %block_number, "State root verified successfully.");
             }
 
             Ok(StageExecutionOutput { last_block_processed: input.to() })