refactor: cache system

Author: prestwich

bin/builder.rs

@@ -1,12 +1,7 @@
 use builder::{
     config::BuilderConfig,
     service::serve_builder,
-    tasks::{
-        block::sim::Simulator,
-        cache::{BundlePoller, CacheTask, TxPoller},
-        metrics::MetricsTask,
-        submit::SubmitTask,
-    },
+    tasks::{block::sim::Simulator, metrics::MetricsTask, submit::SubmitTask},
 };
 use init4_bin_base::{
     deps::tracing::{info, info_span},
@@ -26,12 +21,12 @@ async fn main() -> eyre::Result<()> {
     let config = BuilderConfig::from_env()?.clone();
     let constants = SignetSystemConstants::pecorino();
 
-    // Initialize the oauth token
-    let token = config.oauth_token();
-
-    // Set up the EnvTask
+    // Spawn the EnvTask
     let env_task = config.env_task();
-    let (env_watcher, env_jh) = env_task.spawn();
+    let (block_env, env_jh) = env_task.spawn();
+
+    // Spawn the cache system
+    let cache_system = config.spawn_cache_system(block_env);
 
     // Prep providers and contracts
     let (host_provider, quincey) =
@@ -47,24 +42,12 @@ async fn main() -> eyre::Result<()> {
     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();
-
-    // 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();
-
     // Set up tx submission
     let (submit_channel, submit_jh) = submit.spawn();
 
     // Set up the simulator
     let sim = Simulator::new(&config, ru_provider.clone());
-    let build_jh = sim.spawn_simulator_task(constants, sim_cache, submit_channel);
+    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));
@@ -74,16 +57,17 @@ async fn main() -> eyre::Result<()> {
     drop(init_span_guard);
 
     select! {
-        _ = cache_jh => {
-            info!("cache task finished");
-        },
+
         _ = env_jh => {
             info!("env task finished");
         },
-        _ = tx_poller_jh => {
+        _ = cache_system.cache_task => {
+            info!("cache task finished");
+        },
+        _ = cache_system.tx_poller => {
             info!("tx_poller finished");
         },
-        _ = bundle_poller_jh => {
+        _ = cache_system.bundle_poller => {
             info!("bundle_poller finished");
         },
         _ = submit_jh => {

src/config.rs

@@ -2,6 +2,7 @@ use crate::{
     quincey::Quincey,
     signer::{LocalOrAws, SignerError},
     tasks::{
+        cache::{BundlePoller, CacheSystem, CacheTask, TxPoller},
         env::EnvTask,
         oauth::{Authenticator, SharedToken},
     },
@@ -22,6 +23,8 @@ use init4_bin_base::utils::{calc::SlotCalculator, from_env::FromEnv};
 use oauth2::url;
 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>;
@@ -254,4 +257,25 @@ impl BuilderConfig {
         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 }
+    }
 }

src/tasks/cache/mod.rs

@@ -6,3 +6,22 @@ pub use tx::TxPoller;
 
 mod bundle;
 pub use bundle::BundlePoller;
+
+use signet_sim::SimCache;
+use tokio::task::JoinHandle;
+
+/// Cache tasks for the block builder.
+#[derive(Debug)]
+pub struct CacheSystem {
+    /// The cache task.
+    pub cache_task: JoinHandle<()>,
+
+    /// The transaction poller task.
+    pub tx_poller: JoinHandle<()>,
+
+    /// The bundle poller task.
+    pub bundle_poller: JoinHandle<()>,
+
+    /// The sim cache.
+    pub sim_cache: SimCache,
+}