Implement StreamAttestations RPC Endpoint

beacon-chain/rpc/beacon/BUILD.bazel

@@ -29,6 +29,7 @@ go_library(
         "//shared/hashutil:go_default_library",
         "//shared/pagination:go_default_library",
         "//shared/params:go_default_library",
+        "//shared/slotutil:go_default_library",
         "@com_github_gogo_protobuf//types:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_default_library",
@@ -56,6 +57,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/rpc/testing:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
         "//shared/params:go_default_library",

beacon-chain/rpc/beacon/attestations.go

@@ -11,6 +11,8 @@ import (
 	"github.com/prysmaticlabs/prysm/beacon-chain/db/filters"
 	"github.com/prysmaticlabs/prysm/shared/pagination"
 	"github.com/prysmaticlabs/prysm/shared/params"
+	"github.com/prysmaticlabs/prysm/shared/slotutil"
+
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -113,12 +115,29 @@ func (bs *Server) ListAttestations(
 	}, nil
 }
 
-// StreamAttestations to clients every single time a new attestation is received.
-// TODO(#4184): Implement.
+// StreamAttestations to clients at the end of every slot. This method retrieves the
+// aggregated attestations currently in the pool at the start of a slot and sends
+// them over a gRPC stream.
 func (bs *Server) StreamAttestations(
-	_ *ptypes.Empty, _ ethpb.BeaconChain_StreamAttestationsServer,
+	_ *ptypes.Empty, stream ethpb.BeaconChain_StreamAttestationsServer,
 ) error {
-	return status.Error(codes.Unimplemented, "Not yet implemented")
+	genesisTime := bs.GenesisTimeFetcher.GenesisTime()
+	st := slotutil.GetSlotTicker(genesisTime, params.BeaconConfig().SecondsPerSlot)
+	for {
+		select {
+		case <-st.C():
+			atts := bs.Pool.AggregatedAttestations()
+			for i := 0; i < len(atts); i++ {
+				if err := stream.Send(atts[i]); err != nil {
+					return status.Errorf(codes.Unavailable, "Could not send over stream: %v", err)
+				}
+			}
+		case <-bs.Ctx.Done():
+			return status.Error(codes.Canceled, "Context canceled")
+		case <-stream.Context().Done():
+			return status.Error(codes.Canceled, "Context canceled")
+		}
+	}
 }
 
 // AttestationPool retrieves pending attestations.

beacon-chain/rpc/beacon/attestations_test.go

@@ -7,13 +7,18 @@ import (
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/gogo/protobuf/proto"
+	ptypes "github.com/gogo/protobuf/types"
+	"github.com/golang/mock/gomock"
 	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
 	"github.com/prysmaticlabs/go-bitfield"
 	"github.com/prysmaticlabs/go-ssz"
 	mock "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing"
 	dbTest "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
+	"github.com/prysmaticlabs/prysm/beacon-chain/operations/attestations"
+	mockRPC "github.com/prysmaticlabs/prysm/beacon-chain/rpc/testing"
 	pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	"github.com/prysmaticlabs/prysm/shared/params"
 )
@@ -513,3 +518,78 @@ func TestServer_ListAttestations_Pagination_DefaultPageSize(t *testing.T) {
 		t.Error("Incorrect attestations response")
 	}
 }
+
+func TestServer_StreamAttestations_ContextCanceled(t *testing.T) {
+	db := dbTest.SetupDB(t)
+	defer dbTest.TeardownDB(t, db)
+	ctx := context.Background()
+
+	ctx, cancel := context.WithCancel(ctx)
+	chainService := &mock.ChainService{
+		Genesis: time.Now(),
+	}
+	server := &Server{
+		Ctx:                ctx,
+		GenesisTimeFetcher: chainService,
+	}
+
+	exitRoutine := make(chan bool)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockStream := mockRPC.NewMockBeaconChain_StreamAttestationsServer(ctrl)
+	mockStream.EXPECT().Context().Return(ctx)
+	go func(tt *testing.T) {
+		if err := server.StreamAttestations(
+			&ptypes.Empty{},
+			mockStream,
+		); !strings.Contains(err.Error(), "Context canceled") {
+			tt.Errorf("Expected context canceled error got: %v", err)
+		}
+		<-exitRoutine
+	}(t)
+	cancel()
+	exitRoutine <- true
+}
+
+func TestServer_StreamAttestations_OnSlotTick(t *testing.T) {
+	db := dbTest.SetupDB(t)
+	defer dbTest.TeardownDB(t, db)
+	exitRoutine := make(chan bool)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	ctx := context.Background()
+	secondsPerSlot := time.Second * time.Duration(params.BeaconConfig().SecondsPerSlot)
+	chainService := &mock.ChainService{
+		Genesis: time.Now().Add(-secondsPerSlot),
+	}
+	server := &Server{
+		Ctx:                ctx,
+		GenesisTimeFetcher: chainService,
+		Pool:               attestations.NewPool(),
+	}
+
+	atts := []*ethpb.Attestation{
+		{Data: &ethpb.AttestationData{Slot: 1}, AggregationBits: bitfield.Bitlist{0b1101}},
+		{Data: &ethpb.AttestationData{Slot: 2}, AggregationBits: bitfield.Bitlist{0b1101}},
+		{Data: &ethpb.AttestationData{Slot: 3}, AggregationBits: bitfield.Bitlist{0b1101}},
+	}
+	if err := server.Pool.SaveAggregatedAttestations(atts); err != nil {
+		t.Fatal(err)
+	}
+
+	mockStream := mockRPC.NewMockBeaconChain_StreamAttestationsServer(ctrl)
+	mockStream.EXPECT().Send(atts[0])
+	mockStream.EXPECT().Send(atts[1])
+	mockStream.EXPECT().Send(atts[2]).Do(func(arg0 interface{}) {
+		exitRoutine <- true
+	})
+	mockStream.EXPECT().Context().Return(ctx).AnyTimes()
+
+	go func(tt *testing.T) {
+		if err := server.StreamAttestations(&ptypes.Empty{}, mockStream); err != nil {
+			tt.Errorf("Could not call RPC method: %v", err)
+		}
+	}(t)
+
+	<-exitRoutine
+}

beacon-chain/rpc/beacon/server.go

@@ -27,4 +27,5 @@ type Server struct {
 	IncomingAttestation chan *ethpb.Attestation
 	CanonicalStateChan  chan *pbp2p.BeaconState
 	ChainStartChan      chan time.Time
+	GenesisTimeFetcher  blockchain.GenesisTimeFetcher
 }

beacon-chain/rpc/service.go

@@ -213,6 +213,7 @@ func (s *Service) Start() {
 		ChainStartFetcher:   s.chainStartFetcher,
 		CanonicalStateChan:  s.canonicalStateChan,
 		StateNotifier:       s.stateNotifier,
+		GenesisTimeFetcher:  s.genesisTimeFetcher,
 	}
 	aggregatorServer := &aggregator.Server{
 		BeaconDB:    s.beaconDB,