github/reth
1
1
Fork 0
mirror of https://github.com/paradigmxyz/reth.git synced 2026-01-30 01:28:21 -05:00
Code Issues Packages Projects Releases Wiki Activity

chore(stages): remove unnecessary prune configuration from history index stages (#4018)

Browse Source
This commit is contained in:
joshieDo
2023-08-01 13:17:51 +01:00
committed by GitHub
parent a371cb8bc1
commit 72e6a1ec9f
3 changed files with 22 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
crates/stages/src/stages/index_account_history.rs
View File

@@ -1,9 +1,6 @@
use crate::{ExecInput, ExecOutput, Stage, StageError, UnwindInput, UnwindOutput};
use reth_db::database::Database;
use reth_primitives::{
stage::{StageCheckpoint, StageId},
PruneModes,
};
use reth_primitives::stage::{StageCheckpoint, StageId};
use reth_provider::{AccountExtReader, DatabaseProviderRW, HistoryWriter};
use std::fmt::Debug;
@@ -15,20 +12,18 @@ pub struct IndexAccountHistoryStage {
/// Number of blocks after which the control
/// flow will be returned to the pipeline for commit.
pub commit_threshold: u64,
/// Pruning configuration.
pub prune_modes: PruneModes,
}
impl IndexAccountHistoryStage {
/// Create new instance of [IndexAccountHistoryStage].
pub fn new(commit_threshold: u64) -> Self {
Self { commit_threshold, prune_modes: PruneModes::default() }
Self { commit_threshold }
}
}
impl Default for IndexAccountHistoryStage {
fn default() -> Self {
Self { commit_threshold: 100_000, prune_modes: PruneModes::default() }
Self { commit_threshold: 100_000 }
}
}
@@ -43,16 +38,8 @@ impl<DB: Database> Stage<DB> for IndexAccountHistoryStage {
async fn execute(
&mut self,
provider: &DatabaseProviderRW<'_, &DB>,
mut input: ExecInput,
input: ExecInput,
) -> Result<ExecOutput, StageError> {
if let Some((target_prunable_block, _)) =
self.prune_modes.prune_target_block_account_history(input.target())?
{
if target_prunable_block > input.checkpoint().block_number {
input.checkpoint = Some(StageCheckpoint::new(target_prunable_block));
}
}
if input.target_reached() {
return Ok(ExecOutput::done(input.checkpoint()))
}
@@ -385,16 +372,11 @@ mod tests {
struct IndexAccountHistoryTestRunner {
pub(crate) tx: TestTransaction,
commit_threshold: u64,
prune_modes: PruneModes,
}
impl Default for IndexAccountHistoryTestRunner {
fn default() -> Self {
Self {
tx: TestTransaction::default(),
commit_threshold: 1000,
prune_modes: PruneModes::default(),
}
Self { tx: TestTransaction::default(), commit_threshold: 1000 }
}
}
@@ -406,7 +388,7 @@ mod tests {
}
fn stage(&self) -> Self::S {
Self::S { commit_threshold: self.commit_threshold, prune_modes: self.prune_modes }
Self::S { commit_threshold: self.commit_threshold }
}
}

30
crates/stages/src/stages/index_storage_history.rs
View File

@@ -1,9 +1,6 @@
use crate::{ExecInput, ExecOutput, Stage, StageError, UnwindInput, UnwindOutput};
use reth_db::{database::Database, models::BlockNumberAddress};
use reth_primitives::{
stage::{StageCheckpoint, StageId},
PruneModes,
};
use reth_primitives::stage::{StageCheckpoint, StageId};
use reth_provider::{DatabaseProviderRW, HistoryWriter, StorageReader};
use std::fmt::Debug;
@@ -15,20 +12,18 @@ pub struct IndexStorageHistoryStage {
/// Number of blocks after which the control
/// flow will be returned to the pipeline for commit.
pub commit_threshold: u64,
/// Pruning configuration.
pub prune_modes: PruneModes,
}
impl IndexStorageHistoryStage {
/// Create new instance of [IndexStorageHistoryStage].
pub fn new(commit_threshold: u64) -> Self {
Self { commit_threshold, prune_modes: PruneModes::default() }
Self { commit_threshold }
}
}
impl Default for IndexStorageHistoryStage {
fn default() -> Self {
Self { commit_threshold: 100_000, prune_modes: PruneModes::default() }
Self { commit_threshold: 100_000 }
}
}
@@ -43,16 +38,8 @@ impl<DB: Database> Stage<DB> for IndexStorageHistoryStage {
async fn execute(
&mut self,
provider: &DatabaseProviderRW<'_, &DB>,
mut input: ExecInput,
input: ExecInput,
) -> Result<ExecOutput, StageError> {
if let Some((target_prunable_block, _)) =
self.prune_modes.prune_target_block_storage_history(input.target())?
{
if target_prunable_block > input.checkpoint().block_number {
input.checkpoint = Some(StageCheckpoint::new(target_prunable_block));
}
}
if input.target_reached() {
return Ok(ExecOutput::done(input.checkpoint()))
}
@@ -399,16 +386,11 @@ mod tests {
struct IndexStorageHistoryTestRunner {
pub(crate) tx: TestTransaction,
commit_threshold: u64,
prune_modes: PruneModes,
}
impl Default for IndexStorageHistoryTestRunner {
fn default() -> Self {
Self {
tx: TestTransaction::default(),
commit_threshold: 1000,
prune_modes: PruneModes::default(),
}
Self { tx: TestTransaction::default(), commit_threshold: 1000 }
}
}
@@ -420,7 +402,7 @@ mod tests {
}
fn stage(&self) -> Self::S {
Self::S { commit_threshold: self.commit_threshold, prune_modes: self.prune_modes }
Self::S { commit_threshold: self.commit_threshold }
}
}

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

@@ -154,36 +154,19 @@ mod tests {
);
// Check AccountHistory
let mut acc_indexing_stage =
IndexAccountHistoryStage { prune_modes, ..Default::default() };
if let Some(PruneMode::Full) = prune_modes.account_history {
// Full is not supported
assert!(acc_indexing_stage.execute(&provider, input).await.is_err());
} else {
acc_indexing_stage.execute(&provider, input).await.unwrap();
let mut account_history: Cursor<'_, RW, AccountHistory> =
provider.tx_ref().cursor_read::<tables::AccountHistory>().unwrap();
assert_eq!(account_history.walk(None).unwrap().count(), expect_num_acc_changesets);
}
let mut acc_indexing_stage = IndexAccountHistoryStage::default();
acc_indexing_stage.execute(&provider, input).await.unwrap();
let mut account_history: Cursor<'_, RW, AccountHistory> =
provider.tx_ref().cursor_read::<tables::AccountHistory>().unwrap();
assert_eq!(account_history.walk(None).unwrap().count(), expect_num_acc_changesets);
// Check StorageHistory
let mut storage_indexing_stage =
IndexStorageHistoryStage { prune_modes, ..Default::default() };
let mut storage_indexing_stage = IndexStorageHistoryStage::default();
storage_indexing_stage.execute(&provider, input).await.unwrap();
if let Some(PruneMode::Full) = prune_modes.storage_history {
// Full is not supported
assert!(acc_indexing_stage.execute(&provider, input).await.is_err());
} else {
storage_indexing_stage.execute(&provider, input).await.unwrap();
let mut storage_history =
provider.tx_ref().cursor_read::<tables::StorageHistory>().unwrap();
assert_eq!(
storage_history.walk(None).unwrap().count(),
expect_num_storage_changesets
);
}
let mut storage_history =
provider.tx_ref().cursor_read::<tables::StorageHistory>().unwrap();
assert_eq!(storage_history.walk(None).unwrap().count(), expect_num_storage_changesets);
};
// In an unpruned configuration there is 1 receipt, 3 changed accounts and 1 changed