refactor: integrate env and cache tasks

Author: prestwich

Cargo.toml

@@ -27,7 +27,6 @@ integration = []
 [dependencies]
 init4-bin-base = "0.3"
 
-signet-bundle = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
 signet-constants = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
 signet-sim = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
 signet-tx-cache = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }

bin/builder.rs

@@ -3,16 +3,17 @@ use builder::{
     service::serve_builder,
     tasks::{
         block::sim::Simulator,
-        cache::{BundlePoller, TxPoller},
+        cache::{BundlePoller, CacheTask, TxPoller},
         metrics::MetricsTask,
         submit::SubmitTask,
     },
 };
-use init4_bin_base::{deps::tracing, utils::from_env::FromEnv};
-use signet_sim::SimCache;
+use init4_bin_base::{
+    deps::tracing::{info, info_span},
+    utils::from_env::FromEnv,
+};
 use signet_types::constants::SignetSystemConstants;
 use tokio::select;
-use tracing::info_span;
 
 // Note: Must be set to `multi_thread` to support async tasks.
 // See: https://docs.rs/tokio/latest/tokio/attr.main.html
@@ -21,74 +22,85 @@ async fn main() -> eyre::Result<()> {
     let _guard = init4_bin_base::init4();
     let init_span_guard = info_span!("builder initialization");
 
+    // Pull the configuration from the environment
     let config = BuilderConfig::from_env()?.clone();
     let constants = SignetSystemConstants::pecorino();
+
+    // Initialize the oauth token
     let token = config.oauth_token();
 
+    // Set up the EnvTask
+    let env_task = config.env_task();
+    let (env_watcher, env_jh) = env_task.spawn();
+
+    // Prep providers and contracts
     let (host_provider, quincey) =
         tokio::try_join!(config.connect_host_provider(), config.connect_quincey())?;
     let ru_provider = config.connect_ru_provider();
-
     let zenith = config.connect_zenith(host_provider.clone());
 
+    // Set up the metrics task
     let metrics = MetricsTask { host_provider };
     let (tx_channel, metrics_jh) = metrics.spawn();
 
+    // Make a Tx submission task
     let submit =
         SubmitTask { zenith, quincey, config: config.clone(), outbound_tx_channel: tx_channel };
 
+    // Tx Poller pulls transactions from the cache
     let tx_poller = TxPoller::new(&config);
     let (tx_receiver, tx_poller_jh) = tx_poller.spawn();
 
+    // Bundle Poller pulls bundles from the cache
     let bundle_poller = BundlePoller::new(&config, token);
     let (bundle_receiver, bundle_poller_jh) = bundle_poller.spawn();
 
-    let (submit_channel, submit_jh) = submit.spawn();
-
-    let sim_items = SimCache::new();
-    let slot_calculator = config.slot_calculator;
+    // Set up the cache task
+    let cache_task = CacheTask::new(env_watcher.clone(), bundle_receiver, tx_receiver);
+    let (sim_cache, cache_jh) = cache_task.spawn();
 
-    let sim = Simulator::new(&config, ru_provider.clone(), slot_calculator);
-
-    let (basefee_jh, sim_cache_jh) =
-        sim.spawn_cache_tasks(tx_receiver, bundle_receiver, sim_items.clone());
+    // Set up tx submission
+    let (submit_channel, submit_jh) = submit.spawn();
 
-    let build_jh = sim.spawn_simulator_task(constants, sim_items.clone(), submit_channel);
+    // Set up the simulator
+    let sim = Simulator::new(&config, ru_provider.clone());
+    let build_jh = sim.spawn_simulator_task(constants, sim_cache, submit_channel);
 
+    // Start the healthcheck server
     let server = serve_builder(([0, 0, 0, 0], config.builder_port));
 
     // We have finished initializing the builder, so we can drop the init span
     // guard.
     drop(init_span_guard);
 
     select! {
+        _ = cache_jh => {
+            info!("cache task finished");
+        },
+        _ = env_jh => {
+            info!("env task finished");
+        },
         _ = tx_poller_jh => {
-            tracing::info!("tx_poller finished");
+            info!("tx_poller finished");
         },
         _ = bundle_poller_jh => {
-            tracing::info!("bundle_poller finished");
+            info!("bundle_poller finished");
         },
-        _ = sim_cache_jh => {
-            tracing::info!("sim cache task finished");
-        }
-        _ = basefee_jh => {
-            tracing::info!("basefee task finished");
-        }
         _ = submit_jh => {
-            tracing::info!("submit finished");
+            info!("submit finished");
         },
         _ = metrics_jh => {
-            tracing::info!("metrics finished");
+            info!("metrics finished");
         },
         _ = build_jh => {
-            tracing::info!("build finished");
+            info!("build finished");
         }
         _ = server => {
-            tracing::info!("server finished");
+            info!("server finished");
         }
     }
 
-    tracing::info!("shutting down");
+    info!("shutting down");
 
     Ok(())
 }

