chore: bump version v1.9.4

Co-authored-by: theo <80177219+theochap@users.noreply.github.com>

.github/workflows/release.yml

@@ -78,7 +78,7 @@ jobs:
             profile: maxperf
             allow_fail: false
           - target: x86_64-apple-darwin
-            os: macos-13
+            os: macos-14
             profile: maxperf
             allow_fail: false
           - target: aarch64-apple-darwin

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace.package]
-version = "1.9.0"
+version = "1.9.4"
 edition = "2024"
 rust-version = "1.88"
 license = "MIT OR Apache-2.0"
@@ -466,17 +466,17 @@ reth-ress-protocol = { path = "crates/ress/protocol" }
 reth-ress-provider = { path = "crates/ress/provider" }
 
 # revm
-revm = { version = "31.0.0", default-features = false }
-revm-bytecode = { version = "7.1.0", default-features = false }
-revm-database = { version = "9.0.3", default-features = false }
-revm-state = { version = "8.1.0", default-features = false }
-revm-primitives = { version = "21.0.1", default-features = false }
-revm-interpreter = { version = "29.0.0", default-features = false }
-revm-inspector = { version = "12.0.0", default-features = false }
-revm-context = { version = "11.0.0", default-features = false }
-revm-context-interface = { version = "12.0.0", default-features = false }
-revm-database-interface = { version = "8.0.4", default-features = false }
-op-revm = { version = "12.0.0", default-features = false }
+revm = { version = "31.0.2", default-features = false }
+revm-bytecode = { version = "7.1.1", default-features = false }
+revm-database = { version = "9.0.5", default-features = false }
+revm-state = { version = "8.1.1", default-features = false }
+revm-primitives = { version = "21.0.2", default-features = false }
+revm-interpreter = { version = "29.0.1", default-features = false }
+revm-inspector = { version = "12.0.2", default-features = false }
+revm-context = { version = "11.0.2", default-features = false }
+revm-context-interface = { version = "12.0.1", default-features = false }
+revm-database-interface = { version = "8.0.5", default-features = false }
+op-revm = { version = "12.0.2", default-features = false }
 revm-inspectors = "0.32.0"
 
 # eth

crates/config/src/config.rs

@@ -33,7 +33,7 @@ pub struct Config {
 
 impl Config {
     /// Sets the pruning configuration.
-    pub const fn set_prune_config(&mut self, prune_config: PruneConfig) {
+    pub fn set_prune_config(&mut self, prune_config: PruneConfig) {
         self.prune = prune_config;
     }
 }
@@ -451,14 +451,13 @@ impl PruneConfig {
     }
 
     /// Returns whether there is any kind of receipt pruning configuration.
-    pub const fn has_receipts_pruning(&self) -> bool {
-        self.segments.receipts.is_some()
+    pub fn has_receipts_pruning(&self) -> bool {
+        self.segments.receipts.is_some() || !self.segments.receipts_log_filter.is_empty()
     }
 
     /// Merges another `PruneConfig` into this one, taking values from the other config if and only
     /// if the corresponding value in this config is not set.
     pub fn merge(&mut self, other: Self) {
-        #[expect(deprecated)]
         let Self {
             block_interval,
             segments:
@@ -470,7 +469,7 @@ impl PruneConfig {
                     storage_history,
                     bodies_history,
                     merkle_changesets,
-                    receipts_log_filter: (),
+                    receipts_log_filter,
                 },
         } = other;
 
@@ -488,6 +487,10 @@ impl PruneConfig {
         self.segments.bodies_history = self.segments.bodies_history.or(bodies_history);
         // Merkle changesets is not optional, so we just replace it if provided
         self.segments.merkle_changesets = merkle_changesets;
+
+        if self.segments.receipts_log_filter.0.is_empty() && !receipts_log_filter.0.is_empty() {
+            self.segments.receipts_log_filter = receipts_log_filter;
+        }
     }
 }
 
@@ -514,9 +517,10 @@ where
 mod tests {
     use super::{Config, EXTENSION};
     use crate::PruneConfig;
+    use alloy_primitives::Address;
     use reth_network_peers::TrustedPeer;
-    use reth_prune_types::{PruneMode, PruneModes};
-    use std::{path::Path, str::FromStr, time::Duration};
+    use reth_prune_types::{PruneMode, PruneModes, ReceiptsLogPruneConfig};
+    use std::{collections::BTreeMap, path::Path, str::FromStr, time::Duration};
 
     fn with_tempdir(filename: &str, proc: fn(&std::path::Path)) {
         let temp_dir = tempfile::tempdir().unwrap();
@@ -1005,8 +1009,10 @@ receipts = 'full'
                 storage_history: Some(PruneMode::Before(5000)),
                 bodies_history: None,
                 merkle_changesets: PruneMode::Before(0),
-                #[expect(deprecated)]
-                receipts_log_filter: (),
+                receipts_log_filter: ReceiptsLogPruneConfig(BTreeMap::from([(
+                    Address::random(),
+                    PruneMode::Full,
+                )])),
             },
         };
 
@@ -1020,11 +1026,14 @@ receipts = 'full'
                 storage_history: Some(PruneMode::Distance(3000)),
                 bodies_history: None,
                 merkle_changesets: PruneMode::Distance(10000),
-                #[expect(deprecated)]
-                receipts_log_filter: (),
+                receipts_log_filter: ReceiptsLogPruneConfig(BTreeMap::from([
+                    (Address::random(), PruneMode::Distance(1000)),
+                    (Address::random(), PruneMode::Before(2000)),
+                ])),
             },
         };
 
+        let original_filter = config1.segments.receipts_log_filter.clone();
         config1.merge(config2);
 
         // Check that the configuration has been merged. Any configuration present in config1
@@ -1036,6 +1045,7 @@ receipts = 'full'
         assert_eq!(config1.segments.account_history, Some(PruneMode::Distance(2000)));
         assert_eq!(config1.segments.storage_history, Some(PruneMode::Before(5000)));
         assert_eq!(config1.segments.merkle_changesets, PruneMode::Distance(10000));
+        assert_eq!(config1.segments.receipts_log_filter, original_filter);
     }
 
     #[test]

crates/exex/exex/src/backfill/factory.rs

@@ -39,7 +39,7 @@ impl<E, P> BackfillJobFactory<E, P> {
     }
 
     /// Sets the prune modes
