wip

Author: kariy

crates/node/src/full/mod.rs

@@ -109,7 +109,7 @@ impl Node {
         let block_downloader = BatchBlockDownloader::new_gateway(gateway_client.clone(), 3);
         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
 
@@ -190,8 +190,6 @@ impl Node {
     }
 
     pub async fn launch(self) -> Result<LaunchedNode> {
-        println!("Launching node");
-
         if let Some(ref cfg) = self.config.metrics {
             let reports: Vec<Box<dyn Report>> = vec![Box::new(self.db.clone()) as Box<dyn Report>];
             let exporter = PrometheusRecorder::current().expect("qed; should exist at this point");

crates/sync/pipeline/src/lib.rs

@@ -73,7 +73,7 @@ use katana_provider_api::stage::StageCheckpointProvider;
 use katana_provider_api::ProviderError;
 use katana_stage::{Stage, StageExecutionInput, StageExecutionOutput};
 use tokio::sync::watch;
-use tracing::{debug, error, info, info_span, trace, Instrument};
+use tracing::{debug, error, info, info_span, trace, Instrument, Span};
 
 /// The result of a pipeline execution.
 pub type PipelineResult<T> = Result<T, Error>;
@@ -254,13 +254,6 @@ impl<P: StageCheckpointProvider> Pipeline<P> {
                     continue;
                 }
             }
-
-            debug!(target: "pipeline", "Waiting for new command.");
-
-            // Wait for the next command
-            if self.command_rx.changed().await.is_err() {
-                break;
-            }
         }
 
         info!(target: "pipeline", "Pipeline finished.");
@@ -317,27 +310,29 @@ impl<P: StageCheckpointProvider> Pipeline<P> {
             let from = if checkpoint == 0 {
                 checkpoint
             } else {
-                // plus 1 because the checkpoint is inclusive
+                // plus 1 because the checkpoint is the last block processed, so we need to start from the next block
                 checkpoint + 1
             };
 
             let input = StageExecutionInput::new(from, to);
-            let span = info_span!(target: "pipeline", "execute", stage = %id, %from, %to);
+            let span = info_span!(target: "pipeline", "stage.execute", stage = %id, %from, %to);
+            let _guard = span.enter();
 
-            info!(target: "pipeline", %id, %from, %to, "Executing stage.");
+            info!(target: "pipeline", stage = %id, %from, %to, "[{}/{}] Executing stage.", i + 1, last_stage_idx + 1);
 
             let StageExecutionOutput { last_block_processed } =
                 stage
                 .execute(&input)
-                .instrument(span)
+                .instrument(Span::current())
                 .await
                 .map_err(|error| Error::StageExecution { id, error })?;
 
+            info!(target: "pipeline", stage = %id, %from, %to, "Stage execution completed.");
+
 
             self.provider.set_checkpoint(id, last_block_processed)?;
             last_block_processed_list.push(last_block_processed);
-
-            info!(target: "pipeline", %id, %from, %to, "Stage execution completed.");
+            info!(target: "pipeline", stage = %id, checkpoint = %to, "New checkpoint set.");
         }
 
         Ok(last_block_processed_list.into_iter().min().unwrap_or(to))

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

@@ -14,7 +14,7 @@ use katana_primitives::Felt;
 use katana_provider::api::block::BlockWriter;
 use num_traits::ToPrimitive;
 use starknet::core::types::ResourcePrice;
-use tracing::debug;
+use tracing::{debug, error, trace};
 
 use crate::{Stage, StageExecutionInput, StageExecutionOutput, StageResult};
 
@@ -51,21 +51,27 @@ where
                 .downloader
                 .download_blocks(input.from(), input.to())
                 .await
-                .map_err(Error::Gateway)?;
+                .map_err(Error::Gateway)
+                .inspect_err(|e| error!(error = %e , "Error downloading blocks."))?;
 
             if !blocks.is_empty() {
                 debug!(target: "stage", id = %self.id(), total = %blocks.len(), "Storing blocks to storage.");
 
                 // Store blocks to storage
                 for block in blocks {
                     let (block, receipts, state_updates) = extract_block_data(block)?;
-
-                    self.provider.insert_block_with_states_and_receipts(
-                        block,
-                        state_updates,
-                        receipts,
-                        Vec::new(),
-                    )?;
+                    let block_number = block.block.header.number;
+
+                    self.provider
+                        .insert_block_with_states_and_receipts(
+                            block,
+                            state_updates,
+                            receipts,
+                            Vec::new(),
+                        )
+                        .inspect_err(
+                            |e| error!(error = %e, block = %block_number, "Error storing block."),
+                        )?;
                 }
             }