feat: Snapshotter triggers segment snapshots

bin/reth/src/db/snapshots/headers.rs

@@ -12,8 +12,11 @@ use reth_primitives::{
 use reth_provider::{
     providers::SnapshotProvider, DatabaseProviderRO, HeaderProvider, ProviderError, ProviderFactory,
 };
-use reth_snapshot::segments::{Headers, Segment};
-use std::{path::Path, sync::Arc};
+use reth_snapshot::{segments, segments::Segment};
+use std::{
+    path::{Path, PathBuf},
+    sync::Arc,
+};
 
 impl Command {
     pub(crate) fn generate_headers_snapshot<DB: Database>(
@@ -23,15 +26,24 @@ impl Command {
         inclusion_filter: InclusionFilter,
         phf: PerfectHashingFunction,
     ) -> eyre::Result<()> {
-        let segment = Headers::new(
-            compression,
-            if self.with_filters {
-                Filters::WithFilters(inclusion_filter, phf)
-            } else {
-                Filters::WithoutFilters
-            },
-        );
-        segment.snapshot::<DB>(provider, self.from..=(self.from + self.block_interval - 1))?;
+        let range = self.block_range();
+        let filters = if self.with_filters {
+            Filters::WithFilters(inclusion_filter, phf)
+        } else {
+            Filters::WithoutFilters
+        };
+
+        let segment = segments::Headers::new(compression, filters);
+
+        segment.snapshot::<DB>(provider, PathBuf::default(), range.clone())?;
+
+        // Default name doesn't have any configuration
+        let default_name: PathBuf = SnapshotSegment::Headers.filename(&range).into();
+        let new_name: PathBuf = SnapshotSegment::Headers
+            .filename_with_configuration(filters, compression, &range)
+            .into();
+
+        std::fs::rename(default_name, new_name)?;
 
         Ok(())
     }
@@ -51,12 +63,13 @@ impl Command {
             Filters::WithoutFilters
         };
 
-        let range = self.from..=(self.from + self.block_interval - 1);
+        let range = self.block_range();
 
         let mut row_indexes = range.clone().collect::<Vec<_>>();
         let mut rng = rand::thread_rng();
-        let path =
-            SnapshotSegment::Headers.filename_with_configuration(filters, compression, &range);
+        let path = SnapshotSegment::Headers
+            .filename_with_configuration(filters, compression, &range)
+            .into();
         let provider = SnapshotProvider::default();
         let jar_provider =
             provider.get_segment_provider(SnapshotSegment::Headers, self.from, Some(path))?;

bin/reth/src/db/snapshots/mod.rs

@@ -7,7 +7,7 @@ use reth_primitives::{
     BlockNumber, ChainSpec, SnapshotSegment,
 };
 use reth_provider::ProviderFactory;
-use std::{path::Path, sync::Arc};
+use std::{ops::RangeInclusive, path::Path, sync::Arc};
 
 mod bench;
 mod headers;
@@ -130,4 +130,9 @@ impl Command {
 
         Ok(())
     }
+
+    /// Gives out the inclusive block range for the snapshot requested by the user.
+    fn block_range(&self) -> RangeInclusive<BlockNumber> {
+        self.from..=(self.from + self.block_interval - 1)
+    }
 }

bin/reth/src/db/snapshots/receipts.rs

@@ -14,7 +14,10 @@ use reth_provider::{
     ReceiptProvider, TransactionsProvider, TransactionsProviderExt,
 };
 use reth_snapshot::{segments, segments::Segment};
-use std::{path::Path, sync::Arc};
+use std::{
+    path::{Path, PathBuf},
+    sync::Arc,
+};
 
 impl Command {
     pub(crate) fn generate_receipts_snapshot<DB: Database>(
@@ -24,15 +27,24 @@ impl Command {
         inclusion_filter: InclusionFilter,
         phf: PerfectHashingFunction,
     ) -> eyre::Result<()> {
-        let segment = segments::Receipts::new(
-            compression,
-            if self.with_filters {
-                Filters::WithFilters(inclusion_filter, phf)
-            } else {
-                Filters::WithoutFilters
-            },
-        );
-        segment.snapshot::<DB>(provider, self.from..=(self.from + self.block_interval - 1))?;
+        let range = self.block_range();
+        let filters = if self.with_filters {
+            Filters::WithFilters(inclusion_filter, phf)
+        } else {
+            Filters::WithoutFilters
+        };
+
+        let segment = segments::Receipts::new(compression, filters);
+
+        segment.snapshot::<DB>(provider, PathBuf::default(), range.clone())?;
+
+        // Default name doesn't have any configuration
+        let default_name: PathBuf = SnapshotSegment::Receipts.filename(&range).into();
+        let new_name: PathBuf = SnapshotSegment::Receipts
+            .filename_with_configuration(filters, compression, &range)
+            .into();
+
+        std::fs::rename(default_name, new_name)?;
 
         Ok(())
     }
@@ -62,11 +74,10 @@ impl Command {
 
         let mut row_indexes = tx_range.clone().collect::<Vec<_>>();
 
-        let path = SnapshotSegment::Receipts.filename_with_configuration(
-            filters,
-            compression,
-            &block_range,
-        );
+        let path = SnapshotSegment::Receipts
+            .filename_with_configuration(filters, compression, &block_range)
+            .into();
+
         let provider = SnapshotProvider::default();
         let jar_provider =
             provider.get_segment_provider(SnapshotSegment::Receipts, self.from, Some(path))?;

bin/reth/src/db/snapshots/transactions.rs

@@ -14,7 +14,10 @@ use reth_provider::{
     TransactionsProvider, TransactionsProviderExt,
 };
 use reth_snapshot::{segments, segments::Segment};
-use std::{path::Path, sync::Arc};
+use std::{
+    path::{Path, PathBuf},
+    sync::Arc,
+};
 
 impl Command {
     pub(crate) fn generate_transactions_snapshot<DB: Database>(
@@ -24,15 +27,24 @@ impl Command {
         inclusion_filter: InclusionFilter,
         phf: PerfectHashingFunction,
     ) -> eyre::Result<()> {
-        let segment = segments::Transactions::new(
-            compression,
-            if self.with_filters {
-                Filters::WithFilters(inclusion_filter, phf)
-            } else {
-                Filters::WithoutFilters
-            },
-        );
-        segment.snapshot::<DB>(provider, self.from..=(self.from + self.block_interval - 1))?;
+        let range = self.block_range();
+        let filters = if self.with_filters {
+            Filters::WithFilters(inclusion_filter, phf)
+        } else {
+            Filters::WithoutFilters
+        };
+
+        let segment = segments::Transactions::new(compression, filters);
+
+        segment.snapshot::<DB>(provider, PathBuf::default(), range.clone())?;
+
+        // Default name doesn't have any configuration
+        let default_name: PathBuf = SnapshotSegment::Transactions.filename(&range).into();
+        let new_name: PathBuf = SnapshotSegment::Transactions
+            .filename_with_configuration(filters, compression, &range)
+            .into();
+
+        std::fs::rename(default_name, new_name)?;
 
         Ok(())
     }
@@ -62,11 +74,9 @@ impl Command {
 
         let mut row_indexes = tx_range.clone().collect::<Vec<_>>();
 
-        let path = SnapshotSegment::Transactions.filename_with_configuration(
-            filters,
-            compression,
-            &block_range,
-        );
+        let path = SnapshotSegment::Transactions
+            .filename_with_configuration(filters, compression, &block_range)
+            .into();
         let provider = SnapshotProvider::default();
         let jar_provider =
             provider.get_segment_provider(SnapshotSegment::Transactions, self.from, Some(path))?;

bin/reth/src/dirs.rs

@@ -266,6 +266,11 @@ impl<D> ChainPath<D> {
         self.0.join("db").into()
     }
 
+    /// Returns the path to the snapshots directory for this chain.
+    pub fn snapshots_path(&self) -> PathBuf {
+        self.0.join("snapshots").into()
+    }
+
     /// Returns the path to the reth p2p secret key for this chain.
     ///
     /// `<DIR>/<CHAIN_ID>/discovery-secret`

bin/reth/src/node/mod.rs

@@ -298,8 +298,17 @@ impl<Ext: RethCliExt> NodeCommand<Ext> {
         // fetch the head block from the database
         let head = self.lookup_head(Arc::clone(&db)).wrap_err("the head block is missing")?;
 
+        // configure snapshotter
+        let snapshotter = reth_snapshot::Snapshotter::new(
+            db.clone(),
+            data_dir.snapshots_path(),
+            self.chain.clone(),
+            self.chain.snapshot_block_interval,
+        )?;
+
         // setup the blockchain provider
-        let factory = ProviderFactory::new(Arc::clone(&db), Arc::clone(&self.chain));
+        let factory = ProviderFactory::new(Arc::clone(&db), Arc::clone(&self.chain))
+            .with_snapshots(data_dir.snapshots_path(), snapshotter.highest_snapshot_receiver());
         let blockchain_db = BlockchainProvider::new(factory, blockchain_tree.clone())?;
         let blob_store = InMemoryBlobStore::default();
         let validator = TransactionValidationTaskExecutor::eth_builder(Arc::clone(&self.chain))
@@ -454,12 +463,14 @@ impl<Ext: RethCliExt> NodeCommand<Ext> {
             None
         };
 
-        let (highest_snapshots_tx, highest_snapshots_rx) = watch::channel(None);
-
         let mut hooks = EngineHooks::new();
 
         let pruner_events = if let Some(prune_config) = prune_config {
-            let mut pruner = self.build_pruner(&prune_config, db.clone(), highest_snapshots_rx);
+            let mut pruner = self.build_pruner(
+                &prune_config,
+                db.clone(),
+                snapshotter.highest_snapshot_receiver(),
+            );
 
             let events = pruner.events();
             hooks.add(PruneHook::new(pruner, Box::new(ctx.task_executor.clone())));
@@ -470,13 +481,6 @@ impl<Ext: RethCliExt> NodeCommand<Ext> {
             Either::Right(stream::empty())
         };
 
-        let _snapshotter = reth_snapshot::Snapshotter::new(
-            db,
-            self.chain.clone(),
-            self.chain.snapshot_block_interval,
-            highest_snapshots_tx,
-        );
-
         // Configure the consensus engine
         let (beacon_consensus_engine, beacon_engine_handle) = BeaconConsensusEngine::with_channel(
             client,

crates/interfaces/src/error.rs

@@ -39,6 +39,12 @@ impl From<reth_nippy_jar::NippyJarError> for RethError {
     }
 }
 
+impl From<reth_primitives::fs::FsPathError> for RethError {
+    fn from(err: reth_primitives::fs::FsPathError) -> Self {
+        RethError::Custom(err.to_string())
+    }
+}
+
 // We don't want these types to be too large because they're used in a lot of places.
 const _SIZE_ASSERTIONS: () = {
     // Main error.

crates/primitives/src/fs.rs

@@ -1,6 +1,7 @@
 //! Wrapper for `std::fs` methods
 use std::{
-    fs, io,
+    fs::{self, ReadDir},
+    io,
     path::{Path, PathBuf},
 };
 
@@ -30,6 +31,9 @@ pub enum FsPathError {
     /// Provides additional path context for [`std::fs::remove_dir`].
     #[error("failed to remove dir {path:?}: {source}")]
     RemoveDir { source: io::Error, path: PathBuf },
+    /// Provides additional path context for [`std::fs::read_dir`].
+    #[error("failed to read dir {path:?}: {source}")]
+    ReadDir { source: io::Error, path: PathBuf },
     /// Provides additional path context for [`std::fs::File::open`].
     #[error("failed to open file {path:?}: {source}")]
     Open { source: io::Error, path: PathBuf },
@@ -77,6 +81,11 @@ impl FsPathError {
         FsPathError::RemoveDir { source, path: path.into() }
     }
 
+    /// Returns the complementary error variant for [`std::fs::read_dir`].
+    pub fn read_dir(source: io::Error, path: impl Into<PathBuf>) -> Self {
+        FsPathError::ReadDir { source, path: path.into() }
+    }
+
     /// Returns the complementary error variant for [`std::fs::File::open`].
     pub fn open(source: io::Error, path: impl Into<PathBuf>) -> Self {
         FsPathError::Open { source, path: path.into() }
@@ -108,3 +117,9 @@ pub fn create_dir_all(path: impl AsRef<Path>) -> Result<()> {
     let path = path.as_ref();
     fs::create_dir_all(path).map_err(|err| FsPathError::create_dir(err, path))
 }
+
+/// Wrapper for `std::fs::read_dir`
+pub fn read_dir(path: impl AsRef<Path>) -> Result<ReadDir> {
+    let path = path.as_ref();
+    fs::read_dir(path).map_err(|err| FsPathError::read_dir(err, path))
+}

crates/primitives/src/snapshot/compression.rs

@@ -1,11 +1,17 @@
-#[derive(Debug, Copy, Clone, Default)]
+use strum::AsRefStr;
+
+#[derive(Debug, Copy, Clone, Default, AsRefStr)]
 #[cfg_attr(feature = "clap", derive(clap::ValueEnum))]
 #[allow(missing_docs)]
 /// Snapshot compression
 pub enum Compression {
+    #[strum(serialize = "lz4")]
     Lz4,
+    #[strum(serialize = "zstd")]
     Zstd,
+    #[strum(serialize = "zstd-dict")]
     ZstdWithDictionary,
+    #[strum(serialize = "uncompressed")]
     #[default]
     Uncompressed,
 }

crates/primitives/src/snapshot/filters.rs

@@ -1,3 +1,5 @@
+use strum::AsRefStr;
+
 #[derive(Debug, Copy, Clone)]
 /// Snapshot filters.
 pub enum Filters {
@@ -14,20 +16,23 @@ impl Filters {
     }
 }
 
-#[derive(Debug, Copy, Clone)]
+#[derive(Debug, Copy, Clone, AsRefStr)]
 #[cfg_attr(feature = "clap", derive(clap::ValueEnum))]
 /// Snapshot inclusion filter. Also see [Filters].
 pub enum InclusionFilter {
+    #[strum(serialize = "cuckoo")]
     /// Cuckoo filter
     Cuckoo,
 }
 
-#[derive(Debug, Copy, Clone)]
+#[derive(Debug, Copy, Clone, AsRefStr)]
 #[cfg_attr(feature = "clap", derive(clap::ValueEnum))]
 /// Snapshot perfect hashing  function. Also see [Filters].
 pub enum PerfectHashingFunction {
+    #[strum(serialize = "fmph")]
     /// Fingerprint-Based Minimal Perfect Hash Function
     Fmph,
+    #[strum(serialize = "gofmph")]
     /// Fingerprint-Based Minimal Perfect Hash Function with Group Optimization
     GoFmph,
 }