Implement Eth1Data RPC Server Functionality (#1615)

* eth1 data comments

* eth1 data RPC server side impl

* logic is data objects is empty

* finished eth1 server logic

* update comments

* fmt

* big int comparisons

* begin adding eth1 data tests

* empty data votes test runs

* all tests complete with every code branch triggered, behavior correct

* fmt, imports

* address comments

* wrapped up tests

* addressed all comments

beacon-chain/powchain/service.go

@@ -83,7 +83,7 @@ type Web3Service struct {
 	reader                  Reader
 	logger                  bind.ContractFilterer
 	blockFetcher            POWBlockFetcher
-	blockNumber             *big.Int    // the latest ETH1.0 chain blockNumber.
+	blockHeight             *big.Int    // the latest ETH1.0 chain blockHeight.
 	blockHash               common.Hash // the latest ETH1.0 chain blockHash.
 	depositContractCaller   *contracts.DepositContractCaller
 	depositRoot             []byte
@@ -135,7 +135,7 @@ func NewWeb3Service(ctx context.Context, config *Web3ServiceConfig) (*Web3Servic
 		cancel:                  cancel,
 		headerChan:              make(chan *gethTypes.Header),
 		endpoint:                config.Endpoint,
-		blockNumber:             nil,
+		blockHeight:             nil,
 		blockHash:               common.BytesToHash([]byte{}),
 		depositContractAddress:  config.DepositContract,
 		chainStartFeed:          new(event.Feed),
@@ -190,16 +190,34 @@ func (w *Web3Service) Status() error {
 	return nil
 }
 
-// LatestBlockNumber in the ETH1.0 chain.
-func (w *Web3Service) LatestBlockNumber() *big.Int {
-	return w.blockNumber
+// DepositRoot returns the Merkle root of the latest deposit trie
+// from the ETH1.0 deposit contract.
+func (w *Web3Service) DepositRoot() [32]byte {
+	return w.depositTrie.Root()
+}
+
+// LatestBlockHeight in the ETH1.0 chain.
+func (w *Web3Service) LatestBlockHeight() *big.Int {
+	return w.blockHeight
 }
 
 // LatestBlockHash in the ETH1.0 chain.
 func (w *Web3Service) LatestBlockHash() common.Hash {
 	return w.blockHash
 }
 
+// BlockExists --
+// TODO(#1657): Unimplemented, Work in Progress.
+func (w *Web3Service) BlockExists(hash common.Hash) (bool, *big.Int, error) {
+	return false, big.NewInt(0), nil
+}
+
+// BlockHashByHeight --
+// TODO(#1657): Unimplemented, Work in Progress.
+func (w *Web3Service) BlockHashByHeight(height *big.Int) (common.Hash, error) {
+	return [32]byte{}, nil
+}
+
 // Client for interacting with the ETH1.0 chain.
 func (w *Web3Service) Client() Client {
 	return w.client
@@ -355,7 +373,7 @@ func (w *Web3Service) run(done <-chan struct{}) {
 		return
 	}
 
-	w.blockNumber = header.Number
+	w.blockHeight = header.Number
 	w.blockHash = header.Hash()
 
 	// Only process logs if the chain start delay flag is not enabled.
@@ -380,14 +398,14 @@ func (w *Web3Service) run(done <-chan struct{}) {
 			return
 		case header := <-w.headerChan:
 			blockNumberGauge.Set(float64(header.Number.Int64()))
-			w.blockNumber = header.Number
+			w.blockHeight = header.Number
 			w.blockHash = header.Hash()
 			log.WithFields(logrus.Fields{
-				"blockNumber": w.blockNumber,
+				"blockNumber": w.blockHeight,
 				"blockHash":   w.blockHash.Hex(),
 			}).Debug("Latest web3 chain event")
 		case <-ticker.C:
-			if w.lastRequestedBlock.Cmp(w.blockNumber) == 0 {
+			if w.lastRequestedBlock.Cmp(w.blockHeight) == 0 {
 				continue
 			}
 			if err := w.requestBatchedLogs(); err != nil {
@@ -436,7 +454,7 @@ func (w *Web3Service) processPastLogs() error {
 	for _, log := range logs {
 		w.ProcessLog(log)
 	}
-	w.lastRequestedBlock.Set(w.blockNumber)
+	w.lastRequestedBlock.Set(w.blockHeight)
 	return nil
 }
 
@@ -446,7 +464,7 @@ func (w *Web3Service) requestBatchedLogs() error {
 
 	// We request for the nth block behind the current head, in order to have
 	// stabilised logs when we retrieve it from the 1.0 chain.
-	requestedBlock := big.NewInt(0).Sub(w.blockNumber, big.NewInt(params.BeaconConfig().LogBlockDelay))
+	requestedBlock := big.NewInt(0).Sub(w.blockHeight, big.NewInt(params.BeaconConfig().LogBlockDelay))
 	query := ethereum.FilterQuery{
 		Addresses: []common.Address{
 			w.depositContractAddress,

beacon-chain/powchain/service_test.go

@@ -383,8 +383,8 @@ func TestLatestMainchainInfo(t *testing.T) {
 	web3Service.cancel()
 	exitRoutine <- true
 
-	if web3Service.blockNumber.Cmp(header.Number) != 0 {
-		t.Errorf("block number not set, expected %v, got %v", header.Number, web3Service.blockNumber)
+	if web3Service.blockHeight.Cmp(header.Number) != 0 {
+		t.Errorf("block number not set, expected %v, got %v", header.Number, web3Service.blockHeight)
 	}
 
 	if web3Service.blockHash.Hex() != header.Hash().Hex() {

beacon-chain/rpc/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "//shared/hashutil:go_default_library",
         "//shared/params:go_default_library",
         "//shared/ssz:go_default_library",
+        "@com_github_ethereum_go_ethereum//common:go_default_library",
         "@com_github_gogo_protobuf//types:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@org_golang_google_grpc//:go_default_library",
@@ -50,10 +51,12 @@ go_test(
         "//beacon-chain/internal:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
         "//proto/beacon/rpc/v1:go_default_library",
+        "//shared/bytesutil:go_default_library",
         "//shared/event:go_default_library",
         "//shared/params:go_default_library",
         "//shared/ssz:go_default_library",
         "//shared/testutil:go_default_library",
+        "@com_github_ethereum_go_ethereum//common:go_default_library",
         "@com_github_gogo_protobuf//proto:go_default_library",
         "@com_github_gogo_protobuf//types:go_default_library",
         "@com_github_golang_mock//gomock:go_default_library",

beacon-chain/rpc/beacon_server.go

@@ -7,6 +7,8 @@ import (
 	"math/big"
 	"time"
 
+	"github.com/prysmaticlabs/prysm/shared/bytesutil"
+
 	ptypes "github.com/gogo/protobuf/types"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
@@ -97,24 +99,122 @@ func (bs *BeaconServer) LatestAttestation(req *ptypes.Empty, stream pb.BeaconSer
 	}
 }
 
-// Eth1Data fetches the current ETH 1 data which should be used when voting via
-// block proposal.
-// TODO(1463): Implement this.
+// Eth1Data is a mechanism used by block proposers vote on a recent Ethereum 1.0 block hash and an
+// associated deposit root found in the Ethereum 1.0 deposit contract. When consensus is formed,
+// state.latest_eth1_data is updated, and validator deposits up to this root can be processed.
+// The deposit root can be calculated by calling the get_deposit_root() function of
+// the deposit contract using the post-state of the block hash.
 func (bs *BeaconServer) Eth1Data(ctx context.Context, _ *ptypes.Empty) (*pb.Eth1DataResponse, error) {
-	return &pb.Eth1DataResponse{}, nil
+	beaconState, err := bs.beaconDB.State()
+	if err != nil {
+		return nil, fmt.Errorf("could not fetch beacon state: %v", err)
+	}
+	dataVotes := []*pbp2p.Eth1DataVote{}
+	eth1FollowDistance := int64(params.BeaconConfig().Eth1FollowDistance)
+	for _, vote := range beaconState.Eth1DataVotes {
+		eth1Hash := bytesutil.ToBytes32(vote.Eth1Data.BlockHash32)
+		// Verify the block from the vote's block hash exists in the eth1.0 chain and fetch its height.
+		blockExists, blockHeight, err := bs.powChainService.BlockExists(eth1Hash)
+		if err != nil {
+			log.Errorf("Could not verify block with hash exists in Eth1 chain: %#x: %v", eth1Hash, err)
+			continue
+		}
+		if !blockExists {
+			continue
+		}
+		// Fetch the current canonical chain height from the eth1.0 chain.
+		currentHeight := bs.powChainService.LatestBlockHeight()
+		// Fetch the height of the block pointed to by the beacon state's latest_eth1_data.block_hash
+		// in the canonical, eth1.0 chain.
+		stateLatestEth1Hash := bytesutil.ToBytes32(beaconState.LatestEth1Data.BlockHash32)
+		_, stateLatestEth1Height, err := bs.powChainService.BlockExists(stateLatestEth1Hash)
+		if err != nil {
+			log.Errorf("Could not verify block with hash exists in Eth1 chain: %#x: %v", eth1Hash, err)
+			continue
+		}
+		// Let dataVotes be the set of Eth1DataVote objects vote in state.eth1_data_votes where:
+		// vote.eth1_data.block_hash is the hash of an eth1.0 block that is:
+		//   (i) part of the canonical chain
+		//   (ii) >= ETH1_FOLLOW_DISTANCE blocks behind the head
+		//   (iii) newer than state.latest_eth1_data.block_data.
+		// vote.eth1_data.deposit_root is the deposit root of the eth1.0 deposit contract
+		// at the block defined by vote.eth1_data.block_hash.
+		isBehindFollowDistance := blockHeight.Add(blockHeight, big.NewInt(eth1FollowDistance)).Cmp(currentHeight) >= -1
+		isAheadStateLatestEth1Data := blockHeight.Cmp(stateLatestEth1Height) == 1
+		if blockExists && isBehindFollowDistance && isAheadStateLatestEth1Data {
+			dataVotes = append(dataVotes, vote)
+		}
+	}
+
+	// Now we handle the following two scenarios:
+	// If dataVotes is empty:
+	// Let block_hash be the block hash of the ETH1_FOLLOW_DISTANCE'th ancestor of the head of
+	// the canonical eth1.0 chain.
+	// Let deposit_root be the deposit root of the eth1.0 deposit contract in the
+	// post-state of the block referenced by block_hash.
+	if len(dataVotes) == 0 {
+		// Fetch the current canonical chain height from the eth1.0 chain.
+		currentHeight := bs.powChainService.LatestBlockHeight()
+		ancestorHeight := currentHeight.Sub(currentHeight, big.NewInt(eth1FollowDistance))
+		blockHash, err := bs.powChainService.BlockHashByHeight(ancestorHeight)
+		if err != nil {
+			return nil, fmt.Errorf("could not fetch ETH1_FOLLOW_DISTANCE ancestor: %v", err)
+		}
+		// TODO(#1656): Fetch the deposit root of the post-state deposit contract of the block
+		// references by the block hash of the ancestor instead.
+		depositRoot := bs.powChainService.DepositRoot()
+		return &pb.Eth1DataResponse{
+			Eth1Data: &pbp2p.Eth1Data{
+				DepositRootHash32: depositRoot[:],
+				BlockHash32:       blockHash[:],
+			},
+		}, nil
+	}
+
+	// If dataVotes is non-empty:
+	// Let best_vote be the member of D that has the highest vote.eth1_data.vote_count,
+	// breaking ties by favoring block hashes with higher associated block height.
+	// Let block_hash = best_vote.eth1_data.block_hash.
+	// Let deposit_root = best_vote.eth1_data.deposit_root.
+	bestVote := dataVotes[0]
+	for i := 1; i < len(dataVotes); i++ {
+		vote := dataVotes[i]
+		if vote.VoteCount > bestVote.VoteCount {
+			bestVote = vote
+		} else if vote.VoteCount == bestVote.VoteCount {
+			bestVoteHash := bytesutil.ToBytes32(bestVote.Eth1Data.BlockHash32)
+			voteHash := bytesutil.ToBytes32(vote.Eth1Data.BlockHash32)
+			_, bestVoteHeight, err := bs.powChainService.BlockExists(bestVoteHash)
+			if err != nil {
+				log.Errorf("Could not fetch block height: %v", err)
+				continue
+			}
+			_, voteHeight, err := bs.powChainService.BlockExists(voteHash)
+			if err != nil {
+				log.Errorf("Could not fetch block height: %v", err)
+				continue
+			}
+			if voteHeight.Cmp(bestVoteHeight) == 1 {
+				bestVote = vote
+			}
+		}
+	}
+	return &pb.Eth1DataResponse{
+		Eth1Data: &pbp2p.Eth1Data{
+			BlockHash32:       bestVote.Eth1Data.BlockHash32,
+			DepositRootHash32: bestVote.Eth1Data.DepositRootHash32,
+		},
+	}, nil
 }
 
 // PendingDeposits returns a list of pending deposits that are ready for
 // inclusion in the next beacon block.
 func (bs *BeaconServer) PendingDeposits(ctx context.Context, _ *ptypes.Empty) (*pb.PendingDepositsResponse, error) {
-	bNum := bs.powChainService.LatestBlockNumber()
-
+	bNum := bs.powChainService.LatestBlockHeight()
 	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
 }