feat(costs): Stream receipt fetching

primitives/Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-alloy = { version = "0.1.1", default-features = false, features = [
+alloy = { version = "0.7.0", default-features = false, features = [
     "sol-types",
 ] }
 tendermint = { version = "0.35.0", default-features = false }

script/bin/costs.rs

@@ -89,37 +89,44 @@ async fn get_receipts_for_chain(
         .map(|chunk_start| {
             let chunk_end = (chunk_start + ALCHEMY_CHUNK_SIZE - 1).min(end_block.1);
             let provider = provider.clone();
-            let to_addr = to_addr;
 
             async move {
                 let filter = Filter::new()
                     .from_block(chunk_start)
                     .to_block(chunk_end)
                     .address(to_addr)
                     .event_signature(HeadUpdate::SIGNATURE_HASH);
-
                 provider.get_logs(&filter).await
             }
         });
 
-    let logs = futures::stream::iter(chunks)
-        .buffer_unordered(10)
-        .collect::<Vec<_>>()
-        .await;
-
-    for result in logs {
+    let mut stream = futures::stream::iter(chunks).buffer_unordered(3);
+    while let Some(result) = stream.next().await {
         for log in result? {
-            tx_hashes.push(log.transaction_hash.unwrap());
+            if let Some(tx_hash) = log.transaction_hash {
+                tx_hashes.push(tx_hash);
+            }
         }
     }
 
+    println!("Collected all transaction hashes for chain {}.", chain_id);
+
     let mut all_transactions = Vec::new();
     // Get the receipts for the transactions.
-    for tx_hash in tx_hashes {
-        let receipt = provider.get_transaction_receipt(tx_hash).await?;
-        all_transactions.push(receipt.unwrap());
+    let mut stream = futures::stream::iter(tx_hashes.into_iter().map(|tx_hash| {
+        let provider = provider.clone();
+        async move { provider.get_transaction_receipt(tx_hash).await }
+    }))
+    .buffer_unordered(10);
+
+    while let Some(receipt) = stream.next().await {
+        if let Ok(Some(receipt)) = receipt {
+            all_transactions.push(receipt);
+        }
     }
 
+    println!("Collected all receipts for chain {}.", chain_id);
+
     Ok(all_transactions
         .into_iter()
         .filter(|receipt| receipt.from == from_addr)
@@ -128,7 +135,7 @@ async fn get_receipts_for_chain(
             tx_hash: receipt.transaction_hash,
             tx_fee_wei: receipt.gas_used * receipt.effective_gas_price,
             from: receipt.from,
-            to: receipt.to.unwrap(),
+            to: receipt.to.unwrap_or_default(),
         })
         .collect())
 }
@@ -190,11 +197,11 @@ async fn main() -> Result<()> {
     let total = eth_total + base_total + arbitrum_total;
 
     println!(
-        "\n{} paid the following in relaying fees in {}/{}:\n  Ethereum: {:.4} ETH\n  Base: {:.4} ETH\n  Arbitrum: {:.4} ETH\n  Total: {:.4} ETH",
+        "\n{} paid the following in SP1 Blobstream relaying fees in {}/{}:\n  Ethereum: {:.4} ETH\n  Base: {:.4} ETH\n  Arbitrum: {:.4} ETH\n  Total: {:.4} ETH",
         args.from_address, args.month, args.year, eth_total, base_total, arbitrum_total, total
     );
 
-    csv_writer.flush().unwrap();
+    csv_writer.flush()?;
     Ok(())
 }
 
@@ -210,22 +217,26 @@ where
 {
     let latest_block = provider
         .get_block(BlockId::latest(), BlockTransactionsKind::Hashes)
-        .await?
-        .unwrap();
+        .await?;
+    let Some(latest_block) = latest_block else {
+        return Err(anyhow::anyhow!("No latest block found"));
+    };
     let mut low = 0;
     let mut high = latest_block.header().number();
 
     while low <= high {
         let mid = (low + high) / 2;
         let block = provider
             .get_block(mid.into(), BlockTransactionsKind::Hashes)
-            .await?
-            .unwrap();
+            .await?;
+        let Some(block) = block else {
+            return Err(anyhow::anyhow!("No block found"));
+        };
         let block_timestamp = block.header().timestamp();
 
         match block_timestamp.cmp(&target_timestamp) {
             Ordering::Equal => {
-                return Ok((block.header().hash().into(), block.header().number()));
+                return Ok((block.header().hash(), block.header().number()));
             }
             Ordering::Less => low = mid + 1,
             Ordering::Greater => high = mid - 1,
@@ -235,7 +246,9 @@ where
     // Return the block hash of the closest block after the target timestamp
     let block = provider
         .get_block((low - 10).into(), BlockTransactionsKind::Hashes)
-        .await?
-        .unwrap();
-    Ok((block.header().hash().into(), block.header().number()))
+        .await?;
+    let Some(block) = block else {
+        return Err(anyhow::anyhow!("No block found"));
+    };
+    Ok((block.header().hash(), block.header().number()))
 }

script/bin/operator.rs

@@ -2,7 +2,10 @@ use alloy::{
     network::{Ethereum, EthereumWallet},
     primitives::{Address, B256},
     providers::{
-        fillers::{ChainIdFiller, FillProvider, GasFiller, JoinFill, NonceFiller, WalletFiller},
+        fillers::{
+            BlobGasFiller, ChainIdFiller, FillProvider, GasFiller, JoinFill, NonceFiller,
+            WalletFiller,
+        },
         Identity, Provider, ProviderBuilder, RootProvider,
     },
     signers::local::PrivateKeySigner,
@@ -28,7 +31,10 @@ const ELF: &[u8] = include_bytes!("../../elf/blobstream-elf");
 /// ProviderBuilder. Recommended method for passing around a ProviderBuilder.
 type EthereumFillProvider = FillProvider<
     JoinFill<
-        JoinFill<JoinFill<JoinFill<Identity, GasFiller>, NonceFiller>, ChainIdFiller>,
+        JoinFill<
+            Identity,
+            JoinFill<GasFiller, JoinFill<BlobGasFiller, JoinFill<NonceFiller, ChainIdFiller>>>,
+        >,
         WalletFiller<EthereumWallet>,
     >,
     RootProvider<Http<Client>>,

script/src/relay.rs

@@ -12,7 +12,7 @@ use serde_json::json;
 
 /// Get the gas limit associated with the chain id. Note: These values have been found through
 /// trial and error and can be configured.
-pub fn get_gas_limit(chain_id: u64) -> u128 {
+pub fn get_gas_limit(chain_id: u64) -> u64 {
     if chain_id == 42161 || chain_id == 421614 {
         25_000_000
     } else {