feat(bin, engine, prune): spawn pruning task from the engine (paradig…

…mxyz#3566)

Cargo.toml

@@ -21,6 +21,7 @@ members = [
     "crates/net/downloaders",
     "crates/payload/basic",
     "crates/primitives",
+    "crates/prune",
     "crates/revm",
     "crates/revm/revm-primitives",
     "crates/revm/revm-inspectors",

bin/reth/Cargo.toml

@@ -36,6 +36,7 @@ reth-payload-builder = { workspace = true }
 reth-basic-payload-builder = { path = "../../crates/payload/basic" }
 reth-discv4 = { path = "../../crates/net/discv4" }
 reth-metrics = { workspace = true }
+reth-prune = { path = "../../crates/prune" }
 jemallocator = { version = "0.5.0", optional = true }
 jemalloc-ctl = { version = "0.5.0", optional = true }
 

bin/reth/src/node/mod.rs

@@ -360,6 +360,11 @@ impl Command {
             None
         };
 
+        let pruner = config.prune.map(|prune_config| {
+            info!(target: "reth::cli", "Pruner initialized");
+            reth_prune::Pruner::new(prune_config.block_interval, tree_config.max_reorg_depth())
+        });
+
         // Configure the consensus engine
         let (beacon_consensus_engine, beacon_engine_handle) = BeaconConsensusEngine::with_channel(
             client,
@@ -374,6 +379,7 @@ impl Command {
             MIN_BLOCKS_FOR_PIPELINE_RUN,
             consensus_engine_tx,
             consensus_engine_rx,
+            pruner,
         )?;
         info!(target: "reth::cli", "Consensus engine initialized");
 

crates/blockchain-tree/src/blockchain_tree.rs

@@ -750,12 +750,21 @@ impl<DB: Database, C: Consensus, EF: ExecutorFactory> BlockchainTree<DB, C, EF>
     ///
     /// This finalizes `last_finalized_block` prior to reading the canonical hashes (using
     /// [`BlockchainTree::finalize_block`]).
-    pub fn restore_canonical_hashes(
+    pub fn restore_canonical_hashes_and_finalize(
         &mut self,
         last_finalized_block: BlockNumber,
     ) -> Result<(), Error> {
         self.finalize_block(last_finalized_block);
 
+        self.restore_canonical_hashes()
+    }
+
+    /// Reads the last `N` canonical hashes from the database and updates the block indices of the
+    /// tree.
+    ///
+    /// `N` is the `max_reorg_depth` plus the number of block hashes needed to satisfy the
+    /// `BLOCKHASH` opcode in the EVM.
+    pub fn restore_canonical_hashes(&mut self) -> Result<(), Error> {
         let num_of_canonical_hashes =
             self.config.max_reorg_depth() + self.config.num_of_additional_canonical_block_hashes();
 
@@ -1578,7 +1587,7 @@ mod tests {
             .assert(&tree);
 
         // update canonical block to b2, this would make b2a be removed
-        assert_eq!(tree.restore_canonical_hashes(12), Ok(()));
+        assert_eq!(tree.restore_canonical_hashes_and_finalize(12), Ok(()));
 
         assert_eq!(tree.is_block_known(block2.num_hash()).unwrap(), Some(BlockStatus::Valid));
 

crates/blockchain-tree/src/config.rs

@@ -1,7 +1,7 @@
 //! Blockchain tree configuration
 
 /// The configuration for the blockchain tree.
-#[derive(Clone, Debug)]
+#[derive(Clone, Copy, Debug)]
 pub struct BlockchainTreeConfig {
     /// Number of blocks after the last finalized block that we are storing.
     ///

crates/blockchain-tree/src/shareable.rs

@@ -66,10 +66,21 @@ impl<DB: Database, C: Consensus, EF: ExecutorFactory> BlockchainTreeEngine
         tree.update_chains_metrics();
     }
 
-    fn restore_canonical_hashes(&self, last_finalized_block: BlockNumber) -> Result<(), Error> {
+    fn restore_canonical_hashes_and_finalize(
+        &self,
+        last_finalized_block: BlockNumber,
+    ) -> Result<(), Error> {
         trace!(target: "blockchain_tree", ?last_finalized_block, "Restoring canonical hashes for last finalized block");
         let mut tree = self.tree.write();
-        let res = tree.restore_canonical_hashes(last_finalized_block);
+        let res = tree.restore_canonical_hashes_and_finalize(last_finalized_block);
+        tree.update_chains_metrics();
+        res
+    }
+
+    fn restore_canonical_hashes(&self) -> Result<(), Error> {
+        trace!(target: "blockchain_tree", "Restoring canonical hashes");
+        let mut tree = self.tree.write();
+        let res = tree.restore_canonical_hashes();
         tree.update_chains_metrics();
         res
     }

crates/consensus/beacon/Cargo.toml

@@ -19,6 +19,7 @@ reth-rpc-types = { workspace = true }
 reth-tasks = { workspace = true }
 reth-payload-builder = { workspace = true }
 reth-metrics = { workspace = true }
+reth-prune = { path = "../../prune" }
 
 # async
 tokio = { workspace = true, features = ["sync"] }

crates/consensus/beacon/src/engine/error.rs

@@ -1,3 +1,4 @@
+use reth_prune::PrunerError;
 use reth_rpc_types::engine::ForkchoiceUpdateError;
 use reth_stages::PipelineError;
 
@@ -16,6 +17,12 @@ pub enum BeaconConsensusEngineError {
     /// Pipeline error.
     #[error(transparent)]
     Pipeline(#[from] Box<PipelineError>),
+    /// Pruner channel closed.
+    #[error("Pruner channel closed")]
+    PrunerChannelClosed,
+    /// Pruner error.
+    #[error(transparent)]
+    Pruner(#[from] PrunerError),
     /// Common error. Wrapper around [reth_interfaces::Error].
     #[error(transparent)]
     Common(#[from] reth_interfaces::Error),

crates/consensus/beacon/src/engine/metrics.rs

@@ -13,6 +13,8 @@ pub(crate) struct EngineMetrics {
     pub(crate) forkchoice_updated_messages: Counter,
     /// The total count of new payload messages received.
     pub(crate) new_payload_messages: Counter,
+    /// The number of times the pruner was run.
+    pub(crate) pruner_runs: Counter,
 }
 
 /// Metrics for the `EngineSyncController`.