sync: use randomized peers from the set of 20 peers with good latency (…

…#5263)

this change improves peer selection logic to split the load between 20 peers with good latency,
additionally it avoids suboptimal behavior when node will stick to the same peer when it would benefit switching to another:
- peer had the best latency in the past, but temporarily faulty due to being overloaded
- peer fails to negotiate protocols, but we didn't collect latency info to prioritize peers yet

in second case we should disconnect such peer, but our software registers protocol asynchronously
so we actually may drop honest peer, therefore we only workaround such problem.

fetch/fetch.go

@@ -31,7 +31,7 @@ const (
 
 	cacheSize = 1000
 
-	RedundantPeers = 10
+	RedundantPeers = 20
 )
 
 var (
@@ -686,6 +686,12 @@ func (f *Fetch) RegisterPeerHashes(peer p2p.Peer, hashes []types.Hash32) {
 	f.hashToPeers.RegisterPeerHashes(peer, hashes)
 }
 
-func (f *Fetch) SelectBest(n int) []p2p.Peer {
-	return f.peers.SelectBest(n)
+func (f *Fetch) SelectBestShuffled(n int) []p2p.Peer {
+	// shuffle to split the load between peers with good latency.
+	// and it avoids sticky behavior, when temporarily faulty peer had good latency in the past.
+	peers := f.peers.SelectBest(n)
+	rand.Shuffle(len(peers), func(i, j int) {
+		peers[i], peers[j] = peers[j], peers[i]
+	})
+	return peers
 }

syncer/data_fetch.go

@@ -83,7 +83,7 @@ func NewDataFetch(
 
 // PollMaliciousProofs polls all peers for malicious NodeIDs.
 func (d *DataFetch) PollMaliciousProofs(ctx context.Context) error {
-	peers := d.fetcher.SelectBest(fetch.RedundantPeers)
+	peers := d.fetcher.SelectBestShuffled(fetch.RedundantPeers)
 	logger := d.logger.WithContext(ctx)
 	req := &maliciousIDRequest{
 		peers: peers,
@@ -137,7 +137,7 @@ func (d *DataFetch) PollMaliciousProofs(ctx context.Context) error {
 // PollLayerData polls all peers for data in the specified layer.
 func (d *DataFetch) PollLayerData(ctx context.Context, lid types.LayerID, peers ...p2p.Peer) error {
 	if len(peers) == 0 {
-		peers = d.fetcher.SelectBest(fetch.RedundantPeers)
+		peers = d.fetcher.SelectBestShuffled(fetch.RedundantPeers)
 	}
 	if len(peers) == 0 {
 		return errNoPeers
@@ -471,7 +471,7 @@ func (d *DataFetch) updateAtxPeer(epoch types.EpochID, peer p2p.Peer) {
 
 // GetEpochATXs fetches all ATXs published in the specified epoch from a peer.
 func (d *DataFetch) GetEpochATXs(ctx context.Context, epoch types.EpochID) error {
-	peers := d.fetcher.SelectBest(fetch.RedundantPeers)
+	peers := d.fetcher.SelectBestShuffled(fetch.RedundantPeers)
 	if len(peers) == 0 {
 		return errNoPeers
 	}

syncer/data_fetch_test.go

@@ -103,7 +103,7 @@ func TestDataFetch_PollMaliciousIDs(t *testing.T) {
 	errUnknown := errors.New("unknown")
 	newTestDataFetchWithMocks := func(_ *testing.T, exits bool) *testDataFetch {
 		td := newTestDataFetch(t)
-		td.mFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
+		td.mFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
 		td.mFetcher.EXPECT().GetMaliciousIDs(gomock.Any(), peers, gomock.Any(), gomock.Any()).DoAndReturn(
 			func(_ context.Context, _ []p2p.Peer, okCB func([]byte, p2p.Peer), errCB func(error, p2p.Peer)) error {
 				for _, peer := range peers {
@@ -144,9 +144,15 @@ func TestDataFetch_PollLayerData(t *testing.T) {
 	errUnknown := errors.New("unknown")
 	newTestDataFetchWithMocks := func(*testing.T) *testDataFetch {
 		td := newTestDataFetch(t)
-		td.mFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
-		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).DoAndReturn(
-			func(_ context.Context, _ []p2p.Peer, _ types.LayerID, okCB func([]byte, p2p.Peer), errCB func(error, p2p.Peer)) error {
+		td.mFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
+		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).
+			DoAndReturn(func(
+				_ context.Context,
+				_ []p2p.Peer,
+				_ types.LayerID,
+				okCB func([]byte, p2p.Peer),
+				errCB func(error, p2p.Peer),
+			) error {
 				for _, peer := range peers {
 					td.mFetcher.EXPECT().RegisterPeerHashes(peer, gomock.Any())
 					okCB(generateLayerContent(t), peer)
@@ -183,9 +189,15 @@ func TestDataFetch_PollLayerData_PeerErrors(t *testing.T) {
 	t.Run("only one peer has data", func(t *testing.T) {
 		t.Parallel()
 		td := newTestDataFetch(t)
-		td.mFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
-		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).DoAndReturn(
-			func(_ context.Context, _ []p2p.Peer, _ types.LayerID, okCB func([]byte, p2p.Peer), errCB func(error, p2p.Peer)) error {
+		td.mFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
+		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).
+			DoAndReturn(func(
+				_ context.Context,
+				_ []p2p.Peer,
+				_ types.LayerID,
+				okCB func([]byte, p2p.Peer),
+				errCB func(error, p2p.Peer),
+			) error {
 				td.mFetcher.EXPECT().RegisterPeerHashes(peers[0], gomock.Any())
 				okCB(generateLayerContent(t), peers[0])
 				for i := 1; i < numPeers; i++ {
@@ -200,9 +212,15 @@ func TestDataFetch_PollLayerData_PeerErrors(t *testing.T) {
 	t.Run("only one peer has empty layer", func(t *testing.T) {
 		t.Parallel()
 		td := newTestDataFetch(t)
-		td.mFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
-		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).DoAndReturn(
-			func(_ context.Context, _ []p2p.Peer, _ types.LayerID, okCB func([]byte, p2p.Peer), errCB func(error, p2p.Peer)) error {
+		td.mFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
+		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).
+			DoAndReturn(func(
+				_ context.Context,
+				_ []p2p.Peer,
+				_ types.LayerID,
+				okCB func([]byte, p2p.Peer),
+				errCB func(error, p2p.Peer),
+			) error {
 				okCB(generateEmptyLayer(t), peers[0])
 				for i := 1; i < numPeers; i++ {
 					errCB(errors.New("not available"), peers[i])
@@ -353,7 +371,7 @@ func TestDataFetch_GetEpochATXs(t *testing.T) {
 			ed := &fetch.EpochData{
 				AtxIDs: types.RandomActiveSet(11),
 			}
-			td.mFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
+			td.mFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
 			if tc.getErr == nil {
 				td.mAtxCache.EXPECT().GetMissingActiveSet(epoch+1, ed.AtxIDs).Return(ed.AtxIDs[1:])
 			}

syncer/find_fork.go

@@ -82,7 +82,7 @@ func (ff *ForkFinder) Purge(all bool, toPurge ...p2p.Peer) {
 		return
 	}
 
-	peers := ff.fetcher.SelectBest(fetch.RedundantPeers)
+	peers := ff.fetcher.SelectBestShuffled(fetch.RedundantPeers)
 	uniquePeers := make(map[p2p.Peer]struct{})
 	for _, p := range peers {
 		uniquePeers[p] = struct{}{}

syncer/find_fork_test.go

@@ -48,7 +48,7 @@ func TestResynced(t *testing.T) {
 	tf.AddResynced(lid, hash)
 	require.False(t, tf.NeedResync(lid, hash))
 
-	tf.mFetcher.EXPECT().SelectBest(gomock.Any()).Return([]p2p.Peer{})
+	tf.mFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return([]p2p.Peer{})
 	tf.Purge(false)
 	require.True(t, tf.NeedResync(lid, hash))
 }
@@ -60,7 +60,7 @@ func TestForkFinder_Purge(t *testing.T) {
 	for i := 1; i < numCached; i++ {
 		tf.UpdateAgreement(p2p.Peer(strconv.Itoa(i)), types.LayerID(uint32(i+1)), types.RandomHash(), time.Now())
 	}
-	tf.mFetcher.EXPECT().SelectBest(gomock.Any()).Return([]p2p.Peer{})
+	tf.mFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return([]p2p.Peer{})
 	require.Equal(t, numCached, tf.NumPeersCached())
 	tf.Purge(false)
 	require.Equal(t, 9, tf.NumPeersCached())

syncer/interface.go

@@ -69,7 +69,7 @@ type fetcher interface {
 	GetBlocks(context.Context, []types.BlockID) error
 	RegisterPeerHashes(peer p2p.Peer, hashes []types.Hash32)
 
-	SelectBest(int) []p2p.Peer
+	SelectBestShuffled(int) []p2p.Peer
 	PeerEpochInfo(context.Context, p2p.Peer, types.EpochID) (*fetch.EpochData, error)
 	PeerMeshHashes(context.Context, p2p.Peer, *fetch.MeshHashRequest) (*fetch.MeshHashes, error)
 }

syncer/state_syncer.go

@@ -148,7 +148,7 @@ func (s *Syncer) layerOpinions(
 	ctx context.Context,
 	lid types.LayerID,
 ) ([]*peerOpinion, []*types.Certificate, error) {
-	peers := s.dataFetcher.SelectBest(fetch.RedundantPeers)
+	peers := s.dataFetcher.SelectBestShuffled(fetch.RedundantPeers)
 	if len(peers) == 0 {
 		return nil, nil, errNoPeers
 	}

syncer/state_syncer_test.go

@@ -44,7 +44,7 @@ func TestProcessLayers_MultiLayers(t *testing.T) {
 	ts.mTicker.advanceToLayer(current)
 
 	peers := test.GeneratePeerIDs(3)
-	ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers).AnyTimes()
+	ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers).AnyTimes()
 	ts.mForkFinder.EXPECT().
 		UpdateAgreement(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		AnyTimes()
@@ -156,7 +156,7 @@ func TestProcessLayers_OpinionsNotAdopted(t *testing.T) {
 			ts.syncer.setLastSyncedLayer(current.Sub(1))
 			ts.mTicker.advanceToLayer(current)
 			peers := test.GeneratePeerIDs(3)
-			ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers).AnyTimes()
+			ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers).AnyTimes()
 
 			hasCert := false
 			for _, opn := range tc.opns {
@@ -258,7 +258,7 @@ func TestProcessLayers_HareIsStillWorking(t *testing.T) {
 
 	ts.mLyrPatrol.EXPECT().IsHareInCharge(lastSynced).Return(false)
 	peers := test.GeneratePeerIDs(3)
-	ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
+	ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
 	ts.mDataFetcher.EXPECT().
 		PollLayerOpinions(gomock.Any(), lastSynced, true, peers).
 		Return(nil, nil, nil)
@@ -287,7 +287,7 @@ func TestProcessLayers_HareTakesTooLong(t *testing.T) {
 			ts.mLyrPatrol.EXPECT().IsHareInCharge(lid).Return(false)
 		}
 		peers := test.GeneratePeerIDs(3)
-		ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
+		ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
 		ts.mDataFetcher.EXPECT().
 			PollLayerOpinions(gomock.Any(), lid, gomock.Any(), peers).
 			Return(nil, nil, nil)
@@ -310,7 +310,7 @@ func TestProcessLayers_OpinionsOptional(t *testing.T) {
 	ts.mTicker.advanceToLayer(lastSynced.Add(1))
 	ts.mLyrPatrol.EXPECT().IsHareInCharge(lastSynced).Return(false)
 	peers := test.GeneratePeerIDs(5)
-	ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
+	ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
 	ts.mDataFetcher.EXPECT().
 		PollLayerOpinions(gomock.Any(), lastSynced, true, peers).
 		Return(nil, nil, errors.New("meh"))
@@ -376,7 +376,7 @@ func TestProcessLayers_MeshHashDiverged(t *testing.T) {
 
 	ts.mLyrPatrol.EXPECT().IsHareInCharge(instate).Return(false)
 	peers := test.GeneratePeerIDs(3)
-	ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
+	ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
 	ts.mDataFetcher.EXPECT().
 		PollLayerOpinions(gomock.Any(), instate, false, peers).
 		Return(opns, nil, nil)
@@ -503,7 +503,7 @@ func TestProcessLayers_NoHashResolutionForNewlySyncedNode(t *testing.T) {
 	for lid := instate; lid <= current; lid++ {
 		ts.mLyrPatrol.EXPECT().IsHareInCharge(lid)
 		peers := test.GeneratePeerIDs(3)
-		ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return(peers)
+		ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(peers)
 		ts.mDataFetcher.EXPECT().
 			PollLayerOpinions(gomock.Any(), lid, gomock.Any(), peers).
 			Return(opns, nil, nil)

syncer/syncer.go

@@ -372,11 +372,12 @@ func (s *Syncer) synchronize(ctx context.Context) bool {
 			return true
 		}
 		// check that we have any peers
-		if len(s.dataFetcher.SelectBest(1)) == 0 {
+		if len(s.dataFetcher.SelectBestShuffled(1)) == 0 {
 			return false
 		}
 
 		if err := s.syncAtx(ctx); err != nil {
+			s.logger.With().Error("failed to sync atxs", log.Context(ctx), log.Err(err))
 			return false
 		}
 

syncer/syncer_test.go

@@ -123,7 +123,7 @@ func newTestSyncer(t *testing.T, interval time.Duration) *testSyncer {
 
 func newSyncerWithoutPeriodicRuns(t *testing.T) *testSyncer {
 	ts := newTestSyncer(t, never)
-	ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return([]p2p.Peer{"non-empty"}).AnyTimes()
+	ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return([]p2p.Peer{"non-empty"}).AnyTimes()
 	return ts
 }
 
@@ -145,7 +145,7 @@ func TestStartAndShutdown(t *testing.T) {
 	ts.syncer.Start()
 
 	ts.mForkFinder.EXPECT().Purge(false).AnyTimes()
-	ts.mDataFetcher.EXPECT().SelectBest(gomock.Any()).Return(nil).AnyTimes()
+	ts.mDataFetcher.EXPECT().SelectBestShuffled(gomock.Any()).Return(nil).AnyTimes()
 	require.Eventually(t, func() bool {
 		return ts.syncer.ListenToATXGossip() && ts.syncer.ListenToGossip() &&
 			ts.syncer.IsSynced(ctx)