Only Sync With the Peer With the Highest Observed Slot

beacon-chain/sync/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "//shared/params:go_default_library",
         "@com_github_ethereum_go_ethereum//common:go_default_library",
         "@com_github_gogo_protobuf//proto:go_default_library",
+        "@com_github_libp2p_go_libp2p_peer//:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
@@ -39,11 +40,9 @@ go_test(
         "receive_block_test.go",
         "regular_sync_test.go",
         "service_test.go",
-        "simulated_sync_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
-        "//beacon-chain/chaintest/backend:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/internal:go_default_library",

beacon-chain/sync/initial-sync/BUILD.bazel

@@ -23,6 +23,7 @@ go_library(
         "//shared/params:go_default_library",
         "@com_github_ethereum_go_ethereum//common:go_default_library",
         "@com_github_gogo_protobuf//proto:go_default_library",
+        "@com_github_libp2p_go_libp2p_peer//:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",

beacon-chain/sync/initial-sync/service.go

@@ -17,14 +17,14 @@ import (
 	"sync"
 	"time"
 
-	"github.com/prysmaticlabs/prysm/shared/hashutil"
-
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/gogo/protobuf/proto"
+	peer "github.com/libp2p/go-libp2p-peer"
 	"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	"github.com/prysmaticlabs/prysm/shared/event"
+	"github.com/prysmaticlabs/prysm/shared/hashutil"
 	"github.com/prysmaticlabs/prysm/shared/p2p"
 	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/sirupsen/logrus"
@@ -110,6 +110,7 @@ type InitialSync struct {
 	finalizedStateRoot  [32]byte
 	mutex               *sync.Mutex
 	nodeIsSynced        bool
+	bestPeer            peer.ID
 }
 
 // NewInitialSyncService constructs a new InitialSyncService.
@@ -155,7 +156,6 @@ func (s *InitialSync) Start() {
 	}
 	s.currentSlot = cHead.Slot
 	go s.run()
-	go s.listenForNewBlocks()
 	go s.checkInMemoryBlocks()
 }
 
@@ -181,6 +181,11 @@ func (s *InitialSync) InitializeFinalizedStateRoot(root [32]byte) {
 	s.finalizedStateRoot = root
 }
 
+// InitializeBestPeer sets the peer ID of the highest observed peer.
+func (s *InitialSync) InitializeBestPeer(p peer.ID) {
+	s.bestPeer = p
+}
+
 // HighestObservedSlot returns the highest observed slot.
 func (s *InitialSync) HighestObservedSlot() uint64 {
 	return s.highestObservedSlot
@@ -274,24 +279,6 @@ func (s *InitialSync) checkInMemoryBlocks() {
 	}
 }
 
-// listenForNewBlocks listens for block announcements beyond the canonical head slot that may
-// be received during initial sync - we must process these blocks to catch up with peers.
-func (s *InitialSync) listenForNewBlocks() {
-	blockAnnounceSub := s.p2p.Subscribe(&pb.BeaconBlockAnnounce{}, s.blockAnnounceBuf)
-	defer func() {
-		blockAnnounceSub.Unsubscribe()
-		close(s.blockAnnounceBuf)
-	}()
-	for {
-		select {
-		case <-s.ctx.Done():
-			return
-		case msg := <-s.blockAnnounceBuf:
-			safelyHandleMessage(s.processBlockAnnounce, msg)
-		}
-	}
-}
-
 // run is the main goroutine for the initial sync service.
 // delayChan is explicitly passed into this function to facilitate tests that don't require a timeout.
 // It is assumed that the goroutine `run` is only called once per instance.
@@ -308,7 +295,8 @@ func (s *InitialSync) run() {
 		close(s.stateBuf)
 	}()
 
-	if err := s.requestStateFromPeer(s.ctx, s.finalizedStateRoot); err != nil {
+	// We send out a state request to all peers.
+	if err := s.requestStateFromPeer(s.ctx, s.finalizedStateRoot, s.bestPeer); err != nil {
 		log.Errorf("Could not request state from peer %v", err)
 	}
 

beacon-chain/sync/initial-sync/service_test.go

@@ -136,7 +136,6 @@ func TestSavingBlock_InSync(t *testing.T) {
 
 	go func() {
 		ss.run()
-		ss.listenForNewBlocks()
 		exitRoutine <- true
 	}()
 

beacon-chain/sync/initial-sync/sync_state.go

@@ -3,6 +3,7 @@ package initialsync
 import (
 	"context"
 
+	peer "github.com/libp2p/go-libp2p-peer"
 	"github.com/prysmaticlabs/prysm/shared/hashutil"
 
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/validators"
@@ -96,11 +97,11 @@ func (s *InitialSync) processState(msg p2p.Message) {
 }
 
 // requestStateFromPeer requests for the canonical state, finalized state, and justified state from a peer.
-func (s *InitialSync) requestStateFromPeer(ctx context.Context, lastFinalizedRoot [32]byte) error {
+func (s *InitialSync) requestStateFromPeer(ctx context.Context, lastFinalizedRoot [32]byte, peerID peer.ID) error {
 	ctx, span := trace.StartSpan(ctx, "beacon-chain.sync.initial-sync.requestStateFromPeer")
 	defer span.End()
 	stateReq.Inc()
 	return s.p2p.Send(ctx, &pb.BeaconStateRequest{
 		FinalizedStateRootHash32S: lastFinalizedRoot[:],
-	}, p2p.AnyPeer)
+	}, peerID)
 }

beacon-chain/sync/querier.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	peer "github.com/libp2p/go-libp2p-peer"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
@@ -57,6 +58,7 @@ type Querier struct {
 	powchain                  powChainService
 	chainStarted              bool
 	atGenesis                 bool
+	bestPeer                  peer.ID
 }
 
 // NewQuerierService constructs a new Sync Querier Service.
@@ -151,34 +153,42 @@ func (q *Querier) run() {
 		ticker.Stop()
 	}()
 
-	q.RequestLatestHead()
-
+	timeout := time.After(5 * time.Second)
 	for {
 		select {
 		case <-q.ctx.Done():
 			queryLog.Info("Exiting goroutine")
 			return
 		case <-ticker.C:
 			q.RequestLatestHead()
-		case msg := <-q.responseBuf:
-			response := msg.Data.(*pb.ChainHeadResponse)
+		case <-timeout:
+			queryLog.Infof("Peer with highest canonical head: %v", q.bestPeer.Pretty())
 			queryLog.Infof(
 				"Latest chain head is at slot: %d and state root: %#x",
-				response.CanonicalSlot-params.BeaconConfig().GenesisSlot, response.CanonicalStateRootHash32,
+				q.currentHeadSlot-params.BeaconConfig().GenesisSlot, q.currentStateRoot,
 			)
-			q.currentHeadSlot = response.CanonicalSlot
-			q.currentStateRoot = response.CanonicalStateRootHash32
-			q.currentFinalizedStateRoot = bytesutil.ToBytes32(response.FinalizedStateRootHash32S)
-
 			ticker.Stop()
 			responseSub.Unsubscribe()
 			q.cancel()
+		case msg := <-q.responseBuf:
+			response := msg.Data.(*pb.ChainHeadResponse)
+			queryLog.WithFields(logrus.Fields{
+				"peerID":      msg.Peer.Pretty(),
+				"highestSlot": response.CanonicalSlot - params.BeaconConfig().GenesisSlot,
+			}).Info("Received chain head from peer")
+			if response.CanonicalSlot > q.currentHeadSlot {
+				q.currentHeadSlot = response.CanonicalSlot
+				q.bestPeer = msg.Peer
+				q.currentHeadSlot = response.CanonicalSlot
+				q.currentStateRoot = response.CanonicalStateRootHash32
+				q.currentFinalizedStateRoot = bytesutil.ToBytes32(response.FinalizedStateRootHash32S)
+			}
 		}
 	}
 }
 
-// RequestLatestHead broadcasts out a request for all
-// the latest chain heads from the node's peers.
+// RequestLatestHead sends a request for all
+// the latest chain head slot and state root to a peer.
 func (q *Querier) RequestLatestHead() {
 	request := &pb.ChainHeadRequest{}
 	q.p2p.Broadcast(context.Background(), request)

beacon-chain/sync/querier_test.go

@@ -146,7 +146,7 @@ func TestQuerier_ChainReqResponse(t *testing.T) {
 	}()
 
 	response := &pb.ChainHeadResponse{
-		CanonicalSlot:            0,
+		CanonicalSlot:            1,
 		CanonicalStateRootHash32: []byte{'a', 'b'},
 	}
 

beacon-chain/sync/service.go

@@ -129,6 +129,7 @@ func (ss *Service) run() {
 
 	// Sets the highest observed slot from querier.
 	ss.InitialSync.InitializeObservedSlot(ss.Querier.currentHeadSlot)
+	ss.InitialSync.InitializeBestPeer(ss.Querier.bestPeer)
 	ss.InitialSync.InitializeObservedStateRoot(bytesutil.ToBytes32(ss.Querier.currentStateRoot))
 	// Sets the state root of the highest observed slot.
 	ss.InitialSync.InitializeFinalizedStateRoot(ss.Querier.currentFinalizedStateRoot)