chore(tree, engine, prune, stages, storage): improve logs (#4790)

crates/prune/Cargo.toml

@@ -26,6 +26,7 @@ tracing.workspace = true
 thiserror.workspace = true
 itertools.workspace = true
 rayon.workspace = true
+tokio-stream.workspace = true
 
 [dev-dependencies]
 # reth

crates/prune/src/event.rs

@@ -0,0 +1,14 @@
+use reth_primitives::{BlockNumber, PrunePart};
+use std::{collections::BTreeMap, time::Duration};
+
+/// An event emitted by a [Pruner][crate::Pruner].
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum PrunerEvent {
+    /// Emitted when pruner finished running.
+    Finished {
+        tip_block_number: BlockNumber,
+        elapsed: Duration,
+        done: bool,
+        parts_done: BTreeMap<PrunePart, bool>,
+    },
+}

crates/prune/src/lib.rs

@@ -10,9 +10,11 @@
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 
 mod error;
+mod event;
 mod metrics;
 mod pruner;
 
 use crate::metrics::Metrics;
 pub use error::PrunerError;
+pub use event::PrunerEvent;
 pub use pruner::{Pruner, PrunerResult, PrunerWithResult};

crates/prune/src/pruner.rs

@@ -1,6 +1,6 @@
 //! Support for pruning.
 
-use crate::{Metrics, PrunerError};
+use crate::{Metrics, PrunerError, PrunerEvent};
 use rayon::prelude::*;
 use reth_db::{
     abstraction::cursor::{DbCursorRO, DbCursorRW},
@@ -13,15 +13,16 @@ use reth_db::{
 };
 use reth_interfaces::RethResult;
 use reth_primitives::{
-    BlockNumber, ChainSpec, PruneBatchSizes, PruneCheckpoint, PruneMode, PruneModes, PrunePart,
-    TxNumber, MINIMUM_PRUNING_DISTANCE,
+    listener::EventListeners, BlockNumber, ChainSpec, PruneBatchSizes, PruneCheckpoint, PruneMode,
+    PruneModes, PrunePart, TxNumber, MINIMUM_PRUNING_DISTANCE,
 };
 use reth_provider::{
     BlockReader, DatabaseProviderRW, ProviderFactory, PruneCheckpointReader, PruneCheckpointWriter,
     TransactionsProvider,
 };
-use std::{collections::HashMap, ops::RangeInclusive, sync::Arc, time::Instant};
-use tracing::{debug, error, info, instrument, trace};
+use std::{collections::BTreeMap, ops::RangeInclusive, sync::Arc, time::Instant};
+use tokio_stream::wrappers::UnboundedReceiverStream;
+use tracing::{debug, error, instrument, trace};
 
 /// Result of [Pruner::run] execution.
 ///
@@ -46,6 +47,7 @@ pub struct Pruner<DB> {
     modes: PruneModes,
     /// Maximum entries to prune per block, per prune part.
     batch_sizes: PruneBatchSizes,
+    listeners: EventListeners<PrunerEvent>,
 }
 
 impl<DB: Database> Pruner<DB> {
@@ -64,9 +66,15 @@ impl<DB: Database> Pruner<DB> {
             last_pruned_block_number: None,
             modes,
             batch_sizes,
+            listeners: Default::default(),
         }
     }
 
+    /// Listen for events on the prune.
+    pub fn events(&mut self) -> UnboundedReceiverStream<PrunerEvent> {
+        self.listeners.new_listener()
+    }
+
     /// Run the pruner
     pub fn run(&mut self, tip_block_number: BlockNumber) -> PrunerResult {
         if tip_block_number == 0 {
@@ -83,7 +91,7 @@ impl<DB: Database> Pruner<DB> {
 
         let mut done = true;
 
-        let mut parts_done = HashMap::new();
+        let mut parts_done = BTreeMap::new();
 
         if let Some((to_block, prune_mode)) =
             self.modes.prune_target_block_receipts(tip_block_number)?
@@ -235,7 +243,7 @@ impl<DB: Database> Pruner<DB> {
         let elapsed = start.elapsed();
         self.metrics.duration_seconds.record(elapsed);
 
-        info!(
+        trace!(
             target: "pruner",
             %tip_block_number,
             ?elapsed,
@@ -243,6 +251,14 @@ impl<DB: Database> Pruner<DB> {
             ?parts_done,
             "Pruner finished"
         );
+
+        self.listeners.notify(PrunerEvent::Finished {
+            tip_block_number,
+            elapsed,
+            done,
+            parts_done,
+        });
+
         Ok(done)
     }