-    pub const fn with_prune_modes(mut self, prune_modes: PruneModes) -> Self {
+    pub fn with_prune_modes(mut self, prune_modes: PruneModes) -> Self {
         self.prune_modes = prune_modes;
         self
     }

crates/node/builder/src/launch/common.rs

@@ -169,7 +169,7 @@ impl LaunchContext {
     }
 
     /// Save prune config to the toml file if node is a full node or has custom pruning CLI
-    /// arguments.
+    /// arguments. Also migrates deprecated prune config values to new defaults.
     fn save_pruning_config<ChainSpec>(
         reth_config: &mut reth_config::Config,
         config: &NodeConfig<ChainSpec>,
@@ -178,15 +178,22 @@ impl LaunchContext {
     where
         ChainSpec: EthChainSpec + reth_chainspec::EthereumHardforks,
     {
+        let mut should_save = reth_config.prune.segments.migrate();
+
         if let Some(prune_config) = config.prune_config() {
             if reth_config.prune != prune_config {
                 reth_config.set_prune_config(prune_config);
-                info!(target: "reth::cli", "Saving prune config to toml file");
-                reth_config.save(config_path.as_ref())?;
+                should_save = true;
             }
         } else if !reth_config.prune.is_default() {
             warn!(target: "reth::cli", "Pruning configuration is present in the config file, but no CLI arguments are provided. Using config from file.");
         }
+
+        if should_save {
+            info!(target: "reth::cli", "Saving prune config to toml file");
+            reth_config.save(config_path.as_ref())?;
+        }
+
         Ok(())
     }
 
@@ -405,14 +412,13 @@ impl<R, ChainSpec: EthChainSpec> LaunchContextWith<Attached<WithConfigs<ChainSpe
     where
         ChainSpec: reth_chainspec::EthereumHardforks,
     {
-        let toml_config = self.toml_config().prune.clone();
         let Some(mut node_prune_config) = self.node_config().prune_config() else {
             // No CLI config is set, use the toml config.
-            return toml_config;
+            return self.toml_config().prune.clone();
         };
 
         // Otherwise, use the CLI configuration and merge with toml config.
-        node_prune_config.merge(toml_config);
+        node_prune_config.merge(self.toml_config().prune.clone());
         node_prune_config
     }
 
@@ -1171,7 +1177,6 @@ mod tests {
                     storage_history_before: None,
                     bodies_pre_merge: false,
                     bodies_distance: None,
-                    #[expect(deprecated)]
                     receipts_log_filter: None,
                     bodies_before: None,
                 },

crates/node/core/Cargo.toml

@@ -52,6 +52,7 @@ derive_more.workspace = true
 toml.workspace = true
 serde.workspace = true
 strum = { workspace = true, features = ["derive"] }
+thiserror.workspace = true
 url.workspace = true
 
 # io

crates/node/core/src/args/error.rs

@@ -0,0 +1,22 @@
+use std::num::ParseIntError;
+
+/// Error while parsing a `ReceiptsLogPruneConfig`
+#[derive(thiserror::Error, Debug)]
+#[expect(clippy::enum_variant_names)]
+pub(crate) enum ReceiptsLogError {
+    /// The format of the filter is invalid.
+    #[error("invalid filter format: {0}")]
+    InvalidFilterFormat(String),
+    /// Address is invalid.
+    #[error("address is invalid: {0}")]
+    InvalidAddress(String),
+    /// The prune mode is not one of full, distance, before.
+    #[error("prune mode is invalid: {0}")]
+    InvalidPruneMode(String),
+    /// The distance value supplied is invalid.
+    #[error("distance is invalid: {0}")]
+    InvalidDistance(ParseIntError),
+    /// The block number supplied is invalid.
+    #[error("block number is invalid: {0}")]
+    InvalidBlockNumber(ParseIntError),
+}

crates/node/core/src/args/mod.rs

@@ -76,4 +76,5 @@ pub use ress_args::RessArgs;
 mod era;
 pub use era::{DefaultEraHost, EraArgs, EraSourceArgs};
 
+mod error;
 pub mod types;

crates/node/core/src/args/pruning.rs

@@ -1,13 +1,15 @@
 //! Pruning and full node arguments
 
-use std::ops::Not;
-
-use crate::primitives::EthereumHardfork;
-use alloy_primitives::BlockNumber;
+use crate::{args::error::ReceiptsLogError, primitives::EthereumHardfork};
+use alloy_primitives::{Address, BlockNumber};
 use clap::{builder::RangedU64ValueParser, Args};
 use reth_chainspec::EthereumHardforks;
 use reth_config::config::PruneConfig;
-use reth_prune_types::{PruneMode, PruneModes, MINIMUM_PRUNING_DISTANCE};
+use reth_prune_types::{
+    PruneMode, PruneModes, ReceiptsLogPruneConfig, MERKLE_CHANGESETS_RETENTION_BLOCKS,
+    MINIMUM_PRUNING_DISTANCE,
+};
+use std::{collections::BTreeMap, ops::Not};
 
 /// Parameters for pruning and full node
 #[derive(Debug, Clone, Args, PartialEq, Eq, Default)]
@@ -60,15 +62,12 @@ pub struct PruningArgs {
     /// Prune receipts before the specified block number. The specified block number is not pruned.
     #[arg(long = "prune.receipts.before", value_name = "BLOCK_NUMBER", conflicts_with_all = &["receipts_full", "receipts_pre_merge", "receipts_distance"])]
     pub receipts_before: Option<BlockNumber>,
-    /// Receipts Log Filter
-    #[arg(
-        long = "prune.receipts-log-filter",
-        alias = "prune.receiptslogfilter",
-        value_name = "FILTER_CONFIG",
-        hide = true
-    )]
-    #[deprecated]
-    pub receipts_log_filter: Option<String>,
+    // Receipts Log Filter
+    /// Configure receipts log filter. Format:
+    /// <`address`>:<`prune_mode`>... where <`prune_mode`> can be 'full', 'distance:<`blocks`>', or
+    /// 'before:<`block_number`>'
+    #[arg(long = "prune.receiptslogfilter", value_name = "FILTER_CONFIG", conflicts_with_all = &["receipts_full", "receipts_pre_merge", "receipts_distance",  "receipts_before"], value_parser = parse_receipts_log_filter)]
+    pub receipts_log_filter: Option<ReceiptsLogPruneConfig>,
 
     // Account History
     /// Prunes all account history.
@@ -135,9 +134,8 @@ impl PruningArgs {
                         .ethereum_fork_activation(EthereumHardfork::Paris)
                         .block_number()
                         .map(PruneMode::Before),
-                    merkle_changesets: PruneMode::Distance(MINIMUM_PRUNING_DISTANCE),
-                    #[expect(deprecated)]
-                    receipts_log_filter: (),
+                    merkle_changesets: PruneMode::Distance(MERKLE_CHANGESETS_RETENTION_BLOCKS),
+                    receipts_log_filter: Default::default(),
                 },
             }
         }
@@ -164,14 +162,13 @@ impl PruningArgs {
         if let Some(mode) = self.storage_history_prune_mode() {
             config.segments.storage_history = Some(mode);
         }
-
-        // Log warning if receipts_log_filter is set (deprecated feature)
-        #[expect(deprecated)]
-        if self.receipts_log_filter.is_some() {
-            tracing::warn!(
-                target: "reth::cli",
-                "The --prune.receiptslogfilter flag is deprecated and has no effect. It will be removed in a future release."
-            );
+        if let Some(receipt_logs) =
+            self.receipts_log_filter.as_ref().filter(|c| !c.is_empty()).cloned()
+        {
+            config.segments.receipts_log_filter = receipt_logs;
+            // need to remove the receipts segment filter entirely because that takes precedence
+            // over the logs filter
+            config.segments.receipts.take();
         }
 
         config.is_default().not().then_some(config)
@@ -259,3 +256,141 @@ impl PruningArgs {
         }
     }
 }
+
+/// Parses `,` separated pruning info into [`ReceiptsLogPruneConfig`].
+pub(crate) fn parse_receipts_log_filter(
+    value: &str,
+) -> Result<ReceiptsLogPruneConfig, ReceiptsLogError> {
+    let mut config = BTreeMap::new();
+    // Split out each of the filters.
+    let filters = value.split(',');
+    for filter in filters {
+        let parts: Vec<&str> = filter.split(':').collect();
+        if parts.len() < 2 {
+            return Err(ReceiptsLogError::InvalidFilterFormat(filter.to_string()));
+        }
+        // Parse the address
+        let address = parts[0]
+            .parse::<Address>()
+            .map_err(|_| ReceiptsLogError::InvalidAddress(parts[0].to_string()))?;
+
+        // Parse the prune mode
+        let prune_mode = match parts[1] {
+            "full" => PruneMode::Full,
+            s if s.starts_with("distance") => {
+                if parts.len() < 3 {
+                    return Err(ReceiptsLogError::InvalidFilterFormat(filter.to_string()));
+                }
+                let distance =
+                    parts[2].parse::<u64>().map_err(ReceiptsLogError::InvalidDistance)?;
+                PruneMode::Distance(distance)
+            }
+            s if s.starts_with("before") => {
+                if parts.len() < 3 {
+                    return Err(ReceiptsLogError::InvalidFilterFormat(filter.to_string()));
+                }
+                let block_number =
+                    parts[2].parse::<u64>().map_err(ReceiptsLogError::InvalidBlockNumber)?;
+                PruneMode::Before(block_number)
+            }
+            _ => return Err(ReceiptsLogError::InvalidPruneMode(parts[1].to_string())),
+        };
+        config.insert(address, prune_mode);
+    }
+    Ok(ReceiptsLogPruneConfig(config))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy_primitives::address;
+    use clap::Parser;
+
+    /// A helper type to parse Args more easily
+    #[derive(Parser)]
+    struct CommandParser<T: Args> {
+        #[command(flatten)]
+        args: T,
+    }
+
+    #[test]
+    fn pruning_args_sanity_check() {
+        let args = CommandParser::<PruningArgs>::parse_from([
+            "reth",
+            "--prune.receiptslogfilter",
+            "0x0000000000000000000000000000000000000003:before:5000000",
+        ])
+        .args;
+        let mut config = ReceiptsLogPruneConfig::default();
+        config.0.insert(
+            address!("0x0000000000000000000000000000000000000003"),
+            PruneMode::Before(5000000),
+        );
+        assert_eq!(args.receipts_log_filter, Some(config));
+    }
+
+    #[test]
+    fn parse_receiptslogfilter() {
+        let default_args = PruningArgs::default();
+        let args = CommandParser::<PruningArgs>::parse_from(["reth"]).args;
+        assert_eq!(args, default_args);
+    }
+
+    #[test]
+    fn test_parse_receipts_log_filter() {
+        let filter1 = "0x0000000000000000000000000000000000000001:full";
+        let filter2 = "0x0000000000000000000000000000000000000002:distance:1000";
+        let filter3 = "0x0000000000000000000000000000000000000003:before:5000000";
+        let filters = [filter1, filter2, filter3].join(",");
+
+        // Args can be parsed.
+        let result = parse_receipts_log_filter(&filters);
+        assert!(result.is_ok());
+        let config = result.unwrap();
+        assert_eq!(config.0.len(), 3);
+
+        // Check that the args were parsed correctly.
+        let addr1: Address = "0x0000000000000000000000000000000000000001".parse().unwrap();
+        let addr2: Address = "0x0000000000000000000000000000000000000002".parse().unwrap();
+        let addr3: Address = "0x0000000000000000000000000000000000000003".parse().unwrap();
+
+        assert_eq!(config.0.get(&addr1), Some(&PruneMode::Full));
+        assert_eq!(config.0.get(&addr2), Some(&PruneMode::Distance(1000)));
+        assert_eq!(config.0.get(&addr3), Some(&PruneMode::Before(5000000)));
+    }
+
+    #[test]
+    fn test_parse_receipts_log_filter_invalid_filter_format() {
+        let result = parse_receipts_log_filter("invalid_format");
+        assert!(matches!(result, Err(ReceiptsLogError::InvalidFilterFormat(_))));
+    }
+
+    #[test]
+    fn test_parse_receipts_log_filter_invalid_address() {
+        let result = parse_receipts_log_filter("invalid_address:full");
+        assert!(matches!(result, Err(ReceiptsLogError::InvalidAddress(_))));
+    }
+
+    #[test]
+    fn test_parse_receipts_log_filter_invalid_prune_mode() {
+        let result =
+            parse_receipts_log_filter("0x0000000000000000000000000000000000000000:invalid_mode");
+        assert!(matches!(result, Err(ReceiptsLogError::InvalidPruneMode(_))));
+    }
+
+    #[test]
+    fn test_parse_receipts_log_filter_invalid_distance() {
+        let result = parse_receipts_log_filter(
+            "0x0000000000000000000000000000000000000000:distance:invalid_distance",
+        );
+        assert!(matches!(result, Err(ReceiptsLogError::InvalidDistance(_))));
+    }
+
+    #[test]
+    fn test_parse_receipts_log_filter_invalid_block_number() {
+        let result = parse_receipts_log_filter(
+            "0x0000000000000000000000000000000000000000:before:invalid_block",
+        );
+        assert!(matches!(result, Err(ReceiptsLogError::InvalidBlockNumber(_))));
+    }
+}

