mirror of
https://github.com/hl-archive-node/nanoreth.git
synced 2025-12-06 19:09:54 +00:00
feat(pruner, storage): prune receipts & save checkpoints to database (#3733)
Co-authored-by: joshieDo <ranriver@protonmail.com>
This commit is contained in:
6
Cargo.lock
generated
6
Cargo.lock
generated
@ -5586,7 +5586,13 @@ dependencies = [
|
|||||||
name = "reth-prune"
|
name = "reth-prune"
|
||||||
version = "0.1.0-alpha.4"
|
version = "0.1.0-alpha.4"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
"assert_matches",
|
||||||
|
"itertools",
|
||||||
|
"reth-db",
|
||||||
|
"reth-interfaces",
|
||||||
"reth-primitives",
|
"reth-primitives",
|
||||||
|
"reth-provider",
|
||||||
|
"reth-stages",
|
||||||
"thiserror",
|
"thiserror",
|
||||||
"tracing",
|
"tracing",
|
||||||
]
|
]
|
||||||
|
|||||||
@ -8,7 +8,7 @@ use reth_db::{cursor::DbCursorRO, init_db, tables, transaction::DbTx};
|
|||||||
use reth_primitives::{
|
use reth_primitives::{
|
||||||
fs,
|
fs,
|
||||||
stage::{StageCheckpoint, StageId},
|
stage::{StageCheckpoint, StageId},
|
||||||
ChainSpec, PruneTargets,
|
ChainSpec, PruneModes,
|
||||||
};
|
};
|
||||||
use reth_provider::{ProviderFactory, StageCheckpointReader};
|
use reth_provider::{ProviderFactory, StageCheckpointReader};
|
||||||
use reth_stages::{
|
use reth_stages::{
|
||||||
@ -96,7 +96,7 @@ impl Command {
|
|||||||
let mut execution_stage = ExecutionStage::new(
|
let mut execution_stage = ExecutionStage::new(
|
||||||
factory,
|
factory,
|
||||||
ExecutionStageThresholds { max_blocks: Some(1), max_changes: None },
|
ExecutionStageThresholds { max_blocks: Some(1), max_changes: None },
|
||||||
PruneTargets::all(),
|
PruneModes::all(),
|
||||||
);
|
);
|
||||||
|
|
||||||
let mut account_hashing_stage = AccountHashingStage::default();
|
let mut account_hashing_stage = AccountHashingStage::default();
|
||||||
|
|||||||
@ -78,6 +78,7 @@ use reth_interfaces::p2p::headers::client::HeadersClient;
|
|||||||
use reth_payload_builder::PayloadBuilderService;
|
use reth_payload_builder::PayloadBuilderService;
|
||||||
use reth_primitives::DisplayHardforks;
|
use reth_primitives::DisplayHardforks;
|
||||||
use reth_provider::providers::BlockchainProvider;
|
use reth_provider::providers::BlockchainProvider;
|
||||||
|
use reth_prune::BatchSizes;
|
||||||
use reth_stages::stages::{
|
use reth_stages::stages::{
|
||||||
AccountHashingStage, IndexAccountHistoryStage, IndexStorageHistoryStage, MerkleStage,
|
AccountHashingStage, IndexAccountHistoryStage, IndexStorageHistoryStage, MerkleStage,
|
||||||
StorageHashingStage, TransactionLookupStage,
|
StorageHashingStage, TransactionLookupStage,
|
||||||
@ -364,7 +365,14 @@ impl Command {
|
|||||||
|
|
||||||
let pruner = config.prune.map(|prune_config| {
|
let pruner = config.prune.map(|prune_config| {
|
||||||
info!(target: "reth::cli", "Pruner initialized");
|
info!(target: "reth::cli", "Pruner initialized");
|
||||||
reth_prune::Pruner::new(prune_config.block_interval, tree_config.max_reorg_depth())
|
reth_prune::Pruner::new(
|
||||||
|
db.clone(),
|
||||||
|
self.chain.clone(),
|
||||||
|
prune_config.block_interval,
|
||||||
|
tree_config.max_reorg_depth(),
|
||||||
|
prune_config.parts,
|
||||||
|
BatchSizes::default(),
|
||||||
|
)
|
||||||
});
|
});
|
||||||
|
|
||||||
// Configure the consensus engine
|
// Configure the consensus engine
|
||||||
|
|||||||
@ -2,7 +2,7 @@ use super::setup;
|
|||||||
use crate::utils::DbTool;
|
use crate::utils::DbTool;
|
||||||
use eyre::Result;
|
use eyre::Result;
|
||||||
use reth_db::{database::Database, table::TableImporter, tables, DatabaseEnv};
|
use reth_db::{database::Database, table::TableImporter, tables, DatabaseEnv};
|
||||||
use reth_primitives::{stage::StageCheckpoint, BlockNumber, ChainSpec, PruneTargets};
|
use reth_primitives::{stage::StageCheckpoint, BlockNumber, ChainSpec, PruneModes};
|
||||||
use reth_provider::ProviderFactory;
|
use reth_provider::ProviderFactory;
|
||||||
use reth_stages::{
|
use reth_stages::{
|
||||||
stages::{
|
stages::{
|
||||||
@ -70,7 +70,7 @@ async fn unwind_and_copy<DB: Database>(
|
|||||||
let mut exec_stage = ExecutionStage::new(
|
let mut exec_stage = ExecutionStage::new(
|
||||||
reth_revm::Factory::new(db_tool.chain.clone()),
|
reth_revm::Factory::new(db_tool.chain.clone()),
|
||||||
ExecutionStageThresholds { max_blocks: Some(u64::MAX), max_changes: None },
|
ExecutionStageThresholds { max_blocks: Some(u64::MAX), max_changes: None },
|
||||||
PruneTargets::all(),
|
PruneModes::all(),
|
||||||
);
|
);
|
||||||
|
|
||||||
exec_stage
|
exec_stage
|
||||||
|
|||||||
@ -5,7 +5,7 @@ use reth_downloaders::{
|
|||||||
headers::reverse_headers::ReverseHeadersDownloaderBuilder,
|
headers::reverse_headers::ReverseHeadersDownloaderBuilder,
|
||||||
};
|
};
|
||||||
use reth_network::{NetworkConfigBuilder, PeersConfig, SessionsConfig};
|
use reth_network::{NetworkConfigBuilder, PeersConfig, SessionsConfig};
|
||||||
use reth_primitives::PruneTargets;
|
use reth_primitives::PruneModes;
|
||||||
use secp256k1::SecretKey;
|
use secp256k1::SecretKey;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
@ -285,12 +285,12 @@ pub struct PruneConfig {
|
|||||||
/// Minimum pruning interval measured in blocks.
|
/// Minimum pruning interval measured in blocks.
|
||||||
pub block_interval: u64,
|
pub block_interval: u64,
|
||||||
/// Pruning configuration for every part of the data that can be pruned.
|
/// Pruning configuration for every part of the data that can be pruned.
|
||||||
pub parts: PruneTargets,
|
pub parts: PruneModes,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for PruneConfig {
|
impl Default for PruneConfig {
|
||||||
fn default() -> Self {
|
fn default() -> Self {
|
||||||
Self { block_interval: 10, parts: PruneTargets::default() }
|
Self { block_interval: 10, parts: PruneModes::default() }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -193,7 +193,7 @@ where
|
|||||||
/// be used to download and execute the missing blocks.
|
/// be used to download and execute the missing blocks.
|
||||||
pipeline_run_threshold: u64,
|
pipeline_run_threshold: u64,
|
||||||
/// Controls pruning triggered by engine updates.
|
/// Controls pruning triggered by engine updates.
|
||||||
prune: Option<EnginePruneController>,
|
prune: Option<EnginePruneController<DB>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<DB, BT, Client> BeaconConsensusEngine<DB, BT, Client>
|
impl<DB, BT, Client> BeaconConsensusEngine<DB, BT, Client>
|
||||||
@ -220,7 +220,7 @@ where
|
|||||||
payload_builder: PayloadBuilderHandle,
|
payload_builder: PayloadBuilderHandle,
|
||||||
target: Option<H256>,
|
target: Option<H256>,
|
||||||
pipeline_run_threshold: u64,
|
pipeline_run_threshold: u64,
|
||||||
pruner: Option<Pruner>,
|
pruner: Option<Pruner<DB>>,
|
||||||
) -> Result<(Self, BeaconConsensusEngineHandle), Error> {
|
) -> Result<(Self, BeaconConsensusEngineHandle), Error> {
|
||||||
let (to_engine, rx) = mpsc::unbounded_channel();
|
let (to_engine, rx) = mpsc::unbounded_channel();
|
||||||
Self::with_channel(
|
Self::with_channel(
|
||||||
@ -266,7 +266,7 @@ where
|
|||||||
pipeline_run_threshold: u64,
|
pipeline_run_threshold: u64,
|
||||||
to_engine: UnboundedSender<BeaconEngineMessage>,
|
to_engine: UnboundedSender<BeaconEngineMessage>,
|
||||||
rx: UnboundedReceiver<BeaconEngineMessage>,
|
rx: UnboundedReceiver<BeaconEngineMessage>,
|
||||||
pruner: Option<Pruner>,
|
pruner: Option<Pruner<DB>>,
|
||||||
) -> Result<(Self, BeaconConsensusEngineHandle), Error> {
|
) -> Result<(Self, BeaconConsensusEngineHandle), Error> {
|
||||||
let handle = BeaconConsensusEngineHandle { to_engine };
|
let handle = BeaconConsensusEngineHandle { to_engine };
|
||||||
let sync = EngineSyncController::new(
|
let sync = EngineSyncController::new(
|
||||||
@ -1727,11 +1727,14 @@ mod tests {
|
|||||||
test_utils::{NoopFullBlockClient, TestConsensus},
|
test_utils::{NoopFullBlockClient, TestConsensus},
|
||||||
};
|
};
|
||||||
use reth_payload_builder::test_utils::spawn_test_payload_service;
|
use reth_payload_builder::test_utils::spawn_test_payload_service;
|
||||||
use reth_primitives::{stage::StageCheckpoint, ChainSpec, ChainSpecBuilder, H256, MAINNET};
|
use reth_primitives::{
|
||||||
|
stage::StageCheckpoint, ChainSpec, ChainSpecBuilder, PruneModes, H256, MAINNET,
|
||||||
|
};
|
||||||
use reth_provider::{
|
use reth_provider::{
|
||||||
providers::BlockchainProvider, test_utils::TestExecutorFactory, BlockExecutor, BlockWriter,
|
providers::BlockchainProvider, test_utils::TestExecutorFactory, BlockExecutor, BlockWriter,
|
||||||
ExecutorFactory, ProviderFactory, StateProvider,
|
ExecutorFactory, ProviderFactory, StateProvider,
|
||||||
};
|
};
|
||||||
|
use reth_prune::BatchSizes;
|
||||||
use reth_revm::Factory;
|
use reth_revm::Factory;
|
||||||
use reth_rpc_types::engine::{
|
use reth_rpc_types::engine::{
|
||||||
ExecutionPayload, ForkchoiceState, ForkchoiceUpdated, PayloadStatus,
|
ExecutionPayload, ForkchoiceState, ForkchoiceUpdated, PayloadStatus,
|
||||||
@ -2071,7 +2074,14 @@ mod tests {
|
|||||||
let latest = self.chain_spec.genesis_header().seal_slow();
|
let latest = self.chain_spec.genesis_header().seal_slow();
|
||||||
let blockchain_provider = BlockchainProvider::with_latest(shareable_db, tree, latest);
|
let blockchain_provider = BlockchainProvider::with_latest(shareable_db, tree, latest);
|
||||||
|
|
||||||
let pruner = Pruner::new(5, 0);
|
let pruner = Pruner::new(
|
||||||
|
db.clone(),
|
||||||
|
self.chain_spec.clone(),
|
||||||
|
5,
|
||||||
|
0,
|
||||||
|
PruneModes::default(),
|
||||||
|
BatchSizes::default(),
|
||||||
|
);
|
||||||
|
|
||||||
let (mut engine, handle) = BeaconConsensusEngine::new(
|
let (mut engine, handle) = BeaconConsensusEngine::new(
|
||||||
client,
|
client,
|
||||||
|
|||||||
@ -1,6 +1,7 @@
|
|||||||
//! Prune management for the engine implementation.
|
//! Prune management for the engine implementation.
|
||||||
|
|
||||||
use futures::FutureExt;
|
use futures::FutureExt;
|
||||||
|
use reth_db::database::Database;
|
||||||
use reth_primitives::BlockNumber;
|
use reth_primitives::BlockNumber;
|
||||||
use reth_prune::{Pruner, PrunerError, PrunerWithResult};
|
use reth_prune::{Pruner, PrunerError, PrunerWithResult};
|
||||||
use reth_tasks::TaskSpawner;
|
use reth_tasks::TaskSpawner;
|
||||||
@ -10,16 +11,16 @@ use tokio::sync::oneshot;
|
|||||||
/// Manages pruning under the control of the engine.
|
/// Manages pruning under the control of the engine.
|
||||||
///
|
///
|
||||||
/// This type controls the [Pruner].
|
/// This type controls the [Pruner].
|
||||||
pub(crate) struct EnginePruneController {
|
pub(crate) struct EnginePruneController<DB> {
|
||||||
/// The current state of the pruner.
|
/// The current state of the pruner.
|
||||||
pruner_state: PrunerState,
|
pruner_state: PrunerState<DB>,
|
||||||
/// The type that can spawn the pruner task.
|
/// The type that can spawn the pruner task.
|
||||||
pruner_task_spawner: Box<dyn TaskSpawner>,
|
pruner_task_spawner: Box<dyn TaskSpawner>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl EnginePruneController {
|
impl<DB: Database + 'static> EnginePruneController<DB> {
|
||||||
/// Create a new instance
|
/// Create a new instance
|
||||||
pub(crate) fn new(pruner: Pruner, pruner_task_spawner: Box<dyn TaskSpawner>) -> Self {
|
pub(crate) fn new(pruner: Pruner<DB>, pruner_task_spawner: Box<dyn TaskSpawner>) -> Self {
|
||||||
Self { pruner_state: PrunerState::Idle(Some(pruner)), pruner_task_spawner }
|
Self { pruner_state: PrunerState::Idle(Some(pruner)), pruner_task_spawner }
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -131,14 +132,14 @@ pub(crate) enum EnginePruneEvent {
|
|||||||
/// running, it acquires the write lock over the database. This means that we cannot forward to the
|
/// running, it acquires the write lock over the database. This means that we cannot forward to the
|
||||||
/// blockchain tree any messages that would result in database writes, since it would result in a
|
/// blockchain tree any messages that would result in database writes, since it would result in a
|
||||||
/// deadlock.
|
/// deadlock.
|
||||||
enum PrunerState {
|
enum PrunerState<DB> {
|
||||||
/// Pruner is idle.
|
/// Pruner is idle.
|
||||||
Idle(Option<Pruner>),
|
Idle(Option<Pruner<DB>>),
|
||||||
/// Pruner is running and waiting for a response
|
/// Pruner is running and waiting for a response
|
||||||
Running(oneshot::Receiver<PrunerWithResult>),
|
Running(oneshot::Receiver<PrunerWithResult<DB>>),
|
||||||
}
|
}
|
||||||
|
|
||||||
impl PrunerState {
|
impl<DB> PrunerState<DB> {
|
||||||
/// Returns `true` if the state matches idle.
|
/// Returns `true` if the state matches idle.
|
||||||
fn is_idle(&self) -> bool {
|
fn is_idle(&self) -> bool {
|
||||||
matches!(self, PrunerState::Idle(_))
|
matches!(self, PrunerState::Idle(_))
|
||||||
|
|||||||
@ -78,7 +78,7 @@ pub use net::{
|
|||||||
SEPOLIA_BOOTNODES,
|
SEPOLIA_BOOTNODES,
|
||||||
};
|
};
|
||||||
pub use peer::{PeerId, WithPeerId};
|
pub use peer::{PeerId, WithPeerId};
|
||||||
pub use prune::{PruneCheckpoint, PruneMode, PrunePart, PruneTargets};
|
pub use prune::{PruneCheckpoint, PruneMode, PruneModes, PrunePart};
|
||||||
pub use receipt::{Receipt, ReceiptWithBloom, ReceiptWithBloomRef};
|
pub use receipt::{Receipt, ReceiptWithBloom, ReceiptWithBloomRef};
|
||||||
pub use revm_primitives::JumpMap;
|
pub use revm_primitives::JumpMap;
|
||||||
pub use serde_helper::JsonU256;
|
pub use serde_helper::JsonU256;
|
||||||
|
|||||||
@ -7,7 +7,7 @@ use reth_codecs::{main_codec, Compact};
|
|||||||
#[cfg_attr(test, derive(Default))]
|
#[cfg_attr(test, derive(Default))]
|
||||||
pub struct PruneCheckpoint {
|
pub struct PruneCheckpoint {
|
||||||
/// Highest pruned block number.
|
/// Highest pruned block number.
|
||||||
block_number: BlockNumber,
|
pub block_number: BlockNumber,
|
||||||
/// Prune mode.
|
/// Prune mode.
|
||||||
prune_mode: PruneMode,
|
pub prune_mode: PruneMode,
|
||||||
}
|
}
|
||||||
|
|||||||
@ -6,4 +6,4 @@ mod target;
|
|||||||
pub use checkpoint::PruneCheckpoint;
|
pub use checkpoint::PruneCheckpoint;
|
||||||
pub use mode::PruneMode;
|
pub use mode::PruneMode;
|
||||||
pub use part::PrunePart;
|
pub use part::PrunePart;
|
||||||
pub use target::PruneTargets;
|
pub use target::PruneModes;
|
||||||
|
|||||||
@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
|
|||||||
/// Pruning configuration for every part of the data that can be pruned.
|
/// Pruning configuration for every part of the data that can be pruned.
|
||||||
#[derive(Debug, Clone, Default, Copy, Deserialize, Eq, PartialEq, Serialize)]
|
#[derive(Debug, Clone, Default, Copy, Deserialize, Eq, PartialEq, Serialize)]
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub struct PruneTargets {
|
pub struct PruneModes {
|
||||||
/// Sender Recovery pruning configuration.
|
/// Sender Recovery pruning configuration.
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
pub sender_recovery: Option<PruneMode>,
|
pub sender_recovery: Option<PruneMode>,
|
||||||
@ -26,25 +26,42 @@ pub struct PruneTargets {
|
|||||||
pub storage_history: Option<PruneMode>,
|
pub storage_history: Option<PruneMode>,
|
||||||
}
|
}
|
||||||
|
|
||||||
macro_rules! should_prune_method {
|
macro_rules! impl_prune_parts {
|
||||||
($($config:ident),+) => {
|
($(($part:ident, $human_part:expr)),+) => {
|
||||||
$(
|
$(
|
||||||
paste! {
|
paste! {
|
||||||
#[allow(missing_docs)]
|
#[doc = concat!(
|
||||||
pub fn [<should_prune_ $config>](&self, block: BlockNumber, tip: BlockNumber) -> bool {
|
"Check if ",
|
||||||
if let Some(config) = &self.$config {
|
$human_part,
|
||||||
return self.should_prune(config, block, tip)
|
" should be pruned at the target block according to the provided tip."
|
||||||
|
)]
|
||||||
|
pub fn [<should_prune_ $part>](&self, block: BlockNumber, tip: BlockNumber) -> bool {
|
||||||
|
if let Some(mode) = &self.$part {
|
||||||
|
return self.should_prune(mode, block, tip)
|
||||||
}
|
}
|
||||||
false
|
false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
)+
|
)+
|
||||||
|
|
||||||
|
$(
|
||||||
|
paste! {
|
||||||
|
#[doc = concat!(
|
||||||
|
"Returns block up to which ",
|
||||||
|
$human_part,
|
||||||
|
" pruning needs to be done, inclusive, according to the provided tip."
|
||||||
|
)]
|
||||||
|
pub fn [<prune_to_block_ $part>](&self, tip: BlockNumber) -> Option<(BlockNumber, PruneMode)> {
|
||||||
|
self.$part.as_ref().map(|mode| (self.prune_to_block(mode, tip), *mode))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
)+
|
||||||
|
|
||||||
/// Sets pruning to all targets.
|
/// Sets pruning to all targets.
|
||||||
pub fn all() -> Self {
|
pub fn all() -> Self {
|
||||||
PruneTargets {
|
Self {
|
||||||
$(
|
$(
|
||||||
$config: Some(PruneMode::Full),
|
$part: Some(PruneMode::Full),
|
||||||
)+
|
)+
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -52,15 +69,15 @@ macro_rules! should_prune_method {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
impl PruneTargets {
|
impl PruneModes {
|
||||||
/// Sets pruning to no target.
|
/// Sets pruning to no target.
|
||||||
pub fn none() -> Self {
|
pub fn none() -> Self {
|
||||||
PruneTargets::default()
|
PruneModes::default()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Check if target block should be pruned
|
/// Check if target block should be pruned according to the provided prune mode and tip.
|
||||||
pub fn should_prune(&self, target: &PruneMode, block: BlockNumber, tip: BlockNumber) -> bool {
|
pub fn should_prune(&self, mode: &PruneMode, block: BlockNumber, tip: BlockNumber) -> bool {
|
||||||
match target {
|
match mode {
|
||||||
PruneMode::Full => true,
|
PruneMode::Full => true,
|
||||||
PruneMode::Distance(distance) => {
|
PruneMode::Distance(distance) => {
|
||||||
if *distance > tip {
|
if *distance > tip {
|
||||||
@ -72,11 +89,21 @@ impl PruneTargets {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
should_prune_method!(
|
/// Returns block up to which pruning needs to be done, inclusive, according to the provided
|
||||||
sender_recovery,
|
/// prune mode and tip.
|
||||||
transaction_lookup,
|
pub fn prune_to_block(&self, mode: &PruneMode, tip: BlockNumber) -> BlockNumber {
|
||||||
receipts,
|
match mode {
|
||||||
account_history,
|
PruneMode::Full => tip,
|
||||||
storage_history
|
PruneMode::Distance(distance) => tip.saturating_sub(*distance),
|
||||||
|
PruneMode::Before(n) => *n,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl_prune_parts!(
|
||||||
|
(sender_recovery, "Sender Recovery"),
|
||||||
|
(transaction_lookup, "Transaction Lookup"),
|
||||||
|
(receipts, "Receipts"),
|
||||||
|
(account_history, "Account History"),
|
||||||
|
(storage_history, "Storage History")
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|||||||
@ -13,8 +13,20 @@ Pruning implementation
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
# reth
|
# reth
|
||||||
reth-primitives = { workspace = true }
|
reth-primitives = { workspace = true }
|
||||||
|
reth-db = { workspace = true }
|
||||||
|
reth-provider = { workspace = true }
|
||||||
|
reth-interfaces = { workspace = true }
|
||||||
|
|
||||||
# misc
|
# misc
|
||||||
tracing = { workspace = true }
|
tracing = { workspace = true }
|
||||||
thiserror = { workspace = true }
|
thiserror = { workspace = true }
|
||||||
|
itertools = "0.10"
|
||||||
|
|
||||||
|
[dev-dependencies]
|
||||||
|
# reth
|
||||||
|
reth-db = { workspace = true, features = ["test-utils"] }
|
||||||
|
reth-stages = { path = "../stages", features = ["test-utils"] }
|
||||||
|
|
||||||
|
# misc
|
||||||
|
|
||||||
|
assert_matches = "1.5.0"
|
||||||
|
|||||||
@ -1,4 +1,15 @@
|
|||||||
|
use reth_db::DatabaseError;
|
||||||
|
use reth_provider::ProviderError;
|
||||||
use thiserror::Error;
|
use thiserror::Error;
|
||||||
|
|
||||||
#[derive(Error, Debug)]
|
#[derive(Error, Debug)]
|
||||||
pub enum PrunerError {}
|
pub enum PrunerError {
|
||||||
|
#[error("An interface error occurred.")]
|
||||||
|
Interface(#[from] reth_interfaces::Error),
|
||||||
|
|
||||||
|
#[error(transparent)]
|
||||||
|
Database(#[from] DatabaseError),
|
||||||
|
|
||||||
|
#[error(transparent)]
|
||||||
|
Provider(#[from] ProviderError),
|
||||||
|
}
|
||||||
|
|||||||
@ -2,4 +2,4 @@ mod error;
|
|||||||
mod pruner;
|
mod pruner;
|
||||||
|
|
||||||
pub use error::PrunerError;
|
pub use error::PrunerError;
|
||||||
pub use pruner::{Pruner, PrunerResult, PrunerWithResult};
|
pub use pruner::{BatchSizes, Pruner, PrunerResult, PrunerWithResult};
|
||||||
|
|||||||
@ -1,17 +1,31 @@
|
|||||||
//! Support for pruning.
|
//! Support for pruning.
|
||||||
|
|
||||||
use crate::PrunerError;
|
use crate::PrunerError;
|
||||||
use reth_primitives::BlockNumber;
|
use reth_db::{database::Database, tables};
|
||||||
use tracing::debug;
|
use reth_primitives::{BlockNumber, ChainSpec, PruneCheckpoint, PruneMode, PruneModes, PrunePart};
|
||||||
|
use reth_provider::{BlockReader, DatabaseProviderRW, ProviderFactory, PruneCheckpointWriter};
|
||||||
|
use std::sync::Arc;
|
||||||
|
use tracing::{debug, instrument, trace};
|
||||||
|
|
||||||
/// Result of [Pruner::run] execution
|
/// Result of [Pruner::run] execution
|
||||||
pub type PrunerResult = Result<(), PrunerError>;
|
pub type PrunerResult = Result<(), PrunerError>;
|
||||||
|
|
||||||
/// The pipeline type itself with the result of [Pruner::run]
|
/// The pipeline type itself with the result of [Pruner::run]
|
||||||
pub type PrunerWithResult = (Pruner, PrunerResult);
|
pub type PrunerWithResult<DB> = (Pruner<DB>, PrunerResult);
|
||||||
|
|
||||||
|
pub struct BatchSizes {
|
||||||
|
receipts: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for BatchSizes {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self { receipts: 10000 }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Pruning routine. Main pruning logic happens in [Pruner::run].
|
/// Pruning routine. Main pruning logic happens in [Pruner::run].
|
||||||
pub struct Pruner {
|
pub struct Pruner<DB> {
|
||||||
|
provider_factory: ProviderFactory<DB>,
|
||||||
/// Minimum pruning interval measured in blocks. All prune parts are checked and, if needed,
|
/// Minimum pruning interval measured in blocks. All prune parts are checked and, if needed,
|
||||||
/// pruned, when the chain advances by the specified number of blocks.
|
/// pruned, when the chain advances by the specified number of blocks.
|
||||||
min_block_interval: u64,
|
min_block_interval: u64,
|
||||||
@ -22,17 +36,39 @@ pub struct Pruner {
|
|||||||
/// Last pruned block number. Used in conjunction with `min_block_interval` to determine
|
/// Last pruned block number. Used in conjunction with `min_block_interval` to determine
|
||||||
/// when the pruning needs to be initiated.
|
/// when the pruning needs to be initiated.
|
||||||
last_pruned_block_number: Option<BlockNumber>,
|
last_pruned_block_number: Option<BlockNumber>,
|
||||||
|
modes: PruneModes,
|
||||||
|
batch_sizes: BatchSizes,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Pruner {
|
impl<DB: Database> Pruner<DB> {
|
||||||
/// Creates a new [Pruner].
|
/// Creates a new [Pruner].
|
||||||
pub fn new(min_block_interval: u64, max_prune_depth: u64) -> Self {
|
pub fn new(
|
||||||
Self { min_block_interval, max_prune_depth, last_pruned_block_number: None }
|
db: DB,
|
||||||
|
chain_spec: Arc<ChainSpec>,
|
||||||
|
min_block_interval: u64,
|
||||||
|
max_prune_depth: u64,
|
||||||
|
modes: PruneModes,
|
||||||
|
batch_sizes: BatchSizes,
|
||||||
|
) -> Self {
|
||||||
|
Self {
|
||||||
|
provider_factory: ProviderFactory::new(db, chain_spec),
|
||||||
|
min_block_interval,
|
||||||
|
max_prune_depth,
|
||||||
|
last_pruned_block_number: None,
|
||||||
|
modes,
|
||||||
|
batch_sizes,
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Run the pruner
|
/// Run the pruner
|
||||||
pub fn run(&mut self, tip_block_number: BlockNumber) -> PrunerResult {
|
pub fn run(&mut self, tip_block_number: BlockNumber) -> PrunerResult {
|
||||||
// Pruning logic
|
let provider = self.provider_factory.provider_rw()?;
|
||||||
|
|
||||||
|
if let Some((to_block, prune_mode)) = self.modes.prune_to_block_receipts(tip_block_number) {
|
||||||
|
self.prune_receipts(&provider, to_block, prune_mode)?;
|
||||||
|
}
|
||||||
|
|
||||||
|
provider.commit()?;
|
||||||
|
|
||||||
self.last_pruned_block_number = Some(tip_block_number);
|
self.last_pruned_block_number = Some(tip_block_number);
|
||||||
Ok(())
|
Ok(())
|
||||||
@ -58,15 +94,62 @@ impl Pruner {
|
|||||||
false
|
false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Prune receipts up to the provided block, inclusive.
|
||||||
|
#[instrument(level = "trace", skip(self, provider), target = "pruner")]
|
||||||
|
fn prune_receipts(
|
||||||
|
&self,
|
||||||
|
provider: &DatabaseProviderRW<'_, DB>,
|
||||||
|
to_block: BlockNumber,
|
||||||
|
prune_mode: PruneMode,
|
||||||
|
) -> PrunerResult {
|
||||||
|
let to_block_body = match provider.block_body_indices(to_block)? {
|
||||||
|
Some(body) => body,
|
||||||
|
None => {
|
||||||
|
trace!(target: "pruner", "No receipts to prune");
|
||||||
|
return Ok(())
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
provider.prune_table_in_batches::<tables::Receipts, _, _>(
|
||||||
|
..=to_block_body.last_tx_num(),
|
||||||
|
self.batch_sizes.receipts,
|
||||||
|
|receipts| {
|
||||||
|
trace!(
|
||||||
|
target: "pruner",
|
||||||
|
%receipts,
|
||||||
|
"Pruned receipts"
|
||||||
|
);
|
||||||
|
},
|
||||||
|
)?;
|
||||||
|
|
||||||
|
provider.save_prune_checkpoint(
|
||||||
|
PrunePart::Receipts,
|
||||||
|
PruneCheckpoint { block_number: to_block, prune_mode },
|
||||||
|
)?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use crate::Pruner;
|
use crate::{pruner::BatchSizes, Pruner};
|
||||||
|
use assert_matches::assert_matches;
|
||||||
|
use reth_db::{tables, test_utils::create_test_rw_db};
|
||||||
|
use reth_interfaces::test_utils::{
|
||||||
|
generators,
|
||||||
|
generators::{random_block_range, random_receipt},
|
||||||
|
};
|
||||||
|
use reth_primitives::{PruneCheckpoint, PruneMode, PruneModes, PrunePart, H256, MAINNET};
|
||||||
|
use reth_provider::PruneCheckpointReader;
|
||||||
|
use reth_stages::test_utils::TestTransaction;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn pruner_is_pruning_needed() {
|
fn is_pruning_needed() {
|
||||||
let pruner = Pruner::new(5, 0);
|
let db = create_test_rw_db();
|
||||||
|
let pruner =
|
||||||
|
Pruner::new(db, MAINNET.clone(), 5, 0, PruneModes::default(), BatchSizes::default());
|
||||||
|
|
||||||
// No last pruned block number was set before
|
// No last pruned block number was set before
|
||||||
let first_block_number = 1;
|
let first_block_number = 1;
|
||||||
@ -80,4 +163,61 @@ mod tests {
|
|||||||
let third_block_number = second_block_number;
|
let third_block_number = second_block_number;
|
||||||
assert!(pruner.is_pruning_needed(third_block_number));
|
assert!(pruner.is_pruning_needed(third_block_number));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn prune_receipts() {
|
||||||
|
let tx = TestTransaction::default();
|
||||||
|
let mut rng = generators::rng();
|
||||||
|
|
||||||
|
let blocks = random_block_range(&mut rng, 0..=100, H256::zero(), 0..10);
|
||||||
|
tx.insert_blocks(blocks.iter(), None).expect("insert blocks");
|
||||||
|
|
||||||
|
let mut receipts = Vec::new();
|
||||||
|
for block in &blocks {
|
||||||
|
for transaction in &block.body {
|
||||||
|
receipts
|
||||||
|
.push((receipts.len() as u64, random_receipt(&mut rng, transaction, Some(0))));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tx.insert_receipts(receipts).expect("insert receipts");
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
tx.table::<tables::Transactions>().unwrap().len(),
|
||||||
|
blocks.iter().map(|block| block.body.len()).sum::<usize>()
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
tx.table::<tables::Transactions>().unwrap().len(),
|
||||||
|
tx.table::<tables::Receipts>().unwrap().len()
|
||||||
|
);
|
||||||
|
|
||||||
|
let prune_to_block = 10;
|
||||||
|
let prune_mode = PruneMode::Before(prune_to_block);
|
||||||
|
let pruner = Pruner::new(
|
||||||
|
tx.inner_raw(),
|
||||||
|
MAINNET.clone(),
|
||||||
|
5,
|
||||||
|
0,
|
||||||
|
PruneModes { receipts: Some(prune_mode), ..Default::default() },
|
||||||
|
BatchSizes {
|
||||||
|
// Less than total amount of blocks to prune to test the batching logic
|
||||||
|
receipts: 10,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
let provider = tx.inner_rw();
|
||||||
|
assert_matches!(pruner.prune_receipts(&provider, prune_to_block, prune_mode), Ok(()));
|
||||||
|
provider.commit().expect("commit");
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
tx.table::<tables::Receipts>().unwrap().len(),
|
||||||
|
blocks[prune_to_block as usize + 1..]
|
||||||
|
.iter()
|
||||||
|
.map(|block| block.body.len())
|
||||||
|
.sum::<usize>()
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
tx.inner().get_prune_checkpoint(PrunePart::Receipts).unwrap(),
|
||||||
|
Some(PruneCheckpoint { block_number: prune_to_block, prune_mode })
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -70,7 +70,7 @@ criterion = { version = "0.5", features = ["async_futures"] }
|
|||||||
serde_json = { workspace = true }
|
serde_json = { workspace = true }
|
||||||
|
|
||||||
[features]
|
[features]
|
||||||
test-utils = []
|
test-utils = ["reth-interfaces/test-utils"]
|
||||||
|
|
||||||
[[bench]]
|
[[bench]]
|
||||||
name = "criterion"
|
name = "criterion"
|
||||||
|
|||||||
@ -14,7 +14,7 @@ use reth_primitives::{
|
|||||||
stage::{
|
stage::{
|
||||||
CheckpointBlockRange, EntitiesCheckpoint, ExecutionCheckpoint, StageCheckpoint, StageId,
|
CheckpointBlockRange, EntitiesCheckpoint, ExecutionCheckpoint, StageCheckpoint, StageId,
|
||||||
},
|
},
|
||||||
BlockNumber, Header, PruneTargets, U256,
|
BlockNumber, Header, PruneModes, U256,
|
||||||
};
|
};
|
||||||
use reth_provider::{
|
use reth_provider::{
|
||||||
post_state::PostState, BlockExecutor, BlockReader, DatabaseProviderRW, ExecutorFactory,
|
post_state::PostState, BlockExecutor, BlockReader, DatabaseProviderRW, ExecutorFactory,
|
||||||
@ -60,7 +60,7 @@ pub struct ExecutionStage<EF: ExecutorFactory> {
|
|||||||
/// The commit thresholds of the execution stage.
|
/// The commit thresholds of the execution stage.
|
||||||
thresholds: ExecutionStageThresholds,
|
thresholds: ExecutionStageThresholds,
|
||||||
/// Pruning configuration.
|
/// Pruning configuration.
|
||||||
prune_targets: PruneTargets,
|
prune_targets: PruneModes,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<EF: ExecutorFactory> ExecutionStage<EF> {
|
impl<EF: ExecutorFactory> ExecutionStage<EF> {
|
||||||
@ -68,7 +68,7 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
|
|||||||
pub fn new(
|
pub fn new(
|
||||||
executor_factory: EF,
|
executor_factory: EF,
|
||||||
thresholds: ExecutionStageThresholds,
|
thresholds: ExecutionStageThresholds,
|
||||||
prune_targets: PruneTargets,
|
prune_targets: PruneModes,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self { metrics_tx: None, executor_factory, thresholds, prune_targets }
|
Self { metrics_tx: None, executor_factory, thresholds, prune_targets }
|
||||||
}
|
}
|
||||||
@ -77,7 +77,7 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
|
|||||||
///
|
///
|
||||||
/// The commit threshold will be set to 10_000.
|
/// The commit threshold will be set to 10_000.
|
||||||
pub fn new_with_factory(executor_factory: EF) -> Self {
|
pub fn new_with_factory(executor_factory: EF) -> Self {
|
||||||
Self::new(executor_factory, ExecutionStageThresholds::default(), PruneTargets::default())
|
Self::new(executor_factory, ExecutionStageThresholds::default(), PruneModes::default())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Set the metric events sender.
|
/// Set the metric events sender.
|
||||||
@ -425,7 +425,7 @@ mod tests {
|
|||||||
use reth_db::{models::AccountBeforeTx, test_utils::create_test_rw_db};
|
use reth_db::{models::AccountBeforeTx, test_utils::create_test_rw_db};
|
||||||
use reth_primitives::{
|
use reth_primitives::{
|
||||||
hex_literal::hex, keccak256, stage::StageUnitCheckpoint, Account, Bytecode,
|
hex_literal::hex, keccak256, stage::StageUnitCheckpoint, Account, Bytecode,
|
||||||
ChainSpecBuilder, PruneMode, PruneTargets, SealedBlock, StorageEntry, H160, H256, MAINNET,
|
ChainSpecBuilder, PruneMode, PruneModes, SealedBlock, StorageEntry, H160, H256, MAINNET,
|
||||||
U256,
|
U256,
|
||||||
};
|
};
|
||||||
use reth_provider::{AccountReader, BlockWriter, ProviderFactory, ReceiptProvider};
|
use reth_provider::{AccountReader, BlockWriter, ProviderFactory, ReceiptProvider};
|
||||||
@ -439,7 +439,7 @@ mod tests {
|
|||||||
ExecutionStage::new(
|
ExecutionStage::new(
|
||||||
factory,
|
factory,
|
||||||
ExecutionStageThresholds { max_blocks: Some(100), max_changes: None },
|
ExecutionStageThresholds { max_blocks: Some(100), max_changes: None },
|
||||||
PruneTargets::none(),
|
PruneModes::none(),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -943,7 +943,7 @@ mod tests {
|
|||||||
provider.commit().unwrap();
|
provider.commit().unwrap();
|
||||||
|
|
||||||
let check_pruning = |factory: Arc<ProviderFactory<_>>,
|
let check_pruning = |factory: Arc<ProviderFactory<_>>,
|
||||||
prune_targets: PruneTargets,
|
prune_targets: PruneModes,
|
||||||
expect_num_receipts: usize| async move {
|
expect_num_receipts: usize| async move {
|
||||||
let provider = factory.provider_rw().unwrap();
|
let provider = factory.provider_rw().unwrap();
|
||||||
|
|
||||||
@ -960,7 +960,7 @@ mod tests {
|
|||||||
);
|
);
|
||||||
};
|
};
|
||||||
|
|
||||||
let mut prune = PruneTargets::none();
|
let mut prune = PruneModes::none();
|
||||||
|
|
||||||
check_pruning(factory.clone(), prune, 1).await;
|
check_pruning(factory.clone(), prune, 1).await;
|
||||||
|
|
||||||
|
|||||||
@ -10,8 +10,8 @@ use reth_db::{
|
|||||||
DatabaseEnv, DatabaseError as DbError,
|
DatabaseEnv, DatabaseError as DbError,
|
||||||
};
|
};
|
||||||
use reth_primitives::{
|
use reth_primitives::{
|
||||||
keccak256, Account, Address, BlockNumber, SealedBlock, SealedHeader, StorageEntry, H256,
|
keccak256, Account, Address, BlockNumber, Receipt, SealedBlock, SealedHeader, StorageEntry,
|
||||||
MAINNET, U256,
|
TxNumber, H256, MAINNET, U256,
|
||||||
};
|
};
|
||||||
use reth_provider::{DatabaseProviderRO, DatabaseProviderRW, ProviderFactory};
|
use reth_provider::{DatabaseProviderRO, DatabaseProviderRW, ProviderFactory};
|
||||||
use std::{
|
use std::{
|
||||||
@ -268,6 +268,19 @@ impl TestTransaction {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Insert collection of ([TxNumber], [Receipt]) into the corresponding table.
|
||||||
|
pub fn insert_receipts<I>(&self, receipts: I) -> Result<(), DbError>
|
||||||
|
where
|
||||||
|
I: IntoIterator<Item = (TxNumber, Receipt)>,
|
||||||
|
{
|
||||||
|
self.commit(|tx| {
|
||||||
|
receipts.into_iter().try_for_each(|(tx_num, receipt)| {
|
||||||
|
// Insert into receipts table.
|
||||||
|
tx.put::<tables::Receipts>(tx_num, receipt)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
/// Insert collection of ([Address], [Account]) into corresponding tables.
|
/// Insert collection of ([Address], [Account]) into corresponding tables.
|
||||||
pub fn insert_accounts_and_storages<I, S>(&self, accounts: I) -> Result<(), DbError>
|
pub fn insert_accounts_and_storages<I, S>(&self, accounts: I) -> Result<(), DbError>
|
||||||
where
|
where
|
||||||
|
|||||||
@ -51,7 +51,7 @@ pub enum TableType {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Number of tables that should be present inside database.
|
/// Number of tables that should be present inside database.
|
||||||
pub const NUM_TABLES: usize = 25;
|
pub const NUM_TABLES: usize = 26;
|
||||||
|
|
||||||
/// The general purpose of this is to use with a combination of Tables enum,
|
/// The general purpose of this is to use with a combination of Tables enum,
|
||||||
/// by implementing a `TableViewer` trait you can operate on db tables in an abstract way.
|
/// by implementing a `TableViewer` trait you can operate on db tables in an abstract way.
|
||||||
@ -183,7 +183,8 @@ tables!([
|
|||||||
(StoragesTrie, TableType::DupSort),
|
(StoragesTrie, TableType::DupSort),
|
||||||
(TxSenders, TableType::Table),
|
(TxSenders, TableType::Table),
|
||||||
(SyncStage, TableType::Table),
|
(SyncStage, TableType::Table),
|
||||||
(SyncStageProgress, TableType::Table)
|
(SyncStageProgress, TableType::Table),
|
||||||
|
(PruneCheckpoints, TableType::Table)
|
||||||
]);
|
]);
|
||||||
|
|
||||||
#[macro_export]
|
#[macro_export]
|
||||||
@ -417,7 +418,7 @@ table!(
|
|||||||
|
|
||||||
table!(
|
table!(
|
||||||
/// Stores the highest pruned block number and prune mode of each prune part.
|
/// Stores the highest pruned block number and prune mode of each prune part.
|
||||||
( PruneParts ) PrunePart | PruneCheckpoint
|
( PruneCheckpoints ) PrunePart | PruneCheckpoint
|
||||||
);
|
);
|
||||||
|
|
||||||
/// Alias Types
|
/// Alias Types
|
||||||
@ -459,6 +460,7 @@ mod tests {
|
|||||||
(TableType::Table, TxSenders::const_name()),
|
(TableType::Table, TxSenders::const_name()),
|
||||||
(TableType::Table, SyncStage::const_name()),
|
(TableType::Table, SyncStage::const_name()),
|
||||||
(TableType::Table, SyncStageProgress::const_name()),
|
(TableType::Table, SyncStageProgress::const_name()),
|
||||||
|
(TableType::Table, PruneCheckpoints::const_name()),
|
||||||
];
|
];
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
@ -26,10 +26,10 @@ pub use traits::{
|
|||||||
BlockchainTreePendingStateProvider, CanonChainTracker, CanonStateNotification,
|
BlockchainTreePendingStateProvider, CanonChainTracker, CanonStateNotification,
|
||||||
CanonStateNotificationSender, CanonStateNotifications, CanonStateSubscriptions,
|
CanonStateNotificationSender, CanonStateNotifications, CanonStateSubscriptions,
|
||||||
ChainSpecProvider, ChangeSetReader, EvmEnvProvider, ExecutorFactory, HashingWriter,
|
ChainSpecProvider, ChangeSetReader, EvmEnvProvider, ExecutorFactory, HashingWriter,
|
||||||
HeaderProvider, HistoryWriter, PostStateDataProvider, ReceiptProvider, ReceiptProviderIdExt,
|
HeaderProvider, HistoryWriter, PostStateDataProvider, PruneCheckpointReader,
|
||||||
StageCheckpointReader, StageCheckpointWriter, StateProvider, StateProviderBox,
|
PruneCheckpointWriter, ReceiptProvider, ReceiptProviderIdExt, StageCheckpointReader,
|
||||||
StateProviderFactory, StateRootProvider, StorageReader, TransactionsProvider,
|
StageCheckpointWriter, StateProvider, StateProviderBox, StateProviderFactory,
|
||||||
WithdrawalsProvider,
|
StateRootProvider, StorageReader, TransactionsProvider, WithdrawalsProvider,
|
||||||
};
|
};
|
||||||
|
|
||||||
/// Provider trait implementations.
|
/// Provider trait implementations.
|
||||||
|
|||||||
@ -8,7 +8,7 @@ use reth_db::{
|
|||||||
};
|
};
|
||||||
use reth_primitives::{
|
use reth_primitives::{
|
||||||
bloom::logs_bloom, keccak256, proofs::calculate_receipt_root_ref, Account, Address,
|
bloom::logs_bloom, keccak256, proofs::calculate_receipt_root_ref, Account, Address,
|
||||||
BlockNumber, Bloom, Bytecode, Log, PruneMode, PruneTargets, Receipt, StorageEntry, H256, U256,
|
BlockNumber, Bloom, Bytecode, Log, PruneMode, PruneModes, Receipt, StorageEntry, H256, U256,
|
||||||
};
|
};
|
||||||
use reth_trie::{
|
use reth_trie::{
|
||||||
hashed_cursor::{HashedPostState, HashedPostStateCursorFactory, HashedStorage},
|
hashed_cursor::{HashedPostState, HashedPostStateCursorFactory, HashedStorage},
|
||||||
@ -79,7 +79,7 @@ pub struct PostState {
|
|||||||
/// The receipt(s) of the executed transaction(s).
|
/// The receipt(s) of the executed transaction(s).
|
||||||
receipts: BTreeMap<BlockNumber, Vec<Receipt>>,
|
receipts: BTreeMap<BlockNumber, Vec<Receipt>>,
|
||||||
/// Pruning configuration.
|
/// Pruning configuration.
|
||||||
prune_targets: PruneTargets,
|
prune_targets: PruneModes,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl PostState {
|
impl PostState {
|
||||||
@ -94,7 +94,7 @@ impl PostState {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Add a pruning configuration.
|
/// Add a pruning configuration.
|
||||||
pub fn add_prune_targets(&mut self, prune_targets: PruneTargets) {
|
pub fn add_prune_targets(&mut self, prune_targets: PruneModes) {
|
||||||
self.prune_targets = prune_targets;
|
self.prune_targets = prune_targets;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -2,16 +2,17 @@ use crate::{
|
|||||||
providers::state::{historical::HistoricalStateProvider, latest::LatestStateProvider},
|
providers::state::{historical::HistoricalStateProvider, latest::LatestStateProvider},
|
||||||
traits::{BlockSource, ReceiptProvider},
|
traits::{BlockSource, ReceiptProvider},
|
||||||
BlockHashReader, BlockNumReader, BlockReader, ChainSpecProvider, EvmEnvProvider,
|
BlockHashReader, BlockNumReader, BlockReader, ChainSpecProvider, EvmEnvProvider,
|
||||||
HeaderProvider, ProviderError, StageCheckpointReader, StateProviderBox, TransactionsProvider,
|
HeaderProvider, ProviderError, PruneCheckpointReader, StageCheckpointReader, StateProviderBox,
|
||||||
WithdrawalsProvider,
|
TransactionsProvider, WithdrawalsProvider,
|
||||||
};
|
};
|
||||||
use reth_db::{database::Database, init_db, models::StoredBlockBodyIndices, DatabaseEnv};
|
use reth_db::{database::Database, init_db, models::StoredBlockBodyIndices, DatabaseEnv};
|
||||||
use reth_interfaces::Result;
|
use reth_interfaces::Result;
|
||||||
use reth_primitives::{
|
use reth_primitives::{
|
||||||
stage::{StageCheckpoint, StageId},
|
stage::{StageCheckpoint, StageId},
|
||||||
Address, Block, BlockHash, BlockHashOrNumber, BlockNumber, BlockWithSenders, ChainInfo,
|
Address, Block, BlockHash, BlockHashOrNumber, BlockNumber, BlockWithSenders, ChainInfo,
|
||||||
ChainSpec, Header, Receipt, SealedBlock, SealedHeader, TransactionMeta, TransactionSigned,
|
ChainSpec, Header, PruneCheckpoint, PrunePart, Receipt, SealedBlock, SealedHeader,
|
||||||
TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, H256, U256,
|
TransactionMeta, TransactionSigned, TransactionSignedNoHash, TxHash, TxNumber, Withdrawal,
|
||||||
|
H256, U256,
|
||||||
};
|
};
|
||||||
use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
|
use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
|
||||||
use std::{ops::RangeBounds, sync::Arc};
|
use std::{ops::RangeBounds, sync::Arc};
|
||||||
@ -360,6 +361,12 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<DB: Database> PruneCheckpointReader for ProviderFactory<DB> {
|
||||||
|
fn get_prune_checkpoint(&self, part: PrunePart) -> Result<Option<PruneCheckpoint>> {
|
||||||
|
self.provider()?.get_prune_checkpoint(part)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::ProviderFactory;
|
use super::ProviderFactory;
|
||||||
|
|||||||
@ -5,7 +5,8 @@ use crate::{
|
|||||||
},
|
},
|
||||||
AccountReader, BlockExecutionWriter, BlockHashReader, BlockNumReader, BlockReader, BlockWriter,
|
AccountReader, BlockExecutionWriter, BlockHashReader, BlockNumReader, BlockReader, BlockWriter,
|
||||||
EvmEnvProvider, HashingWriter, HeaderProvider, HistoryWriter, PostState, ProviderError,
|
EvmEnvProvider, HashingWriter, HeaderProvider, HistoryWriter, PostState, ProviderError,
|
||||||
StageCheckpointReader, StorageReader, TransactionsProvider, WithdrawalsProvider,
|
PruneCheckpointReader, PruneCheckpointWriter, StageCheckpointReader, StorageReader,
|
||||||
|
TransactionsProvider, WithdrawalsProvider,
|
||||||
};
|
};
|
||||||
use itertools::{izip, Itertools};
|
use itertools::{izip, Itertools};
|
||||||
use reth_db::{
|
use reth_db::{
|
||||||
@ -16,7 +17,7 @@ use reth_db::{
|
|||||||
sharded_key, storage_sharded_key::StorageShardedKey, AccountBeforeTx, BlockNumberAddress,
|
sharded_key, storage_sharded_key::StorageShardedKey, AccountBeforeTx, BlockNumberAddress,
|
||||||
ShardedKey, StoredBlockBodyIndices, StoredBlockOmmers, StoredBlockWithdrawals,
|
ShardedKey, StoredBlockBodyIndices, StoredBlockOmmers, StoredBlockWithdrawals,
|
||||||
},
|
},
|
||||||
table::Table,
|
table::{Key, Table},
|
||||||
tables,
|
tables,
|
||||||
transaction::{DbTx, DbTxMut},
|
transaction::{DbTx, DbTxMut},
|
||||||
BlockNumberList, DatabaseError,
|
BlockNumberList, DatabaseError,
|
||||||
@ -26,9 +27,10 @@ use reth_primitives::{
|
|||||||
keccak256,
|
keccak256,
|
||||||
stage::{StageCheckpoint, StageId},
|
stage::{StageCheckpoint, StageId},
|
||||||
Account, Address, Block, BlockHash, BlockHashOrNumber, BlockNumber, BlockWithSenders,
|
Account, Address, Block, BlockHash, BlockHashOrNumber, BlockNumber, BlockWithSenders,
|
||||||
ChainInfo, ChainSpec, Hardfork, Head, Header, Receipt, SealedBlock, SealedBlockWithSenders,
|
ChainInfo, ChainSpec, Hardfork, Head, Header, PruneCheckpoint, PrunePart, Receipt, SealedBlock,
|
||||||
SealedHeader, StorageEntry, TransactionMeta, TransactionSigned, TransactionSignedEcRecovered,
|
SealedBlockWithSenders, SealedHeader, StorageEntry, TransactionMeta, TransactionSigned,
|
||||||
TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, H256, U256,
|
TransactionSignedEcRecovered, TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, H256,
|
||||||
|
U256,
|
||||||
};
|
};
|
||||||
use reth_revm_primitives::{
|
use reth_revm_primitives::{
|
||||||
config::revm_spec,
|
config::revm_spec,
|
||||||
@ -617,6 +619,54 @@ impl<'this, TX: DbTxMut<'this> + DbTx<'this>> DatabaseProvider<'this, TX> {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Prune the table for the specified key range.
|
||||||
|
/// Returns number of rows pruned.
|
||||||
|
pub fn prune_table<T, K>(
|
||||||
|
&self,
|
||||||
|
range: impl RangeBounds<K>,
|
||||||
|
) -> std::result::Result<usize, DatabaseError>
|
||||||
|
where
|
||||||
|
T: Table<Key = K>,
|
||||||
|
K: Key,
|
||||||
|
{
|
||||||
|
self.prune_table_in_batches::<T, K, _>(range, usize::MAX, |_| {})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prune the table for the specified key range calling `chunk_callback` after every
|
||||||
|
/// `batch_size` pruned rows.
|
||||||
|
///
|
||||||
|
/// Returns number of rows pruned.
|
||||||
|
pub fn prune_table_in_batches<T, K, F>(
|
||||||
|
&self,
|
||||||
|
range: impl RangeBounds<K>,
|
||||||
|
batch_size: usize,
|
||||||
|
batch_callback: F,
|
||||||
|
) -> std::result::Result<usize, DatabaseError>
|
||||||
|
where
|
||||||
|
T: Table<Key = K>,
|
||||||
|
K: Key,
|
||||||
|
F: Fn(usize),
|
||||||
|
{
|
||||||
|
let mut cursor = self.tx.cursor_write::<T>()?;
|
||||||
|
let mut walker = cursor.walk_range(range)?;
|
||||||
|
let mut deleted = 0;
|
||||||
|
|
||||||
|
while let Some(Ok(_)) = walker.next() {
|
||||||
|
walker.delete_current()?;
|
||||||
|
deleted += 1;
|
||||||
|
|
||||||
|
if deleted % batch_size == 0 {
|
||||||
|
batch_callback(batch_size);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if deleted % batch_size != 0 {
|
||||||
|
batch_callback(deleted % batch_size);
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(deleted)
|
||||||
|
}
|
||||||
|
|
||||||
/// Load shard and remove it. If list is empty, last shard was full or
|
/// Load shard and remove it. If list is empty, last shard was full or
|
||||||
/// there are no shards at all.
|
/// there are no shards at all.
|
||||||
fn take_shard<T>(&self, key: T::Key) -> Result<Vec<u64>>
|
fn take_shard<T>(&self, key: T::Key) -> Result<Vec<u64>>
|
||||||
@ -1816,3 +1866,15 @@ impl<'this, TX: DbTxMut<'this> + DbTx<'this>> BlockWriter for DatabaseProvider<'
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<'this, TX: DbTx<'this>> PruneCheckpointReader for DatabaseProvider<'this, TX> {
|
||||||
|
fn get_prune_checkpoint(&self, part: PrunePart) -> Result<Option<PruneCheckpoint>> {
|
||||||
|
Ok(self.tx.get::<tables::PruneCheckpoints>(part)?)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<'this, TX: DbTxMut<'this>> PruneCheckpointWriter for DatabaseProvider<'this, TX> {
|
||||||
|
fn save_prune_checkpoint(&self, part: PrunePart, checkpoint: PruneCheckpoint) -> Result<()> {
|
||||||
|
Ok(self.tx.put::<tables::PruneCheckpoints>(part, checkpoint)?)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -2,9 +2,9 @@ use crate::{
|
|||||||
BlockHashReader, BlockIdReader, BlockNumReader, BlockReader, BlockReaderIdExt,
|
BlockHashReader, BlockIdReader, BlockNumReader, BlockReader, BlockReaderIdExt,
|
||||||
BlockchainTreePendingStateProvider, CanonChainTracker, CanonStateNotifications,
|
BlockchainTreePendingStateProvider, CanonChainTracker, CanonStateNotifications,
|
||||||
CanonStateSubscriptions, ChainSpecProvider, ChangeSetReader, EvmEnvProvider, HeaderProvider,
|
CanonStateSubscriptions, ChainSpecProvider, ChangeSetReader, EvmEnvProvider, HeaderProvider,
|
||||||
PostStateDataProvider, ProviderError, ReceiptProvider, ReceiptProviderIdExt,
|
PostStateDataProvider, ProviderError, PruneCheckpointReader, ReceiptProvider,
|
||||||
StageCheckpointReader, StateProviderBox, StateProviderFactory, TransactionsProvider,
|
ReceiptProviderIdExt, StageCheckpointReader, StateProviderBox, StateProviderFactory,
|
||||||
WithdrawalsProvider,
|
TransactionsProvider, WithdrawalsProvider,
|
||||||
};
|
};
|
||||||
use reth_db::{database::Database, models::StoredBlockBodyIndices};
|
use reth_db::{database::Database, models::StoredBlockBodyIndices};
|
||||||
use reth_interfaces::{
|
use reth_interfaces::{
|
||||||
@ -15,8 +15,8 @@ use reth_interfaces::{
|
|||||||
use reth_primitives::{
|
use reth_primitives::{
|
||||||
stage::{StageCheckpoint, StageId},
|
stage::{StageCheckpoint, StageId},
|
||||||
Address, Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumHash, BlockNumber,
|
Address, Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumHash, BlockNumber,
|
||||||
BlockNumberOrTag, BlockWithSenders, ChainInfo, ChainSpec, Header, Receipt, SealedBlock,
|
BlockNumberOrTag, BlockWithSenders, ChainInfo, ChainSpec, Header, PruneCheckpoint, PrunePart,
|
||||||
SealedBlockWithSenders, SealedHeader, TransactionMeta, TransactionSigned,
|
Receipt, SealedBlock, SealedBlockWithSenders, SealedHeader, TransactionMeta, TransactionSigned,
|
||||||
TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, H256, U256,
|
TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, H256, U256,
|
||||||
};
|
};
|
||||||
use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
|
use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
|
||||||
@ -438,6 +438,16 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<DB, Tree> PruneCheckpointReader for BlockchainProvider<DB, Tree>
|
||||||
|
where
|
||||||
|
DB: Database,
|
||||||
|
Tree: Send + Sync,
|
||||||
|
{
|
||||||
|
fn get_prune_checkpoint(&self, part: PrunePart) -> Result<Option<PruneCheckpoint>> {
|
||||||
|
self.database.provider()?.get_prune_checkpoint(part)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl<DB, Tree> ChainSpecProvider for BlockchainProvider<DB, Tree>
|
impl<DB, Tree> ChainSpecProvider for BlockchainProvider<DB, Tree>
|
||||||
where
|
where
|
||||||
DB: Send + Sync,
|
DB: Send + Sync,
|
||||||
|
|||||||
@ -2,17 +2,18 @@ use crate::{
|
|||||||
traits::{BlockSource, ReceiptProvider},
|
traits::{BlockSource, ReceiptProvider},
|
||||||
AccountReader, BlockHashReader, BlockIdReader, BlockNumReader, BlockReader, BlockReaderIdExt,
|
AccountReader, BlockHashReader, BlockIdReader, BlockNumReader, BlockReader, BlockReaderIdExt,
|
||||||
ChainSpecProvider, ChangeSetReader, EvmEnvProvider, HeaderProvider, PostState,
|
ChainSpecProvider, ChangeSetReader, EvmEnvProvider, HeaderProvider, PostState,
|
||||||
ReceiptProviderIdExt, StageCheckpointReader, StateProvider, StateProviderBox,
|
PruneCheckpointReader, ReceiptProviderIdExt, StageCheckpointReader, StateProvider,
|
||||||
StateProviderFactory, StateRootProvider, TransactionsProvider, WithdrawalsProvider,
|
StateProviderBox, StateProviderFactory, StateRootProvider, TransactionsProvider,
|
||||||
|
WithdrawalsProvider,
|
||||||
};
|
};
|
||||||
use reth_db::models::{AccountBeforeTx, StoredBlockBodyIndices};
|
use reth_db::models::{AccountBeforeTx, StoredBlockBodyIndices};
|
||||||
use reth_interfaces::Result;
|
use reth_interfaces::Result;
|
||||||
use reth_primitives::{
|
use reth_primitives::{
|
||||||
stage::{StageCheckpoint, StageId},
|
stage::{StageCheckpoint, StageId},
|
||||||
Account, Address, Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumber, Bytecode, Bytes,
|
Account, Address, Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumber, Bytecode, Bytes,
|
||||||
ChainInfo, ChainSpec, Header, Receipt, SealedBlock, SealedHeader, StorageKey, StorageValue,
|
ChainInfo, ChainSpec, Header, PruneCheckpoint, PrunePart, Receipt, SealedBlock, SealedHeader,
|
||||||
TransactionMeta, TransactionSigned, TransactionSignedNoHash, TxHash, TxNumber, H256,
|
StorageKey, StorageValue, TransactionMeta, TransactionSigned, TransactionSignedNoHash, TxHash,
|
||||||
KECCAK_EMPTY, MAINNET, U256,
|
TxNumber, H256, KECCAK_EMPTY, MAINNET, U256,
|
||||||
};
|
};
|
||||||
use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
|
use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
|
||||||
use std::{ops::RangeBounds, sync::Arc};
|
use std::{ops::RangeBounds, sync::Arc};
|
||||||
@ -360,3 +361,9 @@ impl WithdrawalsProvider for NoopProvider {
|
|||||||
Ok(None)
|
Ok(None)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl PruneCheckpointReader for NoopProvider {
|
||||||
|
fn get_prune_checkpoint(&self, _part: PrunePart) -> Result<Option<PruneCheckpoint>> {
|
||||||
|
Ok(None)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -59,3 +59,6 @@ pub use hashing::HashingWriter;
|
|||||||
|
|
||||||
mod history;
|
mod history;
|
||||||
pub use history::HistoryWriter;
|
pub use history::HistoryWriter;
|
||||||
|
|
||||||
|
mod prune_checkpoint;
|
||||||
|
pub use prune_checkpoint::{PruneCheckpointReader, PruneCheckpointWriter};
|
||||||
|
|||||||
16
crates/storage/provider/src/traits/prune_checkpoint.rs
Normal file
16
crates/storage/provider/src/traits/prune_checkpoint.rs
Normal file
@ -0,0 +1,16 @@
|
|||||||
|
use reth_interfaces::Result;
|
||||||
|
use reth_primitives::{PruneCheckpoint, PrunePart};
|
||||||
|
|
||||||
|
/// The trait for fetching prune checkpoint related data.
|
||||||
|
#[auto_impl::auto_impl(&, Arc)]
|
||||||
|
pub trait PruneCheckpointReader: Send + Sync {
|
||||||
|
/// Fetch the checkpoint for the given prune part.
|
||||||
|
fn get_prune_checkpoint(&self, part: PrunePart) -> Result<Option<PruneCheckpoint>>;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The trait for updating prune checkpoint related data.
|
||||||
|
#[auto_impl::auto_impl(&, Arc)]
|
||||||
|
pub trait PruneCheckpointWriter: Send + Sync {
|
||||||
|
/// Save prune checkpoint.
|
||||||
|
fn save_prune_checkpoint(&self, part: PrunePart, checkpoint: PruneCheckpoint) -> Result<()>;
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user