feat(eth/downloader): warn when a sync round stalls

eth/downloader/downloader.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/XinFinOrg/XDPoSChain"
 	"github.com/XinFinOrg/XDPoSChain/common"
+	"github.com/XinFinOrg/XDPoSChain/common/mclock"
 	"github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/engines/engine_v2"
 	"github.com/XinFinOrg/XDPoSChain/core/rawdb"
 	"github.com/XinFinOrg/XDPoSChain/core/state"
@@ -74,6 +75,9 @@ var (
 	fsHeaderSafetyNet      = 2048            // Number of headers to discard in case a chain violation is detected
 	fsHeaderContCheck      = 3 * time.Second // Time interval to check for header continuations during state download
 	fsMinFullBlocks        = 64              // Number of blocks to retrieve fully even in fast sync
+
+	stalledSyncThreshold    = 5 * time.Minute // Minimum age of an unfinished sync round before it is probed for progress
+	stalledSyncWarnInterval = time.Minute     // Minimum interval between two stalled sync warnings
 )
 
 var (
@@ -127,6 +131,16 @@ type Downloader struct {
 	notified        int32
 	committed       int32
 
+	stallProgressHook func() uint64 // Replacement for stallProgress during testing
+	syncRoundStart    atomic.Int64  // Monotonic time at which the in-flight sync round started (0 if idle)
+	syncRoundSeq      atomic.Uint64 // Generation of the active sync round
+	lastStallWarn     atomic.Int64  // Monotonic time at which the last stalled sync warning was emitted
+	stallProbeMu      sync.Mutex    // Serializes round lifecycle and complete stall probe decisions
+	stallProbeRound   uint64        // Sync round associated with the protected probe state
+	stallProbed       bool          // Whether a baseline was captured for stallProbeRound
+	stallBaseline     uint64        // Progress value observed at the previous probe
+	stallProgressSeq  atomic.Uint64 // Lock-free sequence incremented whenever the active round makes progress
+
 	// Pivot block configuration (set before sync starts)
 	pivotNumber uint64      // Fixed pivot block number (0 = use default calculation)
 	pivotHash   common.Hash // Expected pivot block hash for verification
@@ -397,6 +411,80 @@ func (d *Downloader) Synchronise(id string, head common.Hash, td *big.Int, mode
 	return err
 }
 
+// warnIfSyncStalled reports a sync round that has been holding the
+// synchronising flag for an unusually long time without making any progress.
+// Such a round silently rejects every subsequent attempt with errBusy, so
+// without this the node can stop syncing indefinitely without emitting a
+// single log line. The first probe past the threshold only captures a progress
+// baseline, and an actual warning is emitted once a later probe shows the
+// round has not advanced any further.
+func (d *Downloader) warnIfSyncStalled(id string) {
+	started := d.syncRoundStart.Load()
+	round := d.syncRoundSeq.Load()
+	if started == 0 {
+		// The running round is already tearing down, nothing to report.
+		return
+	}
+	now := int64(mclock.Now())
+	elapsed := time.Duration(now - started)
+	if elapsed < stalledSyncThreshold {
+		return
+	}
+	if !d.stallProbeMu.TryLock() {
+		// Another busy attempt is already probing this round. Do not make the
+		// errBusy path wait for diagnostics.
+		return
+	}
+	defer d.stallProbeMu.Unlock()
+	if started != d.syncRoundStart.Load() || round != d.syncRoundSeq.Load() {
+		// The observed round ended while this attempt was entering the probe.
+		return
+	}
+	progress := d.stallProgress()
+	if started != d.syncRoundStart.Load() || round != d.syncRoundSeq.Load() {
+		// Reading progress may yield to a finishing round or test hook.
+		return
+	}
+	if d.stallProbeRound != round {
+		d.stallProbeRound = round
+		d.stallProbed = false
+	}
+	if d.stallProbed {
+		// Baseline already captured, only report when the round stopped
+		// advancing since then. Otherwise shift the baseline forward and
+		// stay quiet.
+		if progress > d.stallBaseline {
+			d.stallBaseline = progress
+			return
+		}
+	} else {
+		// First probe of this round, just remember the baseline.
+		d.stallProbed = true
+		d.stallBaseline = progress
+		return
+	}
+	last := d.lastStallWarn.Load()
+	if last != 0 && time.Duration(now-last) < stalledSyncWarnInterval {
+		return
+	}
+	d.lastStallWarn.Store(now)
+	args := []interface{}{"elapsed", common.PrettyDuration(elapsed)}
+	if id != "" {
+		args = append(args, "attemptedPeer", id)
+	}
+	log.Warn("Sync round has not finished and made no progress, downloader may be stalled", args...)
+}
+
+// stallProgress returns a lock-free monotonic proxy for how far the in-flight
+// sync round has advanced. It must stay non-blocking because it runs while a
+// later synchronization attempt is trying to return errBusy.
+func (d *Downloader) stallProgress() uint64 {
+	if d.stallProgressHook != nil {
+		return d.stallProgressHook()
+	}
+	return d.stallProgressSeq.Load()
+}
+
 // synchronise will select the peer and use it for synchronising. If an empty string is given
 // it will use the best peer possible and synchronize if its TD is higher than our own. If any of the
 // checks fail an error will be returned. This method is synchronous
@@ -407,9 +495,19 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode
 	}
 	// Make sure only one goroutine is ever allowed past this point at once
 	if !atomic.CompareAndSwapInt32(&d.synchronising, 0, 1) {
+		d.warnIfSyncStalled(id)
 		return errBusy
 	}
-	defer atomic.StoreInt32(&d.synchronising, 0)
+	d.stallProbeMu.Lock()
+	d.syncRoundSeq.Add(1)
+	d.syncRoundStart.Store(int64(mclock.Now()))
+	d.stallProbeMu.Unlock()
+	defer func() {
+		d.stallProbeMu.Lock()
+		d.syncRoundStart.Store(0)
+		d.stallProbeMu.Unlock()
+		atomic.StoreInt32(&d.synchronising, 0)
+	}()
 
 	// Post a user notification of the sync (only once per session)
 	if atomic.CompareAndSwapInt32(&d.notified, 0, 1) {
@@ -1511,6 +1609,7 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er
 						return fmt.Errorf("%w: stale headers", errBadPeer)
 					}
 				}
