use randomized peers from the set of 20 peers with good latency

fetch/fetch.go

@@ -31,7 +31,7 @@ const (
 
 	cacheSize = 1000
 
-	RedundantPeers = 10
+	RedundantPeers = 20
 )
 
 var (
@@ -742,6 +742,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,7 +144,7 @@ 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().SelectBestShuffled(gomock.Any()).Return(peers)
 		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).
 			DoAndReturn(func(
 				_ context.Context,
@@ -189,7 +189,7 @@ 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().SelectBestShuffled(gomock.Any()).Return(peers)
 		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).
 			DoAndReturn(func(
 				_ context.Context,
@@ -212,7 +212,7 @@ 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().SelectBestShuffled(gomock.Any()).Return(peers)
 		td.mFetcher.EXPECT().GetLayerData(gomock.Any(), peers, layerID, gomock.Any(), gomock.Any()).
 			DoAndReturn(func(
 				_ context.Context,
@@ -377,7 +377,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))
 }
@@ -65,7 +65,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()
@@ -161,7 +161,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 {
@@ -263,7 +263,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)
@@ -292,7 +292,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)
@@ -315,7 +315,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"))
@@ -381,7 +381,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)
@@ -508,7 +508,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

@@ -370,11 +370,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

@@ -124,7 +124,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
 }
 
@@ -146,7 +146,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)