feat: adds bundle submission call

Author: dylanlott

Cargo.toml

@@ -22,23 +22,20 @@ name = "transaction-submitter"
 path = "bin/submit_transaction.rs"
 
 [dependencies]
-zenith-types = { git = "https://github.com/init4tech/zenith-rs", tag = "v0.10.1" }
+zenith-types = { path = "../zenith-rs" }
 
-alloy-primitives = { version = "=0.8.8", features = ["serde", "tiny-keccak"] }
-alloy-sol-types = { version = "=0.8.8", features = ["json"] }
+alloy = { version = "0.7.3", features = ["full", "json-rpc", "signer-aws", "rpc-types-mev"] }
+alloy-primitives = { version = "=0.8.16", features = ["serde", "tiny-keccak"] }
+alloy-sol-types = { version = "=0.8.16", features = ["json"] }
 alloy-rlp = { version = "0.3.4" }
 
-alloy = { version = "0.5.4", features = ["full", "json-rpc", "signer-aws"] }
-
 aws-config = "1.1.7"
 aws-sdk-kms = "1.15.0"
 
 hex = { package = "const-hex", version = "1", default-features = false, features = [
     "alloc",
 ] }
 
-signet-types = { git = "ssh://[email protected]/init4tech/signet-node.git" }
-
 serde = { version = "1.0.197", features = ["derive"] }
 tracing = "0.1.40"
 

src/config.rs

@@ -140,8 +140,7 @@ pub type WalletlessProvider = FillProvider<
     BoxTransport,
     Ethereum,
 >;
