Add /eth/v2/validator/aggregate_attestation

CHANGELOG.md

@@ -20,6 +20,7 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - Add ability to rollback node's internal state during processing.
 - Change how unsafe protobuf state is created to prevent unnecessary copies.
 - Added benchmarks for process slots for Capella, Deneb, Electra
+- Added GetAggregatedAttestationV2 endpoint.
 
 ### Changed
 

api/server/structs/endpoints_validator.go

@@ -7,7 +7,8 @@ import (
 )
 
 type AggregateAttestationResponse struct {
-	Data *Attestation `json:"data"`
+	Version string          `json:"version,omitempty"`
+	Data    json.RawMessage `json:"data"`
 }
 
 type SubmitContributionAndProofsRequest struct {

beacon-chain/rpc/endpoints.go

@@ -199,6 +199,15 @@ func (s *Service) validatorEndpoints(
 			handler: server.GetAggregateAttestation,
 			methods: []string{http.MethodGet},
 		},
+		{
+			template: "/eth/v2/validator/aggregate_attestation",
+			name:     namespace + ".GetAggregateAttestationV2",
+			middleware: []middleware.Middleware{
+				middleware.AcceptHeaderHandler([]string{api.JsonMediaType}),
+			},
+			handler: server.GetAggregateAttestationV2,
+			methods: []string{http.MethodGet},
+		},
 		{
 			template: "/eth/v1/validator/contribution_and_proofs",
 			name:     namespace + ".SubmitContributionAndProofs",

beacon-chain/rpc/endpoints_test.go

@@ -101,6 +101,7 @@ func Test_endpoints(t *testing.T) {
 		"/eth/v1/validator/blinded_blocks/{slot}":          {http.MethodGet},
 		"/eth/v1/validator/attestation_data":               {http.MethodGet},
 		"/eth/v1/validator/aggregate_attestation":          {http.MethodGet},
+		"/eth/v2/validator/aggregate_attestation":          {http.MethodGet},
 		"/eth/v1/validator/aggregate_and_proofs":           {http.MethodPost},
 		"/eth/v2/validator/aggregate_and_proofs":           {http.MethodPost},
 		"/eth/v1/validator/beacon_committee_subscriptions": {http.MethodPost},

beacon-chain/rpc/eth/validator/BUILD.bazel

@@ -40,6 +40,7 @@ go_library(
         "//monitoring/tracing/trace:go_default_library",
         "//network/httputil:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
+        "//proto/prysm/v1alpha1/attestation/aggregation/attestations:go_default_library",
         "//runtime/version:go_default_library",
         "//time/slots:go_default_library",
         "@com_github_ethereum_go_ethereum//common/hexutil:go_default_library",
@@ -82,6 +83,7 @@ go_test(
         "//config/params:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//crypto/bls:go_default_library",
+        "//crypto/bls/common:go_default_library",
         "//encoding/bytesutil:go_default_library",
         "//network/httputil:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
@@ -93,6 +95,7 @@ go_test(
         "//time/slots:go_default_library",
         "@com_github_ethereum_go_ethereum//common/hexutil:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
+        "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
         "@com_github_sirupsen_logrus//hooks/test:go_default_library",
         "@org_uber_go_mock//gomock:go_default_library",
     ],

beacon-chain/rpc/eth/validator/handlers.go

@@ -2,11 +2,13 @@ package validator
 
 import (
 	"bytes"
+	"cmp"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
+	"slices"
 	"sort"
 	"strconv"
 	"time"
@@ -32,6 +34,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing/trace"
 	"github.com/prysmaticlabs/prysm/v5/network/httputil"
 	ethpbalpha "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1/attestation/aggregation/attestations"
 	"github.com/prysmaticlabs/prysm/v5/runtime/version"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
 	"github.com/sirupsen/logrus"
@@ -48,71 +51,159 @@ func (s *Server) GetAggregateAttestation(w http.ResponseWriter, r *http.Request)
 	if !ok {
 		return
 	}
-
 	_, slot, ok := shared.UintFromQuery(w, r, "slot", true)
 	if !ok {
 		return
 	}
 
-	var match ethpbalpha.Att
-	var err error
-
-	match, err = matchingAtt(s.AttestationsPool.AggregatedAttestations(), primitives.Slot(slot), attDataRoot)
+	agg := s.aggregatedAttestation(w, primitives.Slot(slot), attDataRoot, 0)
+	if agg == nil {
+		return
+	}
+	typedAgg, ok := agg.(*ethpbalpha.Attestation)
+	if !ok {
+		httputil.HandleError(w, fmt.Sprintf("Attestation is not of type %T", &ethpbalpha.Attestation{}), http.StatusInternalServerError)
+		return
+	}
+	data, err := json.Marshal(structs.AttFromConsensus(typedAgg))
 	if err != nil {
-		httputil.HandleError(w, "Could not get matching attestation: "+err.Error(), http.StatusInternalServerError)
+		httputil.HandleError(w, "Could not marshal attestation: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+	httputil.WriteJson(w, &structs.AggregateAttestationResponse{Data: data})
+}
+
+// GetAggregateAttestationV2 aggregates all attestations matching the given attestation data root and slot, returning the aggregated result.
+func (s *Server) GetAggregateAttestationV2(w http.ResponseWriter, r *http.Request) {
+	_, span := trace.StartSpan(r.Context(), "validator.GetAggregateAttestationV2")
+	defer span.End()
+
+	_, attDataRoot, ok := shared.HexFromQuery(w, r, "attestation_data_root", fieldparams.RootLength, true)
+	if !ok {
+		return
+	}
+	_, slot, ok := shared.UintFromQuery(w, r, "slot", true)
+	if !ok {
+		return
+	}
+	_, index, ok := shared.UintFromQuery(w, r, "committee_index", true)
+	if !ok {
 		return
 	}
-	if match == nil {
-		atts, err := s.AttestationsPool.UnaggregatedAttestations()
+
+	agg := s.aggregatedAttestation(w, primitives.Slot(slot), attDataRoot, primitives.CommitteeIndex(index))
+	if agg == nil {
+		return
+	}
+	resp := &structs.AggregateAttestationResponse{
+		Version: version.String(agg.Version()),
+	}
+	if agg.Version() >= version.Electra {
+		typedAgg, ok := agg.(*ethpbalpha.AttestationElectra)
+		if !ok {
+			httputil.HandleError(w, fmt.Sprintf("Attestation is not of type %T", &ethpbalpha.AttestationElectra{}), http.StatusInternalServerError)
+			return
+		}
+		data, err := json.Marshal(structs.AttElectraFromConsensus(typedAgg))
 		if err != nil {
-			httputil.HandleError(w, "Could not get unaggregated attestations: "+err.Error(), http.StatusInternalServerError)
+			httputil.HandleError(w, "Could not marshal attestation: "+err.Error(), http.StatusInternalServerError)
 			return
 		}
-		match, err = matchingAtt(atts, primitives.Slot(slot), attDataRoot)
+		resp.Data = data
+	} else {
+		typedAgg, ok := agg.(*ethpbalpha.Attestation)
+		if !ok {
+			httputil.HandleError(w, fmt.Sprintf("Attestation is not of type %T", &ethpbalpha.Attestation{}), http.StatusInternalServerError)
+			return
+		}
+		data, err := json.Marshal(structs.AttFromConsensus(typedAgg))
 		if err != nil {
-			httputil.HandleError(w, "Could not get matching attestation: "+err.Error(), http.StatusInternalServerError)
+			httputil.HandleError(w, "Could not marshal attestation: "+err.Error(), http.StatusInternalServerError)
 			return
 		}
+		resp.Data = data
 	}
-	if match == nil {
-		httputil.HandleError(w, "No matching attestation found", http.StatusNotFound)
-		return
+	httputil.WriteJson(w, resp)
+}
+
+func (s *Server) aggregatedAttestation(w http.ResponseWriter, slot primitives.Slot, attDataRoot []byte, index primitives.CommitteeIndex) ethpbalpha.Att {
+	var err error
+
+	match, err := matchingAtts(s.AttestationsPool.AggregatedAttestations(), slot, attDataRoot, index)
+	if err != nil {
+		httputil.HandleError(w, "Could not get matching attestations: "+err.Error(), http.StatusInternalServerError)
+		return nil
+	}
+	if len(match) > 0 {
+		// If there are multiple matching aggregated attestations,
+		// then we return the one with the most aggregation bits.
+		slices.SortFunc(match, func(a, b ethpbalpha.Att) int {
+			return cmp.Compare(b.GetAggregationBits().Count(), a.GetAggregationBits().Count())
+		})
+		return match[0]
 	}
 
-	response := &structs.AggregateAttestationResponse{
-		Data: &structs.Attestation{
-			AggregationBits: hexutil.Encode(match.GetAggregationBits()),
-			Data: &structs.AttestationData{
-				Slot:            strconv.FormatUint(uint64(match.GetData().Slot), 10),
-				CommitteeIndex:  strconv.FormatUint(uint64(match.GetData().CommitteeIndex), 10),
-				BeaconBlockRoot: hexutil.Encode(match.GetData().BeaconBlockRoot),
-				Source: &structs.Checkpoint{
-					Epoch: strconv.FormatUint(uint64(match.GetData().Source.Epoch), 10),
-					Root:  hexutil.Encode(match.GetData().Source.Root),
-				},
-				Target: &structs.Checkpoint{
-					Epoch: strconv.FormatUint(uint64(match.GetData().Target.Epoch), 10),
-					Root:  hexutil.Encode(match.GetData().Target.Root),
-				},
-			},
-			Signature: hexutil.Encode(match.GetSignature()),
-		}}
-	httputil.WriteJson(w, response)
+	atts, err := s.AttestationsPool.UnaggregatedAttestations()
+	if err != nil {
+		httputil.HandleError(w, "Could not get unaggregated attestations: "+err.Error(), http.StatusInternalServerError)
+		return nil
+	}
+	match, err = matchingAtts(atts, slot, attDataRoot, index)
+	if err != nil {
+		httputil.HandleError(w, "Could not get matching attestations: "+err.Error(), http.StatusInternalServerError)
+		return nil
+	}
+	if len(match) == 0 {
+		httputil.HandleError(w, "No matching attestations found", http.StatusNotFound)
+		return nil
+	}
+	agg, err := attestations.Aggregate(match)
+	if err != nil {
+		httputil.HandleError(w, "Could not aggregate unaggregated attestations: "+err.Error(), http.StatusInternalServerError)
+		return nil
+	}
+
+	// Aggregating unaggregated attestations will in theory always return just one aggregate,
+	// so we can take the first one and be done with it.
+	return agg[0]
 }
 
-func matchingAtt(atts []ethpbalpha.Att, slot primitives.Slot, attDataRoot []byte) (ethpbalpha.Att, error) {
+func matchingAtts(atts []ethpbalpha.Att, slot primitives.Slot, attDataRoot []byte, index primitives.CommitteeIndex) ([]ethpbalpha.Att, error) {
+	if len(atts) == 0 {
+		return []ethpbalpha.Att{}, nil
+	}
+
+	postElectra := atts[0].Version() >= version.Electra
+
+	result := make([]ethpbalpha.Att, 0)
 	for _, att := range atts {
-		if att.GetData().Slot == slot {
-			root, err := att.GetData().HashTreeRoot()
+		if att.GetData().Slot != slot {
+			continue
+		}
+		// We ignore the committee index from the request before Electra.
+		// This is because before Electra the committee index is part of the attestation data,
+		// meaning that comparing the data root is sufficient.
+		// Post-Electra the committee index in the data root is always 0, so we need to
+		// compare the committee index separately.
+		if postElectra {
+			ci, err := att.GetCommitteeIndex()
 			if err != nil {
-				return nil, errors.Wrap(err, "could not get attestation data root")
+				return nil, err
 			}
-			if bytes.Equal(root[:], attDataRoot) {
-				return att, nil
+			if ci != index {
+				continue
 			}
 		}
+		root, err := att.GetData().HashTreeRoot()
+		if err != nil {
+			return nil, errors.Wrap(err, "could not get attestation data root")
+		}
+		if bytes.Equal(root[:], attDataRoot) {
+			result = append(result, att)
+		}
 	}
-	return nil, nil
+
+	return result, nil
 }
 
 // SubmitContributionAndProofs publishes multiple signed sync committee contribution and proofs.