feat(engine): implement engine_getBlobsV2

crates/rpc/rpc-engine-api/Cargo.toml

@@ -23,7 +23,6 @@ reth-tasks.workspace = true
 reth-engine-primitives.workspace = true
 reth-transaction-pool.workspace = true
 reth-primitives-traits.workspace = true
-reth-rpc-server-types.workspace = true
 
 # ethereum
 alloy-eips.workspace = true

crates/rpc/rpc-engine-api/src/engine_api.rs

@@ -26,7 +26,6 @@ use reth_payload_primitives::{
 };
 use reth_primitives_traits::{Block, BlockBody};
 use reth_rpc_api::{EngineApiServer, IntoEngineApiRpcModule};
-use reth_rpc_server_types::result::internal_rpc_err;
 use reth_storage_api::{BlockReader, HeaderProvider, StateProviderFactory};
 use reth_tasks::TaskSpawner;
 use reth_transaction_pool::TransactionPool;
@@ -40,7 +39,7 @@ pub type EngineApiSender<Ok> = oneshot::Sender<EngineApiResult<Ok>>;
 /// The upper limit for payload bodies request.
 const MAX_PAYLOAD_BODIES_LIMIT: u64 = 1024;
 
-/// The upper limit blobs `eth_getBlobs`.
+/// The upper limit for blobs in `engine_getBlobsVx`.
 const MAX_BLOB_LIMIT: usize = 128;
 
 /// The Engine API implementation that grants the Consensus layer access to data and
@@ -800,6 +799,55 @@ where
 
         res
     }
+
+    fn get_blobs_v2(
+        &self,
+        versioned_hashes: Vec<B256>,
+    ) -> EngineApiResult<Option<Vec<BlobAndProofV2>>> {
+        if versioned_hashes.len() > MAX_BLOB_LIMIT {
+            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
+        }
+
+        self.inner
+            .tx_pool
+            .get_blobs_for_versioned_hashes_v2(&versioned_hashes)
+            .map_err(|err| EngineApiError::Internal(Box::new(err)))
+    }
+
+    fn get_blobs_v2_metered(
+        &self,
+        versioned_hashes: Vec<B256>,
+    ) -> EngineApiResult<Option<Vec<BlobAndProofV2>>> {
+        let hashes_len = versioned_hashes.len();
+        let start = Instant::now();
+        let res = Self::get_blobs_v2(self, versioned_hashes);
+        self.inner.metrics.latency.get_blobs_v2.record(start.elapsed());
+
+        if let Ok(blobs) = &res {
+            let blobs_found = blobs.iter().flatten().count();
+
+            self.inner
+                .metrics
+                .blob_metrics
+                .get_blobs_requests_blobs_total
+                .increment(hashes_len as u64);
+            self.inner
+                .metrics
+                .blob_metrics
+                .get_blobs_requests_blobs_in_blobpool_total
+                .increment(blobs_found as u64);
+
+            if blobs_found == hashes_len {
+                self.inner.metrics.blob_metrics.get_blobs_requests_success_total.increment(1);
+            } else {
+                self.inner.metrics.blob_metrics.get_blobs_requests_failure_total.increment(1);
+            }
+        } else {
+            self.inner.metrics.blob_metrics.get_blobs_requests_failure_total.increment(1);
+        }
+
+        res
+    }
 }
 
 impl<Provider, PayloadT, Pool, Validator, ChainSpec>
@@ -1087,10 +1135,10 @@ where
 
     async fn get_blobs_v2(
         &self,
-        _versioned_hashes: Vec<B256>,
+        versioned_hashes: Vec<B256>,
     ) -> RpcResult<Option<Vec<BlobAndProofV2>>> {
         trace!(target: "rpc::engine", "Serving engine_getBlobsV2");
-        Err(internal_rpc_err("unimplemented"))
+        Ok(self.get_blobs_v2_metered(versioned_hashes)?)
     }
 }
 

crates/rpc/rpc-engine-api/src/metrics.rs

@@ -54,6 +54,8 @@ pub(crate) struct EngineApiLatencyMetrics {
     pub(crate) get_payload_bodies_by_hash_v1: Histogram,
     /// Latency for `engine_getBlobsV1`
     pub(crate) get_blobs_v1: Histogram,
+    /// Latency for `engine_getBlobsV2`
+    pub(crate) get_blobs_v2: Histogram,
 }
 
 /// Metrics for engine API forkchoiceUpdated responses.
@@ -115,6 +117,14 @@ pub(crate) struct BlobMetrics {
     pub(crate) blob_count: Counter,
     /// Count of blob misses
     pub(crate) blob_misses: Counter,
+    /// Number of blobs requested via getBlobsV2
+    pub(crate) get_blobs_requests_blobs_total: Counter,
+    /// Number of blobs requested via getBlobsV2 that are present in the blobpool
+    pub(crate) get_blobs_requests_blobs_in_blobpool_total: Counter,
+    /// Number of times getBlobsV2 responded with “hit”
+    pub(crate) get_blobs_requests_success_total: Counter,
+    /// Number of times getBlobsV2 responded with “miss”
+    pub(crate) get_blobs_requests_failure_total: Counter,
 }
 
 impl NewPayloadStatusResponseMetrics {