crates/optimism/chainspec/res/superchain_registry_commit

@@ -1 +1 @@
-9e3f71cee0e4e2acb4864cb00f5fbee3555d8e9f
+59e22d265b7a423b7f51a67a722471a6f3c3cc39

crates/optimism/payload/src/payload.rs

@@ -342,6 +342,9 @@ where
 /// Generates the payload id for the configured payload from the [`OpPayloadAttributes`].
 ///
 /// Returns an 8-byte identifier by hashing the payload components with sha256 hash.
+///
+/// Note: This must be updated whenever the [`OpPayloadAttributes`] changes for a hardfork.
+/// See also <https://github.com/ethereum-optimism/op-geth/blob/d401af16f2dd94b010a72eaef10e07ac10b31931/miner/payload_building.go#L59-L59>
 pub fn payload_id_optimism(
     parent: &B256,
     attributes: &OpPayloadAttributes,
@@ -387,6 +390,10 @@ pub fn payload_id_optimism(
         hasher.update(eip_1559_params.as_slice());
     }
 
+    if let Some(min_base_fee) = attributes.min_base_fee {
+        hasher.update(min_base_fee.to_be_bytes());
+    }
+
     let mut out = hasher.finalize();
     out[0] = payload_version;
     PayloadId::new(out.as_slice()[..8].try_into().expect("sufficient length"))
@@ -473,6 +480,37 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_payload_id_parity_op_geth_jovian() {
+        // <https://github.com/ethereum-optimism/op-geth/compare/optimism...mattsse:op-geth:matt/check-payload-id-equality>
+        let expected =
+            PayloadId::new(FixedBytes::<8>::from_str("0x046c65ffc4d659ec").unwrap().into());
+        let attrs = OpPayloadAttributes {
+            payload_attributes: PayloadAttributes {
+                timestamp: 1728933301,
+                prev_randao: b256!("0x9158595abbdab2c90635087619aa7042bbebe47642dfab3c9bfb934f6b082765"),
+                suggested_fee_recipient: address!("0x4200000000000000000000000000000000000011"),
+                withdrawals: Some([].into()),
+                parent_beacon_block_root: b256!("0x8fe0193b9bf83cb7e5a08538e494fecc23046aab9a497af3704f4afdae3250ff").into(),
+            },
+            transactions: Some([bytes!("7ef8f8a0dc19cfa777d90980e4875d0a548a881baaa3f83f14d1bc0d3038bc329350e54194deaddeaddeaddeaddeaddeaddeaddeaddead00019442000000000000000000000000000000000000158080830f424080b8a4440a5e20000f424000000000000000000000000300000000670d6d890000000000000125000000000000000000000000000000000000000000000000000000000000000700000000000000000000000000000000000000000000000000000000000000014bf9181db6e381d4384bbf69c48b0ee0eed23c6ca26143c6d2544f9d39997a590000000000000000000000007f83d659683caf2767fd3c720981d51f5bc365bc")].into()),
+            no_tx_pool: None,
+            gas_limit: Some(30000000),
+            eip_1559_params: None,
+            min_base_fee: Some(100),
+        };
+
+        // Reth's `PayloadId` should match op-geth's `PayloadId`. This fails
+        assert_eq!(
+            expected,
+            payload_id_optimism(
+                &b256!("0x3533bf30edaf9505d0810bf475cbe4e5f4b9889904b9845e83efdeab4e92eb1e"),
+                &attrs,
+                EngineApiMessageVersion::V4 as u8
+            )
+        );
+    }
+
     #[test]
     fn test_get_extra_data_post_holocene() {
         let attributes: OpPayloadBuilderAttributes<OpTransactionSigned> =

crates/prune/prune/Cargo.toml

@@ -24,6 +24,7 @@ reth-primitives-traits.workspace = true
 reth-static-file-types.workspace = true
 
 # ethereum
+alloy-consensus.workspace = true
 alloy-eips.workspace = true
 
 # metrics

crates/prune/prune/src/builder.rs

@@ -43,7 +43,7 @@ impl PrunerBuilder {
     }
 
     /// Sets the configuration for every part of the data that can be pruned.
-    pub const fn segments(mut self, segments: PruneModes) -> Self {
+    pub fn segments(mut self, segments: PruneModes) -> Self {
         self.segments = segments;
         self
     }

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

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

crates/prune/prune/src/segments/set.rs

@@ -1,6 +1,6 @@
 use crate::segments::{
-    AccountHistory, Bodies, MerkleChangeSets, Segment, SenderRecovery, StorageHistory,
-    TransactionLookup, UserReceipts,
+    user::ReceiptsByLogs, AccountHistory, Bodies, MerkleChangeSets, Segment, SenderRecovery,
+    StorageHistory, TransactionLookup, UserReceipts,
 };
 use alloy_eips::eip2718::Encodable2718;
 use reth_db_api::{table::Value, transaction::DbTxMut};
@@ -59,7 +59,6 @@ where
         _static_file_provider: StaticFileProvider<Provider::Primitives>,
         prune_modes: PruneModes,
     ) -> Self {
-        #[expect(deprecated)]
         let PruneModes {
             sender_recovery,
             transaction_lookup,
@@ -68,7 +67,7 @@ where
             storage_history,
             bodies_history,
             merkle_changesets,
-            receipts_log_filter: (),
+            receipts_log_filter,
         } = prune_modes;
 
         Self::default()
@@ -82,6 +81,11 @@ where
             .segment_opt(storage_history.map(StorageHistory::new))
             // User receipts
             .segment_opt(receipts.map(UserReceipts::new))
+            // Receipts by logs
+            .segment_opt(
+                (!receipts_log_filter.is_empty())
+                    .then(|| ReceiptsByLogs::new(receipts_log_filter.clone())),
+            )
             // Transaction lookup
             .segment_opt(transaction_lookup.map(TransactionLookup::new))
             // Sender recovery

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

@@ -3,6 +3,7 @@ mod bodies;
 mod history;
 mod merkle_change_sets;
 mod receipts;
+mod receipts_by_logs;
 mod sender_recovery;
 mod storage_history;
 mod transaction_lookup;
@@ -11,6 +12,7 @@ pub use account_history::AccountHistory;
 pub use bodies::Bodies;
 pub use merkle_change_sets::MerkleChangeSets;
 pub use receipts::Receipts;
+pub use receipts_by_logs::ReceiptsByLogs;
 pub use sender_recovery::SenderRecovery;
 pub use storage_history::StorageHistory;
 pub use transaction_lookup::TransactionLookup;

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

@@ -0,0 +1,362 @@
+use crate::{
+    db_ext::DbTxPruneExt,
+    segments::{PruneInput, Segment},
+    PrunerError,
+};
+use alloy_consensus::TxReceipt;
+use reth_db_api::{table::Value, tables, transaction::DbTxMut};
+use reth_primitives_traits::NodePrimitives;
+use reth_provider::{
+    BlockReader, DBProvider, NodePrimitivesProvider, PruneCheckpointWriter, TransactionsProvider,
+};
+use reth_prune_types::{
+    PruneCheckpoint, PruneMode, PrunePurpose, PruneSegment, ReceiptsLogPruneConfig, SegmentOutput,
+    MINIMUM_PRUNING_DISTANCE,
+};
+use tracing::{instrument, trace};
+#[derive(Debug)]
+pub struct ReceiptsByLogs {
+    config: ReceiptsLogPruneConfig,
+}
+
+impl ReceiptsByLogs {
+    pub const fn new(config: ReceiptsLogPruneConfig) -> Self {
+        Self { config }
+    }
+}
+
+impl<Provider> Segment<Provider> for ReceiptsByLogs
+where
+    Provider: DBProvider<Tx: DbTxMut>
+        + PruneCheckpointWriter
+        + TransactionsProvider
+        + BlockReader
+        + NodePrimitivesProvider<Primitives: NodePrimitives<Receipt: Value>>,
+{
+    fn segment(&self) -> PruneSegment {
+        PruneSegment::ContractLogs
+    }
+
+    fn mode(&self) -> Option<PruneMode> {
+        None
+    }
+
+    fn purpose(&self) -> PrunePurpose {
+        PrunePurpose::User
+    }
+
+    #[instrument(target = "pruner", skip(self, provider), ret(level = "trace"))]
+    fn prune(&self, provider: &Provider, input: PruneInput) -> Result<SegmentOutput, PrunerError> {
+        // Contract log filtering removes every receipt possible except the ones in the list. So,
+        // for the other receipts it's as if they had a `PruneMode::Distance()` of
+        // `MINIMUM_PRUNING_DISTANCE`.
+        let to_block = PruneMode::Distance(MINIMUM_PRUNING_DISTANCE)
+            .prune_target_block(input.to_block, PruneSegment::ContractLogs, PrunePurpose::User)?
+            .map(|(bn, _)| bn)
+            .unwrap_or_default();
+
+        // Get status checkpoint from latest run
+        let mut last_pruned_block =
+            input.previous_checkpoint.and_then(|checkpoint| checkpoint.block_number);
+
+        let initial_last_pruned_block = last_pruned_block;
+
+        let mut from_tx_number = match initial_last_pruned_block {
+            Some(block) => provider
+                .block_body_indices(block)?
+                .map(|block| block.last_tx_num() + 1)
+                .unwrap_or(0),
+            None => 0,
+        };
+
+        // Figure out what receipts have already been pruned, so we can have an accurate
+        // `address_filter`
+        let address_filter = self.config.group_by_block(input.to_block, last_pruned_block)?;
+
+        // Splits all transactions in different block ranges. Each block range will have its own
+        // filter address list and will check it while going through the table
+        //
+        // Example:
+        // For an `address_filter` such as:
+        // { block9: [a1, a2], block20: [a3, a4, a5] }
+        //
+        // The following structures will be created in the exact order as showed:
+        // `block_ranges`: [
+        //    (block0, block8, 0 addresses),
+        //    (block9, block19, 2 addresses),
+        //    (block20, to_block, 5 addresses)
+        //  ]
+        // `filtered_addresses`: [a1, a2, a3, a4, a5]
+        //
+        // The first range will delete all receipts between block0 - block8
+        // The second range will delete all receipts between block9 - 19, except the ones with
+        //     emitter logs from these addresses: [a1, a2].
+        // The third range will delete all receipts between block20 - to_block, except the ones with
+        //     emitter logs from these addresses: [a1, a2, a3, a4, a5]
+        let mut block_ranges = vec![];
+        let mut blocks_iter = address_filter.iter().peekable();
+        let mut filtered_addresses = vec![];
+
+        while let Some((start_block, addresses)) = blocks_iter.next() {
+            filtered_addresses.extend_from_slice(addresses);
+
+            // This will clear all receipts before the first  appearance of a contract log or since
+            // the block after the last pruned one.
+            if block_ranges.is_empty() {
+                let init = last_pruned_block.map(|b| b + 1).unwrap_or_default();
+                if init < *start_block {
+                    block_ranges.push((init, *start_block - 1, 0));
+                }
+            }
+
+            let end_block =
+                blocks_iter.peek().map(|(next_block, _)| *next_block - 1).unwrap_or(to_block);
+
+            // Addresses in lower block ranges, are still included in the inclusion list for future
+            // ranges.
+            block_ranges.push((*start_block, end_block, filtered_addresses.len()));
+        }
+
+        trace!(
+            target: "pruner",
+            ?block_ranges,
+            ?filtered_addresses,
+            "Calculated block ranges and filtered addresses",
+        );
+
+        let mut limiter = input.limiter;
+
+        let mut done = true;
+        let mut pruned = 0;
+        let mut last_pruned_transaction = None;
+        for (start_block, end_block, num_addresses) in block_ranges {
+            let block_range = start_block..=end_block;
+
+            // Calculate the transaction range from this block range
+            let tx_range_end = match provider.block_body_indices(end_block)? {
+                Some(body) => body.last_tx_num(),
+                None => {
+                    trace!(
+                        target: "pruner",
+                        ?block_range,
+                        "No receipts to prune."
+                    );
+                    continue
+                }
+            };
+            let tx_range = from_tx_number..=tx_range_end;
+
+            // Delete receipts, except the ones in the inclusion list
+            let mut last_skipped_transaction = 0;
+            let deleted;
+            (deleted, done) = provider.tx_ref().prune_table_with_range::<tables::Receipts<
+                <Provider::Primitives as NodePrimitives>::Receipt,
+            >>(
+                tx_range,
+                &mut limiter,
+                |(tx_num, receipt)| {
+                    let skip = num_addresses > 0 &&
+                        receipt.logs().iter().any(|log| {
+                            filtered_addresses[..num_addresses].contains(&&log.address)
+                        });
+
+                    if skip {
+                        last_skipped_transaction = *tx_num;
+                    }
+                    skip
+                },
+                |row| last_pruned_transaction = Some(row.0),
+            )?;
+
+            trace!(target: "pruner", %deleted, %done, ?block_range, "Pruned receipts");
+
+            pruned += deleted;
+
+            // For accurate checkpoints we need to know that we have checked every transaction.
+            // Example: we reached the end of the range, and the last receipt is supposed to skip
+            // its deletion.
+            let last_pruned_transaction = *last_pruned_transaction
+                .insert(last_pruned_transaction.unwrap_or_default().max(last_skipped_transaction));
+
+            last_pruned_block = Some(
+                provider
+                    .block_by_transaction_id(last_pruned_transaction)?
+                    .ok_or(PrunerError::InconsistentData("Block for transaction is not found"))?
+                    // If there's more receipts to prune, set the checkpoint block number to
+                    // previous, so we could finish pruning its receipts on the
+                    // next run.
+                    .saturating_sub(if done { 0 } else { 1 }),
+            );
+
+            if limiter.is_limit_reached() {
+                done &= end_block == to_block;
+                break
+            }
+
+            from_tx_number = last_pruned_transaction + 1;
+        }
+
+        // If there are contracts using `PruneMode::Distance(_)` there will be receipts before
+        // `to_block` that become eligible to be pruned in future runs. Therefore, our checkpoint is
+        // not actually `to_block`, but the `lowest_block_with_distance` from any contract.
+        // This ensures that in future pruner runs we can prune all these receipts between the
+        // previous `lowest_block_with_distance` and the new one using
+        // `get_next_tx_num_range_from_checkpoint`.
+        //
+        // Only applies if we were able to prune everything intended for this run, otherwise the
+        // checkpoint is the `last_pruned_block`.
+        let prune_mode_block = self
+            .config
+            .lowest_block_with_distance(input.to_block, initial_last_pruned_block)?
+            .unwrap_or(to_block);
+
+        provider.save_prune_checkpoint(
+            PruneSegment::ContractLogs,
+            PruneCheckpoint {
+                block_number: Some(prune_mode_block.min(last_pruned_block.unwrap_or(u64::MAX))),
+                tx_number: last_pruned_transaction,
+                prune_mode: PruneMode::Before(prune_mode_block),
+            },
+        )?;
+
+        let progress = limiter.progress(done);
+
+        Ok(SegmentOutput { progress, pruned, checkpoint: None })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::segments::{user::ReceiptsByLogs, PruneInput, PruneLimiter, Segment};
+    use alloy_primitives::B256;
+    use assert_matches::assert_matches;
+    use reth_db_api::{cursor::DbCursorRO, tables, transaction::DbTx};
+    use reth_primitives_traits::InMemorySize;
+    use reth_provider::{BlockReader, DBProvider, DatabaseProviderFactory, PruneCheckpointReader};
+    use reth_prune_types::{PruneMode, PruneSegment, ReceiptsLogPruneConfig};
+    use reth_stages::test_utils::{StorageKind, TestStageDB};
+    use reth_testing_utils::generators::{
+        self, random_block_range, random_eoa_account, random_log, random_receipt, BlockRangeParams,
+    };
+    use std::collections::BTreeMap;
+
+    #[test]
+    fn prune_receipts_by_logs() {
+        reth_tracing::init_test_tracing();
+
+        let db = TestStageDB::default();
+        let mut rng = generators::rng();
+
+        let tip = 20000;
+        let blocks = [
+            random_block_range(
+                &mut rng,
+                0..=100,
+                BlockRangeParams { parent: Some(B256::ZERO), tx_count: 1..5, ..Default::default() },
+            ),
+            random_block_range(
+                &mut rng,
+                (100 + 1)..=(tip - 100),
+                BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
+            ),
+            random_block_range(
+                &mut rng,
+                (tip - 100 + 1)..=tip,
+                BlockRangeParams { parent: Some(B256::ZERO), tx_count: 1..5, ..Default::default() },
+            ),
+        ]
+        .concat();
+        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
+
+        let mut receipts = Vec::new();
+
+        let (deposit_contract_addr, _) = random_eoa_account(&mut rng);
+        for block in &blocks {
+            receipts.reserve_exact(block.body().size());
+            for (txi, transaction) in block.body().transactions.iter().enumerate() {
+                let mut receipt = random_receipt(&mut rng, transaction, Some(1), None);
+                receipt.logs.push(random_log(
+                    &mut rng,
+                    (txi == (block.transaction_count() - 1)).then_some(deposit_contract_addr),
+                    Some(1),
+                ));
+                receipts.push((receipts.len() as u64, receipt));
+            }
+        }
+        db.insert_receipts(receipts).expect("insert receipts");
+
+        assert_eq!(
+            db.table::<tables::Transactions>().unwrap().len(),
+            blocks.iter().map(|block| block.transaction_count()).sum::<usize>()
+        );
+        assert_eq!(
+            db.table::<tables::Transactions>().unwrap().len(),
+            db.table::<tables::Receipts>().unwrap().len()
+        );
+
+        let run_prune = || {
+            let provider = db.factory.database_provider_rw().unwrap();
+
+            let prune_before_block: usize = 20;
+            let prune_mode = PruneMode::Before(prune_before_block as u64);
+            let receipts_log_filter =
+                ReceiptsLogPruneConfig(BTreeMap::from([(deposit_contract_addr, prune_mode)]));
+
+            let limiter = PruneLimiter::default().set_deleted_entries_limit(10);
+
+            let result = ReceiptsByLogs::new(receipts_log_filter).prune(
+                &provider,
+                PruneInput {
+                    previous_checkpoint: db
+                        .factory
+                        .provider()
+                        .unwrap()
+                        .get_prune_checkpoint(PruneSegment::ContractLogs)
+                        .unwrap(),
+                    to_block: tip,
+                    limiter,
+                },
+            );
+            provider.commit().expect("commit");
+
+            assert_matches!(result, Ok(_));
+            let output = result.unwrap();
+
+            let (pruned_block, pruned_tx) = db
+                .factory
+                .provider()
+                .unwrap()
+                .get_prune_checkpoint(PruneSegment::ContractLogs)
+                .unwrap()
+                .map(|checkpoint| (checkpoint.block_number.unwrap(), checkpoint.tx_number.unwrap()))
+                .unwrap_or_default();
+
+            // All receipts are in the end of the block
+            let unprunable = pruned_block.saturating_sub(prune_before_block as u64 - 1);
+
+            assert_eq!(
+                db.table::<tables::Receipts>().unwrap().len(),
+                blocks.iter().map(|block| block.transaction_count()).sum::<usize>() -
+                    ((pruned_tx + 1) - unprunable) as usize
+            );
+
+            output.progress.is_finished()
+        };
+
+        while !run_prune() {}
+
+        let provider = db.factory.provider().unwrap();
+        let mut cursor = provider.tx_ref().cursor_read::<tables::Receipts>().unwrap();
+        let walker = cursor.walk(None).unwrap();
+        for receipt in walker {
+            let (tx_num, receipt) = receipt.unwrap();
+
+            // Either we only find our contract, or the receipt is part of the unprunable receipts
+            // set by tip - 128
+            assert!(
+                receipt.logs.iter().any(|l| l.address == deposit_contract_addr) ||
+                    provider.block_by_transaction_id(tx_num).unwrap().unwrap() > tip - 128,
+            );
+        }
+    }
+}

crates/prune/types/src/lib.rs

@@ -18,6 +18,10 @@ mod pruner;
 mod segment;
 mod target;
 
+use alloc::{collections::BTreeMap, vec::Vec};
+use alloy_primitives::{Address, BlockNumber};
+use core::ops::Deref;
+
 pub use checkpoint::PruneCheckpoint;
 pub use event::PrunerEvent;
 pub use mode::PruneMode;
@@ -26,4 +30,304 @@ pub use pruner::{
     SegmentOutputCheckpoint,
 };
 pub use segment::{PrunePurpose, PruneSegment, PruneSegmentError};
-pub use target::{PruneModes, UnwindTargetPrunedError, MINIMUM_PRUNING_DISTANCE};
+pub use target::{
+    PruneModes, UnwindTargetPrunedError, MERKLE_CHANGESETS_RETENTION_BLOCKS,
+    MINIMUM_PRUNING_DISTANCE,
+};
+
+/// Configuration for pruning receipts not associated with logs emitted by the specified contracts.
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+#[cfg_attr(any(test, feature = "serde"), derive(serde::Serialize, serde::Deserialize))]
+pub struct ReceiptsLogPruneConfig(pub BTreeMap<Address, PruneMode>);
+
+impl ReceiptsLogPruneConfig {
+    /// Checks if the configuration is empty
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
+    /// Given the `tip` block number, consolidates the structure so it can easily be queried for
+    /// filtering across a range of blocks.
+    ///
+    /// Example:
+    ///
+    /// `{ addrA: Before(872), addrB: Before(500), addrC: Distance(128) }`
+    ///
+    ///    for `tip: 1000`, gets transformed to a map such as:
+    ///
+    /// `{ 500: [addrB], 872: [addrA, addrC] }`
+    ///
+    /// The [`BlockNumber`] key of the new map should be viewed as `PruneMode::Before(block)`, which
+    /// makes the previous result equivalent to
+    ///
+    /// `{ Before(500): [addrB], Before(872): [addrA, addrC] }`
+    pub fn group_by_block(
+        &self,
+        tip: BlockNumber,
+        pruned_block: Option<BlockNumber>,
+    ) -> Result<BTreeMap<BlockNumber, Vec<&Address>>, PruneSegmentError> {
+        let mut map = BTreeMap::new();
+        let base_block = pruned_block.unwrap_or_default() + 1;
+
+        for (address, mode) in &self.0 {
+            // Getting `None`, means that there is nothing to prune yet, so we need it to include in
+            // the BTreeMap (block = 0), otherwise it will be excluded.
+            // Reminder that this BTreeMap works as an inclusion list that excludes (prunes) all
+            // other receipts.
+            //
+            // Reminder, that we increment because the [`BlockNumber`] key of the new map should be
+            // viewed as `PruneMode::Before(block)`
+            let block = base_block.max(
+                mode.prune_target_block(tip, PruneSegment::ContractLogs, PrunePurpose::User)?
+                    .map(|(block, _)| block)
+                    .unwrap_or_default() +
+                    1,
+            );
+
+            map.entry(block).or_insert_with(Vec::new).push(address)
+        }
+        Ok(map)
+    }
+
+    /// Returns the lowest block where we start filtering logs which use `PruneMode::Distance(_)`.
+    pub fn lowest_block_with_distance(
+        &self,
+        tip: BlockNumber,
+        pruned_block: Option<BlockNumber>,
+    ) -> Result<Option<BlockNumber>, PruneSegmentError> {
+        let pruned_block = pruned_block.unwrap_or_default();
+        let mut lowest = None;
+
+        for mode in self.values() {
+            if mode.is_distance() &&
+                let Some((block, _)) =
+                    mode.prune_target_block(tip, PruneSegment::ContractLogs, PrunePurpose::User)?
+            {
+                lowest = Some(lowest.unwrap_or(u64::MAX).min(block));
+            }
+        }
+
+        Ok(lowest.map(|lowest| lowest.max(pruned_block)))
+    }
+}
+
+impl Deref for ReceiptsLogPruneConfig {
+    type Target = BTreeMap<Address, PruneMode>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_group_by_block_empty_config() {
+        let config = ReceiptsLogPruneConfig(BTreeMap::new());
+        let tip = 1000;
+        let pruned_block = None;
+
+        let result = config.group_by_block(tip, pruned_block).unwrap();
+        assert!(result.is_empty(), "The result should be empty when the config is empty");
+    }
+
+    #[test]
+    fn test_group_by_block_single_entry() {
+        let mut config_map = BTreeMap::new();
+        let address = Address::new([1; 20]);
+        let prune_mode = PruneMode::Before(500);
+        config_map.insert(address, prune_mode);
+
+        let config = ReceiptsLogPruneConfig(config_map);
+        // Big tip to have something to prune for the target block
+        let tip = 3000000;
+        let pruned_block = Some(400);
+
+        let result = config.group_by_block(tip, pruned_block).unwrap();
+
+        // Expect one entry with block 500 and the corresponding address
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[&500], vec![&address], "Address should be grouped under block 500");
+
+        // Tip smaller than the target block, so that we have nothing to prune for the block
+        let tip = 300;
+        let pruned_block = Some(400);
+
+        let result = config.group_by_block(tip, pruned_block).unwrap();
+
+        // Expect one entry with block 400 and the corresponding address
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[&401], vec![&address], "Address should be grouped under block 400");
+    }
+
+    #[test]
+    fn test_group_by_block_multiple_entries() {
+        let mut config_map = BTreeMap::new();
+        let address1 = Address::new([1; 20]);
+        let address2 = Address::new([2; 20]);
+        let prune_mode1 = PruneMode::Before(600);
+        let prune_mode2 = PruneMode::Before(800);
+        config_map.insert(address1, prune_mode1);
+        config_map.insert(address2, prune_mode2);
+
+        let config = ReceiptsLogPruneConfig(config_map);
+        let tip = 900000;
+        let pruned_block = Some(400);
+
+        let result = config.group_by_block(tip, pruned_block).unwrap();
+
+        // Expect two entries: one for block 600 and another for block 800
+        assert_eq!(result.len(), 2);
+        assert_eq!(result[&600], vec![&address1], "Address1 should be grouped under block 600");
+        assert_eq!(result[&800], vec![&address2], "Address2 should be grouped under block 800");
+    }
+
+    #[test]
+    fn test_group_by_block_with_distance_prune_mode() {
+        let mut config_map = BTreeMap::new();
+        let address = Address::new([1; 20]);
+        let prune_mode = PruneMode::Distance(100000);
+        config_map.insert(address, prune_mode);
+
+        let config = ReceiptsLogPruneConfig(config_map);
+        let tip = 100100;
+        // Pruned block is smaller than the target block
+        let pruned_block = Some(50);
+
+        let result = config.group_by_block(tip, pruned_block).unwrap();
+
+        // Expect the entry to be grouped under block 100 (tip - distance)
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[&101], vec![&address], "Address should be grouped under block 100");
+
+        let tip = 100100;
+        // Pruned block is larger than the target block
+        let pruned_block = Some(800);
+
+        let result = config.group_by_block(tip, pruned_block).unwrap();
+
+        // Expect the entry to be grouped under block 800 which is larger than tip - distance
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[&801], vec![&address], "Address should be grouped under block 800");
+    }
+
+    #[test]
+    fn test_lowest_block_with_distance_empty_config() {
+        let config = ReceiptsLogPruneConfig(BTreeMap::new());
+        let tip = 1000;
+        let pruned_block = None;
+
+        let result = config.lowest_block_with_distance(tip, pruned_block).unwrap();
+        assert_eq!(result, None, "The result should be None when the config is empty");
+    }
+
+    #[test]
+    fn test_lowest_block_with_distance_no_distance_mode() {
+        let mut config_map = BTreeMap::new();
+        let address = Address::new([1; 20]);
+        let prune_mode = PruneMode::Before(500);
+        config_map.insert(address, prune_mode);
+
+        let config = ReceiptsLogPruneConfig(config_map);
+        let tip = 1000;
+        let pruned_block = None;
+
+        let result = config.lowest_block_with_distance(tip, pruned_block).unwrap();
+        assert_eq!(result, None, "The result should be None when there are no Distance modes");
+    }
+
+    #[test]
+    fn test_lowest_block_with_distance_single_entry() {
+        let mut config_map = BTreeMap::new();
+        let address = Address::new([1; 20]);
+        let prune_mode = PruneMode::Distance(100000);
+        config_map.insert(address, prune_mode);
+
+        let config = ReceiptsLogPruneConfig(config_map);
+
+        let tip = 100100;
+        let pruned_block = Some(400);
+
+        // Expect the lowest block to be 400 as 400 > 100100 - 100000 (tip - distance)
+        assert_eq!(
+            config.lowest_block_with_distance(tip, pruned_block).unwrap(),
+            Some(400),
+            "The lowest block should be 400"
+        );
+
+        let tip = 100100;
+        let pruned_block = Some(50);
+
+        // Expect the lowest block to be 100 as 100 > 50 (pruned block)
+        assert_eq!(
+            config.lowest_block_with_distance(tip, pruned_block).unwrap(),
+            Some(100),
+            "The lowest block should be 100"
+        );
+    }
+
+    #[test]
+    fn test_lowest_block_with_distance_multiple_entries_last() {
+        let mut config_map = BTreeMap::new();
+        let address1 = Address::new([1; 20]);
+        let address2 = Address::new([2; 20]);
+        let prune_mode1 = PruneMode::Distance(100100);
+        let prune_mode2 = PruneMode::Distance(100300);
+        config_map.insert(address1, prune_mode1);
+        config_map.insert(address2, prune_mode2);
+
+        let config = ReceiptsLogPruneConfig(config_map);
+        let tip = 200300;
+        let pruned_block = Some(100);
+
+        // The lowest block should be 200300 - 100300 = 100000:
+        // - First iteration will return 100200 => 200300 - 100100 = 100200
+        // - Second iteration will return 100000 => 200300 - 100300 = 100000 < 100200
+        // - Final result is 100000
+        assert_eq!(config.lowest_block_with_distance(tip, pruned_block).unwrap(), Some(100000));
+    }
+
+    #[test]
+    fn test_lowest_block_with_distance_multiple_entries_first() {
+        let mut config_map = BTreeMap::new();
+        let address1 = Address::new([1; 20]);
+        let address2 = Address::new([2; 20]);
+        let prune_mode1 = PruneMode::Distance(100400);
+        let prune_mode2 = PruneMode::Distance(100300);
+        config_map.insert(address1, prune_mode1);
+        config_map.insert(address2, prune_mode2);
+
+        let config = ReceiptsLogPruneConfig(config_map);
+        let tip = 200300;
+        let pruned_block = Some(100);
+
+        // The lowest block should be 200300 - 100400 = 99900:
+        // - First iteration, lowest block is 200300 - 100400 = 99900
+        // - Second iteration, lowest block is still 99900 < 200300 - 100300 = 100000
+        // - Final result is 99900
+        assert_eq!(config.lowest_block_with_distance(tip, pruned_block).unwrap(), Some(99900));
+    }
+
+    #[test]
+    fn test_lowest_block_with_distance_multiple_entries_pruned_block() {
+        let mut config_map = BTreeMap::new();
+        let address1 = Address::new([1; 20]);
+        let address2 = Address::new([2; 20]);
+        let prune_mode1 = PruneMode::Distance(100400);
+        let prune_mode2 = PruneMode::Distance(100300);
+        config_map.insert(address1, prune_mode1);
+        config_map.insert(address2, prune_mode2);
+
+        let config = ReceiptsLogPruneConfig(config_map);
+        let tip = 200300;
+        let pruned_block = Some(100000);
+
+        // The lowest block should be 100000 because:
+        // - Lowest is 200300 - 100400 = 99900 < 200300 - 100300 = 100000
+        // - Lowest is compared to the pruned block 100000: 100000 > 99900
+        // - Finally the lowest block is 100000
+        assert_eq!(config.lowest_block_with_distance(tip, pruned_block).unwrap(), Some(100000));
+    }
+}

