Do Not Broadcast Attestations in Operations Service (#2509)

* no att broadcast

* broadcast in rpc but not operations

* fix space

* tests

beacon-chain/node/node.go

@@ -338,6 +338,11 @@ func (b *BeaconNode) registerRPCService(ctx *cli.Context) error {
 		return err
 	}
 
+	var p2pService *p2p.Server
+	if err := b.services.FetchService(&p2pService); err != nil {
+		return err
+	}
+
 	var operationService *operations.Service
 	if err := b.services.FetchService(&operationService); err != nil {
 		return err
@@ -361,6 +366,7 @@ func (b *BeaconNode) registerRPCService(ctx *cli.Context) error {
 		CertFlag:         cert,
 		KeyFlag:          key,
 		BeaconDB:         b.db,
+		Broadcaster:      p2pService,
 		ChainService:     chainService,
 		OperationService: operationService,
 		POWChainService:  web3Service,

beacon-chain/operations/service.go

@@ -205,9 +205,6 @@ func (s *Service) HandleAttestations(ctx context.Context, message proto.Message)
 	if err := s.beaconDB.SaveAttestation(ctx, attestation); err != nil {
 		return err
 	}
-	s.p2p.Broadcast(ctx, &pb.AttestationAnnounce{
-		Hash: hash[:],
-	})
 	return nil
 }
 

beacon-chain/operations/service_test.go

@@ -21,11 +21,9 @@ import (
 var _ = OperationFeeds(&Service{})
 
 type mockBroadcaster struct {
-	broadcastCalled bool
 }
 
 func (mb *mockBroadcaster) Broadcast(_ context.Context, _ proto.Message) {
-	mb.broadcastCalled = true
 }
 
 func init() {
@@ -116,10 +114,6 @@ func TestIncomingAttestation_OK(t *testing.T) {
 	if err := service.HandleAttestations(context.Background(), attestation); err != nil {
 		t.Error(err)
 	}
-
-	if !broadcaster.broadcastCalled {
-		t.Error("Attestation was not broadcasted")
-	}
 }
 
 func TestRetrieveAttestations_OK(t *testing.T) {

beacon-chain/rpc/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "//shared/event:go_default_library",
         "//shared/featureconfig:go_default_library",
         "//shared/hashutil:go_default_library",
+        "//shared/p2p:go_default_library",
         "//shared/params:go_default_library",
         "//shared/trieutil:go_default_library",
         "@com_github_ethereum_go_ethereum//common:go_default_library",

beacon-chain/rpc/attester_server.go

@@ -11,12 +11,14 @@ import (
 	pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1"
 	"github.com/prysmaticlabs/prysm/shared/hashutil"
+	"github.com/prysmaticlabs/prysm/shared/p2p"
 	"github.com/prysmaticlabs/prysm/shared/params"
 )
 
 // AttesterServer defines a server implementation of the gRPC Attester service,
 // providing RPC methods for validators acting as attesters to broadcast votes on beacon blocks.
 type AttesterServer struct {
+	p2p              p2p.Broadcaster
 	beaconDB         *db.BeaconDB
 	operationService operationService
 }
@@ -32,7 +34,9 @@ func (as *AttesterServer) AttestHead(ctx context.Context, att *pbp2p.Attestation
 	if err := as.operationService.HandleAttestations(ctx, att); err != nil {
 		return nil, err
 	}
-
+	as.p2p.Broadcast(ctx, &pbp2p.AttestationAnnounce{
+		Hash: h[:],
+	})
 	return &pb.AttestResponse{AttestationHash: h[:]}, nil
 }
 

beacon-chain/rpc/attester_server_test.go

@@ -13,10 +13,16 @@ import (
 	"github.com/prysmaticlabs/prysm/shared/params"
 )
 
+type mockBroadcaster struct{}
+
+func (m *mockBroadcaster) Broadcast(ctx context.Context, msg proto.Message) {
+}
+
 func TestAttestHead_OK(t *testing.T) {
 	mockOperationService := &mockOperationService{}
 	attesterServer := &AttesterServer{
 		operationService: mockOperationService,
+		p2p:              &mockBroadcaster{},
 	}
 	req := &pbp2p.Attestation{
 		Data: &pbp2p.AttestationData{
@@ -73,6 +79,7 @@ func TestAttestationDataAtSlot_OK(t *testing.T) {
 	beaconState.LatestBlockRootHash32S[2*params.BeaconConfig().SlotsPerEpoch] = justifiedBlockRoot[:]
 	attesterServer := &AttesterServer{
 		beaconDB: db,
+		p2p:      &mockBroadcaster{},
 	}
 	if err := attesterServer.beaconDB.SaveBlock(epochBoundaryBlock); err != nil {
 		t.Fatalf("Could not save block in test db: %v", err)
@@ -168,6 +175,7 @@ func TestAttestationDataAtSlot_handlesFarAwayJustifiedEpoch(t *testing.T) {
 	beaconState.LatestBlockRootHash32S[2*params.BeaconConfig().SlotsPerEpoch] = justifiedBlockRoot[:]
 	attesterServer := &AttesterServer{
 		beaconDB: db,
+		p2p:      &mockBroadcaster{},
 	}
 	if err := attesterServer.beaconDB.SaveBlock(epochBoundaryBlock); err != nil {
 		t.Fatalf("Could not save block in test db: %v", err)

beacon-chain/rpc/service.go

@@ -18,6 +18,7 @@ import (
 	pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1"
 	"github.com/prysmaticlabs/prysm/shared/event"
+	"github.com/prysmaticlabs/prysm/shared/p2p"
 	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/trieutil"
 	"github.com/sirupsen/logrus"
@@ -78,6 +79,7 @@ type Service struct {
 	canonicalStateChan  chan *pbp2p.BeaconState
 	incomingAttestation chan *pbp2p.Attestation
 	credentialError     error
+	p2p                 p2p.Broadcaster
 }
 
 // Config options for the beacon node RPC server.
@@ -90,6 +92,7 @@ type Config struct {
 	POWChainService  powChainService
 	OperationService operationService
 	SyncService      syncService
+	Broadcaster      p2p.Broadcaster
 }
 
 // NewRPCService creates a new instance of a struct implementing the BeaconServiceServer
@@ -100,6 +103,7 @@ func NewRPCService(ctx context.Context, cfg *Config) *Service {
 		ctx:                 ctx,
 		cancel:              cancel,
 		beaconDB:            cfg.BeaconDB,
+		p2p:                 cfg.Broadcaster,
 		chainService:        cfg.ChainService,
 		powChainService:     cfg.POWChainService,
 		operationService:    cfg.OperationService,
@@ -167,6 +171,7 @@ func (s *Service) Start() {
 	attesterServer := &AttesterServer{
 		beaconDB:         s.beaconDB,
 		operationService: s.operationService,
+		p2p:              s.p2p,
 	}
 	validatorServer := &ValidatorServer{
 		ctx:                s.ctx,