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.from..=(self.from + self.block_interval - 1);
+        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, "".into(), 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(())
     }
@@ -55,8 +67,9 @@ impl Command {
 
         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/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.from..=(self.from + self.block_interval - 1);
+        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, "".into(), 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.from..=(self.from + self.block_interval - 1);
+        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, "".into(), 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

@@ -264,6 +264,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.
     pub fn p2p_secret_path(&self) -> PathBuf {
         self.0.join("discovery-secret").into()

bin/reth/src/node/mod.rs

@@ -294,8 +294,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))
@@ -450,12 +459,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())));
@@ -466,13 +477,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/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,
 }

crates/primitives/src/snapshot/mod.rs

@@ -4,9 +4,35 @@ mod compression;
 mod filters;
 mod segment;
 
+use alloy_primitives::BlockNumber;
 pub use compression::Compression;
 pub use filters::{Filters, InclusionFilter, PerfectHashingFunction};
-pub use segment::{SegmentHeader, SnapshotSegment};
+pub use segment::{SegmentConfig, SegmentHeader, SnapshotSegment};
 
 /// Default snapshot block count.
 pub const BLOCKS_PER_SNAPSHOT: u64 = 500_000;
+
+/// Highest snapshotted block numbers, per data part.
+#[derive(Debug, Clone, Copy, Default, Eq, PartialEq)]
+pub struct HighestSnapshots {
+    /// Highest snapshotted block of headers, inclusive.
+    /// If [`None`], no snapshot is available.
+    pub headers: Option<BlockNumber>,
+    /// Highest snapshotted block of receipts, inclusive.
+    /// If [`None`], no snapshot is available.
+    pub receipts: Option<BlockNumber>,
+    /// Highest snapshotted block of transactions, inclusive.
+    /// If [`None`], no snapshot is available.
+    pub transactions: Option<BlockNumber>,
+}
+
+impl HighestSnapshots {
+    /// Returns the highest snapshot if it exists for a segment
+    pub fn highest(&self, segment: SnapshotSegment) -> Option<BlockNumber> {
+        match segment {
+            SnapshotSegment::Headers => self.headers,
+            SnapshotSegment::Transactions => self.transactions,
+            SnapshotSegment::Receipts => self.receipts,
+        }
+    }
+}