Repository: paradigmxyz/reth (language: rust)

Issue:
Engine api metered methods
### Describe the feature

Extract api trait methods into metered methods on `EngineApi`, as has been done for `new_payload` methods.
https://github.com/paradigmxyz/reth/blob/5b19846baae8ac7de9159d846d7a0076dd405fe4/crates/rpc/rpc-engine-api/src/engine_api.rs#L829-L842
https://github.com/paradigmxyz/reth/blob/5b19846baae8ac7de9159d846d7a0076dd405fe4/crates/rpc/rpc-engine-api/src/engine_api.rs#L759-L769

### Additional context

This is necessary in order to reuse this code for `OpEngineApiServer for OpEngineApi`, since it complains about conflicting trait bounds on `ExecutionData` with inner `EngineApi` type otherwise.


Discussion / hints:
makes sense!
Hi, I'd like to work on this

Fix patch:
diff --git a/crates/rpc/rpc-engine-api/src/engine_api.rs b/crates/rpc/rpc-engine-api/src/engine_api.rs
index 37405a8d1580..3e76b8c9ff62 100644
--- a/crates/rpc/rpc-engine-api/src/engine_api.rs
+++ b/crates/rpc/rpc-engine-api/src/engine_api.rs
@@ -142,6 +142,156 @@ where
             .ok_or(EngineApiError::UnknownPayload)??)
     }
 
