Update engine_getBlobsV2 response type and add getBlobsV2 tests (#7505)

Update `engine_getBlobsV2` response type to `Option<Vec<BlobsAndProofV2>>`. See recent spec change [here](ethereum/execution-apis#630).

Added some tests to cover basic fetch blob scenarios.

Author: jimmygchen

Cargo.lock

@@ -188,6 +188,8 @@ maplit = "1"
 merkle_proof = { path = "consensus/merkle_proof" }
 metrics = { path = "common/metrics" }
 milhouse = "0.5"
+mockall = "0.13"
+mockall_double = "0.3"
 mockito = "1.5.0"
 monitoring_api = { path = "common/monitoring_api" }
 network = { path = "beacon_node/network" }

Cargo.toml

@@ -1,3 +1,4 @@
+ 
 [package]
 name = "beacon_chain"
 version = "0.2.0"
@@ -69,6 +70,8 @@ types = { workspace = true }
 [dev-dependencies]
 criterion = { workspace = true }
 maplit = { workspace = true }
+mockall = { workspace = true }
+mockall_double = { workspace = true }
 serde_json = { workspace = true }
 
 [[bench]]

beacon_node/beacon_chain/Cargo.toml

@@ -0,0 +1,95 @@
+use crate::blob_verification::{GossipBlobError, GossipVerifiedBlob};
+use crate::fetch_blobs::{EngineGetBlobsOutput, FetchEngineBlobError};
+use crate::observed_data_sidecars::DoNotObserve;
+use crate::{AvailabilityProcessingStatus, BeaconChain, BeaconChainTypes};
+use execution_layer::json_structures::{BlobAndProofV1, BlobAndProofV2};
+use kzg::Kzg;
+#[cfg(test)]
+use mockall::automock;
+use std::sync::Arc;
+use task_executor::TaskExecutor;
+use types::{BlobSidecar, ChainSpec, Hash256, Slot};
+
+/// An adapter to the `BeaconChain` functionalities to remove `BeaconChain` from direct dependency to enable testing fetch blobs logic.
+pub(crate) struct FetchBlobsBeaconAdapter<T: BeaconChainTypes> {
+    chain: Arc<BeaconChain<T>>,
+    spec: Arc<ChainSpec>,
+}
+
+#[cfg_attr(test, automock, allow(dead_code))]
+impl<T: BeaconChainTypes> FetchBlobsBeaconAdapter<T> {
+    pub(crate) fn new(chain: Arc<BeaconChain<T>>) -> Self {
+        let spec = chain.spec.clone();
+        Self { chain, spec }
+    }
+
+    pub(crate) fn spec(&self) -> &Arc<ChainSpec> {
+        &self.spec
+    }
+
+    pub(crate) fn kzg(&self) -> &Arc<Kzg> {
+        &self.chain.kzg
+    }
+
+    pub(crate) fn executor(&self) -> &TaskExecutor {
+        &self.chain.task_executor
+    }
+
+    pub(crate) async fn get_blobs_v1(
+        &self,
+        versioned_hashes: Vec<Hash256>,
+    ) -> Result<Vec<Option<BlobAndProofV1<T::EthSpec>>>, FetchEngineBlobError> {
+        let execution_layer = self
+            .chain
+            .execution_layer
+            .as_ref()
+            .ok_or(FetchEngineBlobError::ExecutionLayerMissing)?;
+
+        execution_layer
+            .get_blobs_v1(versioned_hashes)
+            .await
+            .map_err(FetchEngineBlobError::RequestFailed)
+    }
+
+    pub(crate) async fn get_blobs_v2(
+        &self,
+        versioned_hashes: Vec<Hash256>,
+    ) -> Result<Option<Vec<BlobAndProofV2<T::EthSpec>>>, FetchEngineBlobError> {
+        let execution_layer = self
+            .chain
+            .execution_layer
+            .as_ref()
+            .ok_or(FetchEngineBlobError::ExecutionLayerMissing)?;
+
+        execution_layer
+            .get_blobs_v2(versioned_hashes)
+            .await
+            .map_err(FetchEngineBlobError::RequestFailed)
+    }
+
+    pub(crate) fn verify_blob_for_gossip(
+        &self,
+        blob: &Arc<BlobSidecar<T::EthSpec>>,
+    ) -> Result<GossipVerifiedBlob<T, DoNotObserve>, GossipBlobError> {
+        GossipVerifiedBlob::<T, DoNotObserve>::new(blob.clone(), blob.index, &self.chain)
+    }
+
+    pub(crate) async fn process_engine_blobs(
+        &self,
+        slot: Slot,
+        block_root: Hash256,
+        blobs: EngineGetBlobsOutput<T::EthSpec>,
+    ) -> Result<AvailabilityProcessingStatus, FetchEngineBlobError> {
+        self.chain
+            .process_engine_blobs(slot, block_root, blobs)
+            .await
+            .map_err(FetchEngineBlobError::BlobProcessingError)
+    }
+
+    pub(crate) fn fork_choice_contains_block(&self, block_root: &Hash256) -> bool {
+        self.chain
+            .canonical_head
+            .fork_choice_read_lock()
+            .contains_block(block_root)
+    }
+}