RPC: Stream altair block (#9521)

* Add tests

* Fix tests

* Fix test

Co-authored-by: Raul Jordan <raul@prysmaticlabs.com>
Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

beacon-chain/rpc/prysm/v1alpha1/validator/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "aggregator.go",
         "assignments.go",
         "attester.go",
+        "blocks.go",
         "exit.go",
         "log.go",
         "proposer.go",
@@ -49,6 +50,7 @@ go_library(
         "//shared/bytesutil:go_default_library",
         "//shared/copyutil:go_default_library",
         "//shared/depositutil:go_default_library",
+        "//shared/event:go_default_library",
         "//shared/featureconfig:go_default_library",
         "//shared/hashutil:go_default_library",
         "//shared/params:go_default_library",
@@ -79,6 +81,7 @@ go_test(
         "aggregator_test.go",
         "assignments_test.go",
         "attester_test.go",
+        "blocks_test.go",
         "exit_test.go",
         "proposer_attestations_test.go",
         "proposer_sync_aggregate_test.go",
@@ -96,6 +99,7 @@ go_test(
         "//beacon-chain/core/altair:go_default_library",
         "//beacon-chain/core/blocks:go_default_library",
         "//beacon-chain/core/feed:go_default_library",
+        "//beacon-chain/core/feed/block:go_default_library",
         "//beacon-chain/core/feed/operation:go_default_library",
         "//beacon-chain/core/feed/state:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",

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

@@ -0,0 +1,108 @@
+package validator
+
+import (
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/feed"
+	blockfeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/block"
+	statefeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/state"
+	ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/shared/event"
+	"github.com/prysmaticlabs/prysm/shared/version"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// StreamBlocksAltair to clients every single time a block is received by the beacon node.
+func (bs *Server) StreamBlocksAltair(req *ethpb.StreamBlocksRequest, stream ethpb.BeaconNodeValidator_StreamBlocksAltairServer) error {
+	blocksChannel := make(chan *feed.Event, 1)
+	var blockSub event.Subscription
+	if req.VerifiedOnly {
+		blockSub = bs.StateNotifier.StateFeed().Subscribe(blocksChannel)
+	} else {
+		blockSub = bs.BlockNotifier.BlockFeed().Subscribe(blocksChannel)
+	}
+	defer blockSub.Unsubscribe()
+
+	for {
+		select {
+		case blockEvent := <-blocksChannel:
+			if req.VerifiedOnly {
+				if blockEvent.Type == statefeed.BlockProcessed {
+					data, ok := blockEvent.Data.(*statefeed.BlockProcessedData)
+					if !ok || data == nil {
+						continue
+					}
+					b := &ethpb.StreamBlocksResponse{}
+					switch data.SignedBlock.Version() {
+					case version.Phase0:
+						phBlk, ok := data.SignedBlock.Proto().(*ethpb.SignedBeaconBlock)
+						if !ok {
+							log.Warn("Mismatch between version and block type, was expecting *ethpb.SignedBeaconBlock")
+							continue
+						}
+						b.Block = &ethpb.StreamBlocksResponse_Phase0Block{Phase0Block: phBlk}
+					case version.Altair:
+						phBlk, ok := data.SignedBlock.Proto().(*ethpb.SignedBeaconBlockAltair)
+						if !ok {
+							log.Warn("Mismatch between version and block type, was expecting *v2.SignedBeaconBlockAltair")
+							continue
+						}
+						b.Block = &ethpb.StreamBlocksResponse_AltairBlock{AltairBlock: phBlk}
+					}
+
+					if err := stream.Send(b); err != nil {
+						return status.Errorf(codes.Unavailable, "Could not send over stream: %v", err)
+					}
+				}
+			} else {
+				if blockEvent.Type == blockfeed.ReceivedBlock {
+					data, ok := blockEvent.Data.(*blockfeed.ReceivedBlockData)
+					if !ok {
+						// Got bad data over the stream.
+						continue
+					}
+					if data.SignedBlock == nil {
+						// One nil block shouldn't stop the stream.
+						continue
+					}
+					headState, err := bs.HeadFetcher.HeadState(bs.Ctx)
+					if err != nil {
+						log.WithError(err).WithField("blockSlot", data.SignedBlock.Block().Slot()).Error("Could not get head state")
+						continue
+					}
+					signed := data.SignedBlock
+					if err := blocks.VerifyBlockSignature(headState, signed.Block().ProposerIndex(), signed.Signature(), signed.Block().HashTreeRoot); err != nil {
+						log.WithError(err).WithField("blockSlot", data.SignedBlock.Block().Slot()).Error("Could not verify block signature")
+						continue
+					}
+					b := &ethpb.StreamBlocksResponse{}
+					switch data.SignedBlock.Version() {
+					case version.Phase0:
+						phBlk, ok := data.SignedBlock.Proto().(*ethpb.SignedBeaconBlock)
+						if !ok {
+							log.Warn("Mismatch between version and block type, was expecting *ethpb.SignedBeaconBlock")
+							continue
+						}
+						b.Block = &ethpb.StreamBlocksResponse_Phase0Block{Phase0Block: phBlk}
+					case version.Altair:
+						phBlk, ok := data.SignedBlock.Proto().(*ethpb.SignedBeaconBlockAltair)
+						if !ok {
+							log.Warn("Mismatch between version and block type, was expecting *v2.SignedBeaconBlockAltair")
+							continue
+						}
+						b.Block = &ethpb.StreamBlocksResponse_AltairBlock{AltairBlock: phBlk}
+					}
+					if err := stream.Send(b); err != nil {
+						return status.Errorf(codes.Unavailable, "Could not send over stream: %v", err)
+					}
+				}
+			}
+		case <-blockSub.Err():
+			return status.Error(codes.Aborted, "Subscriber closed, exiting goroutine")
+		case <-bs.Ctx.Done():
+			return status.Error(codes.Canceled, "Context canceled")
+		case <-stream.Context().Done():
+			return status.Error(codes.Canceled, "Context canceled")
+		}
+	}
+}

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

@@ -0,0 +1,158 @@
+package validator
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	chainMock "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/altair"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/feed"
+	blockfeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/block"
+	statefeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/state"
+	dbTest "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
+	ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1/wrapper"
+	"github.com/prysmaticlabs/prysm/shared/mock"
+	"github.com/prysmaticlabs/prysm/shared/params"
+	"github.com/prysmaticlabs/prysm/shared/testutil"
+	"github.com/prysmaticlabs/prysm/shared/testutil/assert"
+	"github.com/prysmaticlabs/prysm/shared/testutil/require"
+)
+
+func TestServer_StreamAltairBlocksVerified_ContextCanceled(t *testing.T) {
+	ctx := context.Background()
+
+	chainService := &chainMock.ChainService{}
+	ctx, cancel := context.WithCancel(ctx)
+	server := &Server{
+		Ctx:           ctx,
+		StateNotifier: chainService.StateNotifier(),
+		HeadFetcher:   chainService,
+	}
+
+	exitRoutine := make(chan bool)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockStream := mock.NewMockBeaconNodeValidatorAltair_StreamBlocksServer(ctrl)
+	mockStream.EXPECT().Context().Return(ctx)
+	go func(tt *testing.T) {
+		assert.ErrorContains(tt, "Context canceled", server.StreamBlocksAltair(&ethpb.StreamBlocksRequest{
+			VerifiedOnly: true,
+		}, mockStream))
+		<-exitRoutine
+	}(t)
+	cancel()
+	exitRoutine <- true
+}
+
+func TestServer_StreamAltairBlocks_ContextCanceled(t *testing.T) {
+	ctx := context.Background()
+
+	chainService := &chainMock.ChainService{}
+	ctx, cancel := context.WithCancel(ctx)
+	server := &Server{
+		Ctx:           ctx,
+		BlockNotifier: chainService.BlockNotifier(),
+		HeadFetcher:   chainService,
+	}
+
+	exitRoutine := make(chan bool)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockStream := mock.NewMockBeaconNodeValidatorAltair_StreamBlocksServer(ctrl)
+	mockStream.EXPECT().Context().Return(ctx)
+	go func(tt *testing.T) {
+		assert.ErrorContains(tt, "Context canceled", server.StreamBlocksAltair(&ethpb.StreamBlocksRequest{}, mockStream))
+		<-exitRoutine
+	}(t)
+	cancel()
+	exitRoutine <- true
+}
+
+func TestServer_StreamAltairBlocks_OnHeadUpdated(t *testing.T) {
+	params.UseMainnetConfig()
+	ctx := context.Background()
+	beaconState, privs := testutil.DeterministicGenesisStateAltair(t, 64)
+	c, err := altair.NextSyncCommittee(ctx, beaconState)
+	require.NoError(t, err)
+	require.NoError(t, beaconState.SetCurrentSyncCommittee(c))
+
+	b, err := testutil.GenerateFullBlockAltair(beaconState, privs, testutil.DefaultBlockGenConfig(), 1)
+	require.NoError(t, err)
+	chainService := &chainMock.ChainService{State: beaconState}
+	server := &Server{
+		Ctx:           ctx,
+		BlockNotifier: chainService.BlockNotifier(),
+		HeadFetcher:   chainService,
+	}
+	exitRoutine := make(chan bool)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockStream := mock.NewMockBeaconNodeValidatorAltair_StreamBlocksServer(ctrl)
+
+	mockStream.EXPECT().Send(&ethpb.StreamBlocksResponse{Block: &ethpb.StreamBlocksResponse_AltairBlock{AltairBlock: b}}).Do(func(arg0 interface{}) {
+		exitRoutine <- true
+	})
+	mockStream.EXPECT().Context().Return(ctx).AnyTimes()
+
+	go func(tt *testing.T) {
+		assert.NoError(tt, server.StreamBlocksAltair(&ethpb.StreamBlocksRequest{}, mockStream), "Could not call RPC method")
+	}(t)
+	wrappedBlk, err := wrapper.WrappedAltairSignedBeaconBlock(b)
+	require.NoError(t, err)
+	// Send in a loop to ensure it is delivered (busy wait for the service to subscribe to the state feed).
+	for sent := 0; sent == 0; {
+		sent = server.BlockNotifier.BlockFeed().Send(&feed.Event{
+			Type: blockfeed.ReceivedBlock,
+			Data: &blockfeed.ReceivedBlockData{SignedBlock: wrappedBlk},
+		})
+	}
+	<-exitRoutine
+}
+
+func TestServer_StreamAltairBlocksVerified_OnHeadUpdated(t *testing.T) {
+	params.UseMainnetConfig()
+	db := dbTest.SetupDB(t)
+	ctx := context.Background()
+	beaconState, privs := testutil.DeterministicGenesisStateAltair(t, 32)
+	c, err := altair.NextSyncCommittee(ctx, beaconState)
+	require.NoError(t, err)
+	require.NoError(t, beaconState.SetCurrentSyncCommittee(c))
+
+	b, err := testutil.GenerateFullBlockAltair(beaconState, privs, testutil.DefaultBlockGenConfig(), 1)
+	require.NoError(t, err)
+	r, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+	wrappedBlk, err := wrapper.WrappedAltairSignedBeaconBlock(b)
+	require.NoError(t, err)
+	require.NoError(t, db.SaveBlock(ctx, wrappedBlk))
+	chainService := &chainMock.ChainService{State: beaconState}
+	server := &Server{
+		Ctx:           ctx,
+		StateNotifier: chainService.StateNotifier(),
+		HeadFetcher:   chainService,
+	}
+	exitRoutine := make(chan bool)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockStream := mock.NewMockBeaconNodeValidatorAltair_StreamBlocksServer(ctrl)
+	mockStream.EXPECT().Send(&ethpb.StreamBlocksResponse{Block: &ethpb.StreamBlocksResponse_AltairBlock{AltairBlock: b}}).Do(func(arg0 interface{}) {
+		exitRoutine <- true
+	})
+	mockStream.EXPECT().Context().Return(ctx).AnyTimes()
+
+	go func(tt *testing.T) {
+		assert.NoError(tt, server.StreamBlocksAltair(&ethpb.StreamBlocksRequest{
+			VerifiedOnly: true,
+		}, mockStream), "Could not call RPC method")
+	}(t)
+	// Send in a loop to ensure it is delivered (busy wait for the service to subscribe to the state feed).
+	for sent := 0; sent == 0; {
+		sent = server.StateNotifier.StateFeed().Send(&feed.Event{
+			Type: statefeed.BlockProcessed,
+			Data: &statefeed.BlockProcessedData{Slot: b.Block.Slot, BlockRoot: r, SignedBlock: wrappedBlk},
+		})
+	}
+	<-exitRoutine
+}

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

@@ -1197,6 +1197,7 @@ func TestProposer_ValidateDepositTrie(t *testing.T) {
 }
 
 func TestProposer_Eth1Data_NoBlockExists(t *testing.T) {
+	params.UseMinimalConfig()
 	ctx := context.Background()
 
 	height := big.NewInt(int64(params.BeaconConfig().Eth1FollowDistance))

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

@@ -73,10 +73,6 @@ func (vs *Server) ProposeBlockAltair(ctx context.Context, altair *ethpb.SignedBe
 	return nil, status.Error(codes.Unimplemented, "Unimplemented")
 }
 
-func (vs *Server) StreamBlocksAltair(request *ethpb.StreamBlocksRequest, server ethpb.BeaconNodeValidator_StreamBlocksAltairServer) error {
-	return status.Error(codes.Unimplemented, "Unimplemented")
-}
-
 // WaitForActivation checks if a validator public key exists in the active validator registry of the current
 // beacon state, if not, then it creates a stream which listens for canonical states which contain
 // the validator with the public key as an active validator record.