Add POST /eth/v2/beacon/pool/attestations endpoint

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 SubmitAttestationsV2 endpoint.
 
 ### Changed
 

api/server/structs/endpoints_beacon.go

@@ -26,7 +26,7 @@ type ListAttestationsResponse struct {
 }
 
 type SubmitAttestationsRequest struct {
-	Data []*Attestation `json:"data"`
+	Data json.RawMessage `json:"data"`
 }
 
 type ListVoluntaryExitsResponse struct {

beacon-chain/rpc/endpoints.go

@@ -650,6 +650,16 @@ func (s *Service) beaconEndpoints(
 			handler: server.SubmitAttestations,
 			methods: []string{http.MethodPost},
 		},
+		{
+			template: "/eth/v2/beacon/pool/attestations",
+			name:     namespace + ".SubmitAttestationsV2",
+			middleware: []middleware.Middleware{
+				middleware.ContentTypeHandler([]string{api.JsonMediaType}),
+				middleware.AcceptHeaderHandler([]string{api.JsonMediaType}),
+			},
+			handler: server.SubmitAttestationsV2,
+			methods: []string{http.MethodPost},
+		},
 		{
 			template: "/eth/v1/beacon/pool/voluntary_exits",
 			name:     namespace + ".ListVoluntaryExits",

beacon-chain/rpc/endpoints_test.go

@@ -41,7 +41,7 @@ func Test_endpoints(t *testing.T) {
 		"/eth/v1/beacon/deposit_snapshot":                            {http.MethodGet},
 		"/eth/v1/beacon/blinded_blocks/{block_id}":                   {http.MethodGet},
 		"/eth/v1/beacon/pool/attestations":                           {http.MethodGet, http.MethodPost},
-		"/eth/v2/beacon/pool/attestations":                           {http.MethodGet},
+		"/eth/v2/beacon/pool/attestations":                           {http.MethodGet, http.MethodPost},
 		"/eth/v1/beacon/pool/attester_slashings":                     {http.MethodGet, http.MethodPost},
 		"/eth/v2/beacon/pool/attester_slashings":                     {http.MethodGet, http.MethodPost},
 		"/eth/v1/beacon/pool/proposer_slashings":                     {http.MethodGet, http.MethodPost},

beacon-chain/rpc/eth/beacon/handlers_pool.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -189,14 +190,109 @@ func (s *Server) SubmitAttestations(w http.ResponseWriter, r *http.Request) {
 		httputil.HandleError(w, "Could not decode request body: "+err.Error(), http.StatusBadRequest)
 		return
 	}
-	if len(req.Data) == 0 {
+
+	attFailures, failedBroadcasts, err := s.handleAttestations(ctx, req.Data)
+	if err != nil {
+		httputil.HandleError(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	if len(failedBroadcasts) > 0 {
+		httputil.HandleError(
+			w,
+			fmt.Sprintf("Attestations at index %s could not be broadcasted", strings.Join(failedBroadcasts, ", ")),
+			http.StatusInternalServerError,
+		)
+		return
+	}
+
+	if len(attFailures) > 0 {
+		failuresErr := &server.IndexedVerificationFailureError{
+			Code:     http.StatusBadRequest,
+			Message:  "One or more attestations failed validation",
+			Failures: attFailures,
+		}
+		httputil.WriteError(w, failuresErr)
+	}
+}
+
+// SubmitAttestationsV2 submits an attestation object to node. If the attestation passes all validation
+// constraints, node MUST publish the attestation on an appropriate subnet.
+func (s *Server) SubmitAttestationsV2(w http.ResponseWriter, r *http.Request) {
+	ctx, span := trace.StartSpan(r.Context(), "beacon.SubmitAttestationsV2")
+	defer span.End()
+
+	versionHeader := r.Header.Get(api.VersionHeader)
+	if versionHeader == "" {
+		httputil.HandleError(w, api.VersionHeader+" header is required", http.StatusBadRequest)
+		return
+	}
+	v, err := version.FromString(versionHeader)
+	if err != nil {
+		httputil.HandleError(w, "Invalid version: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	var req structs.SubmitAttestationsRequest
+	err = json.NewDecoder(r.Body).Decode(&req.Data)
+	switch {
+	case errors.Is(err, io.EOF):
 		httputil.HandleError(w, "No data submitted", http.StatusBadRequest)
 		return
+	case err != nil:
+		httputil.HandleError(w, "Could not decode request body: "+err.Error(), http.StatusBadRequest)
+		return
 	}
 
-	var validAttestations []*eth.Attestation
 	var attFailures []*server.IndexedVerificationFailure
-	for i, sourceAtt := range req.Data {
+	var failedBroadcasts []string
+
+	if v >= version.Electra {
+		attFailures, failedBroadcasts, err = s.handleAttestationsElectra(ctx, req.Data)
+	} else {
+		attFailures, failedBroadcasts, err = s.handleAttestations(ctx, req.Data)
+	}
+	if err != nil {
+		httputil.HandleError(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	if len(failedBroadcasts) > 0 {
+		httputil.HandleError(
+			w,
+			fmt.Sprintf("Attestations at index %s could not be broadcasted", strings.Join(failedBroadcasts, ", ")),
+			http.StatusInternalServerError,
+		)
+		return
+	}
+
+	if len(attFailures) > 0 {
+		failuresErr := &server.IndexedVerificationFailureError{
+			Code:     http.StatusBadRequest,
+			Message:  "One or more attestations failed validation",
+			Failures: attFailures,
+		}
+		httputil.WriteError(w, failuresErr)
+	}
+}
+
+func (s *Server) handleAttestationsElectra(ctx context.Context, data json.RawMessage) (attFailures []*server.IndexedVerificationFailure, failedBroadcasts []string, err error) {
+	var sourceAttestations []*structs.AttestationElectra
+
+	if err = json.Unmarshal(data, &sourceAttestations); err != nil {
+		var singleAttestation *structs.AttestationElectra
+		if err = json.Unmarshal(data, &singleAttestation); err != nil {
+			return nil, nil, errors.New("Failed to unmarshal attestation")
+		}
+		sourceAttestations = append(sourceAttestations, singleAttestation)
+	}
+
+	if len(sourceAttestations) == 0 {
+		return nil, nil, errors.New("No data submitted")
+	}
+
+	var validAttestations []*eth.AttestationElectra
+	for i, sourceAtt := range sourceAttestations {
 		att, err := sourceAtt.ToConsensus()
 		if err != nil {
 			attFailures = append(attFailures, &server.IndexedVerificationFailure{
@@ -212,7 +308,10 @@ func (s *Server) SubmitAttestations(w http.ResponseWriter, r *http.Request) {
 			})
 			continue
 		}
+		validAttestations = append(validAttestations, att)
+	}
 
+	for i, att := range validAttestations {
 		// Broadcast the unaggregated attestation on a feed to notify other services in the beacon node
 		// of a received unaggregated attestation.
 		// Note we can't send for aggregated att because we don't have selection proof.
@@ -225,22 +324,94 @@ func (s *Server) SubmitAttestations(w http.ResponseWriter, r *http.Request) {
 			})
 		}
 
+		wantedEpoch := slots.ToEpoch(att.Data.Slot)
+		vals, err := s.HeadFetcher.HeadValidatorsIndices(ctx, wantedEpoch)
+		if err != nil {
+			failedBroadcasts = append(failedBroadcasts, strconv.Itoa(i))
+			continue
+		}
+
+		subnet := corehelpers.ComputeSubnetFromCommitteeAndSlot(uint64(len(vals)), att.Data.CommitteeIndex, att.Data.Slot)
+		if err = s.Broadcaster.BroadcastAttestation(ctx, subnet, att); err != nil {
+			log.WithError(err).Errorf("could not broadcast attestation at index %d", i)
+			failedBroadcasts = append(failedBroadcasts, strconv.Itoa(i))
+			continue
+		}
+
+		if corehelpers.IsAggregated(att) {
+			if err = s.AttestationsPool.SaveAggregatedAttestation(att); err != nil {
+				log.WithError(err).Error("could not save aggregated attestation")
+			}
+		} else {
+			if err = s.AttestationsPool.SaveUnaggregatedAttestation(att); err != nil {
+				log.WithError(err).Error("could not save unaggregated attestation")
+			}
+		}
+	}
+
+	return attFailures, failedBroadcasts, nil
+}
+
+func (s *Server) handleAttestations(ctx context.Context, data json.RawMessage) (attFailures []*server.IndexedVerificationFailure, failedBroadcasts []string, err error) {
+	var sourceAttestations []*structs.Attestation
+
+	if err = json.Unmarshal(data, &sourceAttestations); err != nil {
+		var singleAttestation *structs.Attestation
+		if err = json.Unmarshal(data, &singleAttestation); err != nil {
+			return nil, nil, errors.New("Failed to unmarshal attestation")
+		}
+		sourceAttestations = append(sourceAttestations, singleAttestation)
+	}
+
+	if len(sourceAttestations) == 0 {
+		return nil, nil, errors.New("No data submitted")
+	}
+
+	var validAttestations []*eth.Attestation
+	for i, sourceAtt := range sourceAttestations {
+		att, err := sourceAtt.ToConsensus()
+		if err != nil {
+			attFailures = append(attFailures, &server.IndexedVerificationFailure{
+				Index:   i,
+				Message: "Could not convert request attestation to consensus attestation: " + err.Error(),
+			})
+			continue
+		}
+		if _, err = bls.SignatureFromBytes(att.Signature); err != nil {
+			attFailures = append(attFailures, &server.IndexedVerificationFailure{
+				Index:   i,
+				Message: "Incorrect attestation signature: " + err.Error(),
+			})
+			continue
+		}
 		validAttestations = append(validAttestations, att)
 	}
 
-	failedBroadcasts := make([]string, 0)
 	for i, att := range validAttestations {
-		// Determine subnet to broadcast attestation to
+		// Broadcast the unaggregated attestation on a feed to notify other services in the beacon node
+		// of a received unaggregated attestation.
+		// Note we can't send for aggregated att because we don't have selection proof.
+		if !corehelpers.IsAggregated(att) {
+			s.OperationNotifier.OperationFeed().Send(&feed.Event{
+				Type: operation.UnaggregatedAttReceived,
+				Data: &operation.UnAggregatedAttReceivedData{
+					Attestation: att,
+				},
+			})
+		}
+
 		wantedEpoch := slots.ToEpoch(att.Data.Slot)
 		vals, err := s.HeadFetcher.HeadValidatorsIndices(ctx, wantedEpoch)
 		if err != nil {
-			httputil.HandleError(w, "Could not get head validator indices: "+err.Error(), http.StatusInternalServerError)
-			return
+			failedBroadcasts = append(failedBroadcasts, strconv.Itoa(i))
+			continue
 		}
-		subnet := corehelpers.ComputeSubnetFromCommitteeAndSlot(uint64(len(vals)), att.Data.CommitteeIndex, att.Data.Slot)
 
+		subnet := corehelpers.ComputeSubnetFromCommitteeAndSlot(uint64(len(vals)), att.Data.CommitteeIndex, att.Data.Slot)
 		if err = s.Broadcaster.BroadcastAttestation(ctx, subnet, att); err != nil {
 			log.WithError(err).Errorf("could not broadcast attestation at index %d", i)
+			failedBroadcasts = append(failedBroadcasts, strconv.Itoa(i))
+			continue
 		}
 
 		if corehelpers.IsAggregated(att) {
@@ -253,23 +424,8 @@ func (s *Server) SubmitAttestations(w http.ResponseWriter, r *http.Request) {
 			}
 		}
 	}
-	if len(failedBroadcasts) > 0 {
-		httputil.HandleError(
-			w,
-			fmt.Sprintf("Attestations at index %s could not be broadcasted", strings.Join(failedBroadcasts, ", ")),
-			http.StatusInternalServerError,
-		)
-		return
-	}
 
-	if len(attFailures) > 0 {
-		failuresErr := &server.IndexedVerificationFailureError{
-			Code:     http.StatusBadRequest,
-			Message:  "One or more attestations failed validation",
-			Failures: attFailures,
-		}
-		httputil.WriteError(w, failuresErr)
-	}
+	return attFailures, failedBroadcasts, nil
 }
 
 // ListVoluntaryExits retrieves voluntary exits known by the node but