feat: adds bundles to in progress block creation

- checks the cache for bundles
- ingests any new bundles that it finds into the in progress block
- adds a unit test for bundle ingestion

Author: dylanlott

src/tasks/block.rs

@@ -12,7 +12,7 @@ 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 tracing::{debug, error, Instrument};
 use zenith_types::{encode_txns, Alloy2718Coder};
 
 /// Ethereum's slot time in seconds.
@@ -71,7 +71,7 @@ impl InProgressBlock {
     /// Ingest a bundle into the in-progress block.
     /// Ignores Signed Orders for now.
     pub fn ingest_bundle(&mut self, bundle: Bundle) {
-        tracing::trace!(bundle = %bundle.id, "ingesting bundle");
+        debug!(bundle = %bundle.id, "ingesting bundle");
 
         let txs = bundle
             .bundle
@@ -83,11 +83,9 @@ impl InProgressBlock {
 
         if let Ok(txs) = txs {
             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.
             self.transactions.extend(txs);
         } else {
-            tracing::error!("failed to decode bundle. dropping");
+            error!("failed to decode bundle. dropping");
         }
     }
 
@@ -155,12 +153,12 @@ impl BlockBuilder {
         }
     }
 
-    async fn _get_bundles(&mut self, in_progress: &mut InProgressBlock) {
+    async fn get_bundles(&mut self, in_progress: &mut InProgressBlock) {
         tracing::trace!("query bundles from cache");
         let bundles = self.bundle_poller.check_bundle_cache().await;
+        // OPTIMIZE: Sort bundles received from cache
         match bundles {
             Ok(bundles) => {
-                tracing::trace!("got bundles response");
                 for bundle in bundles {
                     in_progress.ingest_bundle(bundle);
                 }
@@ -218,9 +216,7 @@ impl BlockBuilder {
                     // Build a block
                     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;
+                    self.get_bundles(&mut in_progress).await;
 
                     // Filter confirmed transactions from the block
                     self.filter_transactions(&mut in_progress).await;
@@ -242,3 +238,71 @@ impl BlockBuilder {
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy::primitives::Address;
+    use alloy::{
+        eips::eip2718::Encodable2718,
+        network::{EthereumWallet, TransactionBuilder},
+        rpc::types::{mev::EthSendBundle, TransactionRequest},
+        signers::local::PrivateKeySigner,
+    };
+    use zenith_types::ZenithEthBundle;
+
+    /// Create a mock bundle for testing with a single transaction
+    async fn create_mock_bundle(wallet: &EthereumWallet) -> Bundle {
+        let tx = TransactionRequest::default()
+            .to(Address::ZERO)
+            .from(wallet.default_signer().address())
+            .nonce(1)
+            .max_fee_per_gas(2)
+            .max_priority_fee_per_gas(3)
+            .gas_limit(4)
+            .build(wallet)
+            .await
+            .unwrap()
+            .encoded_2718();
+
+        let eth_bundle = EthSendBundle {
+            txs: vec![tx.into()],
+            block_number: 1,
+            min_timestamp: Some(u64::MIN),
+            max_timestamp: Some(u64::MAX),
+            reverting_tx_hashes: vec![],
+            replacement_uuid: Some("replacement_uuid".to_owned()),
+        };
+
+        let zenith_bundle = ZenithEthBundle { bundle: eth_bundle, host_fills: None };
+
+        Bundle { id: "mock_bundle".to_owned(), bundle: zenith_bundle }
+    }
+
+    #[tokio::test]
+    async fn test_ingest_bundle() {
+        // Setup random creds
+        let signer = PrivateKeySigner::random();
+        let wallet = EthereumWallet::from(signer);
+
+        // Create an empty InProgressBlock and bundle
+        let mut in_progress_block = InProgressBlock::new();
+        let bundle = create_mock_bundle(&wallet).await;
+
+        // Save previous hash for comparison
+        let prev_hash = in_progress_block.contents_hash();
+
+        // Ingest the bundle
+        in_progress_block.ingest_bundle(bundle);
+
+        // Assert hash is changed after ingest
+        assert_ne!(prev_hash, in_progress_block.contents_hash(), "Bundle should change block hash");
+
+        // Assert that the transaction was persisted into block
+        assert_eq!(in_progress_block.len(), 1, "Bundle should be persisted");
+
+        // Assert that the block is properly sealed
+        let raw_encoding = in_progress_block.encode_raw();
+        assert!(!raw_encoding.is_empty(), "Raw encoding should not be empty");
+    }
+}

src/tasks/submit.rs

@@ -21,7 +21,11 @@ use oauth2::TokenResponse;
 use std::time::Instant;
 use tokio::{sync::mpsc, task::JoinHandle};
 use tracing::{debug, error, instrument, trace};
-use zenith_types::{SignRequest, SignResponse, Zenith, Zenith::IncorrectHostBlock};
+use zenith_types::{
+    BundleHelper::{self, FillPermit2},
+    SignRequest, SignResponse,
+    Zenith::IncorrectHostBlock,
+};
 
 macro_rules! spawn_provider_send {
     ($provider:expr, $tx:expr) => {
@@ -108,20 +112,23 @@ impl SubmitTask {
     /// Builds blob transaction from the provided header and signature values
     fn build_blob_tx(
         &self,
-        header: Zenith::BlockHeader,
+        fills: Vec<FillPermit2>,
+        header: BundleHelper::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 data = zenith_types::BundleHelper::submitCall { fills, header, v, r, s }.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))
     }
 
+    /// Returns the next host block height
     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,16 +145,17 @@ impl SubmitTask {
         let r: FixedBytes<32> = resp.sig.r().into();
         let s: FixedBytes<32> = resp.sig.s().into();
 
-        let header = Zenith::BlockHeader {
+        let header = zenith_types::BundleHelper::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 fills = vec![]; // NB: ignored until fills are implemented
         let tx = self
-            .build_blob_tx(header, v, r, s, in_progress)?
+            .build_blob_tx(fills, 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);
@@ -168,6 +176,8 @@ impl SubmitTask {
 
             return Ok(ControlFlow::Skip);
         }
+
+        // All validation checks have passed, send the transaction
         self.send_transaction(resp, tx).await
     }