feat: add reth db snapshot transactions | receipts commands (#5007)

Co-authored-by: Alexey Shekhirin <a.shekhirin@gmail.com>

crates/storage/provider/src/lib.rs

@@ -25,7 +25,7 @@ pub use traits::{
     PruneCheckpointWriter, ReceiptProvider, ReceiptProviderIdExt, StageCheckpointReader,
     StageCheckpointWriter, StateProvider, StateProviderBox, StateProviderFactory,
     StateRootProvider, StorageReader, TransactionVariant, TransactionsProvider,
-    WithdrawalsProvider,
+    TransactionsProviderExt, WithdrawalsProvider,
 };
 
 /// Provider trait implementations.

crates/storage/provider/src/providers/database/provider.rs

@@ -6,7 +6,8 @@ use crate::{
     AccountReader, BlockExecutionWriter, BlockHashReader, BlockNumReader, BlockReader, BlockWriter,
     Chain, EvmEnvProvider, HashingWriter, HeaderProvider, HistoryWriter, OriginalValuesKnown,
     ProviderError, PruneCheckpointReader, PruneCheckpointWriter, StageCheckpointReader,
-    StorageReader, TransactionVariant, TransactionsProvider, WithdrawalsProvider,
+    StorageReader, TransactionVariant, TransactionsProvider, TransactionsProviderExt,
+    WithdrawalsProvider,
 };
 use itertools::{izip, Itertools};
 use reth_db::{
@@ -24,7 +25,7 @@ use reth_db::{
 };
 use reth_interfaces::{
     executor::{BlockExecutionError, BlockValidationError},
-    RethResult,
+    RethError, RethResult,
 };
 use reth_primitives::{
     keccak256,
@@ -46,7 +47,7 @@ use std::{
     collections::{hash_map, BTreeMap, BTreeSet, HashMap, HashSet},
     fmt::Debug,
     ops::{Deref, DerefMut, Range, RangeBounds, RangeInclusive},
-    sync::Arc,
+    sync::{mpsc, Arc},
 };
 
 /// A [`DatabaseProvider`] that holds a read-only database transaction.
@@ -1140,6 +1141,65 @@ impl<TX: DbTx> BlockReader for DatabaseProvider<TX> {
     }
 }
 
+impl<TX: DbTx> TransactionsProviderExt for DatabaseProvider<TX> {
+    /// Recovers transaction hashes by walking through `Transactions` table and
+    /// calculating them in a parallel manner. Returned unsorted.
+    fn transaction_hashes_by_range(
+        &self,
+        tx_range: Range<TxNumber>,
+    ) -> RethResult<Vec<(TxHash, TxNumber)>> {
+        let mut tx_cursor = self.tx.cursor_read::<tables::Transactions>()?;
+        let tx_range_size = tx_range.clone().count();
+        let tx_walker = tx_cursor.walk_range(tx_range)?;
+
+        let chunk_size = (tx_range_size / rayon::current_num_threads()).max(1);
+        let mut channels = Vec::with_capacity(chunk_size);
+        let mut transaction_count = 0;
+
+        #[inline]
+        fn calculate_hash(
+            entry: Result<(TxNumber, TransactionSignedNoHash), DatabaseError>,
+            rlp_buf: &mut Vec<u8>,
+        ) -> Result<(B256, TxNumber), Box<RethError>> {
+            let (tx_id, tx) = entry.map_err(|e| Box::new(e.into()))?;
+            tx.transaction.encode_with_signature(&tx.signature, rlp_buf, false);
+            Ok((keccak256(rlp_buf), tx_id))
+        }
+
+        for chunk in &tx_walker.chunks(chunk_size) {
+            let (tx, rx) = mpsc::channel();
+            channels.push(rx);
+
+            // Note: Unfortunate side-effect of how chunk is designed in itertools (it is not Send)
+            let chunk: Vec<_> = chunk.collect();
+            transaction_count += chunk.len();
+
+            // Spawn the task onto the global rayon pool
+            // This task will send the results through the channel after it has calculated the hash.
+            rayon::spawn(move || {
+                let mut rlp_buf = Vec::with_capacity(128);
+                for entry in chunk {
+                    rlp_buf.clear();
+                    let _ = tx.send(calculate_hash(entry, &mut rlp_buf));
+                }
+            });
+        }
+        let mut tx_list = Vec::with_capacity(transaction_count);
+
+        // Iterate over channels and append the tx hashes unsorted
+        for channel in channels {
+            while let Ok(tx) = channel.recv() {
+                let (tx_hash, tx_id) = tx.map_err(|boxed| *boxed)?;
+                tx_list.push((tx_hash, tx_id));
+            }
+        }
+
+        Ok(tx_list)
+    }
+}
+
+/// Calculates the hash of the given transaction
+
 impl<TX: DbTx> TransactionsProvider for DatabaseProvider<TX> {
     fn transaction_id(&self, tx_hash: TxHash) -> RethResult<Option<TxNumber>> {
         Ok(self.tx.get::<tables::TxHashNumber>(tx_hash)?)

crates/storage/provider/src/providers/snapshot.rs

@@ -1,11 +1,15 @@
-use crate::HeaderProvider;
+use crate::{BlockHashReader, BlockNumReader, HeaderProvider, TransactionsProvider};
 use reth_db::{
     table::{Decompress, Table},
     HeaderTD,
 };
 use reth_interfaces::{provider::ProviderError, RethResult};
 use reth_nippy_jar::{compression::Decompressor, NippyJar, NippyJarCursor};
-use reth_primitives::{BlockHash, BlockNumber, Header, SealedHeader, U256};
+use reth_primitives::{
+    snapshot::SegmentHeader, Address, BlockHash, BlockHashOrNumber, BlockNumber, ChainInfo, Header,
+    SealedHeader, TransactionMeta, TransactionSigned, TransactionSignedNoHash, TxHash, TxNumber,
+    B256, U256,
+};
 use std::ops::RangeBounds;
 
 /// SnapshotProvider
@@ -16,14 +20,12 @@ use std::ops::RangeBounds;
 #[derive(Debug)]
 pub struct SnapshotProvider<'a> {
     /// NippyJar
-    pub jar: &'a NippyJar,
-    /// Starting snapshot block
-    pub jar_start_block: u64,
+    pub jar: &'a NippyJar<SegmentHeader>,
 }
 
 impl<'a> SnapshotProvider<'a> {
     /// Creates cursor
-    pub fn cursor(&self) -> NippyJarCursor<'a> {
+    pub fn cursor(&self) -> NippyJarCursor<'a, SegmentHeader> {
         NippyJarCursor::new(self.jar, None).unwrap()
     }
 
@@ -31,7 +33,7 @@ impl<'a> SnapshotProvider<'a> {
     pub fn cursor_with_decompressors(
         &self,
         decompressors: Vec<Decompressor<'a>>,
-    ) -> NippyJarCursor<'a> {
+    ) -> NippyJarCursor<'a, SegmentHeader> {
         NippyJarCursor::new(self.jar, Some(decompressors)).unwrap()
     }
 }
@@ -57,7 +59,9 @@ impl<'a> HeaderProvider for SnapshotProvider<'a> {
     fn header_by_number(&self, num: BlockNumber) -> RethResult<Option<Header>> {
         Header::decompress(
             self.cursor()
-                .row_by_number_with_cols::<0b01, 2>((num - self.jar_start_block) as usize)?
+                .row_by_number_with_cols::<0b01, 2>(
+                    (num - self.jar.user_header().block_start()) as usize,
+                )?
                 .ok_or(ProviderError::HeaderNotFound(num.into()))?[0],
         )
         .map(Some)
@@ -101,6 +105,122 @@ impl<'a> HeaderProvider for SnapshotProvider<'a> {
     }
 }
 
+impl<'a> BlockHashReader for SnapshotProvider<'a> {
+    fn block_hash(&self, _number: u64) -> RethResult<Option<B256>> {
+        todo!()
+    }
+
+    fn canonical_hashes_range(
+        &self,
+        _start: BlockNumber,
+        _end: BlockNumber,
+    ) -> RethResult<Vec<B256>> {
+        todo!()
+    }
+}
+
+impl<'a> BlockNumReader for SnapshotProvider<'a> {
+    fn chain_info(&self) -> RethResult<ChainInfo> {
+        todo!()
+    }
+
+    fn best_block_number(&self) -> RethResult<BlockNumber> {
+        todo!()
+    }
+
+    fn last_block_number(&self) -> RethResult<BlockNumber> {
+        todo!()
+    }
+
+    fn block_number(&self, _hash: B256) -> RethResult<Option<BlockNumber>> {
+        todo!()
+    }
+}
+
+impl<'a> TransactionsProvider for SnapshotProvider<'a> {
+    fn transaction_id(&self, _tx_hash: TxHash) -> RethResult<Option<TxNumber>> {
+        todo!()
+    }
+
+    fn transaction_by_id(&self, num: TxNumber) -> RethResult<Option<TransactionSigned>> {
+        TransactionSignedNoHash::decompress(
+            self.cursor()
+                .row_by_number_with_cols::<0b1, 1>(
+                    (num - self.jar.user_header().tx_start()) as usize,
+                )?
+                .ok_or(ProviderError::TransactionNotFound(num.into()))?[0],
+        )
+        .map(Into::into)
+        .map(Some)
+        .map_err(Into::into)
+    }
+
+    fn transaction_by_id_no_hash(
+        &self,
+        _id: TxNumber,
+    ) -> RethResult<Option<TransactionSignedNoHash>> {
+        todo!()
+    }
+
+    fn transaction_by_hash(&self, hash: TxHash) -> RethResult<Option<TransactionSigned>> {
+        // WIP
+        let mut cursor = self.cursor();
+
+        let tx = TransactionSignedNoHash::decompress(
+            cursor.row_by_key_with_cols::<0b1, 1>(&hash.0).unwrap().unwrap()[0],
+        )
+        .unwrap()
+        .with_hash();
+
+        if tx.hash() == hash {
+            return Ok(Some(tx))
+        } else {
+            // check next snapshot
+        }
+        Ok(None)
+    }
+
+    fn transaction_by_hash_with_meta(
+        &self,
+        _hash: TxHash,
+    ) -> RethResult<Option<(TransactionSigned, TransactionMeta)>> {
+        todo!()
+    }
+
+    fn transaction_block(&self, _id: TxNumber) -> RethResult<Option<BlockNumber>> {
+        todo!()
+    }
+
+    fn transactions_by_block(
+        &self,
+        _block_id: BlockHashOrNumber,
+    ) -> RethResult<Option<Vec<TransactionSigned>>> {
+        todo!()
+    }
+
+    fn transactions_by_block_range(
+        &self,
+        _range: impl RangeBounds<BlockNumber>,
+    ) -> RethResult<Vec<Vec<TransactionSigned>>> {
+        todo!()
+    }
+
+    fn senders_by_tx_range(&self, _range: impl RangeBounds<TxNumber>) -> RethResult<Vec<Address>> {
+        todo!()
+    }
+
+    fn transactions_by_tx_range(
+        &self,
+        _range: impl RangeBounds<TxNumber>,
+    ) -> RethResult<Vec<reth_primitives::TransactionSignedNoHash>> {
+        todo!()
+    }
+
+    fn transaction_sender(&self, _id: TxNumber) -> RethResult<Option<Address>> {
+        todo!()
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::*;
@@ -123,6 +243,7 @@ mod test {
         // Ranges
         let row_count = 100u64;
         let range = 0..=(row_count - 1);
+        let segment_header = SegmentHeader::new(range.clone(), range.clone());
 
         // Data sources
         let db = create_test_rw_db();
@@ -157,7 +278,7 @@ mod test {
             let with_compression = true;
             let with_filter = true;
 
-            let mut nippy_jar = NippyJar::new_without_header(2, snap_file.path());
+            let mut nippy_jar = NippyJar::new(2, snap_file.path(), segment_header);
 
             if with_compression {
                 nippy_jar = nippy_jar.with_zstd(false, 0);
@@ -180,7 +301,7 @@ mod test {
                 .unwrap()
                 .map(|row| row.map(|(_key, value)| value.into_value()).map_err(|e| e.into()));
 
-            create_snapshot_T1_T2::<Headers, HeaderTD, BlockNumber>(
+            create_snapshot_T1_T2::<Headers, HeaderTD, BlockNumber, SegmentHeader>(
                 &tx,
                 range,
                 None,
@@ -194,10 +315,10 @@ mod test {
 
         // Use providers to query Header data and compare if it matches
         {
-            let jar = NippyJar::load_without_header(snap_file.path()).unwrap();
+            let jar = NippyJar::load(snap_file.path()).unwrap();
 
             let db_provider = factory.provider().unwrap();
-            let snap_provider = SnapshotProvider { jar: &jar, jar_start_block: 0 };
+            let snap_provider = SnapshotProvider { jar: &jar };
 
             assert!(!headers.is_empty());
 

crates/storage/provider/src/traits/mod.rs

@@ -37,7 +37,7 @@ pub use state::{
 };
 
 mod transactions;
-pub use transactions::TransactionsProvider;
+pub use transactions::{TransactionsProvider, TransactionsProviderExt};
 
 mod withdrawals;
 pub use withdrawals::WithdrawalsProvider;

crates/storage/provider/src/traits/transactions.rs

@@ -1,10 +1,10 @@
-use crate::BlockNumReader;
-use reth_interfaces::RethResult;
+use crate::{BlockNumReader, BlockReader};
+use reth_interfaces::{provider::ProviderError, RethResult};
 use reth_primitives::{
     Address, BlockHashOrNumber, BlockNumber, TransactionMeta, TransactionSigned,
     TransactionSignedNoHash, TxHash, TxNumber,
 };
-use std::ops::RangeBounds;
+use std::ops::{Range, RangeBounds, RangeInclusive};
 
 ///  Client trait for fetching [TransactionSigned] related data.
 #[auto_impl::auto_impl(&, Arc)]
@@ -63,3 +63,31 @@ pub trait TransactionsProvider: BlockNumReader + Send + Sync {
     /// Returns None if the transaction is not found.
     fn transaction_sender(&self, id: TxNumber) -> RethResult<Option<Address>>;
 }
+
+///  Client trait for fetching additional [TransactionSigned] related data.
+#[auto_impl::auto_impl(&, Arc)]
+pub trait TransactionsProviderExt: BlockReader + Send + Sync {
+    /// Get transactions range by block range.
+    fn transaction_range_by_block_range(
+        &self,
+        block_range: RangeInclusive<BlockNumber>,
+    ) -> RethResult<RangeInclusive<TxNumber>> {
+        let from = self
+            .block_body_indices(*block_range.start())?
+            .ok_or(ProviderError::BlockBodyIndicesNotFound(*block_range.start()))?
+            .first_tx_num();
+
+        let to = self
+            .block_body_indices(*block_range.end())?
+            .ok_or(ProviderError::BlockBodyIndicesNotFound(*block_range.end()))?
+            .last_tx_num();
+
+        Ok(from..=to)
+    }
+
+    /// Get transaction hashes from a transaction range.
+    fn transaction_hashes_by_range(
+        &self,
+        tx_range: Range<TxNumber>,
+    ) -> RethResult<Vec<(TxHash, TxNumber)>>;
+}