jleony/nanoreth
1
0
Fork 0
mirror of https://github.com/hl-archive-node/nanoreth.git synced 2025-12-06 10:59:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: run StaticFileProvider::check_consistency on start up (#8143)

Browse Source
This commit is contained in:
joshieDo
2024-06-01 12:56:13 +02:00
committed by GitHub
parent 105570ded0
commit a4df6bbd62
58 changed files with 1335 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
bin/reth/src/commands/db/mod.rs
View File

@ -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)?;
}

2
bin/reth/src/commands/db/static_files/headers.rs
View File

@ -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,

8
bin/reth/src/commands/db/static_files/mod.rs
View File

@ -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 {

2
bin/reth/src/commands/db/static_files/receipts.rs
View File

@ -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,

2
bin/reth/src/commands/db/static_files/transactions.rs
View File

@ -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,

2
bin/reth/src/commands/db/stats.rs
View File

@ -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;

15
bin/reth/src/commands/debug_cmd/build_block.rs
View File

@ -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<dyn Consensus> =
Arc::new(EthBeaconConsensus::new(Arc::clone(&self.chain)));

20
bin/reth/src/commands/debug_cmd/execution.rs
View File

@ -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<NetworkHandle> {
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())?;

18
bin/reth/src/commands/debug_cmd/in_memory_merkle.rs
View File

@ -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

17
bin/reth/src/commands/debug_cmd/merkle.rs
View File

@ -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

17
bin/reth/src/commands/debug_cmd/replay_engine.rs
View File

@ -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<dyn Consensus> =
Arc::new(EthBeaconConsensus::new(Arc::clone(&self.chain)));

12
bin/reth/src/commands/import.rs
View File

@ -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");

12
bin/reth/src/commands/import_op.rs
View File

@ -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");

11
bin/reth/src/commands/import_receipts_op.rs
View File

@ -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,

8
bin/reth/src/commands/init_cmd.rs
View File

@ -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");

8
bin/reth/src/commands/init_state.rs
View File

@ -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(),

6
bin/reth/src/commands/p2p/mod.rs
View File

@ -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?;

10
bin/reth/src/commands/recover/storage_tries.rs
View File

@ -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())?;

12
bin/reth/src/commands/stage/drop.rs
View File

@ -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())?;

8
bin/reth/src/commands/stage/dump/execution.rs
View File

@ -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<DB: Database>(
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,
)

8
bin/reth/src/commands/stage/dump/hashing_account.rs
View File

@ -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<DB: Database>(
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,
)

8
bin/reth/src/commands/stage/dump/hashing_storage.rs
View File

@ -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<DB: Database>(
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,
)

8
bin/reth/src/commands/stage/dump/merkle.rs
View File

@ -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<DB: Database>(
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,
)

9
bin/reth/src/commands/stage/dump/mod.rs
View File

@ -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");

14
bin/reth/src/commands/stage/run.rs
View File

@ -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

11
bin/reth/src/commands/stage/unwind.rs
View File

@ -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 {

1
crates/blockchain-tree-api/src/error.rs
View File

@ -296,7 +296,6 @@ impl InsertBlockErrorKind {
BlockExecutionError::CanonicalRevert { .. } |
BlockExecutionError::CanonicalCommit { .. } |
BlockExecutionError::AppendChainDoesntConnect { .. } |
BlockExecutionError::UnavailableForTest => false,
BlockExecutionError::Other(_) => false,
}
}

7
crates/config/src/config.rs
View File

@ -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<Option<Duration>, D::Error>
where

76
crates/consensus/beacon/src/engine/mod.rs
View File

@ -2179,6 +2179,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;
@ -2235,8 +2236,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
@ -2292,8 +2296,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(),
);
@ -2313,8 +2320,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(),
);
@ -2354,8 +2364,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(),
);
@ -2406,8 +2419,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(),
);
@ -2452,8 +2468,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(),
);
@ -2479,9 +2498,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();
@ -2551,8 +2571,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(),
);
@ -2620,8 +2643,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(),
);
@ -2664,8 +2690,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(),
);
@ -2728,8 +2757,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(),
);

5
crates/evm/execution-errors/src/lib.rs
View File

@ -131,11 +131,6 @@ pub enum BlockExecutionError {
/// The fork on the other chain
other_chain_fork: Box<BlockNumHash>,
},
/// 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),

2
crates/evm/src/execute.rs
View File

@ -245,7 +245,7 @@ mod tests {
type Error = BlockExecutionError;
fn execute(self, _input: Self::Input<'_>) -> Result<Self::Output, Self::Error> {
Err(BlockExecutionError::UnavailableForTest)
Err(BlockExecutionError::msg("execution unavailable for tests"))
}
}

1
crates/evm/src/lib.rs
View File

@ -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"))]

68
crates/evm/src/noop.rs Normal file
View File

