Pending Deposits RPC Server Implementation (#1508)

* Add impl for pending deposits via gRPC

* Add test for pending deposits RPC

beacon-chain/rpc/BUILD.bazel

@@ -44,6 +44,7 @@ go_test(
     deps = [
         "//beacon-chain/core/blocks:go_default_library",
         "//beacon-chain/core/state:go_default_library",
+        "//beacon-chain/db:go_default_library",
         "//beacon-chain/internal:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
         "//proto/beacon/rpc/v1:go_default_library",

beacon-chain/rpc/beacon_server.go

@@ -2,13 +2,16 @@ package rpc
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"math/big"
 	"time"
 
 	ptypes "github.com/gogo/protobuf/types"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1"
+	"github.com/prysmaticlabs/prysm/shared/params"
 )
 
 // BeaconServer defines a server implementation of the gRPC Beacon service,
@@ -102,7 +105,15 @@ func (bs *BeaconServer) Eth1Data(ctx context.Context, _ *ptypes.Empty) (*pb.Eth1
 
 // PendingDeposits returns a list of pending deposits that are ready for
 // inclusion in the next beacon block.
-// TODO(1464): Implement this.
 func (bs *BeaconServer) PendingDeposits(ctx context.Context, _ *ptypes.Empty) (*pb.PendingDepositsResponse, error) {
-	return &pb.PendingDepositsResponse{PendingDeposits: []*pbp2p.Deposit{}}, nil
+	bNum := bs.powChainService.LatestBlockNumber()
+
+	if bNum == nil {
+		return nil, errors.New("latest PoW block number is unknown")
+	}
+
+	// Only request deposits that have passed the ETH1 follow distance window.
+	bNum = bNum.Sub(bNum, big.NewInt(int64(params.BeaconConfig().Eth1FollowDistance)))
+
+	return &pb.PendingDepositsResponse{PendingDeposits: bs.beaconDB.PendingDeposits(ctx, bNum)}, nil
 }

beacon-chain/rpc/beacon_server_test.go

@@ -3,6 +3,8 @@ package rpc
 import (
 	"context"
 	"errors"
+	"math/big"
+	"reflect"
 	"testing"
 	"time"
 
@@ -10,9 +12,11 @@ import (
 
 	ptypes "github.com/gogo/protobuf/types"
 	"github.com/golang/mock/gomock"
+	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/beacon-chain/internal"
 	pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1"
+	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/testutil"
 	logTest "github.com/sirupsen/logrus/hooks/test"
 )
@@ -27,9 +31,13 @@ func (f *faultyPOWChainService) HasChainStartLogOccurred() (bool, uint64, error)
 func (f *faultyPOWChainService) ChainStartFeed() *event.Feed {
 	return f.chainStartFeed
 }
+func (f *faultyPOWChainService) LatestBlockNumber() *big.Int {
+	return big.NewInt(0)
+}
 
 type mockPOWChainService struct {
-	chainStartFeed *event.Feed
+	chainStartFeed    *event.Feed
+	latestBlockNumber *big.Int
 }
 
 func (m *mockPOWChainService) HasChainStartLogOccurred() (bool, uint64, error) {
@@ -38,6 +46,9 @@ func (m *mockPOWChainService) HasChainStartLogOccurred() (bool, uint64, error) {
 func (m *mockPOWChainService) ChainStartFeed() *event.Feed {
 	return m.chainStartFeed
 }
+func (m *mockPOWChainService) LatestBlockNumber() *big.Int {
+	return m.latestBlockNumber
+}
 
 func TestWaitForChainStart_ContextClosed(t *testing.T) {
 	hook := logTest.NewGlobal()
@@ -195,3 +206,72 @@ func TestLatestAttestation_SendsCorrectly(t *testing.T) {
 
 	testutil.AssertLogsContain(t, hook, "Sending attestation to RPC clients")
 }
+
+func TestPendingDeposits_ReturnsErrorOnUnknownBlockNum(t *testing.T) {
+	p := &mockPOWChainService{
+		latestBlockNumber: nil,
+	}
+	bs := BeaconServer{powChainService: p}
+
+	_, err := bs.PendingDeposits(context.Background(), nil)
+	if err.Error() != "latest PoW block number is unknown" {
+		t.Errorf("Received unexpected error: %v", err)
+	}
+}
+
+func TestPendingDeposits_ReturnsDepositsOutsideEth1FollowWindow(t *testing.T) {
+	p := &mockPOWChainService{
+		latestBlockNumber: big.NewInt(int64(10 + params.BeaconConfig().Eth1FollowDistance)),
+	}
+	d := &db.BeaconDB{}
+
+	// Using the merkleTreeIndex as the block number for this test...
+	readyDeposits := []*pbp2p.Deposit{
+		&pbp2p.Deposit{
+			MerkleTreeIndex: 1,
+		},
+		&pbp2p.Deposit{
+			MerkleTreeIndex: 2,
+		},
+	}
+
+	recentDeposits := []*pbp2p.Deposit{
+		&pbp2p.Deposit{
+			MerkleTreeIndex: params.BeaconConfig().Eth1FollowDistance + 100,
+		},
+		&pbp2p.Deposit{
+			MerkleTreeIndex: params.BeaconConfig().Eth1FollowDistance + 101,
+		},
+	}
+	ctx := context.Background()
+	for _, dp := range append(recentDeposits, readyDeposits...) {
+		d.InsertPendingDeposit(ctx, dp, big.NewInt(int64(dp.MerkleTreeIndex)))
+	}
+
+	bs := &BeaconServer{
+		beaconDB:        d,
+		powChainService: p,
+	}
+
+	result, err := bs.PendingDeposits(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(result.PendingDeposits, readyDeposits) {
+		t.Errorf("Received unexpected list of deposits: %+v, wanted: %+v", result, readyDeposits)
+	}
+
+	// It should also return the recent deposits after their follow window.
+	p.latestBlockNumber = big.NewInt(0).Add(p.latestBlockNumber, big.NewInt(10000))
+	allResp, err := bs.PendingDeposits(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(allResp.PendingDeposits) != len(recentDeposits)+len(readyDeposits) {
+		t.Errorf(
+			"Received unexpected number of pending deposits: %d, wanted: %d",
+			len(allResp.PendingDeposits),
+			len(recentDeposits)+len(readyDeposits),
+		)
+	}
+}

beacon-chain/rpc/service.go

@@ -4,6 +4,7 @@ package rpc
 import (
 	"context"
 	"fmt"
+	"math/big"
 	"net"
 	"time"
 
@@ -41,6 +42,7 @@ type operationService interface {
 type powChainService interface {
 	HasChainStartLogOccurred() (bool, uint64, error)
 	ChainStartFeed() *event.Feed
+	LatestBlockNumber() *big.Int
 }
 
 // Service defining an RPC server for a beacon node.