+    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_newpayloadv1>
+    /// Caution: This should not accept the `withdrawals` field
+    pub async fn new_payload_v1(
+        &self,
+        payload: EngineT::ExecutionData,
+    ) -> EngineApiResult<PayloadStatus> {
+        let payload_or_attrs = PayloadOrAttributes::<
+            '_,
+            EngineT::ExecutionData,
+            EngineT::PayloadAttributes,
+        >::from_execution_payload(&payload);
+
+        self.inner
+            .validator
+            .validate_version_specific_fields(EngineApiMessageVersion::V1, payload_or_attrs)?;
+
+        Ok(self
+            .inner
+            .beacon_consensus
+            .new_payload(payload)
+            .await
+            .inspect(|_| self.inner.on_new_payload_response())?)
+    }
+
+    /// Metered version of `new_payload_v1`.
+    async fn new_payload_v1_metered(
+        &self,
+        payload: EngineT::ExecutionData,
+    ) -> EngineApiResult<PayloadStatus> {
+        let start = Instant::now();
+        let gas_used = payload.gas_used();
+
+        let res = Self::new_payload_v1(self, payload).await;
+        let elapsed = start.elapsed();
+        self.inner.metrics.latency.new_payload_v1.record(elapsed);
+        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
+        res
+    }
+
+    /// See also <https://github.com/ethereum/execution-apis/blob/584905270d8ad665718058060267061ecfd79ca5/src/engine/shanghai.md#engine_newpayloadv2>
+    pub async fn new_payload_v2(
+        &self,
+        payload: EngineT::ExecutionData,
+    ) -> EngineApiResult<PayloadStatus> {
+        let payload_or_attrs = PayloadOrAttributes::<
+            '_,
+            EngineT::ExecutionData,
+            EngineT::PayloadAttributes,
+        >::from_execution_payload(&payload);
+        self.inner
+            .validator
+            .validate_version_specific_fields(EngineApiMessageVersion::V2, payload_or_attrs)?;
+        Ok(self
+            .inner
+            .beacon_consensus
+            .new_payload(payload)
+            .await
+            .inspect(|_| self.inner.on_new_payload_response())?)
+    }
+
+    /// Metered version of `new_payload_v2`.
+    pub async fn new_payload_v2_metered(
+        &self,
+        payload: EngineT::ExecutionData,
+    ) -> EngineApiResult<PayloadStatus> {
+        let start = Instant::now();
+        let gas_used = payload.gas_used();
+
+        let res = Self::new_payload_v2(self, payload).await;
+        let elapsed = start.elapsed();
+        self.inner.metrics.latency.new_payload_v2.record(elapsed);
+        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
+        res
+    }
+
+    /// See also <https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/cancun.md#engine_newpayloadv3>
+    pub async fn new_payload_v3(
+        &self,
+        payload: EngineT::ExecutionData,
+    ) -> EngineApiResult<PayloadStatus> {
+        let payload_or_attrs = PayloadOrAttributes::<
+            '_,
+            EngineT::ExecutionData,
+            EngineT::PayloadAttributes,
+        >::from_execution_payload(&payload);
+        self.inner
+            .validator
+            .validate_version_specific_fields(EngineApiMessageVersion::V3, payload_or_attrs)?;
+
+        Ok(self
+            .inner
+            .beacon_consensus
+            .new_payload(payload)
+            .await
+            .inspect(|_| self.inner.on_new_payload_response())?)
+    }
+
+    // Metrics version of `new_payload_v3`
+    async fn new_payload_v3_metered(
+        &self,
+        payload: EngineT::ExecutionData,
+    ) -> RpcResult<PayloadStatus> {
+        let start = Instant::now();
+        let gas_used = payload.gas_used();
+
+        let res = Self::new_payload_v3(self, payload).await;
+        let elapsed = start.elapsed();
+        self.inner.metrics.latency.new_payload_v3.record(elapsed);
+        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
+        Ok(res?)
+    }
+
+    /// See also <https://github.com/ethereum/execution-apis/blob/7907424db935b93c2fe6a3c0faab943adebe8557/src/engine/prague.md#engine_newpayloadv4>
+    pub async fn new_payload_v4(
+        &self,
+        payload: EngineT::ExecutionData,
+    ) -> EngineApiResult<PayloadStatus> {
+        let payload_or_attrs = PayloadOrAttributes::<
+            '_,
+            EngineT::ExecutionData,
+            EngineT::PayloadAttributes,
+        >::from_execution_payload(&payload);
+        self.inner
+            .validator
+            .validate_version_specific_fields(EngineApiMessageVersion::V4, payload_or_attrs)?;
+
+        Ok(self
+            .inner
+            .beacon_consensus
+            .new_payload(payload)
+            .await
+            .inspect(|_| self.inner.on_new_payload_response())?)
+    }
+
+    /// Metrics version of `new_payload_v4`
+    async fn new_payload_v4_metered(
+        &self,
+        payload: EngineT::ExecutionData,
+    ) -> RpcResult<PayloadStatus> {
+        let start = Instant::now();
+        let gas_used = payload.gas_used();
+
+        let res = Self::new_payload_v4(self, payload).await;
+
+        let elapsed = start.elapsed();
+        self.inner.metrics.latency.new_payload_v4.record(elapsed);
+        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
+        Ok(res?)
+    }
+
     /// Sends a message to the beacon consensus engine to update the fork choice _without_
     /// withdrawals.
     ///
@@ -157,6 +307,19 @@ where
             .await
     }
 
+    /// Metrics version of `fork_choice_updated_v1`
+    async fn fork_choice_updated_v1_metered(
+        &self,
+        state: ForkchoiceState,
+        payload_attrs: Option<EngineT::PayloadAttributes>,
+    ) -> EngineApiResult<ForkchoiceUpdated> {
+        let start = Instant::now();
+        let res = Self::fork_choice_updated_v1(self, state, payload_attrs).await;
+        self.inner.metrics.latency.fork_choice_updated_v1.record(start.elapsed());
+        self.inner.metrics.fcu_response.update_response_metrics(&res);
+        res
+    }
+
     /// Sends a message to the beacon consensus engine to update the fork choice _with_ withdrawals,
     /// but only _after_ shanghai.
     ///
@@ -170,6 +333,19 @@ where
             .await
     }
 
+    /// Metrics version of `fork_choice_updated_v2`
+    async fn fork_choice_updated_v2_metered(
+        &self,
+        state: ForkchoiceState,
+        payload_attrs: Option<EngineT::PayloadAttributes>,
+    ) -> EngineApiResult<ForkchoiceUpdated> {
+        let start = Instant::now();
+        let res = Self::fork_choice_updated_v2(self, state, payload_attrs).await;
+        self.inner.metrics.latency.fork_choice_updated_v2.record(start.elapsed());
+        self.inner.metrics.fcu_response.update_response_metrics(&res);
+        res
+    }
+
     /// Sends a message to the beacon consensus engine to update the fork choice _with_ withdrawals,
     /// but only _after_ cancun.
     ///
