[feat] adds simulation to block loop (#71)

# [feat] adds simulation to block loop

This PR adds simulation to the block builder loop.

- Updates the builder to use a `SlotCalculator` for managing slot timings. 
- Removes the tx and bundle poller from the block builder and instead leans into the actor model and feeds them in via channels to the simulation process.
- This PR relies on the [signet-sdk Sim Crate PR](https://github.com/init4tech/signet-sdk/pull/33/files)

## Configuration

Deploying this PR will require additional environment variables to be set 

- `CONCURRENCY_LIMIT` to limit how many concurrent simulations the block builder runs. 
- `START_TIMESTAMP` to set the starting timestamp of the chain for anchoring slot calculation.

##  Testing

This code is still being tested before final merge.

The current test plan is to

- Build a Docker image of the builder with this branch
- Update environment variables to include the above values
- Deploy the test image to Pecorino
- Add a bundle to the cache and see if it is picked up
- See what, if any, errors spring up
- Land blocks and see that the block building rate is not decreased with new timing updates

Author: dylanlott

.gitignore

@@ -23,3 +23,6 @@ Cargo.lock
 # Added by cargo
 
 /target
+
+# VSCode debug launcher
+.vscode/launch.json

Cargo.toml

@@ -22,11 +22,24 @@ name = "transaction-submitter"
 path = "bin/submit_transaction.rs"
 
 [dependencies]
-init4-bin-base = "0.1.0"
+init4-bin-base = "0.3"
 
-zenith-types = "0.13"
+signet-zenith = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
+signet-types = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
+signet-bundle = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
+signet-sim = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
 
-alloy = { version = "0.7.3", features = ["full", "json-rpc", "signer-aws", "rpc-types-mev", "rlp"] }
+trevm = { version = "0.20.10", features = ["concurrent-db", "test-utils"] }
+
+alloy = { version = "0.12.6", features = [
+    "full",
+    "json-rpc",
+    "signer-aws",
+    "rpc-types-mev",
+    "rlp",
+    "node-bindings",
+    "serde",
+] }
 
 aws-config = "1.1.7"
 aws-sdk-kms = "1.15.0"
@@ -48,3 +61,5 @@ tokio = { version = "1.36.0", features = ["full", "macros", "rt-multi-thread"] }
 
 async-trait = "0.1.80"
 oauth2 = "4.4.2"
+tracing-subscriber = "0.3.19"
+chrono = "0.4.41"

bin/builder.rs

@@ -1,34 +1,39 @@
-#![allow(dead_code)]
-
-use builder::config::BuilderConfig;
-use builder::service::serve_builder_with_span;
-use builder::tasks::block::BlockBuilder;
-use builder::tasks::metrics::MetricsTask;
-use builder::tasks::oauth::Authenticator;
-use builder::tasks::submit::SubmitTask;
-
+use builder::{
+    config::BuilderConfig,
+    service::serve_builder_with_span,
+    tasks::{
+        block::Simulator, bundler, metrics::MetricsTask, oauth::Authenticator, submit::SubmitTask,
+        tx_poller,
+    },
+};
+use signet_sim::SimCache;
+use signet_types::SlotCalculator;
+use std::sync::Arc;
 use tokio::select;
 
-#[tokio::main]
+// Note: Must be set to `multi_thread` to support async tasks.
+// See: https://docs.rs/tokio/latest/tokio/attr.main.html
+#[tokio::main(flavor = "multi_thread")]
 async fn main() -> eyre::Result<()> {
     let _guard = init4_bin_base::init4();
 
     let span = tracing::info_span!("zenith-builder");
 
     let config = BuilderConfig::load_from_env()?.clone();
-    let host_provider = config.connect_host_provider().await?;
-    let ru_provider = config.connect_ru_provider().await?;
+    let constants = config.load_pecorino_constants();
     let authenticator = Authenticator::new(&config);
 
-    tracing::debug!(rpc_url = config.host_rpc_url.as_ref(), "instantiated provider");
+    let (host_provider, ru_provider, sequencer_signer) = tokio::try_join!(
+        config.connect_host_provider(),
+        config.connect_ru_provider(),
+        config.connect_sequencer_signer(),
+    )?;
 
-    let sequencer_signer = config.connect_sequencer_signer().await?;
     let zenith = config.connect_zenith(host_provider.clone());
 
     let metrics = MetricsTask { host_provider: host_provider.clone() };
     let (tx_channel, metrics_jh) = metrics.spawn();
 
-    let builder = BlockBuilder::new(&config, authenticator.clone(), ru_provider.clone());
     let submit = SubmitTask {
         authenticator: authenticator.clone(),
         host_provider,
@@ -39,14 +44,43 @@ async fn main() -> eyre::Result<()> {
         outbound_tx_channel: tx_channel,
     };
 
+    let tx_poller = tx_poller::TxPoller::new(&config);
+    let (tx_receiver, tx_poller_jh) = tx_poller.spawn();
+
+    let bundle_poller = bundler::BundlePoller::new(&config, authenticator.clone());
+    let (bundle_receiver, bundle_poller_jh) = bundle_poller.spawn();
+
     let authenticator_jh = authenticator.spawn();
+
     let (submit_channel, submit_jh) = submit.spawn();
-    let build_jh = builder.spawn(submit_channel);
+
+    let sim_items = SimCache::new();
+    let slot_calculator =
+        SlotCalculator::new(config.start_timestamp, config.chain_offset, config.target_slot_time);
+
+    let sim = Arc::new(Simulator::new(&config, ru_provider.clone(), slot_calculator));
+
+    let (basefee_jh, sim_cache_jh) =
+        sim.clone().spawn_cache_tasks(tx_receiver, bundle_receiver, sim_items.clone());
+
+    let build_jh = sim.clone().spawn_simulator_task(constants, sim_items.clone(), submit_channel);
 
     let port = config.builder_port;
     let server = serve_builder_with_span(([0, 0, 0, 0], port), span);
 
     select! {
+        _ = tx_poller_jh => {
+            tracing::info!("tx_poller finished");
+        },
+        _ = bundle_poller_jh => {
+            tracing::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");
         },

bin/submit_transaction.rs

@@ -6,7 +6,7 @@ use alloy::{
     signers::aws::AwsSigner,
 };
 use aws_config::BehaviorVersion;
-use builder::config::{Provider, load_address, load_string, load_u64, load_url};
+use builder::config::{HostProvider, load_address, load_string, load_u64, load_url};
 use init4_bin_base::{
     deps::metrics::{counter, histogram},
     init4,
@@ -30,7 +30,7 @@ async fn main() {
     }
 }
 
-async fn send_transaction(provider: Provider, recipient_address: Address) {
+async fn send_transaction(provider: HostProvider, recipient_address: Address) {
     // construct simple transaction to send ETH to a recipient
     let tx = TransactionRequest::default()
         .with_from(provider.default_signer_address())
@@ -67,7 +67,7 @@ async fn send_transaction(provider: Provider, recipient_address: Address) {
     histogram!("txn_submitter.tx_mine_time").record(mine_time as f64);
 }
 
-async fn connect_from_config() -> (Provider, Address, u64) {
+async fn connect_from_config() -> (HostProvider, Address, u64) {
     // load signer config values from .env
     let rpc_url = load_url("RPC_URL").unwrap();
     let chain_id = load_u64("CHAIN_ID").unwrap();
@@ -82,9 +82,8 @@ async fn connect_from_config() -> (Provider, Address, u64) {
     let signer = AwsSigner::new(client, kms_key_id.to_string(), Some(chain_id)).await.unwrap();
 
     let provider = ProviderBuilder::new()
-        .with_recommended_fillers()
         .wallet(EthereumWallet::from(signer))
-        .on_builtin(&rpc_url)
+        .connect(&rpc_url)
         .await
         .unwrap();
 

src/config.rs

@@ -1,4 +1,7 @@
-use crate::signer::{LocalOrAws, SignerError};
+use crate::{
+    constants,
+    signer::{LocalOrAws, SignerError},
+};
 use alloy::{
     network::{Ethereum, EthereumWallet},
     primitives::Address,
@@ -9,10 +12,12 @@ use alloy::{
             WalletFiller,
         },
     },
-    transports::BoxTransport,
 };
+use eyre::Result;
+use oauth2::url;
+use signet_types::config::{HostConfig, PredeployTokens, RollupConfig, SignetSystemConstants};
+use signet_zenith::Zenith;
 use std::{borrow::Cow, env, num, str::FromStr};
-use zenith_types::Zenith;
 
 // Keys for .env variables that need to be set to configure the builder.
 const HOST_CHAIN_ID: &str = "HOST_CHAIN_ID";
@@ -38,6 +43,8 @@ const OAUTH_CLIENT_ID: &str = "OAUTH_CLIENT_ID";
 const OAUTH_CLIENT_SECRET: &str = "OAUTH_CLIENT_SECRET";
 const OAUTH_AUTHENTICATE_URL: &str = "OAUTH_AUTHENTICATE_URL";
 const OAUTH_TOKEN_URL: &str = "OAUTH_TOKEN_URL";
+const CONCURRENCY_LIMIT: &str = "CONCURRENCY_LIMIT";
+const START_TIMESTAMP: &str = "START_TIMESTAMP";
 
 /// Configuration for a builder running a specific rollup on a specific host
 /// chain.
@@ -93,6 +100,10 @@ pub struct BuilderConfig {
     pub oauth_token_url: String,
     /// The oauth token refresh interval in seconds.
     pub oauth_token_refresh_interval: u64,
+    /// The max number of simultaneous block simulations to run.
+    pub concurrency_limit: usize,
+    /// The anchor for slot time and number calculations before adjusting for chain offset.
+    pub start_timestamp: u64,
 }
 
 /// Error loading the configuration.
@@ -116,6 +127,9 @@ pub enum ConfigError {
     /// Error connecting to the signer
     #[error("failed to connect to signer: {0}")]
     Signer(#[from] SignerError),
+    /// I/O error
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
 }
 
 impl ConfigError {
@@ -125,35 +139,24 @@ impl ConfigError {
     }
 }
 
-/// Provider type used to read & write.
-pub type Provider = FillProvider<
+/// Type alias for the provider used to build and submit blocks to the host.
+pub type HostProvider = FillProvider<
     JoinFill<
         JoinFill<
             Identity,
             JoinFill<GasFiller, JoinFill<BlobGasFiller, JoinFill<NonceFiller, ChainIdFiller>>>,
         >,
         WalletFiller<EthereumWallet>,
     >,
-    RootProvider<BoxTransport>,
-    BoxTransport,
+    RootProvider,
     Ethereum,
 >;
 
-/// Provider type used to read-only.
-pub type WalletlessProvider = FillProvider<
-    JoinFill<
-        Identity,
-        JoinFill<GasFiller, JoinFill<BlobGasFiller, JoinFill<NonceFiller, ChainIdFiller>>>,
-    >,
-    RootProvider<BoxTransport>,
-    BoxTransport,
-    Ethereum,
->;
+/// Type alias for the provider used to simulate against rollup state.
+pub type RuProvider = RootProvider<Ethereum>;
 
-/// A Zenith contract instance, using some provider `P` (defaults to
-/// [`Provider`]).
-pub type ZenithInstance<P = Provider> =
-    Zenith::ZenithInstance<BoxTransport, P, alloy::network::Ethereum>;
+/// A [`Zenith`] contract instance using [`Provider`] as the provider.
+pub type ZenithInstance<P = HostProvider> = Zenith::ZenithInstance<(), P, alloy::network::Ethereum>;
 
 impl BuilderConfig {
     /// Load the builder configuration from environment variables.
@@ -189,6 +192,8 @@ impl BuilderConfig {
             oauth_authenticate_url: load_string(OAUTH_AUTHENTICATE_URL)?,
             oauth_token_url: load_string(OAUTH_TOKEN_URL)?,
             oauth_token_refresh_interval: load_u64(AUTH_TOKEN_REFRESH_INTERVAL)?,
+            concurrency_limit: load_concurrency_limit()?,
+            start_timestamp: load_u64(START_TIMESTAMP)?,
         })
     }
 
@@ -209,42 +214,67 @@ impl BuilderConfig {
     }
 
     /// Connect to the Rollup rpc provider.
-    pub async fn connect_ru_provider(&self) -> Result<WalletlessProvider, ConfigError> {
-        ProviderBuilder::new()
-            .with_recommended_fillers()
-            .on_builtin(&self.ru_rpc_url)
-            .await
-            .map_err(Into::into)
+    pub async fn connect_ru_provider(&self) -> Result<RootProvider<Ethereum>, ConfigError> {
+        let url = url::Url::parse(&self.ru_rpc_url).expect("failed to parse URL");
+        let provider = RootProvider::<Ethereum>::new_http(url);
+        Ok(provider)
     }
 
     /// Connect to the Host rpc provider.
-    pub async fn connect_host_provider(&self) -> Result<Provider, ConfigError> {
+    pub async fn connect_host_provider(&self) -> Result<HostProvider, ConfigError> {
         let builder_signer = self.connect_builder_signer().await?;
-        ProviderBuilder::new()
-            .with_recommended_fillers()
+        let provider = ProviderBuilder::new()
             .wallet(EthereumWallet::from(builder_signer))
-            .on_builtin(&self.host_rpc_url)
+            .connect(&self.host_rpc_url)
             .await
-            .map_err(Into::into)
+            .map_err(ConfigError::Provider)?;
+
+        Ok(provider)
     }
 
     /// Connect additional broadcast providers.
-    pub async fn connect_additional_broadcast(
-        &self,
-    ) -> Result<Vec<RootProvider<BoxTransport>>, ConfigError> {
-        let mut providers = Vec::with_capacity(self.tx_broadcast_urls.len());
-        for url in self.tx_broadcast_urls.iter() {
-            let provider =
-                ProviderBuilder::new().on_builtin(url).await.map_err(Into::<ConfigError>::into)?;
+    pub async fn connect_additional_broadcast(&self) -> Result<Vec<RootProvider>, ConfigError> {
+        let mut providers: Vec<RootProvider> = Vec::with_capacity(self.tx_broadcast_urls.len());
+
+        for url_str in self.tx_broadcast_urls.iter() {
+            let url = url::Url::parse(url_str).expect("failed to parse URL");
+            let provider = RootProvider::new_http(url);
             providers.push(provider);
         }
+
         Ok(providers)
     }
 
     /// Connect to the Zenith instance, using the specified provider.
-    pub const fn connect_zenith(&self, provider: Provider) -> ZenithInstance {
+    pub const fn connect_zenith(&self, provider: HostProvider) -> ZenithInstance {
         Zenith::new(self.zenith_address, provider)
     }
+
+    /// Loads the Signet system constants for Pecorino.
+    pub const fn load_pecorino_constants(&self) -> SignetSystemConstants {
+        let host = HostConfig::new(
+            self.host_chain_id,
+            constants::PECORINO_DEPLOY_HEIGHT,
+            self.zenith_address,
+            constants::HOST_ORDERS,
+            constants::HOST_PASSAGE,
+            constants::HOST_TRANSACTOR,
+            PredeployTokens::new(constants::HOST_USDC, constants::HOST_USDT, constants::HOST_WBTC),
+        );
+        let rollup = RollupConfig::new(
+            self.ru_chain_id,
+            constants::ROLLUP_ORDERS,
+            constants::ROLLUP_PASSAGE,
+            constants::BASE_FEE_RECIPIENT,
+            PredeployTokens::new(
+                constants::ROLLUP_USDC,
+                constants::ROLLUP_USDT,
+                constants::ROLLUP_WBTC,
+            ),
+        );
+
+        SignetSystemConstants::new(host, rollup)
+    }
 }
 
 /// Load a string from an environment variable.
@@ -278,5 +308,18 @@ pub fn load_url(key: &str) -> Result<Cow<'static, str>, ConfigError> {
 /// Load an address from an environment variable.
 pub fn load_address(key: &str) -> Result<Address, ConfigError> {
     let address = load_string(key)?;
-    Address::from_str(&address).map_err(Into::into)
+    Address::from_str(&address)
+        .map_err(|_| ConfigError::Var(format!("Invalid address format for {}", key)))
+}
+
+/// Checks the configured concurrency parameter and, if none is set, checks the available
+/// system concurrency with `std::thread::available_parallelism` and returns that.
+pub fn load_concurrency_limit() -> Result<usize, ConfigError> {
+    match load_u16(CONCURRENCY_LIMIT) {
+        Ok(env) => Ok(env as usize),
+        Err(_) => {
+            let limit = std::thread::available_parallelism()?.get();
+            Ok(limit)
+        }
+    }
 }