diff --git a/crates/cli/commands/src/download/mod.rs b/crates/cli/commands/src/download/mod.rs index 22bcb5ce5f6..ad12629c1e5 100644 --- a/crates/cli/commands/src/download/mod.rs +++ b/crates/cli/commands/src/download/mod.rs @@ -113,7 +113,7 @@ use reth_db::{init_db, Database}; use reth_db_api::transaction::DbTx; use reth_fs_util as fs; use reth_node_core::args::DefaultPruningValues; -use reth_prune_types::{PruneMode, PruneModes}; +use reth_prune_types::{PruneMode, PruneModes, PruneSegment}; use source::{ discover_manifest_url, fetch_manifest_from_source, fetch_snapshot_api_entries, print_snapshot_listing, resolve_manifest_base_url, @@ -886,11 +886,17 @@ impl> DownloadCo SnapshotComponentType::State | SnapshotComponentType::Headers => { return ComponentSelection::All } - SnapshotComponentType::Transactions => prune_modes.bodies_history, + SnapshotComponentType::Transactions => { + prune_modes.write_prune_mode(PruneSegment::Bodies) + } SnapshotComponentType::TransactionSenders => prune_modes.sender_recovery, - SnapshotComponentType::Receipts => prune_modes.receipts, - SnapshotComponentType::AccountChangesets => prune_modes.account_history, - SnapshotComponentType::StorageChangesets => prune_modes.storage_history, + SnapshotComponentType::Receipts => prune_modes.write_prune_mode(PruneSegment::Receipts), + SnapshotComponentType::AccountChangesets => { + prune_modes.write_prune_mode(PruneSegment::AccountHistory) + } + SnapshotComponentType::StorageChangesets => { + prune_modes.write_prune_mode(PruneSegment::StorageHistory) + } SnapshotComponentType::RocksdbIndices => return ComponentSelection::None, }; diff --git a/crates/cli/commands/src/stage/run.rs b/crates/cli/commands/src/stage/run.rs index c99946219d7..31dc2fde4f3 100644 --- a/crates/cli/commands/src/stage/run.rs +++ b/crates/cli/commands/src/stage/run.rs @@ -33,6 +33,7 @@ use reth_provider::{ providers::BlockchainProvider, ChainSpecProvider, DBProvider, DatabaseProviderFactory, StageCheckpointReader, StageCheckpointWriter, }; +use reth_prune_types::PruneSegment; use reth_stages::{ stages::{ AccountHashingStage, BodyStage, ExecutionStage, HeaderStage, IndexAccountHistoryStage, @@ -314,7 +315,7 @@ impl Box::new(IndexAccountHistoryStage::new( config.stages.index_account_history, etl_config, - prune_modes.account_history, + prune_modes.write_prune_mode(PruneSegment::AccountHistory), )), None, ), @@ -322,7 +323,7 @@ impl Box::new(IndexStorageHistoryStage::new( config.stages.index_storage_history, etl_config, - prune_modes.storage_history, + prune_modes.write_prune_mode(PruneSegment::StorageHistory), )), None, ), diff --git a/crates/config/src/config.rs b/crates/config/src/config.rs index e0d28692579..87c131b2459 100644 --- a/crates/config/src/config.rs +++ b/crates/config/src/config.rs @@ -587,6 +587,7 @@ impl PruneConfig { block_interval, segments: PruneModes { + time_based, sender_recovery, transaction_lookup, receipts, @@ -609,6 +610,7 @@ impl PruneConfig { } // Merge the various segment prune modes + self.segments.time_based = self.segments.time_based.or(time_based); self.segments.sender_recovery = self.segments.sender_recovery.or(sender_recovery); self.segments.transaction_lookup = self.segments.transaction_lookup.or(transaction_lookup); self.segments.receipts = self.segments.receipts.or(receipts); @@ -1115,12 +1117,30 @@ receipts = { distance = 16384 } let _conf: Config = toml::from_str(s).unwrap(); } + #[test] + fn time_based_pruning_config_roundtrip() { + let config: Config = toml::from_str( + r#" + [prune.segments] + account_history = { distance = 13392000 } + [prune.segments.time_based] + retention_seconds = 2678400 + catch_up_percent = 130 + "#, + ) + .unwrap(); + assert_eq!(config.prune.segments.time_based, Some(Default::default())); + let restored: Config = toml::from_str(&toml::to_string(&config).unwrap()).unwrap(); + assert_eq!(restored.prune, config.prune); + } + #[test] fn test_prune_config_merge() { let mut config1 = PruneConfig { block_interval: 5, minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE, segments: PruneModes { + time_based: None, sender_recovery: Some(PruneMode::Full), transaction_lookup: None, receipts: Some(PruneMode::Distance(1000)), @@ -1138,6 +1158,7 @@ receipts = { distance = 16384 } block_interval: 10, minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE, segments: PruneModes { + time_based: Some(Default::default()), sender_recovery: Some(PruneMode::Distance(500)), transaction_lookup: Some(PruneMode::Full), receipts: Some(PruneMode::Full), @@ -1157,6 +1178,7 @@ receipts = { distance = 16384 } // Check that the configuration has been merged. Any configuration present in config1 // should not be overwritten by config2 assert_eq!(config1.block_interval, 10); + assert_eq!(config1.segments.time_based, Some(Default::default())); assert_eq!(config1.segments.sender_recovery, Some(PruneMode::Full)); assert_eq!(config1.segments.transaction_lookup, Some(PruneMode::Full)); assert_eq!(config1.segments.receipts, Some(PruneMode::Distance(1000))); diff --git a/crates/node/core/src/args/pruning.rs b/crates/node/core/src/args/pruning.rs index 557d9b3d4a2..11b99dc7106 100644 --- a/crates/node/core/src/args/pruning.rs +++ b/crates/node/core/src/args/pruning.rs @@ -68,6 +68,7 @@ impl Default for DefaultPruningValues { fn default() -> Self { Self { full_prune_modes: PruneModes { + time_based: None, sender_recovery: Some(PruneMode::Full), transaction_lookup: None, receipts: Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)), @@ -79,6 +80,7 @@ impl Default for DefaultPruningValues { }, full_bodies_history_use_pre_merge: true, minimal_prune_modes: PruneModes { + time_based: None, sender_recovery: Some(PruneMode::Full), transaction_lookup: Some(PruneMode::Full), receipts: Some(PruneMode::Distance(MINIMUM_DISTANCE)), diff --git a/crates/prune/prune/src/builder.rs b/crates/prune/prune/src/builder.rs index d30f053939d..9d67d33400c 100644 --- a/crates/prune/prune/src/builder.rs +++ b/crates/prune/prune/src/builder.rs @@ -99,6 +99,7 @@ impl PrunerBuilder { Primitives = ::Primitives, >, { + let time_based = self.segments.time_based; let segments = SegmentSet::from_components(provider_factory.static_file_provider(), self.segments); @@ -109,7 +110,8 @@ impl PrunerBuilder { self.delete_limit, self.timeout, self.finished_exex_height, - ); + ) + .with_time_based(time_based); if let Some(distance) = self.minimum_pruning_distance { pruner = pruner.with_minimum_pruning_distance(distance); } @@ -135,6 +137,7 @@ impl PrunerBuilder { + StorageChangeSetReader + RocksDBProviderFactory, { + let time_based = self.segments.time_based; let segments = SegmentSet::::from_components(static_file_provider, self.segments); let mut pruner = Pruner::new( @@ -143,7 +146,8 @@ impl PrunerBuilder { self.delete_limit, self.timeout, self.finished_exex_height, - ); + ) + .with_time_based(time_based); if let Some(distance) = self.minimum_pruning_distance { pruner = pruner.with_minimum_pruning_distance(distance); } diff --git a/crates/prune/prune/src/lib.rs b/crates/prune/prune/src/lib.rs index 4da07d495e7..75fd2a23fbb 100644 --- a/crates/prune/prune/src/lib.rs +++ b/crates/prune/prune/src/lib.rs @@ -16,6 +16,7 @@ mod limiter; mod metrics; mod pruner; pub mod segments; +mod time_based; use crate::metrics::Metrics; pub use builder::PrunerBuilder; diff --git a/crates/prune/prune/src/pruner.rs b/crates/prune/prune/src/pruner.rs index a98b9f18974..f8505f60ef4 100644 --- a/crates/prune/prune/src/pruner.rs +++ b/crates/prune/prune/src/pruner.rs @@ -11,8 +11,11 @@ use reth_provider::{ DBProvider, DatabaseProviderFactory, PruneCheckpointReader, PruneCheckpointWriter, StageCheckpointReader, }; -use reth_prune_types::{PruneProgress, PrunedSegmentInfo, PrunerOutput}; +use reth_prune_types::{ + PruneProgress, PruneSegment, PrunedSegmentInfo, PrunerOutput, TimeBasedPruning, +}; use reth_stages_types::StageId; +use reth_storage_api::{HeaderProvider, MetadataProvider, MetadataWriter}; use reth_tokio_util::{EventSender, EventStream}; use std::time::Duration; use tokio::sync::watch; @@ -47,6 +50,8 @@ pub struct Pruner { /// Optional override for the minimum pruning distance. When set, this replaces the /// per-segment hardcoded minimums (e.g. `MINIMUM_UNWIND_SAFE_DISTANCE`). minimum_pruning_distance: Option, + /// Optional time-based replacement for user history distance modes. + time_based: Option, /// The finished height of all `ExEx`'s. finished_exex_height: watch::Receiver, #[doc(hidden)] @@ -71,6 +76,7 @@ impl Pruner { delete_limit, timeout, minimum_pruning_distance: None, + time_based: None, finished_exex_height, metrics: Metrics::default(), event_sender: Default::default(), @@ -99,6 +105,7 @@ where delete_limit, timeout, minimum_pruning_distance: None, + time_based: None, finished_exex_height, metrics: Metrics::default(), event_sender: Default::default(), @@ -107,6 +114,12 @@ where } impl Pruner { + /// Sets the time-based retention policy without changing pruning execution limits. + pub const fn with_time_based(mut self, policy: Option) -> Self { + self.time_based = policy; + self + } + /// Sets the minimum pruning distance, overriding per-segment hardcoded minimums. pub const fn with_minimum_pruning_distance(mut self, distance: u64) -> Self { self.minimum_pruning_distance = Some(distance); @@ -116,7 +129,12 @@ impl Pruner { impl Pruner where - Provider: PruneCheckpointReader + PruneCheckpointWriter + StageCheckpointReader, + Provider: PruneCheckpointReader + + PruneCheckpointWriter + + StageCheckpointReader + + HeaderProvider + + MetadataProvider + + MetadataWriter, { /// Listen for events on the pruner. pub fn events(&self) -> EventStream { @@ -139,6 +157,13 @@ where provider: &Provider, tip_block_number: BlockNumber, ) -> PrunerResult { + if let Some(policy) = self.time_based && + (policy.retention_seconds == 0 || policy.catch_up_percent < 100) + { + return Err(PrunerError::InconsistentData( + "time-based pruning requires positive retention_seconds and catch_up_percent >= 100", + )) + } let Some(tip_block_number) = self.adjust_tip_block_number_to_finished_exex_height(tip_block_number) else { @@ -203,32 +228,51 @@ where break } + // Do not initialize time-based anchors until the segment can actually run. + if let Some(required_stage) = segment.required_stage() && + !is_stage_finished(provider, required_stage)? + { + debug!(target: "pruner", segment = ?segment.segment(), ?required_stage, + "Segment's required stage not finished, skipping"); + continue + } + let previous_checkpoint = provider.get_prune_checkpoint(segment.segment())?; if let Some((to_block, prune_mode)) = segment .mode() .map(|mode| { + if let Some(policy) = self.time_based && + mode.is_distance() && + segment.purpose().is_user() && + matches!( + segment.segment(), + PruneSegment::Bodies | + PruneSegment::Receipts | + PruneSegment::AccountHistory | + PruneSegment::StorageHistory + ) + { + return crate::time_based::prune_target( + provider, + policy, + segment.segment(), + tip_block_number, + previous_checkpoint, + self.minimum_pruning_distance + .unwrap_or_else(|| segment.segment().min_blocks()), + ) + .map(|target| target.map(|block| (block, mode))) + } mode.prune_target_block_with_min( tip_block_number, segment.segment(), segment.purpose(), self.minimum_pruning_distance, ) + .map_err(PrunerError::from) }) .transpose()? .flatten() { - // Check if segment has a required stage that must be finished first - if let Some(required_stage) = segment.required_stage() && - !is_stage_finished(provider, required_stage)? - { - debug!( - target: "pruner", - segment = ?segment.segment(), - ?required_stage, - "Segment's required stage not finished, skipping" - ); - continue - } - debug!( target: "pruner", segment = ?segment.segment(), @@ -239,7 +283,6 @@ where ); let segment_start = Instant::now(); - let previous_checkpoint = provider.get_prune_checkpoint(segment.segment())?; let segment_output = segment.prune( provider, PruneInput { previous_checkpoint, to_block, limiter: limiter.clone() }, @@ -348,7 +391,12 @@ where impl Pruner where PF: DatabaseProviderFactory< - ProviderRW: PruneCheckpointWriter + PruneCheckpointReader + StageCheckpointReader, + ProviderRW: PruneCheckpointWriter + + PruneCheckpointReader + + StageCheckpointReader + + HeaderProvider + + MetadataProvider + + MetadataWriter, >, { /// Run the pruner. This will only prune data up to the highest finished ExEx height, if there @@ -380,9 +428,159 @@ fn is_stage_finished( #[cfg(test)] mod tests { - use crate::Pruner; + use crate::{Pruner, PrunerBuilder}; + use alloy_consensus::Header; + use alloy_primitives::Address; + use reth_config::PruneConfig; + use reth_db_api::{models::StorageSettings, tables}; use reth_exex_types::FinishedExExHeight; - use reth_provider::test_utils::create_test_provider_factory; + use reth_primitives_traits::{Account, SealedHeader, StorageEntry}; + use reth_provider::{test_utils::create_test_provider_factory, PruneCheckpointReader}; + use reth_prune_types::{PruneMode, PruneModes, PruneSegment, TimeBasedPruning}; + use reth_stages::test_utils::TestStageDB; + use reth_storage_api::StorageSettingsCache; + + #[test] + fn time_based_catch_up_across_block_time_change_and_restart() { + let db = TestStageDB::default(); + db.factory.set_storage_settings_cache(StorageSettings::v1()); + let headers = (0..=700) + .map(|number| { + SealedHeader::seal_slow(Header { + number, + timestamp: if number <= 100 { number * 2 } else { 200 + (number - 100) / 5 }, + ..Default::default() + }) + }) + .collect::>(); + db.insert_headers(&headers).unwrap(); + let changesets = (0..=700) + .map(|_| vec![(Address::ZERO, Account::default(), vec![StorageEntry::default()])]) + .collect::>(); + db.insert_changesets(changesets.clone(), None).unwrap(); + db.insert_history(changesets, None).unwrap(); + + // Establish different actual checkpoints with the original block-distance policy. + let mut config = PruneConfig { + minimum_pruning_distance: 5, + segments: PruneModes { + account_history: Some(PruneMode::Distance(100)), + storage_history: Some(PruneMode::Distance(110)), + ..Default::default() + }, + ..Default::default() + }; + PrunerBuilder::new(config.clone()) + .build_with_provider_factory(db.factory.clone()) + .run(150) + .unwrap(); + + // The nominal distance would retain everything in this small chain. Time retention + // must replace it, without jumping either segment straight to the desired cutoff. + config.segments.account_history = Some(PruneMode::Distance(13_392_000)); + config.segments.storage_history = Some(PruneMode::Distance(13_392_000)); + config.segments.time_based = + Some(TimeBasedPruning { retention_seconds: 100, catch_up_percent: 130 }); + let (exex_tx, exex_rx) = tokio::sync::watch::channel(FinishedExExHeight::NoExExs); + + // An interrupted first run must not leave an anchor ahead of committed progress. + { + let provider = db.factory.provider_rw().unwrap(); + PrunerBuilder::new(config.clone()) + .build_with_provider_factory(db.factory.clone()) + .run_with_provider(&provider, 100) + .unwrap(); + // Dropping the provider aborts this transaction, including anchor initialization. + } + + // At head timestamp 220, the cutoffs are 113 and 93: six old blocks per segment, + // not fifty. Later, both converge to 100 seconds and cross into repeated timestamps. + for (tip, account_cutoff, storage_cutoff) in + [(150, 50, 40), (200, 56, 46), (350, 74, 65), (650, 149, 149), (700, 199, 199)] + { + // Recreate the pruner every time; catch-up must survive process restarts. + let mut pruner = PrunerBuilder::new(config.clone()) + .finished_exex_height(exex_rx.clone()) + .build_with_provider_factory(db.factory.clone()); + pruner.run(tip).unwrap(); + let provider = db.factory.provider().unwrap(); + for (segment, cutoff) in [ + (PruneSegment::AccountHistory, account_cutoff), + (PruneSegment::StorageHistory, storage_cutoff), + ] { + assert_eq!( + provider.get_prune_checkpoint(segment).unwrap().unwrap().block_number, + Some(cutoff) + ); + } + assert_eq!( + db.table::() + .unwrap() + .into_iter() + .map(|(block, _)| block) + .collect::>(), + ((account_cutoff + 1)..=700).collect::>() + ); + assert_eq!( + db.table::() + .unwrap() + .into_iter() + .map(|(key, _)| key.block_number()) + .collect::>(), + ((storage_cutoff + 1)..=700).collect::>() + ); + assert_eq!( + db.table::().unwrap()[0].1.iter().collect::>(), + ((account_cutoff + 1)..=700).collect::>() + ); + assert_eq!( + db.table::().unwrap()[0].1.iter().collect::>(), + ((storage_cutoff + 1)..=700).collect::>() + ); + } + + // A shorter window and faster multiplier must re-anchor, not accelerate retroactively. + config.segments.time_based = + Some(TimeBasedPruning { retention_seconds: 50, catch_up_percent: 200 }); + let mut pruner = + PrunerBuilder::new(config.clone()).build_with_provider_factory(db.factory.clone()); + pruner.run(700).unwrap(); + assert_eq!(db.table::().unwrap()[0].0, 200); + + // A lower head (unwind) or lagging ExEx cannot advance pruning. + pruner.run(650).unwrap(); + exex_tx.send(FinishedExExHeight::Height(650)).unwrap(); + let mut pruner = PrunerBuilder::new(config.clone()) + .finished_exex_height(exex_rx) + .build_with_provider_factory(db.factory.clone()); + pruner.run(700).unwrap(); + assert_eq!(db.table::().unwrap()[0].0, 200); + + // Time-based eligibility still cannot cross the configured block safety floor. + let headers = (701..=750) + .map(|number| { + SealedHeader::seal_slow(Header { + number, + timestamp: 200 + (number - 100) / 5, + ..Default::default() + }) + }) + .collect::>(); + db.insert_headers(&headers).unwrap(); + config.minimum_pruning_distance = 530; + PrunerBuilder::new(config.clone()) + .build_with_provider_factory(db.factory.clone()) + .run(750) + .unwrap(); + assert_eq!(db.table::().unwrap()[0].0, 221); + + config.segments.time_based.as_mut().unwrap().retention_seconds = 0; + assert!(PrunerBuilder::new(config) + .build_with_provider_factory(db.factory.clone()) + .run(750) + .is_err()); + assert_eq!(db.table::().unwrap()[0].0, 221); + } #[test] fn is_pruning_needed() { diff --git a/crates/prune/prune/src/segments/set.rs b/crates/prune/prune/src/segments/set.rs index d4541393b66..bbbe1107289 100644 --- a/crates/prune/prune/src/segments/set.rs +++ b/crates/prune/prune/src/segments/set.rs @@ -72,6 +72,7 @@ where storage_history, bodies_history, receipts_log_filter, + .. } = prune_modes; Self::default() diff --git a/crates/prune/prune/src/time_based.rs b/crates/prune/prune/src/time_based.rs new file mode 100644 index 00000000000..6f4dc8909dc --- /dev/null +++ b/crates/prune/prune/src/time_based.rs @@ -0,0 +1,116 @@ +use crate::PrunerError; +use alloy_consensus::BlockHeader; +use reth_prune_types::{PruneCheckpoint, PruneSegment, TimeBasedPruning}; +use reth_storage_api::{HeaderProvider, MetadataProvider, MetadataWriter}; +use tracing::debug; + +/// Resolves a time-based cutoff using an anchor committed with the prune checkpoints. +pub(crate) fn prune_target( + provider: &Provider, + policy: TimeBasedPruning, + segment: PruneSegment, + tip: u64, + previous: Option, + minimum_distance: u64, +) -> Result, PrunerError> +where + Provider: HeaderProvider + MetadataProvider + MetadataWriter, +{ + let head_timestamp = timestamp(provider, tip)?; + let key = format!("pruning.time.v1.{segment}"); + let stored = provider.get_metadata(&key)?; + let anchor = stored.as_deref().map(Anchor::decode).transpose()?; + let anchor = match anchor { + Some(anchor) if anchor.policy == policy => anchor, + _ => { + // Changing the policy must not apply a new multiplier to past elapsed time, or + // suddenly make a shorter retention window eligible. Re-anchor at actual progress. + let anchor = Anchor { + head_timestamp, + cutoff_timestamp: timestamp( + provider, + previous.and_then(|cp| cp.block_number).unwrap_or_default().min(tip), + )?, + policy, + }; + provider.write_metadata(&key, anchor.encode())?; + debug!(target: "pruner", ?segment, head_timestamp, cutoff_timestamp = anchor.cutoff_timestamp, + ?policy, "Initialized time-based pruning anchor"); + anchor + } + }; + + // Use canonical chain time, never wall time. Saturation pauses advancement during an unwind + // before the anchor. Whole-second headers conservatively retain the entire boundary second, + // including on chains with multiple blocks sharing a timestamp. + let elapsed = head_timestamp.saturating_sub(anchor.head_timestamp); + let advance = (u128::from(elapsed) * u128::from(policy.catch_up_percent) / 100) + .min(u128::from(u64::MAX)) as u64; + let cutoff = head_timestamp + .saturating_sub(policy.retention_seconds) + .min(anchor.cutoff_timestamp.saturating_add(advance)); + let Some(max_block) = tip.checked_sub(minimum_distance) else { return Ok(None) }; + + // Find the first block whose timestamp is >= cutoff. Headers are retained even when bodies + // and state history are pruned. Binary search also handles the 2s -> 200ms transition without + // chain-specific fork configuration. + let mut low = 0; + let mut high = max_block; + if timestamp(provider, high)? < cutoff { + return Ok(Some(high)) + } + while low < high { + let mid = low + (high - low) / 2; + if timestamp(provider, mid)? < cutoff { + low = mid + 1; + } else { + high = mid; + } + } + let target = low.checked_sub(1); + debug!(target: "pruner", ?segment, head_timestamp, cutoff_timestamp = cutoff, ?target, + target_lag_seconds = head_timestamp.saturating_sub(cutoff).saturating_sub(policy.retention_seconds), + "Resolved time-based pruning target"); + Ok(target) +} + +fn timestamp(provider: &impl HeaderProvider, block: u64) -> Result { + Ok(provider + .header_by_number(block)? + .ok_or(PrunerError::InconsistentData("missing header for time-based pruning"))? + .timestamp()) +} + +/// A fixed-width, versioned metadata value avoids changing the prune checkpoint database codec. +struct Anchor { + head_timestamp: u64, + cutoff_timestamp: u64, + policy: TimeBasedPruning, +} + +impl Anchor { + fn encode(&self) -> Vec { + [ + self.head_timestamp, + self.cutoff_timestamp, + self.policy.retention_seconds, + self.policy.catch_up_percent, + ] + .into_iter() + .flat_map(u64::to_be_bytes) + .collect() + } + + fn decode(bytes: &[u8]) -> Result { + let bytes: &[u8; 32] = bytes.try_into().map_err(|_| { + PrunerError::InconsistentData("invalid time-based pruning anchor metadata") + })?; + let words: [u64; 4] = + core::array::from_fn(|i| u64::from_be_bytes(bytes.as_chunks::<8>().0[i])); + Ok(Self { + head_timestamp: words[0], + cutoff_timestamp: words[1], + policy: TimeBasedPruning { retention_seconds: words[2], catch_up_percent: words[3] }, + }) + } +} diff --git a/crates/prune/types/src/lib.rs b/crates/prune/types/src/lib.rs index 88d2d6490e6..53d92bb8d61 100644 --- a/crates/prune/types/src/lib.rs +++ b/crates/prune/types/src/lib.rs @@ -17,6 +17,8 @@ mod mode; mod pruner; mod segment; mod target; +mod time_based; +pub use time_based::TimeBasedPruning; use alloc::{collections::BTreeMap, vec::Vec}; use alloy_primitives::{Address, BlockNumber}; diff --git a/crates/prune/types/src/target.rs b/crates/prune/types/src/target.rs index 59610957304..4e5551865db 100644 --- a/crates/prune/types/src/target.rs +++ b/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, ReceiptsLogPruneConfig}; +use crate::{PruneCheckpoint, PruneMode, PruneSegment, ReceiptsLogPruneConfig, TimeBasedPruning}; /// 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 @@ -46,6 +46,10 @@ pub enum HistoryType { #[cfg_attr(any(test, feature = "serde"), derive(serde::Serialize, serde::Deserialize))] #[cfg_attr(any(test, feature = "serde"), serde(default))] pub struct PruneModes { + /// Optional time-based replacement for distance-pruned bodies, receipts, and history. + /// Disabled by default. Explicit `Full`/`Before` modes and other segments are unaffected. + #[cfg_attr(any(test, feature = "serde"), serde(skip_serializing_if = "Option::is_none"))] + pub time_based: Option, /// Sender Recovery pruning configuration. #[cfg_attr(any(test, feature = "serde"), serde(skip_serializing_if = "Option::is_none"))] pub sender_recovery: Option, @@ -93,6 +97,7 @@ impl PruneModes { /// Sets pruning to all targets. pub fn all() -> Self { Self { + time_based: None, sender_recovery: Some(PruneMode::Full), transaction_lookup: Some(PruneMode::Full), receipts: Some(PruneMode::Full), @@ -108,6 +113,23 @@ impl PruneModes { self.receipts.is_some() || !self.receipts_log_filter.is_empty() } + /// Returns the mode that may skip data during writes or history indexing. + /// + /// Time-based pruning must use its persisted catch-up anchor. Without that context, retain + /// the data and let the pruner delete it instead of applying the replaced block distance. + pub fn write_prune_mode(&self, segment: PruneSegment) -> Option { + let mode = match segment { + PruneSegment::AccountHistory => self.account_history, + PruneSegment::StorageHistory => self.storage_history, + PruneSegment::Receipts => self.receipts, + PruneSegment::Bodies => self.bodies_history, + PruneSegment::SenderRecovery => return self.sender_recovery, + PruneSegment::TransactionLookup => return self.transaction_lookup, + _ => return None, + }; + mode.filter(|mode| self.time_based.is_none() || !mode.is_distance()) + } + /// Migrates deprecated prune mode values to their new defaults. /// /// Returns `true` if any migration was performed. @@ -147,22 +169,30 @@ impl PruneModes { ), ] { if let Some(PruneMode::Distance(limit)) = prune_mode { - // check if distance exceeds the configured limit - if distance > *limit { - // but only if we haven't pruned the target yet, if we don't have a checkpoint - // yet, it's fully unpruned yet - let pruned_height = checkpoint - .and_then(|checkpoint| checkpoint.1.block_number) - .unwrap_or(latest_block); - if pruned_height >= target_block { - // we've pruned the target block already and can't unwind past it - return Err(UnwindTargetPrunedError::TargetBeyondHistoryLimit { - latest_block, - target_block, - history_type: history_type.clone(), - limit: *limit, - }) - } + // A time-based cutoff can be newer than the nominal block-distance cutoff. + // Respect actual deletions even after the time-based policy is disabled. + let pruned_height = checkpoint.and_then(|cp| cp.1.block_number); + let use_checkpoint = self.time_based.is_some() || + pruned_height + .is_some_and(|height| latest_block.saturating_sub(height) < *limit); + let target_pruned = pruned_height.map_or_else( + // Preserve the distance policy's conservative check without a checkpoint. + || self.time_based.is_none() && distance > *limit, + // A checkpoint can precede a partially pruned block, so retain the + // existing conservative boundary when checking actual deletions. + |height| (use_checkpoint || distance > *limit) && height >= target_block, + ); + if target_pruned { + return Err(UnwindTargetPrunedError::TargetBeyondHistoryLimit { + latest_block, + target_block, + history_type: history_type.clone(), + limit: if self.time_based.is_some() || distance <= *limit { + latest_block.saturating_sub(pruned_height.unwrap_or(latest_block)) + } else { + *limit + }, + }) } } } @@ -248,6 +278,30 @@ mod tests { #[test] fn test_unwind_target_unpruned() { + // Time-based retention can prune inside the nominal block distance. Actual checkpoints + // must protect unwinds, even when that distance would normally allow the target. + let time_based = PruneModes { + time_based: Some(TimeBasedPruning::default()), + account_history: Some(PruneMode::Distance(13_392_000)), + ..Default::default() + }; + let checkpoints = [( + PruneSegment::AccountHistory, + PruneCheckpoint { + block_number: Some(850), + tx_number: None, + prune_mode: PruneMode::Distance(13_392_000), + }, + )]; + assert!(time_based.ensure_unwind_target_unpruned(1000, 800, &checkpoints).is_err()); + assert!(time_based.ensure_unwind_target_unpruned(1000, 850, &checkpoints).is_err()); + assert!(time_based.ensure_unwind_target_unpruned(1000, 851, &checkpoints).is_ok()); + assert!(time_based.ensure_unwind_target_unpruned(1000, 800, &[]).is_ok()); + let disabled = PruneModes { time_based: None, ..time_based }; + assert!(disabled.ensure_unwind_target_unpruned(1000, 849, &checkpoints).is_err()); + assert!(disabled.ensure_unwind_target_unpruned(1000, 850, &checkpoints).is_err()); + assert!(disabled.ensure_unwind_target_unpruned(1000, 851, &checkpoints).is_ok()); + // Test case 1: No pruning configured - should always succeed let prune_modes = PruneModes::default(); assert!(prune_modes.ensure_unwind_target_unpruned(1000, 500, &[]).is_ok()); diff --git a/crates/prune/types/src/time_based.rs b/crates/prune/types/src/time_based.rs new file mode 100644 index 00000000000..2b26887708f --- /dev/null +++ b/crates/prune/types/src/time_based.rs @@ -0,0 +1,21 @@ +/// Time-based retention for distance-pruned bodies, receipts, and account/storage history. +/// +/// The configured block distance is replaced by this policy. Other modes and segments are +/// unchanged. Catch-up starts at each segment's actual prune checkpoint, not its configured +/// distance, and advances in historical seconds per second of chain progress. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[cfg_attr(any(test, feature = "serde"), derive(serde::Serialize, serde::Deserialize))] +#[cfg_attr(any(test, feature = "serde"), serde(default, deny_unknown_fields))] +pub struct TimeBasedPruning { + /// Minimum history age to retain, in seconds. Defaults to 31 days. Must be nonzero. + pub retention_seconds: u64, + /// Cutoff advancement as a percentage of elapsed chain time. Defaults to 130 (1.3x). + /// Must be at least 100; 100 maintains the initial window without catching up. + pub catch_up_percent: u64, +} + +impl Default for TimeBasedPruning { + fn default() -> Self { + Self { retention_seconds: 31 * 24 * 60 * 60, catch_up_percent: 130 } + } +} diff --git a/crates/stages/stages/src/sets.rs b/crates/stages/stages/src/sets.rs index 0edb14e0c85..2327f74d6a9 100644 --- a/crates/stages/stages/src/sets.rs +++ b/crates/stages/stages/src/sets.rs @@ -52,7 +52,7 @@ use reth_evm::ConfigureEvm; use reth_network_p2p::{bodies::downloader::BodyDownloader, headers::downloader::HeaderDownloader}; use reth_primitives_traits::{Block, NodePrimitives}; use reth_provider::HeaderSyncGapProvider; -use reth_prune_types::{PruneMode, PruneModes}; +use reth_prune_types::{PruneMode, PruneModes, PruneSegment}; use reth_stages_api::Stage; use std::sync::Arc; use tokio::sync::watch; @@ -468,12 +468,12 @@ where .add_stage(IndexStorageHistoryStage::new( self.stages_config.index_storage_history, self.stages_config.etl.clone(), - self.prune_modes.storage_history, + self.prune_modes.write_prune_mode(PruneSegment::StorageHistory), )) .add_stage(IndexAccountHistoryStage::new( self.stages_config.index_account_history, self.stages_config.etl.clone(), - self.prune_modes.account_history, + self.prune_modes.write_prune_mode(PruneSegment::AccountHistory), )) } } diff --git a/crates/stages/stages/src/stages/execution/mod.rs b/crates/stages/stages/src/stages/execution/mod.rs index 61253135c85..6f7f09eafc8 100644 --- a/crates/stages/stages/src/stages/execution/mod.rs +++ b/crates/stages/stages/src/stages/execution/mod.rs @@ -17,6 +17,7 @@ use reth_provider::{ ProviderError, StateWriteConfig, StateWriter, StaticFileProviderFactory, StatsReader, StoragePath, StorageSettingsCache, TransactionVariant, }; +use reth_prune_types::PruneSegment; use reth_revm::database::StateProviderDatabase; use reth_stages_api::{ BlockErrorKind, CheckpointBlockRange, EntitiesCheckpoint, ExecInput, ExecOutput, @@ -460,10 +461,10 @@ where // If both account history and storage history pruning is configured, clear reverts // for this block. if prune_modes - .account_history + .write_prune_mode(PruneSegment::AccountHistory) .is_some_and(|m| m.should_prune(block_number, max_block)) && prune_modes - .storage_history + .write_prune_mode(PruneSegment::StorageHistory) .is_some_and(|m| m.should_prune(block_number, max_block)) { reverts.clear(); diff --git a/crates/stages/stages/src/stages/prune.rs b/crates/stages/stages/src/stages/prune.rs index 8ae849d1560..6b1d62da2c1 100644 --- a/crates/stages/stages/src/stages/prune.rs +++ b/crates/stages/stages/src/stages/prune.rs @@ -10,7 +10,9 @@ use reth_prune::{ use reth_stages_api::{ ExecInput, ExecOutput, Stage, StageCheckpoint, StageError, StageId, UnwindInput, UnwindOutput, }; -use reth_storage_api::{ChangeSetReader, StorageChangeSetReader, StorageSettingsCache}; +use reth_storage_api::{ + ChangeSetReader, MetadataProvider, MetadataWriter, StorageChangeSetReader, StorageSettingsCache, +}; use tracing::info; /// The prune stage that runs the pruner with the provided prune modes. @@ -50,6 +52,8 @@ where > + StorageSettingsCache + ChangeSetReader + StorageChangeSetReader + + MetadataProvider + + MetadataWriter + RocksDBProviderFactory, { fn id(&self) -> StageId { @@ -158,6 +162,8 @@ where > + StorageSettingsCache + ChangeSetReader + StorageChangeSetReader + + MetadataProvider + + MetadataWriter + RocksDBProviderFactory, { fn id(&self) -> StageId { diff --git a/crates/storage/provider/src/providers/database/provider.rs b/crates/storage/provider/src/providers/database/provider.rs index 22a219bb360..e687f14a9a5 100644 --- a/crates/storage/provider/src/providers/database/provider.rs +++ b/crates/storage/provider/src/providers/database/provider.rs @@ -550,7 +550,7 @@ impl DatabaseProvider StateWriter // Skip writing receipts if pruning configuration requires us to. if prunable_receipts && self.prune_modes - .receipts + .write_prune_mode(PruneSegment::Receipts) .is_some_and(|mode| mode.should_prune(block_number, tip)) { continue @@ -5115,6 +5115,31 @@ mod tests { provider_rw.commit().unwrap(); }; + // Time-based retention replaces the distance during backfill too. Do not discard + // receipts before the pruner can apply its persisted catch-up boundary. + for settings in [StorageSettings::v1(), StorageSettings::v2()] { + for time_based in [None, Some(Default::default())] { + let factory = create_test_provider_factory(); + factory.set_storage_settings_cache(settings); + let factory = factory.with_prune_modes(PruneModes { + time_based, + receipts: Some(PruneMode::Distance(100)), + ..Default::default() + }); + let provider_rw = factory.provider_rw().unwrap(); + insert_blocks(&provider_rw, 200, 1); + provider_rw.commit().unwrap(); + write_receipts( + factory.provider_rw().unwrap().with_minimum_pruning_distance(100), + 0, + ); + assert_eq!( + factory.provider().unwrap().receipt(0).unwrap().is_some(), + time_based.is_some() + ); + } + } + // Legacy mode (receipts in DB) - should be prunable { let factory = create_test_provider_factory(); diff --git a/docs/vocs/docs/pages/run/storage/pruning.mdx b/docs/vocs/docs/pages/run/storage/pruning.mdx index e5a3117003a..74143128589 100644 --- a/docs/vocs/docs/pages/run/storage/pruning.mdx +++ b/docs/vocs/docs/pages/run/storage/pruning.mdx @@ -87,6 +87,46 @@ Use `reth.toml` or individual `--prune.*` flags to configure each pruning segment yourself. See the [pruning configuration](/run/configuration#the-prune-section) reference for the TOML format. +### Experimental time-based retention + +To retain 31 days across block-time changes, enable this opt-in policy in `reth.toml`: + +```toml +[prune.segments.time_based] +retention_seconds = 2678400 +catch_up_percent = 130 +``` + +It replaces the block distance for **distance-configured** bodies, receipts, account history, +and storage history. Configure those segments with `distance` modes (or use a full preset +that does so). Disabled segments, `full`/`before` modes, transaction lookup, sender recovery, +and log filters are unchanged. In particular, an Ethereum full preset's pre-Merge bodies +policy is not replaced. The policy is disabled by default. + +On first use, each segment records its actual prune checkpoint and current chain timestamp +in database metadata, in the same transaction as pruning. Its cutoff then advances through +1.3 seconds of history per second of chain progress, capped at the head timestamp minus +31 days. Existing prune runs, worker scheduling, and execution limits are unchanged. This +is not a time or deletion budget and does not guarantee a maximum prune-run duration. + +- Anchors survive restarts and snapshot restores. Changing either setting re-anchors at + actual progress, rather than applying the new rate retroactively. +- No checkpoint means catch-up starts at genesis. Already-deleted history cannot be restored. +- Only blocks strictly older than the cutoff are eligible. Whole-second header timestamps, + static-file boundaries, ExEx progress, and unwind safety floors can retain extra history. +- At 130%, seven days of excess history takes about 23 days of chain progress to eliminate. + The multiplier controls historical time, not block count, CPU usage, or database operations. +- Backfill uses the same policy. It retains data during writes/indexing rather than skipping + it using the superseded distance. Catch-up can therefore temporarily require extra disk. +- Removing this setting restores block-distance pruning and can cause a sudden catch-up run. + Test configuration changes on a disposable snapshot copy before using them on a live node. + +For snapshot trials, compare `pruner` debug logs (`head_timestamp`, `cutoff_timestamp`, +`target_lag_seconds`) and existing per-segment duration/checkpoint metrics with persistence +lag and disk growth. A copied snapshot must include the database metadata and prune checkpoints. +Offline `prune` also uses this policy; repeatedly running it at an unchanged head does not +advance a catch-up cutoff to 31 days immediately. + ## Disk usage Disk usage changes as networks grow and depends on the selected node mode. For