-
-pub type ZenithInstance = Zenith::ZenithInstance<BoxTransport, Provider>;
+pub type ZenithInstance = Zenith::ZenithInstance<BoxTransport, Provider, alloy::network::Ethereum>;
 
 impl BuilderConfig {
     /// Load the builder configuration from environment variables.

src/tasks/block.rs

@@ -1,18 +1,25 @@
+use alloy::eips::eip2718::Eip2718Error;
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::{sync::OnceLock, time::Duration};
+use tokio::{sync::mpsc, task::JoinHandle};
+use tracing::Instrument;
+
 use super::bundler::{Bundle, BundlePoller};
 use super::oauth::Authenticator;
 use super::tx_poller::TxPoller;
+
 use crate::config::{BuilderConfig, WalletlessProvider};
+
 use alloy::providers::Provider;
+use alloy::rpc::types::mev::EthSendBundle;
 use alloy::{
     consensus::{SidecarBuilder, SidecarCoder, TxEnvelope},
     eips::eip2718::Decodable2718,
 };
 use alloy_primitives::{keccak256, Bytes, B256};
 use alloy_rlp::Buf;
-use std::time::{SystemTime, UNIX_EPOCH};
-use std::{sync::OnceLock, time::Duration};
-use tokio::{sync::mpsc, task::JoinHandle};
-use tracing::Instrument;
+
+use zenith_types::BundleHelper::FillPermit2;
 use zenith_types::{encode_txns, Alloy2718Coder};
 
 /// Ethereum's slot time in seconds.
@@ -29,7 +36,11 @@ pub struct InProgressBlock {
 impl InProgressBlock {
     /// Create a new `InProgressBlock`
     pub fn new() -> Self {
-        Self { transactions: Vec::new(), raw_encoding: OnceLock::new(), hash: OnceLock::new() }
+        Self {
+            transactions: Vec::new(),
+            raw_encoding: OnceLock::new(),
+            hash: OnceLock::new(),
+        }
     }
 
     /// Get the number of transactions in the block.
@@ -73,15 +84,28 @@ impl InProgressBlock {
     pub fn ingest_bundle(&mut self, bundle: Bundle) {
         tracing::trace!(bundle = %bundle.id, "ingesting bundle");
 
-        let txs = bundle
-            .bundle
-            .bundle
-            .txs
-            .into_iter()
-            .map(|tx| TxEnvelope::decode_2718(&mut tx.chunk()))
-            .collect::<Result<Vec<_>, _>>();
+        // If host fills exists, then it's a Signet bundle
+        if let Some(_) = bundle.bundle.host_fills {
+            self.process_signet_bundle(bundle);
+        } else {
+            // otherwise, treat the bundle as a regular eth bundle and ingest it's tx list in order to maintain compatibility
+            self.process_eth_bundle(bundle);
+        }
+    }
+
+    /// Process a Signet bundle by checking outputs for corresponding host fills and appending the txs in order
+    fn process_signet_bundle(&mut self, bundle: Bundle) {
+        let host_fills = bundle.bundle.host_fills.unwrap(); // safe because we already checked it above
+        let host_fill_list = host_fills.clone().outputs;
+        for output in host_fill_list.iter() {
+            tracing::debug!(?output, "found host fill");
+            // Parse the output as a FillPermit2
+        }
+    }
 
-        if let Ok(txs) = txs {
+    /// Processes a typical bundle of transactions, appending them in order to the in progress block.
+    fn process_eth_bundle(&mut self, bundle: Bundle) {
+        if let Ok(txs) = self.decode_2718_txs(bundle.bundle.bundle) {
             self.unseal();
             // extend the transactions with the decoded transactions.
             // As this builder does not provide bundles landing "top of block", its fine to just extend.
@@ -91,6 +115,15 @@ impl InProgressBlock {
         }
     }
 
+    /// Extracts and decodes a list of 2718 transactions from a EthSendBundle
+    fn decode_2718_txs(&mut self, bundle: EthSendBundle) -> Result<Vec<TxEnvelope>, Eip2718Error> {
+        bundle
+            .txs
+            .into_iter()
+            .map(|tx| TxEnvelope::decode_2718(&mut tx.chunk()))
+            .collect::<Result<Vec<_>, _>>()
+    }
+
     /// Encode the in-progress block
     fn encode_raw(&self) -> &Bytes {
         self.seal();
@@ -125,7 +158,7 @@ pub struct BlockBuilder {
 }
 
 impl BlockBuilder {
-    // create a new block builder with the given config.
+    // Creates a new block builder with the given config
     pub fn new(
         config: &BuilderConfig,
         authenticator: Authenticator,
@@ -139,9 +172,10 @@ impl BlockBuilder {
         }
     }
 
+    /// Fetches transactions from the cache and ingests them into the current block
     async fn get_transactions(&mut self, in_progress: &mut InProgressBlock) {
         tracing::trace!("query transactions from cache");
-        let txns = self.tx_poller.check_tx_cache().await;
+        let txns: Result<Vec<TxEnvelope>, eyre::Error> = self.tx_poller.check_tx_cache().await;
         match txns {
             Ok(txns) => {
                 tracing::trace!("got transactions response");
@@ -155,12 +189,16 @@ impl BlockBuilder {
         }
     }
 
+    /// Returns bundles from the cache and ingests them into the current block
     async fn _get_bundles(&mut self, in_progress: &mut InProgressBlock) {
         tracing::trace!("query bundles from cache");
+
+        // fetch latest bundles from the cache
         let bundles = self.bundle_poller.check_bundle_cache().await;
         match bundles {
             Ok(bundles) => {
-                tracing::trace!("got bundles response");
+                tracing::trace!(?bundles, "got bundles response");
+                // QUESTION: When does this need to stop attempting to ingest for the current block?
                 for bundle in bundles {
                     in_progress.ingest_bundle(bundle);
                 }
@@ -172,6 +210,7 @@ impl BlockBuilder {
         self.bundle_poller.evict();
     }
 
+    // Filters confirmed transactions from the block
     async fn filter_transactions(&self, in_progress: &mut InProgressBlock) {
         // query the rollup node to see which transaction(s) have been included
         let mut confirmed_transactions = Vec::new();
@@ -193,14 +232,14 @@ impl BlockBuilder {
         }
     }
 
-    // calculate the duration in seconds until the beginning of the next block slot.
+    // Calculate the duration in seconds until the beginning of the next block slot.
     fn secs_to_next_slot(&self) -> u64 {
         let curr_timestamp: u64 = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
         let current_slot_time = (curr_timestamp - self.config.chain_offset) % ETHEREUM_SLOT_TIME;
         (ETHEREUM_SLOT_TIME - current_slot_time) % ETHEREUM_SLOT_TIME
     }
 
-    // add a buffer to the beginning of the block slot.
+    // Add a buffer to the beginning of the block slot.
     fn secs_to_next_target(&self) -> u64 {
         self.secs_to_next_slot() + self.config.target_slot_time
     }
@@ -219,8 +258,8 @@ impl BlockBuilder {
                     let mut in_progress = InProgressBlock::default();
                     self.get_transactions(&mut in_progress).await;
 
-                    // TODO: Implement bundle ingestion #later
-                    // self.get_bundles(&mut in_progress).await;
+                    // Ingest bundles
+                    self._get_bundles(&mut in_progress).await;
 
                     // Filter confirmed transactions from the block
                     self.filter_transactions(&mut in_progress).await;

src/tasks/bundler.rs

@@ -1,20 +1,19 @@
-//! Bundler service responsible for polling and submitting bundles to the in-progress block.
-use std::time::{Duration, Instant};
+//! Bundler service responsible for managing bundles. 
+use super::oauth::Authenticator;
 
 pub use crate::config::BuilderConfig;
-use alloy_primitives::map::HashMap;
-use reqwest::Url;
-use serde::{Deserialize, Serialize};
-use signet_types::SignetEthBundle;
 
+use zenith_types::ZenithEthBundle;
 use oauth2::TokenResponse;
-
-use super::oauth::Authenticator;
+use reqwest::Url;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::time::{Duration, Instant};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Bundle {
     pub id: String,
-    pub bundle: SignetEthBundle,
+    pub bundle: ZenithEthBundle,
 }
 
 /// Response from the tx-pool containing a list of bundles.

src/tasks/submit.rs

@@ -4,11 +4,11 @@ use crate::{
     tasks::block::InProgressBlock,
 };
 use alloy::{
-    consensus::{constants::GWEI_TO_WEI, SimpleCoder},
+    consensus::constants::GWEI_TO_WEI,
     eips::BlockNumberOrTag,
-    network::{TransactionBuilder, TransactionBuilder4844},
+    network::TransactionBuilder,
+    providers::Provider as _,
     providers::SendableTx,
-    providers::{Provider as _, WalletProvider},
     rpc::types::eth::TransactionRequest,
     signers::Signer,
     sol_types::SolCall,
@@ -23,8 +23,9 @@ use std::time::Instant;
 use tokio::{sync::mpsc, task::JoinHandle};
 use tracing::{debug, error, instrument, trace};
 use zenith_types::{
+    BundleHelper::{submitCall, BlockHeader, FillPermit2},
     SignRequest, SignResponse,
-    Zenith::{self, IncorrectHostBlock},
+    Zenith::IncorrectHostBlock,
 };
 
 macro_rules! spawn_provider_send {
@@ -109,23 +110,6 @@ impl SubmitTask {
         })
     }
 
-    /// Builds blob transaction from the provided header and signature values
-    fn build_blob_tx(
-        &self,
-        header: Zenith::BlockHeader,
-        v: u8,
-        r: FixedBytes<32>,
-        s: FixedBytes<32>,
-        in_progress: &InProgressBlock,
-    ) -> eyre::Result<TransactionRequest> {
-        let data = Zenith::submitBlockCall { header, v, r, s, _4: Default::default() }.abi_encode();
-        let sidecar = in_progress.encode_blob::<SimpleCoder>().build()?;
-        Ok(TransactionRequest::default()
-            .with_blob_sidecar(sidecar)
-            .with_input(data)
-            .with_max_priority_fee_per_gas((GWEI_TO_WEI * 16) as u128))
-    }
-
     async fn next_host_block_height(&self) -> eyre::Result<u64> {
         let result = self.host_provider.get_block_number().await?;
         let next = result.checked_add(1).ok_or_else(|| eyre!("next host block height overflow"))?;
@@ -138,23 +122,48 @@ impl SubmitTask {
         resp: &SignResponse,
         in_progress: &InProgressBlock,
     ) -> eyre::Result<ControlFlow> {
-        let v: u8 = resp.sig.v().y_parity_byte() + 27;
-        let r: FixedBytes<32> = resp.sig.r().into();
-        let s: FixedBytes<32> = resp.sig.s().into();
-
-        let header = Zenith::BlockHeader {
+        // There are two types of bundles: 
+        // - Bundles with permit2s that require checking and enforcing host fills 
+        // - Bundles that are just groups of transactions and can be executed without sorting host fills
+        // 
+        // Block building depends on the type of bundles being built, and so our builder 
+        // needs to be able to detect and handle both types of bundles or else we will go down. 
+        // This means all builders that build blocks containing bundles on our network need to handle multiple bundle types.
+        //
+        // If a bundle has a permit2 fill, it must be checked if it's one of _our_ permit2 fills. 
+        // If it is, then we can begin the host fills check process.
+
+        let fills: Vec<FillPermit2> = vec![];
+
+        let header = BlockHeader {
             hostBlockNumber: resp.req.host_block_number,
             rollupChainId: U256::from(self.config.ru_chain_id),
             gasLimit: resp.req.gas_limit,
             rewardAddress: resp.req.ru_reward_address,
             blockDataHash: in_progress.contents_hash(),
         };
 
-        let tx = self
-            .build_blob_tx(header, v, r, s, in_progress)?
-            .with_from(self.host_provider.default_signer_address())
-            .with_to(self.config.zenith_address)
-            .with_gas_limit(1_000_000);
+        // get sig details for signing the tx
+        let v = resp.sig.v().into();
+        let r: FixedBytes<32> = resp.sig.r().into();
+        let s: FixedBytes<32> = resp.sig.s().into();
+
+        let submit_call = submitCall { fills, header, v, r, s }.abi_encode();
+
+        let tx = TransactionRequest::default()
+            .with_input(submit_call)
+            .with_max_priority_fee_per_gas((GWEI_TO_WEI * 16) as u128);
+
+        if let Err(TransportError::ErrorResp(e)) =
+            self.host_provider.call(&tx).block(BlockNumberOrTag::Pending.into()).await
+        {
+            error!(
+                code = e.code,
+                message = %e.message,
+                data = ?e.data,
+                "error in transaction submission"
+            );
+        }
 
         if let Err(TransportError::ErrorResp(e)) =
             self.host_provider.call(&tx).block(BlockNumberOrTag::Pending.into()).await