@@ -183,6 +359,19 @@ where
             .await
     }
 
+    /// Metrics version of `fork_choice_updated_v3`
+    async fn fork_choice_updated_v3_metered(
+        &self,
+        state: ForkchoiceState,
+        payload_attrs: Option<EngineT::PayloadAttributes>,
+    ) -> EngineApiResult<ForkchoiceUpdated> {
+        let start = Instant::now();
+        let res = Self::fork_choice_updated_v3(self, state, payload_attrs).await;
+        self.inner.metrics.latency.fork_choice_updated_v3.record(start.elapsed());
+        self.inner.metrics.fcu_response.update_response_metrics(&res);
+        res
+    }
+
     /// Returns the most recent version of the payload that is available in the corresponding
     /// payload build process at the time of receiving this call.
     ///
@@ -209,6 +398,17 @@ where
             })
     }
 
+    /// Metrics version of `get_payload_v1`
+    async fn get_payload_v1_metered(
+        &self,
+        payload_id: PayloadId,
+    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV1> {
+        let start = Instant::now();
+        let res = Self::get_payload_v1(self, payload_id).await;
+        self.inner.metrics.latency.get_payload_v1.record(start.elapsed());
+        res
+    }
+
     /// Returns the most recent version of the payload that is available in the corresponding
     /// payload build process at the time of receiving this call.
     ///
@@ -244,6 +444,17 @@ where
             })
     }
 
+    /// Metrics version of `get_payload_v2`
+    async fn get_payload_v2_metered(
+        &self,
+        payload_id: PayloadId,
+    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV2> {
+        let start = Instant::now();
+        let res = Self::get_payload_v2(self, payload_id).await;
+        self.inner.metrics.latency.get_payload_v2.record(start.elapsed());
+        res
+    }
+
     /// Returns the most recent version of the payload that is available in the corresponding
     /// payload build process at the time of receiving this call.
     ///
@@ -279,6 +490,17 @@ where
             })
     }
 
+    /// Metrics version of `get_payload_v3`
+    async fn get_payload_v3_metered(
+        &self,
+        payload_id: PayloadId,
+    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV3> {
+        let start = Instant::now();
+        let res = Self::get_payload_v3(self, payload_id).await;
+        self.inner.metrics.latency.get_payload_v3.record(start.elapsed());
+        res
+    }
+
     /// Returns the most recent version of the payload that is available in the corresponding
     /// payload build process at the time of receiving this call.
     ///
@@ -314,6 +536,17 @@ where
             })
     }
 
+    /// Metrics version of `get_payload_v4`
+    async fn get_payload_v4_metered(
+        &self,
+        payload_id: PayloadId,
+    ) -> EngineApiResult<EngineT::ExecutionPayloadEnvelopeV4> {
+        let start = Instant::now();
+        let res = Self::get_payload_v4(self, payload_id).await;
+        self.inner.metrics.latency.get_payload_v4.record(start.elapsed());
+        res
+    }
+
     /// Fetches all the blocks for the provided range starting at `start`, containing `count`
     /// blocks and returns the mapped payload bodies.
     async fn get_payload_bodies_by_range_with<F, R>(
@@ -393,6 +626,18 @@ where
         .await
     }
 
+    /// Metrics version of `get_payload_bodies_by_range_v1`
+    async fn get_payload_bodies_by_range_v1_metered(
+        &self,
+        start: BlockNumber,
+        count: u64,
+    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
+        let start_time = Instant::now();
+        let res = Self::get_payload_bodies_by_range_v1(self, start, count).await;
+        self.inner.metrics.latency.get_payload_bodies_by_range_v1.record(start_time.elapsed());
+        res
+    }
+
     /// Called to retrieve execution payload bodies by hashes.
     async fn get_payload_bodies_by_hash_with<F, R>(
         &self,
@@ -443,6 +688,17 @@ where
         .await
     }
 
