diff --git a/bin/reth/src/commands/db/mod.rs b/bin/reth/src/commands/db/mod.rs index 6eedabcc7714..1aecef717f1e 100644 --- a/bin/reth/src/commands/db/mod.rs +++ b/bin/reth/src/commands/db/mod.rs @@ -14,7 +14,7 @@ use reth_db::{ version::{get_db_version, DatabaseVersionError, DB_VERSION}, }; use reth_primitives::ChainSpec; -use reth_provider::ProviderFactory; +use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use std::{ io::{self, Write}, sync::Arc, @@ -96,7 +96,8 @@ pub enum Subcommands { macro_rules! db_ro_exec { ($chain:expr, $db_path:expr, $db_args:ident, $sfp:ident, $tool:ident, $command:block) => { let db = open_db_read_only($db_path, $db_args)?; - let provider_factory = ProviderFactory::new(db, $chain.clone(), $sfp)?; + let provider_factory = + ProviderFactory::new(db, $chain.clone(), StaticFileProvider::read_only($sfp)?); let $tool = DbTool::new(provider_factory, $chain.clone())?; $command; @@ -156,16 +157,22 @@ impl Command { } let db = open_db(&db_path, db_args)?; - let provider_factory = - ProviderFactory::new(db, self.chain.clone(), static_files_path.clone())?; + let provider_factory = ProviderFactory::new( + db, + self.chain.clone(), + StaticFileProvider::read_write(&static_files_path)?, + ); let tool = DbTool::new(provider_factory, self.chain.clone())?; tool.drop(db_path, static_files_path)?; } Subcommands::Clear(command) => { let db = open_db(&db_path, db_args)?; - let provider_factory = - ProviderFactory::new(db, self.chain.clone(), static_files_path)?; + let provider_factory = ProviderFactory::new( + db, + self.chain.clone(), + StaticFileProvider::read_write(static_files_path)?, + ); command.execute(provider_factory)?; } diff --git a/bin/reth/src/commands/db/static_files/headers.rs b/bin/reth/src/commands/db/static_files/headers.rs index 7584f614c9bf..e9dd8802679e 100644 --- a/bin/reth/src/commands/db/static_files/headers.rs +++ b/bin/reth/src/commands/db/static_files/headers.rs @@ -38,7 +38,7 @@ impl Command { let path: PathBuf = StaticFileSegment::Headers .filename_with_configuration(filters, compression, &block_range) .into(); - let provider = StaticFileProvider::new(PathBuf::default())?; + let provider = StaticFileProvider::read_only(PathBuf::default())?; let jar_provider = provider.get_segment_provider_from_block( StaticFileSegment::Headers, self.from, diff --git a/bin/reth/src/commands/db/static_files/mod.rs b/bin/reth/src/commands/db/static_files/mod.rs index 8f5930e10835..f7532237fbaa 100644 --- a/bin/reth/src/commands/db/static_files/mod.rs +++ b/bin/reth/src/commands/db/static_files/mod.rs @@ -16,7 +16,7 @@ use reth_primitives::{ }, BlockNumber, ChainSpec, StaticFileSegment, }; -use reth_provider::{BlockNumReader, ProviderFactory}; +use reth_provider::{providers::StaticFileProvider, BlockNumReader, ProviderFactory}; use reth_static_file::{segments as static_file_segments, segments::Segment}; use std::{ path::{Path, PathBuf}, @@ -99,7 +99,11 @@ impl Command { data_dir.db().as_path(), db_args.with_max_read_transaction_duration(Some(MaxReadTransactionDuration::Unbounded)), )?; - let provider_factory = Arc::new(ProviderFactory::new(db, chain, data_dir.static_files())?); + let provider_factory = Arc::new(ProviderFactory::new( + db, + chain, + StaticFileProvider::read_only(data_dir.static_files())?, + )); { if !self.only_bench { diff --git a/bin/reth/src/commands/db/static_files/receipts.rs b/bin/reth/src/commands/db/static_files/receipts.rs index 50ebd42caf6b..5c2e8ea7aee9 100644 --- a/bin/reth/src/commands/db/static_files/receipts.rs +++ b/bin/reth/src/commands/db/static_files/receipts.rs @@ -43,7 +43,7 @@ impl Command { .filename_with_configuration(filters, compression, &block_range) .into(); - let provider = StaticFileProvider::new(PathBuf::default())?; + let provider = StaticFileProvider::read_only(PathBuf::default())?; let jar_provider = provider.get_segment_provider_from_block( StaticFileSegment::Receipts, self.from, diff --git a/bin/reth/src/commands/db/static_files/transactions.rs b/bin/reth/src/commands/db/static_files/transactions.rs index a5939c0f41bb..cd8b8811df79 100644 --- a/bin/reth/src/commands/db/static_files/transactions.rs +++ b/bin/reth/src/commands/db/static_files/transactions.rs @@ -42,7 +42,7 @@ impl Command { let path: PathBuf = StaticFileSegment::Transactions .filename_with_configuration(filters, compression, &block_range) .into(); - let provider = StaticFileProvider::new(PathBuf::default())?; + let provider = StaticFileProvider::read_only(PathBuf::default())?; let jar_provider = provider.get_segment_provider_from_block( StaticFileSegment::Transactions, self.from, diff --git a/bin/reth/src/commands/db/stats.rs b/bin/reth/src/commands/db/stats.rs index 8ea473e8fe6b..c3675f387b0b 100644 --- a/bin/reth/src/commands/db/stats.rs +++ b/bin/reth/src/commands/db/stats.rs @@ -168,7 +168,7 @@ impl Command { } let static_files = iter_static_files(data_dir.static_files())?; - let static_file_provider = StaticFileProvider::new(data_dir.static_files())?; + let static_file_provider = StaticFileProvider::read_only(data_dir.static_files())?; let mut total_data_size = 0; let mut total_index_size = 0; diff --git a/bin/reth/src/commands/debug_cmd/build_block.rs b/bin/reth/src/commands/debug_cmd/build_block.rs index 632f11d758d8..f4720d81708d 100644 --- a/bin/reth/src/commands/debug_cmd/build_block.rs +++ b/bin/reth/src/commands/debug_cmd/build_block.rs @@ -35,8 +35,9 @@ use reth_primitives::{ U256, }; use reth_provider::{ - providers::BlockchainProvider, BlockHashReader, BlockReader, BlockWriter, - BundleStateWithReceipts, ProviderFactory, StageCheckpointReader, StateProviderFactory, + providers::{BlockchainProvider, StaticFileProvider}, + BlockHashReader, BlockReader, BlockWriter, BundleStateWithReceipts, ProviderFactory, + StageCheckpointReader, StateProviderFactory, }; use reth_revm::database::StateProviderDatabase; use reth_rpc_types::engine::{BlobsBundleV1, PayloadAttributes}; @@ -113,8 +114,10 @@ impl Command { let factory = ProviderFactory::new( db, self.chain.clone(), - self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), - )?; + StaticFileProvider::read_only( + self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), + )?, + ); let provider = factory.provider()?; let best_number = @@ -155,8 +158,8 @@ impl Command { let provider_factory = ProviderFactory::new( Arc::clone(&db), Arc::clone(&self.chain), - data_dir.static_files(), - )?; + StaticFileProvider::read_only(data_dir.static_files())?, + ); let consensus: Arc = Arc::new(EthBeaconConsensus::new(Arc::clone(&self.chain))); diff --git a/bin/reth/src/commands/debug_cmd/execution.rs b/bin/reth/src/commands/debug_cmd/execution.rs index 87e122b4c061..5403f450c1cc 100644 --- a/bin/reth/src/commands/debug_cmd/execution.rs +++ b/bin/reth/src/commands/debug_cmd/execution.rs @@ -31,8 +31,8 @@ use reth_primitives::{ stage::StageId, BlockHashOrNumber, BlockNumber, ChainSpec, PruneModes, B256, }; use reth_provider::{ - BlockExecutionWriter, HeaderSyncMode, ProviderFactory, StageCheckpointReader, - StaticFileProviderFactory, + providers::StaticFileProvider, BlockExecutionWriter, HeaderSyncMode, ProviderFactory, + StageCheckpointReader, StaticFileProviderFactory, }; use reth_stages::{ sets::DefaultStages, @@ -156,6 +156,9 @@ impl Command { default_peers_path: PathBuf, ) -> eyre::Result { let secret_key = get_secret_key(&network_secret_path)?; + let static_files = StaticFileProvider::read_only( + self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), + )?; let network = self .network .network_config(config, self.chain.clone(), secret_key, default_peers_path) @@ -165,11 +168,7 @@ impl Command { self.network.discovery.addr, self.network.discovery.port, )) - .build(ProviderFactory::new( - db, - self.chain.clone(), - self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), - )?) + .build(ProviderFactory::new(db, self.chain.clone(), static_files)) .start_network() .await?; info!(target: "reth::cli", peer_id = %network.peer_id(), local_addr = %network.local_addr(), "Connected to P2P network"); @@ -210,8 +209,11 @@ impl Command { fs::create_dir_all(&db_path)?; let db = Arc::new(init_db(db_path, self.db.database_args())?); - let provider_factory = - ProviderFactory::new(db.clone(), self.chain.clone(), data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db.clone(), + self.chain.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); debug!(target: "reth::cli", chain=%self.chain.chain, genesis=?self.chain.genesis_hash(), "Initializing genesis"); init_genesis(provider_factory.clone())?; diff --git a/bin/reth/src/commands/debug_cmd/in_memory_merkle.rs b/bin/reth/src/commands/debug_cmd/in_memory_merkle.rs index 4a38bec01f8a..3e1d984686c3 100644 --- a/bin/reth/src/commands/debug_cmd/in_memory_merkle.rs +++ b/bin/reth/src/commands/debug_cmd/in_memory_merkle.rs @@ -22,9 +22,9 @@ use reth_network::NetworkHandle; use reth_network_api::NetworkInfo; use reth_primitives::{stage::StageId, BlockHashOrNumber, ChainSpec, Receipts}; use reth_provider::{ - AccountExtReader, BundleStateWithReceipts, HashingWriter, HeaderProvider, - LatestStateProviderRef, OriginalValuesKnown, ProviderFactory, StageCheckpointReader, - StateWriter, StaticFileProviderFactory, StorageReader, + providers::StaticFileProvider, AccountExtReader, BundleStateWithReceipts, HashingWriter, + HeaderProvider, LatestStateProviderRef, OriginalValuesKnown, ProviderFactory, + StageCheckpointReader, StateWriter, StaticFileProviderFactory, StorageReader, }; use reth_revm::database::StateProviderDatabase; use reth_tasks::TaskExecutor; @@ -97,8 +97,10 @@ impl Command { .build(ProviderFactory::new( db, self.chain.clone(), - self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), - )?) + StaticFileProvider::read_only( + self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), + )?, + )) .start_network() .await?; info!(target: "reth::cli", peer_id = %network.peer_id(), local_addr = %network.local_addr(), "Connected to P2P network"); @@ -117,7 +119,11 @@ impl Command { // initialize the database let db = Arc::new(init_db(db_path, self.db.database_args())?); - let factory = ProviderFactory::new(&db, self.chain.clone(), data_dir.static_files())?; + let factory = ProviderFactory::new( + &db, + self.chain.clone(), + StaticFileProvider::read_only(data_dir.static_files())?, + ); let provider = factory.provider()?; // Look up merkle checkpoint diff --git a/bin/reth/src/commands/debug_cmd/merkle.rs b/bin/reth/src/commands/debug_cmd/merkle.rs index 83562d967157..081764ac4294 100644 --- a/bin/reth/src/commands/debug_cmd/merkle.rs +++ b/bin/reth/src/commands/debug_cmd/merkle.rs @@ -24,8 +24,9 @@ use reth_network_api::NetworkInfo; use reth_network_p2p::full_block::FullBlockClient; use reth_primitives::{stage::StageCheckpoint, BlockHashOrNumber, ChainSpec, PruneModes}; use reth_provider::{ - BlockNumReader, BlockWriter, BundleStateWithReceipts, HeaderProvider, LatestStateProviderRef, - OriginalValuesKnown, ProviderError, ProviderFactory, StateWriter, + providers::StaticFileProvider, BlockNumReader, BlockWriter, BundleStateWithReceipts, + HeaderProvider, LatestStateProviderRef, OriginalValuesKnown, ProviderError, ProviderFactory, + StateWriter, }; use reth_revm::database::StateProviderDatabase; use reth_stages::{ @@ -102,8 +103,10 @@ impl Command { .build(ProviderFactory::new( db, self.chain.clone(), - self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), - )?) + StaticFileProvider::read_only( + self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), + )?, + )) .start_network() .await?; info!(target: "reth::cli", peer_id = %network.peer_id(), local_addr = %network.local_addr(), "Connected to P2P network"); @@ -122,7 +125,11 @@ impl Command { // initialize the database let db = Arc::new(init_db(db_path, self.db.database_args())?); - let factory = ProviderFactory::new(&db, self.chain.clone(), data_dir.static_files())?; + let factory = ProviderFactory::new( + &db, + self.chain.clone(), + StaticFileProvider::read_only(data_dir.static_files())?, + ); let provider_rw = factory.provider_rw()?; // Configure and build network diff --git a/bin/reth/src/commands/debug_cmd/replay_engine.rs b/bin/reth/src/commands/debug_cmd/replay_engine.rs index 72031ce1b86b..98666888722b 100644 --- a/bin/reth/src/commands/debug_cmd/replay_engine.rs +++ b/bin/reth/src/commands/debug_cmd/replay_engine.rs @@ -25,8 +25,8 @@ use reth_node_core::engine::engine_store::{EngineMessageStore, StoredEngineApiMe use reth_payload_builder::{PayloadBuilderHandle, PayloadBuilderService}; use reth_primitives::{ChainSpec, PruneModes}; use reth_provider::{ - providers::BlockchainProvider, CanonStateSubscriptions, ProviderFactory, - StaticFileProviderFactory, + providers::{BlockchainProvider, StaticFileProvider}, + CanonStateSubscriptions, ProviderFactory, StaticFileProviderFactory, }; use reth_stages::Pipeline; use reth_static_file::StaticFileProducer; @@ -100,8 +100,10 @@ impl Command { .build(ProviderFactory::new( db, self.chain.clone(), - self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), - )?) + StaticFileProvider::read_only( + self.datadir.unwrap_or_chain_default(self.chain.chain).static_files(), + )?, + )) .start_network() .await?; info!(target: "reth::cli", peer_id = %network.peer_id(), local_addr = %network.local_addr(), "Connected to P2P network"); @@ -120,8 +122,11 @@ impl Command { // Initialize the database let db = Arc::new(init_db(db_path, self.db.database_args())?); - let provider_factory = - ProviderFactory::new(db.clone(), self.chain.clone(), data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db.clone(), + self.chain.clone(), + StaticFileProvider::read_only(data_dir.static_files())?, + ); let consensus: Arc = Arc::new(EthBeaconConsensus::new(Arc::clone(&self.chain))); diff --git a/bin/reth/src/commands/import.rs b/bin/reth/src/commands/import.rs index 869bcefbd643..9fb967dd9e1d 100644 --- a/bin/reth/src/commands/import.rs +++ b/bin/reth/src/commands/import.rs @@ -29,8 +29,9 @@ use reth_network_p2p::{ use reth_node_events::node::NodeEvent; use reth_primitives::{stage::StageId, ChainSpec, PruneModes, B256}; use reth_provider::{ - BlockNumReader, ChainSpecProvider, HeaderProvider, HeaderSyncMode, ProviderError, - ProviderFactory, StageCheckpointReader, StaticFileProviderFactory, + providers::StaticFileProvider, BlockNumReader, ChainSpecProvider, HeaderProvider, + HeaderSyncMode, ProviderError, ProviderFactory, StageCheckpointReader, + StaticFileProviderFactory, }; use reth_stages::{prelude::*, Pipeline, StageSet}; use reth_static_file::StaticFileProducer; @@ -117,8 +118,11 @@ impl ImportCommand { info!(target: "reth::cli", path = ?db_path, "Opening database"); let db = Arc::new(init_db(db_path, self.db.database_args())?); info!(target: "reth::cli", "Database opened"); - let provider_factory = - ProviderFactory::new(db.clone(), self.chain.clone(), data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db.clone(), + self.chain.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); debug!(target: "reth::cli", chain=%self.chain.chain, genesis=?self.chain.genesis_hash(), "Initializing genesis"); diff --git a/bin/reth/src/commands/import_op.rs b/bin/reth/src/commands/import_op.rs index 1da75951b991..bb2ef4567b69 100644 --- a/bin/reth/src/commands/import_op.rs +++ b/bin/reth/src/commands/import_op.rs @@ -20,7 +20,10 @@ use reth_downloaders::file_client::{ }; use reth_optimism_primitives::bedrock_import::is_dup_tx; use reth_primitives::{stage::StageId, PruneModes}; -use reth_provider::{ProviderFactory, StageCheckpointReader, StaticFileProviderFactory}; +use reth_provider::{ + providers::StaticFileProvider, ProviderFactory, StageCheckpointReader, + StaticFileProviderFactory, +}; use reth_static_file::StaticFileProducer; use std::{path::PathBuf, sync::Arc}; use tracing::{debug, error, info}; @@ -91,8 +94,11 @@ impl ImportOpCommand { let db = Arc::new(init_db(db_path, self.db.database_args())?); info!(target: "reth::cli", "Database opened"); - let provider_factory = - ProviderFactory::new(db.clone(), chain_spec.clone(), data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db.clone(), + chain_spec.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); debug!(target: "reth::cli", chain=%chain_spec.chain, genesis=?chain_spec.genesis_hash(), "Initializing genesis"); diff --git a/bin/reth/src/commands/import_receipts_op.rs b/bin/reth/src/commands/import_receipts_op.rs index 44c79cd2fe6a..6fefc4ea2a35 100644 --- a/bin/reth/src/commands/import_receipts_op.rs +++ b/bin/reth/src/commands/import_receipts_op.rs @@ -16,8 +16,8 @@ use reth_node_core::version::SHORT_VERSION; use reth_optimism_primitives::bedrock_import::is_dup_tx; use reth_primitives::{stage::StageId, Receipts, StaticFileSegment}; use reth_provider::{ - BundleStateWithReceipts, OriginalValuesKnown, ProviderFactory, StageCheckpointReader, - StateWriter, StaticFileProviderFactory, StaticFileWriter, StatsReader, + providers::StaticFileProvider, BundleStateWithReceipts, OriginalValuesKnown, ProviderFactory, + StageCheckpointReader, StateWriter, StaticFileProviderFactory, StaticFileWriter, StatsReader, }; use tracing::{debug, error, info, trace}; @@ -77,8 +77,11 @@ impl ImportReceiptsOpCommand { let db = Arc::new(init_db(db_path, self.db.database_args())?); info!(target: "reth::cli", "Database opened"); - let provider_factory = - ProviderFactory::new(db.clone(), chain_spec.clone(), data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db.clone(), + chain_spec.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); import_receipts_from_file( provider_factory, diff --git a/bin/reth/src/commands/init_cmd.rs b/bin/reth/src/commands/init_cmd.rs index 3b900b3f01a2..0da65aba5033 100644 --- a/bin/reth/src/commands/init_cmd.rs +++ b/bin/reth/src/commands/init_cmd.rs @@ -11,7 +11,7 @@ use clap::Parser; use reth_db::init_db; use reth_db_common::init::init_genesis; use reth_primitives::ChainSpec; -use reth_provider::ProviderFactory; +use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use std::sync::Arc; use tracing::info; @@ -56,7 +56,11 @@ impl InitCommand { let db = Arc::new(init_db(&db_path, self.db.database_args())?); info!(target: "reth::cli", "Database opened"); - let provider_factory = ProviderFactory::new(db, self.chain, data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db, + self.chain, + StaticFileProvider::read_write(data_dir.static_files())?, + ); info!(target: "reth::cli", "Writing genesis block"); diff --git a/bin/reth/src/commands/init_state.rs b/bin/reth/src/commands/init_state.rs index f5ee0c4b1c15..b672f11f2063 100644 --- a/bin/reth/src/commands/init_state.rs +++ b/bin/reth/src/commands/init_state.rs @@ -12,7 +12,7 @@ use reth_config::config::EtlConfig; use reth_db::{database::Database, init_db}; use reth_db_common::init::init_from_state_dump; use reth_primitives::{ChainSpec, B256}; -use reth_provider::ProviderFactory; +use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use std::{fs::File, io::BufReader, path::PathBuf, sync::Arc}; use tracing::info; @@ -78,7 +78,11 @@ impl InitStateCommand { let db = Arc::new(init_db(&db_path, self.db.database_args())?); info!(target: "reth::cli", "Database opened"); - let provider_factory = ProviderFactory::new(db, self.chain, data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db, + self.chain, + StaticFileProvider::read_write(data_dir.static_files())?, + ); let etl_config = EtlConfig::new( Some(EtlConfig::from_datadir(data_dir.data_dir())), EtlConfig::default_file_size(), diff --git a/bin/reth/src/commands/p2p/mod.rs b/bin/reth/src/commands/p2p/mod.rs index 64233d6e5b36..627c2d36b44f 100644 --- a/bin/reth/src/commands/p2p/mod.rs +++ b/bin/reth/src/commands/p2p/mod.rs @@ -17,7 +17,7 @@ use reth_db::create_db; use reth_network::NetworkConfigBuilder; use reth_network_p2p::bodies::client::BodiesClient; use reth_primitives::{BlockHashOrNumber, ChainSpec}; -use reth_provider::ProviderFactory; +use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use std::{ net::{IpAddr, SocketAddrV4, SocketAddrV6}, path::PathBuf, @@ -165,8 +165,8 @@ impl Command { .build(Arc::new(ProviderFactory::new( noop_db, self.chain.clone(), - data_dir.static_files(), - )?)) + StaticFileProvider::read_write(data_dir.static_files())?, + ))) .start_network() .await?; diff --git a/bin/reth/src/commands/recover/storage_tries.rs b/bin/reth/src/commands/recover/storage_tries.rs index 583829bc39bc..8aa568974925 100644 --- a/bin/reth/src/commands/recover/storage_tries.rs +++ b/bin/reth/src/commands/recover/storage_tries.rs @@ -12,7 +12,9 @@ use reth_db::{ use reth_db_common::init::init_genesis; use reth_node_core::args::DatabaseArgs; use reth_primitives::ChainSpec; -use reth_provider::{BlockNumReader, HeaderProvider, ProviderError, ProviderFactory}; +use reth_provider::{ + providers::StaticFileProvider, BlockNumReader, HeaderProvider, ProviderError, ProviderFactory, +}; use reth_trie::StateRoot; use std::{fs, sync::Arc}; use tracing::*; @@ -55,7 +57,11 @@ impl Command { fs::create_dir_all(&db_path)?; let db = Arc::new(init_db(db_path, self.db.database_args())?); - let factory = ProviderFactory::new(&db, self.chain.clone(), data_dir.static_files())?; + let factory = ProviderFactory::new( + &db, + self.chain.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); debug!(target: "reth::cli", chain=%self.chain.chain, genesis=?self.chain.genesis_hash(), "Initializing genesis"); init_genesis(factory.clone())?; diff --git a/bin/reth/src/commands/stage/drop.rs b/bin/reth/src/commands/stage/drop.rs index 47af150ce526..042bafa3ada7 100644 --- a/bin/reth/src/commands/stage/drop.rs +++ b/bin/reth/src/commands/stage/drop.rs @@ -16,7 +16,10 @@ use reth_fs_util as fs; use reth_primitives::{ stage::StageId, static_file::find_fixed_range, ChainSpec, StaticFileSegment, }; -use reth_provider::{providers::StaticFileWriter, ProviderFactory, StaticFileProviderFactory}; +use reth_provider::{ + providers::{StaticFileProvider, StaticFileWriter}, + ProviderFactory, StaticFileProviderFactory, +}; use std::sync::Arc; /// `reth drop-stage` command @@ -59,8 +62,11 @@ impl Command { fs::create_dir_all(&db_path)?; let db = open_db(db_path.as_ref(), self.db.database_args())?; - let provider_factory = - ProviderFactory::new(db, self.chain.clone(), data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db, + self.chain.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); let static_file_provider = provider_factory.static_file_provider(); let tool = DbTool::new(provider_factory, self.chain.clone())?; diff --git a/bin/reth/src/commands/stage/dump/execution.rs b/bin/reth/src/commands/stage/dump/execution.rs index d8f12b50af7c..abe1fccd70fc 100644 --- a/bin/reth/src/commands/stage/dump/execution.rs +++ b/bin/reth/src/commands/stage/dump/execution.rs @@ -6,7 +6,7 @@ use reth_db::{ }; use reth_node_core::dirs::{ChainPath, DataDirPath}; use reth_primitives::stage::StageCheckpoint; -use reth_provider::{ChainSpecProvider, ProviderFactory}; +use reth_provider::{providers::StaticFileProvider, ChainSpecProvider, ProviderFactory}; use reth_stages::{stages::ExecutionStage, Stage, UnwindInput}; use tracing::info; @@ -25,7 +25,11 @@ pub(crate) async fn dump_execution_stage( if should_run { dry_run( - ProviderFactory::new(output_db, db_tool.chain.clone(), output_datadir.static_files())?, + ProviderFactory::new( + output_db, + db_tool.chain.clone(), + StaticFileProvider::read_only(output_datadir.static_files())?, + ), to, from, ) diff --git a/bin/reth/src/commands/stage/dump/hashing_account.rs b/bin/reth/src/commands/stage/dump/hashing_account.rs index 2f28ba129a10..ebcf1ad8c093 100644 --- a/bin/reth/src/commands/stage/dump/hashing_account.rs +++ b/bin/reth/src/commands/stage/dump/hashing_account.rs @@ -4,7 +4,7 @@ use eyre::Result; use reth_db::{database::Database, table::TableImporter, tables, DatabaseEnv}; use reth_node_core::dirs::{ChainPath, DataDirPath}; use reth_primitives::{stage::StageCheckpoint, BlockNumber}; -use reth_provider::ProviderFactory; +use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use reth_stages::{stages::AccountHashingStage, Stage, UnwindInput}; use tracing::info; @@ -30,7 +30,11 @@ pub(crate) async fn dump_hashing_account_stage( if should_run { dry_run( - ProviderFactory::new(output_db, db_tool.chain.clone(), output_datadir.static_files())?, + ProviderFactory::new( + output_db, + db_tool.chain.clone(), + StaticFileProvider::read_only(output_datadir.static_files())?, + ), to, from, ) diff --git a/bin/reth/src/commands/stage/dump/hashing_storage.rs b/bin/reth/src/commands/stage/dump/hashing_storage.rs index 7d38892dc8bd..f990357d1895 100644 --- a/bin/reth/src/commands/stage/dump/hashing_storage.rs +++ b/bin/reth/src/commands/stage/dump/hashing_storage.rs @@ -4,7 +4,7 @@ use eyre::Result; use reth_db::{database::Database, table::TableImporter, tables, DatabaseEnv}; use reth_node_core::dirs::{ChainPath, DataDirPath}; use reth_primitives::stage::StageCheckpoint; -use reth_provider::ProviderFactory; +use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use reth_stages::{stages::StorageHashingStage, Stage, UnwindInput}; use tracing::info; @@ -21,7 +21,11 @@ pub(crate) async fn dump_hashing_storage_stage( if should_run { dry_run( - ProviderFactory::new(output_db, db_tool.chain.clone(), output_datadir.static_files())?, + ProviderFactory::new( + output_db, + db_tool.chain.clone(), + StaticFileProvider::read_only(output_datadir.static_files())?, + ), to, from, ) diff --git a/bin/reth/src/commands/stage/dump/merkle.rs b/bin/reth/src/commands/stage/dump/merkle.rs index 9b421be7ca3f..bb3ab80bfb16 100644 --- a/bin/reth/src/commands/stage/dump/merkle.rs +++ b/bin/reth/src/commands/stage/dump/merkle.rs @@ -6,7 +6,7 @@ use reth_db::{database::Database, table::TableImporter, tables, DatabaseEnv}; use reth_exex::ExExManagerHandle; use reth_node_core::dirs::{ChainPath, DataDirPath}; use reth_primitives::{stage::StageCheckpoint, BlockNumber, PruneModes}; -use reth_provider::ProviderFactory; +use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use reth_stages::{ stages::{ AccountHashingStage, ExecutionStage, ExecutionStageThresholds, MerkleStage, @@ -45,7 +45,11 @@ pub(crate) async fn dump_merkle_stage( if should_run { dry_run( - ProviderFactory::new(output_db, db_tool.chain.clone(), output_datadir.static_files())?, + ProviderFactory::new( + output_db, + db_tool.chain.clone(), + StaticFileProvider::read_only(output_datadir.static_files())?, + ), to, from, ) diff --git a/bin/reth/src/commands/stage/dump/mod.rs b/bin/reth/src/commands/stage/dump/mod.rs index fa4184356558..03cff6055ce5 100644 --- a/bin/reth/src/commands/stage/dump/mod.rs +++ b/bin/reth/src/commands/stage/dump/mod.rs @@ -17,7 +17,7 @@ use reth_db::{ }; use reth_node_core::dirs::PlatformPath; use reth_primitives::ChainSpec; -use reth_provider::ProviderFactory; +use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use std::{path::PathBuf, sync::Arc}; use tracing::info; @@ -105,8 +105,11 @@ impl Command { let db_path = data_dir.db(); info!(target: "reth::cli", path = ?db_path, "Opening database"); let db = Arc::new(init_db(db_path, self.db.database_args())?); - let provider_factory = - ProviderFactory::new(db, self.chain.clone(), data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db, + self.chain.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); info!(target: "reth::cli", "Database opened"); diff --git a/bin/reth/src/commands/stage/run.rs b/bin/reth/src/commands/stage/run.rs index d34b67db42a1..e6b96bdc6a4a 100644 --- a/bin/reth/src/commands/stage/run.rs +++ b/bin/reth/src/commands/stage/run.rs @@ -25,7 +25,8 @@ use reth_downloaders::bodies::bodies::BodiesDownloaderBuilder; use reth_exex::ExExManagerHandle; use reth_primitives::ChainSpec; use reth_provider::{ - ProviderFactory, StageCheckpointReader, StageCheckpointWriter, StaticFileProviderFactory, + providers::StaticFileProvider, ProviderFactory, StageCheckpointReader, StageCheckpointWriter, + StaticFileProviderFactory, }; use reth_stages::{ stages::{ @@ -145,8 +146,11 @@ impl Command { let db = Arc::new(init_db(db_path, self.db.database_args())?); info!(target: "reth::cli", "Database opened"); - let factory = - ProviderFactory::new(Arc::clone(&db), self.chain.clone(), data_dir.static_files())?; + let factory = ProviderFactory::new( + Arc::clone(&db), + self.chain.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); let mut provider_rw = factory.provider_rw()?; if let Some(listen_addr) = self.metrics { @@ -195,8 +199,8 @@ impl Command { let provider_factory = Arc::new(ProviderFactory::new( db.clone(), self.chain.clone(), - data_dir.static_files(), - )?); + StaticFileProvider::read_write(data_dir.static_files())?, + )); let network = self .network diff --git a/bin/reth/src/commands/stage/unwind.rs b/bin/reth/src/commands/stage/unwind.rs index 320539280975..e5714f57b708 100644 --- a/bin/reth/src/commands/stage/unwind.rs +++ b/bin/reth/src/commands/stage/unwind.rs @@ -10,8 +10,8 @@ use reth_exex::ExExManagerHandle; use reth_node_core::args::NetworkArgs; use reth_primitives::{BlockHashOrNumber, ChainSpec, PruneModes, B256}; use reth_provider::{ - BlockExecutionWriter, BlockNumReader, ChainSpecProvider, HeaderSyncMode, ProviderFactory, - StaticFileProviderFactory, + providers::StaticFileProvider, BlockExecutionWriter, BlockNumReader, ChainSpecProvider, + HeaderSyncMode, ProviderFactory, StaticFileProviderFactory, }; use reth_stages::{ sets::DefaultStages, @@ -81,8 +81,11 @@ impl Command { let config: Config = confy::load_path(config_path).unwrap_or_default(); let db = Arc::new(open_db(db_path.as_ref(), self.db.database_args())?); - let provider_factory = - ProviderFactory::new(db, self.chain.clone(), data_dir.static_files())?; + let provider_factory = ProviderFactory::new( + db, + self.chain.clone(), + StaticFileProvider::read_write(data_dir.static_files())?, + ); let range = self.command.unwind_range(provider_factory.clone())?; if *range.start() == 0 { diff --git a/crates/blockchain-tree-api/src/error.rs b/crates/blockchain-tree-api/src/error.rs index ad9f7abcd3e8..ae9365aa6b06 100644 --- a/crates/blockchain-tree-api/src/error.rs +++ b/crates/blockchain-tree-api/src/error.rs @@ -296,7 +296,6 @@ impl InsertBlockErrorKind { BlockExecutionError::CanonicalRevert { .. } | BlockExecutionError::CanonicalCommit { .. } | BlockExecutionError::AppendChainDoesntConnect { .. } | - BlockExecutionError::UnavailableForTest => false, BlockExecutionError::Other(_) => false, } } diff --git a/crates/config/src/config.rs b/crates/config/src/config.rs index b804f7f9aa17..65fc1c647ab8 100644 --- a/crates/config/src/config.rs +++ b/crates/config/src/config.rs @@ -327,6 +327,13 @@ impl Default for PruneConfig { } } +impl PruneConfig { + /// Returns whether there is any kind of receipt pruning configuration. + pub fn has_receipts_pruning(&self) -> bool { + self.segments.receipts.is_some() || !self.segments.receipts_log_filter.is_empty() + } +} + /// Helper type to support older versions of Duration deserialization. fn deserialize_duration<'de, D>(deserializer: D) -> Result, D::Error> where diff --git a/crates/consensus/beacon/src/engine/mod.rs b/crates/consensus/beacon/src/engine/mod.rs index f86452b5fc12..4a47de7b7a56 100644 --- a/crates/consensus/beacon/src/engine/mod.rs +++ b/crates/consensus/beacon/src/engine/mod.rs @@ -2155,6 +2155,7 @@ mod tests { use super::*; use reth_db::{tables, test_utils::create_test_static_files_dir, transaction::DbTxMut}; use reth_primitives::U256; + use reth_provider::providers::StaticFileProvider; use reth_rpc_types::engine::ForkchoiceUpdateError; use reth_testing_utils::generators::random_block; @@ -2211,8 +2212,11 @@ mod tests { let (_static_dir, static_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&genesis, &block1].into_iter(), ); env.db @@ -2268,8 +2272,11 @@ mod tests { let (_static_dir, static_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&genesis, &block1].into_iter(), ); @@ -2289,8 +2296,11 @@ mod tests { // Insert next head immediately after sending forkchoice update insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&next_head].into_iter(), ); @@ -2330,8 +2340,11 @@ mod tests { let (_static_dir, static_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&genesis, &block1].into_iter(), ); @@ -2382,8 +2395,11 @@ mod tests { let (_static_dir, static_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&genesis, &block1, &block2, &block3].into_iter(), ); @@ -2428,8 +2444,11 @@ mod tests { let (_temp_dir, temp_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), temp_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(temp_dir_path).unwrap(), + ), [&genesis, &block1].into_iter(), ); @@ -2455,9 +2474,10 @@ mod tests { use super::*; use reth_db::test_utils::create_test_static_files_dir; use reth_primitives::{genesis::Genesis, Hardfork, U256}; - use reth_provider::test_utils::blocks::BlockchainTestData; + use reth_provider::{ + providers::StaticFileProvider, test_utils::blocks::BlockchainTestData, + }; use reth_testing_utils::{generators::random_block, GenesisAllocator}; - #[tokio::test] async fn new_payload_before_forkchoice() { let mut rng = generators::rng(); @@ -2527,8 +2547,11 @@ mod tests { let (_static_dir, static_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&genesis, &block1, &block2].into_iter(), ); @@ -2596,8 +2619,11 @@ mod tests { let (_static_dir, static_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&genesis, &block1].into_iter(), ); @@ -2640,8 +2666,11 @@ mod tests { let (_static_dir, static_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&genesis].into_iter(), ); @@ -2704,8 +2733,11 @@ mod tests { let (_static_dir, static_dir_path) = create_test_static_files_dir(); insert_blocks( - ProviderFactory::new(env.db.as_ref(), chain_spec.clone(), static_dir_path) - .expect("create provider factory with static_files"), + ProviderFactory::new( + env.db.as_ref(), + chain_spec.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), [&data.genesis, &block1].into_iter(), ); diff --git a/crates/evm/execution-errors/src/lib.rs b/crates/evm/execution-errors/src/lib.rs index 4d954ba1e178..08db509d8e0c 100644 --- a/crates/evm/execution-errors/src/lib.rs +++ b/crates/evm/execution-errors/src/lib.rs @@ -131,11 +131,6 @@ pub enum BlockExecutionError { /// The fork on the other chain other_chain_fork: Box, }, - /// Only used for TestExecutor - /// - /// Note: this is not feature gated for convenience. - #[error("execution unavailable for tests")] - UnavailableForTest, /// Error when fetching latest block state. #[error(transparent)] LatestBlock(#[from] ProviderError), diff --git a/crates/evm/src/execute.rs b/crates/evm/src/execute.rs index 8409f5c3742d..beefdac7fc42 100644 --- a/crates/evm/src/execute.rs +++ b/crates/evm/src/execute.rs @@ -245,7 +245,7 @@ mod tests { type Error = BlockExecutionError; fn execute(self, _input: Self::Input<'_>) -> Result { - Err(BlockExecutionError::UnavailableForTest) + Err(BlockExecutionError::msg("execution unavailable for tests")) } } diff --git a/crates/evm/src/lib.rs b/crates/evm/src/lib.rs index 93e8035258cf..c7f90057f4b8 100644 --- a/crates/evm/src/lib.rs +++ b/crates/evm/src/lib.rs @@ -16,6 +16,7 @@ use revm_primitives::{BlockEnv, CfgEnvWithHandlerCfg, EnvWithHandlerCfg, SpecId, pub mod either; pub mod execute; +pub mod noop; pub mod provider; #[cfg(any(test, feature = "test-utils"))] diff --git a/crates/evm/src/noop.rs b/crates/evm/src/noop.rs new file mode 100644 index 000000000000..220bbb39a84b --- /dev/null +++ b/crates/evm/src/noop.rs @@ -0,0 +1,68 @@ +//! A no operation block executor implementation. + +use reth_execution_errors::BlockExecutionError; +use reth_primitives::{BlockNumber, BlockWithSenders, PruneModes, Receipt}; +use reth_storage_errors::provider::ProviderError; +use revm_primitives::db::Database; + +use crate::execute::{ + BatchBlockExecutionOutput, BatchExecutor, BlockExecutionInput, BlockExecutionOutput, + BlockExecutorProvider, Executor, +}; + +const UNAVAILABLE_FOR_NOOP: &str = "execution unavailable for noop"; + +/// A [BlockExecutorProvider] implementation that does nothing. +#[derive(Debug, Default, Clone)] +#[non_exhaustive] +pub struct NoopBlockExecutorProvider; + +impl BlockExecutorProvider for NoopBlockExecutorProvider { + type Executor> = Self; + + type BatchExecutor> = Self; + + fn executor(&self, _: DB) -> Self::Executor + where + DB: Database, + { + Self + } + + fn batch_executor(&self, _: DB, _: PruneModes) -> Self::BatchExecutor + where + DB: Database, + { + Self + } +} + +impl Executor for NoopBlockExecutorProvider { + type Input<'a> = BlockExecutionInput<'a, BlockWithSenders>; + type Output = BlockExecutionOutput; + type Error = BlockExecutionError; + + fn execute(self, _: Self::Input<'_>) -> Result { + Err(BlockExecutionError::msg(UNAVAILABLE_FOR_NOOP)) + } +} + +impl BatchExecutor for NoopBlockExecutorProvider { + type Input<'a> = BlockExecutionInput<'a, BlockWithSenders>; + type Output = BatchBlockExecutionOutput; + type Error = BlockExecutionError; + + fn execute_and_verify_one(&mut self, _: Self::Input<'_>) -> Result<(), Self::Error> { + Err(BlockExecutionError::msg(UNAVAILABLE_FOR_NOOP)) + } + + fn finalize(self) -> Self::Output { + unreachable!() + } + + fn set_tip(&mut self, _: BlockNumber) {} + + fn size_hint(&self) -> Option { + None + } +} diff --git a/crates/net/downloaders/src/bodies/bodies.rs b/crates/net/downloaders/src/bodies/bodies.rs index e5042691ed6f..1f7a78d31092 100644 --- a/crates/net/downloaders/src/bodies/bodies.rs +++ b/crates/net/downloaders/src/bodies/bodies.rs @@ -607,7 +607,7 @@ mod tests { use reth_consensus::test_utils::TestConsensus; use reth_db::test_utils::{create_test_rw_db, create_test_static_files_dir}; use reth_primitives::{BlockBody, B256, MAINNET}; - use reth_provider::ProviderFactory; + use reth_provider::{providers::StaticFileProvider, ProviderFactory}; use reth_testing_utils::{generators, generators::random_block_range}; use std::collections::HashMap; @@ -629,7 +629,11 @@ mod tests { let mut downloader = BodiesDownloaderBuilder::default().build( client.clone(), Arc::new(TestConsensus::default()), - ProviderFactory::new(db, MAINNET.clone(), static_dir_path).unwrap(), + ProviderFactory::new( + db, + MAINNET.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), ); downloader.set_download_range(0..=19).expect("failed to set download range"); @@ -675,7 +679,11 @@ mod tests { BodiesDownloaderBuilder::default().with_request_limit(request_limit).build( client.clone(), Arc::new(TestConsensus::default()), - ProviderFactory::new(db, MAINNET.clone(), static_dir_path).unwrap(), + ProviderFactory::new( + db, + MAINNET.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), ); downloader.set_download_range(0..=199).expect("failed to set download range"); @@ -705,7 +713,11 @@ mod tests { .build( client.clone(), Arc::new(TestConsensus::default()), - ProviderFactory::new(db, MAINNET.clone(), static_dir_path).unwrap(), + ProviderFactory::new( + db, + MAINNET.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), ); let mut range_start = 0; @@ -737,7 +749,11 @@ mod tests { let mut downloader = BodiesDownloaderBuilder::default().with_stream_batch_size(100).build( client.clone(), Arc::new(TestConsensus::default()), - ProviderFactory::new(db, MAINNET.clone(), static_dir_path).unwrap(), + ProviderFactory::new( + db, + MAINNET.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), ); // Set and download the first range @@ -779,7 +795,11 @@ mod tests { .build( client.clone(), Arc::new(TestConsensus::default()), - ProviderFactory::new(db, MAINNET.clone(), static_dir_path).unwrap(), + ProviderFactory::new( + db, + MAINNET.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), ); // Set and download the entire range @@ -812,7 +832,11 @@ mod tests { .build( client.clone(), Arc::new(TestConsensus::default()), - ProviderFactory::new(db, MAINNET.clone(), static_dir_path).unwrap(), + ProviderFactory::new( + db, + MAINNET.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), ); // Download the requested range diff --git a/crates/node/builder/src/launch/common.rs b/crates/node/builder/src/launch/common.rs index 0f16369fda44..8105b9dfae26 100644 --- a/crates/node/builder/src/launch/common.rs +++ b/crates/node/builder/src/launch/common.rs @@ -3,24 +3,32 @@ use eyre::Context; use rayon::ThreadPoolBuilder; use reth_auto_seal_consensus::MiningMode; +use reth_beacon_consensus::EthBeaconConsensus; use reth_config::{config::EtlConfig, PruneConfig}; use reth_db::{database::Database, database_metrics::DatabaseMetrics}; use reth_db_common::init::{init_genesis, InitDatabaseError}; +use reth_downloaders::{bodies::noop::NoopBodiesDownloader, headers::noop::NoopHeaderDownloader}; +use reth_evm::noop::NoopBlockExecutorProvider; use reth_network_p2p::headers::client::HeadersClient; use reth_node_core::{ cli::config::RethRpcConfig, dirs::{ChainPath, DataDirPath}, node_config::NodeConfig, }; -use reth_primitives::{BlockNumber, Chain, ChainSpec, Head, PruneModes, B256}; -use reth_provider::{providers::StaticFileProvider, ProviderFactory, StaticFileProviderFactory}; +use reth_primitives::{ + stage::PipelineTarget, BlockNumber, Chain, ChainSpec, Head, PruneModes, B256, +}; +use reth_provider::{ + providers::StaticFileProvider, HeaderSyncMode, ProviderFactory, StaticFileProviderFactory, +}; use reth_prune::PrunerBuilder; use reth_rpc_layer::JwtSecret; +use reth_stages::{sets::DefaultStages, Pipeline}; use reth_static_file::StaticFileProducer; use reth_tasks::TaskExecutor; use reth_tracing::tracing::{debug, error, info, warn}; use std::{sync::Arc, thread::available_parallelism}; -use tokio::sync::mpsc::Receiver; +use tokio::sync::{mpsc::Receiver, oneshot}; /// Reusable setup for launching a node. /// @@ -319,25 +327,81 @@ impl LaunchContextWith> { impl LaunchContextWith> where - DB: Clone, + DB: Database + Clone + 'static, { - /// Returns the [ProviderFactory] for the attached database. - pub fn create_provider_factory(&self) -> eyre::Result> { + /// Returns the [ProviderFactory] for the attached storage after executing a consistent check + /// between the database and static files. **It may execute a pipeline unwind if it fails this + /// check.** + pub async fn create_provider_factory(&self) -> eyre::Result> { let factory = ProviderFactory::new( self.right().clone(), self.chain_spec(), - self.data_dir().static_files(), - )? + StaticFileProvider::read_write(self.data_dir().static_files())?, + ) .with_static_files_metrics(); + let has_receipt_pruning = + self.toml_config().prune.as_ref().map_or(false, |a| a.has_receipts_pruning()); + + info!(target: "reth::cli", "Verifying storage consistency."); + + // Check for consistency between database and static files. If it fails, it unwinds to + // the first block that's consistent between database and static files. + if let Some(unwind_target) = factory + .static_file_provider() + .check_consistency(&factory.provider()?, has_receipt_pruning)? + { + // Highly unlikely to happen, and given its destructive nature, it's better to panic + // instead. + if PipelineTarget::Unwind(0) == unwind_target { + panic!("A static file <> database inconsistency was found that would trigger an unwind to block 0.") + } + + info!(target: "reth::cli", unwind_target = %unwind_target, "Executing an unwind after a failed storage consistency check."); + + // Builds an unwind-only pipeline + let pipeline = Pipeline::builder() + .add_stages(DefaultStages::new( + factory.clone(), + HeaderSyncMode::Continuous, + Arc::new(EthBeaconConsensus::new(self.chain_spec())), + NoopHeaderDownloader::default(), + NoopBodiesDownloader::default(), + NoopBlockExecutorProvider::default(), + self.toml_config().stages.clone(), + self.prune_modes().unwrap_or_default(), + )) + .build( + factory.clone(), + StaticFileProducer::new( + factory.clone(), + factory.static_file_provider(), + self.prune_modes().unwrap_or_default(), + ), + ); + + // Unwinds to block + let (tx, rx) = oneshot::channel(); + + // Pipeline should be run as blocking and panic if it fails. + self.task_executor().spawn_critical_blocking( + "pipeline task", + Box::pin(async move { + let (_, result) = pipeline.run_as_fut(Some(unwind_target)).await; + let _ = tx.send(result); + }), + ); + rx.await??; + } + Ok(factory) } /// Creates a new [ProviderFactory] and attaches it to the launch context. - pub fn with_provider_factory( + pub async fn with_provider_factory( self, ) -> eyre::Result>>> { - let factory = self.create_provider_factory()?; + let factory = self.create_provider_factory().await?; let ctx = LaunchContextWith { inner: self.inner, attachment: self.attachment.map_right(|_| factory), diff --git a/crates/node/builder/src/launch/mod.rs b/crates/node/builder/src/launch/mod.rs index 2df6e7824962..8146461d6974 100644 --- a/crates/node/builder/src/launch/mod.rs +++ b/crates/node/builder/src/launch/mod.rs @@ -100,7 +100,7 @@ where // ensure certain settings take effect .with_adjusted_configs() // Create the provider factory - .with_provider_factory()? + .with_provider_factory().await? .inspect(|_| { info!(target: "reth::cli", "Database opened"); }) diff --git a/crates/primitives/src/stage/mod.rs b/crates/primitives/src/stage/mod.rs index 74f84409b10e..6637cb41e8fa 100644 --- a/crates/primitives/src/stage/mod.rs +++ b/crates/primitives/src/stage/mod.rs @@ -12,7 +12,7 @@ pub use checkpoints::{ }; /// Direction and target block for pipeline operations. -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum PipelineTarget { /// Target for forward synchronization, indicating a block hash to sync to. Sync(BlockHash), @@ -53,3 +53,14 @@ impl From for PipelineTarget { Self::Sync(hash) } } + +impl std::fmt::Display for PipelineTarget { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Sync(block) => { + write!(f, "Sync({block})") + } + Self::Unwind(block) => write!(f, "Unwind({block})"), + } + } +} diff --git a/crates/prune/src/pruner.rs b/crates/prune/src/pruner.rs index f4111f131a50..c6e0fffae8eb 100644 --- a/crates/prune/src/pruner.rs +++ b/crates/prune/src/pruner.rs @@ -335,14 +335,17 @@ mod tests { use crate::Pruner; use reth_db::test_utils::{create_test_rw_db, create_test_static_files_dir}; use reth_primitives::{FinishedExExHeight, MAINNET}; - use reth_provider::ProviderFactory; + use reth_provider::{providers::StaticFileProvider, ProviderFactory}; #[test] fn is_pruning_needed() { let db = create_test_rw_db(); let (_static_dir, static_dir_path) = create_test_static_files_dir(); - let provider_factory = ProviderFactory::new(db, MAINNET.clone(), static_dir_path) - .expect("create provide factory with static_files"); + let provider_factory = ProviderFactory::new( + db, + MAINNET.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ); let (finished_exex_height_tx, finished_exex_height_rx) = tokio::sync::watch::channel(FinishedExExHeight::NoExExs); diff --git a/crates/stages/api/src/pipeline/mod.rs b/crates/stages/api/src/pipeline/mod.rs index e8c0cefd736a..6b7b65b9b89f 100644 --- a/crates/stages/api/src/pipeline/mod.rs +++ b/crates/stages/api/src/pipeline/mod.rs @@ -316,7 +316,7 @@ where continue } - debug!( + info!( target: "sync::pipeline", from = %checkpoint.block_number, %to, @@ -353,8 +353,12 @@ where self.event_sender .notify(PipelineEvent::Unwound { stage_id, result: unwind_output }); - self.provider_factory.static_file_provider().commit()?; + // For unwinding it makes more sense to commit the database first, since if + // this function is interrupted before the static files commit, we can just + // truncate the static files according to the + // checkpoints on the next start-up. provider_rw.commit()?; + self.provider_factory.static_file_provider().commit()?; provider_rw = self.provider_factory.provider_rw()?; } @@ -459,6 +463,10 @@ where result: out.clone(), }); + // For execution it makes more sense to commit the static files first, since if + // this function is interrupted before the database commit, we can just truncate + // the static files according to the checkpoints on the next + // start-up. self.provider_factory.static_file_provider().commit()?; provider_rw.commit()?; diff --git a/crates/stages/stages/src/stages/execution.rs b/crates/stages/stages/src/stages/execution.rs index 2388d0d6ff9d..e76b210925de 100644 --- a/crates/stages/stages/src/stages/execution.rs +++ b/crates/stages/stages/src/stages/execution.rs @@ -606,7 +606,9 @@ where // Check if we had any unexpected shutdown after committing to static files, but // NOT committing to database. match next_static_file_receipt_num.cmp(&next_receipt_num) { - Ordering::Greater => static_file_producer.prune_receipts( + // It can be equal when it's a chain of empty blocks, but we still need to update the last + // block in the range. + Ordering::Greater | Ordering::Equal => static_file_producer.prune_receipts( next_static_file_receipt_num - next_receipt_num, start_block.saturating_sub(1), )?, @@ -641,7 +643,6 @@ where segment: StaticFileSegment::Receipts, }) } - Ordering::Equal => {} } Ok(static_file_producer) diff --git a/crates/stages/stages/src/stages/mod.rs b/crates/stages/stages/src/stages/mod.rs index 9af26247348e..1ecc8ac1e3fc 100644 --- a/crates/stages/stages/src/stages/mod.rs +++ b/crates/stages/stages/src/stages/mod.rs @@ -40,11 +40,12 @@ use utils::*; #[cfg(test)] mod tests { use super::*; - use crate::test_utils::TestStageDB; + use crate::test_utils::{StorageKind, TestStageDB}; use alloy_rlp::Decodable; use reth_db::{ - cursor::DbCursorRO, + cursor::{DbCursorRO, DbCursorRW}, mdbx::{cursor::Cursor, RW}, + table::Table, tables, test_utils::TempDatabase, transaction::{DbTx, DbTxMut}, @@ -53,16 +54,21 @@ mod tests { use reth_evm_ethereum::execute::EthExecutorProvider; use reth_exex::ExExManagerHandle; use reth_primitives::{ - address, hex_literal::hex, keccak256, Account, Bytecode, ChainSpecBuilder, PruneMode, - PruneModes, SealedBlock, StaticFileSegment, U256, + address, + hex_literal::hex, + keccak256, + stage::{PipelineTarget, StageCheckpoint, StageId}, + Account, BlockNumber, Bytecode, ChainSpecBuilder, PruneMode, PruneModes, SealedBlock, + StaticFileSegment, B256, U256, }; use reth_provider::{ - providers::StaticFileWriter, AccountExtReader, ProviderFactory, ReceiptProvider, - StorageReader, + providers::StaticFileWriter, AccountExtReader, BlockReader, DatabaseProviderFactory, + ProviderFactory, ProviderResult, ReceiptProvider, StageCheckpointWriter, + StaticFileProviderFactory, StorageReader, }; use reth_stages_api::{ExecInput, Stage}; - use reth_testing_utils::generators::{self, random_block}; - use std::sync::Arc; + use reth_testing_utils::generators::{self, random_block, random_block_range, random_receipt}; + use std::{io::Write, sync::Arc}; #[tokio::test] #[ignore] @@ -239,4 +245,252 @@ mod tests { // The one account is the miner check_pruning(test_db.factory.clone(), prune.clone(), 0, 1, 0).await; } + + /// It will generate `num_blocks`, push them to static files and set all stage checkpoints to + /// `num_blocks - 1`. + fn seed_data(num_blocks: usize) -> ProviderResult { + let db = TestStageDB::default(); + let mut rng = generators::rng(); + let genesis_hash = B256::ZERO; + let tip = (num_blocks - 1) as u64; + + let blocks = random_block_range(&mut rng, 0..=tip, genesis_hash, 2..3); + db.insert_blocks(blocks.iter(), StorageKind::Static)?; + + let mut receipts = Vec::new(); + let mut tx_num = 0u64; + for block in &blocks { + let mut block_receipts = Vec::with_capacity(block.body.len()); + for transaction in &block.body { + block_receipts.push((tx_num, random_receipt(&mut rng, transaction, Some(0)))); + tx_num += 1; + } + receipts.push((block.number, block_receipts)); + } + db.insert_receipts_by_block(receipts, StorageKind::Static)?; + + // simulate pipeline by setting all checkpoints to inserted height. + let provider_rw = db.factory.provider_rw()?; + for stage in StageId::ALL { + provider_rw.save_stage_checkpoint(stage, StageCheckpoint::new(tip))?; + } + provider_rw.commit()?; + + Ok(db) + } + + /// Simulates losing data to corruption and compare the check consistency result + /// against the expected one. + fn simulate_behind_checkpoint_corruption( + db: &TestStageDB, + prune_count: usize, + segment: StaticFileSegment, + is_full_node: bool, + expected: Option, + ) { + let static_file_provider = db.factory.static_file_provider(); + + // Simulate corruption by removing `prune_count` rows from the data file without updating + // its offset list and configuration. + { + let mut headers_writer = static_file_provider.latest_writer(segment).unwrap(); + let reader = headers_writer.inner().jar().open_data_reader().unwrap(); + let columns = headers_writer.inner().jar().columns(); + let data_file = headers_writer.inner().data_file(); + let last_offset = reader.reverse_offset(prune_count * columns).unwrap(); + data_file.get_mut().set_len(last_offset).unwrap(); + data_file.flush().unwrap(); + data_file.get_ref().sync_all().unwrap(); + } + + assert_eq!( + static_file_provider + .check_consistency(&db.factory.database_provider_ro().unwrap(), is_full_node,), + Ok(expected) + ); + } + + /// Saves a checkpoint with `checkpoint_block_number` and compare the check consistency result + /// against the expected one. + fn save_checkpoint_and_check( + db: &TestStageDB, + stage_id: StageId, + checkpoint_block_number: BlockNumber, + expected: Option, + ) { + let provider_rw = db.factory.provider_rw().unwrap(); + provider_rw + .save_stage_checkpoint(stage_id, StageCheckpoint::new(checkpoint_block_number)) + .unwrap(); + provider_rw.commit().unwrap(); + + assert_eq!( + db.factory + .static_file_provider() + .check_consistency(&db.factory.database_provider_ro().unwrap(), false,), + Ok(expected) + ); + } + + /// Inserts a dummy value at key and compare the check consistency result against the expected + /// one. + fn update_db_and_check>( + db: &TestStageDB, + key: u64, + expected: Option, + ) where + ::Value: Default, + { + let provider_rw = db.factory.provider_rw().unwrap(); + let mut cursor = provider_rw.tx_ref().cursor_write::().unwrap(); + cursor.insert(key, Default::default()).unwrap(); + provider_rw.commit().unwrap(); + + assert_eq!( + db.factory + .static_file_provider() + .check_consistency(&db.factory.database_provider_ro().unwrap(), false), + Ok(expected) + ); + } + + #[test] + fn test_consistency() { + let db = seed_data(90).unwrap(); + let db_provider = db.factory.database_provider_ro().unwrap(); + + assert_eq!( + db.factory.static_file_provider().check_consistency(&db_provider, false), + Ok(None) + ); + } + + #[test] + fn test_consistency_no_commit_prune() { + let db = seed_data(90).unwrap(); + let full_node = true; + let archive_node = !full_node; + + // Full node does not use receipts, therefore doesn't check for consistency on receipts + // segment + simulate_behind_checkpoint_corruption(&db, 1, StaticFileSegment::Receipts, full_node, None); + + // there are 2 to 3 transactions per block. however, if we lose one tx, we need to unwind to + // the previous block. + simulate_behind_checkpoint_corruption( + &db, + 1, + StaticFileSegment::Receipts, + archive_node, + Some(PipelineTarget::Unwind(88)), + ); + + simulate_behind_checkpoint_corruption( + &db, + 3, + StaticFileSegment::Headers, + archive_node, + Some(PipelineTarget::Unwind(86)), + ); + } + + #[test] + fn test_consistency_checkpoints() { + let db = seed_data(90).unwrap(); + + // When a checkpoint is behind, we delete data from static files. + let block = 87; + save_checkpoint_and_check(&db, StageId::Bodies, block, None); + assert_eq!( + db.factory + .static_file_provider() + .get_highest_static_file_block(StaticFileSegment::Transactions), + Some(block) + ); + assert_eq!( + db.factory + .static_file_provider() + .get_highest_static_file_tx(StaticFileSegment::Transactions), + db.factory.block_body_indices(block).unwrap().map(|b| b.last_tx_num()) + ); + + let block = 86; + save_checkpoint_and_check(&db, StageId::Execution, block, None); + assert_eq!( + db.factory + .static_file_provider() + .get_highest_static_file_block(StaticFileSegment::Receipts), + Some(block) + ); + assert_eq!( + db.factory + .static_file_provider() + .get_highest_static_file_tx(StaticFileSegment::Receipts), + db.factory.block_body_indices(block).unwrap().map(|b| b.last_tx_num()) + ); + + let block = 80; + save_checkpoint_and_check(&db, StageId::Headers, block, None); + assert_eq!( + db.factory + .static_file_provider() + .get_highest_static_file_block(StaticFileSegment::Headers), + Some(block) + ); + + // When a checkpoint is ahead, we request a pipeline unwind. + save_checkpoint_and_check(&db, StageId::Headers, 91, Some(PipelineTarget::Unwind(block))); + } + + #[test] + fn test_consistency_headers_gap() { + let db = seed_data(90).unwrap(); + let current = db + .factory + .static_file_provider() + .get_highest_static_file_block(StaticFileSegment::Headers) + .unwrap(); + + // Creates a gap of one header: static_file db + update_db_and_check::(&db, current + 2, Some(PipelineTarget::Unwind(89))); + + // Fill the gap, and ensure no unwind is necessary. + update_db_and_check::(&db, current + 1, None); + } + + #[test] + fn test_consistency_tx_gap() { + let db = seed_data(90).unwrap(); + let current = db + .factory + .static_file_provider() + .get_highest_static_file_tx(StaticFileSegment::Transactions) + .unwrap(); + + // Creates a gap of one transaction: static_file db + update_db_and_check::( + &db, + current + 2, + Some(PipelineTarget::Unwind(89)), + ); + + // Fill the gap, and ensure no unwind is necessary. + update_db_and_check::(&db, current + 1, None); + } + + #[test] + fn test_consistency_receipt_gap() { + let db = seed_data(90).unwrap(); + let current = db + .factory + .static_file_provider() + .get_highest_static_file_tx(StaticFileSegment::Receipts) + .unwrap(); + + // Creates a gap of one receipt: static_file db + update_db_and_check::(&db, current + 2, Some(PipelineTarget::Unwind(89))); + + // Fill the gap, and ensure no unwind is necessary. + update_db_and_check::(&db, current + 1, None); + } } diff --git a/crates/stages/stages/src/test_utils/test_db.rs b/crates/stages/stages/src/test_utils/test_db.rs index 234aace74413..c92ccc15bf32 100644 --- a/crates/stages/stages/src/test_utils/test_db.rs +++ b/crates/stages/stages/src/test_utils/test_db.rs @@ -16,7 +16,7 @@ use reth_primitives::{ StaticFileSegment, StorageEntry, TxHash, TxNumber, B256, MAINNET, U256, }; use reth_provider::{ - providers::{StaticFileProviderRWRefMut, StaticFileWriter}, + providers::{StaticFileProvider, StaticFileProviderRWRefMut, StaticFileWriter}, HistoryWriter, ProviderError, ProviderFactory, StaticFileProviderFactory, }; use reth_storage_errors::provider::ProviderResult; @@ -37,8 +37,11 @@ impl Default for TestStageDB { let (static_dir, static_dir_path) = create_test_static_files_dir(); Self { temp_static_files_dir: static_dir, - factory: ProviderFactory::new(create_test_rw_db(), MAINNET.clone(), static_dir_path) - .unwrap(), + factory: ProviderFactory::new( + create_test_rw_db(), + MAINNET.clone(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), } } } @@ -52,9 +55,8 @@ impl TestStageDB { factory: ProviderFactory::new( create_test_rw_db_with_path(path), MAINNET.clone(), - static_dir_path, - ) - .unwrap(), + StaticFileProvider::read_write(static_dir_path).unwrap(), + ), } } @@ -225,13 +227,17 @@ impl TestStageDB { let blocks = blocks.into_iter().collect::>(); { - let mut headers_writer = provider.latest_writer(StaticFileSegment::Headers)?; + let mut headers_writer = storage_kind + .is_static() + .then(|| provider.latest_writer(StaticFileSegment::Headers).unwrap()); blocks.iter().try_for_each(|block| { - Self::insert_header(Some(&mut headers_writer), &tx, &block.header, U256::ZERO) + Self::insert_header(headers_writer.as_mut(), &tx, &block.header, U256::ZERO) })?; - headers_writer.commit()?; + if let Some(mut writer) = headers_writer { + writer.commit()?; + } } { @@ -315,6 +321,42 @@ impl TestStageDB { }) } + /// Insert collection of ([TxNumber], [Receipt]) organized by respective block numbers into the + /// corresponding table or static file segment. + pub fn insert_receipts_by_block( + &self, + receipts: I, + storage_kind: StorageKind, + ) -> ProviderResult<()> + where + I: IntoIterator, + J: IntoIterator, + { + match storage_kind { + StorageKind::Database(_) => self.commit(|tx| { + receipts.into_iter().try_for_each(|(_, receipts)| { + for (tx_num, receipt) in receipts { + tx.put::(tx_num, receipt)?; + } + Ok(()) + }) + }), + StorageKind::Static => { + let provider = self.factory.static_file_provider(); + let mut writer = provider.latest_writer(StaticFileSegment::Receipts)?; + let res = receipts.into_iter().try_for_each(|(block_num, receipts)| { + writer.increment_block(StaticFileSegment::Receipts, block_num)?; + for (tx_num, receipt) in receipts { + writer.append_receipt(tx_num, receipt)?; + } + Ok(()) + }); + writer.commit_without_sync_all()?; + res + } + } + } + pub fn insert_transaction_senders(&self, transaction_senders: I) -> ProviderResult<()> where I: IntoIterator, diff --git a/crates/static-file-types/src/segment.rs b/crates/static-file-types/src/segment.rs index a69fab525c18..666d47258ad1 100644 --- a/crates/static-file-types/src/segment.rs +++ b/crates/static-file-types/src/segment.rs @@ -136,6 +136,11 @@ impl StaticFileSegment { pub const fn is_headers(&self) -> bool { matches!(self, Self::Headers) } + + /// Returns `true` if the segment is `StaticFileSegment::Receipts`. + pub const fn is_receipts(&self) -> bool { + matches!(self, Self::Receipts) + } } /// A segment header that contains information common to all segments. Used for storage. diff --git a/crates/storage/db-common/src/init.rs b/crates/storage/db-common/src/init.rs index 27728313b105..b193f04583d1 100644 --- a/crates/storage/db-common/src/init.rs +++ b/crates/storage/db-common/src/init.rs @@ -579,14 +579,11 @@ mod tests { init_genesis(factory.clone()).unwrap(); // Try to init db with a different genesis block - let genesis_hash = init_genesis( - ProviderFactory::new( - factory.into_db(), - MAINNET.clone(), - static_file_provider.path().into(), - ) - .unwrap(), - ); + let genesis_hash = init_genesis(ProviderFactory::new( + factory.into_db(), + MAINNET.clone(), + StaticFileProvider::read_write(static_file_provider.path()).unwrap(), + )); assert_eq!( genesis_hash.unwrap_err(), diff --git a/crates/storage/errors/src/provider.rs b/crates/storage/errors/src/provider.rs index aeecf64a7b41..5420b8f3c6cb 100644 --- a/crates/storage/errors/src/provider.rs +++ b/crates/storage/errors/src/provider.rs @@ -122,6 +122,9 @@ pub enum ProviderError { /// Trying to insert data from an unexpected block number. #[error("trying to append data to {0} as block #{1} but expected block #{2}")] UnexpectedStaticFileBlockNumber(StaticFileSegment, BlockNumber, BlockNumber), + /// Static File Provider was initialized as read-only. + #[error("cannot get a writer on a read-only environment.")] + ReadOnlyStaticFileAccess, /// Error encountered when the block number conversion from U256 to u64 causes an overflow. #[error("failed to convert block number U256 to u64: {0}")] BlockNumberOverflow(U256), diff --git a/crates/storage/nippy-jar/src/error.rs b/crates/storage/nippy-jar/src/error.rs index d59500842c7a..58e27a76b4c5 100644 --- a/crates/storage/nippy-jar/src/error.rs +++ b/crates/storage/nippy-jar/src/error.rs @@ -57,4 +57,6 @@ pub enum NippyJarError { InvalidPruning(u64, u64), #[error("jar has been frozen and cannot be modified.")] FrozenJar, + #[error("File is in an inconsistent state.")] + InconsistentState, } diff --git a/crates/storage/nippy-jar/src/lib.rs b/crates/storage/nippy-jar/src/lib.rs index 3258ebd10595..b1da4429355a 100644 --- a/crates/storage/nippy-jar/src/lib.rs +++ b/crates/storage/nippy-jar/src/lib.rs @@ -41,7 +41,7 @@ mod cursor; pub use cursor::NippyJarCursor; mod writer; -pub use writer::NippyJarWriter; +pub use writer::{ConsistencyFailStrategy, NippyJarWriter}; const NIPPY_JAR_VERSION: usize = 1; @@ -382,7 +382,7 @@ impl NippyJar { self.freeze_filters()?; // Creates the writer, data and offsets file - let mut writer = NippyJarWriter::new(self)?; + let mut writer = NippyJarWriter::new(self, ConsistencyFailStrategy::Heal)?; // Append rows to file while holding offsets in memory writer.append_rows(columns, total_rows)?; @@ -1114,7 +1114,7 @@ mod tests { assert!(initial_offset_size > 0); // Appends a third row - let mut writer = NippyJarWriter::new(nippy).unwrap(); + let mut writer = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); writer.append_column(Some(Ok(&col1[2]))).unwrap(); writer.append_column(Some(Ok(&col2[2]))).unwrap(); @@ -1145,7 +1145,7 @@ mod tests { // Writer will execute a consistency check and verify first that the offset list on disk // doesn't match the nippy.rows, and prune it. Then, it will prune the data file // accordingly as well. - let writer = NippyJarWriter::new(nippy).unwrap(); + let writer = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); assert_eq!(initial_rows, writer.rows()); assert_eq!( initial_offset_size, @@ -1171,7 +1171,7 @@ mod tests { // Appends a third row, so we have an offset list in memory, which is not flushed to disk, // while the data has been. - let mut writer = NippyJarWriter::new(nippy).unwrap(); + let mut writer = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); writer.append_column(Some(Ok(&col1[2]))).unwrap(); writer.append_column(Some(Ok(&col2[2]))).unwrap(); @@ -1194,7 +1194,7 @@ mod tests { // Writer will execute a consistency check and verify that the data file has more data than // it should, and resets it to the last offset of the list (on disk here) - let writer = NippyJarWriter::new(nippy).unwrap(); + let writer = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); assert_eq!(initial_rows, writer.rows()); assert_eq!( initial_data_size, @@ -1210,7 +1210,7 @@ mod tests { assert_eq!(nippy.max_row_size, 0); assert_eq!(nippy.rows, 0); - let mut writer = NippyJarWriter::new(nippy).unwrap(); + let mut writer = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); assert_eq!(writer.column(), 0); writer.append_column(Some(Ok(&col1[0]))).unwrap(); @@ -1245,7 +1245,7 @@ mod tests { assert_eq!(nippy.max_row_size, col1[0].len() + col2[0].len()); assert_eq!(nippy.rows, 1); - let mut writer = NippyJarWriter::new(nippy).unwrap(); + let mut writer = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); assert_eq!(writer.column(), 0); writer.append_column(Some(Ok(&col1[1]))).unwrap(); @@ -1276,7 +1276,7 @@ mod tests { fn prune_rows(num_columns: usize, file_path: &Path, col1: &[Vec], col2: &[Vec]) { let nippy = NippyJar::load_without_header(file_path).unwrap(); - let mut writer = NippyJarWriter::new(nippy).unwrap(); + let mut writer = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); // Appends a third row, so we have an offset list in memory, which is not flushed to disk writer.append_column(Some(Ok(&col1[2]))).unwrap(); @@ -1306,7 +1306,7 @@ mod tests { } // This should prune from the ondisk offset list and clear the jar. - let mut writer = NippyJarWriter::new(nippy).unwrap(); + let mut writer = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); writer.prune_rows(1).unwrap(); assert_eq!(writer.rows(), 0); assert_eq!(writer.max_row_size(), 0); @@ -1343,6 +1343,6 @@ mod tests { data_file.set_len(data_len - 32 * missing_offsets).unwrap(); // runs the consistency check. - let _ = NippyJarWriter::new(nippy).unwrap(); + let _ = NippyJarWriter::new(nippy, ConsistencyFailStrategy::Heal).unwrap(); } } diff --git a/crates/storage/nippy-jar/src/writer.rs b/crates/storage/nippy-jar/src/writer.rs index 8bd47b1b4113..5c405856d132 100644 --- a/crates/storage/nippy-jar/src/writer.rs +++ b/crates/storage/nippy-jar/src/writer.rs @@ -43,7 +43,13 @@ pub struct NippyJarWriter { impl NippyJarWriter { /// Creates a [`NippyJarWriter`] from [`NippyJar`]. - pub fn new(jar: NippyJar) -> Result { + /// + /// If `read_only` is set to `true`, any inconsistency issue won't be healed, and will return + /// [NippyJarError::InconsistentState] instead. + pub fn new( + jar: NippyJar, + check_mode: ConsistencyFailStrategy, + ) -> Result { let (data_file, offsets_file, is_created) = Self::create_or_open_files(jar.data_path(), &jar.offsets_path())?; @@ -63,8 +69,10 @@ impl NippyJarWriter { // If we are opening a previously created jar, we need to check its consistency, and make // changes if necessary. if !is_created { - writer.check_consistency_and_heal()?; - writer.commit()?; + writer.ensure_file_consistency(check_mode)?; + if check_mode.should_heal() { + writer.commit()?; + } } Ok(writer) @@ -118,13 +126,17 @@ impl NippyJarWriter { Ok((data_file, offsets_file, is_created)) } - /// Performs consistency checks on the [`NippyJar`] file and acts upon any issues: + /// Performs consistency checks on the [`NippyJar`] file and might self-heal or throw an error + /// according to [ConsistencyFailStrategy]. /// * Is the offsets file size expected? /// * Is the data file size expected? /// /// This is based on the assumption that [`NippyJar`] configuration is **always** the last one /// to be updated when something is written, as by the `commit()` function shows. - fn check_consistency_and_heal(&mut self) -> Result<(), NippyJarError> { + pub fn ensure_file_consistency( + &mut self, + check_mode: ConsistencyFailStrategy, + ) -> Result<(), NippyJarError> { let reader = self.jar.open_data_reader()?; // When an offset size is smaller than the initial (8), we are dealing with immutable @@ -138,6 +150,12 @@ impl NippyJarWriter { OFFSET_SIZE_BYTES as usize) as u64; // expected size of the data file let actual_offsets_file_size = self.offsets_file.get_ref().metadata()?.len(); + if check_mode.should_err() && + expected_offsets_file_size.cmp(&actual_offsets_file_size) != Ordering::Equal + { + return Err(NippyJarError::InconsistentState) + } + // Offsets configuration wasn't properly committed match expected_offsets_file_size.cmp(&actual_offsets_file_size) { Ordering::Less => { @@ -165,6 +183,10 @@ impl NippyJarWriter { let last_offset = reader.reverse_offset(0)?; let data_file_len = self.data_file.get_ref().metadata()?.len(); + if check_mode.should_err() && last_offset.cmp(&data_file_len) != Ordering::Equal { + return Err(NippyJarError::InconsistentState) + } + // Offset list wasn't properly committed match last_offset.cmp(&data_file_len) { Ordering::Less => { @@ -177,7 +199,8 @@ impl NippyJarWriter { // find the matching one. for index in 0..reader.offsets_count()? { let offset = reader.reverse_offset(index + 1)?; - if offset == data_file_len { + // It would only be equal if the previous row was fully pruned. + if offset <= data_file_len { let new_len = self .offsets_file .get_ref() @@ -190,7 +213,7 @@ impl NippyJarWriter { // Since we decrease the offset list, we need to check the consistency of // `self.jar.rows` again - self.check_consistency_and_heal()?; + self.ensure_file_consistency(ConsistencyFailStrategy::Heal)?; break } } @@ -479,4 +502,35 @@ impl NippyJarWriter { pub fn data_path(&self) -> &Path { self.jar.data_path() } + + #[cfg(any(test, feature = "test-utils"))] + pub fn data_file(&mut self) -> &mut BufWriter { + &mut self.data_file + } + + #[cfg(any(test, feature = "test-utils"))] + pub const fn jar(&self) -> &NippyJar { + &self.jar + } +} + +/// Strategy on encountering an inconsistent state when creating a [NippyJarWriter]. +#[derive(Debug, Copy, Clone)] +pub enum ConsistencyFailStrategy { + /// Writer should heal. + Heal, + /// Writer should throw an error. + ThrowError, +} + +impl ConsistencyFailStrategy { + /// Whether writer should heal. + const fn should_heal(&self) -> bool { + matches!(self, Self::Heal) + } + + /// Whether writer should throw an error. + const fn should_err(&self) -> bool { + matches!(self, Self::ThrowError) + } } diff --git a/crates/storage/provider/src/lib.rs b/crates/storage/provider/src/lib.rs index 021372bb647e..17c6af875818 100644 --- a/crates/storage/provider/src/lib.rs +++ b/crates/storage/provider/src/lib.rs @@ -21,7 +21,7 @@ pub mod providers; pub use providers::{ DatabaseProvider, DatabaseProviderRO, DatabaseProviderRW, HistoricalStateProvider, HistoricalStateProviderRef, LatestStateProvider, LatestStateProviderRef, ProviderFactory, - StaticFileWriter, + StaticFileAccess, StaticFileWriter, }; #[cfg(any(test, feature = "test-utils"))] diff --git a/crates/storage/provider/src/providers/database/mod.rs b/crates/storage/provider/src/providers/database/mod.rs index 8b83cc26a44a..08148e95e1f8 100644 --- a/crates/storage/provider/src/providers/database/mod.rs +++ b/crates/storage/provider/src/providers/database/mod.rs @@ -25,7 +25,7 @@ use reth_storage_errors::provider::ProviderResult; use revm::primitives::{BlockEnv, CfgEnvWithHandlerCfg}; use std::{ ops::{RangeBounds, RangeInclusive}, - path::{Path, PathBuf}, + path::Path, sync::Arc, }; use tracing::trace; @@ -53,13 +53,9 @@ impl ProviderFactory { pub fn new( db: DB, chain_spec: Arc, - static_files_path: PathBuf, - ) -> ProviderResult { - Ok(Self { - db: Arc::new(db), - chain_spec, - static_file_provider: StaticFileProvider::new(static_files_path)?, - }) + static_file_provider: StaticFileProvider, + ) -> Self { + Self { db: Arc::new(db), chain_spec, static_file_provider } } /// Enables metrics on the static file provider. @@ -87,12 +83,12 @@ impl ProviderFactory { path: P, chain_spec: Arc, args: DatabaseArguments, - static_files_path: PathBuf, + static_file_provider: StaticFileProvider, ) -> RethResult { Ok(Self { db: Arc::new(init_db(path, args).map_err(RethError::msg)?), chain_spec, - static_file_provider: StaticFileProvider::new(static_files_path)?, + static_file_provider, }) } } @@ -588,8 +584,10 @@ impl Clone for ProviderFactory { mod tests { use super::*; use crate::{ - providers::StaticFileWriter, test_utils::create_test_provider_factory, BlockHashReader, - BlockNumReader, BlockWriter, HeaderSyncGapProvider, HeaderSyncMode, TransactionsProvider, + providers::{StaticFileProvider, StaticFileWriter}, + test_utils::create_test_provider_factory, + BlockHashReader, BlockNumReader, BlockWriter, HeaderSyncGapProvider, HeaderSyncMode, + TransactionsProvider, }; use alloy_rlp::Decodable; use assert_matches::assert_matches; @@ -645,7 +643,7 @@ mod tests { tempfile::TempDir::new().expect(ERROR_TEMPDIR).into_path(), Arc::new(chain_spec), DatabaseArguments::new(Default::default()), - static_dir_path, + StaticFileProvider::read_write(static_dir_path).unwrap(), ) .unwrap(); diff --git a/crates/storage/provider/src/providers/mod.rs b/crates/storage/provider/src/providers/mod.rs index 9bbdcb26f1d1..9b72b3bd9740 100644 --- a/crates/storage/provider/src/providers/mod.rs +++ b/crates/storage/provider/src/providers/mod.rs @@ -40,8 +40,8 @@ pub use database::*; mod static_file; pub use static_file::{ - StaticFileJarProvider, StaticFileProvider, StaticFileProviderRW, StaticFileProviderRWRefMut, - StaticFileWriter, + StaticFileAccess, StaticFileJarProvider, StaticFileProvider, StaticFileProviderRW, + StaticFileProviderRWRefMut, StaticFileWriter, }; mod state; diff --git a/crates/storage/provider/src/providers/static_file/manager.rs b/crates/storage/provider/src/providers/static_file/manager.rs index fd877a4470eb..0052868ad9c0 100644 --- a/crates/storage/provider/src/providers/static_file/manager.rs +++ b/crates/storage/provider/src/providers/static_file/manager.rs @@ -3,22 +3,25 @@ use super::{ StaticFileProviderRWRefMut, BLOCKS_PER_STATIC_FILE, }; use crate::{ - to_range, BlockHashReader, BlockNumReader, BlockReader, BlockSource, HeaderProvider, - ReceiptProvider, RequestsProvider, StatsReader, TransactionVariant, TransactionsProvider, - TransactionsProviderExt, WithdrawalsProvider, + to_range, BlockHashReader, BlockNumReader, BlockReader, BlockSource, DatabaseProvider, + HeaderProvider, ReceiptProvider, RequestsProvider, StageCheckpointReader, StatsReader, + TransactionVariant, TransactionsProvider, TransactionsProviderExt, WithdrawalsProvider, }; use dashmap::{mapref::entry::Entry as DashMapEntry, DashMap}; use parking_lot::RwLock; use reth_db::{ codecs::CompactU256, + cursor::DbCursorRO, models::StoredBlockBodyIndices, static_file::{iter_static_files, HeaderMask, ReceiptMask, StaticFileCursor, TransactionMask}, table::Table, tables, + transaction::DbTx, }; use reth_nippy_jar::NippyJar; use reth_primitives::{ keccak256, + stage::{PipelineTarget, StageId}, static_file::{find_fixed_range, HighestStaticFiles, SegmentHeader, SegmentRangeInclusive}, Address, Block, BlockHash, BlockHashOrNumber, BlockNumber, BlockWithSenders, ChainInfo, Header, Receipt, SealedBlock, SealedBlockWithSenders, SealedHeader, StaticFileSegment, TransactionMeta, @@ -32,24 +35,52 @@ use std::{ path::{Path, PathBuf}, sync::{mpsc, Arc}, }; -use tracing::warn; +use strum::IntoEnumIterator; +use tracing::{info, warn}; /// Alias type for a map that can be queried for block ranges from a transaction /// segment respectively. It uses `TxNumber` to represent the transaction end of a static file /// range. type SegmentRanges = HashMap>; +/// Access mode on a static file provider. RO/RW. +#[derive(Debug, Default, PartialEq, Eq)] +pub enum StaticFileAccess { + /// Read-only access. + #[default] + RO, + /// Read-write access. + RW, +} + +impl StaticFileAccess { + /// Returns `true` if read-only access. + pub const fn is_read_only(&self) -> bool { + matches!(self, Self::RO) + } +} + /// [`StaticFileProvider`] manages all existing [`StaticFileJarProvider`]. #[derive(Debug, Default, Clone)] pub struct StaticFileProvider(pub(crate) Arc); impl StaticFileProvider { /// Creates a new [`StaticFileProvider`]. - pub fn new(path: impl AsRef) -> ProviderResult { - let provider = Self(Arc::new(StaticFileProviderInner::new(path)?)); + fn new(path: impl AsRef, access: StaticFileAccess) -> ProviderResult { + let provider = Self(Arc::new(StaticFileProviderInner::new(path, access)?)); provider.initialize_index()?; Ok(provider) } + + /// Creates a new [`StaticFileProvider`] with read-only access. + pub fn read_only(path: impl AsRef) -> ProviderResult { + Self::new(path, StaticFileAccess::RO) + } + + /// Creates a new [`StaticFileProvider`] with read-write access. + pub fn read_write(path: impl AsRef) -> ProviderResult { + Self::new(path, StaticFileAccess::RW) + } } impl Deref for StaticFileProvider { @@ -78,11 +109,13 @@ pub struct StaticFileProviderInner { /// Maintains a map of StaticFile writers for each [`StaticFileSegment`] writers: DashMap, metrics: Option>, + /// Access rights of the provider. + access: StaticFileAccess, } impl StaticFileProviderInner { /// Creates a new [`StaticFileProviderInner`]. - fn new(path: impl AsRef) -> ProviderResult { + fn new(path: impl AsRef, access: StaticFileAccess) -> ProviderResult { let provider = Self { map: Default::default(), writers: Default::default(), @@ -91,10 +124,15 @@ impl StaticFileProviderInner { path: path.as_ref().to_path_buf(), load_filters: false, metrics: None, + access, }; Ok(provider) } + + pub const fn is_read_only(&self) -> bool { + self.access.is_read_only() + } } impl StaticFileProvider { @@ -448,6 +486,209 @@ impl StaticFileProvider { Ok(()) } + /// Ensures that any broken invariants which cannot be healed on the spot return a pipeline + /// target to unwind to. + /// + /// Two types of consistency checks are done for: + /// + /// 1) When a static file fails to commit but the underlying data was changed. + /// 2) When a static file was committed, but the required database transaction was not. + /// + /// For 1) it can self-heal if `self.access.is_read_only()` is set to `false`. Otherwise, it + /// will return an error. + /// For 2) the invariants below are checked, and if broken, might require a pipeline unwind + /// to heal. + /// + /// For each static file segment: + /// * the corresponding database table should overlap or have continuity in their keys + /// ([TxNumber] or [BlockNumber]). + /// * its highest block should match the stage checkpoint block number if it's equal or higher + /// than the corresponding database table last entry. + /// + /// Returns a [`Option`] of [`PipelineTarget::Unwind`] if any healing is further required. + /// + /// WARNING: No static file writer should be held before calling this function, otherwise it + /// will deadlock. + #[allow(clippy::while_let_loop)] + pub fn check_consistency( + &self, + provider: &DatabaseProvider, + has_receipt_pruning: bool, + ) -> ProviderResult> { + let mut unwind_target: Option = None; + let mut update_unwind_target = |new_target: BlockNumber| { + if let Some(target) = unwind_target.as_mut() { + *target = (*target).min(new_target); + } else { + unwind_target = Some(new_target); + } + }; + + for segment in StaticFileSegment::iter() { + if has_receipt_pruning && segment.is_receipts() { + // Pruned nodes (including full node) do not store receipts as static files. + continue + } + + let initial_highest_block = self.get_highest_static_file_block(segment); + + // File consistency is broken if: + // + // * appending data was interrupted before a config commit, then data file will be + // truncated according to the config. + // + // * pruning data was interrupted before a config commit, then we have deleted data that + // we are expected to still have. We need to check the Database and unwind everything + // accordingly. + self.ensure_file_consistency(segment)?; + + // Only applies to block-based static files. (Headers) + // + // The updated `highest_block` may have decreased if we healed from a pruning + // interruption. + let mut highest_block = self.get_highest_static_file_block(segment); + if initial_highest_block != highest_block { + update_unwind_target(highest_block.unwrap_or_default()); + } + + // Only applies to transaction-based static files. (Receipts & Transactions) + // + // Make sure the last transaction matches the last block from its indices, since a heal + // from a pruning interruption might have decreased the number of transactions without + // being able to update the last block of the static file segment. + let highest_tx = self.get_highest_static_file_tx(segment); + if let Some(highest_tx) = highest_tx { + let mut last_block = highest_block.unwrap_or_default(); + loop { + if let Some(indices) = provider.block_body_indices(last_block)? { + if indices.last_tx_num() <= highest_tx { + break + } + } else { + // If the block body indices can not be found, then it means that static + // files is ahead of database, and the `ensure_invariants` check will fix + // it by comparing with stage checkpoints. + break + } + if last_block == 0 { + break + } + last_block -= 1; + + highest_block = Some(last_block); + update_unwind_target(last_block); + } + } + + if let Some(unwind) = match segment { + StaticFileSegment::Headers => self.ensure_invariants::<_, tables::Headers>( + provider, + segment, + highest_block, + highest_block, + )?, + StaticFileSegment::Transactions => self + .ensure_invariants::<_, tables::Transactions>( + provider, + segment, + highest_tx, + highest_block, + )?, + StaticFileSegment::Receipts => self.ensure_invariants::<_, tables::Receipts>( + provider, + segment, + highest_tx, + highest_block, + )?, + } { + update_unwind_target(unwind); + } + } + + Ok(unwind_target.map(PipelineTarget::Unwind)) + } + + /// Check invariants for each corresponding table and static file segment: + /// + /// * the corresponding database table should overlap or have continuity in their keys + /// ([TxNumber] or [BlockNumber]). + /// * its highest block should match the stage checkpoint block number if it's equal or higher + /// than the corresponding database table last entry. + /// * If the checkpoint block is higher, then request a pipeline unwind to the static file + /// block. + /// * If the checkpoint block is lower, then heal by removing rows from the static file. + fn ensure_invariants>( + &self, + provider: &DatabaseProvider, + segment: StaticFileSegment, + highest_static_file_entry: Option, + highest_static_file_block: Option, + ) -> ProviderResult> { + let highest_static_file_entry = highest_static_file_entry.unwrap_or_default(); + let highest_static_file_block = highest_static_file_block.unwrap_or_default(); + let mut db_cursor = provider.tx_ref().cursor_read::()?; + + if let Some((db_first_entry, _)) = db_cursor.first()? { + // If there is a gap between the entry found in static file and + // database, then we have most likely lost static file data and need to unwind so we can + // load it again + if !(db_first_entry <= highest_static_file_entry || + highest_static_file_entry + 1 == db_first_entry) + { + return Ok(Some(highest_static_file_block)) + } + + if let Some((db_last_entry, _)) = db_cursor.last()? { + if db_last_entry > highest_static_file_entry { + return Ok(None) + } + } + } + + // If static file entry is ahead of the database entries, then ensure the checkpoint block + // number matches. + let checkpoint_block_number = provider + .get_stage_checkpoint(match segment { + StaticFileSegment::Headers => StageId::Headers, + StaticFileSegment::Transactions => StageId::Bodies, + StaticFileSegment::Receipts => StageId::Execution, + })? + .unwrap_or_default() + .block_number; + + // If the checkpoint is ahead, then we lost static file data. May be data corruption. + if checkpoint_block_number > highest_static_file_block { + return Ok(Some(highest_static_file_block)); + } + + // If the checkpoint is behind, then we failed to do a database commit **but committed** to + // static files on executing a stage, or the reverse on unwinding a stage. + // All we need to do is to prune the extra static file rows. + if checkpoint_block_number < highest_static_file_block { + info!( + target: "reth::providers", + ?segment, + from = highest_static_file_block, + to = checkpoint_block_number, + "Unwinding static file segment." + ); + let mut writer = self.latest_writer(segment)?; + if segment.is_headers() { + writer.prune_headers(highest_static_file_block - checkpoint_block_number)?; + } else if let Some(block) = provider.block_body_indices(checkpoint_block_number)? { + let number = highest_static_file_entry - block.last_tx_num(); + if segment.is_receipts() { + writer.prune_receipts(number, checkpoint_block_number)?; + } else { + writer.prune_transactions(number, checkpoint_block_number)?; + } + } + writer.commit()?; + } + + Ok(None) + } + /// Gets the highest static file block if it exists for a static file segment. pub fn get_highest_static_file_block(&self, segment: StaticFileSegment) -> Option { self.static_files_max_block.read().get(&segment).copied() @@ -717,6 +958,9 @@ pub trait StaticFileWriter { /// Commits all changes of all [`StaticFileProviderRW`] of all [`StaticFileSegment`]. fn commit(&self) -> ProviderResult<()>; + + /// Checks consistency of the segment latest file and heals if possible. + fn ensure_file_consistency(&self, segment: StaticFileSegment) -> ProviderResult<()>; } impl StaticFileWriter for StaticFileProvider { @@ -725,6 +969,10 @@ impl StaticFileWriter for StaticFileProvider { block: BlockNumber, segment: StaticFileSegment, ) -> ProviderResult> { + if self.access.is_read_only() { + return Err(ProviderError::ReadOnlyStaticFileAccess) + } + tracing::trace!(target: "providers::static_file", ?block, ?segment, "Getting static file writer."); Ok(match self.writers.entry(segment) { DashMapEntry::Occupied(entry) => entry.into_ref(), @@ -753,6 +1001,28 @@ impl StaticFileWriter for StaticFileProvider { } Ok(()) } + + fn ensure_file_consistency(&self, segment: StaticFileSegment) -> ProviderResult<()> { + match self.access { + StaticFileAccess::RO => { + let latest_block = self.get_highest_static_file_block(segment).unwrap_or_default(); + + let mut writer = StaticFileProviderRW::new( + segment, + latest_block, + Arc::downgrade(&self.0), + self.metrics.clone(), + )?; + + writer.ensure_file_consistency(self.access.is_read_only())?; + } + StaticFileAccess::RW => { + self.latest_writer(segment)?.ensure_file_consistency(self.access.is_read_only())?; + } + } + + Ok(()) + } } impl HeaderProvider for StaticFileProvider { @@ -771,8 +1041,15 @@ impl HeaderProvider for StaticFileProvider { } fn header_by_number(&self, num: BlockNumber) -> ProviderResult> { - self.get_segment_provider_from_block(StaticFileSegment::Headers, num, None)? - .header_by_number(num) + self.get_segment_provider_from_block(StaticFileSegment::Headers, num, None) + .and_then(|provider| provider.header_by_number(num)) + .or_else(|err| { + if let ProviderError::MissingStaticFileBlock(_, _) = err { + Ok(None) + } else { + Err(err) + } + }) } fn header_td(&self, block_hash: &BlockHash) -> ProviderResult> { @@ -785,8 +1062,15 @@ impl HeaderProvider for StaticFileProvider { } fn header_td_by_number(&self, num: BlockNumber) -> ProviderResult> { - self.get_segment_provider_from_block(StaticFileSegment::Headers, num, None)? - .header_td_by_number(num) + self.get_segment_provider_from_block(StaticFileSegment::Headers, num, None) + .and_then(|provider| provider.header_td_by_number(num)) + .or_else(|err| { + if let ProviderError::MissingStaticFileBlock(_, _) = err { + Ok(None) + } else { + Err(err) + } + }) } fn headers_range(&self, range: impl RangeBounds) -> ProviderResult> { @@ -799,8 +1083,15 @@ impl HeaderProvider for StaticFileProvider { } fn sealed_header(&self, num: BlockNumber) -> ProviderResult> { - self.get_segment_provider_from_block(StaticFileSegment::Headers, num, None)? - .sealed_header(num) + self.get_segment_provider_from_block(StaticFileSegment::Headers, num, None) + .and_then(|provider| provider.sealed_header(num)) + .or_else(|err| { + if let ProviderError::MissingStaticFileBlock(_, _) = err { + Ok(None) + } else { + Err(err) + } + }) } fn sealed_headers_while( @@ -842,8 +1133,15 @@ impl BlockHashReader for StaticFileProvider { impl ReceiptProvider for StaticFileProvider { fn receipt(&self, num: TxNumber) -> ProviderResult> { - self.get_segment_provider_from_transaction(StaticFileSegment::Receipts, num, None)? - .receipt(num) + self.get_segment_provider_from_transaction(StaticFileSegment::Receipts, num, None) + .and_then(|provider| provider.receipt(num)) + .or_else(|err| { + if let ProviderError::MissingStaticFileTx(_, _) = err { + Ok(None) + } else { + Err(err) + } + }) } fn receipt_by_hash(&self, hash: TxHash) -> ProviderResult> { @@ -947,16 +1245,30 @@ impl TransactionsProvider for StaticFileProvider { } fn transaction_by_id(&self, num: TxNumber) -> ProviderResult> { - self.get_segment_provider_from_transaction(StaticFileSegment::Transactions, num, None)? - .transaction_by_id(num) + self.get_segment_provider_from_transaction(StaticFileSegment::Transactions, num, None) + .and_then(|provider| provider.transaction_by_id(num)) + .or_else(|err| { + if let ProviderError::MissingStaticFileTx(_, _) = err { + Ok(None) + } else { + Err(err) + } + }) } fn transaction_by_id_no_hash( &self, num: TxNumber, ) -> ProviderResult> { - self.get_segment_provider_from_transaction(StaticFileSegment::Transactions, num, None)? - .transaction_by_id_no_hash(num) + self.get_segment_provider_from_transaction(StaticFileSegment::Transactions, num, None) + .and_then(|provider| provider.transaction_by_id_no_hash(num)) + .or_else(|err| { + if let ProviderError::MissingStaticFileTx(_, _) = err { + Ok(None) + } else { + Err(err) + } + }) } fn transaction_by_hash(&self, hash: TxHash) -> ProviderResult> { diff --git a/crates/storage/provider/src/providers/static_file/mod.rs b/crates/storage/provider/src/providers/static_file/mod.rs index 4ea8c525d2e8..97fc649a46cc 100644 --- a/crates/storage/provider/src/providers/static_file/mod.rs +++ b/crates/storage/provider/src/providers/static_file/mod.rs @@ -1,5 +1,5 @@ mod manager; -pub use manager::{StaticFileProvider, StaticFileWriter}; +pub use manager::{StaticFileAccess, StaticFileProvider, StaticFileWriter}; mod jar; pub use jar::StaticFileJarProvider; @@ -150,7 +150,8 @@ mod tests { // Use providers to query Header data and compare if it matches { let db_provider = factory.provider().unwrap(); - let manager = StaticFileProvider::new(static_files_path.path()).unwrap().with_filters(); + let manager = + StaticFileProvider::read_write(static_files_path.path()).unwrap().with_filters(); let jar_provider = manager .get_segment_provider_from_block(StaticFileSegment::Headers, 0, Some(&static_file)) .unwrap(); diff --git a/crates/storage/provider/src/providers/static_file/writer.rs b/crates/storage/provider/src/providers/static_file/writer.rs index 7289da860595..3b88812da7f3 100644 --- a/crates/storage/provider/src/providers/static_file/writer.rs +++ b/crates/storage/provider/src/providers/static_file/writer.rs @@ -6,7 +6,7 @@ use super::{ use dashmap::mapref::one::RefMut; use reth_codecs::Compact; use reth_db::codecs::CompactU256; -use reth_nippy_jar::{NippyJar, NippyJarError, NippyJarWriter}; +use reth_nippy_jar::{ConsistencyFailStrategy, NippyJar, NippyJarError, NippyJarWriter}; use reth_primitives::{ static_file::{find_fixed_range, SegmentHeader, SegmentRangeInclusive}, BlockHash, BlockNumber, Header, Receipt, StaticFileSegment, TransactionSignedNoHash, TxNumber, @@ -90,7 +90,14 @@ impl StaticFileProviderRW { Err(err) => return Err(err), }; - let result = match NippyJarWriter::new(jar) { + let reader = Self::upgrade_provider_to_strong_reference(&reader); + let access = if reader.is_read_only() { + ConsistencyFailStrategy::ThrowError + } else { + ConsistencyFailStrategy::Heal + }; + + let result = match NippyJarWriter::new(jar, access) { Ok(writer) => Ok((writer, path)), Err(NippyJarError::FrozenJar) => { // This static file has been frozen, so we should @@ -110,6 +117,54 @@ impl StaticFileProviderRW { Ok(result) } + /// Checks the consistency of the file and heals it if necessary and `read_only` is set to + /// false. If the check fails, it will return an error. + /// + /// If healing does happen, it will update the end range on the [SegmentHeader]. However, for + /// transaction based segments, the block end range has to be found and healed externally. + /// + /// Check [NippyJarWriter::ensure_file_consistency] for more on healing. + pub fn ensure_file_consistency(&mut self, read_only: bool) -> ProviderResult<()> { + let inconsistent_error = || { + ProviderError::NippyJar( + "Inconsistent state found. Restart the node to heal.".to_string(), + ) + }; + + let check_mode = if read_only { + ConsistencyFailStrategy::ThrowError + } else { + ConsistencyFailStrategy::Heal + }; + + self.writer.ensure_file_consistency(check_mode).map_err(|error| { + if matches!(error, NippyJarError::InconsistentState) { + return inconsistent_error() + } + ProviderError::NippyJar(error.to_string()) + })?; + + // If we have lost rows (in this run or previous), we need to update the [SegmentHeader]. + let expected_rows = if self.user_header().segment().is_headers() { + self.user_header().block_len().unwrap_or_default() + } else { + self.user_header().tx_len().unwrap_or_default() + }; + let pruned_rows = expected_rows - self.writer.rows() as u64; + if pruned_rows > 0 { + if read_only { + return Err(inconsistent_error()) + } + self.user_header_mut().prune(pruned_rows); + } + + self.writer.commit().map_err(|error| ProviderError::NippyJar(error.to_string()))?; + + // Updates the [SnapshotProvider] manager + self.update_index()?; + Ok(()) + } + /// Commits configuration changes to disk and updates the reader index with the new changes. pub fn commit(&mut self) -> ProviderResult<()> { let start = Instant::now(); @@ -285,10 +340,11 @@ impl StaticFileProviderRW { fn truncate( &mut self, segment: StaticFileSegment, - mut num_rows: u64, + num_rows: u64, last_block: Option, ) -> ProviderResult<()> { - while num_rows > 0 { + let mut remaining_rows = num_rows; + while remaining_rows > 0 { let len = match segment { StaticFileSegment::Headers => { self.writer.user_header().block_len().unwrap_or_default() @@ -298,26 +354,13 @@ impl StaticFileProviderRW { } }; - if num_rows >= len { + if remaining_rows >= len { // If there's more rows to delete than this static file contains, then just // delete the whole file and go to the next static file - let previous_snap = self.data_path.clone(); let block_start = self.writer.user_header().expected_block_start(); if block_start != 0 { - let (writer, data_path) = Self::open( - segment, - self.writer.user_header().expected_block_start() - 1, - self.reader.clone(), - self.metrics.clone(), - )?; - self.writer = writer; - self.data_path = data_path; - - NippyJar::::load(&previous_snap) - .map_err(|e| ProviderError::NippyJar(e.to_string()))? - .delete() - .map_err(|e| ProviderError::NippyJar(e.to_string()))?; + self.delete_current_and_open_previous()?; } else { // Update `SegmentHeader` self.writer.user_header_mut().prune(len); @@ -327,23 +370,33 @@ impl StaticFileProviderRW { break } - num_rows -= len; + remaining_rows -= len; } else { // Update `SegmentHeader` - self.writer.user_header_mut().prune(num_rows); + self.writer.user_header_mut().prune(remaining_rows); // Truncate data self.writer - .prune_rows(num_rows as usize) + .prune_rows(remaining_rows as usize) .map_err(|e| ProviderError::NippyJar(e.to_string()))?; - num_rows = 0; + remaining_rows = 0; } } // Only Transactions and Receipts if let Some(last_block) = last_block { - let header = self.writer.user_header_mut(); - header.set_block_range(header.expected_block_start(), last_block); + let mut expected_block_start = self.writer.user_header().expected_block_start(); + + if num_rows == 0 { + // Edge case for when we are unwinding a chain of empty blocks that goes across + // files, and therefore, the only reference point to know which file + // we are supposed to be at is `last_block`. + while last_block < expected_block_start { + self.delete_current_and_open_previous()?; + expected_block_start = self.writer.user_header().expected_block_start(); + } + } + self.writer.user_header_mut().set_block_range(expected_block_start, last_block); } // Commits new changes to disk. @@ -352,6 +405,25 @@ impl StaticFileProviderRW { Ok(()) } + /// Delete the current static file, and replace this provider writer with the previous static + /// file. + fn delete_current_and_open_previous(&mut self) -> Result<(), ProviderError> { + let current_path = self.data_path.clone(); + let (previous_writer, data_path) = Self::open( + self.user_header().segment(), + self.writer.user_header().expected_block_start() - 1, + self.reader.clone(), + self.metrics.clone(), + )?; + self.writer = previous_writer; + self.data_path = data_path; + NippyJar::::load(¤t_path) + .map_err(|e| ProviderError::NippyJar(e.to_string()))? + .delete() + .map_err(|e| ProviderError::NippyJar(e.to_string()))?; + Ok(()) + } + /// Appends column to static file. fn append_column(&mut self, column: T) -> ProviderResult<()> { self.buf.clear(); @@ -613,16 +685,26 @@ impl StaticFileProviderRW { provider.upgrade().map(StaticFileProvider).expect("StaticFileProvider is dropped") } - #[cfg(any(test, feature = "test-utils"))] + /// Helper function to access [`SegmentHeader`]. + pub const fn user_header(&self) -> &SegmentHeader { + self.writer.user_header() + } + + /// Helper function to access a mutable reference to [`SegmentHeader`]. + pub fn user_header_mut(&mut self) -> &mut SegmentHeader { + self.writer.user_header_mut() + } + /// Helper function to override block range for testing. + #[cfg(any(test, feature = "test-utils"))] pub fn set_block_range(&mut self, block_range: std::ops::RangeInclusive) { self.writer.user_header_mut().set_block_range(*block_range.start(), *block_range.end()) } + /// Helper function to override block range for testing. #[cfg(any(test, feature = "test-utils"))] - /// Helper function to access [`SegmentHeader`]. - pub const fn user_header(&self) -> &SegmentHeader { - self.writer.user_header() + pub fn inner(&mut self) -> &mut NippyJarWriter { + &mut self.writer } } diff --git a/crates/storage/provider/src/test_utils/mod.rs b/crates/storage/provider/src/test_utils/mod.rs index 2f5462309442..6f5ecd526783 100644 --- a/crates/storage/provider/src/test_utils/mod.rs +++ b/crates/storage/provider/src/test_utils/mod.rs @@ -1,4 +1,4 @@ -use crate::ProviderFactory; +use crate::{providers::StaticFileProvider, ProviderFactory}; use reth_db::{ test_utils::{create_test_rw_db, create_test_static_files_dir, TempDatabase}, DatabaseEnv, @@ -26,6 +26,9 @@ pub fn create_test_provider_factory_with_chain_spec( ) -> ProviderFactory>> { let (static_dir, _) = create_test_static_files_dir(); let db = create_test_rw_db(); - ProviderFactory::new(db, chain_spec, static_dir.into_path()) - .expect("create provider factory with static_files") + ProviderFactory::new( + db, + chain_spec, + StaticFileProvider::read_write(static_dir.into_path()).expect("static file provider"), + ) } diff --git a/examples/db-access/src/main.rs b/examples/db-access/src/main.rs index c076b76dcf24..c43aec47ce0c 100644 --- a/examples/db-access/src/main.rs +++ b/examples/db-access/src/main.rs @@ -1,8 +1,8 @@ use reth_db::open_db_read_only; use reth_primitives::{Address, ChainSpecBuilder, B256}; use reth_provider::{ - AccountReader, BlockReader, BlockSource, HeaderProvider, ProviderFactory, ReceiptProvider, - StateProvider, TransactionsProvider, + providers::StaticFileProvider, AccountReader, BlockReader, BlockSource, HeaderProvider, + ProviderFactory, ReceiptProvider, StateProvider, TransactionsProvider, }; use reth_rpc_types::{Filter, FilteredParams}; use std::path::Path; @@ -24,7 +24,11 @@ fn main() -> eyre::Result<()> { // Instantiate a provider factory for Ethereum mainnet using the provided DB. // TODO: Should the DB version include the spec so that you do not need to specify it here? let spec = ChainSpecBuilder::mainnet().build(); - let factory = ProviderFactory::new(db, spec.into(), db_path.join("static_files"))?; + let factory = ProviderFactory::new( + db, + spec.into(), + StaticFileProvider::read_only(db_path.join("static_files"))?, + ); // This call opens a RO transaction on the database. To write to the DB you'd need to call // the `provider_rw` function and look for the `Writer` variants of the traits. diff --git a/examples/rpc-db/src/main.rs b/examples/rpc-db/src/main.rs index 627da093c591..e51f8fe1091a 100644 --- a/examples/rpc-db/src/main.rs +++ b/examples/rpc-db/src/main.rs @@ -14,7 +14,10 @@ use reth::{ primitives::ChainSpecBuilder, - providers::{providers::BlockchainProvider, ProviderFactory}, + providers::{ + providers::{BlockchainProvider, StaticFileProvider}, + ProviderFactory, + }, utils::db::open_db_read_only, }; use reth_db::{mdbx::DatabaseArguments, models::client_version::ClientVersion}; @@ -44,7 +47,11 @@ async fn main() -> eyre::Result<()> { DatabaseArguments::new(ClientVersion::default()), )?); let spec = Arc::new(ChainSpecBuilder::mainnet().build()); - let factory = ProviderFactory::new(db.clone(), spec.clone(), db_path.join("static_files"))?; + let factory = ProviderFactory::new( + db.clone(), + spec.clone(), + StaticFileProvider::read_only(db_path.join("static_files"))?, + ); // 2. Setup the blockchain provider using only the database provider and a noop for the tree to // satisfy trait bounds. Tree is not used in this example since we are only operating on the diff --git a/testing/ef-tests/src/cases/blockchain_test.rs b/testing/ef-tests/src/cases/blockchain_test.rs index fb71d1a868a7..a0aee453fe63 100644 --- a/testing/ef-tests/src/cases/blockchain_test.rs +++ b/testing/ef-tests/src/cases/blockchain_test.rs @@ -8,7 +8,10 @@ use alloy_rlp::Decodable; use rayon::iter::{ParallelBridge, ParallelIterator}; use reth_db::test_utils::{create_test_rw_db, create_test_static_files_dir}; use reth_primitives::{BlockBody, SealedBlock, StaticFileSegment}; -use reth_provider::{providers::StaticFileWriter, HashingWriter, ProviderFactory}; +use reth_provider::{ + providers::{StaticFileProvider, StaticFileWriter}, + HashingWriter, ProviderFactory, +}; use reth_stages::{stages::ExecutionStage, ExecInput, Stage}; use std::{collections::BTreeMap, fs, path::Path, sync::Arc}; @@ -86,8 +89,8 @@ impl Case for BlockchainTestCase { let provider = ProviderFactory::new( db.as_ref(), Arc::new(case.network.clone().into()), - static_files_dir_path, - )? + StaticFileProvider::read_write(static_files_dir_path).unwrap(), + ) .provider_rw() .unwrap();