Add Recent Blocks RPC Request Handler (#3281)

* add new rpc handler

* gaz

* add it back

* remove ok

* preston's comments

beacon-chain/sync/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "rpc_beacon_blocks.go",
         "rpc_goodbye.go",
         "rpc_hello.go",
+        "rpc_recent_beacon_blocks.go",
         "service.go",
         "subscriber.go",
         "subscriber_beacon_blocks.go",
@@ -57,6 +58,7 @@ go_test(
         "rpc_beacon_blocks_test.go",
         "rpc_goodbye_test.go",
         "rpc_hello_test.go",
+        "rpc_recent_beacon_blocks_test.go",
         "rpc_test.go",
         "subscriber_test.go",
         "validate_attetser_slashing_test.go",

beacon-chain/sync/rpc.go

@@ -46,13 +46,8 @@ func (r *RegularSync) registerRPCHandlers() {
 	)
 	r.registerRPC(
 		"/eth2/beacon_chain/req/recent_beacon_blocks/1",
-		nil,
-		notImplementedRPCHandler, // TODO(3147): Implement.
-	)
-	r.registerRPC(
-		"/eth2/beacon_chain/req/beacon_blocks/1",
-		nil,
-		notImplementedRPCHandler, // TODO(3147): Implement.
+		&pb.RecentBeaconBlocksRequest{},
+		r.recentBeaconBlocksRPCHandler,
 	)
 }
 

beacon-chain/sync/rpc_recent_beacon_blocks.go

@@ -0,0 +1,59 @@
+package sync
+
+import (
+	"context"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+	libp2pcore "github.com/libp2p/go-libp2p-core"
+	"github.com/pkg/errors"
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	"github.com/prysmaticlabs/prysm/shared/bytesutil"
+)
+
+// recentBeaconBlocksRPCHandler looks up the request blocks from the database from the given block roots.
+func (r *RegularSync) recentBeaconBlocksRPCHandler(ctx context.Context, msg proto.Message, stream libp2pcore.Stream) error {
+	defer stream.Close()
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	setRPCStreamDeadlines(stream)
+	log := log.WithField("handler", "recent_beacon_blocks")
+
+	m := msg.(*pb.RecentBeaconBlocksRequest)
+	blockRoots := m.BlockRoots
+	if len(blockRoots) == 0 {
+		resp, err := r.generateErrorResponse(responseCodeInvalidRequest, "no block roots provided in request")
+		if err != nil {
+			log.WithError(err).Error("Failed to generate a response error")
+		} else {
+			if _, err := stream.Write(resp); err != nil {
+				log.WithError(err).Errorf("Failed to write to stream")
+			}
+		}
+		return errors.New("no block roots provided")
+	}
+	ret := &pb.BeaconBlocksResponse{}
+	for _, root := range blockRoots {
+		blk, err := r.db.Block(ctx, bytesutil.ToBytes32(root))
+		if err != nil {
+			log.WithError(err).Error("Failed to fetch block")
+			resp, err := r.generateErrorResponse(responseCodeServerError, genericError)
+			if err != nil {
+				log.WithError(err).Error("Failed to generate a response error")
+			} else {
+				if _, err := stream.Write(resp); err != nil {
+					log.WithError(err).Errorf("Failed to write to stream")
+				}
+			}
+			return err
+		}
+		// if block returned is nil, it appends nil to the slice
+		ret.Blocks = append(ret.Blocks, blk)
+	}
+
+	if _, err := stream.Write([]byte{responseCodeSuccess}); err != nil {
+		log.WithError(err).Error("Failed to write to stream")
+	}
+	_, err := r.p2p.Encoding().Encode(stream, ret)
+	return err
+}

beacon-chain/sync/rpc_recent_beacon_blocks_test.go

@@ -0,0 +1,83 @@
+package sync
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/network"
+	"github.com/libp2p/go-libp2p-core/protocol"
+	"github.com/prysmaticlabs/go-ssz"
+	db "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
+	p2ptest "github.com/prysmaticlabs/prysm/beacon-chain/p2p/testing"
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
+	"github.com/prysmaticlabs/prysm/shared/testutil"
+)
+
+func TestRecentBeaconBlocksRPCHandler_ReturnsBlocks(t *testing.T) {
+	p1 := p2ptest.NewTestP2P(t)
+	p2 := p2ptest.NewTestP2P(t)
+	p1.Connect(p2)
+	if len(p1.Host.Network().Peers()) != 1 {
+		t.Error("Expected peers to be connected")
+	}
+	d := db.SetupDB(t)
+	defer db.TeardownDB(t, d)
+
+	var blkRoots [][]byte
+	// Populate the database with blocks that would match the request.
+	for i := 1; i < 11; i++ {
+		blk := &ethpb.BeaconBlock{
+			Slot: uint64(i),
+		}
+		root, err := ssz.SigningRoot(blk)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := d.SaveBlock(context.Background(), blk); err != nil {
+			t.Fatal(err)
+		}
+		blkRoots = append(blkRoots, root[:])
+	}
+	req := &pb.RecentBeaconBlocksRequest{
+		BlockRoots: blkRoots,
+	}
+
+	r := &RegularSync{p2p: p1, db: d}
+	pcl := protocol.ID("/testing")
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	p2.Host.SetStreamHandler(pcl, func(stream network.Stream) {
+		defer wg.Done()
+		expectSuccess(t, r, stream)
+		res := &pb.BeaconBlocksResponse{}
+		if err := r.p2p.Encoding().Decode(stream, res); err != nil {
+			t.Error(err)
+		}
+		if len(res.Blocks) != len(req.BlockRoots) {
+			t.Errorf("Received only %d blocks, expected %d", len(res.Blocks), len(req.BlockRoots))
+		}
+		for i, blk := range res.Blocks {
+			if blk.Slot != uint64(i+1) {
+				t.Errorf("Received unexpected block slot %d but wanted %d", blk.Slot, i+1)
+			}
+		}
+	})
+
+	stream1, err := p1.Host.NewStream(context.Background(), p2.Host.ID(), pcl)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = r.recentBeaconBlocksRPCHandler(context.Background(), req, stream1)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	if testutil.WaitTimeout(&wg, 1*time.Second) {
+		t.Fatal("Did not receive stream within 1 sec")
+	}
+}