Apply nightly rustfmt formatting

Author: niran

crates/metering/src/annotator.rs

@@ -1,10 +1,12 @@
-use crate::{MeteredTransaction, MeteringCache};
+use std::sync::Arc;
+
 use alloy_primitives::TxHash;
 use parking_lot::RwLock;
-use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedReceiver;
 use tracing::{debug, info, warn};
 
+use crate::{MeteredTransaction, MeteringCache};
+
 /// Message received from the flashblocks websocket feed indicating which
 /// transactions were included in a specific flashblock.
 #[derive(Debug)]

crates/metering/src/cache.rs

@@ -1,8 +1,7 @@
-use alloy_primitives::B256;
-use indexmap::IndexMap;
 use std::collections::{BTreeMap, HashMap, VecDeque};
 
-use alloy_primitives::U256;
+use alloy_primitives::{B256, U256};
+use indexmap::IndexMap;
 
 #[derive(Debug, Clone)]
 pub struct MeteredTransaction {
@@ -39,23 +38,17 @@ impl ResourceTotals {
     fn accumulate(&mut self, tx: &MeteredTransaction) {
         self.gas_used = self.gas_used.saturating_add(tx.gas_used);
         self.execution_time_us = self.execution_time_us.saturating_add(tx.execution_time_us);
-        self.state_root_time_us = self
-            .state_root_time_us
-            .saturating_add(tx.state_root_time_us);
-        self.data_availability_bytes = self
-            .data_availability_bytes
-            .saturating_add(tx.data_availability_bytes);
+        self.state_root_time_us = self.state_root_time_us.saturating_add(tx.state_root_time_us);
+        self.data_availability_bytes =
+            self.data_availability_bytes.saturating_add(tx.data_availability_bytes);
     }
 
     fn subtract(&mut self, tx: &MeteredTransaction) {
         self.gas_used = self.gas_used.saturating_sub(tx.gas_used);
         self.execution_time_us = self.execution_time_us.saturating_sub(tx.execution_time_us);
-        self.state_root_time_us = self
-            .state_root_time_us
-            .saturating_sub(tx.state_root_time_us);
-        self.data_availability_bytes = self
-            .data_availability_bytes
-            .saturating_sub(tx.data_availability_bytes);
+        self.state_root_time_us = self.state_root_time_us.saturating_sub(tx.state_root_time_us);
+        self.data_availability_bytes =
+            self.data_availability_bytes.saturating_sub(tx.data_availability_bytes);
     }
 }
 
@@ -129,11 +122,7 @@ pub struct BlockMetrics {
 
 impl BlockMetrics {
     pub fn new(block_number: u64) -> Self {
-        Self {
-            block_number,
-            flashblocks: BTreeMap::new(),
-            totals: ResourceTotals::default(),
-        }
+        Self { block_number, flashblocks: BTreeMap::new(), totals: ResourceTotals::default() }
     }
 
     pub fn flashblock_count(&self) -> usize {
@@ -166,18 +155,12 @@ impl BlockMetrics {
         for flashblock in self.flashblocks.values() {
             let totals = flashblock.totals();
             self.totals.gas_used = self.totals.gas_used.saturating_add(totals.gas_used);
-            self.totals.execution_time_us = self
-                .totals
-                .execution_time_us
-                .saturating_add(totals.execution_time_us);
-            self.totals.state_root_time_us = self
-                .totals
-                .state_root_time_us
-                .saturating_add(totals.state_root_time_us);
-            self.totals.data_availability_bytes = self
-                .totals
-                .data_availability_bytes
-                .saturating_add(totals.data_availability_bytes);
+            self.totals.execution_time_us =
+                self.totals.execution_time_us.saturating_add(totals.execution_time_us);
+            self.totals.state_root_time_us =
+                self.totals.state_root_time_us.saturating_add(totals.state_root_time_us);
+            self.totals.data_availability_bytes =
+                self.totals.data_availability_bytes.saturating_add(totals.data_availability_bytes);
         }
     }
 }
@@ -193,21 +176,15 @@ pub struct MeteringCache {
 impl MeteringCache {
     /// Creates a new cache retaining at most `max_blocks` recent blocks.
     pub fn new(max_blocks: usize) -> Self {
-        Self {
-            max_blocks,
-            blocks: VecDeque::new(),
-            block_index: HashMap::new(),
-        }
+        Self { max_blocks, blocks: VecDeque::new(), block_index: HashMap::new() }
     }
 
     pub fn max_blocks(&self) -> usize {
         self.max_blocks
     }
 
     pub fn block(&self, block_number: u64) -> Option<&BlockMetrics> {
-        self.block_index
-            .get(&block_number)
-            .and_then(|&idx| self.blocks.get(idx))
+        self.block_index.get(&block_number).and_then(|&idx| self.blocks.get(idx))
     }
 
     pub fn block_mut(&mut self, block_number: u64) -> &mut BlockMetrics {
@@ -221,18 +198,15 @@ impl MeteringCache {
         self.block_index.insert(block_number, idx);
 
         self.evict_if_needed();
-        self.blocks
-            .get_mut(*self.block_index.get(&block_number).unwrap())
-            .unwrap()
+        self.blocks.get_mut(*self.block_index.get(&block_number).unwrap()).unwrap()
     }
 
     pub fn flashblock(
         &self,
         block_number: u64,
         flashblock_index: u64,
     ) -> Option<&FlashblockMetrics> {
-        self.block(block_number)
-            .and_then(|block| block.flashblock(flashblock_index))
+        self.block(block_number).and_then(|block| block.flashblock(flashblock_index))
     }
 
     pub fn upsert_transaction(
@@ -320,10 +294,7 @@ mod tests {
         let block = cache.block(100).unwrap();
         let flashblock = block.flashblocks().next().unwrap();
         assert_eq!(flashblock.len(), 1);
-        assert_eq!(
-            flashblock.transactions().next().unwrap().tx_hash,
-            tx1.tx_hash
-        );
+        assert_eq!(flashblock.transactions().next().unwrap().tx_hash, tx1.tx_hash);
     }
 
     #[test]
@@ -337,10 +308,7 @@ mod tests {
         let block = cache.block(100).unwrap();
         let flashblock = block.flashblocks().next().unwrap();
         assert_eq!(flashblock.len(), 1);
-        assert_eq!(
-            flashblock.transactions().next().unwrap().gas_used,
-            tx1.gas_used
-        );
+        assert_eq!(flashblock.transactions().next().unwrap().gas_used, tx1.gas_used);
     }
 
     #[test]
@@ -368,9 +336,6 @@ mod tests {
             .iter()
             .map(|tx| tx.priority_fee_per_gas)
             .collect();
-        assert_eq!(
-            sorted,
-            vec![U256::from(10u64), U256::from(20u64), U256::from(30u64)]
-        );
+        assert_eq!(sorted, vec![U256::from(10u64), U256::from(20u64), U256::from(30u64)]);
     }
 }

crates/metering/src/estimator.rs

@@ -1,28 +1,22 @@
-use crate::{MeteredTransaction, MeteringCache};
+use std::sync::Arc;
+
 use alloy_primitives::U256;
 use parking_lot::RwLock;
 use reth_optimism_payload_builder::config::OpDAConfig;
-use std::sync::Arc;
+
+use crate::{MeteredTransaction, MeteringCache};
 
 /// Errors that can occur during priority fee estimation.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum EstimateError {
     /// The bundle's resource demand exceeds the configured capacity limit.
-    DemandExceedsCapacity {
-        resource: ResourceKind,
-        demand: u128,
-        limit: u128,
-    },
+    DemandExceedsCapacity { resource: ResourceKind, demand: u128, limit: u128 },
 }
 
 impl std::fmt::Display for EstimateError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            EstimateError::DemandExceedsCapacity {
-                resource,
-                demand,
-                limit,
-            } => {
+            EstimateError::DemandExceedsCapacity { resource, demand, limit } => {
                 write!(
                     f,
                     "bundle {} demand ({}) exceeds capacity limit ({})",
@@ -256,13 +250,7 @@ impl PriorityFeeEstimator {
         default_priority_fee: U256,
         da_config: Option<OpDAConfig>,
     ) -> Self {
-        Self {
-            cache,
-            percentile,
-            limits,
-            default_priority_fee,
-            da_config,
-        }
+        Self { cache, percentile, limits, default_priority_fee, da_config }
     }
 
     /// Returns the current DA block size limit, preferring the dynamic `OpDAConfig` value
@@ -310,11 +298,8 @@ impl PriorityFeeEstimator {
         let mut flashblock_transactions = Vec::new();
         let mut total_tx_count = 0usize;
         for flashblock in block_metrics.flashblocks() {
-            let sorted: Vec<MeteredTransaction> = flashblock
-                .transactions_sorted_by_priority_fee()
-                .into_iter()
-                .cloned()
-                .collect();
+            let sorted: Vec<MeteredTransaction> =
+                flashblock.transactions_sorted_by_priority_fee().into_iter().cloned().collect();
             if sorted.is_empty() {
                 continue;
             }
@@ -350,11 +335,8 @@ impl PriorityFeeEstimator {
                     continue;
                 };
 
-                let transactions: &[&MeteredTransaction] = if resource.use_it_or_lose_it() {
-                    &aggregate_refs
-                } else {
-                    &txs_refs
-                };
+                let transactions: &[&MeteredTransaction] =
+                    if resource.use_it_or_lose_it() { &aggregate_refs } else { &txs_refs };
                 let estimate = compute_estimate(
                     resource,
                     transactions,
@@ -505,11 +487,7 @@ fn compute_estimate(
 ) -> Result<ResourceEstimate, EstimateError> {
     // Bundle demand exceeds the resource limit entirely.
     if demand > limit {
-        return Err(EstimateError::DemandExceedsCapacity {
-            resource,
-            demand,
-            limit,
-        });
+        return Err(EstimateError::DemandExceedsCapacity { resource, demand, limit });
     }
 
     // No transactions or zero demand means no competition for this resource.
@@ -646,10 +624,12 @@ fn compute_min_max_estimates(
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use std::sync::Arc;
+
     use alloy_primitives::{B256, U256};
     use parking_lot::RwLock;
-    use std::sync::Arc;
+
+    use super::*;
 
     const DEFAULT_FEE: U256 = U256::from_limbs([1, 0, 0, 0]); // 1 wei
     const DEFAULT_LIMITS: ResourceLimits = ResourceLimits {
@@ -847,16 +827,11 @@ mod tests {
         let mut demand = ResourceDemand::default();
         demand.gas_used = Some(15);
 
-        let estimates = estimator
-            .estimate_for_block(Some(1), demand)
-            .expect("no error")
-            .expect("cached block");
+        let estimates =
+            estimator.estimate_for_block(Some(1), demand).expect("no error").expect("cached block");
 
         assert_eq!(estimates.block_number, 1);
-        let gas_estimate = estimates
-            .max_across_flashblocks
-            .gas_used
-            .expect("gas estimate present");
+        let gas_estimate = estimates.max_across_flashblocks.gas_used.expect("gas estimate present");
         assert_eq!(gas_estimate.threshold_priority_fee, U256::from(10));
     }
 
@@ -902,10 +877,8 @@ mod tests {
         let mut demand = ResourceDemand::default();
         demand.gas_used = Some(15);
 
-        let rolling = estimator
-            .estimate_rolling(demand)
-            .expect("no error")
-            .expect("estimates available");
+        let rolling =
+            estimator.estimate_rolling(demand).expect("no error").expect("estimates available");
 
         assert_eq!(rolling.blocks_sampled, 2);
         let gas_estimate = rolling.estimates.gas_used.expect("gas estimate present");

crates/metering/src/kafka.rs

@@ -1,20 +1,22 @@
-use crate::MeteredTransaction;
-use alloy_consensus::Transaction;
-use alloy_consensus::transaction::Recovered;
+use std::time::Duration;
+
+use alloy_consensus::{Transaction, transaction::Recovered};
 use alloy_eips::Encodable2718;
 use alloy_primitives::U256;
 use chrono::Utc;
 use eyre::Result;
 use op_alloy_consensus::OpTxEnvelope;
 use op_alloy_flz::tx_estimated_size_fjord_bytes;
-use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
-use rdkafka::{ClientConfig, Message};
-use std::time::Duration;
+use rdkafka::{
+    ClientConfig, Message,
+    consumer::{CommitMode, Consumer, StreamConsumer},
+};
 use tips_core::types::AcceptedBundle;
-use tokio::sync::mpsc::UnboundedSender;
-use tokio::time::sleep;
+use tokio::{sync::mpsc::UnboundedSender, time::sleep};
 use tracing::{debug, error, info, trace, warn};
 
+use crate::MeteredTransaction;
+
 /// Configuration required to connect to the Kafka topic publishing accepted bundles.
 pub struct KafkaBundleConsumerConfig {
     pub client_config: ClientConfig,
@@ -36,19 +38,12 @@ impl KafkaBundleConsumer {
         config: KafkaBundleConsumerConfig,
         tx_sender: UnboundedSender<MeteredTransaction>,
     ) -> Result<Self> {
-        let KafkaBundleConsumerConfig {
-            client_config,
-            topic,
-        } = config;
+        let KafkaBundleConsumerConfig { client_config, topic } = config;
 
         let consumer: StreamConsumer = client_config.create()?;
         consumer.subscribe(&[topic.as_str()])?;
 
-        Ok(Self {
-            consumer,
-            tx_sender,
-            topic,
-        })
+        Ok(Self { consumer, tx_sender, topic })
     }
 
     /// Starts listening for Kafka messages until the task is cancelled.
@@ -88,9 +83,8 @@ impl KafkaBundleConsumer {
     }
 
     async fn handle_message(&self, message: rdkafka::message::BorrowedMessage<'_>) -> Result<()> {
-        let payload = message
-            .payload()
-            .ok_or_else(|| eyre::eyre!("Kafka message missing payload"))?;
+        let payload =
+            message.payload().ok_or_else(|| eyre::eyre!("Kafka message missing payload"))?;
 
         let bundle: AcceptedBundle = serde_json::from_slice(payload)?;
         metrics::counter!("metering.kafka.messages_total").increment(1);
@@ -137,11 +131,7 @@ impl KafkaBundleConsumer {
             return Ok(());
         }
 
-        for (tx, result) in bundle
-            .txs
-            .iter()
-            .zip(bundle.meter_bundle_response.results.iter())
-        {
+        for (tx, result) in bundle.txs.iter().zip(bundle.meter_bundle_response.results.iter()) {
             let priority_fee_per_gas = calculate_priority_fee(tx);
             let data_availability_bytes = tx_estimated_size_fjord_bytes(&tx.encoded_2718());