feat(prune): Add MerkleChangeSets segment

crates/prune/prune/src/segments/mod.rs

@@ -15,8 +15,8 @@ pub use static_file::{
 use std::{fmt::Debug, ops::RangeInclusive};
 use tracing::error;
 pub use user::{
-    AccountHistory, Receipts as UserReceipts, ReceiptsByLogs, SenderRecovery, StorageHistory,
-    TransactionLookup,
+    AccountHistory, MerkleChangeSets, Receipts as UserReceipts, ReceiptsByLogs, SenderRecovery,
+    StorageHistory, TransactionLookup,
 };
 
 /// A segment represents a pruning of some portion of the data.

crates/prune/prune/src/segments/user/merkle_change_sets.rs

@@ -0,0 +1,107 @@
+use crate::{
+    db_ext::DbTxPruneExt,
+    segments::{PruneInput, Segment},
+    PrunerError,
+};
+use reth_db_api::{models::BlockNumberAddress, table::Value, tables, transaction::DbTxMut};
+use reth_primitives_traits::NodePrimitives;
+use reth_provider::{
+    errors::provider::ProviderResult, BlockReader, DBProvider, NodePrimitivesProvider,
+    PruneCheckpointWriter, TransactionsProvider,
+};
+use reth_prune_types::{
+    PruneCheckpoint, PruneMode, PrunePurpose, PruneSegment, SegmentOutput, SegmentOutputCheckpoint,
+};
+use tracing::{instrument, trace};
+
+#[derive(Debug)]
+pub struct MerkleChangeSets {
+    mode: PruneMode,
+}
+
+impl MerkleChangeSets {
+    pub const fn new(mode: PruneMode) -> Self {
+        Self { mode }
+    }
+}
+
+impl<Provider> Segment<Provider> for MerkleChangeSets
+where
+    Provider: DBProvider<Tx: DbTxMut>
+        + PruneCheckpointWriter
+        + TransactionsProvider
+        + BlockReader
+        + NodePrimitivesProvider<Primitives: NodePrimitives<Receipt: Value>>,
+{
+    fn segment(&self) -> PruneSegment {
+        PruneSegment::MerkleChangeSets
+    }
+
+    fn mode(&self) -> Option<PruneMode> {
+        Some(self.mode)
+    }
+
+    fn purpose(&self) -> PrunePurpose {
+        PrunePurpose::User
+    }
+
+    #[instrument(level = "trace", target = "pruner", skip(self, provider), ret)]
+    fn prune(&self, provider: &Provider, input: PruneInput) -> Result<SegmentOutput, PrunerError> {
+        let Some(block_range) = input.get_next_block_range() else {
+            trace!(target: "pruner", "No change sets to prune");
+            return Ok(SegmentOutput::done())
+        };
+
+        let block_range_end = *block_range.end();
+        let mut limiter = input.limiter;
+
+        let mut last_storages_pruned_block = None;
+        let (storages_pruned, done) =
+            provider.tx_ref().prune_table_with_range::<tables::StorageChangeSets>(
+                BlockNumberAddress::range(block_range.clone()),
+                &mut limiter,
+                |_| false,
+                |(BlockNumberAddress((block_number, ..)), ..)| {
+                    last_storages_pruned_block = Some(block_number);
+                },
+            )?;
+
+        trace!(target: "pruner", %storages_pruned, %done, "Pruned storages change sets");
+
+        let mut last_accounts_pruned_block = block_range_end;
+        let last_storages_pruned_block = last_storages_pruned_block
+            // If there's more storage changesets to prune, set the checkpoint block number to
+            // previous, so we could finish pruning its storage changesets on the next run.
+            .map(|block_number| if done { block_number } else { block_number.saturating_sub(1) })
+            .unwrap_or(block_range_end);
+
+        let (accounts_pruned, done) =
+            provider.tx_ref().prune_table_with_range::<tables::AccountsTrieChangeSets>(
+                block_range,
+                &mut limiter,
+                |_| false,
+                |row| last_accounts_pruned_block = row.0,
+            )?;
+
+        trace!(target: "pruner", %accounts_pruned, %done, "Pruned accounts change sets");
+
+        let progress = limiter.progress(done);
+
+        Ok(SegmentOutput {
+            progress,
+            pruned: accounts_pruned + storages_pruned,
+            checkpoint: Some(SegmentOutputCheckpoint {
+                block_number: Some(last_storages_pruned_block.min(last_accounts_pruned_block)),
+                tx_number: None,
+            }),
+        })
+    }
+
+    fn save_checkpoint(
+        &self,
+        provider: &Provider,
+        checkpoint: PruneCheckpoint,
+    ) -> ProviderResult<()> {
+        provider.save_prune_checkpoint(PruneSegment::MerkleChangeSets, checkpoint)
+    }
+}

crates/prune/prune/src/segments/user/mod.rs

@@ -1,12 +1,14 @@
 mod account_history;
 mod history;
+mod merkle_change_sets;
 mod receipts;
 mod receipts_by_logs;
 mod sender_recovery;
 mod storage_history;
 mod transaction_lookup;
 
 pub use account_history::AccountHistory;
+pub use merkle_change_sets::MerkleChangeSets;
 pub use receipts::Receipts;
 pub use receipts_by_logs::ReceiptsByLogs;
 pub use sender_recovery::SenderRecovery;

crates/prune/types/src/segment.rs

@@ -15,6 +15,9 @@ pub enum PruneSegment {
     TransactionLookup,
     /// Prune segment responsible for all rows in `Receipts` table.
     Receipts,
+    /// Prune segment responsible for all rows in `AccountsTrieChangeSets` and
+    /// `StoragesTrieChangeSets` table.
+    MerkleChangeSets,
     /// Prune segment responsible for some rows in `Receipts` table filtered by logs.
     ContractLogs,
     /// Prune segment responsible for the `AccountChangeSets` and `AccountsHistory` tables.
@@ -44,9 +47,10 @@ impl PruneSegment {
                 0
             }
             Self::Receipts if purpose.is_static_file() => 0,
-            Self::ContractLogs | Self::AccountHistory | Self::StorageHistory => {
-                MINIMUM_PRUNING_DISTANCE
-            }
+            Self::ContractLogs |
+            Self::AccountHistory |
+            Self::StorageHistory |
+            Self::MerkleChangeSets |
             Self::Receipts => MINIMUM_PRUNING_DISTANCE,
         }
     }