github/reth
1
1
Fork 0
mirror of https://github.com/paradigmxyz/reth.git synced 2026-02-17 18:31:42 -05:00
Code Issues Packages Projects Releases Wiki Activity

refactor: make reth-prune independent of concrete DatabaseProvider (#10921)

Browse Source
This commit is contained in:
Arsenii Kulikov
2024-09-16 14:46:53 +03:00
committed by GitHub
parent 664f8b23be
commit 0fa8e83e16
31 changed files with 458 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
crates/storage/provider/src/providers/blockchain_provider.rs
View File

@@ -13,6 +13,7 @@ use reth_chain_state::{
MemoryOverlayStateProvider,
};
use reth_chainspec::ChainInfo;
use reth_db::Database;
use reth_db_api::models::{AccountBeforeTx, StoredBlockBodyIndices};
use reth_evm::ConfigureEvmEnv;
use reth_execution_types::ExecutionOutcome;
@@ -34,7 +35,7 @@ use std::{
};
use tracing::trace;
use super::ProviderNodeTypes;
use super::{DatabaseProvider, ProviderNodeTypes};
/// The main type for interacting with the blockchain.
///
@@ -263,10 +264,15 @@ impl<N: ProviderNodeTypes> BlockchainProvider2<N> {
impl<N: ProviderNodeTypes> DatabaseProviderFactory for BlockchainProvider2<N> {
type DB = N::DB;
type Provider = DatabaseProviderRO<N::DB>;
type Provider = DatabaseProvider<<N::DB as Database>::TX>;
type ProviderRW = DatabaseProvider<<N::DB as Database>::TXMut>;
fn database_provider_ro(&self) -> ProviderResult<Self::Provider> {
self.database.provider()
self.database.database_provider_ro()
}
fn database_provider_rw(&self) -> ProviderResult<Self::ProviderRW> {
self.database.database_provider_rw()
}
}

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

@@ -186,10 +186,15 @@ impl<N: ProviderNodeTypes> ProviderFactory<N> {
impl<N: ProviderNodeTypes> DatabaseProviderFactory for ProviderFactory<N> {
type DB = N::DB;
type Provider = DatabaseProviderRO<N::DB>;
type ProviderRW = DatabaseProvider<<N::DB as Database>::TXMut>;
fn database_provider_ro(&self) -> ProviderResult<Self::Provider> {
self.provider()
}
fn database_provider_rw(&self) -> ProviderResult<Self::ProviderRW> {
self.provider_rw().map(|provider| provider.0)
}
}
impl<N: NodeTypesWithDB> StaticFileProviderFactory for ProviderFactory<N> {

125
crates/storage/provider/src/providers/database/provider.rs
View File

@@ -23,13 +23,13 @@ use reth_db::{
};
use reth_db_api::{
common::KeyValue,
cursor::{DbCursorRO, DbCursorRW, DbDupCursorRO, RangeWalker},
cursor::{DbCursorRO, DbCursorRW, DbDupCursorRO},
database::Database,
models::{
sharded_key, storage_sharded_key::StorageShardedKey, AccountBeforeTx, BlockNumberAddress,
ShardedKey, StoredBlockBodyIndices, StoredBlockOmmers, StoredBlockWithdrawals,
},
table::{Table, TableRow},
table::Table,
transaction::{DbTx, DbTxMut},
DatabaseError,
};
@@ -43,7 +43,7 @@ use reth_primitives::{
TransactionSigned, TransactionSignedEcRecovered, TransactionSignedNoHash, TxHash, TxNumber,
Withdrawal, Withdrawals, B256, U256,
};
use reth_prune_types::{PruneCheckpoint, PruneLimiter, PruneModes, PruneSegment};
use reth_prune_types::{PruneCheckpoint, PruneModes, PruneSegment};
use reth_stages_types::{StageCheckpoint, StageId};
use reth_storage_api::TryIntoHistoricalStateProvider;
use reth_storage_errors::provider::{ProviderResult, RootMismatch};
@@ -1519,119 +1519,6 @@ impl<TX: DbTxMut + DbTx> DatabaseProvider<TX> {
Ok(())
}
/// Prune the table for the specified pre-sorted key iterator.
///
/// Returns number of rows pruned.
pub fn prune_table_with_iterator<T: Table>(
&self,
keys: impl IntoIterator<Item = T::Key>,
limiter: &mut PruneLimiter,
mut delete_callback: impl FnMut(TableRow<T>),
) -> Result<(usize, bool), DatabaseError> {
let mut cursor = self.tx.cursor_write::<T>()?;
let mut keys = keys.into_iter();
let mut deleted_entries = 0;
for key in &mut keys {
if limiter.is_limit_reached() {
debug!(
target: "providers::db",
?limiter,
deleted_entries_limit = %limiter.is_deleted_entries_limit_reached(),
time_limit = %limiter.is_time_limit_reached(),
table = %T::NAME,
"Pruning limit reached"
);
break
}
let row = cursor.seek_exact(key)?;
if let Some(row) = row {
cursor.delete_current()?;
limiter.increment_deleted_entries_count();
deleted_entries += 1;
delete_callback(row);
}
}
let done = keys.next().is_none();
Ok((deleted_entries, done))
}
/// Prune the table for the specified key range.
///
/// Returns number of rows pruned.
pub fn prune_table_with_range<T: Table>(
&self,
keys: impl RangeBounds<T::Key> + Clone + Debug,
limiter: &mut PruneLimiter,
mut skip_filter: impl FnMut(&TableRow<T>) -> bool,
mut delete_callback: impl FnMut(TableRow<T>),
) -> Result<(usize, bool), DatabaseError> {
let mut cursor = self.tx.cursor_write::<T>()?;
let mut walker = cursor.walk_range(keys)?;
let mut deleted_entries = 0;
let done = loop {
// check for time out must be done in this scope since it's not done in
// `prune_table_with_range_step`
if limiter.is_limit_reached() {
debug!(
target: "providers::db",
?limiter,
deleted_entries_limit = %limiter.is_deleted_entries_limit_reached(),
time_limit = %limiter.is_time_limit_reached(),
table = %T::NAME,
"Pruning limit reached"
);
break false
}
let done = self.prune_table_with_range_step(
&mut walker,
limiter,
&mut skip_filter,
&mut delete_callback,
)?;
if done {
break true
}
deleted_entries += 1;
};
Ok((deleted_entries, done))
}
/// Steps once with the given walker and prunes the entry in the table.
///
/// Returns `true` if the walker is finished, `false` if it may have more data to prune.
///
/// CAUTION: Pruner limits are not checked. This allows for a clean exit of a prune run that's
/// pruning different tables concurrently, by letting them step to the same height before
/// timing out.
pub fn prune_table_with_range_step<T: Table>(
&self,
walker: &mut RangeWalker<'_, T, <TX as DbTxMut>::CursorMut<T>>,
limiter: &mut PruneLimiter,
skip_filter: &mut impl FnMut(&TableRow<T>) -> bool,
delete_callback: &mut impl FnMut(TableRow<T>),
) -> Result<bool, DatabaseError> {
let Some(res) = walker.next() else { return Ok(true) };
let row = res?;
if !skip_filter(&row) {
walker.delete_current()?;
limiter.increment_deleted_entries_count();
delete_callback(row);
}
Ok(false)
}
/// Load shard and remove it. If list is empty, last shard was full or
/// there are no shards at all.
fn take_shard<T>(&self, key: T::Key) -> ProviderResult<Vec<u64>>
@@ -3690,7 +3577,7 @@ impl<TX: DbTxMut> FinalizedBlockWriter for DatabaseProvider<TX> {
}
}
impl<TX: DbTx> DBProvider for DatabaseProvider<TX> {
impl<TX: DbTx + 'static> DBProvider for DatabaseProvider<TX> {
type Tx = TX;
fn tx_ref(&self) -> &Self::Tx {
@@ -3700,6 +3587,10 @@ impl<TX: DbTx> DBProvider for DatabaseProvider<TX> {
fn tx_mut(&mut self) -> &mut Self::Tx {
&mut self.tx
}
fn into_tx(self) -> Self::Tx {
self.tx
}
}
/// Helper method to recover senders for any blocks in the db which do not have senders. This

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

@@ -14,6 +14,7 @@ use reth_blockchain_tree_api::{
};
use reth_chain_state::{ChainInfoTracker, ForkChoiceNotifications, ForkChoiceSubscriptions};
use reth_chainspec::{ChainInfo, ChainSpec};
use reth_db::Database;
use reth_db_api::models::{AccountBeforeTx, StoredBlockBodyIndices};
use reth_evm::ConfigureEvmEnv;
use reth_node_types::NodeTypesWithDB;
@@ -171,11 +172,16 @@ where
impl<N: ProviderNodeTypes> DatabaseProviderFactory for BlockchainProvider<N> {
type DB = N::DB;
type Provider = DatabaseProviderRO<N::DB>;
type Provider = DatabaseProvider<<N::DB as Database>::TX>;
type ProviderRW = DatabaseProvider<<N::DB as Database>::TXMut>;
fn database_provider_ro(&self) -> ProviderResult<Self::Provider> {
self.database.provider()
}
fn database_provider_rw(&self) -> ProviderResult<Self::ProviderRW> {
self.database.provider_rw().map(|p| p.0)
}
}
impl<N: ProviderNodeTypes> StaticFileProviderFactory for BlockchainProvider<N> {

5
crates/storage/provider/src/test_utils/mock.rs
View File

@@ -146,10 +146,15 @@ impl MockEthProvider {
impl DatabaseProviderFactory for MockEthProvider {
type DB = DatabaseMock;
type Provider = DatabaseProvider<TxMock>;
type ProviderRW = DatabaseProvider<TxMock>;
fn database_provider_ro(&self) -> ProviderResult<Self::Provider> {
Err(ConsistentViewError::Syncing { best_block: GotExpected::new(0, 0) }.into())
}
fn database_provider_rw(&self) -> ProviderResult<Self::ProviderRW> {
Err(ConsistentViewError::Syncing { best_block: GotExpected::new(0, 0) }.into())
}
}
impl HeaderProvider for MockEthProvider {