feat(snapshots): write Receipt to snapshots during ExecutionStage

bin/reth/src/commands/debug_cmd/in_memory_merkle.rs

@@ -204,7 +204,7 @@ impl Command {
                 .map_err(|_| BlockValidationError::SenderRecoveryError)?,
             None,
         )?;
-        block_state.write_to_db(provider_rw.tx_ref(), OriginalValuesKnown::No)?;
+        block_state.write_to_storage(provider_rw.tx_ref(), None, OriginalValuesKnown::No)?;
         let storage_lists = provider_rw.changed_storages_with_range(block.number..=block.number)?;
         let storages = provider_rw.plain_state_storages(storage_lists)?;
         provider_rw.insert_storage_for_hashing(storages)?;

crates/node-core/src/init.rs

@@ -153,7 +153,7 @@ pub fn insert_genesis_state<DB: Database>(
         0,
     );
 
-    bundle.write_to_db(tx, OriginalValuesKnown::Yes)?;
+    bundle.write_to_storage(tx, None, OriginalValuesKnown::Yes)?;
 
     Ok(())
 }

crates/storage/db/src/snapshot/cursor.rs

@@ -28,6 +28,10 @@ impl<'a> SnapshotCursor<'a> {
         key_or_num: KeyOrNumber<'_>,
         mask: usize,
     ) -> ProviderResult<Option<Vec<&'_ [u8]>>> {
