refactor: integrate env and cache tasks

Author: prestwich

Cargo.toml

@@ -27,7 +27,6 @@ integration = []
 [dependencies]
 init4-bin-base = { version = "0.3.4", features = ["perms"] }
 
-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" }
@@ -56,6 +55,5 @@ serde_json = "1.0"
 tokio = { version = "1.36.0", features = ["full", "macros", "rt-multi-thread"] }
 
 oauth2 = "5"
-chrono = "0.4.41"
 tokio-stream = "0.1.17"
 url = "2.5.4"

bin/builder.rs

@@ -1,18 +1,14 @@
 use builder::{
     config::BuilderConfig,
     service::serve_builder,
-    tasks::{
-        block::sim::Simulator,
-        cache::{BundlePoller, TxPoller},
-        metrics::MetricsTask,
-        submit::SubmitTask,
-    },
+    tasks::{block::sim::Simulator, metrics::MetricsTask, submit::SubmitTask},
+};
+use init4_bin_base::{
+    deps::tracing::{info, info_span},
+    utils::from_env::FromEnv,
 };
-use init4_bin_base::{deps::tracing, utils::from_env::FromEnv};
-use signet_sim::SimCache;
 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 +17,74 @@ 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();
-    let token = config.oauth_token();
 
+    // Spawn the EnvTask
+    let env_task = config.env_task();
+    let (block_env, env_jh) = env_task.spawn();
+
+    // Spawn the cache system
+    let cache_system = config.spawn_cache_system(block_env.clone());
+
+    // 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 };
 
-    let tx_poller = TxPoller::new(&config);
-    let (tx_receiver, tx_poller_jh) = tx_poller.spawn();
-
-    let bundle_poller = BundlePoller::new(&config, token);
-    let (bundle_receiver, bundle_poller_jh) = bundle_poller.spawn();
-
+    // Set up tx submission
     let (submit_channel, submit_jh) = submit.spawn();
 