crates/prune/types/src/segment.rs

@@ -1,6 +1,6 @@
 #![allow(deprecated)] // necessary to all defining deprecated `PruneSegment` variants
 
-use crate::MINIMUM_PRUNING_DISTANCE;
+use crate::{MERKLE_CHANGESETS_RETENTION_BLOCKS, MINIMUM_PRUNING_DISTANCE};
 use derive_more::Display;
 use strum::{EnumIter, IntoEnumIterator};
 use thiserror::Error;
@@ -68,9 +68,9 @@ impl PruneSegment {
             Self::ContractLogs |
             Self::AccountHistory |
             Self::StorageHistory |
-            Self::MerkleChangeSets |
             Self::Bodies |
             Self::Receipts => MINIMUM_PRUNING_DISTANCE,
+            Self::MerkleChangeSets => MERKLE_CHANGESETS_RETENTION_BLOCKS,
             #[expect(deprecated)]
             #[expect(clippy::match_same_arms)]
             Self::Headers | Self::Transactions => 0,

crates/prune/types/src/target.rs

@@ -2,7 +2,7 @@ use alloy_primitives::BlockNumber;
 use derive_more::Display;
 use thiserror::Error;
 
-use crate::{PruneCheckpoint, PruneMode, PruneSegment};
+use crate::{PruneCheckpoint, PruneMode, PruneSegment, ReceiptsLogPruneConfig};
 
 /// Minimum distance from the tip necessary for the node to work correctly:
 /// 1. Minimum 2 epochs (32 blocks per epoch) required to handle any reorg according to the
@@ -36,9 +36,13 @@ pub enum HistoryType {
     StorageHistory,
 }
 
+/// Default number of blocks to retain for merkle changesets.
+/// This is used by both the `MerkleChangeSets` stage and the pruner segment.
+pub const MERKLE_CHANGESETS_RETENTION_BLOCKS: u64 = 128;
+
 /// Default pruning mode for merkle changesets
 const fn default_merkle_changesets_mode() -> PruneMode {
-    PruneMode::Distance(MINIMUM_PRUNING_DISTANCE)
+    PruneMode::Distance(MERKLE_CHANGESETS_RETENTION_BLOCKS)
 }
 
 /// Pruning configuration for every segment of the data that can be pruned.
@@ -91,18 +95,18 @@ pub struct PruneModes {
     pub bodies_history: Option<PruneMode>,
     /// Merkle Changesets pruning configuration for `AccountsTrieChangeSets` and
     /// `StoragesTrieChangeSets`.
+    #[cfg_attr(any(test, feature = "serde"), serde(default = "default_merkle_changesets_mode"))]
+    pub merkle_changesets: PruneMode,
+    /// Receipts pruning configuration by retaining only those receipts that contain logs emitted
+    /// by the specified addresses, discarding others. This setting is overridden by `receipts`.
+    ///
+    /// The [`BlockNumber`](`crate::BlockNumber`) represents the starting block from which point
+    /// onwards the receipts are preserved.
     #[cfg_attr(
         any(test, feature = "serde"),
-        serde(
-            default = "default_merkle_changesets_mode",
-            deserialize_with = "deserialize_prune_mode_with_min_blocks::<MINIMUM_PRUNING_DISTANCE, _>"
-        )
+        serde(skip_serializing_if = "ReceiptsLogPruneConfig::is_empty")
     )]