@ -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<DB: Database<Error = ProviderError>> = Self;
type BatchExecutor<DB: Database<Error = ProviderError>> = Self;
fn executor<DB>(&self, _: DB) -> Self::Executor<DB>
where
DB: Database<Error = ProviderError>,
{
Self
}
fn batch_executor<DB>(&self, _: DB, _: PruneModes) -> Self::BatchExecutor<DB>
where
DB: Database<Error = ProviderError>,
{
Self
}
}
impl<DB> Executor<DB> for NoopBlockExecutorProvider {
type Input<'a> = BlockExecutionInput<'a, BlockWithSenders>;
type Output = BlockExecutionOutput<Receipt>;
type Error = BlockExecutionError;
fn execute(self, _: Self::Input<'_>) -> Result<Self::Output, Self::Error> {
Err(BlockExecutionError::msg(UNAVAILABLE_FOR_NOOP))
}
}
impl<DB> BatchExecutor<DB> 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<usize> {
None
}
}

38
crates/net/downloaders/src/bodies/bodies.rs
View File

@ -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

84
crates/node/builder/src/launch/common.rs
View File

@ -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<R> LaunchContextWith<Attached<WithConfigs, R>> {
impl<DB> LaunchContextWith<Attached<WithConfigs, DB>>
where
DB: Clone,
DB: Database + Clone + 'static,
{
/// Returns the [ProviderFactory] for the attached database.
pub fn create_provider_factory(&self) -> eyre::Result<ProviderFactory<DB>> {
/// 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<ProviderFactory<DB>> {
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<LaunchContextWith<Attached<WithConfigs, ProviderFactory<DB>>>> {
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),

2
crates/node/builder/src/launch/mod.rs
View File

@ -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");
})

13
crates/primitives/src/stage/mod.rs
View File

@ -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<BlockHash> 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})"),
}
}
}

9
crates/prune/src/pruner.rs
View File

@ -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);

12
crates/stages/api/src/pipeline/mod.rs
View File

@ -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()?;

5
crates/stages/stages/src/stages/execution.rs
View File

@ -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)

270
crates/stages/stages/src/stages/mod.rs
View File