+				d.stallProgressSeq.Add(1)
 				headers = headers[limit:]
 				origin += uint64(limit)
 			}
@@ -1615,6 +1714,7 @@ func (d *Downloader) importBlockResults(results []*fetchResult) error {
 		}
 		return fmt.Errorf("%w: %v", errInvalidChain, err)
 	}
+	d.stallProgressSeq.Add(1)
 	if d.handleProposedBlock != nil {
 		header := blocks[len(blocks)-1].Header()
 		err := d.handleProposedBlock(header)
@@ -1878,6 +1978,7 @@ func (d *Downloader) commitFastSyncData(results []*fetchResult, stateSync *state
 		log.Debug("Downloaded item processing failed", "number", results[index].Header.Number, "hash", results[index].Header.Hash(), "err", err)
 		return fmt.Errorf("%w: %v", errInvalidChain, err)
 	}
+	d.stallProgressSeq.Add(1)
 	return nil
 }
 
@@ -1887,6 +1988,7 @@ func (d *Downloader) commitPivotBlock(result *fetchResult) error {
 	if _, err := d.blockchain.InsertReceiptChain([]*types.Block{block}, []types.Receipts{result.Receipts}); err != nil {
 		return err
 	}
+	d.stallProgressSeq.Add(1)
 	if err := d.blockchain.FastSyncCommitHead(block.Hash()); err != nil {
 		return err
 	}

eth/downloader/downloader_test.go

@@ -30,6 +30,7 @@ import (
 
 	ethereum "github.com/XinFinOrg/XDPoSChain"
 	"github.com/XinFinOrg/XDPoSChain/common"
+	"github.com/XinFinOrg/XDPoSChain/common/mclock"
 	engine_v2 "github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/engines/engine_v2"
 	"github.com/XinFinOrg/XDPoSChain/core"
 	"github.com/XinFinOrg/XDPoSChain/core/rawdb"
@@ -2299,3 +2300,202 @@ func TestRequestTTL(t *testing.T) {
 		t.Fatalf("ttlLimit (%v) is below rttMaxEstimate (%v)", ttlLimit, rttMaxEstimate)
 	}
 }
+
+// TestSynchroniseStalledSyncWarning checks that a sync round which never
+// releases the synchronising flag is reported once it stops making progress,
+// that the report is rate limited, and that fresh, progressing and finishing
+// rounds stay quiet. Without it a wedged round silently rejects every
+// subsequent attempt with errBusy and the node stops syncing without any log
+// line.
+func TestSynchroniseStalledSyncWarning(t *testing.T) {
+	var progress uint64
+	d := new(Downloader)
+	d.stallProgressHook = func() uint64 { return progress }
+	atomic.StoreInt32(&d.synchronising, 1)
+
+	// A round that just started must not be reported yet.
+	d.syncRoundStart.Store(int64(mclock.Now()))
+	if err := d.synchronise("peer", common.Hash{}, nil, FullSync); err != errBusy {
+		t.Fatalf("synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	if warned := d.lastStallWarn.Load(); warned != 0 {
+		t.Fatalf("fresh sync round reported as stalled")
+	}
+
+	// An old round that is still progressing must stay quiet: the first probe
+	// only captures a progress baseline.
+	progress = 10
+	old := int64(mclock.Now()) - int64(stalledSyncThreshold) - int64(time.Second)
+	d.syncRoundStart.Store(old)
+	if err := d.synchronise("peer", common.Hash{}, nil, FullSync); err != errBusy {
+		t.Fatalf("synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	if warned := d.lastStallWarn.Load(); warned != 0 {
+		t.Fatalf("progressing sync round reported as stalled")
+	}
+	if baseline := d.stallBaseline; baseline != progress {
+		t.Fatalf("progress baseline mismatch: have %d, want %d", baseline, progress)
+	}
+
+	// An old round that made no progress since the baseline must be reported.
+	if err := d.synchronise("peer", common.Hash{}, nil, FullSync); err != errBusy {
+		t.Fatalf("synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	warned := d.lastStallWarn.Load()
+	if warned == 0 {
+		t.Fatal("stalled sync round not reported")
+	}
+
+	// Further attempts within the rate limit window must stay quiet.
+	if err := d.synchronise("peer", common.Hash{}, nil, FullSync); err != errBusy {
+		t.Fatalf("synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	if again := d.lastStallWarn.Load(); again != warned {
+		t.Fatalf("stalled sync warning not rate limited: have %d, want %d", again, warned)
+	}
+
+	// A round that resumed progress must shift the baseline forward and stay
+	// quiet, so healthy long syncs never trip the stall detector.
+	progress = 20
+	if err := d.synchronise("peer", common.Hash{}, nil, FullSync); err != errBusy {
+		t.Fatalf("synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	if again := d.lastStallWarn.Load(); again != warned {
+		t.Fatalf("recovered sync round reported as stalled: have %d, want %d", again, warned)
+	}
+	if baseline := d.stallBaseline; baseline != progress {
+		t.Fatalf("progress baseline not shifted: have %d, want %d", baseline, progress)
+	}
+
+	// A round that is already tearing down must not be reported.
+	d.syncRoundStart.Store(0)
+	if err := d.synchronise("peer", common.Hash{}, nil, FullSync); err != errBusy {
+		t.Fatalf("synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	if again := d.lastStallWarn.Load(); again != warned {
+		t.Fatalf("finishing sync round reported as stalled: have %d, want %d", again, warned)
+	}
+}
+
+// TestSynchroniseStalledProbeNonBlocking checks that diagnostics on the busy
+// path never delay a synchronization attempt behind progress-reporting locks.
+func TestSynchroniseStalledProbeNonBlocking(t *testing.T) {
+	tester := newTester()
+	d := tester.downloader
+	atomic.StoreInt32(&d.synchronising, 1)
+	d.syncRoundStart.Store(int64(mclock.Now()) - int64(stalledSyncThreshold) - int64(time.Second))
+
+	d.syncStatsLock.Lock()
+	done := make(chan error, 1)
+	go func() {
+		done <- d.synchronise("peer", common.Hash{}, nil, FullSync)
+	}()
+	select {
+	case err := <-done:
+		d.syncStatsLock.Unlock()
+		if err != errBusy {
+			t.Fatalf("synchronise error mismatch: have %v, want %v", err, errBusy)
+		}
+	case <-time.After(100 * time.Millisecond):
+		d.syncStatsLock.Unlock()
+		<-done
+		t.Fatal("busy synchronisation attempt blocked on sync statistics lock")
+	}
+}
+
+// TestSynchroniseConcurrentStallProbes checks that overlapping busy attempts
+// cannot observe or update a partially completed stall-probe decision.
+func TestSynchroniseConcurrentStallProbes(t *testing.T) {
+	var calls atomic.Uint64
+	firstEntered := make(chan struct{})
+	releaseFirst := make(chan struct{})
+	d := new(Downloader)
+	d.stallProgressHook = func() uint64 {
+		if calls.Add(1) == 1 {
+			close(firstEntered)
+			<-releaseFirst
+		}
+		return 0
+	}
+	atomic.StoreInt32(&d.synchronising, 1)
+	d.syncRoundStart.Store(int64(mclock.Now()) - int64(stalledSyncThreshold) - int64(time.Second))
+
+	firstDone := make(chan error, 1)
+	go func() {
+		firstDone <- d.synchronise("first", common.Hash{}, nil, FullSync)
+	}()
+	<-firstEntered
+	if err := d.synchronise("second", common.Hash{}, nil, FullSync); err != errBusy {
+		t.Fatalf("second synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	close(releaseFirst)
+	if err := <-firstDone; err != errBusy {
+		t.Fatalf("first synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	if warned := d.lastStallWarn.Load(); warned != 0 {
+		t.Fatal("overlapping initial probes reported the sync as stalled")
+	}
+}
+
+// TestSynchroniseStallProbeRoundIsolation checks that a probe from a finishing
+// round cannot initialize or warn against the next round's state.
+func TestSynchroniseStallProbeRoundIsolation(t *testing.T) {
+	probeEntered := make(chan struct{})
+	releaseProbe := make(chan struct{})
+	d := new(Downloader)
+	d.stallProgressHook = func() uint64 {
+		close(probeEntered)
+		<-releaseProbe
+		return 0
+	}
+	atomic.StoreInt32(&d.synchronising, 1)
+	d.syncRoundSeq.Store(1)
+	old := int64(mclock.Now()) - int64(stalledSyncThreshold) - int64(time.Second)
+	d.syncRoundStart.Store(old)
+
+	oldDone := make(chan error, 1)
+	go func() {
+		oldDone <- d.synchronise("old", common.Hash{}, nil, FullSync)
+	}()
+	<-probeEntered
+	d.syncRoundSeq.Store(2)
+	d.syncRoundStart.Store(old + 1)
+	close(releaseProbe)
+	if err := <-oldDone; err != errBusy {
+		t.Fatalf("old synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	if warned := d.lastStallWarn.Load(); warned != 0 {
+		t.Fatal("probe from old round emitted a warning for the new round")
+	}
+	if d.stallProbed {
+		t.Fatal("probe from old round initialized the new round baseline")
+	}
+
+	d.stallProgressHook = func() uint64 { return 0 }
+	if err := d.synchronise("new", common.Hash{}, nil, FullSync); err != errBusy {
+		t.Fatalf("new synchronise error mismatch: have %v, want %v", err, errBusy)
+	}
+	if !d.stallProbed || d.stallProbeRound != 2 {
+		t.Fatalf("new round baseline mismatch: probed %t, round %d", d.stallProbed, d.stallProbeRound)
+	}
+}
+
+// TestSynchroniseRoundLifecycle checks that a successfully started round
+// stamps the round start marker and clears it together with the synchronising
+// flag on the way out, so stale state can never leak into the next round.
+func TestSynchroniseRoundLifecycle(t *testing.T) {
+	tester := newTester()
+	d := tester.downloader
+
+	// A round started without any known peer fails fast, but must still mark
+	// the round as in-flight and clean up both markers when it returns.
+	if err := d.synchronise("peer", common.Hash{}, nil, FullSync); err != errUnknownPeer {
+		t.Fatalf("synchronise error mismatch: have %v, want %v", err, errUnknownPeer)
+	}
+	if started := d.syncRoundStart.Load(); started != 0 {
+		t.Fatalf("round start marker not cleared: %d", started)
+	}
+	if flag := atomic.LoadInt32(&d.synchronising); flag != 0 {
+		t.Fatalf("synchronising flag not cleared: %d", flag)
+	}
+}

eth/downloader/statesync.go

@@ -625,6 +625,9 @@ func (s *stateSync) updateStats(written, duplicate, unexpected int, duration tim
 	s.d.syncStatsState.processed += uint64(written)
 	s.d.syncStatsState.duplicate += uint64(duplicate)
 	s.d.syncStatsState.unexpected += uint64(unexpected)
+	if written > 0 {
+		s.d.stallProgressSeq.Add(1)
+	}
 
 	if written > 0 || duplicate > 0 || unexpected > 0 {
 		log.Info("Imported new state entries", "count", written, "elapsed", common.PrettyDuration(duration), "processed", s.d.syncStatsState.processed, "pending", s.d.syncStatsState.pending, "trieretry", len(s.trieTasks), "coderetry", len(s.codeTasks), "duplicate", s.d.syncStatsState.duplicate, "unexpected", s.d.syncStatsState.unexpected)