feat(cli): reth prune (#9055)

crates/stages/api/src/pipeline/mod.rs

@@ -13,7 +13,6 @@ use reth_provider::{
 };
 use reth_prune::PrunerBuilder;
 use reth_static_file::StaticFileProducer;
-use reth_static_file_types::HighestStaticFiles;
 use reth_tokio_util::{EventSender, EventStream};
 use std::pin::Pin;
 use tokio::sync::watch;
@@ -248,26 +247,9 @@ where
     /// CAUTION: This method locks the static file producer Mutex, hence can block the thread if the
     /// lock is occupied.
     pub fn move_to_static_files(&self) -> RethResult<()> {
-        let static_file_producer = self.static_file_producer.lock();
-
         // Copies data from database to static files
-        let lowest_static_file_height = {
-            let provider = self.provider_factory.provider()?;
-            let stages_checkpoints = [StageId::Headers, StageId::Execution, StageId::Bodies]
-                .into_iter()
-                .map(|stage| {
-                    provider.get_stage_checkpoint(stage).map(|c| c.map(|c| c.block_number))
-                })
-                .collect::<Result<Vec<_>, _>>()?;
-
-            let targets = static_file_producer.get_static_file_targets(HighestStaticFiles {
-                headers: stages_checkpoints[0],
-                receipts: stages_checkpoints[1],
-                transactions: stages_checkpoints[2],
-            })?;
-            static_file_producer.run(targets)?;
-            stages_checkpoints.into_iter().min().expect("exists")
-        };
+        let lowest_static_file_height =
+            self.static_file_producer.lock().copy_to_static_files()?.min();
 
         // Deletes data which has been copied to static files.
         if let Some(prune_tip) = lowest_static_file_height {

crates/static-file/static-file/Cargo.toml

@@ -21,6 +21,7 @@ reth-nippy-jar.workspace = true
 reth-tokio-util.workspace = true
 reth-prune-types.workspace = true
 reth-static-file-types.workspace = true
+reth-stages-types.workspace = true
 
 alloy-primitives.workspace = true
 

crates/static-file/static-file/src/static_file_producer.rs

@@ -5,8 +5,12 @@ use alloy_primitives::BlockNumber;
 use parking_lot::Mutex;
 use rayon::prelude::*;
 use reth_db_api::database::Database;
-use reth_provider::{providers::StaticFileWriter, ProviderFactory, StaticFileProviderFactory};
+use reth_provider::{
+    providers::StaticFileWriter, ProviderFactory, StageCheckpointReader as _,
+    StaticFileProviderFactory,
+};
 use reth_prune_types::PruneModes;
+use reth_stages_types::StageId;
 use reth_static_file_types::HighestStaticFiles;
 use reth_storage_errors::provider::ProviderResult;
 use reth_tokio_util::{EventSender, EventStream};
@@ -56,7 +60,7 @@ pub struct StaticFileProducerInner<DB> {
     event_sender: EventSender<StaticFileProducerEvent>,
 }
 
-/// Static File targets, per data part, measured in [`BlockNumber`].
+/// Static File targets, per data segment, measured in [`BlockNumber`].
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct StaticFileTargets {
     headers: Option<RangeInclusive<BlockNumber>>,
@@ -167,6 +171,28 @@ impl<DB: Database> StaticFileProducerInner<DB> {
         Ok(targets)
     }
 
+    /// Copies data from database to static files according to
+    /// [stage checkpoints](reth_stages_types::StageCheckpoint).
+    ///
+    /// Returns highest block numbers for all static file segments.
+    pub fn copy_to_static_files(&self) -> ProviderResult<HighestStaticFiles> {
+        let provider = self.provider_factory.provider()?;
+        let stages_checkpoints = [StageId::Headers, StageId::Execution, StageId::Bodies]
+            .into_iter()
+            .map(|stage| provider.get_stage_checkpoint(stage).map(|c| c.map(|c| c.block_number)))
+            .collect::<Result<Vec<_>, _>>()?;
+
+        let highest_static_files = HighestStaticFiles {
+            headers: stages_checkpoints[0],
+            receipts: stages_checkpoints[1],
+            transactions: stages_checkpoints[2],
+        };
+        let targets = self.get_static_file_targets(highest_static_files)?;
+        self.run(targets)?;
+
+        Ok(highest_static_files)
+    }
+
     /// Returns a static file targets at the provided finalized block numbers per segment.
     /// The target is determined by the check against highest `static_files` using
     /// [`reth_provider::providers::StaticFileProvider::get_highest_static_files`].

crates/static-file/types/src/lib.rs

@@ -20,7 +20,7 @@ pub use segment::{SegmentConfig, SegmentHeader, SegmentRangeInclusive, StaticFil
 /// Default static file block count.
 pub const BLOCKS_PER_STATIC_FILE: u64 = 500_000;
 
-/// Highest static file block numbers, per data part.
+/// Highest static file block numbers, per data segment.
 #[derive(Debug, Clone, Copy, Default, Eq, PartialEq)]
 pub struct HighestStaticFiles {
     /// Highest static file block of headers, inclusive.
@@ -53,6 +53,11 @@ impl HighestStaticFiles {
         }
     }
 
+    /// Returns the minimum block of all segments.
+    pub fn min(&self) -> Option<u64> {
+        [self.headers, self.transactions, self.receipts].iter().filter_map(|&option| option).min()
+    }
+
     /// Returns the maximum block of all segments.
     pub fn max(&self) -> Option<u64> {
         [self.headers, self.transactions, self.receipts].iter().filter_map(|&option| option).max()