Fix aggregate attestation v2 response (#6926)

michaelsproul · michaelsproul · commit 10931428d108 · 2025-02-07T16:59:37.000+11:00
Squashed commit of the following: commit c11c343 Author: Eitan Seri-Levi <eserilev@ucsc.edu> Date: Thu Feb 6 17:32:34 2025 +0200 fix test commit 4203b56 Author: Eitan Seri-Levi <eserilev@ucsc.edu> Date: Thu Feb 6 16:38:13 2025 +0200 add consensus version header commit 37c9b67 Author: Eitan Seri-Levi <eserilev@ucsc.edu> Date: Thu Feb 6 14:39:11 2025 +0200 Return fork versioned response and a small refactor
diff --git a/beacon_node/http_api/src/aggregate_attestation.rs b/beacon_node/http_api/src/aggregate_attestation.rs
@@ -0,0 +1,65 @@
+use crate::api_types::GenericResponse;
+use crate::unsupported_version_rejection;
+use crate::version::{add_consensus_version_header, V1, V2};
+use beacon_chain::{BeaconChain, BeaconChainTypes};
+use eth2::types::{self, EndpointVersion, Hash256, Slot};
+use std::sync::Arc;
+use types::fork_versioned_response::EmptyMetadata;
+use types::{CommitteeIndex, ForkVersionedResponse};
+use warp::{
+    hyper::{Body, Response},
+    reply::Reply,
+};
+
+pub fn get_aggregate_attestation<T: BeaconChainTypes>(
+    slot: Slot,
+    attestation_data_root: &Hash256,
+    committee_index: Option<CommitteeIndex>,
+    endpoint_version: EndpointVersion,
+    chain: Arc<BeaconChain<T>>,
+) -> Result<Response<Body>, warp::reject::Rejection> {
+    if endpoint_version == V2 {
+        let Some(committee_index) = committee_index else {
+            return Err(warp_utils::reject::custom_bad_request(
+                "missing committee index".to_string(),
+            ));
+        };
+        let aggregate_attestation = chain
+            .get_aggregated_attestation_electra(slot, attestation_data_root, committee_index)
+            .map_err(|e| {
+                warp_utils::reject::custom_bad_request(format!(
+                    "unable to fetch aggregate: {:?}",
+                    e
+                ))
+            })?
+            .ok_or_else(|| {
+                warp_utils::reject::custom_not_found("no matching aggregate found".to_string())
+            })?;
+        let fork_name = chain.spec.fork_name_at_slot::<T::EthSpec>(slot);
+        let fork_versioned_response = ForkVersionedResponse {
+            version: Some(fork_name),
+            metadata: EmptyMetadata {},
+            data: aggregate_attestation,
+        };
+        Ok(add_consensus_version_header(
+            warp::reply::json(&fork_versioned_response).into_response(),
+            fork_name,
+        ))
+    } else if endpoint_version == V1 {
+        let aggregate_attestation = chain
+            .get_pre_electra_aggregated_attestation_by_slot_and_root(slot, attestation_data_root)
+            .map_err(|e| {
+                warp_utils::reject::custom_bad_request(format!(
+                    "unable to fetch aggregate: {:?}",
+                    e
+                ))
+            })?
+            .map(GenericResponse::from)
+            .ok_or_else(|| {
+                warp_utils::reject::custom_not_found("no matching aggregate found".to_string())
+            })?;
+        Ok(warp::reply::json(&aggregate_attestation).into_response())
+    } else {
+        return Err(unsupported_version_rejection(endpoint_version));
+    }
+}
diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs
@@ -5,6 +5,7 @@
 //! There are also some additional, non-standard endpoints behind the `/lighthouse/` path which are
 //! used for development.
 
+mod aggregate_attestation;
 mod attestation_performance;
 mod attester_duties;
 mod block_id;
@@ -3384,40 +3385,15 @@ pub fn serve<T: BeaconChainTypes>(
              not_synced_filter: Result<(), Rejection>,
              task_spawner: TaskSpawner<T::EthSpec>,
              chain: Arc<BeaconChain<T>>| {
-                task_spawner.blocking_json_task(Priority::P0, move || {
+                task_spawner.blocking_response_task(Priority::P0, move || {
                     not_synced_filter?;
-                    let res = if endpoint_version == V2 {
-                        let Some(committee_index) = query.committee_index else {
-                            return Err(warp_utils::reject::custom_bad_request(
-                                "missing committee index".to_string(),
-                            ));
-                        };
-                        chain.get_aggregated_attestation_electra(
-                            query.slot,
-                            &query.attestation_data_root,
-                            committee_index,
-                        )
-                    } else if endpoint_version == V1 {
-                        // Do nothing
-                        chain.get_pre_electra_aggregated_attestation_by_slot_and_root(
-                            query.slot,
-                            &query.attestation_data_root,
-                        )
-                    } else {
-                        return Err(unsupported_version_rejection(endpoint_version));
-                    };
-                    res.map_err(|e| {
-                        warp_utils::reject::custom_bad_request(format!(
-                            "unable to fetch aggregate: {:?}",
-                            e
-                        ))
-                    })?
-                    .map(api_types::GenericResponse::from)
-                    .ok_or_else(|| {
-                        warp_utils::reject::custom_not_found(
-                            "no matching aggregate found".to_string(),
-                        )
-                    })
+                    crate::aggregate_attestation::get_aggregate_attestation(
+                        query.slot,
+                        &query.attestation_data_root,
+                        query.committee_index,
+                        endpoint_version,
+                        chain,
+                    )
                 })
             },
         );
