catchup: fetchAndWrite/fetchRound quit early on errNoBlockForRound

catchup/service.go

@@ -243,6 +243,8 @@
 	return
 }
 
+const errNoBlockForRoundThreshold = 5
+
 // fetchAndWrite fetches a block, checks the cert, and writes it to the ledger. Cert checking and ledger writing both wait for the ledger to advance if necessary.
 // Returns false if we should stop trying to catch up.  This may occur for several reasons:
 //   - If the context is canceled (e.g. if the node is shutting down)
@@ -254,6 +256,11 @@
 	if dontSyncRound := s.GetDisableSyncRound(); dontSyncRound != 0 && r >= basics.Round(dontSyncRound) {
 		return false
 	}
+
+	// peerErrors tracks errNoBlockForRound in order to quite earlier without making
+	// tons of requests for a block that most likely not happened yet
+	peerErrors := map[network.Peer]int{}
+
 	i := 0
 	for {
 		i++
@@ -302,6 +309,15 @@
 				s.log.Infof("fetchAndWrite(%d): the block is already in the ledger. The catchup is complete", r)
 				return false
 			}
+			if err == errNoBlockForRound {
+				// remote peer doesn't have the block, try another peer
+				// quit if the the same peer peer encountered errNoBlockForRound more than errNoBlockForRoundThreshold times
+				if count := peerErrors[peer]; count > errNoBlockForRoundThreshold {
+					s.log.Infof("fetchAndWrite(%d): remote peers do not have the block. Quitting", r)
+					return false
+				}
+				peerErrors[peer]++
+			}
 			s.log.Debugf("fetchAndWrite(%v): Could not fetch: %v (attempt %d)", r, err, i)
 			peerSelector.rankPeer(psp, peerRankDownloadFailed)
 
@@ -689,6 +705,8 @@
 		return
 	}
 
+	peerErrors := map[network.Peer]int{}
+
 	blockHash := bookkeeping.BlockHash(cert.Proposal.BlockDigest) // semantic digest (i.e., hash of the block header), not byte-for-byte digest
 	peerSelector := createPeerSelector(s.net, s.cfg, false)
 	for s.ledger.LastRound() < cert.Round {
@@ -710,6 +728,15 @@
 				return
 			default:
 			}
+			if err == errNoBlockForRound {
+				// remote peer doesn't have the block, try another peer
+				// quit if the the same peer peer encountered errNoBlockForRound more than errNoBlockForRoundThreshold times
+				if count := peerErrors[peer]; count > errNoBlockForRoundThreshold {
+					s.log.Debugf("fetchAndWrite(%d): remote peers do not have the block. Quitting", cert.Round)
+					return
+				}
+				peerErrors[peer]++
+			}
 			logging.Base().Warnf("fetchRound could not acquire block, fetcher errored out: %v", err)
 			peerSelector.rankPeer(psp, peerRankDownloadFailed)
 			continue

catchup/service_test.go

