update pruner

paradigmxyz · May 10, 2024 · 9d5d81d · 9d5d81d
1 parent 7c78137
commit 9d5d81d
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 4 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/prune/Cargo.toml b/crates/prune/Cargo.toml
@@ -19,6 +19,7 @@ reth-provider.workspace = true
 reth-interfaces.workspace = true
 reth-tokio-util.workspace = true
 reth-config.workspace = true
+reth-network-api.workspace = true
 
 # metrics
 reth-metrics.workspace = true

diff --git a/crates/prune/src/error.rs b/crates/prune/src/error.rs
@@ -1,8 +1,11 @@
+use crate::PrunerEvent;
 use reth_db::DatabaseError;
 use reth_interfaces::RethError;
+use reth_network_api::NetworkError;
 use reth_primitives::PruneSegmentError;
 use reth_provider::ProviderError;
 use thiserror::Error;
+use tokio::sync::broadcast::error::SendError;
 
 #[derive(Error, Debug)]
 pub enum PrunerError {
@@ -34,3 +37,9 @@ impl From<PrunerError> for RethError {
         }
     }
 }
+
+impl From<SendError<PrunerEvent>> for PrunerError {
+    fn from(_: SendError<PrunerEvent>) -> Self {
+        Self::Interface(RethError::Network(NetworkError::ChannelClosed))
+    }
+}
diff --git a/crates/prune/src/pruner.rs b/crates/prune/src/pruner.rs
@@ -19,7 +19,7 @@ use std::{
     time::{Duration, Instant},
 };
 use tokio::sync::watch;
-use tokio_stream::wrappers::UnboundedReceiverStream;
+use tokio_stream::wrappers::BroadcastStream;
 use tracing::debug;
 
 /// Result of [Pruner::run] execution.
@@ -82,7 +82,7 @@ impl<DB: Database> Pruner<DB> {
     }
 
     /// Listen for events on the pruner.
-    pub fn events(&mut self) -> UnboundedReceiverStream<PrunerEvent> {
+    pub fn events(&mut self) -> BroadcastStream<PrunerEvent> {
         self.listeners.new_listener()
     }
 
@@ -100,7 +100,7 @@ impl<DB: Database> Pruner<DB> {
             return Ok(PruneProgress::Finished)
         }
 
-        self.listeners.notify(PrunerEvent::Started { tip_block_number });
+        self.listeners.notify(PrunerEvent::Started { tip_block_number })?;
 
         debug!(target: "pruner", %tip_block_number, "Pruner started");
         let start = Instant::now();
@@ -154,7 +154,7 @@ impl<DB: Database> Pruner<DB> {
             "{message}",
         );
 
-        self.listeners.notify(PrunerEvent::Finished { tip_block_number, elapsed, stats });
+        self.listeners.notify(PrunerEvent::Finished { tip_block_number, elapsed, stats })?;
 
         Ok(progress)
     }