fix(blocksync): survive transient peer failures and stalls

internal/blocksync/peer_store.go

@@ -21,7 +21,10 @@ type (
 	}
 	// PeerData uses to keep peer related data like base height and the current height etc
 	PeerData struct {
-		numPending  int32
+		numPending int32
+		// numFailures counts block requests this peer failed in a row. It resets
+		// on the first successful response.
+		numFailures int32
 		height      int64
 		base        int64
 		peerID      types.NodeID
@@ -124,6 +127,22 @@ func (p *InMemPeerStore) FindPeer(height int64) (PeerData, bool) {
 	return peers[0], true
 }
 
+// AddFailure records a block request that this peer failed to answer: the
+// request stops being pending, and the peer's consecutive failure count grows.
+//
+// It reports true once the peer has failed maxFailures requests in a row and
+// should be dropped. An unknown peer reports false, so the failures of a peer
+// that was already removed do not report it again.
+func (p *InMemPeerStore) AddFailure(peerID types.NodeID, maxFailures int32) bool {
+	tooMany := false
+	p.store.Update(peerID, func(_ types.NodeID, peer *PeerData) {
+		peer.numPending--
+		peer.numFailures++
+		tooMany = peer.numFailures >= maxFailures
+	})
+	return tooMany
+}
+
 // FindTimedoutPeers finds and returns the timed out peers
 func (p *InMemPeerStore) FindTimedoutPeers() []PeerData {
 	return p.Query(store.AndX(
@@ -214,6 +233,14 @@ func AddNumPending(val int32) store.UpdateFunc[types.NodeID, PeerData] {
 	}
 }
 
+// ResetFailures clears the count of consecutive failed requests, so that the
+// threshold only ever trips on an unbroken run of failures
+func ResetFailures() store.UpdateFunc[types.NodeID, PeerData] {
+	return func(_ types.NodeID, peer *PeerData) {
+		peer.numFailures = 0
+	}
+}
+
 // UpdateMonitor adds a block size value to the peer monitor if numPending is greater than zero
 func UpdateMonitor(recvSize int) store.UpdateFunc[types.NodeID, PeerData] {
 	return func(peerID types.NodeID, peer *PeerData) {

internal/blocksync/peer_store_test.go

@@ -119,3 +119,45 @@ func TestInMemPeerStoreFindPeer(t *testing.T) {
 	require.Len(t, timedoutPeers, 1)
 	require.Equal(t, peers[3].peerID, timedoutPeers[0].peerID)
 }
+
+// TestAddFailure checks that a peer is only reported once it has failed
+// maxFailures requests in a row, and that a success in between clears the run.
+func TestAddFailure(t *testing.T) {
+	const maxFailures int32 = 3
+	peerID := types.NodeID("peer1")
+	inmem := NewInMemPeerStore(newPeerData(peerID, 1, 100))
+
+	// requests below the threshold keep the peer
+	for i := int32(1); i < maxFailures; i++ {
+		require.False(t, inmem.AddFailure(peerID, maxFailures),
+			"peer must survive failure %d of %d", i, maxFailures)
+	}
+	require.True(t, inmem.AddFailure(peerID, maxFailures),
+		"peer must be reported on the last failure")
+
+	// a success clears the run, so the next failure starts over
+	inmem.Update(peerID, ResetFailures())
+	require.False(t, inmem.AddFailure(peerID, maxFailures),
+		"a success must clear the failure count")
+
+	// an unknown peer is never reported, so failures arriving after a peer was
+	// already removed do not report it a second time
+	require.False(t, inmem.AddFailure(types.NodeID("nope"), maxFailures))
+}
+
+// TestAddFailureClearsPending checks that a failed request stops being counted
+// as pending; otherwise the peer's pending count only grows and it stops being
+// selected once it reaches maxPendingRequestsPerPeer.
+func TestAddFailureClearsPending(t *testing.T) {
+	peerID := types.NodeID("peer1")
+	inmem := NewInMemPeerStore(newPeerData(peerID, 1, 100))
+
+	inmem.Update(peerID, AddNumPending(2))
+	inmem.AddFailure(peerID, 10)
+	inmem.AddFailure(peerID, 10)
+
+	peer, found := inmem.Get(peerID)
+	require.True(t, found)
+	require.EqualValues(t, 0, peer.numPending)
+	require.EqualValues(t, 2, peer.numFailures)
+}

internal/blocksync/reactor.go

@@ -29,8 +29,13 @@ const (
 	// check if we should switch to consensus reactor
 	switchToConsensusIntervalSeconds = 1
 
-	// switch to consensus after this duration of inactivity
+	// consider block sync stalled after this duration of inactivity
 	syncTimeout = 60 * time.Second
+
+	// hand over to consensus after this duration of inactivity even when peers
+	// still report higher blocks, so that a wedged synchronizer cannot keep the
+	// node out of consensus forever
+	maxSyncStall = 10 * time.Minute
 )
 
 type ReactorOption func(*Reactor)
@@ -286,11 +291,13 @@ func (r *Reactor) requestRoutine(ctx context.Context, p2pClient *client.Client)
 //
 // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
 func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool) {
-	r.synchronizer.WaitForSync(ctx)
+	caughtUp := r.synchronizer.WaitForSync(ctx)
 	r.synchronizer.Stop()
 	r.blockSyncFlag.Store(false)
 	if r.consReactor != nil {
-		r.consReactor.SwitchToConsensus(ctx, r.executor.State(), r.synchronizer.IsCaughtUp() || stateSynced)
+		// caughtUp is what WaitForSync actually decided on, rather than a second
+		// IsCaughtUp call that races the synchronizer we just stopped
+		r.consReactor.SwitchToConsensus(ctx, r.executor.State(), caughtUp || stateSynced)
 	}
 }
 

internal/blocksync/synchronizer.go

@@ -37,6 +37,12 @@ const (
 	maxPendingRequestsPerPeer           = 20
 	defaultSyncRateIntervalBlocks int64 = 100
 
+	// maxConsecutiveFailures is how many block requests in a row a peer may fail
+	// before we drop it. Requests time out under load, and a peer serves its
+	// requests one at a time, so a single failure says very little about the
+	// peer. Dropping one costs the rest of its in-flight requests too.
+	maxConsecutiveFailures int32 = 5
+
 	// Minimum recv rate to ensure we're receiving blocks from a peer fast
 	// enough. If a peer is not sending us data at at least that rate, we
 	// consider them to have timed out and we disconnect.
@@ -236,12 +242,28 @@ func (s *Synchronizer) consumeJobResult(ctx context.Context) error {
 			return nil
 		}
 		s.jobGen.pushBack(bfErr.height)
+		// One failed request is usually a timeout under load, not a bad peer.
+		// Dropping the peer also fails its other in-flight requests, each of
+		// which drops another peer in turn, so react only to a sustained run of
+		// failures.
+		if !s.peerStore.AddFailure(bfErr.peerID, maxConsecutiveFailures) {
+			s.logger.Debug("block request failed, keeping peer",
+				"peer", bfErr.peerID,
+				"height", bfErr.height,
+				"error", bfErr.err)
+			return nil
+		}
+		s.logger.Error("removing peer after too many consecutive failed block requests",
+			"peer", bfErr.peerID,
+			"height", bfErr.height,
+			"failures", maxConsecutiveFailures,
+			"error", bfErr.err)
 		s.RemovePeer(bfErr.peerID)
 		_ = s.client.Send(ctx, p2p.PeerError{NodeID: bfErr.peerID, Err: bfErr})
 		return nil
 	}
 	resp := res.Value.(*BlockResponse)
-	s.peerStore.Update(resp.PeerID, AddNumPending(-1), UpdateMonitor(resp.Size))
+	s.peerStore.Update(resp.PeerID, AddNumPending(-1), ResetFailures(), UpdateMonitor(resp.Size))
 	err = s.addBlock(*resp)
 	if err != nil {
 		if !errors.Is(err, errDuplicateBlock) {
@@ -286,32 +308,101 @@ func (s *Synchronizer) IsCaughtUp() bool {
 	return s.height >= maxHeight
 }
 
-func (s *Synchronizer) WaitForSync(ctx context.Context) {
+// WaitForSync blocks until block sync is finished, and reports whether the node
+// actually caught up.
+//
+// A stall on its own is not a reason to stop. Handing over to consensus is a
+// one-way door - only the state sync path ever switches back to block sync - and
+// consensus catch-up is far slower than block sync, so a node that gives up
+// while it is still thousands of blocks behind stays behind. As long as peers
+// report heights above ours there are blocks left to fetch and something to
+// retry, so keep going and say so loudly. Give up on the stall only once no peer
+// can help us, or once the stall has outlasted maxSyncStall, so that a wedged
+// synchronizer can still hand over rather than blocking forever.
+func (s *Synchronizer) WaitForSync(ctx context.Context) (caughtUp bool) {
 	ticker := time.NewTicker(switchToConsensusIntervalSeconds * time.Second)
 	defer ticker.Stop()
 	for {
 		select {
 		case <-ctx.Done():
-			return
+			return s.IsCaughtUp()
 		case <-ticker.C:
+			if s.IsCaughtUp() {
+				return true
+			}
 			var (
-				height, _   = s.GetStatus()
-				lastAdvance = s.LastAdvance()
-				isCaughtUp  = s.IsCaughtUp()
+				height, _     = s.GetStatus()
+				maxPeerHeight = s.MaxPeerHeight()
+				stalledFor    = time.Since(s.LastAdvance())
+				behind        = maxPeerHeight - height
 			)
-			if isCaughtUp || time.Since(lastAdvance) > syncTimeout {
-				return
+			switch stallVerdictFor(behind, stalledFor) {
+			case stopNothingToFetch:
+				return false
+			case stopStalledTooLong:
+				s.logger.Error(
+					"block sync stalled for too long, handing over to consensus while still behind",
+					"height", height,
+					"max_peer_height", maxPeerHeight,
+					"behind", behind,
+					"stalled_for", stalledFor,
+				)
+				return false
+			}
+			if stalledFor > syncTimeout {
+				s.logger.Error(
+					"block sync has stalled but peers report higher blocks, still trying",
+					"height", height,
+					"max_peer_height", maxPeerHeight,
+					"behind", behind,
+					"stalled_for", stalledFor,
+					"giving_up_in", maxSyncStall-stalledFor,
+				)
+				continue
 			}
 			s.logger.Info(
 				"not caught up yet",
 				"height", height,
-				"max_peer_height", s.MaxPeerHeight(),
-				"timeout_in", syncTimeout-time.Since(lastAdvance),
+				"max_peer_height", maxPeerHeight,
+				"timeout_in", syncTimeout-stalledFor,
 			)
 		}
 	}
 }
 
+// stallVerdict says what a lack of progress in block sync means.
+type stallVerdict int
+
+const (
+	// keepSyncing means the stall is not a reason to stop yet
+	keepSyncing stallVerdict = iota
+	// stopNothingToFetch means no peer has a block we are missing
+	stopNothingToFetch
+	// stopStalledTooLong means we are still behind but have to hand over anyway
+	stopStalledTooLong
+)
+
+// stallVerdictFor decides what to do when block sync has made no progress for
+// stalledFor, given how many blocks the best peer is ahead of us.
+//
+// Being behind with peers that can serve us is a reason to keep retrying, not to
+// stop: handing over to consensus is effectively irreversible, so stopping while
+// behind leaves the node grinding through consensus catch-up instead. Only a
+// stall with nothing left to fetch, or one long enough to look like a wedge,
+// ends block sync.
+func stallVerdictFor(behind int64, stalledFor time.Duration) stallVerdict {
+	switch {
+	case stalledFor <= syncTimeout:
+		return keepSyncing
+	case behind <= 0:
+		return stopNothingToFetch
+	case stalledFor > maxSyncStall:
+		return stopStalledTooLong
+	default:
+		return keepSyncing
+	}
+}
+
 // MaxPeerHeight returns the highest reported height.
 func (s *Synchronizer) MaxPeerHeight() int64 {
 	return s.peerStore.MaxHeight()