@@ -20,7 +20,9 @@ import (
 	"context"
 	"errors"
 	"math/rand"
+	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -180,6 +182,27 @@ func (cl *periodicSyncLogger) Warnf(s string, args ...interface{}) {
 	cl.Logger.Warnf(s, args...)
 }
 
+type periodicSyncDebugLogger struct {
+	periodicSyncLogger
+	debugMsgFilter []string
+	debugMsgs      atomic.Uint32
+}
+
+func (cl *periodicSyncDebugLogger) Debugf(s string, args ...interface{}) {
+	// save debug messages for later inspection.
+	if len(cl.debugMsgFilter) > 0 {
+		for _, filter := range cl.debugMsgFilter {
+			if strings.Contains(s, filter) {
+				cl.debugMsgs.Add(1)
+				break
+			}
+		}
+	} else {
+		cl.debugMsgs.Add(1)
+	}
+	cl.Logger.Debugf(s, args...)
+}
+
 func TestSyncRound(t *testing.T) {
 	partitiontest.PartitionTest(t)
 
@@ -208,7 +231,7 @@ func TestSyncRound(t *testing.T) {
 
 	auth := &mockedAuthenticator{fail: true}
 	initialLocalRound := local.LastRound()
-	require.True(t, 0 == initialLocalRound)
+	require.Zero(t, initialLocalRound)
 
 	// Make Service
 	localCfg := config.GetDefaultLocal()
@@ -253,7 +276,7 @@ func TestSyncRound(t *testing.T) {
 	s.UnsetDisableSyncRound()
 	// wait until the catchup is done
 	waitStart = time.Now()
-	for time.Now().Sub(waitStart) < 8*s.deadlineTimeout {
+	for time.Since(waitStart) < 8*s.deadlineTimeout {
 		if remote.LastRound() == local.LastRound() {
 			break
 		}
@@ -298,7 +321,7 @@ func TestPeriodicSync(t *testing.T) {
 
 	auth := &mockedAuthenticator{fail: true}
 	initialLocalRound := local.LastRound()
-	require.True(t, 0 == initialLocalRound)
+	require.Zero(t, initialLocalRound)
 
 	// Make Service
 	s := MakeService(logging.Base(), defaultConfig, net, local, auth, nil, nil)
@@ -315,7 +338,7 @@ func TestPeriodicSync(t *testing.T) {
 	// wait until the catchup is done. Since we've might have missed the sleep window, we need to wait
 	// until the synchronization is complete.
 	waitStart := time.Now()
-	for time.Now().Sub(waitStart) < 10*s.deadlineTimeout {
+	for time.Since(waitStart) < 10*s.deadlineTimeout {
 		if remote.LastRound() == local.LastRound() {
 			break
 		}
@@ -472,6 +495,8 @@ func TestServiceFetchBlocksMultiBlocks(t *testing.T) {
 	}
 	addBlocks(t, remote, blk, int(numberOfBlocks)-1)
 
+	logging.Base().SetLevel(logging.Debug)
+
 	// Create a network and block service
 	blockServiceConfig := config.GetDefaultLocal()
 	net := &httpTestPeerSource{}
@@ -1184,3 +1209,52 @@ func TestServiceLedgerUnavailable(t *testing.T) {
 	require.Greater(t, local.LastRound(), basics.Round(0))
 	require.Less(t, local.LastRound(), remote.LastRound())
 }
+
+// TestServiceNoBlockForRound checks if fetchAndWrite does not repeats 500 times if a block not avaialble
+func TestServiceNoBlockForRound(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	// Make Ledger
+	local := new(mockedLedger)
+	local.blocks = append(local.blocks, bookkeeping.Block{})
+
+	remote, _, blk, err := buildTestLedger(t, bookkeeping.Block{})
+	if err != nil {
+		t.Fatal(err)
+		return
+	}
+	numBlocks := 10
+	addBlocks(t, remote, blk, numBlocks)
+
+	// Create a network and block service
+	blockServiceConfig := config.GetDefaultLocal()
+	net := &httpTestPeerSource{}
+	ls := rpcs.MakeBlockService(logging.Base(), blockServiceConfig, remote, net, "test genesisID")
+
+	nodeA := basicRPCNode{}
+	nodeA.RegisterHTTPHandler(rpcs.BlockServiceBlockPath, ls)
+	nodeA.start()
+	defer nodeA.stop()
+	rootURL := nodeA.rootURL()
+	net.addPeer(rootURL)
+
+	require.Equal(t, basics.Round(0), local.LastRound())
+	require.Equal(t, basics.Round(numBlocks+1), remote.LastRound())
+
+	// Make Service
+	auth := &mockedAuthenticator{fail: false}
+	cfg := config.GetDefaultLocal()
+	cfg.CatchupParallelBlocks = 8
+	s := MakeService(logging.Base(), cfg, net, local, auth, nil, nil)
+	pl := &periodicSyncDebugLogger{periodicSyncLogger: periodicSyncLogger{Logger: logging.Base()}}
+	s.log = pl
+	s.deadlineTimeout = 1 * time.Second
+
+	s.testStart()
+	defer s.Stop()
+	s.sync()
+
+	// without the fix there are about 2k messages (4x catchupRetryLimit)
+	// with the fix expect less than catchupRetryLimit
+	require.Less(t, int(pl.debugMsgs.Load()), catchupRetryLimit)
+}

test/e2e-go/features/catchup/basicCatchup_test.go

@@ -73,24 +73,6 @@ func TestBasicCatchup(t *testing.T) {
 	// Now, catch up
 	err = fixture.LibGoalFixture.ClientWaitForRoundWithTimeout(cloneClient, waitForRound)
 	a.NoError(err)
-
-	cloneNC := fixture.GetNodeControllerForDataDir(cloneDataDir)
-	cloneRestClient := fixture.GetAlgodClientForController(cloneNC)
-
-	// an immediate call for ready will error, for sync time != 0
-	a.Error(cloneRestClient.ReadyCheck())
-
-	for {
-		status, err := cloneRestClient.Status()
-		a.NoError(err)
-
-		if status.LastRound < 10 {
-			time.Sleep(250 * time.Millisecond)
-			continue
-		}
-		a.NoError(cloneRestClient.ReadyCheck())
-		break
-	}
 }
 
 // TestCatchupOverGossip tests catchup across network versions