+        if self.jar().rows() == 0 {
+            return Ok(None)
+        }
+
         let row = match key_or_num {
             KeyOrNumber::Key(k) => self.row_by_key_with_cols(k, mask),
             KeyOrNumber::Number(n) => {

crates/storage/nippy-jar/src/lib.rs

@@ -207,6 +207,11 @@ impl<H: NippyJarHeader> NippyJar<H> {
         &self.user_header
     }
 
+    /// Gets total rows in jar.
+    pub fn rows(&self) -> usize {
+        self.rows
+    }
+
     /// Returns the size of inclusion filter
     pub fn filter_size(&self) -> usize {
         self.size()

crates/storage/provider/src/bundle_state/bundle_state_with_receipts.rs

@@ -1,15 +1,15 @@
-use crate::{StateChanges, StateReverts};
+use crate::{providers::SnapshotProviderRWRefMut, StateChanges, StateReverts};
 use reth_db::{
     cursor::{DbCursorRO, DbCursorRW},
     tables,
     transaction::{DbTx, DbTxMut},
 };
-use reth_interfaces::db::DatabaseError;
+use reth_interfaces::provider::{ProviderError, ProviderResult};
 use reth_primitives::{
     logs_bloom,
     revm::compat::{into_reth_acc, into_revm_acc},
-    Account, Address, BlockNumber, Bloom, Bytecode, Log, Receipt, Receipts, StorageEntry, B256,
-    U256,
+    Account, Address, BlockNumber, Bloom, Bytecode, Log, Receipt, Receipts, SnapshotSegment,
+    StorageEntry, B256, U256,
 };
 use reth_trie::HashedPostState;
 use revm::{
@@ -285,15 +285,21 @@ impl BundleStateWithReceipts {
         std::mem::swap(&mut self.bundle, &mut other)
     }
 
-    /// Write bundle state to database.
+    /// Write bundle state to database and receipts to either database or static files if
+    /// `snapshotter` is `Some`. It should be none if there is any kind of pruning/filtering over
+    /// the receipts.
     ///
     /// `omit_changed_check` should be set to true of bundle has some of it data
     /// detached, This would make some original values not known.
-    pub fn write_to_db<TX: DbTxMut + DbTx>(
+    pub fn write_to_storage<TX>(
         self,
         tx: &TX,
+        mut snapshotter: Option<SnapshotProviderRWRefMut<'_>>,
         is_value_known: OriginalValuesKnown,
-    ) -> Result<(), DatabaseError> {
+    ) -> ProviderResult<()>
+    where
+        TX: DbTxMut + DbTx,
+    {
         let (plain_state, reverts) = self.bundle.into_plain_state_and_reverts(is_value_known);
 
         StateReverts(reverts).write_to_db(tx, self.first_block)?;
@@ -303,15 +309,22 @@ impl BundleStateWithReceipts {
         let mut receipts_cursor = tx.cursor_write::<tables::Receipts>()?;
 
         for (idx, receipts) in self.receipts.into_iter().enumerate() {
-            if !receipts.is_empty() {
-                let block_number = self.first_block + idx as u64;
-                let (_, body_indices) =
-                    bodies_cursor.seek_exact(block_number)?.unwrap_or_else(|| {
-                        let last_available = bodies_cursor.last().ok().flatten().map(|(number, _)| number);
-                        panic!("body indices for block {block_number} must exist. last available block number: {last_available:?}");
-                    });
-
-                let first_tx_index = body_indices.first_tx_num();
+            let block_number = self.first_block + idx as u64;
+            let first_tx_index = bodies_cursor
+                .seek_exact(block_number)?
+                .map(|(_, indices)| indices.first_tx_num())
+                .ok_or_else(|| ProviderError::BlockBodyIndicesNotFound(block_number))?;
+
+            if let Some(snapshotter) = &mut snapshotter {
+                // Increment block on static file header.
+                snapshotter.increment_block(SnapshotSegment::Receipts)?;
+
+                for (tx_idx, receipt) in receipts.into_iter().enumerate() {
+                    let receipt = receipt
+                        .expect("receipt should not be filtered when saving to static files.");
+                    snapshotter.append_receipt(first_tx_index + tx_idx as u64, receipt)?;
+                }
+            } else if !receipts.is_empty() {
                 for (tx_idx, receipt) in receipts.into_iter().enumerate() {
                     if let Some(receipt) = receipt {
                         receipts_cursor.append(first_tx_index + tx_idx as u64, receipt)?;
@@ -556,7 +569,7 @@ mod tests {
         state.merge_transitions(BundleRetention::Reverts);
 
         BundleStateWithReceipts::new(state.take_bundle(), Receipts::new(), 1)
-            .write_to_db(provider.tx_ref(), OriginalValuesKnown::Yes)
+            .write_to_storage(provider.tx_ref(), None, OriginalValuesKnown::Yes)
             .expect("Could not write bundle state to DB");
 
         // Check plain storage state
@@ -654,7 +667,7 @@ mod tests {
 
         state.merge_transitions(BundleRetention::Reverts);
         BundleStateWithReceipts::new(state.take_bundle(), Receipts::new(), 2)
-            .write_to_db(provider.tx_ref(), OriginalValuesKnown::Yes)
+            .write_to_storage(provider.tx_ref(), None, OriginalValuesKnown::Yes)
             .expect("Could not write bundle state to DB");
 
         assert_eq!(
@@ -718,7 +731,7 @@ mod tests {
         )]));
         init_state.merge_transitions(BundleRetention::Reverts);
         BundleStateWithReceipts::new(init_state.take_bundle(), Receipts::new(), 0)
-            .write_to_db(provider.tx_ref(), OriginalValuesKnown::Yes)
+            .write_to_storage(provider.tx_ref(), None, OriginalValuesKnown::Yes)
             .expect("Could not write init bundle state to DB");
 
         let mut state = State::builder().with_bundle_update().build();
@@ -863,7 +876,7 @@ mod tests {
         let bundle = state.take_bundle();
 
         BundleStateWithReceipts::new(bundle, Receipts::new(), 1)
-            .write_to_db(provider.tx_ref(), OriginalValuesKnown::Yes)
+            .write_to_storage(provider.tx_ref(), None, OriginalValuesKnown::Yes)
             .expect("Could not write bundle state to DB");
 
         let mut storage_changeset_cursor = provider
@@ -1026,7 +1039,7 @@ mod tests {
         )]));
         init_state.merge_transitions(BundleRetention::Reverts);
         BundleStateWithReceipts::new(init_state.take_bundle(), Receipts::new(), 0)
-            .write_to_db(provider.tx_ref(), OriginalValuesKnown::Yes)
+            .write_to_storage(provider.tx_ref(), None, OriginalValuesKnown::Yes)
             .expect("Could not write init bundle state to DB");
 
         let mut state = State::builder().with_bundle_update().build();
@@ -1071,7 +1084,7 @@ mod tests {
         // Commit block #1 changes to the database.
         state.merge_transitions(BundleRetention::Reverts);
         BundleStateWithReceipts::new(state.take_bundle(), Receipts::new(), 1)
-            .write_to_db(provider.tx_ref(), OriginalValuesKnown::Yes)
+            .write_to_storage(provider.tx_ref(), None, OriginalValuesKnown::Yes)
             .expect("Could not write bundle state to DB");
 
         let mut storage_changeset_cursor = provider

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

@@ -108,9 +108,9 @@ pub struct DatabaseProvider<TX> {
 }
 
 impl<TX> DatabaseProvider<TX> {
-    /// Returns snapshot provider
-    pub fn snapshot_provider(&self) -> Option<Arc<SnapshotProvider>> {
-        self.snapshot_provider.clone()
+    /// Returns a snapshot provider reference
+    pub fn snapshot_provider(&self) -> Option<&Arc<SnapshotProvider>> {
+        self.snapshot_provider.as_ref()
     }
 }
 
@@ -2499,7 +2499,7 @@ impl<TX: DbTxMut + DbTx> BlockWriter for DatabaseProvider<TX> {
 
         // Write state and changesets to the database.
         // Must be written after blocks because of the receipt lookup.
-        state.write_to_db(self.tx_ref(), OriginalValuesKnown::No)?;
+        state.write_to_storage(self.tx_ref(), None, OriginalValuesKnown::No)?;
         durations_recorder.record_relative(metrics::Action::InsertState);
 
         // insert hashes and intermediate merkle nodes

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

@@ -38,7 +38,10 @@ mod bundle_state_provider;
 mod chain_info;
 mod database;
 mod snapshot;
-pub use snapshot::{SnapshotJarProvider, SnapshotProvider, SnapshotProviderRW, SnapshotWriter};
+pub use snapshot::{
+    SnapshotJarProvider, SnapshotProvider, SnapshotProviderRW, SnapshotProviderRWRefMut,
+    SnapshotWriter,
+};
 mod state;
 use crate::{providers::chain_info::ChainInfoTracker, traits::BlockSource};
 pub use bundle_state_provider::BundleStateProvider;

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

@@ -1,13 +1,13 @@
-use super::{LoadedJar, SnapshotJarProvider, SnapshotProviderRW, BLOCKS_PER_SNAPSHOT};
+use super::{
+    LoadedJar, SnapshotJarProvider, SnapshotProviderRW, SnapshotProviderRWRefMut,
+    BLOCKS_PER_SNAPSHOT,
+};
 use crate::{
     to_range, BlockHashReader, BlockNumReader, BlockReader, BlockSource, HeaderProvider,
     ReceiptProvider, TransactionVariant, TransactionsProvider, TransactionsProviderExt,
     WithdrawalsProvider,
 };
-use dashmap::{
-    mapref::{entry::Entry as DashMapEntry, one::RefMut},
-    DashMap,
-};
+use dashmap::{mapref::entry::Entry as DashMapEntry, DashMap};
 use parking_lot::RwLock;
 use reth_db::{
     codecs::CompactU256,
@@ -264,6 +264,13 @@ impl SnapshotProvider {
                             index.insert(tx_end, current_block_range.clone());
                         })
                         .or_insert_with(|| BTreeMap::from([(tx_end, current_block_range)]));
+                } else if let Some(1) = tx_index.get(&segment).map(|index| index.len()) {
+                    // Only happens if we unwind all the txs/receipts from the first static file.
+                    // Should only happen in test scenarios.
+                    if matches!(segment, SnapshotSegment::Receipts | SnapshotSegment::Transactions)
+                    {
+                        tx_index.remove(&segment);
+                    }
                 }
 
                 // Update the cached provider.
@@ -420,13 +427,13 @@ pub trait SnapshotWriter {
         &self,
         block: BlockNumber,
         segment: SnapshotSegment,
-    ) -> ProviderResult<RefMut<'_, SnapshotSegment, SnapshotProviderRW<'static>>>;
+    ) -> ProviderResult<SnapshotProviderRWRefMut<'_>>;
 
     /// Returns a mutable reference to a [`SnapshotProviderRW`] of the latest [`SnapshotSegment`].
     fn latest_writer(
         &self,
         segment: SnapshotSegment,
-    ) -> ProviderResult<RefMut<'_, SnapshotSegment, SnapshotProviderRW<'static>>>;
+    ) -> ProviderResult<SnapshotProviderRWRefMut<'_>>;
 
     /// Commits all changes of all [`SnapshotProviderRW`] of all [`SnapshotSegment`].
     fn commit(&self) -> ProviderResult<()>;
@@ -437,7 +444,7 @@ impl SnapshotWriter for Arc<SnapshotProvider> {
         &self,
         block: BlockNumber,
         segment: SnapshotSegment,
-    ) -> ProviderResult<RefMut<'_, SnapshotSegment, SnapshotProviderRW<'static>>> {
+    ) -> ProviderResult<SnapshotProviderRWRefMut<'_>> {
         Ok(match self.writers.entry(segment) {
             DashMapEntry::Occupied(entry) => entry.into_ref(),
             DashMapEntry::Vacant(entry) => {
@@ -449,7 +456,7 @@ impl SnapshotWriter for Arc<SnapshotProvider> {
     fn latest_writer(
         &self,
         segment: SnapshotSegment,
-    ) -> ProviderResult<RefMut<'_, SnapshotSegment, SnapshotProviderRW<'static>>> {
+    ) -> ProviderResult<SnapshotProviderRWRefMut<'_>> {
         self.get_writer(self.get_highest_snapshot_block(segment).unwrap_or_default(), segment)
     }
 

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

@@ -5,7 +5,7 @@ mod jar;
 pub use jar::SnapshotJarProvider;
 
 mod writer;
-pub use writer::SnapshotProviderRW;
+pub use writer::{SnapshotProviderRW, SnapshotProviderRWRefMut};
 
 use reth_interfaces::provider::ProviderResult;
 use reth_nippy_jar::NippyJar;

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

@@ -1,4 +1,5 @@
 use super::SnapshotProvider;
+use dashmap::mapref::one::RefMut;
 use reth_codecs::Compact;
 use reth_interfaces::provider::{ProviderError, ProviderResult};
 use reth_nippy_jar::{NippyJar, NippyJarError, NippyJarWriter};
@@ -9,6 +10,9 @@ use reth_primitives::{
 };
 use std::{ops::Deref, path::PathBuf, sync::Arc};
 
+/// Mutable reference to a dashmap element of [`SnapshotProviderRW`].
+pub type SnapshotProviderRWRefMut<'a> = RefMut<'a, SnapshotSegment, SnapshotProviderRW<'static>>;
+
 #[derive(Debug)]
 /// Extends `SnapshotProvider` with writing capabilities
 pub struct SnapshotProviderRW<'a> {
@@ -274,6 +278,21 @@ impl<'a> SnapshotProviderRW<'a> {
         self.truncate(segment, number, Some(last_block))
     }
 
+    /// Prunes `to_delete` number of receipts from snapshots.
+    ///
+    /// # Note
+    /// Commits to the configuration file at the end.
+    pub fn prune_receipts(
+        &mut self,
+        to_delete: u64,
+        last_block: BlockNumber,
+    ) -> ProviderResult<()> {
+        let segment = SnapshotSegment::Receipts;
+        debug_assert!(self.writer.user_header().segment() == segment);
+
+        self.truncate(segment, to_delete, Some(last_block))
+    }
+
     #[cfg(any(test, feature = "test-utils"))]
     /// Helper function to override block range for testing.
     pub fn set_block_range(&mut self, block_range: std::ops::RangeInclusive<BlockNumber>) {

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

@@ -1,6 +1,6 @@
 use crate::ProviderFactory;
 use reth_db::{
-    test_utils::{create_test_rw_db, TempDatabase},
+    test_utils::{create_test_rw_db, create_test_snapshots_dir, TempDatabase},
     DatabaseEnv,
 };
 use reth_primitives::{ChainSpec, MAINNET};
@@ -27,5 +27,5 @@ pub fn create_test_provider_factory_with_chain_spec(
     chain_spec: Arc<ChainSpec>,
 ) -> ProviderFactory<Arc<TempDatabase<DatabaseEnv>>> {
     let db = create_test_rw_db();
-    ProviderFactory::new(db, chain_spec)
+    ProviderFactory::new(db, chain_spec).with_snapshots(create_test_snapshots_dir()).unwrap()
 }

testing/ef-tests/src/cases/blockchain_test.rs

@@ -5,7 +5,7 @@ use crate::{
     Case, Error, Suite,
 };
 use alloy_rlp::Decodable;
-use reth_db::test_utils::create_test_rw_db;
+use reth_db::test_utils::{create_test_rw_db, create_test_snapshots_dir};
 use reth_primitives::{BlockBody, SealedBlock};
 use reth_provider::{BlockWriter, HashingWriter, ProviderFactory};
 use reth_stages::{stages::ExecutionStage, ExecInput, Stage};
@@ -78,6 +78,8 @@ impl Case for BlockchainTestCase {
             // Create a new test database and initialize a provider for the test case.
             let db = create_test_rw_db();
             let provider = ProviderFactory::new(db.as_ref(), Arc::new(case.network.clone().into()))
+                .with_snapshots(create_test_snapshots_dir())
+                .map_err(|err| Error::RethError(err.into()))?
                 .provider_rw()
                 .unwrap();