+    /// Metrics version of `get_payload_bodies_by_hash_v1`
+    async fn get_payload_bodies_by_hash_v1_metered(
+        &self,
+        hashes: Vec<BlockHash>,
+    ) -> EngineApiResult<ExecutionPayloadBodiesV1> {
+        let start = Instant::now();
+        let res = Self::get_payload_bodies_by_hash_v1(self, hashes);
+        self.inner.metrics.latency.get_payload_bodies_by_hash_v1.record(start.elapsed());
+        res.await
+    }
+
     /// Called to verify network configuration parameters and ensure that Consensus and Execution
     /// layers are using the latest configuration.
     pub fn exchange_transition_configuration(
@@ -501,6 +757,17 @@ where
         }
     }
 
+    /// Metrics version of `exchange_transition_configuration`
+    fn exchange_transition_configuration_metered(
+        &self,
+        config: TransitionConfiguration,
+    ) -> EngineApiResult<TransitionConfiguration> {
+        let start = Instant::now();
+        let res = Self::exchange_transition_configuration(self, config);
+        self.inner.metrics.latency.exchange_transition_configuration.record(start.elapsed());
+        res
+    }
+
     /// Validates the `engine_forkchoiceUpdated` payload attributes and executes the forkchoice
     /// update.
     ///
@@ -558,166 +825,40 @@ where
     pub fn capabilities(&self) -> &EngineCapabilities {
         &self.inner.capabilities
     }
-}
 
