Separate type for unaggregated network attestations

CHANGELOG.md

@@ -33,6 +33,7 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - Added a Prometheus error counter metric for SSE requests.
 - Save light client updates and bootstraps in DB.
 - Added more comprehensive tests for `BlockToLightClientHeader`. [PR](https://github.com/prysmaticlabs/prysm/pull/14699)
+- Separate type for unaggregated network attestations. [PR](https://github.com/prysmaticlabs/prysm/pull/14659)
 
 ### Changed
 

api/server/structs/conversions.go

@@ -432,6 +432,32 @@ func (a *AttestationElectra) ToConsensus() (*eth.AttestationElectra, error) {
 	}, nil
 }
 
+func (a *SingleAttestation) ToConsensus() (*eth.SingleAttestation, error) {
+	ci, err := strconv.ParseUint(a.CommitteeIndex, 10, 64)
+	if err != nil {
+		return nil, server.NewDecodeError(err, "CommitteeIndex")
+	}
+	ai, err := strconv.ParseUint(a.AttesterIndex, 10, 64)
+	if err != nil {
+		return nil, server.NewDecodeError(err, "AttesterIndex")
+	}
+	data, err := a.Data.ToConsensus()
+	if err != nil {
+		return nil, server.NewDecodeError(err, "Data")
+	}
+	sig, err := bytesutil.DecodeHexWithLength(a.Signature, fieldparams.BLSSignatureLength)
+	if err != nil {
+		return nil, server.NewDecodeError(err, "Signature")
+	}
+
+	return &eth.SingleAttestation{
+		CommitteeId:   primitives.CommitteeIndex(ci),
+		AttesterIndex: primitives.ValidatorIndex(ai),
+		Data:          data,
+		Signature:     sig,
+	}, nil
+}
+
 func AttElectraFromConsensus(a *eth.AttestationElectra) *AttestationElectra {
 	return &AttestationElectra{
 		AggregationBits: hexutil.Encode(a.AggregationBits),

api/server/structs/other.go

@@ -36,6 +36,13 @@ type AttestationElectra struct {
 	CommitteeBits   string           `json:"committee_bits"`
 }
 
+type SingleAttestation struct {
+	CommitteeIndex string           `json:"committee_index"`
+	AttesterIndex  string           `json:"attester_index"`
+	Data           *AttestationData `json:"data"`
+	Signature      string           `json:"signature"`
+}
+
 type AttestationData struct {
 	Slot            string      `json:"slot"`
 	CommitteeIndex  string      `json:"index"`

beacon-chain/core/helpers/attestation.go

@@ -32,7 +32,7 @@ func ValidateNilAttestation(attestation ethpb.Att) error {
 	if attestation.GetData().Target == nil {
 		return errors.New("attestation's target can't be nil")
 	}
-	if attestation.GetAggregationBits() == nil {
+	if !attestation.IsSingle() && attestation.GetAggregationBits() == nil {
 		return errors.New("attestation's bitfield can't be nil")
 	}
 	return nil

beacon-chain/core/helpers/attestation_test.go

@@ -308,6 +308,16 @@ func TestValidateNilAttestation(t *testing.T) {
 			},
 			errString: "",
 		},
+		{
+			name: "single attestation",
+			attestation: &ethpb.SingleAttestation{
+				Data: &ethpb.AttestationData{
+					Target: &ethpb.Checkpoint{},
+					Source: &ethpb.Checkpoint{},
+				},
+			},
+			errString: "",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

beacon-chain/p2p/gossip_topic_mappings.go

@@ -47,7 +47,7 @@ func GossipTopicMappings(topic string, epoch primitives.Epoch) proto.Message {
 		return gossipMessage(topic)
 	case AttestationSubnetTopicFormat:
 		if epoch >= params.BeaconConfig().ElectraForkEpoch {
-			return &ethpb.AttestationElectra{}
+			return &ethpb.SingleAttestation{}
 		}
 		return gossipMessage(topic)
 	case AttesterSlashingSubnetTopicFormat:
@@ -101,7 +101,7 @@ func init() {
 	GossipTypeMapping[reflect.TypeOf(&ethpb.SignedBeaconBlockDeneb{})] = BlockSubnetTopicFormat
 	// Specially handle Electra objects.
 	GossipTypeMapping[reflect.TypeOf(&ethpb.SignedBeaconBlockElectra{})] = BlockSubnetTopicFormat
-	GossipTypeMapping[reflect.TypeOf(&ethpb.AttestationElectra{})] = AttestationSubnetTopicFormat
+	GossipTypeMapping[reflect.TypeOf(&ethpb.SingleAttestation{})] = AttestationSubnetTopicFormat
 	GossipTypeMapping[reflect.TypeOf(&ethpb.AttesterSlashingElectra{})] = AttesterSlashingSubnetTopicFormat
 	GossipTypeMapping[reflect.TypeOf(&ethpb.SignedAggregateAttestationAndProofElectra{})] = AggregateAndProofSubnetTopicFormat
 }

beacon-chain/p2p/gossip_topic_mappings_test.go

@@ -118,7 +118,7 @@ func TestGossipTopicMappings_CorrectType(t *testing.T) {
 	_, ok = pMessage.(*ethpb.SignedBeaconBlockElectra)
 	assert.Equal(t, true, ok)
 	pMessage = GossipTopicMappings(AttestationSubnetTopicFormat, electraForkEpoch)
-	_, ok = pMessage.(*ethpb.AttestationElectra)
+	_, ok = pMessage.(*ethpb.SingleAttestation)
 	assert.Equal(t, true, ok)
 	pMessage = GossipTopicMappings(AttesterSlashingSubnetTopicFormat, electraForkEpoch)
 	_, ok = pMessage.(*ethpb.AttesterSlashingElectra)

beacon-chain/p2p/types/object_mapping.go

@@ -112,7 +112,7 @@ func InitializeDataMaps() {
 			return &ethpb.Attestation{}, nil
 		},
 		bytesutil.ToBytes4(params.BeaconConfig().ElectraForkVersion): func() (ethpb.Att, error) {
-			return &ethpb.AttestationElectra{}, nil
+			return &ethpb.SingleAttestation{}, nil
 		},
 	}
 

beacon-chain/rpc/endpoints.go

@@ -529,6 +529,7 @@ func (s *Service) beaconEndpoints(
 		FinalizationFetcher:     s.cfg.FinalizationFetcher,
 		ForkchoiceFetcher:       s.cfg.ForkchoiceFetcher,
 		CoreService:             coreService,
+		AttestationStateFetcher: s.cfg.AttestationReceiver,
 	}
 
 	const namespace = "beacon"

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

@@ -274,8 +274,11 @@ func (s *Server) SubmitAttestationsV2(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (s *Server) handleAttestationsElectra(ctx context.Context, data json.RawMessage) (attFailures []*server.IndexedVerificationFailure, failedBroadcasts []string, err error) {
-	var sourceAttestations []*structs.AttestationElectra
+func (s *Server) handleAttestationsElectra(
+	ctx context.Context,
+	data json.RawMessage,
+) (attFailures []*server.IndexedVerificationFailure, failedBroadcasts []string, err error) {
+	var sourceAttestations []*structs.SingleAttestation
 
 	if err = json.Unmarshal(data, &sourceAttestations); err != nil {
 		return nil, nil, errors.Wrap(err, "failed to unmarshal attestation")
@@ -285,7 +288,7 @@ func (s *Server) handleAttestationsElectra(ctx context.Context, data json.RawMes
 		return nil, nil, errors.New("no data submitted")
 	}
 
-	var validAttestations []*eth.AttestationElectra
+	var validAttestations []*eth.SingleAttestation
 	for i, sourceAtt := range sourceAttestations {
 		att, err := sourceAtt.ToConsensus()
 		if err != nil {
@@ -306,14 +309,23 @@ func (s *Server) handleAttestationsElectra(ctx context.Context, data json.RawMes
 	}
 
 	for i, att := range validAttestations {
+		targetState, err := s.AttestationStateFetcher.AttestationTargetState(ctx, att.Data.Target)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "could not get target state for attestation")
+		}
+		committee, err := corehelpers.BeaconCommitteeFromState(ctx, targetState, att.Data.Slot, att.CommitteeId)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "could not get committee for attestation")
+		}
+
 		// 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,
+					Attestation: att.ToAttestationElectra(committee),
 				},
 			})
 		}
@@ -336,11 +348,11 @@ func (s *Server) handleAttestationsElectra(ctx context.Context, data json.RawMes
 		}
 
 		if corehelpers.IsAggregated(att) {
-			if err = s.AttestationsPool.SaveAggregatedAttestation(att); err != nil {
+			if err = s.AttestationsPool.SaveAggregatedAttestation(att.ToAttestationElectra(committee)); err != nil {
 				log.WithError(err).Error("could not save aggregated attestation")
 			}
 		} else {
-			if err = s.AttestationsPool.SaveUnaggregatedAttestation(att); err != nil {
+			if err = s.AttestationsPool.SaveUnaggregatedAttestation(att.ToAttestationElectra(committee)); err != nil {
 				log.WithError(err).Error("could not save unaggregated attestation")
 			}
 		}

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

@@ -498,13 +498,17 @@ func TestSubmitAttestations(t *testing.T) {
 	c.SlotsPerEpoch = 1
 	params.OverrideBeaconConfig(c)
 
-	_, keys, err := util.DeterministicDepositsAndKeys(1)
+	_, keys, err := util.DeterministicDepositsAndKeys(2)
 	require.NoError(t, err)
 	validators := []*ethpbv1alpha1.Validator{
 		{
 			PublicKey: keys[0].PublicKey().Marshal(),
 			ExitEpoch: params.BeaconConfig().FarFutureEpoch,
 		},
+		{
+			PublicKey: keys[1].PublicKey().Marshal(),
+			ExitEpoch: params.BeaconConfig().FarFutureEpoch,
+		},
 	}
 	bs, err := util.NewBeaconState(func(state *ethpbv1alpha1.BeaconState) error {
 		state.Validators = validators
@@ -521,9 +525,10 @@ func TestSubmitAttestations(t *testing.T) {
 
 	chainService := &blockchainmock.ChainService{State: bs}
 	s := &Server{
-		HeadFetcher:       chainService,
-		ChainInfoFetcher:  chainService,
-		OperationNotifier: &blockchainmock.MockOperationNotifier{},
+		HeadFetcher:             chainService,
+		ChainInfoFetcher:        chainService,
+		OperationNotifier:       &blockchainmock.MockOperationNotifier{},
+		AttestationStateFetcher: chainService,
 	}
 	t.Run("V1", func(t *testing.T) {
 		t.Run("single", func(t *testing.T) {
@@ -732,7 +737,7 @@ func TestSubmitAttestations(t *testing.T) {
 				assert.Equal(t, http.StatusOK, writer.Code)
 				assert.Equal(t, true, broadcaster.BroadcastCalled.Load())
 				assert.Equal(t, 1, broadcaster.NumAttestations())
-				assert.Equal(t, "0x03", hexutil.Encode(broadcaster.BroadcastAttestations[0].GetAggregationBits()))
+				assert.Equal(t, primitives.ValidatorIndex(1), broadcaster.BroadcastAttestations[0].GetAttestingIndex())
 				assert.Equal(t, "0x8146f4397bfd8fd057ebbcd6a67327bdc7ed5fb650533edcb6377b650dea0b6da64c14ecd60846d5c0a0cd43893d6972092500f82c9d8a955e2b58c5ed3cbe885d84008ace6bd86ba9e23652f58e2ec207cec494c916063257abf285b9b15b15", hexutil.Encode(broadcaster.BroadcastAttestations[0].GetSignature()))
 				assert.Equal(t, primitives.Slot(0), broadcaster.BroadcastAttestations[0].GetData().Slot)
 				assert.Equal(t, primitives.CommitteeIndex(0), broadcaster.BroadcastAttestations[0].GetData().CommitteeIndex)
@@ -2344,8 +2349,8 @@ var (
 ]`
 	singleAttElectra = `[
   {
-    "aggregation_bits": "0x03",
-	"committee_bits": "0x0100000000000000",
+    "committee_index": "0",
+	"attester_index": "1",
     "signature": "0x8146f4397bfd8fd057ebbcd6a67327bdc7ed5fb650533edcb6377b650dea0b6da64c14ecd60846d5c0a0cd43893d6972092500f82c9d8a955e2b58c5ed3cbe885d84008ace6bd86ba9e23652f58e2ec207cec494c916063257abf285b9b15b15",
     "data": {
       "slot": "0",
@@ -2364,8 +2369,8 @@ var (
 ]`
 	multipleAttsElectra = `[
   {
-    "aggregation_bits": "0x03",
-	"committee_bits": "0x0100000000000000",
+    "committee_index": "0",
+	"attester_index": "0",
     "signature": "0x8146f4397bfd8fd057ebbcd6a67327bdc7ed5fb650533edcb6377b650dea0b6da64c14ecd60846d5c0a0cd43893d6972092500f82c9d8a955e2b58c5ed3cbe885d84008ace6bd86ba9e23652f58e2ec207cec494c916063257abf285b9b15b15",
     "data": {
       "slot": "0",
@@ -2382,8 +2387,8 @@ var (
     }
   },
   {
-    "aggregation_bits": "0x03",
-	"committee_bits": "0x0100000000000000",
+    "committee_index": "0",
+	"attester_index": "1",
     "signature": "0x8146f4397bfd8fd057ebbcd6a67327bdc7ed5fb650533edcb6377b650dea0b6da64c14ecd60846d5c0a0cd43893d6972092500f82c9d8a955e2b58c5ed3cbe885d84008ace6bd86ba9e23652f58e2ec207cec494c916063257abf285b9b15b15",
     "data": {
       "slot": "0",
@@ -2403,8 +2408,8 @@ var (
 	// signature is invalid
 	invalidAttElectra = `[
   {
-    "aggregation_bits": "0x03",
-	"committee_bits": "0x0100000000000000",
+    "committee_index": "0",
+	"attester_index": "0",
 	"signature": "0x000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
     "data": {
       "slot": "0",

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

@@ -48,4 +48,5 @@ type Server struct {
 	BLSChangesPool          blstoexec.PoolManager
 	ForkchoiceFetcher       blockchain.ForkchoiceFetcher
 	CoreService             *core.Service
+	AttestationStateFetcher blockchain.AttestationStateFetcher
 }

beacon-chain/rpc/prysm/v1alpha1/validator/attester.go

@@ -12,6 +12,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/crypto/bls"
 	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing/trace"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v5/runtime/version"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -44,7 +45,7 @@ func (vs *Server) ProposeAttestation(ctx context.Context, att *ethpb.Attestation
 	ctx, span := trace.StartSpan(ctx, "AttesterServer.ProposeAttestation")
 	defer span.End()
 
-	resp, err := vs.proposeAtt(ctx, att, att.GetData().CommitteeIndex)
+	resp, err := vs.proposeAtt(ctx, att, nil, att.GetData().CommitteeIndex)
 	if err != nil {
 		return nil, err
 	}
@@ -62,24 +63,31 @@ func (vs *Server) ProposeAttestation(ctx context.Context, att *ethpb.Attestation
 
 // ProposeAttestationElectra is a function called by an attester to vote
 // on a block via an attestation object as defined in the Ethereum specification.
-func (vs *Server) ProposeAttestationElectra(ctx context.Context, att *ethpb.AttestationElectra) (*ethpb.AttestResponse, error) {
+func (vs *Server) ProposeAttestationElectra(ctx context.Context, att *ethpb.SingleAttestation) (*ethpb.AttestResponse, error) {
 	ctx, span := trace.StartSpan(ctx, "AttesterServer.ProposeAttestationElectra")
 	defer span.End()
 
+	targetState, err := vs.AttestationStateFetcher.AttestationTargetState(ctx, att.Data.Target)
+	if err != nil {
+		return nil, status.Error(codes.Internal, "Could not get target state")
+	}
 	committeeIndex, err := att.GetCommitteeIndex()
 	if err != nil {
-		return nil, err
+		return nil, status.Error(codes.Internal, "Could not get committee index")
+	}
+	committee, err := helpers.BeaconCommitteeFromState(ctx, targetState, att.Data.Slot, committeeIndex)
+	if err != nil {
+		return nil, status.Error(codes.Internal, "Could not get committee")
 	}
 
-	resp, err := vs.proposeAtt(ctx, att, committeeIndex)
+	resp, err := vs.proposeAtt(ctx, att, committee, committeeIndex)
 	if err != nil {
 		return nil, err
 	}
 
 	go func() {
 		ctx = trace.NewContext(context.Background(), trace.FromContext(ctx))
-		attCopy := att.Copy()
-		if err := vs.AttPool.SaveUnaggregatedAttestation(attCopy); err != nil {
+		if err := vs.AttPool.SaveUnaggregatedAttestation(att.ToAttestationElectra(committee)); err != nil {
 			log.WithError(err).Error("Could not save unaggregated attestation")
 			return
 		}
@@ -136,14 +144,29 @@ func (vs *Server) SubscribeCommitteeSubnets(ctx context.Context, req *ethpb.Comm
 	return &emptypb.Empty{}, nil
 }
 
-func (vs *Server) proposeAtt(ctx context.Context, att ethpb.Att, committee primitives.CommitteeIndex) (*ethpb.AttestResponse, error) {
+func (vs *Server) proposeAtt(
+	ctx context.Context,
+	att ethpb.Att,
+	committee []primitives.ValidatorIndex, // required post-Electra
+	committeeIndex primitives.CommitteeIndex,
+) (*ethpb.AttestResponse, error) {
 	if _, err := bls.SignatureFromBytes(att.GetSignature()); err != nil {
 		return nil, status.Error(codes.InvalidArgument, "Incorrect attestation signature")
 	}
 
 	root, err := att.GetData().HashTreeRoot()
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Could not tree hash attestation: %v", err)
+		return nil, status.Errorf(codes.Internal, "Could not get attestation root: %v", err)
+	}
+
+	var singleAtt *ethpb.SingleAttestation
+	if att.Version() >= version.Electra {
+		var ok bool
+		singleAtt, ok = att.(*ethpb.SingleAttestation)
+		if !ok {
+			return nil, status.Errorf(codes.Internal, "Attestation has wrong type (expected %T, got %T)", &ethpb.SingleAttestation{}, att)
+		}
+		att = singleAtt.ToAttestationElectra(committee)
 	}
 
 	// Broadcast the unaggregated attestation on a feed to notify other services in the beacon node
@@ -161,10 +184,16 @@ func (vs *Server) proposeAtt(ctx context.Context, att ethpb.Att, committee primi
 	if err != nil {
 		return nil, err
 	}
-	subnet := helpers.ComputeSubnetFromCommitteeAndSlot(uint64(len(vals)), committee, att.GetData().Slot)
+	subnet := helpers.ComputeSubnetFromCommitteeAndSlot(uint64(len(vals)), committeeIndex, att.GetData().Slot)
 
 	// Broadcast the new attestation to the network.
-	if err := vs.P2P.BroadcastAttestation(ctx, subnet, att); err != nil {
+	var attToBroadcast ethpb.Att
+	if singleAtt != nil {
+		attToBroadcast = singleAtt
+	} else {
+		attToBroadcast = att
+	}
+	if err := vs.P2P.BroadcastAttestation(ctx, subnet, attToBroadcast); err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not broadcast attestation: %v", err)
 	}