feat: Snapshotter triggers segment snapshots

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

@@ -13,7 +13,10 @@ use reth_provider::{
     providers::SnapshotProvider, DatabaseProviderRO, HeaderProvider, ProviderError, ProviderFactory,
 };
 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_headers_snapshot<DB: Database>(
@@ -32,7 +35,7 @@ impl Command {
 
         let segment = segments::Headers::new(compression, filters);
 
-        segment.snapshot::<DB>(provider, range.clone())?;
+        segment.snapshot::<DB>(provider, PathBuf::default(), range.clone())?;
 
         // Default name doesn't have any configuration
         reth_primitives::fs::rename(

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>(
@@ -33,7 +36,7 @@ impl Command {
 
         let segment = segments::Receipts::new(compression, filters);
 
-        segment.snapshot::<DB>(provider, range.clone())?;
+        segment.snapshot::<DB>(provider, PathBuf::default(), range.clone())?;
 
         // Default name doesn't have any configuration
         reth_primitives::fs::rename(

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>(
@@ -33,7 +36,7 @@ impl Command {
 
         let segment = segments::Transactions::new(compression, filters);
 
-        segment.snapshot::<DB>(provider, range.clone())?;
+        segment.snapshot::<DB>(provider, PathBuf::default(), range.clone())?;
 
         // Default name doesn't have any configuration
         reth_primitives::fs::rename(

crates/primitives/src/snapshot/mod.rs

@@ -7,7 +7,7 @@ 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;

crates/primitives/src/snapshot/segment.rs

@@ -36,11 +36,14 @@ pub enum SnapshotSegment {
 
 impl SnapshotSegment {
     /// Returns the default configuration of the segment.
-    pub const fn config(&self) -> (Filters, Compression) {
-        let default_config = (
-            Filters::WithFilters(InclusionFilter::Cuckoo, super::PerfectHashingFunction::Fmph),
-            Compression::Lz4,
-        );
+    pub const fn config(&self) -> SegmentConfig {
+        let default_config = SegmentConfig {
+            filters: Filters::WithFilters(
+                InclusionFilter::Cuckoo,
+                super::PerfectHashingFunction::Fmph,
+            ),
+            compression: Compression::Lz4,
+        };
 
         match self {
             SnapshotSegment::Headers => default_config,
@@ -133,3 +136,12 @@ impl SegmentHeader {
         }
     }
 }
+
+/// Configuration used on the segment.
+#[derive(Debug, Clone, Copy)]
+pub struct SegmentConfig {
+    /// Inclusion filters used on the segment
+    pub filters: Filters,
+    /// Compression used on the segment
+    pub compression: Compression,
+}

crates/snapshot/src/segments/headers.rs

@@ -5,38 +5,48 @@ use reth_db::{
 };
 use reth_interfaces::RethResult;
 use reth_primitives::{
-    snapshot::{Compression, Filters},
+    snapshot::{Compression, Filters, SegmentConfig},
     BlockNumber, SnapshotSegment,
 };
 use reth_provider::DatabaseProviderRO;
-use std::ops::RangeInclusive;
+use std::{ops::RangeInclusive, path::Path};
 
 /// Snapshot segment responsible for [SnapshotSegment::Headers] part of data.
 #[derive(Debug)]
 pub struct Headers {
-    compression: Compression,
-    filters: Filters,
+    config: SegmentConfig,
 }
 
 impl Headers {
     /// Creates new instance of [Headers] snapshot segment.
     pub fn new(compression: Compression, filters: Filters) -> Self {
-        Self { compression, filters }
+        Self { config: SegmentConfig { compression, filters } }
+    }
+}
+
+impl Default for Headers {
+    fn default() -> Self {
+        Self { config: SnapshotSegment::Headers.config() }
     }
 }
 
 impl Segment for Headers {
+    fn segment() -> SnapshotSegment {
+        SnapshotSegment::Headers
+    }
+
     fn snapshot<DB: Database>(
         &self,
         provider: &DatabaseProviderRO<'_, DB>,
+        directory: impl AsRef<Path>,
         range: RangeInclusive<BlockNumber>,
     ) -> RethResult<()> {
         let range_len = range.clone().count();
         let mut jar = prepare_jar::<DB, 3>(
             provider,
-            SnapshotSegment::Headers,
-            self.filters,
-            self.compression,
+            directory,
+            Self::segment(),
+            self.config,
             range.clone(),
             range_len,
             || {
@@ -57,7 +67,7 @@ impl Segment for Headers {
         // Generate list of hashes for filters & PHF
         let mut cursor = provider.tx_ref().cursor_read::<RawTable<tables::CanonicalHeaders>>()?;
         let mut hashes = None;
-        if self.filters.has_filters() {
+        if self.config.filters.has_filters() {
             hashes = Some(
                 cursor
                     .walk(Some(RawKey::from(*range.start())))?

crates/snapshot/src/segments/mod.rs

@@ -15,7 +15,9 @@ use reth_db::{
 use reth_interfaces::RethResult;
 use reth_nippy_jar::NippyJar;
 use reth_primitives::{
-    snapshot::{Compression, Filters, InclusionFilter, PerfectHashingFunction, SegmentHeader},
+    snapshot::{
+        Compression, Filters, InclusionFilter, PerfectHashingFunction, SegmentConfig, SegmentHeader,
+    },
     BlockNumber, SnapshotSegment,
 };
 use reth_provider::{DatabaseProviderRO, TransactionsProviderExt};
@@ -24,14 +26,19 @@ use std::{ops::RangeInclusive, path::Path};
 pub(crate) type Rows<const COLUMNS: usize> = [Vec<Vec<u8>>; COLUMNS];
 
 /// A segment represents a snapshotting of some portion of the data.
-pub trait Segment {
-    /// Snapshot data using the provided range.
+pub trait Segment: Default {
+    /// Snapshot data using the provided range. The `directory` parameter determines the snapshot
+    /// file's save location.
     fn snapshot<DB: Database>(
         &self,
         provider: &DatabaseProviderRO<'_, DB>,
+        directory: impl AsRef<Path>,
         range: RangeInclusive<BlockNumber>,
     ) -> RethResult<()>;
 
+    /// Returns this struct's [`SnapshotSegment`].
+    fn segment() -> SnapshotSegment;
+
     /// Generates the dataset to train a zstd dictionary with the most recent rows (at most 1000).
     fn dataset_for_compression<DB: Database, T: Table<Key = u64>>(
         &self,
@@ -48,24 +55,25 @@ pub trait Segment {
     }
 }
 
-/// Returns a [`NippyJar`] according to the desired configuration.
+/// Returns a [`NippyJar`] according to the desired configuration. The `directory` parameter
+/// determines the snapshot file's save location.
 pub(crate) fn prepare_jar<DB: Database, const COLUMNS: usize>(
     provider: &DatabaseProviderRO<'_, DB>,
+    directory: impl AsRef<Path>,
     segment: SnapshotSegment,
-    filters: Filters,
-    compression: Compression,
+    segment_config: SegmentConfig,
     block_range: RangeInclusive<BlockNumber>,
     total_rows: usize,
     prepare_compression: impl Fn() -> RethResult<Rows<COLUMNS>>,
 ) -> RethResult<NippyJar<SegmentHeader>> {
     let tx_range = provider.transaction_range_by_block_range(block_range.clone())?;
     let mut nippy_jar = NippyJar::new(
         COLUMNS,
-        Path::new(segment.filename(&block_range).as_str()),
+        &directory.as_ref().join(segment.filename(&block_range).as_str()),
         SegmentHeader::new(block_range, tx_range, segment),
     );
 
-    nippy_jar = match compression {
+    nippy_jar = match segment_config.compression {
         Compression::Lz4 => nippy_jar.with_lz4(),
         Compression::Zstd => nippy_jar.with_zstd(false, 0),
         Compression::ZstdWithDictionary => {
@@ -78,7 +86,7 @@ pub(crate) fn prepare_jar<DB: Database, const COLUMNS: usize>(
         Compression::Uncompressed => nippy_jar,
     };
 
-    if let Filters::WithFilters(inclusion_filter, phf) = filters {
+    if let Filters::WithFilters(inclusion_filter, phf) = segment_config.filters {
         nippy_jar = match inclusion_filter {
             InclusionFilter::Cuckoo => nippy_jar.with_cuckoo_filter(total_rows),
         };

crates/snapshot/src/segments/receipts.rs

@@ -2,40 +2,50 @@ use crate::segments::{prepare_jar, Segment};
 use reth_db::{database::Database, snapshot::create_snapshot_T1, tables};
 use reth_interfaces::RethResult;
 use reth_primitives::{
-    snapshot::{Compression, Filters, SegmentHeader},
+    snapshot::{Compression, Filters, SegmentConfig, SegmentHeader},
     BlockNumber, SnapshotSegment, TxNumber,
 };
 use reth_provider::{DatabaseProviderRO, TransactionsProviderExt};
-use std::ops::RangeInclusive;
+use std::{ops::RangeInclusive, path::Path};
 
 /// Snapshot segment responsible for [SnapshotSegment::Receipts] part of data.
 #[derive(Debug)]
 pub struct Receipts {
-    compression: Compression,
-    filters: Filters,
+    config: SegmentConfig,
 }
 
 impl Receipts {
     /// Creates new instance of [Receipts] snapshot segment.
     pub fn new(compression: Compression, filters: Filters) -> Self {
-        Self { compression, filters }
+        Self { config: SegmentConfig { compression, filters } }
+    }
+}
+
+impl Default for Receipts {
+    fn default() -> Self {
+        Self { config: SnapshotSegment::Receipts.config() }
     }
 }
 
 impl Segment for Receipts {
+    fn segment() -> SnapshotSegment {
+        SnapshotSegment::Receipts
+    }
+
     fn snapshot<DB: Database>(
         &self,
         provider: &DatabaseProviderRO<'_, DB>,
+        directory: impl AsRef<Path>,
         block_range: RangeInclusive<BlockNumber>,
     ) -> RethResult<()> {
         let tx_range = provider.transaction_range_by_block_range(block_range.clone())?;
         let tx_range_len = tx_range.clone().count();
 
         let mut jar = prepare_jar::<DB, 1>(
             provider,
-            SnapshotSegment::Receipts,
-            self.filters,
-            self.compression,
+            directory,
+            Self::segment(),
+            self.config,
             block_range,
             tx_range_len,
             || {
@@ -49,7 +59,7 @@ impl Segment for Receipts {
 
         // Generate list of hashes for filters & PHF
         let mut hashes = None;
-        if self.filters.has_filters() {
+        if self.config.filters.has_filters() {
             hashes = Some(
                 provider
                     .transaction_hashes_by_range(*tx_range.start()..(*tx_range.end() + 1))?

crates/snapshot/src/segments/transactions.rs

@@ -2,40 +2,50 @@ use crate::segments::{prepare_jar, Segment};
 use reth_db::{database::Database, snapshot::create_snapshot_T1, tables};
 use reth_interfaces::RethResult;
 use reth_primitives::{
-    snapshot::{Compression, Filters, SegmentHeader},
+    snapshot::{Compression, Filters, SegmentConfig, SegmentHeader},
     BlockNumber, SnapshotSegment, TxNumber,
 };
 use reth_provider::{DatabaseProviderRO, TransactionsProviderExt};
-use std::ops::RangeInclusive;
+use std::{ops::RangeInclusive, path::Path};
 
 /// Snapshot segment responsible for [SnapshotSegment::Transactions] part of data.
 #[derive(Debug)]
 pub struct Transactions {
-    compression: Compression,
-    filters: Filters,
+    config: SegmentConfig,
 }
 
 impl Transactions {
     /// Creates new instance of [Transactions] snapshot segment.
     pub fn new(compression: Compression, filters: Filters) -> Self {
-        Self { compression, filters }
+        Self { config: SegmentConfig { compression, filters } }
+    }
+}
+
+impl Default for Transactions {
+    fn default() -> Self {
+        Self { config: SnapshotSegment::Transactions.config() }
     }
 }
 
 impl Segment for Transactions {
+    fn segment() -> SnapshotSegment {
+        SnapshotSegment::Transactions
+    }
+
     fn snapshot<DB: Database>(
         &self,
         provider: &DatabaseProviderRO<'_, DB>,
+        directory: impl AsRef<Path>,
         block_range: RangeInclusive<BlockNumber>,
     ) -> RethResult<()> {
         let tx_range = provider.transaction_range_by_block_range(block_range.clone())?;
         let tx_range_len = tx_range.clone().count();
 
         let mut jar = prepare_jar::<DB, 1>(
             provider,
-            SnapshotSegment::Transactions,
-            self.filters,
-            self.compression,
+            directory,
+            Self::segment(),
+            self.config,
             block_range,
             tx_range_len,
             || {
@@ -49,7 +59,7 @@ impl Segment for Transactions {
 
         // Generate list of hashes for filters & PHF
         let mut hashes = None;
-        if self.filters.has_filters() {
+        if self.config.filters.has_filters() {
             hashes = Some(
                 provider
                     .transaction_hashes_by_range(*tx_range.start()..(*tx_range.end() + 1))?