src/config.rs

@@ -1,7 +1,10 @@
 use crate::{
     quincey::Quincey,
     signer::{LocalOrAws, SignerError},
-    tasks::oauth::{Authenticator, SharedToken},
+    tasks::{
+        env::EnvTask,
+        oauth::{Authenticator, SharedToken},
+    },
 };
 use alloy::{
     network::{Ethereum, EthereumWallet},
@@ -184,8 +187,13 @@ impl BuilderConfig {
 
     /// Connect to the Rollup rpc provider.
     pub fn connect_ru_provider(&self) -> RootProvider<Ethereum> {
-        let url = url::Url::parse(&self.ru_rpc_url).expect("failed to parse URL");
-        RootProvider::<Ethereum>::new_http(url)
+        static ONCE: std::sync::OnceLock<RootProvider<Ethereum>> = std::sync::OnceLock::new();
+
+        ONCE.get_or_init(|| {
+            let url = url::Url::parse(&self.ru_rpc_url).expect("failed to parse URL");
+            RootProvider::new_http(url)
+        })
+        .clone()
     }
 
     /// Connect to the Host rpc provider.
@@ -240,4 +248,10 @@ impl BuilderConfig {
 
         Ok(Quincey::new_remote(client, url, token))
     }
+
+    /// Create an [`EnvTask`] using this config.
+    pub fn env_task(&self) -> EnvTask {
+        let provider = self.connect_ru_provider();
+        EnvTask::new(self.clone(), provider)
+    }
 }

src/tasks/block/sim.rs

@@ -4,10 +4,9 @@
 use super::cfg::PecorinoBlockEnv;
 use crate::{
     config::{BuilderConfig, RuProvider},
-    tasks::{block::cfg::PecorinoCfg, cache::Bundle},
+    tasks::block::cfg::PecorinoCfg,
 };
 use alloy::{
-    consensus::TxEnvelope,
     eips::{BlockId, BlockNumberOrTag::Latest},
     network::Ethereum,
     providers::Provider,
@@ -20,18 +19,10 @@ use init4_bin_base::{
 };
 use signet_sim::{BlockBuild, BuiltBlock, SimCache};
 use signet_types::constants::SignetSystemConstants;
-use std::{
-    sync::{
-        Arc,
-        atomic::{AtomicU64, Ordering},
-    },
-    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
-};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tokio::{
-    select,
     sync::mpsc::{self},
     task::JoinHandle,
-    time::sleep,
 };
 use trevm::revm::{
     database::{AlloyDB, WrapDatabaseAsync},
@@ -47,8 +38,6 @@ pub struct Simulator {
     pub config: BuilderConfig,
     /// A provider that cannot sign transactions, used for interacting with the rollup.
     pub ru_provider: RuProvider,
-    /// The slot calculator for determining when to wake up and build blocks.
-    pub slot_calculator: SlotCalculator,
 }
 
 type AlloyDatabaseProvider = WrapDatabaseAsync<AlloyDB<Ethereum, RuProvider>>;
@@ -60,17 +49,17 @@ impl Simulator {
     ///
     /// - `config`: The configuration for the builder.
     /// - `ru_provider`: A provider for interacting with the rollup.
-    /// - `slot_calculator`: A slot calculator for managing block timing.
     ///
     /// # Returns
     ///
     /// A new `Simulator` instance.
-    pub fn new(
-        config: &BuilderConfig,
-        ru_provider: RuProvider,
-        slot_calculator: SlotCalculator,
-    ) -> Self {
-        Self { config: config.clone(), ru_provider, slot_calculator }
+    pub fn new(config: &BuilderConfig, ru_provider: RuProvider) -> Self {
+        Self { config: config.clone(), ru_provider }
+    }
+
+    /// Get the slot calculator.
+    pub const fn slot_calculator(&self) -> &SlotCalculator {
+        &self.config.slot_calculator
     }
 
     /// Handles building a single block.
@@ -110,85 +99,6 @@ impl Simulator {
         Ok(block)
     }
 
-    /// Spawns two tasks: one to handle incoming transactions and bundles,
-    /// adding them to the simulation cache, and one to track the latest basefee.
-    ///
-    /// # Arguments
-    ///
-    /// - `tx_receiver`: A channel receiver for incoming transactions.
-    /// - `bundle_receiver`: A channel receiver for incoming bundles.
-    /// - `cache`: The simulation cache to store the received items.
-    ///
-    /// # Returns
-    ///
-    /// A `JoinHandle` for the basefee updater and a `JoinHandle` for the
-    /// cache handler.
-    pub fn spawn_cache_tasks(
-        &self,
-        tx_receiver: mpsc::UnboundedReceiver<TxEnvelope>,
-        bundle_receiver: mpsc::UnboundedReceiver<Bundle>,
-        cache: SimCache,
-    ) -> (JoinHandle<()>, JoinHandle<()>) {
-        debug!("starting up cache handler");
-
-        let basefee_price = Arc::new(AtomicU64::new(0_u64));
-        let basefee_reader = Arc::clone(&basefee_price);
-        let fut = self.basefee_updater_fut(basefee_price);
-
-        // Update the basefee on a per-block cadence
-        let basefee_jh = tokio::spawn(fut);
-
-        // Update the sim cache whenever a transaction or bundle is received with respect to the basefee
-        let cache_jh = tokio::spawn(async move {
-            cache_updater(tx_receiver, bundle_receiver, cache, basefee_reader).await
-        });
-
-        (basefee_jh, cache_jh)
-    }
-
-    /// Periodically updates the shared basefee by querying the latest block.
-    ///
-    /// This function calculates the remaining time until the next slot,
-    /// sleeps until that time, and then retrieves the latest basefee from the rollup provider.
-    /// The updated basefee is stored in the provided `AtomicU64`.
-    ///
-    /// This function runs continuously.
-    ///
-    /// # Arguments
-    ///
-    /// - `price`: A shared `Arc<AtomicU64>` used to store the updated basefee value.
-    fn basefee_updater_fut(&self, price: Arc<AtomicU64>) -> impl Future<Output = ()> + use<> {
-        let slot_calculator = self.slot_calculator;
-        let ru_provider = self.ru_provider.clone();
-
-        async move {
-            debug!("starting basefee updater");
-            loop {
-                // calculate start of next slot plus a small buffer
-                let time_remaining = slot_calculator.slot_duration()
-                    - slot_calculator.current_timepoint_within_slot()
-                    + 1;
-                debug!(time_remaining = ?time_remaining, "basefee updater sleeping until next slot");
-
-                // wait until that point in time
-                sleep(Duration::from_secs(time_remaining)).await;
-
-                // update the basefee with that price
-                let resp = ru_provider.get_block_by_number(Latest).await.inspect_err(|e| {
-                    error!(error = %e, "RPC error during basefee update");
-                });
-
-                if let Ok(Some(block)) = resp {
-                    let basefee = block.header.base_fee_per_gas.unwrap_or(0);
-                    price.store(basefee, Ordering::Relaxed);
-                    debug!(basefee = basefee, "basefee updated");
-                } else {
-                    warn!("get basefee failed - an error likely occurred");
-                }
-            }
-        }
-    }
-
     /// Spawns the simulator task, which handles the setup and sets the deadline
     /// for the each round of simulation.
     ///
@@ -269,7 +179,7 @@ impl Simulator {
         let now = SystemTime::now();
         let unix_seconds = now.duration_since(UNIX_EPOCH).expect("Time went backwards").as_secs();
         // Calculate the time remaining in the current slot
-        let remaining = self.slot_calculator.calculate_timepoint_within_slot(unix_seconds);
+        let remaining = self.slot_calculator().calculate_timepoint_within_slot(unix_seconds);
         //  Deadline is equal to the start of the next slot plus the time remaining in this slot
         Instant::now() + Duration::from_secs(remaining)
     }
@@ -365,41 +275,3 @@ impl Simulator {
         Ok(block.header.base_fee_per_gas)
     }
 }
-
-/// Continuously updates the simulation cache with incoming transactions and bundles.
-///
-/// This function listens for new transactions and bundles on their respective
-/// channels and adds them to the simulation cache using the latest observed basefee.
-///
-/// # Arguments
-///
-/// - `tx_receiver`: A receiver channel for incoming Ethereum transactions.
-/// - `bundle_receiver`: A receiver channel for incoming transaction bundles.
-/// - `cache`: The simulation cache used to store transactions and bundles.
-/// - `price_reader`: An `Arc<AtomicU64>` providing the latest basefee for simulation pricing.
-async fn cache_updater(
-    mut tx_receiver: mpsc::UnboundedReceiver<
-        alloy::consensus::EthereumTxEnvelope<alloy::consensus::TxEip4844Variant>,
-    >,
-    mut bundle_receiver: mpsc::UnboundedReceiver<Bundle>,
-    cache: SimCache,
-    price_reader: Arc<AtomicU64>,
-) -> ! {
-    loop {
-        let p = price_reader.load(Ordering::Relaxed);
-        select! {
-            maybe_tx = tx_receiver.recv() => {
-                if let Some(tx) = maybe_tx {
-                    debug!(tx = ?tx.hash(), "received transaction");
-                    cache.add_item(tx, p);
-                }
-            }
-            maybe_bundle = bundle_receiver.recv() => {
-                if let Some(bundle) = maybe_bundle {
-                    debug!(bundle = ?bundle.id, "received bundle");
-                    cache.add_item(bundle.bundle, p);
-                }
-            }
-        }
-    }
-}