-impl<Provider, EngineT, Pool, Validator, ChainSpec>
-    EngineApi<Provider, EngineT, Pool, Validator, ChainSpec>
-where
-    Provider: HeaderProvider + BlockReader + StateProviderFactory + 'static,
-    EngineT: EngineTypes,
-    Pool: TransactionPool + 'static,
-    Validator: EngineValidator<EngineT>,
-    ChainSpec: EthereumHardforks + Send + Sync + 'static,
-{
-    /// See also <https://github.com/ethereum/execution-apis/blob/3d627c95a4d3510a8187dd02e0250ecb4331d27e/src/engine/paris.md#engine_newpayloadv1>
-    /// Caution: This should not accept the `withdrawals` field
-    pub async fn new_payload_v1(
+    fn get_blobs_v1(
         &self,
-        payload: EngineT::ExecutionData,
-    ) -> EngineApiResult<PayloadStatus> {
-        let payload_or_attrs = PayloadOrAttributes::<
-            '_,
-            EngineT::ExecutionData,
-            EngineT::PayloadAttributes,
-        >::from_execution_payload(&payload);
+        versioned_hashes: Vec<B256>,
+    ) -> EngineApiResult<Vec<Option<BlobAndProofV1>>> {
+        if versioned_hashes.len() > MAX_BLOB_LIMIT {
+            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
+        }
 
         self.inner
-            .validator
-            .validate_version_specific_fields(EngineApiMessageVersion::V1, payload_or_attrs)?;
-
-        Ok(self
-            .inner
-            .beacon_consensus
-            .new_payload(payload)
-            .await
-            .inspect(|_| self.inner.on_new_payload_response())?)
+            .tx_pool
+            .get_blobs_for_versioned_hashes(&versioned_hashes)
+            .map_err(|err| EngineApiError::Internal(Box::new(err)))
     }
 
-    /// Metered version of `new_payload_v1`.
-    async fn new_payload_v1_metered(
+    fn get_blobs_v1_metered(
         &self,
-        payload: EngineT::ExecutionData,
-    ) -> EngineApiResult<PayloadStatus> {
+        versioned_hashes: Vec<B256>,
+    ) -> EngineApiResult<Vec<Option<BlobAndProofV1>>> {
+        let hashes_len = versioned_hashes.len();
         let start = Instant::now();
-        let gas_used = payload.gas_used();
-
-        let res = Self::new_payload_v1(self, payload).await;
-        let elapsed = start.elapsed();
-        self.inner.metrics.latency.new_payload_v1.record(elapsed);
-        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
-        res
-    }
+        let res = Self::get_blobs_v1(self, versioned_hashes);
+        self.inner.metrics.latency.get_blobs_v1.record(start.elapsed());
 
-    /// See also <https://github.com/ethereum/execution-apis/blob/584905270d8ad665718058060267061ecfd79ca5/src/engine/shanghai.md#engine_newpayloadv2>
-    pub async fn new_payload_v2(
-        &self,
-        payload: EngineT::ExecutionData,
-    ) -> EngineApiResult<PayloadStatus> {
-        let payload_or_attrs = PayloadOrAttributes::<
-            '_,
-            EngineT::ExecutionData,
-            EngineT::PayloadAttributes,
-        >::from_execution_payload(&payload);
-        self.inner
-            .validator
-            .validate_version_specific_fields(EngineApiMessageVersion::V2, payload_or_attrs)?;
-        Ok(self
-            .inner
-            .beacon_consensus
-            .new_payload(payload)
-            .await
-            .inspect(|_| self.inner.on_new_payload_response())?)
-    }
+        if let Ok(blobs) = &res {
+            let blobs_found = blobs.iter().flatten().count();
+            let blobs_missed = hashes_len - blobs_found;
 
-    /// Metered version of `new_payload_v2`.
-    pub async fn new_payload_v2_metered(
-        &self,
-        payload: EngineT::ExecutionData,
-    ) -> EngineApiResult<PayloadStatus> {
-        let start = Instant::now();
-        let gas_used = payload.gas_used();
+            self.inner.metrics.blob_metrics.blob_count.increment(blobs_found as u64);
+            self.inner.metrics.blob_metrics.blob_misses.increment(blobs_missed as u64);
+        }
 
-        let res = Self::new_payload_v2(self, payload).await;
-        let elapsed = start.elapsed();
-        self.inner.metrics.latency.new_payload_v2.record(elapsed);
-        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
         res
     }
-
-    /// See also <https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/cancun.md#engine_newpayloadv3>
-    pub async fn new_payload_v3(
-        &self,
-        payload: EngineT::ExecutionData,
-    ) -> EngineApiResult<PayloadStatus> {
-        let payload_or_attrs = PayloadOrAttributes::<
-            '_,
-            EngineT::ExecutionData,
-            EngineT::PayloadAttributes,
-        >::from_execution_payload(&payload);
-        self.inner
-            .validator
-            .validate_version_specific_fields(EngineApiMessageVersion::V3, payload_or_attrs)?;
-
-        Ok(self
-            .inner
-            .beacon_consensus
-            .new_payload(payload)
-            .await
-            .inspect(|_| self.inner.on_new_payload_response())?)
-    }
-
-    // Metrics version of `new_payload_v3`
-    async fn new_payload_v3_metered(
-        &self,
-        payload: EngineT::ExecutionData,
-    ) -> RpcResult<PayloadStatus> {
-        let start = Instant::now();
-        let gas_used = payload.gas_used();
-
-        let res = Self::new_payload_v3(self, payload).await;
-        let elapsed = start.elapsed();
-        self.inner.metrics.latency.new_payload_v3.record(elapsed);
-        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
-        Ok(res?)
-    }
-
-    /// See also <https://github.com/ethereum/execution-apis/blob/7907424db935b93c2fe6a3c0faab943adebe8557/src/engine/prague.md#engine_newpayloadv4>
-    pub async fn new_payload_v4(
-        &self,
-        payload: EngineT::ExecutionData,
-    ) -> EngineApiResult<PayloadStatus> {
-        let payload_or_attrs = PayloadOrAttributes::<
-            '_,
-            EngineT::ExecutionData,
-            EngineT::PayloadAttributes,
-        >::from_execution_payload(&payload);
-        self.inner
-            .validator
-            .validate_version_specific_fields(EngineApiMessageVersion::V4, payload_or_attrs)?;
-
-        Ok(self
-            .inner
-            .beacon_consensus
-            .new_payload(payload)
-            .await
-            .inspect(|_| self.inner.on_new_payload_response())?)
-    }
-
-    /// Metrics version of `new_payload_v4`
-    async fn new_payload_v4_metered(
-        &self,
-        payload: EngineT::ExecutionData,
-    ) -> RpcResult<PayloadStatus> {
-        let start = Instant::now();
-        let gas_used = payload.gas_used();
-
-        let res = Self::new_payload_v4(self, payload).await;
-
-        let elapsed = start.elapsed();
-        self.inner.metrics.latency.new_payload_v4.record(elapsed);
-        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
-        Ok(res?)
-    }
 }
 
 impl<Provider, EngineT, Pool, Validator, ChainSpec>
@@ -824,11 +965,7 @@ where
         payload_attributes: Option<EngineT::PayloadAttributes>,
     ) -> RpcResult<ForkchoiceUpdated> {
         trace!(target: "rpc::engine", "Serving engine_forkchoiceUpdatedV1");
-        let start = Instant::now();
-        let res = Self::fork_choice_updated_v1(self, fork_choice_state, payload_attributes).await;
-        self.inner.metrics.latency.fork_choice_updated_v1.record(start.elapsed());
-        self.inner.metrics.fcu_response.update_response_metrics(&res);
-        Ok(res?)
+        Ok(self.fork_choice_updated_v1_metered(fork_choice_state, payload_attributes).await?)
     }
 
     /// Handler for `engine_forkchoiceUpdatedV2`
@@ -839,11 +976,7 @@ where
         payload_attributes: Option<EngineT::PayloadAttributes>,
     ) -> RpcResult<ForkchoiceUpdated> {
         trace!(target: "rpc::engine", "Serving engine_forkchoiceUpdatedV2");
-        let start = Instant::now();
-        let res = Self::fork_choice_updated_v2(self, fork_choice_state, payload_attributes).await;
-        self.inner.metrics.latency.fork_choice_updated_v2.record(start.elapsed());
-        self.inner.metrics.fcu_response.update_response_metrics(&res);
-        Ok(res?)
+        Ok(self.fork_choice_updated_v2_metered(fork_choice_state, payload_attributes).await?)
     }
 
     /// Handler for `engine_forkchoiceUpdatedV2`