@ -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<TestStageDB> {
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<PipelineTarget>,
) {
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<PipelineTarget>,
) {
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<T: Table<Key = u64>>(
db: &TestStageDB,
key: u64,
expected: Option<PipelineTarget>,
) where
<T as Table>::Value: Default,
{
let provider_rw = db.factory.provider_rw().unwrap();
let mut cursor = provider_rw.tx_ref().cursor_write::<T>().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 <missing> db
update_db_and_check::<tables::Headers>(&db, current + 2, Some(PipelineTarget::Unwind(89)));
// Fill the gap, and ensure no unwind is necessary.
update_db_and_check::<tables::Headers>(&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 <missing> db
update_db_and_check::<tables::Transactions>(
&db,
current + 2,
Some(PipelineTarget::Unwind(89)),
);
// Fill the gap, and ensure no unwind is necessary.
update_db_and_check::<tables::Transactions>(&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 <missing> db
update_db_and_check::<tables::Receipts>(&db, current + 2, Some(PipelineTarget::Unwind(89)));
// Fill the gap, and ensure no unwind is necessary.
update_db_and_check::<tables::Receipts>(&db, current + 1, None);
}
}

60
crates/stages/stages/src/test_utils/test_db.rs
View File

@ -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::<Vec<_>>();
{
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<I, J>(
&self,
receipts: I,
storage_kind: StorageKind,
) -> ProviderResult<()>
where
I: IntoIterator<Item = (BlockNumber, J)>,
J: IntoIterator<Item = (TxNumber, Receipt)>,
{
match storage_kind {
StorageKind::Database(_) => self.commit(|tx| {
receipts.into_iter().try_for_each(|(_, receipts)| {
for (tx_num, receipt) in receipts {
tx.put::<tables::Receipts>(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<I>(&self, transaction_senders: I) -> ProviderResult<()>
where
I: IntoIterator<Item = (TxNumber, Address)>,

5
crates/static-file-types/src/segment.rs
View File

@ -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.

13
crates/storage/db-common/src/init.rs
View File

@ -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(),

3
crates/storage/errors/src/provider.rs
View File

@ -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),

2
crates/storage/nippy-jar/src/error.rs
View File

@ -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,
}

22
crates/storage/nippy-jar/src/lib.rs
View File

@ -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<H: NippyJarHeader> NippyJar<H> {
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<u8>], col2: &[Vec<u8>]) {
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();
}
}

68
crates/storage/nippy-jar/src/writer.rs
View File

@ -43,7 +43,13 @@ pub struct NippyJarWriter<H: NippyJarHeader = ()> {
impl<H: NippyJarHeader> NippyJarWriter<H> {
/// Creates a [`NippyJarWriter`] from [`NippyJar`].
pub fn new(jar: NippyJar<H>) -> Result<Self, NippyJarError> {
///
/// 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<H>,
check_mode: ConsistencyFailStrategy,
) -> Result<Self, NippyJarError> {
let (data_file, offsets_file, is_created) =
Self::create_or_open_files(jar.data_path(), &jar.offsets_path())?;
@ -63,8 +69,10 @@ impl<H: NippyJarHeader> NippyJarWriter<H> {
// 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<H: NippyJarHeader> NippyJarWriter<H> {
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<H: NippyJarHeader> NippyJarWriter<H> {
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<H: NippyJarHeader> NippyJarWriter<H> {
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<H: NippyJarHeader> NippyJarWriter<H> {
// 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<H: NippyJarHeader> NippyJarWriter<H> {
// 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<H: NippyJarHeader> NippyJarWriter<H> {
pub fn data_path(&self) -> &Path {
self.jar.data_path()
}
#[cfg(any(test, feature = "test-utils"))]
pub fn data_file(&mut self) -> &mut BufWriter<File> {
&mut self.data_file
}
#[cfg(any(test, feature = "test-utils"))]
pub const fn jar(&self) -> &NippyJar<H> {
&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)
}
}

2
crates/storage/provider/src/lib.rs
View File

@ -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"))]

24
crates/storage/provider/src/providers/database/mod.rs
View File

@ -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<DB> ProviderFactory<DB> {
pub fn new(
db: DB,
chain_spec: Arc<ChainSpec>,
static_files_path: PathBuf,
) -> ProviderResult<Self> {
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<DatabaseEnv> {
path: P,
chain_spec: Arc<ChainSpec>,
args: DatabaseArguments,
static_files_path: PathBuf,
static_file_provider: StaticFileProvider,
) -> RethResult<Self> {
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<DB> Clone for ProviderFactory<DB> {
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();

4
crates/storage/provider/src/providers/mod.rs
View File

@ -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;

350
crates/storage/provider/src/providers/static_file/manager.rs
View File

@ -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<StaticFileSegment, BTreeMap<TxNumber, SegmentRangeInclusive>>;
/// 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<StaticFileProviderInner>);
impl StaticFileProvider {
/// Creates a new [`StaticFileProvider`].
pub fn new(path: impl AsRef<Path>) -> ProviderResult<Self> {
let provider = Self(Arc::new(StaticFileProviderInner::new(path)?));
fn new(path: impl AsRef<Path>, access: StaticFileAccess) -> ProviderResult<Self> {
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<Path>) -> ProviderResult<Self> {
Self::new(path, StaticFileAccess::RO)
}
/// Creates a new [`StaticFileProvider`] with read-write access.
pub fn read_write(path: impl AsRef<Path>) -> ProviderResult<Self> {
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<StaticFileSegment, StaticFileProviderRW>,
metrics: Option<Arc<StaticFileProviderMetrics>>,
/// Access rights of the provider.
access: StaticFileAccess,
}
impl StaticFileProviderInner {
/// Creates a new [`StaticFileProviderInner`].
fn new(path: impl AsRef<Path>) -> ProviderResult<Self> {
fn new(path: impl AsRef<Path>, access: StaticFileAccess) -> ProviderResult<Self> {
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<TX: DbTx>(
&self,
provider: &DatabaseProvider<TX>,
has_receipt_pruning: bool,
) -> ProviderResult<Option<PipelineTarget>> {
let mut unwind_target: Option<BlockNumber> = 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<TX: DbTx, T: Table<Key = u64>>(
&self,
provider: &DatabaseProvider<TX>,
segment: StaticFileSegment,
highest_static_file_entry: Option<u64>,
highest_static_file_block: Option<BlockNumber>,
) -> ProviderResult<Option<BlockNumber>> {
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::<T>()?;
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<BlockNumber> {
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<StaticFileProviderRWRefMut<'_>> {
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<Option<Header>> {
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<Option<U256>> {
@ -785,8 +1062,15 @@ impl HeaderProvider for StaticFileProvider {
}
fn header_td_by_number(&self, num: BlockNumber) -> ProviderResult<Option<U256>> {
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<BlockNumber>) -> ProviderResult<Vec<Header>> {
@ -799,8 +1083,15 @@ impl HeaderProvider for StaticFileProvider {
}
fn sealed_header(&self, num: BlockNumber) -> ProviderResult<Option<SealedHeader>> {
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<Option<Receipt>> {
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<Option<Receipt>> {
@ -947,16 +1245,30 @@ impl TransactionsProvider for StaticFileProvider {
}
fn transaction_by_id(&self, num: TxNumber) -> ProviderResult<Option<TransactionSigned>> {
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<Option<TransactionSignedNoHash>> {
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<Option<TransactionSigned>> {

5
crates/storage/provider/src/providers/static_file/mod.rs
View File

@ -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();

140
crates/storage/provider/src/providers/static_file/writer.rs
View File

@ -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<u64>,
) -> 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::<SegmentHeader>::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::<SegmentHeader>::load(&current_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<T: Compact>(&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<BlockNumber>) {
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<SegmentHeader> {
&mut self.writer
}
}

9
crates/storage/provider/src/test_utils/mod.rs
View File

@ -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<Arc<TempDatabase<DatabaseEnv>>> {
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"),
)
}

10
examples/db-access/src/main.rs
View File

@ -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.

11
examples/rpc-db/src/main.rs
View File

@ -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

9
testing/ef-tests/src/cases/blockchain_test.rs
View File

@ -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();