refactors

- improves poll interval handling
- cleans up and updates comments
- removes arc usage

Author: dylanlott

src/tasks/bundler.rs

@@ -1,19 +1,14 @@
-//! Bundler service responsible for managing bundles.
-use std::sync::Arc;
-
-use super::oauth::Authenticator;
-
+//! Bundler service responsible for fetching bundles and sending them to the simulator.
 pub use crate::config::BuilderConfig;
-
+use crate::tasks::oauth::Authenticator;
 use oauth2::TokenResponse;
 use reqwest::Url;
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
 use tokio::task::JoinHandle;
 use zenith_types::ZenithEthBundle;
 
-/// Holds a Signet bundle from the cache that has a unique identifier
-/// and a Zenith bundle
+/// Holds a bundle from the cache with a unique ID and a Zenith bundle.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Bundle {
     /// Cache identifier for the bundle
@@ -22,38 +17,37 @@ pub struct Bundle {
     pub bundle: ZenithEthBundle,
 }
 
-impl PartialEq for Bundle {
-    fn eq(&self, other: &Self) -> bool {
-        self.id == other.id
-    }
-}
-
-impl Eq for Bundle {}
-
 /// Response from the tx-pool containing a list of bundles.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TxPoolBundleResponse {
-    /// Bundle responses are availabel on the bundles property
+    /// Bundle responses are available on the bundles property.
     pub bundles: Vec<Bundle>,
 }
 
-/// The BundlePoller polls the tx-pool for bundles and manages the seen bundles.
+/// The BundlePoller polls the tx-pool for bundles.
 #[derive(Debug, Clone)]
 pub struct BundlePoller {
     /// The builder configuration values.
     pub config: BuilderConfig,
     /// Authentication module that periodically fetches and stores auth tokens.
     pub authenticator: Authenticator,
+    /// Defines the interval at which the bundler polls the tx-pool for bundles.
+    pub poll_interval_ms: u64,
 }
 
-/// Implements a poller for the block builder to pull bundles from the tx cache.
+/// Implements a poller for the block builder to pull bundles from the tx-pool.
 impl BundlePoller {
     /// Creates a new BundlePoller from the provided builder config.
     pub fn new(config: &BuilderConfig, authenticator: Authenticator) -> Self {
-        Self { config: config.clone(), authenticator }
+        Self { config: config.clone(), authenticator, poll_interval_ms: 1000 }
     }
 
-    /// Fetches bundles from the transaction cache and returns the (oldest? random?) bundle in the cache.
+    /// Creates a new BundlePoller from the provided builder config and with the specified poll interval in ms.
+    pub fn new_with_poll_interval_ms(config: &BuilderConfig, authenticator: Authenticator) -> Self {
+        Self { config: config.clone(), authenticator, poll_interval_ms: 1000 }
+    }
+
+    /// Fetches bundles from the transaction cache and returns them.
     pub async fn check_bundle_cache(&mut self) -> eyre::Result<Vec<Bundle>> {
         let bundle_url: Url = Url::parse(&self.config.tx_pool_url)?.join("bundles")?;
         let token = self.authenticator.fetch_oauth_token().await?;
@@ -71,23 +65,31 @@ impl BundlePoller {
         Ok(resp.bundles)
     }
 
-    /// Spawns a task that simply sends out any bundles it ever finds
-    pub fn spawn(mut self) -> (UnboundedReceiver<Arc<Bundle>>, JoinHandle<()>) {
+    /// Spawns a task that sends bundles it finds to its channel sender.
+    pub fn spawn(mut self) -> (UnboundedReceiver<Bundle>, JoinHandle<()>) {
         let (outbound, inbound) = unbounded_channel();
         let jh = tokio::spawn(async move {
             loop {
                 if let Ok(bundles) = self.check_bundle_cache().await {
                     tracing::debug!(count = ?bundles.len(), "found bundles");
-                    for bundle in bundles.iter() {
-                        if let Err(err) = outbound.send(Arc::new(bundle.clone())) {
-                            tracing::error!(err = ?err, "Failed to send bundle");
+                    for bundle in bundles.into_iter() {
+                        if let Err(err) = outbound.send(bundle) {
+                            tracing::error!(err = ?err, "Failed to send bundle - channel full");
                         }
                     }
                 }
-                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+                tokio::time::sleep(tokio::time::Duration::from_millis(self.poll_interval_ms)).await;
             }
         });
 
         (inbound, jh)
     }
 }
+
+impl PartialEq for Bundle {
+    fn eq(&self, other: &Self) -> bool {
+        self.id == other.id
+    }
+}
+
+impl Eq for Bundle {}