Refactor fetch blobs to be testable and add tests.

Author: jimmygchen

Cargo.lock

@@ -163,6 +163,8 @@ logroller = "0.1.4"
 lru = "0.12"
 maplit = "1"
 milhouse = "0.5"
+mockall = "0.13"
+mockall_double = "0.3"
 mockito = "1.5.0"
 num_cpus = "1"
 once_cell = "1.17.1"

Cargo.toml

@@ -21,6 +21,8 @@ test_backfill = []
 criterion = { workspace = true }
 maplit = { workspace = true }
 serde_json = { workspace = true }
+mockall = { workspace = true }
+mockall_double = { workspace = true }
 
 [dependencies]
 alloy-primitives = { workspace = true }

beacon_node/beacon_chain/Cargo.toml

@@ -0,0 +1,89 @@
+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};
+#[cfg(test)]
+use mockall::automock;
+use std::sync::Arc;
+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 chain(&self) -> &Arc<BeaconChain<T>> {
+        &self.chain
+    }
+
+    pub(crate) fn spec(&self) -> &Arc<ChainSpec> {
+        &self.spec
+    }
+
+    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)
+    }
+}