-    pub merkle_changesets: PruneMode,
-    /// Receipts log filtering has been deprecated and will be removed in a future release.
-    #[deprecated]
-    #[cfg_attr(any(test, feature = "serde"), serde(skip))]
-    pub receipts_log_filter: (),
+    pub receipts_log_filter: ReceiptsLogPruneConfig,
 }
 
 impl Default for PruneModes {
@@ -115,15 +119,14 @@ impl Default for PruneModes {
             storage_history: None,
             bodies_history: None,
             merkle_changesets: default_merkle_changesets_mode(),
-            #[expect(deprecated)]
-            receipts_log_filter: (),
+            receipts_log_filter: ReceiptsLogPruneConfig::default(),
         }
     }
 }
 
 impl PruneModes {
     /// Sets pruning to all targets.
-    pub const fn all() -> Self {
+    pub fn all() -> Self {
         Self {
             sender_recovery: Some(PruneMode::Full),
             transaction_lookup: Some(PruneMode::Full),
@@ -132,14 +135,29 @@ impl PruneModes {
             storage_history: Some(PruneMode::Full),
             bodies_history: Some(PruneMode::Full),
             merkle_changesets: PruneMode::Full,
-            #[expect(deprecated)]
-            receipts_log_filter: (),
+            receipts_log_filter: Default::default(),
         }
     }
 
     /// Returns whether there is any kind of receipt pruning configuration.
-    pub const fn has_receipts_pruning(&self) -> bool {
-        self.receipts.is_some()
+    pub fn has_receipts_pruning(&self) -> bool {
+        self.receipts.is_some() || !self.receipts_log_filter.is_empty()
+    }
+
+    /// Migrates deprecated prune mode values to their new defaults.
+    ///
+    /// Returns `true` if any migration was performed.
+    ///
+    /// Currently migrates:
+    /// - `merkle_changesets`: `Distance(n)` where `n < 128` or `n == 10064` -> `Distance(128)`
+    pub const fn migrate(&mut self) -> bool {
+        if let PruneMode::Distance(d) = self.merkle_changesets &&
+            (d < MERKLE_CHANGESETS_RETENTION_BLOCKS || d == MINIMUM_PRUNING_DISTANCE)
+        {
+            self.merkle_changesets = PruneMode::Distance(MERKLE_CHANGESETS_RETENTION_BLOCKS);
+            return true;
+        }
+        false
     }
 
     /// Returns an error if we can't unwind to the targeted block because the target block is
@@ -191,28 +209,6 @@ impl PruneModes {
     }
 }
 
-/// Deserializes [`PruneMode`] and validates that the value is not less than the const
-/// generic parameter `MIN_BLOCKS`. This parameter represents the number of blocks that needs to be
-/// left in database after the pruning.
-///
-/// 1. For [`PruneMode::Full`], it fails if `MIN_BLOCKS > 0`.
-/// 2. For [`PruneMode::Distance`], it fails if `distance < MIN_BLOCKS + 1`. `+ 1` is needed because
-///    `PruneMode::Distance(0)` means that we leave zero blocks from the latest, meaning we have one
-///    block in the database.
-#[cfg(any(test, feature = "serde"))]
-fn deserialize_prune_mode_with_min_blocks<
-    'de,
-    const MIN_BLOCKS: u64,
-    D: serde::Deserializer<'de>,
->(
-    deserializer: D,
-) -> Result<PruneMode, D::Error> {
-    use serde::Deserialize;
-    let prune_mode = PruneMode::deserialize(deserializer)?;
-    serde_deserialize_validate::<MIN_BLOCKS, D>(&prune_mode)?;
-    Ok(prune_mode)
-}
-
 /// Deserializes [`Option<PruneMode>`] and validates that the value is not less than the const
 /// generic parameter `MIN_BLOCKS`. This parameter represents the number of blocks that needs to be
 /// left in database after the pruning.

crates/stages/stages/src/stages/execution.rs

@@ -660,7 +660,7 @@ where
 mod tests {
     use super::*;
     use crate::{stages::MERKLE_STAGE_DEFAULT_REBUILD_THRESHOLD, test_utils::TestStageDB};
-    use alloy_primitives::{address, hex_literal::hex, keccak256, B256, U256};
+    use alloy_primitives::{address, hex_literal::hex, keccak256, Address, B256, U256};
     use alloy_rlp::Decodable;
     use assert_matches::assert_matches;
     use reth_chainspec::ChainSpecBuilder;
@@ -677,7 +677,9 @@ mod tests {
         DatabaseProviderFactory, ReceiptProvider, StaticFileProviderFactory,
     };
     use reth_prune::PruneModes;
+    use reth_prune_types::{PruneMode, ReceiptsLogPruneConfig};
     use reth_stages_api::StageUnitCheckpoint;
+    use std::collections::BTreeMap;
 
     fn stage() -> ExecutionStage<EthEvmConfig> {
         let evm_config =
@@ -894,11 +896,20 @@ mod tests {
         // If there is a pruning configuration, then it's forced to use the database.
         // This way we test both cases.
         let modes = [None, Some(PruneModes::default())];
+        let random_filter = ReceiptsLogPruneConfig(BTreeMap::from([(
+            Address::random(),
+            PruneMode::Distance(100000),
+        )]));
 
         // Tests node with database and node with static files
-        for mode in modes {
+        for mut mode in modes {
             let mut provider = factory.database_provider_rw().unwrap();
 
+            if let Some(mode) = &mut mode {
+                // Simulating a full node where we write receipts to database
+                mode.receipts_log_filter = random_filter.clone();
+            }
+
             let mut execution_stage = stage();
             provider.set_prune_modes(mode.clone().unwrap_or_default());
 
@@ -1022,9 +1033,18 @@ mod tests {
         // If there is a pruning configuration, then it's forced to use the database.
         // This way we test both cases.
         let modes = [None, Some(PruneModes::default())];
+        let random_filter = ReceiptsLogPruneConfig(BTreeMap::from([(
+            Address::random(),
+            PruneMode::Before(100000),
+        )]));
 
         // Tests node with database and node with static files
-        for mode in modes {
+        for mut mode in modes {
+            if let Some(mode) = &mut mode {
+                // Simulating a full node where we write receipts to database
+                mode.receipts_log_filter = random_filter.clone();
+            }
+
             // Test Execution
             let mut execution_stage = stage();
             provider.set_prune_modes(mode.clone().unwrap_or_default());

crates/stages/stages/src/stages/merkle_changesets.rs

@@ -7,7 +7,9 @@ use reth_provider::{
     ChainStateBlockReader, DBProvider, HeaderProvider, ProviderError, PruneCheckpointReader,
     PruneCheckpointWriter, StageCheckpointReader, TrieWriter,
 };
-use reth_prune_types::{PruneCheckpoint, PruneMode, PruneSegment};
+use reth_prune_types::{
+    PruneCheckpoint, PruneMode, PruneSegment, MERKLE_CHANGESETS_RETENTION_BLOCKS,
+};
 use reth_stages_api::{
     BlockErrorKind, ExecInput, ExecOutput, Stage, StageCheckpoint, StageError, StageId,
     UnwindInput, UnwindOutput,
@@ -23,14 +25,15 @@ use tracing::{debug, error};
 #[derive(Debug, Clone)]
 pub struct MerkleChangeSets {
     /// The number of blocks to retain changesets for, used as a fallback when the finalized block
-    /// is not found. Defaults to 64 (2 epochs in beacon chain).
+    /// is not found. Defaults to [`MERKLE_CHANGESETS_RETENTION_BLOCKS`] (2 epochs in beacon
+    /// chain).
     retention_blocks: u64,
 }
 
 impl MerkleChangeSets {
-    /// Creates a new `MerkleChangeSets` stage with default retention blocks of 64.
+    /// Creates a new `MerkleChangeSets` stage with the default retention blocks.
     pub const fn new() -> Self {
-        Self { retention_blocks: 64 }
+        Self { retention_blocks: MERKLE_CHANGESETS_RETENTION_BLOCKS }
     }
 
     /// Creates a new `MerkleChangeSets` stage with a custom finalized block height.

crates/static-file/static-file/src/static_file_producer.rs

@@ -194,7 +194,9 @@ where
 
         let targets = StaticFileTargets {
             // StaticFile receipts only if they're not pruned according to the user configuration
-            receipts: if self.prune_modes.receipts.is_none() {
+            receipts: if self.prune_modes.receipts.is_none() &&
+                self.prune_modes.receipts_log_filter.is_empty()
+            {
                 finalized_block_numbers.receipts.and_then(|finalized_block_number| {
                     self.get_static_file_target(
                         highest_static_files.receipts,

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

@@ -96,7 +96,7 @@ impl<N: NodeTypesWithDB> ProviderFactory<N> {
     }
 
     /// Sets the pruning configuration for an existing [`ProviderFactory`].
-    pub const fn with_prune_modes(mut self, prune_modes: PruneModes) -> Self {
+    pub fn with_prune_modes(mut self, prune_modes: PruneModes) -> Self {
         self.prune_modes = prune_modes;
         self
     }

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

@@ -22,7 +22,7 @@ use crate::{
 };
 use alloy_consensus::{
     transaction::{SignerRecoverable, TransactionMeta, TxHashRef},
-    BlockHeader,
+    BlockHeader, TxReceipt,
 };
 use alloy_eips::BlockHashOrNumber;
 use alloy_primitives::{
@@ -214,7 +214,7 @@ impl<TX: DbTx + 'static, N: NodeTypes> DatabaseProvider<TX, N> {
 
     #[cfg(feature = "test-utils")]
     /// Sets the prune modes for provider.
-    pub const fn set_prune_modes(&mut self, prune_modes: PruneModes) {
+    pub fn set_prune_modes(&mut self, prune_modes: PruneModes) {
         self.prune_modes = prune_modes;
     }
 }
@@ -1621,11 +1621,20 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
             .then(|| self.static_file_provider.get_writer(first_block, StaticFileSegment::Receipts))
             .transpose()?;
 
+        let has_contract_log_filter = !self.prune_modes.receipts_log_filter.is_empty();
+        let contract_log_pruner = self.prune_modes.receipts_log_filter.group_by_block(tip, None)?;
+
         // All receipts from the last 128 blocks are required for blockchain tree, even with
         // [`PruneSegment::ContractLogs`].
         let prunable_receipts =
             PruneMode::Distance(MINIMUM_PRUNING_DISTANCE).should_prune(first_block, tip);
 
+        // Prepare set of addresses which logs should not be pruned.
+        let mut allowed_addresses: HashSet<Address, _> = HashSet::new();
+        for (_, addresses) in contract_log_pruner.range(..first_block) {
+            allowed_addresses.extend(addresses.iter().copied());
+        }
+
         for (idx, (receipts, first_tx_index)) in
             execution_outcome.receipts.iter().zip(block_indices).enumerate()
         {
@@ -1645,8 +1654,21 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
                 continue
             }
 
+            // If there are new addresses to retain after this block number, track them
+            if let Some(new_addresses) = contract_log_pruner.get(&block_number) {
+                allowed_addresses.extend(new_addresses.iter().copied());
+            }
+
             for (idx, receipt) in receipts.iter().enumerate() {
                 let receipt_idx = first_tx_index + idx as u64;
+                // Skip writing receipt if log filter is active and it does not have any logs to
+                // retain
+                if prunable_receipts &&
+                    has_contract_log_filter &&
+                    !receipt.logs().iter().any(|log| allowed_addresses.contains(&log.address))
+                {
+                    continue
+                }
 
                 if let Some(writer) = &mut receipts_static_writer {
                     writer.append_receipt(receipt_idx, receipt)?;

docs/vocs/docs/pages/cli/reth/node.mdx

@@ -801,6 +801,9 @@ Pruning:
       --prune.receipts.before <BLOCK_NUMBER>
           Prune receipts before the specified block number. The specified block number is not pruned
 
+      --prune.receiptslogfilter <FILTER_CONFIG>
+          Configure receipts log filter. Format: <`address`>:<`prune_mode`>... where <`prune_mode`> can be 'full', 'distance:<`blocks`>', or 'before:<`block_number`>'
+
       --prune.account-history.full
           Prunes all account history
 

docs/vocs/vocs.config.ts

@@ -21,7 +21,7 @@ export default defineConfig({
     },
     { text: 'GitHub', link: 'https://github.com/paradigmxyz/reth' },
     {
-      text: 'v1.9.0',
+      text: 'v1.9.4',
       items: [
         {
           text: 'Releases',