@@ -855,11 +988,7 @@ where
         payload_attributes: Option<EngineT::PayloadAttributes>,
     ) -> RpcResult<ForkchoiceUpdated> {
         trace!(target: "rpc::engine", "Serving engine_forkchoiceUpdatedV3");
-        let start = Instant::now();
-        let res = Self::fork_choice_updated_v3(self, fork_choice_state, payload_attributes).await;
-        self.inner.metrics.latency.fork_choice_updated_v3.record(start.elapsed());
-        self.inner.metrics.fcu_response.update_response_metrics(&res);
-        Ok(res?)
+        Ok(self.fork_choice_updated_v3_metered(fork_choice_state, payload_attributes).await?)
     }
 
     /// Handler for `engine_getPayloadV1`
@@ -878,10 +1007,7 @@ where
         payload_id: PayloadId,
     ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV1> {
         trace!(target: "rpc::engine", "Serving engine_getPayloadV1");
-        let start = Instant::now();
-        let res = Self::get_payload_v1(self, payload_id).await;
-        self.inner.metrics.latency.get_payload_v1.record(start.elapsed());
-        Ok(res?)
+        Ok(self.get_payload_v1_metered(payload_id).await?)
     }
 
     /// Handler for `engine_getPayloadV2`
@@ -898,10 +1024,7 @@ where
         payload_id: PayloadId,
     ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV2> {
         trace!(target: "rpc::engine", "Serving engine_getPayloadV2");
-        let start = Instant::now();
-        let res = Self::get_payload_v2(self, payload_id).await;
-        self.inner.metrics.latency.get_payload_v2.record(start.elapsed());
-        Ok(res?)
+        Ok(self.get_payload_v2_metered(payload_id).await?)
     }
 
     /// Handler for `engine_getPayloadV3`
