From fdbcf2c142417b91f59f2abf84dca6120c48435f Mon Sep 17 00:00:00 2001 From: Arsenii Kulikov Date: Fri, 21 Aug 2026 09:09:14 +0000 Subject: [PATCH 1/8] feat(download): expose prepared snapshot context (#26777) (cherry picked from commit d892c93b3abf15b7573ca600124024258cf888ec) --- .../cli/commands/src/download/config_gen.rs | 1 + crates/cli/commands/src/download/manifest.rs | 35 ++++++++++ crates/cli/commands/src/download/mod.rs | 68 +++++++++++++------ crates/cli/commands/src/download/planning.rs | 2 + crates/ethereum/cli/src/app.rs | 3 +- 5 files changed, 86 insertions(+), 23 deletions(-) diff --git a/crates/cli/commands/src/download/config_gen.rs b/crates/cli/commands/src/download/config_gen.rs index ed7ee2e904a..34f70b10af2 100644 --- a/crates/cli/commands/src/download/config_gen.rs +++ b/crates/cli/commands/src/download/config_gen.rs @@ -294,6 +294,7 @@ mod tests { base_url: None, reth_version: None, components: BTreeMap::new(), + extensions: Default::default(), } } diff --git a/crates/cli/commands/src/download/manifest.rs b/crates/cli/commands/src/download/manifest.rs index 50d57b78cbe..4f88e3ca907 100644 --- a/crates/cli/commands/src/download/manifest.rs +++ b/crates/cli/commands/src/download/manifest.rs @@ -52,6 +52,12 @@ pub struct SnapshotManifest { pub reth_version: Option, /// Available snapshot components. pub components: BTreeMap, + /// Chain-specific manifest fields not interpreted by Reth. + /// + /// Extensions are retained so downstream commands can consume snapshot metadata without + /// refetching or reparsing the manifest selected by Reth. + #[serde(default, flatten)] + pub extensions: BTreeMap, } /// Manifest entry for a single snapshot component. @@ -764,6 +770,7 @@ pub fn generate_manifest_with_sink( base_url: base_url.map(str::to_owned), reth_version: Some(reth_node_core::version::version_metadata().short_version.to_string()), components, + extensions: Default::default(), }) } @@ -1201,9 +1208,30 @@ mod tests { base_url: Some("https://example.com".to_string()), reth_version: None, components, + extensions: Default::default(), } } + #[test] + fn manifest_preserves_extensions() { + let manifest: SnapshotManifest = serde_json::from_str( + r#"{ + "block": 1, + "chain_id": 1, + "storage_version": 2, + "timestamp": 0, + "components": {}, + "consensus": { "archive": "consensus.tar.zst" } + }"#, + ) + .unwrap(); + + assert_eq!( + manifest.extensions.get("consensus"), + Some(&serde_json::json!({ "archive": "consensus.tar.zst" })) + ); + } + #[test] fn archive_urls_for_distance_all() { let m = test_manifest(); @@ -1253,6 +1281,7 @@ mod tests { base_url: Some("https://example.com".to_string()), reth_version: None, components, + extensions: Default::default(), }; let urls = m.archive_urls_for_distance(SnapshotComponentType::RocksdbIndices, Some(10)); @@ -1312,6 +1341,7 @@ mod tests { base_url: Some("https://example.com".to_string()), reth_version: None, components, + extensions: Default::default(), } } @@ -1404,6 +1434,7 @@ mod tests { base_url: Some("https://example.com".to_string()), reth_version: None, components, + extensions: Default::default(), }; let urls = m.archive_urls(SnapshotComponentType::StorageChangesets); assert_eq!(urls.len(), 49); @@ -1485,6 +1516,7 @@ mod tests { base_url: Some("https://example.com".to_string()), reth_version: None, components, + extensions: Default::default(), }; assert_eq!(manifest.output_size_for_distance(SnapshotComponentType::State, None), 1_000); @@ -1546,6 +1578,7 @@ mod tests { base_url: Some("https://example.com".to_string()), reth_version: None, components, + extensions: Default::default(), }; let state = m.snapshot_archives_for_distance(SnapshotComponentType::State, None); @@ -1692,6 +1725,7 @@ mod tests { base_url: Some("https://example.com/mainnet".to_string()), reth_version: None, components, + extensions: Default::default(), }; let urls = m.archive_urls(SnapshotComponentType::Headers); @@ -1775,6 +1809,7 @@ mod tests { base_url: Some("https://example.com/mainnet".to_string()), reth_version: None, components, + extensions: Default::default(), }; let ComponentManifest::Chunked(chunked) = diff --git a/crates/cli/commands/src/download/mod.rs b/crates/cli/commands/src/download/mod.rs index 3d00f4609c9..d7279afda21 100644 --- a/crates/cli/commands/src/download/mod.rs +++ b/crates/cli/commands/src/download/mod.rs @@ -456,24 +456,21 @@ pub struct DownloadCommand { impl> DownloadCommand { /// Runs the download command in single-archive or manifest mode. - pub async fn execute(self) -> Result<()> { + pub async fn execute(self) -> Result> { let chain = self.env.chain.chain(); - let chain_id = chain.id(); // --list: print available snapshots and exit if self.list { - let entries = fetch_snapshot_api_entries(chain_id).await?; - print_snapshot_listing(&entries, chain_id); - return Ok(()); + let entries = fetch_snapshot_api_entries(chain.id()).await?; + print_snapshot_listing(&entries, chain.id()); + return Ok(None); } - let data_dir = self.env.datadir.clone().resolve_datadir(chain); - - let cancel_token = CancellationToken::new(); - let _cancel_guard = cancel_token.drop_guard(); - // Legacy single-URL mode: download one archive and extract it if let Some(ref url) = self.url { + let cancel_token = CancellationToken::new(); + let _cancel_guard = cancel_token.drop_guard(); + let data_dir = self.env.datadir.clone().resolve_datadir(chain); let target_dir = data_dir.data_dir(); if self.force { clear_existing_datadir(target_dir)?; @@ -498,17 +495,22 @@ impl> DownloadCo .await?; info!(target: "reth::cli", "Snapshot downloaded and extracted successfully"); - return Ok(()); + return Ok(None); } let ResolvedDownload { manifest, selections, preset, planned } = - self.resolve_download(chain_id).await?; + self.resolve_download(chain.id()).await?; + let data_dir = self.env.datadir.clone().resolve_datadir(chain).data_dir().to_path_buf(); + let prepared = PreparedSnapshotDownload { manifest, data_dir }; if self.print_plan_json { - DownloadPlan::from_planned(&manifest, &planned).write_json(std::io::stdout().lock())?; - return Ok(()) + DownloadPlan::from_planned(&prepared.manifest, &planned) + .write_json(std::io::stdout().lock())?; + return Ok(Some(prepared)) } - let target_dir = data_dir.data_dir(); + let target_dir = prepared.data_dir.as_path(); + let cancel_token = CancellationToken::new(); + let _cancel_guard = cancel_token.drop_guard(); if self.force { clear_existing_datadir(target_dir)?; } @@ -535,16 +537,28 @@ impl> DownloadCo ) .await?; - self.finalize_modular_download(&selections, &manifest, preset, target_dir, &data_dir.db())?; + self.finalize_modular_download( + &selections, + &prepared.manifest, + preset, + target_dir, + &target_dir.join("db"), + )?; - Ok(()) + Ok(Some(prepared)) } - /// Resolves the exact modular archive plan without downloading or modifying the data dir. - pub async fn plan(&self) -> Result { - let chain_id = self.env.chain.chain().id(); - let resolved = self.resolve_download(chain_id).await?; - Ok(DownloadPlan::from_planned(&resolved.manifest, &resolved.planned)) + /// Resolves the exact modular archive plan and manifest context without downloading or + /// modifying the data dir. + pub async fn plan(&self) -> Result<(DownloadPlan, PreparedSnapshotDownload)> { + let chain = self.env.chain.chain(); + let resolved = self.resolve_download(chain.id()).await?; + let plan = DownloadPlan::from_planned(&resolved.manifest, &resolved.planned); + let prepared = PreparedSnapshotDownload { + manifest: resolved.manifest, + data_dir: self.env.datadir.clone().resolve_datadir(chain).data_dir().to_path_buf(), + }; + Ok((plan, prepared)) } async fn resolve_download(&self, chain_id: u64) -> Result { @@ -1034,6 +1048,15 @@ impl DownloadCommand { } } +/// A modular snapshot download after manifest and data-directory resolution. +#[derive(Debug)] +pub struct PreparedSnapshotDownload { + /// Manifest selected by the command, with a normalized `base_url`. + pub manifest: SnapshotManifest, + /// Chain-resolved directory where Reth installs the snapshot. + pub data_dir: PathBuf, +} + struct ResolvedDownload { manifest: SnapshotManifest, selections: BTreeMap, @@ -1089,6 +1112,7 @@ mod tests { base_url: Some("https://example.com".to_string()), reth_version: None, components, + extensions: Default::default(), } } diff --git a/crates/cli/commands/src/download/planning.rs b/crates/cli/commands/src/download/planning.rs index f087926d7d0..22bb8b96bc9 100644 --- a/crates/cli/commands/src/download/planning.rs +++ b/crates/cli/commands/src/download/planning.rs @@ -418,6 +418,7 @@ mod tests { base_url: Some("https://example.com".to_string()), reth_version: None, components, + extensions: Default::default(), }; let selections = BTreeMap::from([ @@ -494,6 +495,7 @@ mod tests { base_url: Some("https://example.com/mainnet".to_string()), reth_version: None, components, + extensions: Default::default(), }; let selections = BTreeMap::from([(SnapshotComponentType::Transactions, ComponentSelection::All)]); diff --git a/crates/ethereum/cli/src/app.rs b/crates/ethereum/cli/src/app.rs index cb3d732757a..44fe4dbde98 100644 --- a/crates/ethereum/cli/src/app.rs +++ b/crates/ethereum/cli/src/app.rs @@ -201,7 +201,8 @@ where Commands::Db(command) => { runner.run_blocking_command_until_exit(|ctx| command.execute::(ctx)) } - Commands::Download(command) => runner.run_blocking_until_ctrl_c(command.execute::()), + Commands::Download(command) => runner + .run_blocking_until_ctrl_c(async move { command.execute::().await.map(|_| ()) }), Commands::SnapshotManifest(command) => command.execute(), Commands::Stage(command) => { runner.run_command_until_exit(|ctx| command.execute::(ctx, components)) From b0613f37d24464654feb3d520b414166d4a06c0a Mon Sep 17 00:00:00 2001 From: Dan Cline <6798349+Rjected@users.noreply.github.com> Date: Fri, 4 Sep 2026 11:45:33 +0000 Subject: [PATCH 2/8] fix(cli): use configured minimal download defaults (#26934) Co-authored-by: Arsenii Kulikov <62447812+klkvr@users.noreply.github.com> Co-authored-by: Derek Cofausper <256792747+decofe@users.noreply.github.com> (cherry picked from commit e423a23d8132d02997f73a2a55c197807688f301) --- .../cli/commands/src/download/config_gen.rs | 72 ++++--- crates/cli/commands/src/download/manifest.rs | 22 --- crates/cli/commands/src/download/mod.rs | 176 ++++++++++++------ crates/cli/commands/src/download/tui.rs | 125 ++++++++++--- 4 files changed, 267 insertions(+), 128 deletions(-) diff --git a/crates/cli/commands/src/download/config_gen.rs b/crates/cli/commands/src/download/config_gen.rs index 34f70b10af2..2eda1df907b 100644 --- a/crates/cli/commands/src/download/config_gen.rs +++ b/crates/cli/commands/src/download/config_gen.rs @@ -7,18 +7,13 @@ use reth_config::config::{BlocksPerFileConfig, Config, PruneConfig, StaticFilesC use reth_db::tables; use reth_db_api::transaction::{DbTx, DbTxMut}; use reth_node_core::args::DefaultPruningValues; -use reth_prune_types::{PruneCheckpoint, PruneMode, PruneSegment}; +use reth_prune_types::{ + PruneCheckpoint, PruneMode, PruneSegment, MINIMUM_DISTANCE, MINIMUM_UNWIND_SAFE_DISTANCE, +}; use reth_stages_types::StageCheckpoint; use std::{collections::BTreeMap, path::Path}; use tracing::info; -/// Minimum blocks to keep for receipts, matching `--minimal` prune settings. -const MINIMUM_RECEIPTS_DISTANCE: u64 = 64; - -/// Minimum blocks to keep for history/bodies, matching `--minimal` prune settings -/// (`MINIMUM_UNWIND_SAFE_DISTANCE`). -const MINIMUM_HISTORY_DISTANCE: u64 = 10064; - /// Writes a [`Config`] as TOML to `/reth.toml`. /// /// If the file already exists, it is not overwritten. Returns `true` if the file was written. @@ -181,10 +176,21 @@ pub(crate) fn config_for_selections( }, }; - if is_archive || matches!(preset, Some(SelectionPreset::Archive)) { + if matches!(preset, Some(SelectionPreset::Archive)) { return Config { static_files, ..Default::default() }; } + if matches!(preset, Some(SelectionPreset::Minimal)) { + return Config { + prune: PruneConfig { + segments: DefaultPruningValues::get_global().minimal_prune_modes.clone(), + ..Default::default() + }, + static_files, + ..Default::default() + }; + } + if matches!(preset, Some(SelectionPreset::Full)) { let defaults = DefaultPruningValues::get_global(); let mut segments = defaults.full_prune_modes.clone(); @@ -205,6 +211,10 @@ pub(crate) fn config_for_selections( }; } + if is_archive { + return Config { static_files, ..Default::default() }; + } + let mut config = Config::default(); let mut prune = PruneConfig::default(); @@ -213,19 +223,21 @@ pub(crate) fn config_for_selections( } prune.segments.transaction_lookup = Some(PruneMode::Full); - if let Some(mode) = selection_to_prune_mode(tx_sel, Some(MINIMUM_HISTORY_DISTANCE)) { + if let Some(mode) = selection_to_prune_mode(tx_sel, Some(MINIMUM_UNWIND_SAFE_DISTANCE)) { prune.segments.bodies_history = Some(mode); } - if let Some(mode) = selection_to_prune_mode(receipt_sel, Some(MINIMUM_RECEIPTS_DISTANCE)) { + if let Some(mode) = selection_to_prune_mode(receipt_sel, Some(MINIMUM_DISTANCE)) { prune.segments.receipts = Some(mode); } - if let Some(mode) = selection_to_prune_mode(account_cs_sel, Some(MINIMUM_HISTORY_DISTANCE)) { + if let Some(mode) = selection_to_prune_mode(account_cs_sel, Some(MINIMUM_UNWIND_SAFE_DISTANCE)) + { prune.segments.account_history = Some(mode); } - if let Some(mode) = selection_to_prune_mode(storage_cs_sel, Some(MINIMUM_HISTORY_DISTANCE)) { + if let Some(mode) = selection_to_prune_mode(storage_cs_sel, Some(MINIMUM_UNWIND_SAFE_DISTANCE)) + { prune.segments.storage_history = Some(mode); } @@ -409,19 +421,16 @@ mod tests { // All segments clamped to their minimum distances assert_eq!( config.prune.segments.bodies_history, - Some(PruneMode::Distance(MINIMUM_HISTORY_DISTANCE)) - ); - assert_eq!( - config.prune.segments.receipts, - Some(PruneMode::Distance(MINIMUM_RECEIPTS_DISTANCE)) + Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)) ); + assert_eq!(config.prune.segments.receipts, Some(PruneMode::Distance(MINIMUM_DISTANCE))); assert_eq!( config.prune.segments.account_history, - Some(PruneMode::Distance(MINIMUM_HISTORY_DISTANCE)) + Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)) ); assert_eq!( config.prune.segments.storage_history, - Some(PruneMode::Distance(MINIMUM_HISTORY_DISTANCE)) + Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)) ); } @@ -448,10 +457,7 @@ mod tests { assert_eq!(config.prune.segments.sender_recovery, Some(PruneMode::Full)); // Bodies follows tx selection assert_eq!(config.prune.segments.bodies_history, Some(PruneMode::Distance(10_064))); - assert_eq!( - config.prune.segments.receipts, - Some(PruneMode::Distance(MINIMUM_RECEIPTS_DISTANCE)) - ); + assert_eq!(config.prune.segments.receipts, Some(PruneMode::Distance(MINIMUM_DISTANCE))); assert_eq!(config.prune.segments.account_history, Some(PruneMode::Distance(10_064))); assert_eq!(config.prune.segments.storage_history, Some(PruneMode::Distance(10_064))); } @@ -507,15 +513,15 @@ mod tests { assert_eq!(config.prune.segments.transaction_lookup, None); assert_eq!( config.prune.segments.receipts, - Some(PruneMode::Distance(MINIMUM_HISTORY_DISTANCE)) + Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)) ); assert_eq!( config.prune.segments.account_history, - Some(PruneMode::Distance(MINIMUM_HISTORY_DISTANCE)) + Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)) ); assert_eq!( config.prune.segments.storage_history, - Some(PruneMode::Distance(MINIMUM_HISTORY_DISTANCE)) + Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)) ); let paris_block = chain_spec @@ -525,6 +531,18 @@ mod tests { assert_eq!(config.prune.segments.bodies_history, Some(PruneMode::Before(paris_block))); } + #[test] + fn minimal_preset_matches_default_minimal_prune_config() { + let config = config_for_selections( + &BTreeMap::new(), + &empty_manifest(), + Some(SelectionPreset::Minimal), + None::<&reth_chainspec::ChainSpec>, + ); + + assert_eq!(&config.prune.segments, &DefaultPruningValues::get_global().minimal_prune_modes); + } + #[test] fn describe_selections_all_no_pruning() { let mut selections = BTreeMap::new(); diff --git a/crates/cli/commands/src/download/manifest.rs b/crates/cli/commands/src/download/manifest.rs index 4f88e3ca907..c6db67b3af3 100644 --- a/crates/cli/commands/src/download/manifest.rs +++ b/crates/cli/commands/src/download/manifest.rs @@ -257,28 +257,6 @@ impl SnapshotComponentType { matches!(self, Self::State | Self::Headers) } - /// Returns the default selection for this component in the minimal download preset. - /// - /// Matches the `--minimal` prune configuration: - /// - State/Headers: always All (required) - /// - Transactions/Changesets: Distance(10_064) (`MINIMUM_UNWIND_SAFE_DISTANCE`) - /// - Receipts: Distance(64) (`MINIMUM_DISTANCE`) - /// - TransactionSenders: None (only downloaded for archive nodes) - /// - RocksdbIndices: None (only downloaded for archive nodes) - /// - /// `tx_lookup` and `sender_recovery` are always pruned full regardless. - pub const fn minimal_selection(&self) -> ComponentSelection { - match self { - Self::State | Self::Headers => ComponentSelection::All, - Self::Transactions | Self::AccountChangesets | Self::StorageChangesets => { - ComponentSelection::Distance(10_064) - } - Self::Receipts => ComponentSelection::Distance(64), - Self::TransactionSenders => ComponentSelection::None, - Self::RocksdbIndices => ComponentSelection::None, - } - } - /// Whether this component type uses chunked archives. pub const fn is_chunked(&self) -> bool { !matches!(self, Self::State | Self::RocksdbIndices) diff --git a/crates/cli/commands/src/download/mod.rs b/crates/cli/commands/src/download/mod.rs index d7279afda21..7cd8de7f42d 100644 --- a/crates/cli/commands/src/download/mod.rs +++ b/crates/cli/commands/src/download/mod.rs @@ -108,7 +108,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; +use reth_prune_types::{PruneMode, PruneModes}; use source::{ discover_manifest_url, fetch_manifest_from_source, fetch_snapshot_api_entries, print_snapshot_listing, resolve_manifest_base_url, @@ -741,8 +741,10 @@ impl> DownloadCo } // Interactive TUI + let minimal_preset = self.minimal_preset_selections(manifest); let full_preset = self.full_preset_selections(manifest); - let SelectorOutput { selections, preset } = run_selector(manifest.clone(), &full_preset)?; + let SelectorOutput { selections, preset } = + run_selector(manifest.clone(), &minimal_preset, &full_preset)?; let selected = selections.into_iter().filter(|(_, sel)| *sel != ComponentSelection::None).collect(); @@ -754,12 +756,7 @@ impl> DownloadCo &self, manifest: &SnapshotManifest, ) -> BTreeMap { - SnapshotComponentType::ALL - .iter() - .copied() - .filter(|ty| manifest.component(*ty).is_some()) - .map(|ty| (ty, ty.minimal_selection())) - .collect() + self.pruning_preset_selections(manifest, SelectionPreset::Minimal) } /// Builds the default full-node component selection for the manifest. @@ -767,23 +764,36 @@ impl> DownloadCo &self, manifest: &SnapshotManifest, ) -> BTreeMap { + self.pruning_preset_selections(manifest, SelectionPreset::Full) + } + + /// Builds component selections from the configured pruning preset. + fn pruning_preset_selections( + &self, + manifest: &SnapshotManifest, + preset: SelectionPreset, + ) -> BTreeMap { + let defaults = DefaultPruningValues::get_global(); + let (prune_modes, bodies_history_use_pre_merge) = match preset { + SelectionPreset::Minimal => (&defaults.minimal_prune_modes, false), + SelectionPreset::Full => { + (&defaults.full_prune_modes, defaults.full_bodies_history_use_pre_merge) + } + SelectionPreset::Archive => unreachable!("archive selects every component"), + }; let mut selections = BTreeMap::new(); - for ty in [ - SnapshotComponentType::State, - SnapshotComponentType::Headers, - SnapshotComponentType::Transactions, - SnapshotComponentType::Receipts, - SnapshotComponentType::AccountChangesets, - SnapshotComponentType::StorageChangesets, - SnapshotComponentType::TransactionSenders, - SnapshotComponentType::RocksdbIndices, - ] { + for &ty in &SnapshotComponentType::ALL { if manifest.component(ty).is_none() { continue; } - let selection = self.full_selection_for_component(ty, manifest.block); + let selection = self.pruning_selection_for_component( + ty, + manifest.block, + prune_modes, + bodies_history_use_pre_merge, + ); if selection != ComponentSelection::None { selections.insert(ty, selection); } @@ -792,51 +802,40 @@ impl> DownloadCo selections } - /// Returns the full preset selection for one component type. - fn full_selection_for_component( + /// Returns the component selection for one configured pruning preset. + fn pruning_selection_for_component( &self, ty: SnapshotComponentType, snapshot_block: u64, + prune_modes: &PruneModes, + bodies_history_use_pre_merge: bool, ) -> ComponentSelection { - let defaults = DefaultPruningValues::get_global(); - match ty { - SnapshotComponentType::State | SnapshotComponentType::Headers => { - ComponentSelection::All - } - SnapshotComponentType::Transactions => { - if defaults.full_bodies_history_use_pre_merge { - match self - .env - .chain - .ethereum_fork_activation(EthereumHardfork::Paris) - .block_number() - { - Some(paris) if snapshot_block >= paris => ComponentSelection::Since(paris), - Some(_) => ComponentSelection::None, - None => ComponentSelection::All, - } - } else { - selection_from_prune_mode( - defaults.full_prune_modes.bodies_history, - snapshot_block, - ) - } - } - SnapshotComponentType::Receipts => { - selection_from_prune_mode(defaults.full_prune_modes.receipts, snapshot_block) - } - SnapshotComponentType::AccountChangesets => { - selection_from_prune_mode(defaults.full_prune_modes.account_history, snapshot_block) - } - SnapshotComponentType::StorageChangesets => { - selection_from_prune_mode(defaults.full_prune_modes.storage_history, snapshot_block) - } - SnapshotComponentType::TransactionSenders => { - selection_from_prune_mode(defaults.full_prune_modes.sender_recovery, snapshot_block) + if ty == SnapshotComponentType::Transactions && bodies_history_use_pre_merge { + return match self + .env + .chain + .ethereum_fork_activation(EthereumHardfork::Paris) + .block_number() + { + Some(paris) if snapshot_block >= paris => ComponentSelection::Since(paris), + Some(_) => ComponentSelection::None, + None => ComponentSelection::All, } - // Keep hidden by default in full mode; if users want indices they can use archive. - SnapshotComponentType::RocksdbIndices => ComponentSelection::None, } + + let mode = match ty { + SnapshotComponentType::State | SnapshotComponentType::Headers => { + return ComponentSelection::All + } + SnapshotComponentType::Transactions => prune_modes.bodies_history, + SnapshotComponentType::TransactionSenders => prune_modes.sender_recovery, + SnapshotComponentType::Receipts => prune_modes.receipts, + SnapshotComponentType::AccountChangesets => prune_modes.account_history, + SnapshotComponentType::StorageChangesets => prune_modes.storage_history, + SnapshotComponentType::RocksdbIndices => return ComponentSelection::None, + }; + + selection_from_prune_mode(mode, snapshot_block) } /// Resolves the manifest source from CLI input or snapshot discovery. @@ -1280,6 +1279,67 @@ mod tests { assert!(result.is_err()); } + #[test] + fn minimal_component_selection_uses_configured_prune_modes() { + let args = + CommandParser::>::parse_from(["reth"]).args; + let history_distance = 64_864; + let modes = PruneModes { + sender_recovery: Some(PruneMode::Full), + receipts: Some(PruneMode::Distance(128)), + account_history: Some(PruneMode::Distance(history_distance)), + storage_history: Some(PruneMode::Distance(history_distance)), + bodies_history: Some(PruneMode::Distance(history_distance)), + ..Default::default() + }; + + assert_eq!( + args.pruning_selection_for_component( + SnapshotComponentType::Transactions, + 1_000_000, + &modes, + false, + ), + ComponentSelection::Distance(history_distance) + ); + assert_eq!( + args.pruning_selection_for_component( + SnapshotComponentType::Receipts, + 1_000_000, + &modes, + false, + ), + ComponentSelection::Distance(128) + ); + assert_eq!( + args.pruning_selection_for_component( + SnapshotComponentType::AccountChangesets, + 1_000_000, + &modes, + false, + ), + ComponentSelection::Distance(history_distance) + ); + assert_eq!( + args.pruning_selection_for_component( + SnapshotComponentType::StorageChangesets, + 1_000_000, + &modes, + false, + ), + ComponentSelection::Distance(history_distance) + ); + assert_eq!( + args.pruning_selection_for_component( + SnapshotComponentType::TransactionSenders, + 1_000_000, + &modes, + false, + ), + ComponentSelection::None + ); + } + #[test] fn resolve_manifest_source_requires_explicit_source_for_non_mainnet_defaults() { let args = CommandParser::>::parse_from([ diff --git a/crates/cli/commands/src/download/tui.rs b/crates/cli/commands/src/download/tui.rs index 712ce3bee6a..08319e978f9 100644 --- a/crates/cli/commands/src/download/tui.rs +++ b/crates/cli/commands/src/download/tui.rs @@ -16,6 +16,7 @@ use ratatui::{ widgets::{Block, Borders, List, ListItem, ListState, Paragraph}, Frame, Terminal, }; +use reth_prune_types::{MINIMUM_DISTANCE, MINIMUM_UNWIND_SAFE_DISTANCE}; use std::{ collections::BTreeMap, io, @@ -33,25 +34,25 @@ pub struct SelectorOutput { /// All distance presets. Groups filter this to only valid options. const DISTANCE_PRESETS: [ComponentSelection; 6] = [ ComponentSelection::None, - ComponentSelection::Distance(64), - ComponentSelection::Distance(10_064), + ComponentSelection::Distance(MINIMUM_DISTANCE), + ComponentSelection::Distance(MINIMUM_UNWIND_SAFE_DISTANCE), ComponentSelection::Distance(100_000), ComponentSelection::Distance(1_000_000), ComponentSelection::All, ]; -/// Presets for components that require at least 64 blocks (receipts). +/// Presets for components that require the minimum pruning distance (receipts). const RECEIPTS_PRESETS: [ComponentSelection; 5] = [ - ComponentSelection::Distance(64), - ComponentSelection::Distance(10_064), + ComponentSelection::Distance(MINIMUM_DISTANCE), + ComponentSelection::Distance(MINIMUM_UNWIND_SAFE_DISTANCE), ComponentSelection::Distance(100_000), ComponentSelection::Distance(1_000_000), ComponentSelection::All, ]; -/// Presets for components that require at least 10064 blocks (account/storage history). +/// Presets for components that require the minimum unwind-safe history. const HISTORY_PRESETS: [ComponentSelection; 4] = [ - ComponentSelection::Distance(10_064), + ComponentSelection::Distance(MINIMUM_UNWIND_SAFE_DISTANCE), ComponentSelection::Distance(100_000), ComponentSelection::Distance(1_000_000), ComponentSelection::All, @@ -67,11 +68,43 @@ struct DisplayGroup { required: bool, /// Valid presets for this group. Components with minimum distance requirements /// exclude presets that would produce invalid prune configs. - presets: &'static [ComponentSelection], + presets: Vec, +} + +/// Adds a configured preset to the ordered list of generic TUI choices. +fn presets_with_configured( + presets: &[ComponentSelection], + configured: ComponentSelection, +) -> Vec { + let mut presets = presets.to_vec(); + if presets.contains(&configured) { + return presets; + } + + let insert_at = match configured { + ComponentSelection::None => 0, + ComponentSelection::Distance(distance) => presets + .iter() + .position(|selection| { + matches!(selection, ComponentSelection::Distance(other) if *other > distance) || + matches!(selection, ComponentSelection::All) + }) + .unwrap_or(presets.len()), + ComponentSelection::Since(_) => presets + .iter() + .position(|selection| matches!(selection, ComponentSelection::All)) + .unwrap_or(presets.len()), + ComponentSelection::All => presets.len(), + }; + presets.insert(insert_at, configured); + presets } /// Build the display groups from available components in the manifest. -fn build_groups(manifest: &SnapshotManifest) -> Vec { +fn build_groups( + manifest: &SnapshotManifest, + minimal_preset: &BTreeMap, +) -> Vec { let has = |ty: SnapshotComponentType| manifest.component(ty).is_some(); let mut groups = Vec::new(); @@ -81,7 +114,7 @@ fn build_groups(manifest: &SnapshotManifest) -> Vec { name: "State (mdbx)", types: vec![SnapshotComponentType::State], required: true, - presets: &DISTANCE_PRESETS, + presets: DISTANCE_PRESETS.to_vec(), }); } @@ -90,7 +123,7 @@ fn build_groups(manifest: &SnapshotManifest) -> Vec { name: "Headers", types: vec![SnapshotComponentType::Headers], required: true, - presets: &DISTANCE_PRESETS, + presets: DISTANCE_PRESETS.to_vec(), }); } @@ -99,7 +132,13 @@ fn build_groups(manifest: &SnapshotManifest) -> Vec { name: "Transactions", types: vec![SnapshotComponentType::Transactions], required: false, - presets: &HISTORY_PRESETS, + presets: presets_with_configured( + &HISTORY_PRESETS, + minimal_preset + .get(&SnapshotComponentType::Transactions) + .copied() + .unwrap_or(ComponentSelection::None), + ), }); } @@ -108,7 +147,13 @@ fn build_groups(manifest: &SnapshotManifest) -> Vec { name: "Receipts", types: vec![SnapshotComponentType::Receipts], required: false, - presets: &RECEIPTS_PRESETS, + presets: presets_with_configured( + &RECEIPTS_PRESETS, + minimal_preset + .get(&SnapshotComponentType::Receipts) + .copied() + .unwrap_or(ComponentSelection::None), + ), }); } @@ -123,11 +168,12 @@ fn build_groups(manifest: &SnapshotManifest) -> Vec { if has_stor { types.push(SnapshotComponentType::StorageChangesets); } + let configured = minimal_preset.get(&types[0]).copied().unwrap_or(ComponentSelection::None); groups.push(DisplayGroup { name: "State History", types, required: false, - presets: &HISTORY_PRESETS, + presets: presets_with_configured(&HISTORY_PRESETS, configured), }); } @@ -136,6 +182,7 @@ fn build_groups(manifest: &SnapshotManifest) -> Vec { struct SelectorApp { manifest: SnapshotManifest, + minimal_preset: BTreeMap, full_preset: BTreeMap, /// Display groups shown in the TUI. groups: Vec, @@ -152,18 +199,23 @@ struct SelectorApp { impl SelectorApp { fn new( manifest: SnapshotManifest, + minimal_preset: BTreeMap, full_preset: BTreeMap, ) -> Self { - let groups = build_groups(&manifest); + let groups = build_groups(&manifest, &minimal_preset); // Default to the minimal preset (matches --minimal prune config) - let selections = groups.iter().map(|g| g.types[0].minimal_selection()).collect(); + let selections = groups + .iter() + .map(|g| minimal_preset.get(&g.types[0]).copied().unwrap_or(ComponentSelection::None)) + .collect(); let mut list_state = ListState::default(); list_state.select(Some(0)); Self { manifest, + minimal_preset, full_preset, groups, selections, @@ -178,7 +230,7 @@ impl SelectorApp { if group.required { return; } - let presets = group.presets; + let presets = &group.presets; let current = self.selections[self.cursor]; let idx = presets.iter().position(|p| *p == current).unwrap_or(0); self.selections[self.cursor] = presets[(idx + 1) % presets.len()]; @@ -191,7 +243,7 @@ impl SelectorApp { if group.required { return; } - let presets = group.presets; + let presets = &group.presets; let current = self.selections[self.cursor]; let idx = presets.iter().position(|p| *p == current).unwrap_or(0); self.selections[self.cursor] = presets[(idx + presets.len() - 1) % presets.len()]; @@ -206,14 +258,22 @@ impl SelectorApp { fn select_minimal(&mut self) { for (i, group) in self.groups.iter().enumerate() { - self.selections[i] = group.types[0].minimal_selection(); + self.selections[i] = self + .minimal_preset + .get(&group.types[0]) + .copied() + .unwrap_or(ComponentSelection::None); } self.preset = Some(SelectionPreset::Minimal); } fn select_full(&mut self) { for (i, group) in self.groups.iter().enumerate() { - let mut selection = group.types[0].minimal_selection(); + let mut selection = self + .minimal_preset + .get(&group.types[0]) + .copied() + .unwrap_or(ComponentSelection::None); for ty in &group.types { if let Some(sel) = self.full_preset.get(ty).copied() { selection = sel; @@ -278,6 +338,7 @@ impl SelectorApp { /// Runs the interactive component selector TUI. pub fn run_selector( manifest: SnapshotManifest, + minimal_preset: &BTreeMap, full_preset: &BTreeMap, ) -> eyre::Result { enable_raw_mode()?; @@ -286,7 +347,7 @@ pub fn run_selector( let backend = CrosstermBackend::new(stdout); let mut terminal = Terminal::new(backend)?; - let mut app = SelectorApp::new(manifest, full_preset.clone()); + let mut app = SelectorApp::new(manifest, minimal_preset.clone(), full_preset.clone()); let result = event_loop(&mut terminal, &mut app); disable_raw_mode()?; @@ -436,3 +497,25 @@ fn render(f: &mut Frame<'_>, app: &mut SelectorApp) { .block(Block::default().borders(Borders::ALL)); f.render_widget(footer, chunks[2]); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn configured_distance_is_an_ordered_tui_choice() { + let configured = ComponentSelection::Distance(64_864); + let presets = presets_with_configured(&HISTORY_PRESETS, configured); + + assert_eq!( + presets, + vec![ + ComponentSelection::Distance(MINIMUM_UNWIND_SAFE_DISTANCE), + configured, + ComponentSelection::Distance(100_000), + ComponentSelection::Distance(1_000_000), + ComponentSelection::All, + ] + ); + } +} From c61c472aa58e5835a87c66b045637c84faac3eff Mon Sep 17 00:00:00 2001 From: Matthias Seitz Date: Fri, 4 Sep 2026 18:54:15 +0000 Subject: [PATCH 3/8] test(cli): cover snapshot downloads with a mock HTTP server (#27009) Co-authored-by: Matthias Seitz <19890894+mattsse@users.noreply.github.com> (cherry picked from commit 9c4573f0963c99b5b13f5c0f1b9750b46f10aed5) --- Cargo.lock | 4 + bin/reth/Cargo.toml | 4 + bin/reth/tests/it/download/mod.rs | 213 ++++++++++++++++++++++++ bin/reth/tests/it/download/server.rs | 98 +++++++++++ bin/reth/tests/it/main.rs | 2 + crates/cli/commands/src/download/mod.rs | 5 + 6 files changed, 326 insertions(+) create mode 100644 bin/reth/tests/it/download/mod.rs create mode 100644 bin/reth/tests/it/download/server.rs diff --git a/Cargo.lock b/Cargo.lock index 016f891a405..785ffa6e71f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7670,7 +7670,11 @@ dependencies = [ "alloy-rpc-types-eth", "aquamarine", "backon", + "blake3", "clap", + "http-body-util", + "hyper", + "hyper-util", "reth-chainspec", "reth-cli-runner", "reth-cli-util", diff --git a/bin/reth/Cargo.toml b/bin/reth/Cargo.toml index d616349ceda..dfca817f277 100644 --- a/bin/reth/Cargo.toml +++ b/bin/reth/Cargo.toml @@ -75,6 +75,10 @@ alloy-node-bindings.workspace = true alloy-provider = { workspace = true, features = ["reqwest"] } alloy-rpc-types-eth.workspace = true backon.workspace = true +blake3.workspace = true +http-body-util.workspace = true +hyper = { workspace = true, features = ["server", "http1"] } +hyper-util = { workspace = true, features = ["tokio"] } serde_json.workspace = true tar.workspace = true tempfile.workspace = true diff --git a/bin/reth/tests/it/download/mod.rs b/bin/reth/tests/it/download/mod.rs new file mode 100644 index 00000000000..a29d799f7de --- /dev/null +++ b/bin/reth/tests/it/download/mod.rs @@ -0,0 +1,213 @@ +mod server; + +use self::server::SnapshotServer; +use super::{reth_ok, RETH}; +use serde_json::{json, Value}; +use std::{ + collections::BTreeMap, + fs, + path::Path, + process::{Command, Output}, +}; + +/// Tiny transport fixtures: real tar/zstd archives and checksums, with distinct payloads +/// standing in for database/static files. These test download preparation, not node sync. +struct Snapshot { + files: BTreeMap>, + manifest: Value, +} + +impl Snapshot { + fn new() -> Self { + let mut files = BTreeMap::new(); + let mut entry = |archive: &str, path: &str| { + let data = archive.as_bytes(); + let bytes = archive_bytes(path, data); + let metadata = json!({ + "file": archive, "size": bytes.len(), + "output_files": [{"path": path, "size": data.len(), "blake3": blake3::hash(data).to_hex().to_string()}] + }); + files.insert(format!("/snapshot/{archive}"), bytes); + metadata + }; + let state = entry("state.tar.zst", "db/snapshot"); + let headers = entry("headers.tar.zst", "static_files/headers"); + let chunks = [ + entry("txs-0.tar.zst", "static_files/txs-0"), + entry("txs-1.tar.zst", "static_files/txs-1"), + ]; + let manifest = json!({ + "block": 19, "chain_id": 1, "storage_version": 2, "timestamp": 0, + "components": { + "state": state, "headers": headers, + "transactions": { + "blocks_per_file": 10, "total_blocks": 20, + "chunk_files": ["txs-0.tar.zst", "txs-1.tar.zst"], + "chunk_sizes": [chunks[0]["size"], chunks[1]["size"]], + "chunk_output_files": [chunks[0]["output_files"], chunks[1]["output_files"]] + } + } + }); + Self { files, manifest } + } + + async fn serve(mut self, ranges: bool) -> SnapshotServer { + self.files + .insert("/snapshot/manifest.json".into(), serde_json::to_vec(&self.manifest).unwrap()); + SnapshotServer::start(self.files, ranges).await + } +} + +fn archive_bytes(path: &str, data: &[u8]) -> Vec { + let mut archive = tar::Builder::new(zstd::Encoder::new(Vec::new(), 0).unwrap()); + let mut header = tar::Header::new_gnu(); + header.set_size(data.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + archive.append_data(&mut header, path, data).unwrap(); + archive.into_inner().unwrap().finish().unwrap() +} + +fn download(server: &SnapshotServer, datadir: &Path, args: &[&str]) -> Output { + Command::new(RETH) + .env("RUST_LOG", "off") + .args(["download", "--datadir"]) + .arg(datadir) + .args([ + "--manifest-url", + &format!("{}/snapshot/manifest.json", server.url), + "--non-interactive", + ]) + .args(args) + .output() + .unwrap() +} + +#[track_caller] +fn success(output: Output) -> String { + assert!( + output.status.success(), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout).unwrap() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn manifest_selection_reuse_and_repair() { + let snapshot = Snapshot::new(); + let state = &snapshot.files["/snapshot/state.tar.zst"]; + let offset = state.len() / 2; + let dir = tempfile::tempdir().unwrap(); + fs::create_dir(dir.path().join(".download-cache")).unwrap(); + fs::write(dir.path().join(".download-cache/state.tar.zst.part"), &state[..offset]).unwrap(); + let server = snapshot.serve(true).await; + let args = ["--with-txs-since", "10"]; + success(download(&server, dir.path(), &args)); + for (path, expected) in [ + ("db/snapshot", "state.tar.zst"), + ("static_files/headers", "headers.tar.zst"), + ("static_files/txs-1", "txs-1.tar.zst"), + ] { + assert_eq!(fs::read(dir.path().join(path)).unwrap(), expected.as_bytes()); + } + assert!(server.requests().iter().any(|(path, range)| { + path.ends_with("state.tar.zst") && range.as_deref() == Some(&format!("bytes={offset}-")) + })); + assert!(!dir.path().join("static_files/txs-0").exists()); + assert!(!server.requests().iter().any(|(path, _)| path.ends_with("txs-0.tar.zst"))); + let config: toml::Value = + toml::from_str(&fs::read_to_string(dir.path().join("reth.toml")).unwrap()).unwrap(); + assert_eq!(config["prune"]["segments"]["bodies_history"]["before"].as_integer(), Some(10)); + + server.clear_requests(); + success(download(&server, dir.path(), &args)); + assert_eq!(server.requests(), vec![("/snapshot/manifest.json".into(), None)]); + + // Same-size corruption must be detected by checksum, not just the file length. + fs::write(dir.path().join("static_files/txs-1"), b"bad-1.tar.zst").unwrap(); + server.clear_requests(); + success(download(&server, dir.path(), &args)); + assert_eq!(fs::read(dir.path().join("static_files/txs-1")).unwrap(), b"txs-1.tar.zst"); + assert!(server.requests().iter().any(|(path, _)| path.ends_with("txs-1.tar.zst"))); + assert!(server + .requests() + .iter() + .all(|(path, _)| path.ends_with("manifest.json") || path.ends_with("txs-1.tar.zst"))); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn plan_does_not_download_or_create_datadir() { + let server = Snapshot::new().serve(true).await; + let dir = tempfile::tempdir().unwrap(); + let target = dir.path().join("new"); + let output = success(download(&server, &target, &["--archive", "--print-plan-json"])); + let plan: Value = serde_json::from_str(&output).unwrap(); + assert_eq!(plan["archives"].as_array().unwrap().len(), 4); + assert!(!target.exists()); + assert_eq!(server.requests(), vec![("/snapshot/manifest.json".into(), None)]); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn legacy_streaming_and_resume() { + for (resumable, ranges) in [(false, false), (true, true), (true, false)] { + let bytes = archive_bytes("db/snapshot", b"snapshot contents"); + let offset = bytes.len() / 2; + let server = SnapshotServer::start( + BTreeMap::from([("/state.tar.zst".into(), bytes.clone())]), + ranges, + ) + .await; + let dir = tempfile::tempdir().unwrap(); + if resumable { + fs::write(dir.path().join("state.tar.zst.part"), &bytes[..offset]).unwrap(); + } + reth_ok(&[ + "download", + "--url", + &format!("{}/state.tar.zst", server.url), + "--datadir", + dir.path().to_str().unwrap(), + if resumable { "--resumable=true" } else { "--resumable=false" }, + ]); + assert_eq!(fs::read(dir.path().join("db/snapshot")).unwrap(), b"snapshot contents"); + assert!(!dir.path().join("state.tar.zst.part").exists()); + assert!(!dir.path().join("state.tar.zst").exists()); + if resumable { + assert!(server + .requests() + .iter() + .any(|(_, range)| range.as_deref() == Some(&format!("bytes={offset}-")))); + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn wrong_chain_is_rejected_before_force_removes_data() { + let mut snapshot = Snapshot::new(); + snapshot.manifest["chain_id"] = json!(11155111); + let server = snapshot.serve(true).await; + let dir = tempfile::tempdir().unwrap(); + fs::create_dir(dir.path().join("db")).unwrap(); + fs::write(dir.path().join("db/existing"), b"keep").unwrap(); + let output = download(&server, dir.path(), &["--force"]); + assert!(!output.status.success(), "a Sepolia snapshot must not be installed for mainnet"); + assert!(String::from_utf8_lossy(&output.stderr).contains("chain ID")); + assert_eq!(fs::read(dir.path().join("db/existing")).unwrap(), b"keep"); + assert_eq!(server.requests(), vec![("/snapshot/manifest.json".into(), None)]); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn bad_output_checksum_prevents_finalization() { + let mut snapshot = Snapshot::new(); + snapshot.manifest["components"]["state"]["output_files"][0]["blake3"] = + json!(blake3::hash(b"wrong").to_hex().to_string()); + let server = snapshot.serve(true).await; + let dir = tempfile::tempdir().unwrap(); + let output = download(&server, dir.path(), &["--with-txs-since", "20"]); + assert!(!output.status.success()); + assert!(String::from_utf8_lossy(&output.stderr).contains("Failed integrity validation")); + assert!(!dir.path().join("reth.toml").exists()); + assert!(!dir.path().join("db/mdbx.dat").exists()); +} diff --git a/bin/reth/tests/it/download/server.rs b/bin/reth/tests/it/download/server.rs new file mode 100644 index 00000000000..e2a60d22306 --- /dev/null +++ b/bin/reth/tests/it/download/server.rs @@ -0,0 +1,98 @@ +use http_body_util::Full; +use hyper::{body::Bytes, server::conn::http1, service::service_fn, Request, Response, StatusCode}; +use hyper_util::rt::TokioIo; +use std::{ + collections::BTreeMap, + convert::Infallible, + sync::{Arc, Mutex}, +}; +use tokio::{net::TcpListener, task::JoinHandle}; + +/// Loopback snapshot server with optional byte-range support and a request log. +/// Each test owns its server, so ports and responses are isolated across parallel runs. +pub(super) struct SnapshotServer { + pub(super) url: String, + requests: Arc>>, + task: JoinHandle<()>, +} + +impl SnapshotServer { + pub(super) async fn start(files: BTreeMap>, ranges: bool) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let requests = Arc::new(Mutex::new(Vec::new())); + let log = Arc::clone(&requests); + let files = Arc::new(files); + let task = tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let files = Arc::clone(&files); + let log = Arc::clone(&log); + tokio::spawn(async move { + let service = service_fn(move |request: Request| { + let path = request.uri().path().to_owned(); + let range = + request.headers().get("range").map(|v| v.to_str().unwrap().to_owned()); + log.lock().unwrap().push((path.clone(), range.clone())); + let mut response = Response::builder(); + let body = if let Some(data) = files.get(&path) { + let mut bytes = data.as_slice(); + if let Some(range) = range.filter(|_| ranges) { + let (start, end) = + range.strip_prefix("bytes=").unwrap().split_once('-').unwrap(); + let start: usize = start.parse().unwrap(); + let end = end + .parse::() + .unwrap_or(data.len() - 1) + .min(data.len() - 1); + if start >= data.len() { + response = response + .status(StatusCode::RANGE_NOT_SATISFIABLE) + .header("content-range", format!("bytes */{}", data.len())); + bytes = &[]; + } else { + response = response.status(StatusCode::PARTIAL_CONTENT).header( + "content-range", + format!("bytes {start}-{end}/{}", data.len()), + ); + bytes = &data[start..=end]; + } + } + response = response.header("content-length", bytes.len()); + if request.method() == hyper::Method::HEAD { + Bytes::new() + } else { + Bytes::copy_from_slice(bytes) + } + } else { + response = response.status(StatusCode::NOT_FOUND); + Bytes::new() + }; + std::future::ready(Ok::<_, Infallible>( + response.body(Full::new(body)).unwrap(), + )) + }); + // Range probes may close the connection without consuming the body. + let _ = + http1::Builder::new().serve_connection(TokioIo::new(stream), service).await; + }); + } + }); + Self { url, requests, task } + } + + pub(super) fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } + + pub(super) fn clear_requests(&self) { + self.requests.lock().unwrap().clear(); + } +} + +impl Drop for SnapshotServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +type RecordedRequest = (String, Option); diff --git a/bin/reth/tests/it/main.rs b/bin/reth/tests/it/main.rs index d71390501d6..68e0520d2c4 100644 --- a/bin/reth/tests/it/main.rs +++ b/bin/reth/tests/it/main.rs @@ -2,6 +2,8 @@ use std::{fs, path::Path, process::Command}; +mod download; + const RETH: &str = env!("CARGO_BIN_EXE_reth"); // ── Helpers ────────────────────────────────────────────────────────────────── diff --git a/crates/cli/commands/src/download/mod.rs b/crates/cli/commands/src/download/mod.rs index 7cd8de7f42d..aeab3e04d45 100644 --- a/crates/cli/commands/src/download/mod.rs +++ b/crates/cli/commands/src/download/mod.rs @@ -579,6 +579,11 @@ impl> DownloadCo info!(target: "reth::cli", source = %manifest_source, "Fetching snapshot manifest"); let mut manifest = fetch_manifest_from_source(&manifest_source).await?; + eyre::ensure!( + manifest.chain_id == chain_id, + "Snapshot chain ID {} does not match selected chain ID {chain_id}", + manifest.chain_id + ); manifest.base_url = Some(resolve_manifest_base_url(&manifest, &manifest_source)?); info!(target: "reth::cli", From 0e7e335fee60e4941813008c44a6b4acf3c0f395 Mon Sep 17 00:00:00 2001 From: Matthias Seitz Date: Mon, 7 Sep 2026 10:55:41 +0000 Subject: [PATCH 4/8] feat(download): allow retry delay overrides (#27048) Co-authored-by: Matthias Seitz <19890894+mattsse@users.noreply.github.com> (cherry picked from commit 0297d4f9d076d7edc7e330020da6825928c85a9d) --- bin/reth/tests/it/download/mod.rs | 13 +++++++- crates/cli/commands/src/download/archive.rs | 8 +++-- crates/cli/commands/src/download/extract.rs | 23 ++++++++++---- crates/cli/commands/src/download/fetch.rs | 23 +++++++++++--- crates/cli/commands/src/download/mod.rs | 31 ++++++++++++++++++ crates/cli/commands/src/download/session.rs | 35 ++++++++++++++++++++- docs/vocs/docs/pages/cli/reth/download.mdx | 5 +++ 7 files changed, 123 insertions(+), 15 deletions(-) diff --git a/bin/reth/tests/it/download/mod.rs b/bin/reth/tests/it/download/mod.rs index a29d799f7de..db794e30102 100644 --- a/bin/reth/tests/it/download/mod.rs +++ b/bin/reth/tests/it/download/mod.rs @@ -205,9 +205,20 @@ async fn bad_output_checksum_prevents_finalization() { json!(blake3::hash(b"wrong").to_hex().to_string()); let server = snapshot.serve(true).await; let dir = tempfile::tempdir().unwrap(); - let output = download(&server, dir.path(), &["--with-txs-since", "20"]); + let output = + download(&server, dir.path(), &["--with-txs-since", "20", "--retry-backoff", "0ms"]); assert!(!output.status.success()); assert!(String::from_utf8_lossy(&output.stderr).contains("Failed integrity validation")); + assert!(String::from_utf8_lossy(&output.stderr).contains("after 10 attempts")); + assert_eq!( + server + .requests() + .iter() + .filter(|(path, range)| { path == "/snapshot/state.tar.zst" && range.is_none() }) + .count(), + 10, + "each checksum failure must fetch a fresh archive" + ); assert!(!dir.path().join("reth.toml").exists()); assert!(!dir.path().join("db/mdbx.dat").exists()); } diff --git a/crates/cli/commands/src/download/archive.rs b/crates/cli/commands/src/download/archive.rs index f1189486f29..65573eec561 100644 --- a/crates/cli/commands/src/download/archive.rs +++ b/crates/cli/commands/src/download/archive.rs @@ -31,6 +31,7 @@ pub(crate) async fn run_modular_downloads( target_dir: &Path, download_concurrency: usize, cancel_token: CancellationToken, + retry_backoff: Option, ) -> Result<()> { let download_cache_dir = target_dir.join(DOWNLOAD_CACHE_DIR); fs::create_dir_all(&download_cache_dir)?; @@ -45,7 +46,8 @@ pub(crate) async fn run_modular_downloads( Some(Arc::clone(&shared)), Some(DownloadRequestLimiter::new(download_concurrency)), cancel_token, - ); + ) + .with_retry_backoff(retry_backoff); let ctx = ArchiveProcessContext::new(target_dir.to_path_buf(), Some(download_cache_dir), session); @@ -185,7 +187,9 @@ impl ArchiveProcessor { if attempt >= MAX_DOWNLOAD_RETRIES { state = ArchiveAttemptState::Fail; } else { - std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS)); + std::thread::sleep( + self.ctx.session().retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)), + ); attempt += 1; state = ArchiveAttemptState::RunAttempt; } diff --git a/crates/cli/commands/src/download/extract.rs b/crates/cli/commands/src/download/extract.rs index 29bb49089d4..bc5722cb19d 100644 --- a/crates/cli/commands/src/download/extract.rs +++ b/crates/cli/commands/src/download/extract.rs @@ -300,7 +300,9 @@ pub(crate) fn streaming_download_and_extract( } last_error = Some(err); if attempt < MAX_DOWNLOAD_RETRIES { - std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS)); + std::thread::sleep( + session.retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)), + ); } continue; } @@ -342,7 +344,9 @@ pub(crate) fn streaming_download_and_extract( } last_error = Some(error); if attempt < MAX_DOWNLOAD_RETRIES { - std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS)); + std::thread::sleep( + session.retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)), + ); } } } @@ -417,13 +421,15 @@ fn blocking_download_and_extract( resumable: bool, request_limiter: Option>, cancel_token: CancellationToken, + retry_backoff: Option, ) -> Result<()> { let format = CompressionFormat::from_url(url)?; if let Ok(parsed_url) = Url::parse(url) && parsed_url.scheme() == "file" { - let session = DownloadSession::new(shared, request_limiter, cancel_token); + let session = DownloadSession::new(shared, request_limiter, cancel_token) + .with_retry_backoff(retry_backoff); let file_path = parsed_url .to_file_path() .map_err(|_| eyre::eyre!("Invalid file:// URL path: {}", url))?; @@ -437,14 +443,17 @@ fn blocking_download_and_extract( url, format, target_dir, - DownloadSession::new(shared, Some(request_limiter), cancel_token), + DownloadSession::new(shared, Some(request_limiter), cancel_token) + .with_retry_backoff(retry_backoff), ) } else if resumable { let session = - DownloadSession::new(shared, Some(DownloadRequestLimiter::new(1)), cancel_token); + DownloadSession::new(shared, Some(DownloadRequestLimiter::new(1)), cancel_token) + .with_retry_backoff(retry_backoff); download_and_extract(url, format, target_dir, session) } else { - let session = DownloadSession::new(shared, None, cancel_token); + let session = + DownloadSession::new(shared, None, cancel_token).with_retry_backoff(retry_backoff); let result = streaming_download_and_extract(url, format, target_dir, &session); if result.is_ok() { session.record_archive_output_complete(0); @@ -465,6 +474,7 @@ pub(crate) async fn stream_and_extract( resumable: bool, request_limiter: Option>, cancel_token: CancellationToken, + retry_backoff: Option, ) -> Result<()> { let target_dir = target_dir.to_path_buf(); let url = url.to_string(); @@ -476,6 +486,7 @@ pub(crate) async fn stream_and_extract( resumable, request_limiter, cancel_token, + retry_backoff, ) }) .await??; diff --git a/crates/cli/commands/src/download/fetch.rs b/crates/cli/commands/src/download/fetch.rs index 83eea6ebbcd..db0300344dd 100644 --- a/crates/cli/commands/src/download/fetch.rs +++ b/crates/cli/commands/src/download/fetch.rs @@ -247,9 +247,12 @@ impl ArchiveFetcher { if attempt < max_download_retries { info!(target: "reth::cli", file = %self.paths.file_name(), - "Download failed, retrying in {RETRY_BACKOFF_SECS}s..." + retry_delay = ?self.session.retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)), + "Download failed, retrying" + ); + std::thread::sleep( + self.session.retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)), ); - std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS)); } continue; } @@ -328,9 +331,12 @@ impl ArchiveFetcher { if attempt < max_download_retries { info!(target: "reth::cli", file = %self.paths.file_name(), - "Download interrupted, retrying in {RETRY_BACKOFF_SECS}s..." + retry_delay = ?self.session.retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)), + "Download interrupted, retrying" + ); + std::thread::sleep( + self.session.retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)), ); - std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS)); } continue; } @@ -550,6 +556,8 @@ struct SegmentedWorkerContext<'a> { request_limiter: &'a DownloadRequestLimiter, /// Cancellation token shared by the whole command. cancel_token: &'a CancellationToken, + /// Session carrying the retry-delay override. + session: &'a DownloadSession, } impl SegmentedDownload { @@ -588,6 +596,7 @@ impl SegmentedDownload { let worker_context = SegmentedWorkerContext { url, part_path: paths.part_path(), + session: &session, shared, request_limiter: request_limiter.as_ref(), cancel_token, @@ -669,6 +678,7 @@ impl SegmentedDownload { &piece_progress_bytes, context.request_limiter, context.cancel_token, + context.session, ) { state.note_terminal_failure(); terminal_failure.record(error); @@ -691,6 +701,7 @@ impl SegmentedDownload { piece_progress_bytes: &AtomicU64, request_limiter: &DownloadRequestLimiter, cancel_token: &CancellationToken, + session: &DownloadSession, ) -> Result<()> { for attempt in 1..=SEGMENT_RETRY_ATTEMPTS { if cancel_token.is_cancelled() { @@ -711,7 +722,9 @@ impl SegmentedDownload { Err(PieceAttemptFailure::Retryable { error: _, throttled }) if attempt < SEGMENT_RETRY_ATTEMPTS => { - std::thread::sleep(piece_retry_backoff(attempt, throttled)); + std::thread::sleep( + session.retry_delay(piece_retry_backoff(attempt, throttled)), + ); } Err(PieceAttemptFailure::Retryable { error, .. }) => return Err(error), Err(PieceAttemptFailure::Terminal(error)) => return Err(error), diff --git a/crates/cli/commands/src/download/mod.rs b/crates/cli/commands/src/download/mod.rs index aeab3e04d45..411fa824cba 100644 --- a/crates/cli/commands/src/download/mod.rs +++ b/crates/cli/commands/src/download/mod.rs @@ -118,6 +118,7 @@ use std::{ collections::BTreeMap, path::{Path, PathBuf}, sync::{Arc, OnceLock}, + time::Duration, }; use tracing::info; use tui::{run_selector, SelectorOutput}; @@ -442,6 +443,14 @@ pub struct DownloadCommand { #[arg(long, default_value_t = MAX_CONCURRENT_DOWNLOADS)] download_concurrency: usize, + /// Override the delay between retry attempts (for example, 500ms or 5s). + /// + /// Applies to requests, extraction, output verification, and segmented downloads. + /// By default, retries wait five seconds; segmented requests use adaptive backoff. + /// This does not change the number of attempts. + #[arg(long, value_name = "DURATION", value_parser = reth_cli_util::parse_duration_from_secs_or_ms)] + retry_backoff: Option, + /// List available snapshots and exit. /// /// Queries the snapshots API and prints all available snapshots for the selected chain, @@ -491,6 +500,7 @@ impl> DownloadCo self.resumable, Some(request_limiter), cancel_token.clone(), + self.retry_backoff, ) .await?; info!(target: "reth::cli", "Snapshot downloaded and extracted successfully"); @@ -534,6 +544,7 @@ impl> DownloadCo target_dir, self.download_concurrency.max(1), cancel_token.clone(), + self.retry_backoff, ) .await?; @@ -1228,6 +1239,26 @@ mod tests { assert_eq!(defaults.long_help, Some("Custom help for snapshots".to_string())); } + #[test] + fn test_download_retry_backoff() { + let parse = |args: Vec<&str>| { + CommandParser::>::try_parse_from(args) + }; + assert_eq!(parse(vec!["reth"]).unwrap().args.retry_backoff, None); + for (value, expected) in [ + ("0ms", Duration::ZERO), + ("250ms", Duration::from_millis(250)), + ("2s", Duration::from_secs(2)), + ] { + assert_eq!( + parse(vec!["reth", "--retry-backoff", value]).unwrap().args.retry_backoff, + Some(expected) + ); + } + assert!(parse(vec!["reth", "--retry-backoff=-1s"]).is_err()); + assert!(parse(vec!["reth", "--retry-backoff", "invalid"]).is_err()); + } + #[test] fn test_download_resumable_defaults_to_true() { let args = diff --git a/crates/cli/commands/src/download/session.rs b/crates/cli/commands/src/download/session.rs index 23412e23d4a..bc1e5a941af 100644 --- a/crates/cli/commands/src/download/session.rs +++ b/crates/cli/commands/src/download/session.rs @@ -4,6 +4,7 @@ use reth_cli_util::cancellation::CancellationToken; use std::{ path::{Path, PathBuf}, sync::Arc, + time::Duration, }; /// Shared state for one run of `reth download`. @@ -15,6 +16,8 @@ pub(crate) struct DownloadSession { request_limiter: Option>, /// Cancellation token shared by the whole command. cancel_token: CancellationToken, + /// Optional fixed delay overriding the default retry backoff. + retry_backoff: Option, } impl DownloadSession { @@ -24,7 +27,18 @@ impl DownloadSession { request_limiter: Option>, cancel_token: CancellationToken, ) -> Self { - Self { progress, request_limiter, cancel_token } + Self { progress, request_limiter, cancel_token, retry_backoff: None } + } + + /// Overrides the delay between retry attempts for this download. + pub(crate) const fn with_retry_backoff(mut self, retry_backoff: Option) -> Self { + self.retry_backoff = retry_backoff; + self + } + + /// Returns the configured delay, preserving the caller's default when no override is set. + pub(crate) fn retry_delay(&self, default: Duration) -> Duration { + self.retry_backoff.unwrap_or(default) } /// Returns the shared progress tracker, if this flow uses one. @@ -98,3 +112,22 @@ impl ArchiveProcessContext { &self.session } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retry_delay_preserves_defaults_or_uses_override() { + let session = DownloadSession::new(None, None, CancellationToken::new()); + for default in [Duration::from_secs(2), Duration::from_secs(5), Duration::from_secs(40)] { + assert_eq!(session.retry_delay(default), default); + for delay in [Duration::ZERO, Duration::from_millis(250)] { + assert_eq!( + session.clone().with_retry_backoff(Some(delay)).retry_delay(default), + delay + ); + } + } + } +} diff --git a/docs/vocs/docs/pages/cli/reth/download.mdx b/docs/vocs/docs/pages/cli/reth/download.mdx index 2dc38158f09..6817a3c413b 100644 --- a/docs/vocs/docs/pages/cli/reth/download.mdx +++ b/docs/vocs/docs/pages/cli/reth/download.mdx @@ -223,6 +223,11 @@ Storage: [default: 8] + --retry-backoff + Override the delay between retry attempts (for example, 500ms or 5s). + + Applies to requests, extraction, output verification, and segmented downloads. By default, retries wait five seconds; segmented requests use adaptive backoff. This does not change the number of attempts. + --list List available snapshots and exit. From acebfe7ff6fe589faa5034a023029e660e61f55f Mon Sep 17 00:00:00 2001 From: Matthias Seitz Date: Mon, 14 Sep 2026 14:02:31 +0000 Subject: [PATCH 5/8] fix(download): restore custom static files directory handling (#27065) (cherry picked from commit 4b01c866c8743cb1a26d0786ea858a196a8d15f2) --- bin/reth/tests/it/download/mod.rs | 77 +++++- crates/cli/commands/src/download/archive.rs | 13 +- crates/cli/commands/src/download/extract.rs | 264 ++++++++++++++++--- crates/cli/commands/src/download/mod.rs | 27 +- crates/cli/commands/src/download/planning.rs | 5 +- crates/cli/commands/src/download/session.rs | 10 +- crates/cli/commands/src/download/verify.rs | 47 +++- 7 files changed, 384 insertions(+), 59 deletions(-) diff --git a/bin/reth/tests/it/download/mod.rs b/bin/reth/tests/it/download/mod.rs index db794e30102..d76377e4093 100644 --- a/bin/reth/tests/it/download/mod.rs +++ b/bin/reth/tests/it/download/mod.rs @@ -96,6 +96,15 @@ fn success(output: Output) -> String { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn manifest_selection_reuse_and_repair() { + check_manifest_selection_reuse_and_repair(false).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn custom_static_files_selection_reuse_and_repair() { + check_manifest_selection_reuse_and_repair(true).await; +} + +async fn check_manifest_selection_reuse_and_repair(custom_static_files: bool) { let snapshot = Snapshot::new(); let state = &snapshot.files["/snapshot/state.tar.zst"]; let offset = state.len() / 2; @@ -103,19 +112,28 @@ async fn manifest_selection_reuse_and_repair() { fs::create_dir(dir.path().join(".download-cache")).unwrap(); fs::write(dir.path().join(".download-cache/state.tar.zst.part"), &state[..offset]).unwrap(); let server = snapshot.serve(true).await; - let args = ["--with-txs-since", "10"]; + let custom_dir = tempfile::tempdir().unwrap(); + let static_files = if custom_static_files { + custom_dir.path().join("nested/static") + } else { + dir.path().join("static_files") + }; + let mut args = vec!["--with-txs-since", "10"]; + if custom_static_files { + args.extend(["--datadir.static-files", static_files.to_str().unwrap()]); + } success(download(&server, dir.path(), &args)); for (path, expected) in [ - ("db/snapshot", "state.tar.zst"), - ("static_files/headers", "headers.tar.zst"), - ("static_files/txs-1", "txs-1.tar.zst"), + (dir.path().join("db/snapshot"), "state.tar.zst"), + (static_files.join("headers"), "headers.tar.zst"), + (static_files.join("txs-1"), "txs-1.tar.zst"), ] { - assert_eq!(fs::read(dir.path().join(path)).unwrap(), expected.as_bytes()); + assert_eq!(fs::read(path).unwrap(), expected.as_bytes()); } assert!(server.requests().iter().any(|(path, range)| { path.ends_with("state.tar.zst") && range.as_deref() == Some(&format!("bytes={offset}-")) })); - assert!(!dir.path().join("static_files/txs-0").exists()); + assert!(!static_files.join("txs-0").exists()); assert!(!server.requests().iter().any(|(path, _)| path.ends_with("txs-0.tar.zst"))); let config: toml::Value = toml::from_str(&fs::read_to_string(dir.path().join("reth.toml")).unwrap()).unwrap(); @@ -126,15 +144,23 @@ async fn manifest_selection_reuse_and_repair() { assert_eq!(server.requests(), vec![("/snapshot/manifest.json".into(), None)]); // Same-size corruption must be detected by checksum, not just the file length. - fs::write(dir.path().join("static_files/txs-1"), b"bad-1.tar.zst").unwrap(); + fs::write(static_files.join("txs-1"), b"bad-1.tar.zst").unwrap(); server.clear_requests(); success(download(&server, dir.path(), &args)); - assert_eq!(fs::read(dir.path().join("static_files/txs-1")).unwrap(), b"txs-1.tar.zst"); + assert_eq!(fs::read(static_files.join("txs-1")).unwrap(), b"txs-1.tar.zst"); assert!(server.requests().iter().any(|(path, _)| path.ends_with("txs-1.tar.zst"))); assert!(server .requests() .iter() .all(|(path, _)| path.ends_with("manifest.json") || path.ends_with("txs-1.tar.zst"))); + if custom_static_files { + assert!(!dir.path().join("static_files").exists()); + } + fs::write(static_files.join("stale"), b"stale").unwrap(); + args.push("--force"); + success(download(&server, dir.path(), &args)); + assert!(!static_files.join("stale").exists()); + assert_eq!(fs::read(static_files.join("headers")).unwrap(), b"headers.tar.zst"); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -183,6 +209,41 @@ async fn legacy_streaming_and_resume() { } } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn custom_static_files_legacy_http_and_local_archives() { + for local in [false, true] { + let bytes = archive_bytes("./static_files/nested/headers", b"snapshot contents"); + let archive_name = "headers.tar.zst"; + let source = tempfile::tempdir().unwrap(); + let archive_path = source.path().join(archive_name); + fs::write(&archive_path, &bytes).unwrap(); + let server = + SnapshotServer::start(BTreeMap::from([(format!("/{archive_name}"), bytes)]), true) + .await; + let url = if local { + format!("file://{}", archive_path.display()) + } else { + format!("{}/{archive_name}", server.url) + }; + let dir = tempfile::tempdir().unwrap(); + let static_dir = tempfile::tempdir().unwrap(); + reth_ok(&[ + "download", + "--url", + &url, + "--datadir", + dir.path().to_str().unwrap(), + "--datadir.static-files", + static_dir.path().to_str().unwrap(), + ]); + assert_eq!( + fs::read(static_dir.path().join("nested/headers")).unwrap(), + b"snapshot contents" + ); + assert!(!dir.path().join("static_files").exists()); + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn wrong_chain_is_rejected_before_force_removes_data() { let mut snapshot = Snapshot::new(); diff --git a/crates/cli/commands/src/download/archive.rs b/crates/cli/commands/src/download/archive.rs index 65573eec561..c4e46e1ddbf 100644 --- a/crates/cli/commands/src/download/archive.rs +++ b/crates/cli/commands/src/download/archive.rs @@ -29,6 +29,7 @@ const DOWNLOAD_CACHE_DIR: &str = ".download-cache"; pub(crate) async fn run_modular_downloads( planned_downloads: PlannedDownloads, target_dir: &Path, + static_files_dir: Option<&Path>, download_concurrency: usize, cancel_token: CancellationToken, retry_backoff: Option, @@ -48,8 +49,12 @@ pub(crate) async fn run_modular_downloads( cancel_token, ) .with_retry_backoff(retry_backoff); - let ctx = - ArchiveProcessContext::new(target_dir.to_path_buf(), Some(download_cache_dir), session); + let ctx = ArchiveProcessContext::new( + target_dir.to_path_buf(), + static_files_dir.map(Path::to_path_buf), + Some(download_cache_dir), + session, + ); ModularDownloadJob::new(ctx, download_concurrency).run(planned_downloads).await } @@ -220,7 +225,7 @@ impl ArchiveProcessor { /// Returns the verifier for this archive's output files. fn output_verifier(&self) -> OutputVerifier<'_> { - OutputVerifier::new(self.ctx.target_dir()) + OutputVerifier::new(self.ctx.target_dir(), self.ctx.static_files_dir()) } /// Returns `true` if this archive can be reused from existing verified outputs. @@ -304,6 +309,7 @@ impl ArchiveProcessor { &self.archive().url, format, self.ctx.target_dir(), + self.ctx.static_files_dir(), self.ctx.session(), ) } @@ -316,6 +322,7 @@ impl ArchiveProcessor { file, format, self.ctx.target_dir(), + self.ctx.static_files_dir(), Some(&mut extraction_progress), ); extraction_progress.finish(); diff --git a/crates/cli/commands/src/download/extract.rs b/crates/cli/commands/src/download/extract.rs index bc5722cb19d..ba405ff6fed 100644 --- a/crates/cli/commands/src/download/extract.rs +++ b/crates/cli/commands/src/download/extract.rs @@ -2,7 +2,7 @@ use super::{ fetch::{ArchiveFetcher, DownloadedArchive}, progress::{ ArchiveExtractionProgress, ArchiveExtractionProgressHandle, DownloadProgress, - DownloadRequestLimiter, ProgressReader, SharedProgress, SharedProgressReader, + DownloadRequestLimiter, ProgressReader, SharedProgressReader, }, session::DownloadSession, MAX_DOWNLOAD_RETRIES, RETRY_BACKOFF_SECS, @@ -67,6 +67,7 @@ fn extract_archive( total_size: u64, format: CompressionFormat, target_dir: &Path, + static_files_dir: Option<&Path>, cancel_token: CancellationToken, ) -> Result<()> { let progress_reader = ProgressReader::new(reader, total_size, cancel_token); @@ -74,11 +75,11 @@ fn extract_archive( match format { CompressionFormat::Lz4 => { let decoder = Decoder::new(progress_reader)?; - Archive::new(decoder).unpack(target_dir)?; + unpack_archive(Archive::new(decoder), target_dir, static_files_dir, None)?; } CompressionFormat::Zstd => { let decoder = ZstdDecoder::new(progress_reader)?; - Archive::new(decoder).unpack(target_dir)?; + unpack_archive(Archive::new(decoder), target_dir, static_files_dir, None)?; } } @@ -91,14 +92,25 @@ pub(crate) fn extract_archive_raw( reader: R, format: CompressionFormat, target_dir: &Path, + static_files_dir: Option<&Path>, progress: Option<&mut ArchiveExtractionProgress>, ) -> Result<()> { match format { CompressionFormat::Lz4 => { - unpack_archive(Archive::new(Decoder::new(reader)?), target_dir, progress)?; + unpack_archive( + Archive::new(Decoder::new(reader)?), + target_dir, + static_files_dir, + progress, + )?; } CompressionFormat::Zstd => { - unpack_archive(Archive::new(ZstdDecoder::new(reader)?), target_dir, progress)?; + unpack_archive( + Archive::new(ZstdDecoder::new(reader)?), + target_dir, + static_files_dir, + progress, + )?; } } @@ -108,8 +120,13 @@ pub(crate) fn extract_archive_raw( fn unpack_archive( mut archive: Archive, target_dir: &Path, + static_files_dir: Option<&Path>, mut progress: Option<&mut ArchiveExtractionProgress>, ) -> Result<()> { + if static_files_dir.is_none() && progress.is_none() { + archive.unpack(target_dir)?; + return Ok(()) + } let entries = archive.entries().wrap_err_with(|| { format!("failed to read archive entries for `{}`", target_dir.display()) })?; @@ -118,29 +135,105 @@ fn unpack_archive( let mut entry = entry.wrap_err_with(|| { format!("failed to read archive entry for `{}`", target_dir.display()) })?; - extract_entry_with_progress(&mut entry, target_dir, progress.as_deref_mut())?; + extract_entry_with_progress( + &mut entry, + target_dir, + static_files_dir, + progress.as_deref_mut(), + )?; } Ok(()) } +/// Returns the path within the static files directory, accepting tar's optional `./` prefix. +pub(crate) fn static_file_relative_path(path: &Path) -> Option<&Path> { + path.strip_prefix(".").unwrap_or(path).strip_prefix("static_files").ok() +} + +/// Extracts static files beneath their configured root without allowing archive paths or links +/// to escape it. +fn unpack_static_file( + entry: &mut tar::Entry<'_, R>, + static_files_dir: &Path, + relative_path: &Path, +) -> Result<()> { + eyre::ensure!( + relative_path + .components() + .all(|part| matches!(part, Component::Normal(_) | Component::CurDir)), + "Invalid static file archive path: {}", + relative_path.display() + ); + let entry_type = entry.header().entry_type(); + eyre::ensure!( + entry_type.is_file() || entry_type.is_dir(), + "Unsupported static file archive entry" + ); + fs::create_dir_all(static_files_dir)?; + let root = static_files_dir.to_path_buf(); + let dest = root.join(relative_path); + // Check each existing ancestor before creating directories or replacing an output. + let mut current = root.clone(); + for part in relative_path.components() { + current.push(part); + if let Ok(metadata) = std::fs::symlink_metadata(¤t) { + eyre::ensure!( + !metadata.file_type().is_symlink(), + "Static file archive path contains a symlink" + ); + } + } + if relative_path.as_os_str().is_empty() { + return Ok(()) + } + if let Some(parent) = dest.parent() { + fs::create_dir_all(parent)?; + } + entry.unpack(dest)?; + Ok(()) +} + +fn unpack_entry( + entry: &mut tar::Entry<'_, R>, + target_dir: &Path, + static_files_dir: Option<&Path>, +) -> Result<()> { + // A non-static entry could otherwise plant a symlink at a nested custom root before + // extraction or retry cleanup uses it. + eyre::ensure!( + static_files_dir.is_none() || !entry.header().entry_type().is_symlink(), + "Archive symlinks are unsupported with a custom static files directory" + ); + let path = entry.path()?.into_owned(); + if let Some(static_files_dir) = static_files_dir && + let Some(relative_path) = static_file_relative_path(&path) + { + unpack_static_file(entry, static_files_dir, relative_path) + } else { + entry.unpack_in(target_dir)?; + Ok(()) + } +} + fn extract_entry_with_progress( entry: &mut tar::Entry<'_, R>, target_dir: &Path, + static_files_dir: Option<&Path>, progress: Option<&mut ArchiveExtractionProgress>, ) -> Result<()> { let size = entry.header().entry_size().unwrap_or(0); let entry_type = entry.header().entry_type(); if !entry_type.is_file() || size == 0 { - entry.unpack_in(target_dir).wrap_err_with(|| { + unpack_entry(entry, target_dir, static_files_dir).wrap_err_with(|| { format!("failed to extract archive into `{}`", target_dir.display()) })?; return Ok(()) } if size < STREAMING_EXTRACTION_PROGRESS_MIN_FILE_SIZE { - entry.unpack_in(target_dir).wrap_err_with(|| { + unpack_entry(entry, target_dir, static_files_dir).wrap_err_with(|| { format!("failed to extract archive into `{}`", target_dir.display()) })?; if let Some(progress) = progress { @@ -150,14 +243,14 @@ fn extract_entry_with_progress( } let Some(progress_handle) = progress.as_ref().and_then(|progress| progress.handle()) else { - entry.unpack_in(target_dir).wrap_err_with(|| { + unpack_entry(entry, target_dir, static_files_dir).wrap_err_with(|| { format!("failed to extract archive into `{}`", target_dir.display()) })?; return Ok(()) }; - let Some(entry_path) = entry_destination_path(entry, target_dir)? else { - entry.unpack_in(target_dir).wrap_err_with(|| { + let Some(entry_path) = entry_destination_path(entry, target_dir, static_files_dir)? else { + unpack_entry(entry, target_dir, static_files_dir).wrap_err_with(|| { format!("failed to extract archive into `{}`", target_dir.display()) })?; return Ok(()) @@ -165,8 +258,7 @@ fn extract_entry_with_progress( let stop = Arc::new(AtomicBool::new(false)); let monitor = spawn_extraction_progress_monitor(entry_path, progress_handle, Arc::clone(&stop)); - let unpack_result = entry - .unpack_in(target_dir) + let unpack_result = unpack_entry(entry, target_dir, static_files_dir) .wrap_err_with(|| format!("failed to extract archive into `{}`", target_dir.display())); stop.store(true, Ordering::Relaxed); @@ -180,9 +272,18 @@ fn extract_entry_with_progress( fn entry_destination_path( entry: &tar::Entry<'_, R>, target_dir: &Path, + static_files_dir: Option<&Path>, ) -> Result> { let mut file_dst = target_dir.to_path_buf(); let path = entry.path().wrap_err("invalid path in archive entry")?; + let path = if let Some(static_files_dir) = static_files_dir && + let Some(relative_path) = static_file_relative_path(&path) + { + file_dst = static_files_dir.to_path_buf(); + relative_path + } else { + path.as_ref() + }; for part in path.components() { match part { @@ -231,7 +332,12 @@ fn record_extracted_file_bytes( } /// Extracts a snapshot from a local file. -fn extract_from_file(path: &Path, format: CompressionFormat, target_dir: &Path) -> Result<()> { +fn extract_from_file( + path: &Path, + format: CompressionFormat, + target_dir: &Path, + static_files_dir: Option<&Path>, +) -> Result<()> { let file = std::fs::File::open(path)?; let total_size = file.metadata()?.len(); info!(target: "reth::cli", @@ -240,7 +346,14 @@ fn extract_from_file(path: &Path, format: CompressionFormat, target_dir: &Path) "Extracting local archive" ); let start = Instant::now(); - extract_archive(file, total_size, format, target_dir, CancellationToken::new())?; + extract_archive( + file, + total_size, + format, + target_dir, + static_files_dir, + CancellationToken::new(), + )?; info!(target: "reth::cli", file = %path.display(), elapsed = %DownloadProgress::format_duration(start.elapsed()), @@ -256,11 +369,12 @@ pub(crate) fn streaming_download_and_extract( url: &str, format: CompressionFormat, target_dir: &Path, + static_files_dir: Option<&Path>, session: &DownloadSession, ) -> Result<()> { if let Some(path) = archive_file_url_path(url)? { let size = path.metadata()?.len(); - extract_from_file(&path, format, target_dir)?; + extract_from_file(&path, format, target_dir, static_files_dir)?; session.record_archive_output_complete(size); return Ok(()) } @@ -318,7 +432,7 @@ pub(crate) fn streaming_download_and_extract( let result = if let Some(progress) = shared { let reader = SharedProgressReader { inner: response, progress: Arc::clone(progress) }; - extract_archive_raw(reader, format, target_dir, None) + extract_archive_raw(reader, format, target_dir, static_files_dir, None) } else { let total_size = response.content_length().unwrap_or(0); extract_archive( @@ -326,6 +440,7 @@ pub(crate) fn streaming_download_and_extract( total_size, format, target_dir, + static_files_dir, session.cancel_token().clone(), ) }; @@ -375,6 +490,7 @@ fn download_and_extract( url: &str, format: CompressionFormat, target_dir: &Path, + static_files_dir: Option<&Path>, session: DownloadSession, ) -> Result<()> { let quiet = session.progress().is_some(); @@ -394,9 +510,16 @@ fn download_and_extract( let file = fs::open(&downloaded_path)?; if quiet { - extract_archive_raw(file, format, target_dir, None)?; + extract_archive_raw(file, format, target_dir, static_files_dir, None)?; } else { - extract_archive(file, total_size, format, target_dir, session.cancel_token().clone())?; + extract_archive( + file, + total_size, + format, + target_dir, + static_files_dir, + session.cancel_token().clone(), + )?; info!(target: "reth::cli", file = %file_name, "Extraction complete" @@ -417,7 +540,7 @@ fn download_and_extract( fn blocking_download_and_extract( url: &str, target_dir: &Path, - shared: Option>, + static_files_dir: Option<&Path>, resumable: bool, request_limiter: Option>, cancel_token: CancellationToken, @@ -428,12 +551,12 @@ fn blocking_download_and_extract( if let Ok(parsed_url) = Url::parse(url) && parsed_url.scheme() == "file" { - let session = DownloadSession::new(shared, request_limiter, cancel_token) + let session = DownloadSession::new(None, request_limiter, cancel_token) .with_retry_backoff(retry_backoff); let file_path = parsed_url .to_file_path() .map_err(|_| eyre::eyre!("Invalid file:// URL path: {}", url))?; - let result = extract_from_file(&file_path, format, target_dir); + let result = extract_from_file(&file_path, format, target_dir, static_files_dir); if result.is_ok() { session.record_archive_output_complete(file_path.metadata()?.len()); } @@ -443,18 +566,20 @@ fn blocking_download_and_extract( url, format, target_dir, - DownloadSession::new(shared, Some(request_limiter), cancel_token) + static_files_dir, + DownloadSession::new(None, Some(request_limiter), cancel_token) .with_retry_backoff(retry_backoff), ) } else if resumable { let session = - DownloadSession::new(shared, Some(DownloadRequestLimiter::new(1)), cancel_token) + DownloadSession::new(None, Some(DownloadRequestLimiter::new(1)), cancel_token) .with_retry_backoff(retry_backoff); - download_and_extract(url, format, target_dir, session) + download_and_extract(url, format, target_dir, static_files_dir, session) } else { let session = - DownloadSession::new(shared, None, cancel_token).with_retry_backoff(retry_backoff); - let result = streaming_download_and_extract(url, format, target_dir, &session); + DownloadSession::new(None, None, cancel_token).with_retry_backoff(retry_backoff); + let result = + streaming_download_and_extract(url, format, target_dir, static_files_dir, &session); if result.is_ok() { session.record_archive_output_complete(0); } @@ -464,25 +589,25 @@ fn blocking_download_and_extract( /// Downloads and extracts a snapshot archive asynchronously. /// -/// When `shared` is provided, download progress is reported to the shared -/// counter for aggregated display. Otherwise uses a local progress bar. +/// Download progress is reported using a local progress bar. /// When `resumable` is true, uses two-phase download with `.part` files. pub(crate) async fn stream_and_extract( url: &str, target_dir: &Path, - shared: Option>, + static_files_dir: Option<&Path>, resumable: bool, request_limiter: Option>, cancel_token: CancellationToken, retry_backoff: Option, ) -> Result<()> { let target_dir = target_dir.to_path_buf(); + let static_files_dir = static_files_dir.map(Path::to_path_buf); let url = url.to_string(); task::spawn_blocking(move || { blocking_download_and_extract( &url, &target_dir, - shared, + static_files_dir.as_deref(), resumable, request_limiter, cancel_token, @@ -498,6 +623,83 @@ pub(crate) async fn stream_and_extract( mod tests { use super::*; + #[test] + fn custom_static_root_cannot_be_redirected_by_archive_symlinks() { + for custom_path in ["custom", "custom/nested"] { + let target = tempfile::tempdir().unwrap(); + let outside = tempfile::tempdir().unwrap(); + fs::create_dir_all(outside.path().join("nested")).unwrap(); + for path in ["headers", "nested/headers"] { + fs::write(outside.path().join(path), b"keep").unwrap(); + } + let custom = target.path().join(custom_path); + let mut archive = tar::Builder::new(Vec::new()); + let mut link = tar::Header::new_gnu(); + link.set_entry_type(tar::EntryType::Symlink); + link.set_size(0); + link.set_mode(0o777); + archive.append_link(&mut link, "custom", outside.path()).unwrap(); + let mut file = tar::Header::new_gnu(); + file.set_size(4); + file.set_mode(0o644); + file.set_cksum(); + archive.append_data(&mut file, "static_files/headers", b"data".as_slice()).unwrap(); + let tar = archive.into_inner().unwrap(); + let err = + unpack_archive(Archive::new(tar.as_slice()), target.path(), Some(&custom), None) + .unwrap_err(); + assert!(format!("{err:#}").contains("Archive symlinks are unsupported")); + let outputs = [super::super::manifest::OutputFileChecksum { + path: "static_files/headers".into(), + size: 4, + blake3: String::new(), + }]; + super::super::verify::OutputVerifier::new(target.path(), Some(&custom)) + .cleanup(&outputs); + for path in ["headers", "nested/headers"] { + assert_eq!(fs::read(outside.path().join(path)).unwrap(), b"keep"); + } + } + } + + #[test] + fn remap_static_files_in_both_compression_formats() { + let mut archive = tar::Builder::new(Vec::new()); + let mut directory = tar::Header::new_gnu(); + directory.set_entry_type(tar::EntryType::Directory); + directory.set_size(0); + directory.set_mode(0o755); + directory.set_cksum(); + archive.append_data(&mut directory, "./static_files/", std::io::empty()).unwrap(); + for path in ["./static_files/nested/headers", "db/data"] { + let mut header = tar::Header::new_gnu(); + header.set_size(4); + header.set_mode(0o644); + header.set_cksum(); + archive.append_data(&mut header, path, b"data".as_slice()).unwrap(); + } + let tar = archive.into_inner().unwrap(); + for format in [CompressionFormat::Lz4, CompressionFormat::Zstd] { + let bytes = match format { + CompressionFormat::Lz4 => { + let mut encoder = lz4::EncoderBuilder::new().build(Vec::new()).unwrap(); + std::io::copy(&mut tar.as_slice(), &mut encoder).unwrap(); + let (bytes, result) = encoder.finish(); + result.unwrap(); + bytes + } + CompressionFormat::Zstd => zstd::encode_all(tar.as_slice(), 0).unwrap(), + }; + let target = tempfile::tempdir().unwrap(); + let custom = tempfile::tempdir().unwrap(); + extract_archive_raw(bytes.as_slice(), format, target.path(), Some(custom.path()), None) + .unwrap(); + assert_eq!(fs::read(custom.path().join("nested/headers")).unwrap(), b"data"); + assert_eq!(fs::read(target.path().join("db/data")).unwrap(), b"data"); + assert!(!target.path().join("static_files").exists()); + } + } + #[test] fn test_compression_format_detection() { assert!(matches!( diff --git a/crates/cli/commands/src/download/mod.rs b/crates/cli/commands/src/download/mod.rs index 411fa824cba..3726743f6bc 100644 --- a/crates/cli/commands/src/download/mod.rs +++ b/crates/cli/commands/src/download/mod.rs @@ -475,6 +475,11 @@ impl> DownloadCo return Ok(None); } + let data_dir = self.env.datadir.clone().resolve_datadir(chain); + let static_files_dir = data_dir.static_files(); + let static_files_dir = (static_files_dir != data_dir.data_dir().join("static_files")) + .then_some(static_files_dir); + // Legacy single-URL mode: download one archive and extract it if let Some(ref url) = self.url { let cancel_token = CancellationToken::new(); @@ -482,7 +487,7 @@ impl> DownloadCo let data_dir = self.env.datadir.clone().resolve_datadir(chain); let target_dir = data_dir.data_dir(); if self.force { - clear_existing_datadir(target_dir)?; + clear_existing_datadir(target_dir, static_files_dir.as_deref())?; } fs::create_dir_all(target_dir)?; @@ -496,7 +501,7 @@ impl> DownloadCo stream_and_extract( url, data_dir.data_dir(), - None, + static_files_dir.as_deref(), self.resumable, Some(request_limiter), cancel_token.clone(), @@ -522,10 +527,11 @@ impl> DownloadCo let cancel_token = CancellationToken::new(); let _cancel_guard = cancel_token.drop_guard(); if self.force { - clear_existing_datadir(target_dir)?; + clear_existing_datadir(target_dir, static_files_dir.as_deref())?; } fs::create_dir_all(target_dir)?; - let startup_summary = summarize_download_startup(&planned.archives, target_dir)?; + let startup_summary = + summarize_download_startup(&planned.archives, target_dir, static_files_dir.as_deref())?; info!(target: "reth::cli", reusable = startup_summary.reusable, needs_download = startup_summary.needs_download, @@ -542,6 +548,7 @@ impl> DownloadCo run_modular_downloads( planned, target_dir, + static_files_dir.as_deref(), self.download_concurrency.max(1), cancel_token.clone(), self.retry_backoff, @@ -916,14 +923,14 @@ fn selection_from_prune_mode(mode: Option, snapshot_block: u64) -> Co } /// Removes existing snapshot data that is managed by `reth download`. -fn clear_existing_datadir(target_dir: &Path) -> Result<()> { - if !target_dir.try_exists()? { - return Ok(()); - } - +fn clear_existing_datadir(target_dir: &Path, static_files_dir: Option<&Path>) -> Result<()> { info!(target: "reth::cli", dir = ?target_dir, "Clearing existing snapshot data"); for entry in FORCE_REMOVED_DATADIR_PATHS { - let path = target_dir.join(entry); + let path = if *entry == "static_files" { + static_files_dir.map_or_else(|| target_dir.join(entry), Path::to_path_buf) + } else { + target_dir.join(entry) + }; if !path.try_exists()? { continue; } diff --git a/crates/cli/commands/src/download/planning.rs b/crates/cli/commands/src/download/planning.rs index 22bb8b96bc9..6a3f8457986 100644 --- a/crates/cli/commands/src/download/planning.rs +++ b/crates/cli/commands/src/download/planning.rs @@ -153,9 +153,10 @@ pub(crate) struct DownloadStartupSummary { pub(crate) fn summarize_download_startup( all_downloads: &[PlannedArchive], target_dir: &Path, + static_files_dir: Option<&Path>, ) -> Result { let mut summary = DownloadStartupSummary::default(); - let verifier = OutputVerifier::new(target_dir); + let verifier = OutputVerifier::new(target_dir, static_files_dir); for planned in all_downloads { if verifier.verify(&planned.archive.output_files)? { @@ -303,7 +304,7 @@ mod tests { }, ]; - let summary = summarize_download_startup(&planned, target_dir).unwrap(); + let summary = summarize_download_startup(&planned, target_dir, None).unwrap(); assert_eq!(summary.reusable, 1); assert_eq!(summary.needs_download, 2); } diff --git a/crates/cli/commands/src/download/session.rs b/crates/cli/commands/src/download/session.rs index bc1e5a941af..9aa898e66b5 100644 --- a/crates/cli/commands/src/download/session.rs +++ b/crates/cli/commands/src/download/session.rs @@ -81,6 +81,8 @@ impl DownloadSession { pub(crate) struct ArchiveProcessContext { /// Directory where extracted output files are written. target_dir: PathBuf, + /// Custom location of static file outputs. + static_files_dir: Option, /// Directory used for cached archive downloads, when enabled. cache_dir: Option, /// Shared command-scoped download state. @@ -91,10 +93,11 @@ impl ArchiveProcessContext { /// Creates the context used while processing modular archives. pub(crate) fn new( target_dir: PathBuf, + static_files_dir: Option, cache_dir: Option, session: DownloadSession, ) -> Self { - Self { target_dir, cache_dir, session } + Self { target_dir, static_files_dir, cache_dir, session } } /// Returns the directory where extracted outputs should be written. @@ -102,6 +105,11 @@ impl ArchiveProcessContext { &self.target_dir } + /// Returns the custom static files directory, if configured. + pub(crate) fn static_files_dir(&self) -> Option<&Path> { + self.static_files_dir.as_deref() + } + /// Returns the cache directory for two-phase downloads, if enabled. pub(crate) fn cache_dir(&self) -> Option<&Path> { self.cache_dir.as_deref() diff --git a/crates/cli/commands/src/download/verify.rs b/crates/cli/commands/src/download/verify.rs index a4b84b992eb..d87c7bf6fef 100644 --- a/crates/cli/commands/src/download/verify.rs +++ b/crates/cli/commands/src/download/verify.rs @@ -8,12 +8,14 @@ use std::{io::Read, path::Path}; pub(crate) struct OutputVerifier<'a> { /// Directory containing the output files declared by the manifest. target_dir: &'a Path, + /// Custom location of static file outputs. + static_files_dir: Option<&'a Path>, } impl<'a> OutputVerifier<'a> { /// Creates a verifier for one extraction target directory. - pub(crate) const fn new(target_dir: &'a Path) -> Self { - Self { target_dir } + pub(crate) const fn new(target_dir: &'a Path, static_files_dir: Option<&'a Path>) -> Self { + Self { target_dir, static_files_dir } } /// Returns `true` only when every declared output file exists and matches size and BLAKE3. @@ -34,7 +36,7 @@ impl<'a> OutputVerifier<'a> { } for expected in output_files { - let output_path = self.target_dir.join(&expected.path); + let output_path = self.output_path(&expected.path); let meta = match fs::metadata(&output_path) { Ok(meta) => meta, Err(_) => return Ok(false), @@ -55,10 +57,20 @@ impl<'a> OutputVerifier<'a> { /// Removes any declared output files so a fresh archive attempt can restart cleanly. pub(crate) fn cleanup(&self, output_files: &[OutputFileChecksum]) { for output in output_files { - let _ = fs::remove_file(self.target_dir.join(&output.path)); + let _ = fs::remove_file(self.output_path(&output.path)); } } + /// Resolves archive paths consistently for verification and retry cleanup. + fn output_path(&self, path: &str) -> std::path::PathBuf { + if let Some(static_files_dir) = self.static_files_dir && + let Some(relative_path) = super::extract::static_file_relative_path(Path::new(path)) + { + return static_files_dir.join(relative_path) + } + self.target_dir.join(path) + } + /// Computes the hex-encoded BLAKE3 checksum for one plain output file. fn file_blake3_hex( path: &Path, @@ -82,3 +94,30 @@ impl<'a> OutputVerifier<'a> { Ok(hasher.finalize().to_hex().to_string()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn custom_static_files_verification_and_cleanup() { + let datadir = tempfile::tempdir().unwrap(); + let static_dir = tempfile::tempdir().unwrap(); + let default_file = datadir.path().join("static_files/headers"); + fs::create_dir_all(default_file.parent().unwrap()).unwrap(); + fs::write(&default_file, b"headers").unwrap(); + let outputs = [OutputFileChecksum { + path: "./static_files/headers".into(), + size: 7, + blake3: blake3::hash(b"headers").to_hex().to_string(), + }]; + let verifier = OutputVerifier::new(datadir.path(), Some(static_dir.path())); + assert!(!verifier.verify(&outputs).unwrap()); + let custom_file = static_dir.path().join("headers"); + fs::write(&custom_file, b"headers").unwrap(); + assert!(verifier.verify(&outputs).unwrap()); + verifier.cleanup(&outputs); + assert!(!custom_file.exists()); + assert_eq!(fs::read(default_file).unwrap(), b"headers"); + } +} From 8adbef51a28ec9421089da21422b457a1d1d79bd Mon Sep 17 00:00:00 2001 From: Matthias Seitz Date: Mon, 28 Sep 2026 14:05:25 +0000 Subject: [PATCH 6/8] fix(cli): exit download right after completion (#27498) Co-authored-by: Matthias Seitz <19890894+mattsse@users.noreply.github.com> (cherry picked from commit 84a5fd964434062b326f3359473f85d9952cd3bf) --- crates/cli/commands/src/download/archive.rs | 8 ++---- crates/cli/commands/src/download/progress.rs | 27 ++++++++++++++------ 2 files changed, 21 insertions(+), 14 deletions(-) diff --git a/crates/cli/commands/src/download/archive.rs b/crates/cli/commands/src/download/archive.rs index c4e46e1ddbf..83abd22062f 100644 --- a/crates/cli/commands/src/download/archive.rs +++ b/crates/cli/commands/src/download/archive.rs @@ -15,11 +15,7 @@ use eyre::Result; use futures::stream::{self, StreamExt}; use reth_cli_util::cancellation::CancellationToken; use reth_fs_util as fs; -use std::{ - path::Path, - sync::{atomic::Ordering, Arc}, - time::Duration, -}; +use std::{path::Path, sync::Arc, time::Duration}; use tokio::task; use tracing::{debug, info, warn}; @@ -89,7 +85,7 @@ impl ModularDownloadJob { .collect() .await; - shared.done.store(true, Ordering::Relaxed); + shared.done.notify_one(); let _ = progress_handle.await; for result in results { diff --git a/crates/cli/commands/src/download/progress.rs b/crates/cli/commands/src/download/progress.rs index b0b6b50f696..e8025cab3fa 100644 --- a/crates/cli/commands/src/download/progress.rs +++ b/crates/cli/commands/src/download/progress.rs @@ -3,11 +3,12 @@ use reth_cli_util::cancellation::CancellationToken; use std::{ io::{self, Read, Write}, sync::{ - atomic::{AtomicBool, AtomicU64, Ordering}, + atomic::{AtomicU64, Ordering}, Arc, Condvar, Mutex, }, time::{Duration, Instant}, }; +use tokio::sync::Notify; use tracing::info; const BYTE_UNITS: [&str; 4] = ["B", "KB", "MB", "GB"]; @@ -132,8 +133,8 @@ pub(crate) struct SharedProgress { pub(crate) active_extractions: AtomicU64, /// Number of archives currently verifying extracted outputs. pub(crate) active_verifications: AtomicU64, - /// Signals the background progress task to exit. - pub(crate) done: AtomicBool, + /// Wakes the background progress task to print the final summary and exit. + pub(crate) done: Notify, /// Cancellation token shared by the whole command. cancel_token: CancellationToken, } @@ -164,7 +165,7 @@ impl SharedProgress { active_download_requests: AtomicU64::new(0), active_extractions: AtomicU64::new(0), active_verifications: AtomicU64::new(0), - done: AtomicBool::new(false), + done: Notify::new(), cancel_token, }) } @@ -697,10 +698,9 @@ pub(crate) fn spawn_progress_display(progress: Arc) -> tokio::ta let mut interval = tokio::time::interval(Duration::from_secs(3)); interval.tick().await; loop { - interval.tick().await; - - if progress.done.load(Ordering::Relaxed) { - break; + tokio::select! { + _ = interval.tick() => {} + _ = progress.done.notified() => break, } let download_total = progress.total_download_bytes; @@ -841,4 +841,15 @@ mod tests { 40 ); } + + #[tokio::test] + async fn progress_display_exits_when_done() { + let progress = SharedProgress::new(10, 20, 1, CancellationToken::new()); + let handle = spawn_progress_display(Arc::clone(&progress)); + + progress.done.notify_one(); + + // The task must exit right away instead of on its next 3s progress tick. + tokio::time::timeout(Duration::from_secs(1), handle).await.unwrap().unwrap(); + } } From f5f8cc87d5187b56d65501d89be1b0eb903adb8a Mon Sep 17 00:00:00 2001 From: wbj-cb Date: Sat, 3 Oct 2026 05:19:11 +0000 Subject: [PATCH 7/8] feat(cli): add --prune-unlisted to reth download (#27669) Co-authored-by: OpenCode Co-authored-by: Matthias Seitz (cherry picked from commit b1e9f3f39eb85fd6e65293995878b1fa6fa606e5) --- crates/cli/commands/src/download/mod.rs | 64 +++++- crates/cli/commands/src/download/prune.rs | 248 +++++++++++++++++++++ crates/cli/commands/src/download/verify.rs | 2 +- docs/vocs/docs/pages/cli/reth/download.mdx | 7 + 4 files changed, 313 insertions(+), 8 deletions(-) create mode 100644 crates/cli/commands/src/download/prune.rs diff --git a/crates/cli/commands/src/download/mod.rs b/crates/cli/commands/src/download/mod.rs index 3726743f6bc..0ac48f1281a 100644 --- a/crates/cli/commands/src/download/mod.rs +++ b/crates/cli/commands/src/download/mod.rs @@ -85,6 +85,7 @@ pub mod manifest; pub mod manifest_cmd; mod planning; mod progress; +mod prune; mod session; mod source; mod tui; @@ -101,6 +102,7 @@ use eyre::Result; use manifest::{ComponentSelection, SnapshotComponentType, SnapshotManifest}; use planning::{collect_planned_archives, summarize_download_startup, PlannedDownloads}; use progress::{DownloadProgress, DownloadRequestLimiter}; +use prune::prune_unlisted_outputs; use reth_chainspec::{EthChainSpec, EthereumHardfork, EthereumHardforks, MAINNET}; use reth_cli::chainspec::ChainSpecParser; use reth_cli_util::cancellation::CancellationToken; @@ -127,7 +129,7 @@ const RETH_SNAPSHOTS_BASE_URL: &str = "https://snapshots-r2.reth.rs"; const RETH_SNAPSHOTS_API_URL: &str = "https://snapshots.reth.rs/api/snapshots"; const RETH_SNAPSHOTS_SOURCE: &str = "https://snapshots.reth.rs (default)"; const SNAPSHOT_API_PATH: &str = "/api/snapshots"; -const FORCE_REMOVED_DATADIR_PATHS: &[&str] = &["db", "rocksdb", "static_files", "reth.toml"]; +const MANAGED_DATADIR_PATHS: &[&str] = &["db", "rocksdb", "static_files", "reth.toml"]; /// Maximum number of simultaneous HTTP downloads across the entire snapshot job. const MAX_CONCURRENT_DOWNLOADS: usize = 8; @@ -427,6 +429,22 @@ pub struct DownloadCommand { #[arg(long, conflicts_with = "list")] force: bool, + /// Remove files from db, rocksdb, static_files, and reth.toml that the selected snapshot + /// plan does not list, then reuse the listed files that still verify. + /// + /// Unlike `--force`, this keeps reusable snapshot files, so only missing or changed archives + /// are downloaded. Unlike a plain re-run, the data dir ends up with exactly the snapshot's + /// files, which is required when the node has written data past the snapshot block. + /// + /// Components that are not selected are removed entirely, so `--prune-unlisted --minimal` + /// over an archive node deletes its history. With `--non-interactive`, components must be + /// selected explicitly with a preset or `--with-*` flags. reth.toml is always regenerated, + /// which discards local edits. Symlinks below these paths are removed, not followed. Pruning + /// refuses to run while a node holds the database lock, and happens before downloading, so a + /// failed download leaves the data dir pruned, as with `--force`. + #[arg(long, conflicts_with_all = ["force", "list", "url"])] + prune_unlisted: bool, + /// Enable resumable two-phase downloads (download to disk first, then extract). /// /// Archives are downloaded to a `.part` file with HTTP Range resume support @@ -528,6 +546,8 @@ impl> DownloadCo let _cancel_guard = cancel_token.drop_guard(); if self.force { clear_existing_datadir(target_dir, static_files_dir.as_deref())?; + } else if self.prune_unlisted { + prune_unlisted_outputs(&planned.archives, target_dir, static_files_dir.as_deref())?; } fs::create_dir_all(target_dir)?; let startup_summary = @@ -757,6 +777,13 @@ impl> DownloadCo } if self.non_interactive { + // The implicit minimal default would silently prune history from a full or archive + // node. + eyre::ensure!( + !self.prune_unlisted, + "--prune-unlisted with --non-interactive requires an explicit component selection: \ + --minimal, --full, --archive, or --with-* flags" + ); return Ok(ResolvedComponents { selections: self.minimal_preset_selections(manifest), preset: Some(SelectionPreset::Minimal), @@ -925,12 +952,8 @@ fn selection_from_prune_mode(mode: Option, snapshot_block: u64) -> Co /// Removes existing snapshot data that is managed by `reth download`. fn clear_existing_datadir(target_dir: &Path, static_files_dir: Option<&Path>) -> Result<()> { info!(target: "reth::cli", dir = ?target_dir, "Clearing existing snapshot data"); - for entry in FORCE_REMOVED_DATADIR_PATHS { - let path = if *entry == "static_files" { - static_files_dir.map_or_else(|| target_dir.join(entry), Path::to_path_buf) - } else { - target_dir.join(entry) - }; + for entry in MANAGED_DATADIR_PATHS { + let path = managed_datadir_path(entry, target_dir, static_files_dir); if !path.try_exists()? { continue; } @@ -946,6 +969,19 @@ fn clear_existing_datadir(target_dir: &Path, static_files_dir: Option<&Path>) -> Ok(()) } +/// Resolves one of [`MANAGED_DATADIR_PATHS`], honoring a custom static files directory. +fn managed_datadir_path( + entry: &str, + target_dir: &Path, + static_files_dir: Option<&Path>, +) -> PathBuf { + if entry == "static_files" { + static_files_dir.map_or_else(|| target_dir.join(entry), Path::to_path_buf) + } else { + target_dir.join(entry) + } +} + /// If all data components (txs, receipts, changesets) are `All`, automatically /// include hidden archive-only components when available in the manifest. fn inject_archive_only_components( @@ -1266,6 +1302,20 @@ mod tests { assert!(parse(vec!["reth", "--retry-backoff", "invalid"]).is_err()); } + #[test] + fn test_download_prune_unlisted_conflicts_with_force_list_and_url() { + let parse = |args: &[&str]| { + CommandParser::>::try_parse_from(args) + }; + + assert!(parse(&["reth", "--prune-unlisted"]).unwrap().args.prune_unlisted); + assert!(parse(&["reth", "--prune-unlisted", "--force"]).is_err()); + assert!(parse(&["reth", "--prune-unlisted", "--list"]).is_err()); + assert!( + parse(&["reth", "--prune-unlisted", "--url", "https://example.com/a.tar.zst"]).is_err() + ); + } + #[test] fn test_download_resumable_defaults_to_true() { let args = diff --git a/crates/cli/commands/src/download/prune.rs b/crates/cli/commands/src/download/prune.rs new file mode 100644 index 00000000000..354f42b55bc --- /dev/null +++ b/crates/cli/commands/src/download/prune.rs @@ -0,0 +1,248 @@ +//! Removes snapshot-managed data that the selected download plan does not list. + +use super::{ + managed_datadir_path, planning::PlannedArchive, verify::OutputVerifier, MANAGED_DATADIR_PATHS, +}; +use eyre::{Result, WrapErr}; +use reth_db::lockfile::StorageLock; +use reth_fs_util as fs; +use std::{ + collections::HashSet, + path::{Path, PathBuf}, +}; +use tracing::info; + +/// Removes every file under the paths that `--force` clears (`db`, `rocksdb`, `static_files`, +/// and `reth.toml`) that no planned archive lists as an output. +/// +/// Listed files are kept so the startup integrity check can reuse the ones that still verify. +/// Once the download completes, those paths hold exactly the snapshot's files, with no leftovers +/// such as blocks or history indices from a chain the node extended past the snapshot block. +/// +/// This relies on each archive's `output_files` listing every file the archive extracts. A file +/// an archive extracts without listing it is removed here and not restored, because the archive +/// is reused once its listed outputs verify. +/// +/// A managed path that is itself a symlink is followed, because operators link whole data +/// directories to other disks. Symlinks below it are removed rather than followed, so pruning +/// never deletes files outside the data dir. Nested directories left empty are removed; the +/// managed directories themselves are kept, since they may be mount points. +/// +/// Fails if a running node holds the database lock, since pruning would otherwise delete its +/// lock and files from under it. +pub(crate) fn prune_unlisted_outputs( + planned: &[PlannedArchive], + target_dir: &Path, + static_files_dir: Option<&Path>, +) -> Result<()> { + let db_dir = target_dir.join("db"); + if db_dir.is_dir() { + drop(StorageLock::try_acquire(&db_dir).wrap_err("refusing to prune a data dir in use")?); + } + + let verifier = OutputVerifier::new(target_dir, static_files_dir); + let listed = planned + .iter() + .flat_map(|planned| &planned.archive.output_files) + .map(|output| verifier.output_path(&output.path)) + .collect::>(); + + let mut removed = 0; + for entry in MANAGED_DATADIR_PATHS { + let root = managed_datadir_path(entry, target_dir, static_files_dir); + if !root.try_exists()? { + continue + } + + if fs::metadata(&root)?.is_dir() { + removed += prune_dir(&root, &listed)?; + } else if !listed.contains(&root) { + fs::remove_file(&root)?; + removed += 1; + } + } + + info!(target: "reth::cli", removed, "Removed existing files not listed in the snapshot plan"); + Ok(()) +} + +/// Removes unlisted entries below `dir` without following symlinks, and returns how many files +/// or links it removed. Removing a symlink leaves its target intact. +fn prune_dir(dir: &Path, listed: &HashSet) -> Result { + let mut removed = 0; + for entry in fs::read_dir(dir)? { + let entry = entry?; + let path = entry.path(); + // Unlike `fs::metadata`, this does not follow symlinks. + let file_type = entry + .file_type() + .wrap_err_with(|| format!("failed to read file type of {}", path.display()))?; + if file_type.is_dir() { + removed += prune_dir(&path, listed)?; + remove_dir_if_empty(&path)?; + continue + } + if listed.contains(&path) { + continue + } + + // Windows removes a symlink to a directory with `remove_dir`, not `remove_file`. + #[cfg(windows)] + if std::os::windows::fs::FileTypeExt::is_symlink_dir(&file_type) { + std::fs::remove_dir(&path) + .wrap_err_with(|| format!("failed to remove {}", path.display()))?; + removed += 1; + continue + } + fs::remove_file(&path)?; + removed += 1; + } + Ok(removed) +} + +/// Removes `dir` if pruning left it empty. +fn remove_dir_if_empty(dir: &Path) -> Result<()> { + if fs::read_dir(dir)?.next().is_none() { + std::fs::remove_dir(dir) + .wrap_err_with(|| format!("failed to remove empty directory {}", dir.display()))?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::download::manifest::{OutputFileChecksum, SnapshotArchive, SnapshotComponentType}; + use tempfile::tempdir; + + fn planned_with_outputs(paths: &[&str]) -> Vec { + vec![PlannedArchive { + ty: SnapshotComponentType::State, + component: "State".to_string(), + archive: SnapshotArchive { + url: "https://example.com/state.tar.zst".to_string(), + file_name: "state.tar.zst".to_string(), + size: 1, + blake3: None, + output_files: paths + .iter() + .map(|path| OutputFileChecksum { + path: (*path).to_string(), + size: 1, + blake3: String::new(), + }) + .collect(), + }, + }] + } + + fn write_file(path: &Path) { + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, b"data").unwrap(); + } + + #[test] + fn prune_removes_unlisted_files_and_keeps_listed_ones() { + let dir = tempdir().unwrap(); + let target = dir.path(); + let listed = ["db/mdbx.dat", "static_files/static_file_headers_0_499999"]; + let leftovers = [ + "db/stale.dat", + "rocksdb/000042.sst", + "static_files/static_file_headers_500000_999999", + "reth.toml", + ]; + for path in listed.iter().chain(&leftovers).chain(&["known-peers.json"]) { + write_file(&target.join(path)); + } + + prune_unlisted_outputs(&planned_with_outputs(&listed), target, None).unwrap(); + + assert!(listed.iter().all(|path| target.join(path).exists())); + assert!(leftovers.iter().all(|path| !target.join(path).exists())); + assert!(target.join("known-peers.json").exists()); + } + + #[test] + fn prune_removes_nested_directories_it_empties_and_keeps_managed_ones() { + let dir = tempdir().unwrap(); + write_file(&dir.path().join("rocksdb/nested/000042.sst")); + + prune_unlisted_outputs(&planned_with_outputs(&["db/mdbx.dat"]), dir.path(), None).unwrap(); + + assert!(!dir.path().join("rocksdb/nested").exists()); + assert!(dir.path().join("rocksdb").is_dir()); + } + + #[test] + fn prune_resolves_static_files_in_a_custom_directory() { + let dir = tempdir().unwrap(); + let target = dir.path().join("datadir"); + let static_files = dir.path().join("custom-static-files"); + write_file(&static_files.join("static_file_headers_0_499999")); + write_file(&static_files.join("static_file_headers_500000_999999")); + + prune_unlisted_outputs( + &planned_with_outputs(&["static_files/static_file_headers_0_499999"]), + &target, + Some(&static_files), + ) + .unwrap(); + + assert!(static_files.join("static_file_headers_0_499999").exists()); + assert!(!static_files.join("static_file_headers_500000_999999").exists()); + } + + #[test] + fn prune_skips_missing_managed_paths() { + let dir = tempdir().unwrap(); + + prune_unlisted_outputs(&planned_with_outputs(&["db/mdbx.dat"]), dir.path(), None).unwrap(); + + assert!(!dir.path().join("db").exists()); + } + + #[cfg(unix)] + #[test] + fn prune_removes_nested_symlinks_without_following_them() { + let dir = tempdir().unwrap(); + let target = dir.path().join("datadir"); + let outside = dir.path().join("outside"); + write_file(&outside.join("unrelated.dat")); + write_file(&target.join("static_files/static_file_headers_0_499999")); + std::os::unix::fs::symlink(&outside, target.join("static_files/linked")).unwrap(); + + prune_unlisted_outputs( + &planned_with_outputs(&["static_files/static_file_headers_0_499999"]), + &target, + None, + ) + .unwrap(); + + assert!(std::fs::symlink_metadata(target.join("static_files/linked")).is_err()); + assert!(outside.join("unrelated.dat").exists()); + } + + #[cfg(unix)] + #[test] + fn prune_follows_a_symlinked_managed_directory() { + let dir = tempdir().unwrap(); + let target = dir.path().join("datadir"); + let other_disk = dir.path().join("other-disk"); + write_file(&other_disk.join("static_file_headers_0_499999")); + write_file(&other_disk.join("static_file_headers_500000_999999")); + std::fs::create_dir_all(&target).unwrap(); + std::os::unix::fs::symlink(&other_disk, target.join("static_files")).unwrap(); + + prune_unlisted_outputs( + &planned_with_outputs(&["static_files/static_file_headers_0_499999"]), + &target, + None, + ) + .unwrap(); + + assert!(std::fs::symlink_metadata(target.join("static_files")).unwrap().is_symlink()); + assert!(other_disk.join("static_file_headers_0_499999").exists()); + assert!(!other_disk.join("static_file_headers_500000_999999").exists()); + } +} diff --git a/crates/cli/commands/src/download/verify.rs b/crates/cli/commands/src/download/verify.rs index d87c7bf6fef..e4c2f58631c 100644 --- a/crates/cli/commands/src/download/verify.rs +++ b/crates/cli/commands/src/download/verify.rs @@ -62,7 +62,7 @@ impl<'a> OutputVerifier<'a> { } /// Resolves archive paths consistently for verification and retry cleanup. - fn output_path(&self, path: &str) -> std::path::PathBuf { + pub(crate) fn output_path(&self, path: &str) -> std::path::PathBuf { if let Some(static_files_dir) = self.static_files_dir && let Some(relative_path) = super::extract::static_file_relative_path(Path::new(path)) { diff --git a/docs/vocs/docs/pages/cli/reth/download.mdx b/docs/vocs/docs/pages/cli/reth/download.mdx index 6817a3c413b..04a2a3220d8 100644 --- a/docs/vocs/docs/pages/cli/reth/download.mdx +++ b/docs/vocs/docs/pages/cli/reth/download.mdx @@ -208,6 +208,13 @@ Storage: --force Overwrite existing snapshot data by removing db, rocksdb, static_files, and reth.toml + --prune-unlisted + Remove files from db, rocksdb, static_files, and reth.toml that the selected snapshot plan does not list, then reuse the listed files that still verify. + + Unlike `--force`, this keeps reusable snapshot files, so only missing or changed archives are downloaded. Unlike a plain re-run, the data dir ends up with exactly the snapshot's files, which is required when the node has written data past the snapshot block. + + Components that are not selected are removed entirely, so `--prune-unlisted --minimal` over an archive node deletes its history. With `--non-interactive`, components must be selected explicitly with a preset or `--with-*` flags. reth.toml is always regenerated, which discards local edits. Symlinks below these paths are removed, not followed. Pruning refuses to run while a node holds the database lock, and happens before downloading, so a failed download leaves the data dir pruned, as with `--force`. + --resumable [] Enable resumable two-phase downloads (download to disk first, then extract). From 52802afa4c38d18e3be4c6be12b142db1187ef57 Mon Sep 17 00:00:00 2001 From: wbj-cb Date: Sat, 3 Oct 2026 06:11:55 +0000 Subject: [PATCH 8/8] perf(download): verify reusable outputs once and in parallel (#27665) Co-authored-by: Matthias Seitz (cherry picked from commit 5b01d18dba83043bf65969496e3f14e1a97951ff) --- Cargo.lock | 1 + crates/cli/commands/Cargo.toml | 2 +- crates/cli/commands/src/download/archive.rs | 45 ++---- crates/cli/commands/src/download/mod.rs | 33 ++-- crates/cli/commands/src/download/planning.rs | 95 +++++++----- crates/cli/commands/src/download/progress.rs | 16 +- crates/cli/commands/src/download/session.rs | 7 - crates/cli/commands/src/download/verify.rs | 149 +++++++++++++++---- 8 files changed, 227 insertions(+), 121 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 785ffa6e71f..94eb5d1c845 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1776,6 +1776,7 @@ dependencies = [ "cfg-if", "constant_time_eq", "cpufeatures 0.3.0", + "rayon-core", ] [[package]] diff --git a/crates/cli/commands/Cargo.toml b/crates/cli/commands/Cargo.toml index 3d64068cc39..88cdd8310de 100644 --- a/crates/cli/commands/Cargo.toml +++ b/crates/cli/commands/Cargo.toml @@ -88,7 +88,7 @@ tokio-stream.workspace = true reqwest = { workspace = true, features = ["blocking"] } url.workspace = true metrics.workspace = true -blake3.workspace = true +blake3 = { workspace = true, features = ["rayon"] } rayon.workspace = true # io diff --git a/crates/cli/commands/src/download/archive.rs b/crates/cli/commands/src/download/archive.rs index 83abd22062f..beca5b0688d 100644 --- a/crates/cli/commands/src/download/archive.rs +++ b/crates/cli/commands/src/download/archive.rs @@ -2,7 +2,7 @@ use super::{ extract::{extract_archive_raw, streaming_download_and_extract, CompressionFormat}, fetch::ArchiveFetcher, manifest::SnapshotArchive, - planning::{PlannedArchive, PlannedDownloads}, + planning::{CheckedDownloads, PlannedArchive}, progress::{ spawn_progress_display, ArchiveDownloadProgress, ArchiveExtractionProgress, ArchiveVerificationProgress, DownloadRequestLimiter, SharedProgress, @@ -23,7 +23,7 @@ const DOWNLOAD_CACHE_DIR: &str = ".download-cache"; /// Runs all planned modular archive downloads for one command invocation. pub(crate) async fn run_modular_downloads( - planned_downloads: PlannedDownloads, + planned_downloads: CheckedDownloads, target_dir: &Path, static_files_dir: Option<&Path>, download_concurrency: usize, @@ -70,13 +70,17 @@ impl ModularDownloadJob { } /// Runs all planned archives and waits for the shared progress task to finish. - async fn run(self, planned_downloads: PlannedDownloads) -> Result<()> { + async fn run(self, planned_downloads: CheckedDownloads) -> Result<()> { let shared = Arc::clone( self.ctx.session().progress().expect("modular downloads always use shared progress"), ); let progress_handle = spawn_progress_display(Arc::clone(&shared)); + for reused in &planned_downloads.reused { + info!(target: "reth::cli", file = %reused.archive.file_name, component = %reused.component, "Skipping already verified plain files"); + shared.record_reused_archive(reused.archive.size, reused.archive.output_size()); + } let ctx = self.ctx.clone(); - let results: Vec> = stream::iter(planned_downloads.archives) + let results: Vec> = stream::iter(planned_downloads.pending) .map(move |archive| { let ctx = ctx.clone(); async move { Self::process_archive(ctx, archive).await } @@ -117,7 +121,7 @@ enum ArchiveAttemptState { Fail, } -/// Processes one modular archive from reuse check through extraction and verification. +/// Processes one modular archive through fetch, extraction, and verification. struct ArchiveProcessor { /// The concrete archive and component being processed. archive: PlannedArchive, @@ -132,13 +136,10 @@ impl ArchiveProcessor { } /// Runs the archive retry state machine until outputs are verified or retries are exhausted. + /// + /// Callers only pass archives that failed the startup reuse check. fn run(self) -> Result<()> { let archive = self.archive(); - if self.try_reuse_outputs()? { - info!(target: "reth::cli", file = %archive.file_name, component = %self.archive.component, "Skipping already verified plain files"); - return Ok(()); - } - let mode = ArchiveMode::new(&self.ctx)?; let format = CompressionFormat::from_url(&archive.file_name)?; let mut attempt = 1; @@ -224,33 +225,11 @@ impl ArchiveProcessor { OutputVerifier::new(self.ctx.target_dir(), self.ctx.static_files_dir()) } - /// Returns `true` if this archive can be reused from existing verified outputs. - /// Returns `false` if a fresh archive attempt is still needed. - fn try_reuse_outputs(&self) -> Result { - if self.verify_outputs()? { - self.mark_complete(); - return Ok(true); - } - - Ok(false) - } - /// Removes any partial outputs before a fresh archive attempt. fn cleanup_outputs(&self) { self.output_verifier().cleanup(&self.archive().output_files); } - /// Returns `true` if all declared plain outputs verify. - /// Returns `false` if any output is missing or does not match. - fn verify_outputs(&self) -> Result { - self.output_verifier().verify(&self.archive().output_files) - } - - /// Records archive completion in shared progress once outputs verify. - fn mark_complete(&self) { - self.ctx.session().record_reused_archive(self.archive().size, self.archive().output_size()); - } - /// Executes one archive attempt according to the selected cache-vs-stream mode. fn run_attempt(&self, mode: ArchiveMode, format: CompressionFormat) -> Result<()> { mode.execute(self, format) @@ -332,7 +311,7 @@ impl ArchiveProcessor { ArchiveVerificationProgress::new(self.ctx.session().progress()); let verified = self .output_verifier() - .verify_with_progress(&self.archive().output_files, Some(&mut verification_progress))?; + .verify_with_progress(&self.archive().output_files, Some(&verification_progress))?; if verified { verification_progress.complete(self.archive().output_size()); } diff --git a/crates/cli/commands/src/download/mod.rs b/crates/cli/commands/src/download/mod.rs index 0ac48f1281a..22bcb5ce5f6 100644 --- a/crates/cli/commands/src/download/mod.rs +++ b/crates/cli/commands/src/download/mod.rs @@ -39,10 +39,13 @@ //! Archive processing is modeled around `ModularDownloadJob`, which schedules work, and //! `ArchiveProcessor`, which owns the explicit retry state machine for one archive. //! `ArchiveMode` decides whether that archive should be fetched through the cache or streamed -//! directly: +//! directly. //! -//! - reuse verified plain output files when possible, -//! - otherwise fetch and extract the archive, +//! Before any archive is scheduled, `PlannedDownloads::partition_reusable` hashes the existing +//! output files once and skips archives whose outputs already verify. Each remaining archive is +//! processed as follows: +//! +//! - fetch and extract the archive, //! - verify the declared output files, //! - retry the entire archive attempt if extraction succeeded but verification failed. //! @@ -100,7 +103,7 @@ use config_gen::{config_for_selections, write_config}; use extract::stream_and_extract; use eyre::Result; use manifest::{ComponentSelection, SnapshotComponentType, SnapshotManifest}; -use planning::{collect_planned_archives, summarize_download_startup, PlannedDownloads}; +use planning::{collect_planned_archives, PlannedDownloads}; use progress::{DownloadProgress, DownloadRequestLimiter}; use prune::prune_unlisted_outputs; use reth_chainspec::{EthChainSpec, EthereumHardfork, EthereumHardforks, MAINNET}; @@ -550,23 +553,29 @@ impl> DownloadCo prune_unlisted_outputs(&planned.archives, target_dir, static_files_dir.as_deref())?; } fs::create_dir_all(target_dir)?; - let startup_summary = - summarize_download_startup(&planned.archives, target_dir, static_files_dir.as_deref())?; + let downloads = { + let (target_dir, static_files_dir) = + (target_dir.to_path_buf(), static_files_dir.clone()); + tokio::task::spawn_blocking(move || { + planned.partition_reusable(&target_dir, static_files_dir.as_deref()) + }) + .await?? + }; info!(target: "reth::cli", - reusable = startup_summary.reusable, - needs_download = startup_summary.needs_download, + reusable = downloads.reused.len(), + needs_download = downloads.pending.len(), "Startup integrity summary (plain output files)" ); info!(target: "reth::cli", - archives = planned.total_archives(), - download_total = %DownloadProgress::format_size(planned.total_download_size), - output_total = %DownloadProgress::format_size(planned.total_output_size), + archives = downloads.total_archives(), + download_total = %DownloadProgress::format_size(downloads.total_download_size), + output_total = %DownloadProgress::format_size(downloads.total_output_size), "Downloading all archives" ); run_modular_downloads( - planned, + downloads, target_dir, static_files_dir.as_deref(), self.download_concurrency.max(1), diff --git a/crates/cli/commands/src/download/planning.rs b/crates/cli/commands/src/download/planning.rs index 6a3f8457986..fe2748a9d80 100644 --- a/crates/cli/commands/src/download/planning.rs +++ b/crates/cli/commands/src/download/planning.rs @@ -1,5 +1,6 @@ use super::{manifest::*, verify::OutputVerifier}; use eyre::Result; +use rayon::prelude::*; use serde::Serialize; use std::{collections::BTreeMap, io::Write, path::Path}; use tracing::info; @@ -125,9 +126,54 @@ pub(crate) struct PlannedDownloads { } impl PlannedDownloads { - /// Returns the number of concrete archives queued for this snapshot selection. + /// Verifies every archive's existing outputs in parallel and splits off the archives that + /// already match, so each output file is hashed at most once before work begins. + pub(crate) fn partition_reusable( + self, + target_dir: &Path, + static_files_dir: Option<&Path>, + ) -> Result { + let verifier = OutputVerifier::new(target_dir, static_files_dir); + let checked = self + .archives + .into_par_iter() + .map(|planned| Ok((verifier.verify(&planned.archive.output_files)?, planned))) + .collect::>>()?; + + let mut downloads = CheckedDownloads { + pending: Vec::new(), + reused: Vec::new(), + total_download_size: self.total_download_size, + total_output_size: self.total_output_size, + }; + for (reusable, planned) in checked { + if reusable { + downloads.reused.push(planned); + } else { + downloads.pending.push(planned); + } + } + Ok(downloads) + } +} + +/// Planned archives split by the startup reuse check. +#[derive(Debug)] +pub(crate) struct CheckedDownloads { + /// Archives that still need to be fetched and extracted. + pub(crate) pending: Vec, + /// Archives whose declared outputs already verified on disk. + pub(crate) reused: Vec, + /// Total compressed download size of all planned archives. + pub(crate) total_download_size: u64, + /// Total extracted plain-output size of all planned archives. + pub(crate) total_output_size: u64, +} + +impl CheckedDownloads { + /// Returns the number of concrete archives in this snapshot selection, including reused ones. pub(crate) const fn total_archives(&self) -> usize { - self.archives.len() + self.pending.len() + self.reused.len() } } @@ -140,35 +186,6 @@ pub(crate) const fn archive_priority_rank(ty: SnapshotComponentType) -> u8 { } } -/// Startup summary showing how much of the selected work can be reused. -#[derive(Debug, Default, Clone, Copy)] -pub(crate) struct DownloadStartupSummary { - /// Archives whose declared outputs already verify on disk. - pub(crate) reusable: usize, - /// Archives that still need to be downloaded or retried. - pub(crate) needs_download: usize, -} - -/// Checks selected archives against existing output files before work begins. -pub(crate) fn summarize_download_startup( - all_downloads: &[PlannedArchive], - target_dir: &Path, - static_files_dir: Option<&Path>, -) -> Result { - let mut summary = DownloadStartupSummary::default(); - let verifier = OutputVerifier::new(target_dir, static_files_dir); - - for planned in all_downloads { - if verifier.verify(&planned.archive.output_files)? { - summary.reusable += 1; - } else { - summary.needs_download += 1; - } - } - - Ok(summary) -} - /// Converts a selection into the manifest distance form used for archive lookup. fn selection_archive_distance( selection: &ComponentSelection, @@ -253,7 +270,7 @@ mod tests { use tempfile::tempdir; #[test] - fn summarize_download_startup_counts_reusable_and_needs_download() { + fn partition_reusable_moves_only_verified_archives() { let dir = tempdir().unwrap(); let target_dir = dir.path(); let ok_file = target_dir.join("ok.bin"); @@ -304,9 +321,17 @@ mod tests { }, ]; - let summary = summarize_download_startup(&planned, target_dir, None).unwrap(); - assert_eq!(summary.reusable, 1); - assert_eq!(summary.needs_download, 2); + let downloads = + PlannedDownloads { archives: planned, total_download_size: 30, total_output_size: 5 } + .partition_reusable(target_dir, None) + .unwrap(); + assert_eq!(downloads.total_archives(), 3); + + let names = |archives: &[PlannedArchive]| { + archives.iter().map(|planned| planned.archive.file_name.clone()).collect::>() + }; + assert_eq!(names(&downloads.reused), ["ok.tar.zst"]); + assert_eq!(names(&downloads.pending), ["missing.tar.zst", "bad-size.tar.zst"]); } #[test] diff --git a/crates/cli/commands/src/download/progress.rs b/crates/cli/commands/src/download/progress.rs index e8025cab3fa..a0f0cf508e8 100644 --- a/crates/cli/commands/src/download/progress.rs +++ b/crates/cli/commands/src/download/progress.rs @@ -560,9 +560,11 @@ impl ArchiveExtractionProgressHandle { } /// Tracks one active archive verification attempt. +/// +/// Bytes are recorded through a shared reference because output files are hashed in parallel. pub(crate) struct ArchiveVerificationProgress<'a> { progress: Option<&'a Arc>, - verified: u64, + verified: AtomicU64, completed: bool, } @@ -572,12 +574,12 @@ impl<'a> ArchiveVerificationProgress<'a> { if let Some(progress) = progress { progress.verification_started(); } - Self { progress, verified: 0, completed: false } + Self { progress, verified: AtomicU64::new(0), completed: false } } /// Adds plain-output bytes hashed by this verification attempt. - pub(crate) fn record_verified(&mut self, bytes: u64) { - self.verified += bytes; + pub(crate) fn record_verified(&self, bytes: u64) { + self.verified.fetch_add(bytes, Ordering::Relaxed); if let Some(progress) = self.progress { progress.add_active_verified_output_bytes(bytes); } @@ -588,11 +590,11 @@ impl<'a> ArchiveVerificationProgress<'a> { if self.completed { return; } + let verified = std::mem::take(self.verified.get_mut()); if let Some(progress) = self.progress { - progress.sub_active_verified_output_bytes(self.verified); + progress.sub_active_verified_output_bytes(verified); progress.record_archive_output_complete(total_bytes); } - self.verified = 0; self.completed = true; } } @@ -600,7 +602,7 @@ impl<'a> ArchiveVerificationProgress<'a> { impl Drop for ArchiveVerificationProgress<'_> { fn drop(&mut self) { if let Some(progress) = self.progress { - progress.sub_active_verified_output_bytes(self.verified); + progress.sub_active_verified_output_bytes(*self.verified.get_mut()); progress.verification_finished(); } } diff --git a/crates/cli/commands/src/download/session.rs b/crates/cli/commands/src/download/session.rs index 9aa898e66b5..8663aad3f1c 100644 --- a/crates/cli/commands/src/download/session.rs +++ b/crates/cli/commands/src/download/session.rs @@ -61,13 +61,6 @@ impl DownloadSession { &self.cancel_token } - /// Records one archive whose outputs were already reusable on disk. - pub(crate) fn record_reused_archive(&self, download_bytes: u64, output_bytes: u64) { - if let Some(progress) = self.progress() { - progress.record_reused_archive(download_bytes, output_bytes); - } - } - /// Records one archive whose extracted outputs fully verified. pub(crate) fn record_archive_output_complete(&self, bytes: u64) { if let Some(progress) = self.progress() { diff --git a/crates/cli/commands/src/download/verify.rs b/crates/cli/commands/src/download/verify.rs index e4c2f58631c..897b6c415f0 100644 --- a/crates/cli/commands/src/download/verify.rs +++ b/crates/cli/commands/src/download/verify.rs @@ -1,8 +1,16 @@ use super::{manifest::OutputFileChecksum, progress::ArchiveVerificationProgress}; use blake3::Hasher; use eyre::Result; -use reth_fs_util as fs; -use std::{io::Read, path::Path}; +use rayon::prelude::*; +use reth_fs_util::{self as fs, FsPathError}; +use std::{ + io::{ErrorKind, Read}, + mem, + path::Path, +}; + +/// Read size per hash update. Large enough for `update_rayon` to split across threads. +const HASH_BUFFER_SIZE: usize = 4 * 1024 * 1024; /// Verifies and cleans up extracted output files in one target directory. pub(crate) struct OutputVerifier<'a> { @@ -19,39 +27,49 @@ impl<'a> OutputVerifier<'a> { } /// Returns `true` only when every declared output file exists and matches size and BLAKE3. - /// Returns `false` if any file is missing, mismatched, or no outputs were declared. + /// Returns `false` if any file is missing, mismatched, or no outputs were declared, and an + /// error if a file exists but cannot be inspected or read. pub(crate) fn verify(&self, output_files: &[OutputFileChecksum]) -> Result { self.verify_with_progress(output_files, None) } /// Returns `true` only when every declared output file exists and matches size and BLAKE3, /// updating the optional verification progress as file bytes are hashed. + /// + /// Sizes are checked before any file is hashed, then files are hashed in parallel. pub(crate) fn verify_with_progress( &self, output_files: &[OutputFileChecksum], - mut progress: Option<&mut ArchiveVerificationProgress<'_>>, + progress: Option<&ArchiveVerificationProgress<'_>>, ) -> Result { if output_files.is_empty() { return Ok(false); } + let mut paths = Vec::with_capacity(output_files.len()); for expected in output_files { let output_path = self.output_path(&expected.path); - let meta = match fs::metadata(&output_path) { - Ok(meta) => meta, - Err(_) => return Ok(false), - }; - if meta.len() != expected.size { - return Ok(false); - } - - let actual = Self::file_blake3_hex(&output_path, progress.as_deref_mut())?; - if !actual.eq_ignore_ascii_case(&expected.blake3) { - return Ok(false); + match fs::metadata(&output_path) { + Ok(meta) if meta.len() == expected.size => paths.push(output_path), + Ok(_) => return Ok(false), + Err(FsPathError::Metadata { source, .. }) + if source.kind() == ErrorKind::NotFound => + { + return Ok(false) + } + Err(error) => return Err(error.into()), } } - Ok(true) + output_files + .par_iter() + .zip(paths) + .map(|(expected, path)| { + Self::file_blake3_hex(&path, expected.size, progress) + .map(|actual| actual.eq_ignore_ascii_case(&expected.blake3)) + }) + .find_any(|verified| !matches!(verified, Ok(true))) + .unwrap_or(Ok(true)) } /// Removes any declared output files so a fresh archive attempt can restart cleanly. @@ -72,33 +90,112 @@ impl<'a> OutputVerifier<'a> { } /// Computes the hex-encoded BLAKE3 checksum for one plain output file. + /// + /// The next buffer is read while the current one is hashed, so disk reads overlap hashing + /// instead of alternating with it. fn file_blake3_hex( path: &Path, - mut progress: Option<&mut ArchiveVerificationProgress<'_>>, + size: u64, + progress: Option<&ArchiveVerificationProgress<'_>>, ) -> Result { let mut file = fs::open(path)?; let mut hasher = Hasher::new(); - let mut buf = [0_u8; 64 * 1024]; + let buf_len = size.min(HASH_BUFFER_SIZE as u64) as usize; + let mut current = vec![0_u8; buf_len]; + let mut next = vec![0_u8; buf_len]; - loop { - let n = file.read(&mut buf)?; - if n == 0 { - break; - } - hasher.update(&buf[..n]); - if let Some(progress) = progress.as_deref_mut() { - progress.record_verified(n as u64); + let mut filled = read_full(&mut file, &mut current)?; + while filled > 0 { + let (_, read) = rayon::join( + || { + hasher.update_rayon(¤t[..filled]); + }, + || read_full(&mut file, &mut next), + ); + if let Some(progress) = progress { + progress.record_verified(filled as u64); } + filled = read?; + mem::swap(&mut current, &mut next); } Ok(hasher.finalize().to_hex().to_string()) } } +/// Reads until `buf` is full or the reader is exhausted, returning the number of bytes read. +/// +/// Full buffers keep every `update_rayon` call aligned, so BLAKE3 can hash it as parallel +/// subtrees. +fn read_full(reader: &mut impl Read, buf: &mut [u8]) -> std::io::Result { + let mut filled = 0; + while filled < buf.len() { + match reader.read(&mut buf[filled..]) { + Ok(0) => break, + Ok(n) => filled += n, + Err(error) if error.kind() == ErrorKind::Interrupted => {} + Err(error) => return Err(error), + } + } + Ok(filled) +} + #[cfg(test)] mod tests { use super::*; + fn checksum(path: &str, data: &[u8]) -> OutputFileChecksum { + OutputFileChecksum { + path: path.into(), + size: data.len() as u64, + blake3: blake3::hash(data).to_hex().to_string(), + } + } + + #[test] + fn verify_detects_each_kind_of_output_mismatch() { + let dir = tempfile::tempdir().unwrap(); + let large: Vec = (0..2 * HASH_BUFFER_SIZE + 123).map(|i| (i % 251) as u8).collect(); + let small = b"small".to_vec(); + fs::write(dir.path().join("large"), &large).unwrap(); + fs::write(dir.path().join("small"), &small).unwrap(); + let outputs = vec![checksum("large", &large), checksum("small", &small)]; + let verifier = OutputVerifier::new(dir.path(), None); + + assert!(verifier.verify(&outputs).unwrap(), "matching multi-buffer outputs verify"); + + let mut uppercase = outputs.clone(); + uppercase[0].blake3 = uppercase[0].blake3.to_ascii_uppercase(); + assert!(verifier.verify(&uppercase).unwrap(), "checksums compare case-insensitively"); + + let mut flipped = large.clone(); + flipped[HASH_BUFFER_SIZE + 7] ^= 1; + fs::write(dir.path().join("large"), &flipped).unwrap(); + assert!(!verifier.verify(&outputs).unwrap(), "one flipped byte fails verification"); + + fs::write(dir.path().join("large"), &large[1..]).unwrap(); + assert!(!verifier.verify(&outputs).unwrap(), "size mismatch fails verification"); + + fs::remove_file(dir.path().join("large")).unwrap(); + assert!(!verifier.verify(&outputs).unwrap(), "missing file fails verification"); + + assert!(!verifier.verify(&[]).unwrap(), "archives without outputs never verify"); + } + + #[cfg(unix)] + #[test] + fn verify_returns_metadata_errors_other_than_not_found() { + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("file"), b"data").unwrap(); + let verifier = OutputVerifier::new(dir.path(), None); + + assert!(!verifier.verify(&[checksum("missing", b"data")]).unwrap()); + assert!( + verifier.verify(&[checksum("file/child", b"data")]).is_err(), + "a path through a regular file is an error, not a missing output" + ); + } + #[test] fn custom_static_files_verification_and_cleanup() { let datadir = tempfile::tempdir().unwrap();