From 09530abed9023e80775b05630ab2e6baab7f3b78 Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Mon, 27 Jul 2026 14:53:10 +0700 Subject: [PATCH] fix(blocksync): survive transient peer failures and stalls Two ways block sync gives up far too easily. A single failed block request dropped the peer. consumeJobResult called RemovePeer and reported a PeerError for any errBlockFetch, including a plain 15s request timeout, and PeerManager.Errored evicts on the first report. Peers serve block requests one at a time, so a busy peer times out long before it is unhealthy, and dropping it also fails the up to 20 requests already in flight to it - each of which then drops another peer. Count consecutive failures per peer instead and drop only after maxConsecutiveFailures in a row, resetting the count on the first successful response. A failed request also stops being counted as pending, which it previously only did by virtue of the peer being deleted. The p2p client reported the peer itself on every request timeout, which bypassed any caller-side policy. Reject the promise and let the caller decide: blocksync now counts timeouts, statesync's backfill already said it did not want to punish on timeout, and a genuinely dead connection is still caught by the transport's ping/pong. GetBlock is the only production caller of a promise-returning client method, so this is confined to the block sync path. A 60s stall handed over to consensus even when far behind. Nothing switches back to block sync except the state sync path, so a stalled node was demoted permanently to consensus catch-up, which is much slower. A stall is now only a reason to stop when no peer reports a height above ours, or when it has outlasted maxSyncStall, so a wedged synchronizer can still hand over rather than blocking forever. WaitForSync returns whether the node actually caught up, and poolRoutine uses that instead of a second IsCaughtUp call racing the synchronizer it just stopped. Co-Authored-By: Claude Opus 5 --- internal/blocksync/peer_store.go | 29 ++++++- internal/blocksync/peer_store_test.go | 42 +++++++++ internal/blocksync/reactor.go | 13 ++- internal/blocksync/synchronizer.go | 111 +++++++++++++++++++++--- internal/blocksync/synchronizer_test.go | 110 +++++++++++++++++++++-- internal/p2p/client/client.go | 20 +++-- internal/p2p/client/client_test.go | 7 +- 7 files changed, 299 insertions(+), 33 deletions(-) diff --git a/internal/blocksync/peer_store.go b/internal/blocksync/peer_store.go index 2ec26cbb1b..a8136f24d7 100644 --- a/internal/blocksync/peer_store.go +++ b/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) { diff --git a/internal/blocksync/peer_store_test.go b/internal/blocksync/peer_store_test.go index a954b604ca..7dddee5435 100644 --- a/internal/blocksync/peer_store_test.go +++ b/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) +} diff --git a/internal/blocksync/reactor.go b/internal/blocksync/reactor.go index 018a22aefc..a2c48a9931 100644 --- a/internal/blocksync/reactor.go +++ b/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) } } diff --git a/internal/blocksync/synchronizer.go b/internal/blocksync/synchronizer.go index 7f21ce536d..4c5d7087f1 100644 --- a/internal/blocksync/synchronizer.go +++ b/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() diff --git a/internal/blocksync/synchronizer_test.go b/internal/blocksync/synchronizer_test.go index e236066e10..fb6e7f7c05 100644 --- a/internal/blocksync/synchronizer_test.go +++ b/internal/blocksync/synchronizer_test.go @@ -196,19 +196,22 @@ func (suite *SynchronizerTestSuite) TestConsumeJobResult() { }, }, { + // a lone failure re-requests the height but leaves the peer alone, so + // the client mock deliberately has no Send expectation here result: workerpool.Result{Err: mockErr}, wantPushBack: []int64{1}, - mockFn: func(pool *Synchronizer) { - suite.client. - On("Send", mock.Anything, p2p.PeerError{NodeID: "peer 1", Err: mockErr}). - Once(). - Return(nil) - }, + mockFn: func(_ *Synchronizer) {}, }, { + // the peer starts one failure short of the threshold, so this result + // crosses it: the peer is reported and its pending blocks re-requested result: workerpool.Result{Err: mockErr}, wantPushBack: []int64{1, 2}, mockFn: func(pool *Synchronizer) { + pool.AddPeer(newPeerData(peerID1, 1, 100)) + for i := int32(1); i < maxConsecutiveFailures; i++ { + pool.peerStore.AddFailure(peerID1, maxConsecutiveFailures) + } pool.pendingToApply[2] = *respH2 pool.pendingToApply[3] = *respH3 suite.client. @@ -543,3 +546,98 @@ func makePeers(numPeers int, minHeight, maxHeight int64) map[types.NodeID]PeerDa } return peers } + +// TestStallVerdictFor checks when a lack of progress ends block sync. Handing +// over to consensus is effectively irreversible, so a stall must not end block +// sync while peers still have blocks we need. +func TestStallVerdictFor(t *testing.T) { + testCases := []struct { + name string + behind int64 + stalledFor time.Duration + want stallVerdict + }{ + { + name: "progressing while behind", + behind: 500, + stalledFor: syncTimeout / 2, + want: keepSyncing, + }, + { + name: "progressing and level with peers", + behind: 0, + stalledFor: syncTimeout / 2, + want: keepSyncing, + }, + { + name: "stalled with nothing left to fetch", + behind: 0, + stalledFor: syncTimeout + time.Second, + want: stopNothingToFetch, + }, + { + name: "stalled while ahead of every peer", + behind: -5, + stalledFor: syncTimeout + time.Second, + want: stopNothingToFetch, + }, + { + name: "stalled but peers still have blocks", + behind: 500, + stalledFor: syncTimeout + time.Second, + want: keepSyncing, + }, + { + name: "stalled just short of the wedge limit", + behind: 500, + stalledFor: maxSyncStall, + want: keepSyncing, + }, + { + name: "stalled past the wedge limit", + behind: 500, + stalledFor: maxSyncStall + time.Second, + want: stopStalledTooLong, + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, stallVerdictFor(tc.behind, tc.stalledFor)) + }) + } +} + +// TestConsumeJobResultKeepsPeerOnTransientFailure checks that a single failed +// block request does not drop the peer. Dropping it would also fail its other +// in-flight requests, each of which drops another peer in turn. +func (suite *SynchronizerTestSuite) TestConsumeJobResultKeepsPeerOnTransientFailure() { + ctx := context.Background() + + peerID := types.NodeID("peer 1") + resultCh := make(chan workerpool.Result, 1) + wp := workerpool.New(1, workerpool.WithResultCh(resultCh)) + applier := newBlockApplier(suite.blockExec, suite.store, applierWithState(suite.initialState)) + pool := NewSynchronizer(1, suite.client, applier, WithWorkerPool(wp)) + pool.AddPeer(newPeerData(peerID, 1, 100)) + + // stop one short of the threshold: the peer is kept, and the client mock has + // no Send expectation, so reporting it would fail this test + for i := int32(1); i < maxConsecutiveFailures; i++ { + resultCh <- workerpool.Result{ + Err: &errBlockFetch{peerID: peerID, height: int64(i), err: errors.New("timeout")}, + } + suite.Require().NoError(pool.consumeJobResult(ctx)) + suite.Require().Len(pool.peerStore.All(), 1, "peer dropped after %d failures", i) + } + + // the next failure crosses the threshold and does report the peer + suite.client. + On("Send", mock.Anything, mock.Anything). + Once(). + Return(nil) + resultCh <- workerpool.Result{ + Err: &errBlockFetch{peerID: peerID, height: 99, err: errors.New("timeout")}, + } + suite.Require().NoError(pool.consumeJobResult(ctx)) + suite.Require().Empty(pool.peerStore.All(), "peer must be dropped once it exceeds the threshold") +} diff --git a/internal/p2p/client/client.go b/internal/p2p/client/client.go index 82fabaa233..a228b67ac7 100644 --- a/internal/p2p/client/client.go +++ b/internal/p2p/client/client.go @@ -167,7 +167,7 @@ func (c *Client) GetBlock(ctx context.Context, height int64, peerID types.NodeID if err != nil { return nil, err } - return newPromise[*bcproto.BlockResponse](ctx, peerID, reqID, respCh, c), nil + return newPromise[*bcproto.BlockResponse](ctx, reqID, respCh, c), nil } // GetChunk requests a chunk from a peer and returns promise.Promise which resolve the result @@ -184,7 +184,7 @@ func (c *Client) GetChunk( if err != nil { return nil, err } - return newPromise[*statesync.ChunkResponse](ctx, peerID, reqID, respCh, c), nil + return newPromise[*statesync.ChunkResponse](ctx, reqID, respCh, c), nil } // GetSnapshots requests snapshots from a peer @@ -208,7 +208,7 @@ func (c *Client) GetParams( if err != nil { return nil, err } - return newPromise[*statesync.ParamsResponse](ctx, peerID, reqID, respCh, c), nil + return newPromise[*statesync.ParamsResponse](ctx, reqID, respCh, c), nil } // GetLightBlock returns a promise.Promise which resolve the result if response received in time otherwise reject @@ -223,7 +223,7 @@ func (c *Client) GetLightBlock( if err != nil { return nil, err } - return newPromise[*statesync.LightBlockResponse](ctx, peerID, reqID, respCh, c), nil + return newPromise[*statesync.LightBlockResponse](ctx, reqID, respCh, c), nil } // GetSyncStatus requests a block synchronization status from all connected peers @@ -397,7 +397,6 @@ func (c *Client) timeout() <-chan time.Time { func newPromise[T proto.Message]( ctx context.Context, - peerID types.NodeID, reqID string, respCh chan result, client *Client, @@ -415,10 +414,13 @@ func newPromise[T proto.Message]( } resolve(res.Value.(T)) case <-client.timeout(): - _ = client.Send(ctx, p2p.PeerError{ - NodeID: peerID, - Err: ErrPeerNotResponded, - }) + // Reject and let the caller decide what a timeout is worth. Reporting + // the peer here evicts it on a single slow request, which is a poor + // signal: peers answer requests one at a time, so a busy peer times + // out long before it is unhealthy, and evicting it also fails every + // other request already in flight to it. Callers that want to act on + // repeated timeouts can count them, and a genuinely dead connection + // is still caught by the transport's ping/pong. reject(ErrPeerNotResponded) } }) diff --git a/internal/p2p/client/client_test.go b/internal/p2p/client/client_test.go index 43c4438dc0..059c5f7f8a 100644 --- a/internal/p2p/client/client_test.go +++ b/internal/p2p/client/client_test.go @@ -120,16 +120,15 @@ func (suite *ChannelTestSuite) TestGetBlockTimeout() { On("Send", mock.Anything, mock.MatchedBy(envelopeArg)). Once(). Return(nil) - suite.p2pChannel. - On("SendError", mock.Anything, mock.Anything). - Once(). - Return(nil) p, err := suite.client.GetBlock(ctx, suite.height, suite.peerID) // need to wait for the goroutine is started time.Sleep(time.Millisecond) suite.fakeClock.Advance(peerTimeout) suite.Require().NoError(err) _, err = p.Await() + // A timeout rejects the promise and leaves the peer alone; deciding whether + // the peer is at fault belongs to the caller. The channel mock has no + // SendError expectation, so reporting the peer here would fail this test. tmrequire.Error(suite.T(), ErrPeerNotResponded.Error(), err) err = suite.client.resolve(ctx, newEnvelope(reqID, suite.peerID, suite.response)) tmrequire.Error(suite.T(), "cannot resolve a result", err)