-    let sim_items = SimCache::new();
-    let slot_calculator = config.slot_calculator;
-
-    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());
-
-    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(), block_env);
+    let build_jh = sim.spawn_simulator_task(constants, cache_system.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! {
-        _ = tx_poller_jh => {
-            tracing::info!("tx_poller finished");
+
+        _ = env_jh => {
+            info!("env task finished");
         },
-        _ = bundle_poller_jh => {
-            tracing::info!("bundle_poller finished");
+        _ = cache_system.cache_task => {
+            info!("cache task finished");
+        },
+        _ = cache_system.tx_poller => {
+            info!("tx_poller finished");
+        },
+        _ = cache_system.bundle_poller => {
+            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,4 +1,11 @@
-use crate::quincey::Quincey;
+use crate::{
+    quincey::Quincey,
+    tasks::{
+        block::cfg::SignetCfgEnv,
+        cache::{BundlePoller, CacheSystem, CacheTask, TxPoller},
+        env::EnvTask,
+    },
+};
 use alloy::{
     network::{Ethereum, EthereumWallet},
     primitives::Address,
@@ -21,6 +28,8 @@ use init4_bin_base::{
 };
 use signet_zenith::Zenith;
 use std::borrow::Cow;
+use tokio::sync::watch;
+use trevm::revm::context::BlockEnv;
 
 /// Type alias for the provider used to simulate against rollup state.
 pub type RuProvider = RootProvider<Ethereum>;
@@ -166,8 +175,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.
@@ -222,4 +236,36 @@ 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)
+    }
+
+    /// Spawn a new [`CacheSystem`] using this config. This contains the
+    /// joinhandles for [`TxPoller`] and [`BundlePoller`] and [`CacheTask`], as
+    /// well as the [`SimCache`] and the block env watcher.
+    ///
+    /// [`SimCache`]: signet_sim::SimCache
+    pub fn spawn_cache_system(&self, block_env: watch::Receiver<Option<BlockEnv>>) -> CacheSystem {
+        // Tx Poller pulls transactions from the cache
+        let tx_poller = TxPoller::new(self);
+        let (tx_receiver, tx_poller) = tx_poller.spawn();
+
+        // Bundle Poller pulls bundles from the cache
+        let bundle_poller = BundlePoller::new(self, self.oauth_token());
+        let (bundle_receiver, bundle_poller) = bundle_poller.spawn();
+
+        // Set up the cache task
+        let cache_task = CacheTask::new(block_env.clone(), bundle_receiver, tx_receiver);
+        let (sim_cache, cache_task) = cache_task.spawn();
+
+        CacheSystem { cache_task, tx_poller, bundle_poller, sim_cache }
+    }
+
+    /// Create a [`SignetCfgEnv`] using this config.
+    pub const fn cfg_env(&self) -> SignetCfgEnv {
+        SignetCfgEnv { chain_id: self.ru_chain_id }
+    }
 }

src/tasks/block/cfg.rs

@@ -1,21 +1,14 @@
 //! This file implements the [`trevm::Cfg`] and [`trevm::Block`] traits for Pecorino blocks.
-use alloy::primitives::{Address, B256, FixedBytes, U256};
-use trevm::{
-    Block,
-    revm::{
-        context::{BlockEnv, CfgEnv},
-        context_interface::block::BlobExcessGasAndPrice,
-        primitives::hardfork::SpecId,
-    },
-};
-
-use crate::config::BuilderConfig;
+use trevm::revm::{context::CfgEnv, primitives::hardfork::SpecId};
 
 /// PecorinoCfg holds network-level configuration values.
 #[derive(Debug, Clone, Copy)]
-pub struct PecorinoCfg {}
+pub struct SignetCfgEnv {
+    /// The chain ID.
+    pub chain_id: u64,
+}
 
-impl trevm::Cfg for PecorinoCfg {
+impl trevm::Cfg for SignetCfgEnv {
     /// Fills the configuration environment with Pecorino-specific values.
     ///
     /// # Arguments
@@ -24,76 +17,7 @@ impl trevm::Cfg for PecorinoCfg {
     fn fill_cfg_env(&self, cfg_env: &mut CfgEnv) {
         let CfgEnv { chain_id, spec, .. } = cfg_env;
 
-        *chain_id = signet_constants::pecorino::RU_CHAIN_ID;
+        *chain_id = self.chain_id;
         *spec = SpecId::default();
     }
 }
-
-/// PecorinoBlockEnv holds block-level configurations for Pecorino blocks.
-#[derive(Debug, Clone, Copy)]
-pub struct PecorinoBlockEnv {
-    /// The block number for this block.
-    pub number: u64,
-    /// The address the block reward should be sent to.
-    pub beneficiary: Address,
-    /// Timestamp for the block.
-    pub timestamp: u64,
-    /// The gas limit for this block environment.
-    pub gas_limit: u64,
-    /// The basefee to use for calculating gas usage.
-    pub basefee: u64,
-    /// The prevrandao to use for this block.
-    pub prevrandao: Option<FixedBytes<32>>,
-}
-
-/// Implements [`trevm::Block`] for the Pecorino block.
-impl Block for PecorinoBlockEnv {
-    /// Fills the block environment with the Pecorino specific values
-    fn fill_block_env(&self, block_env: &mut BlockEnv) {
-        // Destructure the fields off of the block_env and modify them
-        let BlockEnv {
-            number,
-            beneficiary,
-            timestamp,
-            gas_limit,
-            basefee,
-            difficulty,
-            prevrandao,
-            blob_excess_gas_and_price,
-        } = block_env;
-        *number = self.number;
-        *beneficiary = self.beneficiary;
-        *timestamp = self.timestamp;
-        *gas_limit = self.gas_limit;
-        *basefee = self.basefee;
-        *prevrandao = self.prevrandao;
-
-        // NB: The following fields are set to sane defaults because they
-        // are not supported by the rollup
-        *difficulty = U256::ZERO;
-        *blob_excess_gas_and_price =
-            Some(BlobExcessGasAndPrice { excess_blob_gas: 0, blob_gasprice: 0 });
-    }
-}
-
-impl PecorinoBlockEnv {
-    /// Returns a new PecorinoBlockEnv with the specified values.
-    ///
-    /// # Arguments
-    ///
-    /// - config: The BuilderConfig for the builder.
-    /// - number: The block number of this block, usually the latest block number plus 1,
-    ///   unless simulating blocks in the past.
-    /// - timestamp: The timestamp of the block, typically set to the deadline of the
-    ///   block building task.
-    pub fn new(config: BuilderConfig, number: u64, timestamp: u64, basefee: u64) -> Self {
-        PecorinoBlockEnv {
-            number,
-            beneficiary: config.builder_rewards_address,
-            timestamp,
-            gas_limit: config.rollup_block_gas_limit,
-            basefee,
-            prevrandao: Some(B256::random()),
-        }
-    }
-}