Implement SubmitAttestations in the beacon API (#8563)

* update ethereumapis deps

* V1AttToV1Alpha1 migration

* Implementation plus happy path test

* fix root variable names

* Invalid attestation test

* gzl

* mod tidy

* use a single append to concatenate two slices

* remove outdated comment from attestation processing

* invoke ProcessAttestationNoVerifySignature when validating attestations

* implement missing PoolMock members

* use new VerifyAttestationNoVerifySignature function

beacon-chain/operations/attestations/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     srcs = [
         "log.go",
         "metrics.go",
+        "mock.go",
         "pool.go",
         "prepare_forkchoice.go",
         "prune_expired.go",

beacon-chain/operations/attestations/mock.go

@@ -0,0 +1,113 @@
+package attestations
+
+import (
+	"context"
+
+	types "github.com/prysmaticlabs/eth2-types"
+	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
+)
+
+type PoolMock struct {
+	AggregatedAtts []*ethpb.Attestation
+}
+
+func (*PoolMock) AggregateUnaggregatedAttestations(_ context.Context) error {
+	panic("implement me")
+}
+
+func (*PoolMock) AggregateUnaggregatedAttestationsBySlotIndex(_ context.Context, _ types.Slot, _ types.CommitteeIndex) error {
+	panic("implement me")
+}
+
+func (*PoolMock) SaveAggregatedAttestation(_ *ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (m *PoolMock) SaveAggregatedAttestations(atts []*ethpb.Attestation) error {
+	m.AggregatedAtts = append(m.AggregatedAtts, atts...)
+	return nil
+}
+
+func (m *PoolMock) AggregatedAttestations() []*ethpb.Attestation {
+	return m.AggregatedAtts
+}
+
+func (*PoolMock) AggregatedAttestationsBySlotIndex(_ context.Context, _ types.Slot, _ types.CommitteeIndex) []*ethpb.Attestation {
+	panic("implement me")
+}
+
+func (*PoolMock) DeleteAggregatedAttestation(_ *ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) HasAggregatedAttestation(_ *ethpb.Attestation) (bool, error) {
+	panic("implement me")
+}
+
+func (*PoolMock) AggregatedAttestationCount() int {
+	panic("implement me")
+}
+
+func (*PoolMock) SaveUnaggregatedAttestation(_ *ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) SaveUnaggregatedAttestations(_ []*ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) UnaggregatedAttestations() ([]*ethpb.Attestation, error) {
+	panic("implement me")
+}
+
+func (*PoolMock) UnaggregatedAttestationsBySlotIndex(_ context.Context, _ types.Slot, _ types.CommitteeIndex) []*ethpb.Attestation {
+	panic("implement me")
+}
+
+func (*PoolMock) DeleteUnaggregatedAttestation(_ *ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) DeleteSeenUnaggregatedAttestations() (int, error) {
+	panic("implement me")
+}
+
+func (*PoolMock) UnaggregatedAttestationCount() int {
+	panic("implement me")
+}
+
+func (*PoolMock) SaveBlockAttestation(_ *ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) SaveBlockAttestations(_ []*ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) BlockAttestations() []*ethpb.Attestation {
+	panic("implement me")
+}
+
+func (*PoolMock) DeleteBlockAttestation(_ *ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) SaveForkchoiceAttestation(_ *ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) SaveForkchoiceAttestations(_ []*ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) ForkchoiceAttestations() []*ethpb.Attestation {
+	panic("implement me")
+}
+
+func (*PoolMock) DeleteForkchoiceAttestation(_ *ethpb.Attestation) error {
+	panic("implement me")
+}
+
+func (*PoolMock) ForkchoiceAttestationCount() int {
+	panic("implement me")
+}

beacon-chain/p2p/testing/mock_broadcaster.go

@@ -9,12 +9,14 @@ import (
 
 // MockBroadcaster implements p2p.Broadcaster for testing.
 type MockBroadcaster struct {
-	BroadcastCalled bool
+	BroadcastCalled   bool
+	BroadcastMessages []proto.Message
 }
 
 // Broadcast records a broadcast occurred.
-func (m *MockBroadcaster) Broadcast(context.Context, proto.Message) error {
+func (m *MockBroadcaster) Broadcast(_ context.Context, msg proto.Message) error {
 	m.BroadcastCalled = true
+	m.BroadcastMessages = append(m.BroadcastMessages, msg)
 	return nil
 }
 

beacon-chain/rpc/beaconv1/BUILD.bazel

@@ -66,6 +66,7 @@ go_test(
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/testing:go_default_library",
+        "//beacon-chain/operations/attestations:go_default_library",
         "//beacon-chain/operations/slashings:go_default_library",
         "//beacon-chain/operations/voluntaryexits:go_default_library",
         "//beacon-chain/p2p/testing:go_default_library",
@@ -84,5 +85,6 @@ go_test(
         "@com_github_prysmaticlabs_eth2_types//:go_default_library",
         "@com_github_prysmaticlabs_ethereumapis//eth/v1:go_default_library",
         "@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_default_library",
+        "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
     ],
 )

beacon-chain/rpc/beaconv1/pool.go

@@ -6,6 +6,7 @@ import (
 
 	ptypes "github.com/gogo/protobuf/types"
 	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1"
+	ethpb_alpha "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
 	"github.com/prysmaticlabs/prysm/proto/migration"
 	"github.com/prysmaticlabs/prysm/shared/featureconfig"
@@ -22,8 +23,44 @@ func (bs *Server) ListPoolAttestations(ctx context.Context, req *ethpb.Attestati
 
 // SubmitAttestation submits Attestation object to node. If attestation passes all validation
 // constraints, node MUST publish attestation on appropriate subnet.
-func (bs *Server) SubmitAttestation(ctx context.Context, req *ethpb.Attestation) (*ptypes.Empty, error) {
-	return nil, errors.New("unimplemented")
+func (bs *Server) SubmitAttestations(ctx context.Context, req *ethpb.SubmitAttestationsRequest) (*ptypes.Empty, error) {
+	ctx, span := trace.StartSpan(ctx, "beaconv1.SubmitAttestation")
+	defer span.End()
+
+	headState, err := bs.ChainInfoFetcher.HeadState(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Could not get head state: %v", err)
+	}
+
+	var validAttestations []*ethpb_alpha.Attestation
+	for _, sourceAtt := range req.Data {
+		att := migration.V1AttToV1Alpha1(sourceAtt)
+		_, err = blocks.VerifyAttestationNoVerifySignature(ctx, headState, att)
+		if err != nil {
+			continue
+		}
+		err = blocks.VerifyAttestationSignature(ctx, headState, att)
+		if err == nil {
+			validAttestations = append(validAttestations, att)
+		}
+	}
+
+	err = bs.AttestationsPool.SaveAggregatedAttestations(validAttestations)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Could not save attestations: %v", err)
+	}
+	broadcastFailed := false
+	for _, att := range validAttestations {
+		if err := bs.Broadcaster.Broadcast(ctx, att); err != nil {
+			broadcastFailed = true
+		}
+	}
+	if broadcastFailed {
+		return nil, status.Errorf(
+			codes.Internal,
+			"Could not publish one or more attestations. Some attestations could be published successfully.")
+	}
+	return &ptypes.Empty{}, nil
 }
 
 // ListPoolAttesterSlashings retrieves attester slashings known by the node but