wip

Author: kariy

.tool-versions

@@ -1 +1,2 @@
 scarb 2.8.2
+katana 1.7.0

crates/sync/pipeline/src/lib.rs

@@ -256,6 +256,7 @@ pub struct Pipeline<P> {
     block_tx: watch::Sender<Option<BlockNumber>>,
     tip: Option<BlockNumber>,
     pruning_config: PruningConfig,
+    /// The block at which the pipeline was last pruned.
     last_pruned_block: Option<BlockNumber>,
 }
 
@@ -456,15 +457,15 @@ impl<P: StageCheckpointProvider> Pipeline<P> {
     }
 
     /// Runs pruning on all stages.
-    pub async fn prune(&mut self, tip: BlockNumber) -> PipelineResult<()> {
+    pub async fn prune(&mut self) -> PipelineResult<()> {
         if self.stages.is_empty() {
             return Ok(());
         }
 
         for stage in self.stages.iter_mut() {
             let id = stage.id();
 
-            let span = info_span!(target: "pipeline", "stage.prune", stage = %id, %tip);
+            let span = info_span!(target: "pipeline", "stage.prune", stage = %id);
             let enter = span.entered();
 
             if let Some(checkpoint) = self.provider.checkpoint(id)? {
@@ -504,7 +505,7 @@ impl<P: StageCheckpointProvider> Pipeline<P> {
                 // Check if we should run pruning
                 if self.should_prune(last_block_processed) {
                     info!(target: "pipeline", block = %last_block_processed, "Starting pruning.");
-                    self.prune(last_block_processed).await?;
+                    self.prune().await?;
                     self.last_pruned_block = Some(last_block_processed);
                 }
 

crates/sync/pipeline/tests/pipeline.rs

@@ -75,6 +75,11 @@ impl Stage for TrackingStage {
             Ok(StageExecutionOutput { last_block_processed: input.to() })
         })
     }
+
+    fn prune<'a>(&'a mut self, input: &'a PruneInput) -> BoxFuture<'a, PruneResult> {
+        let _ = input;
+        Box::pin(async move { Ok(PruneOutput::default()) })
+    }
 }
 
 /// Mock stage that fails on execution
@@ -97,6 +102,11 @@ impl Stage for FailingStage {
     fn execute<'a>(&'a mut self, _: &'a StageExecutionInput) -> BoxFuture<'a, StageResult> {
         Box::pin(async { Err(katana_stage::Error::Other(anyhow!("Stage execution failed"))) })
     }
+
+    fn prune<'a>(&'a mut self, input: &'a PruneInput) -> BoxFuture<'a, PruneResult> {
+        let _ = input;
+        Box::pin(async move { Ok(PruneOutput::default()) })
+    }
 }
 
 /// Mock stage that always reports a fixed `last_block_processed`.
@@ -138,6 +148,11 @@ impl Stage for FixedOutputStage {
             Ok(StageExecutionOutput { last_block_processed })
         })
     }
+
+    fn prune<'a>(&'a mut self, input: &'a PruneInput) -> BoxFuture<'a, PruneResult> {
+        let _ = input;
+        Box::pin(async move { Ok(PruneOutput::default()) })
+    }
 }
 
 // ============================================================================
@@ -497,6 +512,11 @@ async fn run_should_be_cancelled_if_stop_requested() {
                 Ok(StageExecutionOutput { last_block_processed: 100 })
             })
         }
+
+        fn prune<'a>(&'a mut self, input: &'a PruneInput) -> BoxFuture<'a, PruneResult> {
+            let _ = input;
+            Box::pin(async move { Ok(PruneOutput::default()) })
+        }
     }
 
     let provider = Arc::new(test_provider());

crates/sync/stage/tests/trie.rs

@@ -53,12 +53,6 @@ impl MockProvider {
         self
     }
 
-    /// Configure the mock to fail on trie operations.
-    fn with_trie_error(self) -> Self {
-        *self.should_fail.lock().unwrap() = true;
-        self
-    }
-
     /// Get all block numbers that had trie inserts called.
     fn trie_insert_blocks(&self) -> Vec<BlockNumber> {
         self.trie_insert_calls.lock().unwrap().clone()