@@ -918,10 +1041,7 @@ where
         payload_id: PayloadId,
     ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV3> {
         trace!(target: "rpc::engine", "Serving engine_getPayloadV3");
-        let start = Instant::now();
-        let res = Self::get_payload_v3(self, payload_id).await;
-        self.inner.metrics.latency.get_payload_v3.record(start.elapsed());
-        Ok(res?)
+        Ok(self.get_payload_v3_metered(payload_id).await?)
     }
 
     /// Handler for `engine_getPayloadV4`
@@ -938,10 +1058,7 @@ where
         payload_id: PayloadId,
     ) -> RpcResult<EngineT::ExecutionPayloadEnvelopeV4> {
         trace!(target: "rpc::engine", "Serving engine_getPayloadV4");
-        let start = Instant::now();
-        let res = Self::get_payload_v4(self, payload_id).await;
-        self.inner.metrics.latency.get_payload_v4.record(start.elapsed());
-        Ok(res?)
+        Ok(self.get_payload_v4_metered(payload_id).await?)
     }
 
     /// Handler for `engine_getPayloadBodiesByHashV1`
@@ -951,10 +1068,7 @@ where
         block_hashes: Vec<BlockHash>,
     ) -> RpcResult<ExecutionPayloadBodiesV1> {
         trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByHashV1");
-        let start = Instant::now();
-        let res = Self::get_payload_bodies_by_hash_v1(self, block_hashes);
-        self.inner.metrics.latency.get_payload_bodies_by_hash_v1.record(start.elapsed());
-        Ok(res.await?)
+        Ok(self.get_payload_bodies_by_hash_v1_metered(block_hashes).await?)
     }
 
     /// Handler for `engine_getPayloadBodiesByRangeV1`
@@ -979,10 +1093,7 @@ where
         count: U64,
     ) -> RpcResult<ExecutionPayloadBodiesV1> {
         trace!(target: "rpc::engine", "Serving engine_getPayloadBodiesByRangeV1");
-        let start_time = Instant::now();
-        let res = Self::get_payload_bodies_by_range_v1(self, start.to(), count.to()).await;
-        self.inner.metrics.latency.get_payload_bodies_by_range_v1.record(start_time.elapsed());
-        Ok(res?)
+        Ok(self.get_payload_bodies_by_range_v1_metered(start.to(), count.to()).await?)
     }
 
     /// Handler for `engine_exchangeTransitionConfigurationV1`
@@ -992,10 +1103,7 @@ where
         config: TransitionConfiguration,
     ) -> RpcResult<TransitionConfiguration> {
         trace!(target: "rpc::engine", "Serving engine_exchangeTransitionConfigurationV1");
-        let start = Instant::now();
-        let res = Self::exchange_transition_configuration(self, config);
-        self.inner.metrics.latency.exchange_transition_configuration.record(start.elapsed());
-        Ok(res?)
+        Ok(self.exchange_transition_configuration_metered(config)?)
     }
 
     /// Handler for `engine_getClientVersionV1`
@@ -1022,30 +1130,7 @@ where
         versioned_hashes: Vec<B256>,
     ) -> RpcResult<Vec<Option<BlobAndProofV1>>> {
         trace!(target: "rpc::engine", "Serving engine_getBlobsV1");
-        let start = Instant::now();
-
-        if versioned_hashes.len() > MAX_BLOB_LIMIT {
-            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() }.into())
-        }
-
-        let res = self
-            .inner
-            .tx_pool
-            .get_blobs_for_versioned_hashes(&versioned_hashes)
-            .map_err(|err| EngineApiError::Internal(Box::new(err)).into());
-
-        let elapsed = start.elapsed();
-        self.inner.metrics.latency.get_blobs_v1.record(elapsed);
-
-        if let Ok(blobs) = &res {
-            let blobs_found = blobs.iter().flatten().count();
-            let blobs_missed = versioned_hashes.len() - blobs_found;
-
-            self.inner.metrics.blob_metrics.blob_count.increment(blobs_found as u64);
-            self.inner.metrics.blob_metrics.blob_misses.increment(blobs_missed as u64);
-        }
-
-        res
+        Ok(self.get_blobs_v1_metered(versioned_hashes)?)
     }
 }
 


Test patch: