Fix nightly clippy warnings in metering pipeline

- Add const fn annotations where possible (cache, estimator, runner)
- Add Debug impls for ResourceAnnotator and KafkaBundleConsumer
- Remove redundant clones in meter_rpc and runner
- Change pub mod to pub(crate) mod for internal modules
- Replace redundant closure with function reference
- Remove unused cache field from MeteringRuntime

Author: niran

crates/rpc/src/base/annotator.rs

@@ -1,6 +1,6 @@
 //! Resource annotator that correlates Kafka metering data with flashblock inclusions.
 
-use std::sync::Arc;
+use std::{fmt, sync::Arc};
 
 use alloy_primitives::TxHash;
 use parking_lot::RwLock;
@@ -40,6 +40,14 @@ pub struct ResourceAnnotator {
     pending_transactions: indexmap::IndexMap<TxHash, MeteredTransaction>,
 }
 
+impl fmt::Debug for ResourceAnnotator {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ResourceAnnotator")
+            .field("pending_transactions", &self.pending_transactions.len())
+            .finish_non_exhaustive()
+    }
+}
+
 impl ResourceAnnotator {
     /// Creates a new resource annotator.
     pub fn new(

crates/rpc/src/base/cache.rs

@@ -26,7 +26,7 @@ pub struct MeteredTransaction {
 
 impl MeteredTransaction {
     /// Creates a zeroed transaction (placeholder with no resource usage).
-    pub fn zeroed(tx_hash: B256) -> Self {
+    pub const fn zeroed(tx_hash: B256) -> Self {
         Self {
             tx_hash,
             priority_fee_per_gas: U256::ZERO,
@@ -52,7 +52,7 @@ pub struct ResourceTotals {
 }
 
 impl ResourceTotals {
-    fn accumulate(&mut self, tx: &MeteredTransaction) {
+    const 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);
@@ -93,7 +93,7 @@ impl FlashblockMetrics {
     }
 
     /// Returns the resource totals for this flashblock.
-    pub fn totals(&self) -> ResourceTotals {
+    pub const fn totals(&self) -> ResourceTotals {
         self.totals
     }
 
@@ -103,12 +103,12 @@ impl FlashblockMetrics {
     }
 
     /// Returns the number of transactions.
-    pub fn len(&self) -> usize {
+    pub const fn len(&self) -> usize {
         self.transactions.len()
     }
 
     /// Returns true if empty.
-    pub fn is_empty(&self) -> bool {
+    pub const fn is_empty(&self) -> bool {
         self.transactions.is_empty()
     }
 }
@@ -150,7 +150,7 @@ impl BlockMetrics {
     }
 
     /// Returns the resource totals for this block.
-    pub fn totals(&self) -> ResourceTotals {
+    pub const fn totals(&self) -> ResourceTotals {
         self.totals
     }
 
@@ -184,7 +184,7 @@ impl MeteringCache {
     }
 
     /// Returns the maximum number of blocks retained.
-    pub fn max_blocks(&self) -> usize {
+    pub const fn max_blocks(&self) -> usize {
         self.max_blocks
     }
 

crates/rpc/src/base/estimator.rs

@@ -84,7 +84,7 @@ pub enum ResourceKind {
 
 impl ResourceKind {
     /// Returns all resource kinds in a fixed order.
-    pub fn all() -> [Self; 4] {
+    pub const fn all() -> [Self; 4] {
         [Self::GasUsed, Self::ExecutionTime, Self::StateRootTime, Self::DataAvailability]
     }
 
@@ -98,7 +98,7 @@ impl ResourceKind {
     ///
     /// Other resources like gas and DA bytes are bounded per-block but are
     /// evaluated per-flashblock since their limits apply independently.
-    fn use_it_or_lose_it(self) -> bool {
+    const fn use_it_or_lose_it(self) -> bool {
         matches!(self, Self::ExecutionTime)
     }
 
@@ -185,7 +185,7 @@ pub struct ResourceEstimates {
 
 impl ResourceEstimates {
     /// Returns the estimate for the given resource kind.
-    pub fn get(&self, kind: ResourceKind) -> Option<&ResourceEstimate> {
+    pub const fn get(&self, kind: ResourceKind) -> Option<&ResourceEstimate> {
         match kind {
             ResourceKind::GasUsed => self.gas_used.as_ref(),
             ResourceKind::ExecutionTime => self.execution_time.as_ref(),
@@ -195,7 +195,7 @@ impl ResourceEstimates {
     }
 
     /// Sets the estimate for the given resource kind.
-    pub fn set(&mut self, kind: ResourceKind, estimate: ResourceEstimate) {
+    pub const fn set(&mut self, kind: ResourceKind, estimate: ResourceEstimate) {
         match kind {
             ResourceKind::GasUsed => self.gas_used = Some(estimate),
             ResourceKind::ExecutionTime => self.execution_time = Some(estimate),
@@ -278,7 +278,7 @@ impl PriorityFeeEstimator {
     /// - `limits`: Configured resource capacity limits.
     /// - `default_priority_fee`: Fee to return when a resource is not congested.
     /// - `da_config`: Optional shared DA config for dynamic DA limit updates.
-    pub fn new(
+    pub const fn new(
         cache: Arc<RwLock<MeteringCache>>,
         percentile: f64,
         limits: ResourceLimits,
@@ -318,10 +318,8 @@ impl PriorityFeeEstimator {
         demand: ResourceDemand,
     ) -> Result<Option<BlockPriorityEstimates>, EstimateError> {
         let cache_guard = self.cache.read();
-        let block_metrics = match block_number {
-            Some(target) => cache_guard.block(target),
-            None => cache_guard.blocks_desc().next(),
-        };
+        let block_metrics = block_number
+            .map_or_else(|| cache_guard.blocks_desc().next(), |target| cache_guard.block(target));
         let Some(block_metrics) = block_metrics else {
             return Ok(None);
         };
@@ -555,8 +553,15 @@ fn compute_estimate(
         });
     }
 
-    let (supporting_count, threshold_fee, recommended_fee) = match last_included_idx {
-        Some(idx) => {
+    let (supporting_count, threshold_fee, recommended_fee) = last_included_idx.map_or_else(
+        || {
+            // No transactions fit - even the first transaction would crowd out
+            // the bundle. The bundle must beat the highest fee to be included.
+            // Report 0 supporting transactions since none were actually included.
+            let threshold_fee = transactions[0].priority_fee_per_gas;
+            (0, threshold_fee, threshold_fee)
+        },
+        |idx| {
             // At least one transaction fits alongside the bundle.
             // The threshold is the fee of the last included transaction.
             let threshold_fee = transactions[idx].priority_fee_per_gas;
@@ -574,15 +579,8 @@ fn compute_estimate(
             };
 
             (idx + 1, threshold_fee, recommended_fee)
-        }
-        None => {
-            // No transactions fit - even the first transaction would crowd out
-            // the bundle. The bundle must beat the highest fee to be included.
-            // Report 0 supporting transactions since none were actually included.
-            let threshold_fee = transactions[0].priority_fee_per_gas;
-            (0, threshold_fee, threshold_fee)
-        }
-    };
+        },
+    );
 
     Ok(ResourceEstimate {
         threshold_priority_fee: threshold_fee,

crates/rpc/src/base/kafka.rs

@@ -1,6 +1,6 @@
 //! Kafka consumer for accepted bundle events.
 
-use std::time::Duration;
+use std::{fmt, time::Duration};
 
 use alloy_consensus::{Transaction, transaction::Recovered};
 use alloy_eips::Encodable2718;
@@ -38,6 +38,12 @@ pub struct KafkaBundleConsumer {
     topic: String,
 }
 
+impl fmt::Debug for KafkaBundleConsumer {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("KafkaBundleConsumer").field("topic", &self.topic).finish_non_exhaustive()
+    }
+}
+
 impl KafkaBundleConsumer {
     /// Creates a new Kafka bundle consumer.
     pub fn new(

crates/rpc/src/base/meter_rpc.rs

@@ -42,7 +42,7 @@ where
     }
 
     /// Creates a new instance of MeteringApi with priority fee estimation enabled.
-    pub fn with_estimator(
+    pub const fn with_estimator(
         provider: Provider,
         priority_fee_estimator: Arc<PriorityFeeEstimator>,
     ) -> Self {
@@ -100,7 +100,7 @@ where
             )
         })?;
 
-        let chain_spec = self.provider.chain_spec().clone();
+        let chain_spec = self.provider.chain_spec();
 
         let (results, total_gas_used, total_gas_fees, bundle_hash, total_execution_time) =
             meter_bundle(state_provider, chain_spec, parsed_bundle, &header).map_err(|e| {

crates/rpc/src/base/mod.rs

@@ -1,7 +1,7 @@
-pub mod annotator;
-pub mod cache;
-pub mod estimator;
-pub mod kafka;
+pub(crate) mod annotator;
+pub(crate) mod cache;
+pub(crate) mod estimator;
+pub(crate) mod kafka;
 pub(crate) mod meter;
 pub(crate) mod meter_rpc;
 pub(crate) mod pubsub;

crates/runner/src/extensions/rpc.rs

@@ -26,8 +26,6 @@ use crate::{
 /// Runtime state for the metering pipeline.
 #[derive(Clone)]
 struct MeteringRuntime {
-    /// Shared cache for metered transactions.
-    cache: Arc<RwLock<MeteringCache>>,
     /// Priority fee estimator.
     estimator: Arc<PriorityFeeEstimator>,
     /// Sender for metered transactions from Kafka.
@@ -45,7 +43,7 @@ struct CompositeFlashblocksReceiver<Client> {
 }
 
 impl<Client> CompositeFlashblocksReceiver<Client> {
-    fn new(
+    const fn new(
         state: Arc<FlashblocksState<Client>>,
         metering_sender: Option<mpsc::UnboundedSender<FlashblockInclusion>>,
     ) -> Self {
@@ -84,8 +82,7 @@ fn flashblock_inclusion_from_flashblock(flashblock: &Flashblock) -> Option<Flash
         return None;
     }
 
-    let ordered_tx_hashes: Vec<B256> =
-        flashblock.diff.transactions.iter().map(|tx_bytes| keccak256(tx_bytes)).collect();
+    let ordered_tx_hashes: Vec<B256> = flashblock.diff.transactions.iter().map(keccak256).collect();
 
     Some(FlashblockInclusion {
         block_number: flashblock.metadata.block_number,
@@ -187,14 +184,11 @@ impl BaseNodeExtension for BaseRpcExtension {
                     mpsc::unbounded_channel::<FlashblockInclusion>();
 
                 // Spawn the resource annotator
-                let annotator_cache = cache.clone();
                 tokio::spawn(async move {
-                    ResourceAnnotator::new(annotator_cache, tx_receiver, flashblock_receiver)
-                        .run()
-                        .await;
+                    ResourceAnnotator::new(cache, tx_receiver, flashblock_receiver).run().await;
                 });
 
-                Some(MeteringRuntime { cache, estimator, tx_sender, flashblock_sender })
+                Some(MeteringRuntime { estimator, tx_sender, flashblock_sender })
             } else {
                 None
             };