diff --git a/eth/fetcher/block_fetcher.go b/eth/fetcher/block_fetcher.go index dbf87da9b3c7..caea7b6a3483 100644 --- a/eth/fetcher/block_fetcher.go +++ b/eth/fetcher/block_fetcher.go @@ -45,6 +45,21 @@ const ( blockLimit = 64 // Maximum number of unique blocks a peer may have delivered ) +// IsPlausibleAnnouncement reports whether a block announcement at the given +// number is within the fetcher's plausibility window of the current chain +// height. Untrusted announced numbers must be gated on this check before being +// recorded (e.g. as a peer's live tip), so a peer cannot inflate the recorded +// number beyond the window. +func IsPlausibleAnnouncement(number, height uint64) bool { + if number == 0 { + return true + } + if number <= height { + return height <= number+maxUncleDist + } + return number <= height+maxQueueDist +} + var ( blockAnnounceInMeter = metrics.NewRegisteredMeter("eth/fetcher/block/announces/in", nil) blockAnnounceOutTimer = metrics.NewRegisteredTimer("eth/fetcher/block/announces/out", nil) @@ -365,12 +380,11 @@ func (f *BlockFetcher) loop() { break } // If we have a valid block number, check that it's potentially useful - if notification.number > 0 { - if dist := int64(notification.number) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { - log.Debug("Peer discarded announcement", "peer", notification.origin, "number", notification.number, "hash", notification.hash, "distance", dist) - blockAnnounceDropMeter.Mark(1) - break - } + height := f.chainHeight() + if !IsPlausibleAnnouncement(notification.number, height) { + log.Debug("Peer discarded announcement", "peer", notification.origin, "number", notification.number, "hash", notification.hash, "height", height) + blockAnnounceDropMeter.Mark(1) + break } // All is well, schedule the announce if block's not yet downloading if _, ok := f.fetching[notification.hash]; ok { @@ -658,9 +672,11 @@ func (f *BlockFetcher) enqueue(peer string, block *types.Block) { f.forgetHash(hash) return } - // Discard any past or too distant blocks - if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { - log.Debug("Discarded propagated block, too far away", "peer", peer, "number", block.Number(), "hash", hash, "distance", dist) + // Discard any past or too distant blocks, using the same overflow-safe + // plausibility window as block announcements so a far-future number (e.g. + // MaxUint64) cannot wrap to a small distance and slip into the queue. + if !IsPlausibleAnnouncement(block.NumberU64(), f.chainHeight()) { + log.Debug("Discarded propagated block, too far away", "peer", peer, "number", block.Number(), "hash", hash) blockBroadcastDropMeter.Mark(1) f.forgetHash(hash) return diff --git a/eth/fetcher/block_fetcher_test.go b/eth/fetcher/block_fetcher_test.go index 5548b36a9de4..ca5550674d08 100644 --- a/eth/fetcher/block_fetcher_test.go +++ b/eth/fetcher/block_fetcher_test.go @@ -629,6 +629,56 @@ func TestDistantPropagationDiscarding(t *testing.T) { } } +// Tests that a far-future block whose number overflows int64 (e.g. MaxUint64) +// is discarded by enqueue, matching IsPlausibleAnnouncement, instead of being +// wrapped to a negative distance and accepted into the import queue. +func TestMaxUint64PropagationDiscarding(t *testing.T) { + // Build a low chain so the wrapped distance of a MaxUint64 block (-1) minus + // the head would fall inside the acceptance window under the old signed + // subtraction (height <= maxUncleDist). + hashes, blocks := makeChain(6, 0, genesis) + head := hashes[2] // block #4 + + tester := newTester() + tester.lock.Lock() + tester.hashes = []common.Hash{head} + tester.blocks = map[common.Hash]*types.Block{head: blocks[head]} + tester.lock.Unlock() + + // A far-future block must be discarded rather than queued. + maxFuture := types.NewBlockWithHeader(&types.Header{ + Number: new(big.Int).SetUint64(^uint64(0)), + Difficulty: big.NewInt(1), + }) + tester.fetcher.Enqueue("maxfuture", maxFuture) + time.Sleep(10 * time.Millisecond) + if !tester.fetcher.queue.Empty() { + t.Fatalf("fetcher queued far-future block") + } +} + +// Tests that IsPlausibleAnnouncement accepts announcements within the fetcher's +// plausibility window and rejects those too far from the local chain head. +func TestIsPlausibleAnnouncement(t *testing.T) { + height := uint64(1000) + tests := []struct { + number uint64 + want bool + }{ + {number: 0, want: true}, // Unknown height, always plausible + {number: height - maxUncleDist, want: true}, // Backward window edge + {number: height - maxUncleDist - 1, want: false}, // Too far behind + {number: height + maxQueueDist, want: true}, // Forward window edge + {number: height + maxQueueDist + 1, want: false}, // Too far ahead + {number: ^uint64(0), want: false}, // Far future + } + for i, tt := range tests { + if got := IsPlausibleAnnouncement(tt.number, height); got != tt.want { + t.Fatalf("case %d: number=%d got %v, want %v", i, tt.number, got, tt.want) + } + } +} + // Tests that announcements with numbers much lower or higher than out current // head get discarded to prevent wasting resources on useless blocks from faulty // peers. diff --git a/eth/handler.go b/eth/handler.go index 168ec3afc3e9..316c7d5f0f5f 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -756,9 +756,18 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&announces); err != nil { return errResp(ErrDecode, "%v: %v", msg, err) } - // Mark the hashes as present at the remote node + // Mark the hashes as present at the remote node and track the highest + // plausible announced block as the peer's live tip. + height := pm.blockchain.CurrentBlock().Number.Uint64() + var tip uint64 for _, block := range announces { p.MarkBlock(block.Hash) + if fetcher.IsPlausibleAnnouncement(block.Number, height) { + tip = max(tip, block.Number) + } + } + if tip > 0 { + p.SetTipNumber(tip) } // Schedule all the unknown hashes for retrieval unknown := make(newBlockHashesData, 0, len(announces)) @@ -791,8 +800,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { request.Block.ReceivedAt = msg.ReceivedAt request.Block.ReceivedFrom = p - // Mark the peer as owning the block and schedule it for import + // Mark the peer as owning the block, track its tip (only when the block + // number is within the fetcher's plausibility window), and schedule it + // for import p.MarkBlock(request.Block.Hash()) + if fetcher.IsPlausibleAnnouncement(request.Block.NumberU64(), pm.blockchain.CurrentBlock().Number.Uint64()) { + p.SetTipNumber(request.Block.NumberU64()) + } pm.blockFetcher.Enqueue(p.id, request.Block) // Assuming the block is importable by the peer, but possibly not yet done so, diff --git a/eth/peer.go b/eth/peer.go index 0c9916ce24fe..749b85de0cb6 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -96,6 +96,8 @@ type peer struct { td *big.Int lock sync.RWMutex + tipNumber uint64 // Highest block number announced by the peer (live network high-water mark) + knownBlocks mapset.Set[common.Hash] // Set of block hashes known to be known by this peer queuedBlocks chan *propEvent // Queue of blocks to broadcast to the peer queuedBlockAnns chan *types.Block // Queue of blocks to announce to the peer @@ -353,6 +355,25 @@ func (p *peer) SetHead(hash common.Hash, td *big.Int) { p.td.Set(td) } +// TipNumber retrieves the highest block number announced by the peer. +func (p *peer) TipNumber() uint64 { + p.lock.RLock() + defer p.lock.RUnlock() + return p.tipNumber +} + +// SetTipNumber records the highest block number announced by the peer. The tip +// only ever moves forward; unlike SetHead (which conservatively tracks the +// parent of an announced block), this reflects the actual tip the peer has +// advertised, so it can serve as a live network high-water mark. +func (p *peer) SetTipNumber(number uint64) { + p.lock.Lock() + defer p.lock.Unlock() + if number > p.tipNumber { + p.tipNumber = number + } +} + // MarkBlock marks a block as known for the peer, ensuring that the block will // never be propagated to this particular peer. func (p *peer) MarkBlock(hash common.Hash) { @@ -1051,6 +1072,22 @@ func (ps *peerSet) BestPeer() *peer { return bestPeer } +// HighestTipNumber returns the highest block number announced by any known +// peer, providing a live network high-water mark that stays current even when +// the downloader is idle. +func (ps *peerSet) HighestTipNumber() uint64 { + ps.lock.RLock() + defer ps.lock.RUnlock() + + var highest uint64 + for _, p := range ps.peers { + if tip := p.TipNumber(); tip > highest { + highest = tip + } + } + return highest +} + // Close disconnects all peers. // No new peers can be registered after Close has returned. func (ps *peerSet) Close() { diff --git a/eth/sync.go b/eth/sync.go index faa410d6491b..3d131ce64fd8 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -205,17 +205,7 @@ func (pm *ProtocolManager) syncStatusLogger() { for { select { case <-ticker.C: - if pm.downloader.Synchronising() { - progress := pm.downloader.Progress() - log.Warn("Block synchronisation in progress", - "starting", progress.StartingBlock, - "current", progress.CurrentBlock, - "highest", progress.HighestBlock, - "pulledStates", progress.PulledStates, - "knownStates", progress.KnownStates, - "peers", pm.peers.Len(), - ) - } + pm.reportSyncStatus() case <-pm.quitSync: return @@ -223,6 +213,57 @@ func (pm *ProtocolManager) syncStatusLogger() { } } +// reportSyncStatus emits a warn-level periodic sync status line, so that the +// current sync state is always visible in the logs every cycle regardless of +// whether the node is catching up or already in sync. +func (pm *ProtocolManager) reportSyncStatus() { + var ( + current uint64 + highest uint64 + ) + // Seed current/highest from the downloader while it is actively + // synchronising (it knows the discovered sync target and, in fast sync, + // reports the snap block as the current height). Otherwise seed both from + // the local chain head. computeSyncStatus then folds the live per-peer + // announced-tip high-water mark into highest in both states, so the + // reported highest always reflects the freshest known chain tip. + if pm.downloader.Synchronising() { + progress := pm.downloader.Progress() + current, highest = progress.CurrentBlock, progress.HighestBlock + } else { + current = pm.blockchain.CurrentBlock().Number.Uint64() + } + status := computeSyncStatus(current, highest, pm.peers.HighestTipNumber()) + log.Warn("Block synchronisation status", + "current", status.current, + "highest", status.highest, + "behind", status.behind, + "peers", pm.peers.Len(), + ) +} + +// syncStatus holds the values reported by the periodic sync status heartbeat. +type syncStatus struct { + current uint64 // Local head, or the fast-sync snap block while bulk syncing + highest uint64 // Highest known network block (downloader target + announced tips) + behind uint64 // Number of blocks behind the highest known network block +} + +// computeSyncStatus derives the heartbeat values. It always folds the live +// network high-water mark (the highest block announced by any peer, bounded by +// IsPlausibleAnnouncement) into the reported highest, regardless of whether the +// downloader is bulk-syncing. The reported highest is therefore the maximum of +// the downloader target, the local head and the live announced tip in every +// state, so it reflects the freshest known chain tip. +func computeSyncStatus(current, highest, announcedTip uint64) syncStatus { + highest = max(current, max(highest, announcedTip)) + behind := uint64(0) + if highest > current { + behind = highest - current + } + return syncStatus{current: current, highest: highest, behind: behind} +} + // synchronise tries to sync up our local block chain with a remote peer. func (pm *ProtocolManager) synchronise(peer *peer) { // Short circuit if no peers are available diff --git a/eth/sync_test.go b/eth/sync_test.go index 647a3a8a2b6f..348eb77492f7 100644 --- a/eth/sync_test.go +++ b/eth/sync_test.go @@ -17,20 +17,422 @@ package eth import ( + "bytes" + "fmt" + "math/big" + "strings" + "sync" "sync/atomic" "testing" "time" + "github.com/XinFinOrg/XDPoSChain/common" + "github.com/XinFinOrg/XDPoSChain/core/types" "github.com/XinFinOrg/XDPoSChain/eth/downloader" + "github.com/XinFinOrg/XDPoSChain/log" "github.com/XinFinOrg/XDPoSChain/p2p" "github.com/XinFinOrg/XDPoSChain/p2p/enode" ) +// lockedBuffer is a concurrency-safe log capture buffer. The protocol manager +// under test starts background goroutines (syncer, sync status logger, tx +// broadcast loops) and, in some tests, peer handler goroutines that keep +// emitting trace/debug logs while the test reads and resets the captured output, +// so a plain bytes.Buffer would race under -race and could corrupt the output. +// All access (write/read/reset) is serialized through a mutex. +type lockedBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *lockedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +func (b *lockedBuffer) Reset() { + b.mu.Lock() + defer b.mu.Unlock() + b.buf.Reset() +} + +// stalledDownloaderPeer is a downloader.Peer stub that answers the height probe +// and the common-ancestor search, then blocks all subsequent requests until +// released. It lets a test hold the downloader in a deterministic +// "synchronising with a discovered target" state without depending on a real +// sync still being in flight. +type stalledDownloaderPeer struct { + id string + download *downloader.Downloader + genesis *types.Header + + releaseCh chan struct{} + releaseOnce sync.Once + + mu sync.Mutex + answeredNumber bool // Whether the ancestor-search header request was answered +} + +func newStalledDownloaderPeer(dl *downloader.Downloader, id string, genesis *types.Header) *stalledDownloaderPeer { + return &stalledDownloaderPeer{ + id: id, + download: dl, + genesis: genesis, + releaseCh: make(chan struct{}), + } +} + +// release unblocks all pending requests so the stalled sync can wind down. +func (p *stalledDownloaderPeer) release() { + p.releaseOnce.Do(func() { close(p.releaseCh) }) +} + +// Head is unused by the downloader sync path (the head hash/TD are passed into +// Synchronise directly) but must exist to satisfy the downloader.Peer interface. +func (p *stalledDownloaderPeer) Head() (common.Hash, *big.Int) { + return p.genesis.Hash(), big.NewInt(1) +} + +// claimedHead is the header the stub advertises as the remote head, two blocks +// above the local genesis so the ancestor search converges on the genesis. +func (p *stalledDownloaderPeer) claimedHead() *types.Header { + return &types.Header{ + ParentHash: p.genesis.Hash(), + Number: new(big.Int).SetUint64(2), + Difficulty: big.NewInt(1), + } +} + +// RequestHeadersByHash answers the height probe with the claimed head header. +func (p *stalledDownloaderPeer) RequestHeadersByHash(h common.Hash, amount int, skip int, reverse bool) error { + return p.download.DeliverHeaders(p.id, []*types.Header{p.claimedHead()}) +} + +// RequestHeadersByNumber answers the ancestor search (the first call) with the +// genesis and the claimed head so the search converges, then blocks all +// subsequent bulk-download requests until released. +func (p *stalledDownloaderPeer) RequestHeadersByNumber(from uint64, amount int, skip int, reverse bool) error { + p.mu.Lock() + answer := !p.answeredNumber + p.answeredNumber = true + p.mu.Unlock() + if !answer { + <-p.releaseCh + return nil + } + // Reply with the genesis at number 0 plus synthetic headers matching the + // span request, so the downloader finds the genesis as the common ancestor. + step := uint64(skip + 1) + headers := make([]*types.Header, 0, amount) + for i := 0; i < amount; i++ { + n := from + uint64(i)*step + switch { + case n == 0: + headers = append(headers, p.genesis) + case n == 2: + headers = append(headers, p.claimedHead()) + default: + headers = append(headers, &types.Header{Number: new(big.Int).SetUint64(n)}) + } + } + return p.download.DeliverHeaders(p.id, headers) +} + +// RequestBodies blocks until released (the sync is paused during body download). +func (p *stalledDownloaderPeer) RequestBodies([]common.Hash) error { <-p.releaseCh; return nil } + +// RequestReceipts blocks until released. +func (p *stalledDownloaderPeer) RequestReceipts([]common.Hash) error { <-p.releaseCh; return nil } + +// RequestNodeData blocks until released. +func (p *stalledDownloaderPeer) RequestNodeData([]common.Hash) error { <-p.releaseCh; return nil } + // Tests that fast sync is disabled after a successful sync cycle. func TestFastSyncDisabling100(t *testing.T) { testFastSyncDisabling(t, xdc100) } func TestFastSyncDisabling164(t *testing.T) { testFastSyncDisabling(t, xdc164) } func TestFastSyncDisabling165(t *testing.T) { testFastSyncDisabling(t, xdc165) } +// Tests that the periodic sync status logger emits a status line on every cycle, +// reporting the live network high-water mark from the peers' announced tips, +// regardless of whether the node is catching up or already in sync. +func TestSyncStatusLogger(t *testing.T) { + pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil) + defer pm.Stop() + + // Capture the warn-level logs emitted by the sync status logger. + logBuf := new(lockedBuffer) + prevLog := log.Root() + glog := log.NewGlogHandler(log.NewTerminalHandlerWithLevel(logBuf, log.LevelTrace, false)) + glog.Verbosity(log.LevelTrace) + log.SetDefault(log.NewLogger(glog)) + defer log.SetDefault(prevLog) + + // Register a peer and give it a tip above our chain head. + app, net := p2p.MsgPipe() + defer app.Close() + peer := pm.newPeer(xdc100, p2p.NewPeer(enode.ID{1}, "sync-status-peer", nil), net, pm.txpool.Get) + if err := pm.peers.Register(peer); err != nil { + t.Fatalf("failed to register peer: %v", err) + } + defer pm.peers.Unregister(peer.id) + + current := pm.blockchain.CurrentBlock().Number.Uint64() + + // A peer ahead of us must surface the gap... + peer.SetTipNumber(current + 5) + pm.reportSyncStatus() + got := logBuf.String() + if count := strings.Count(got, "Block synchronisation status"); count != 1 { + t.Fatalf("expected exactly one sync status log, got %d, log: %q", count, got) + } + if !strings.Contains(got, fmt.Sprintf("highest=%d", current+5)) || + !strings.Contains(got, fmt.Sprintf("behind=%d", 5)) || !strings.Contains(got, "peers=1") { + t.Fatalf("expected live high-water mark and gap, got %q", got) + } + // ...and once no peer advertises a higher tip, no gap is reported. + logBuf.Reset() + if err := pm.peers.Unregister(peer.id); err != nil { + t.Fatalf("failed to unregister peer: %v", err) + } + pm.reportSyncStatus() + got = logBuf.String() + if !strings.Contains(got, "behind=0") { + t.Fatalf("expected no gap when at the peer's head, got %q", got) + } +} + +// Tests that a real NewBlockHashesMsg announcement updates the peer's tip, which +// the sync status heartbeat then surfaces as the network high-water mark. +func TestAnnouncementUpdatesPeerTip(t *testing.T) { + pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 8, nil, nil) + defer pm.Stop() + + // Create a connected peer and deliver a real block-hash announcement for a + // block the peer is known to have. + tp, _ := newTestPeer("announce", xdc165, pm, true) + defer tp.close() + + block := pm.blockchain.GetBlockByNumber(5) + if block == nil { + t.Fatalf("block #5 not found") + } + if err := p2p.Send(tp.app, NewBlockHashesMsg, newBlockHashesData{{Hash: block.Hash(), Number: block.NumberU64()}}); err != nil { + t.Fatalf("failed to send block announcement: %v", err) + } + // Wait for the protocol handler to record the announced tip. + deadline := time.After(5 * time.Second) + for tp.peer.TipNumber() != block.NumberU64() { + select { + case <-deadline: + t.Fatalf("peer tip not updated by NewBlockHashesMsg, have %d", tp.peer.TipNumber()) + case <-time.After(10 * time.Millisecond): + } + } + // The heartbeat must reflect the announced tip. + if highest := pm.peers.HighestTipNumber(); highest != block.NumberU64() { + t.Fatalf("expected highest tip %d, got %d", block.NumberU64(), highest) + } +} + +// Tests that a far-future block announcement (which the fetcher would discard) +// does not inflate the peer's recorded tip. +func TestAnnouncementRejectsFarFutureTip(t *testing.T) { + pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 8, nil, nil) + defer pm.Stop() + + tp, _ := newTestPeer("announce", xdc165, pm, true) + defer tp.close() + + // A plausible announcement within the fetcher's window is recorded as the + // peer's tip. + block := pm.blockchain.GetBlockByNumber(5) + if err := p2p.Send(tp.app, NewBlockHashesMsg, newBlockHashesData{{Hash: block.Hash(), Number: block.NumberU64()}}); err != nil { + t.Fatalf("failed to send block announcement: %v", err) + } + deadline := time.After(5 * time.Second) + for tp.peer.TipNumber() != block.NumberU64() { + select { + case <-deadline: + t.Fatalf("peer tip not updated by plausible announcement, have %d", tp.peer.TipNumber()) + case <-time.After(10 * time.Millisecond): + } + } + // A far-future announcement must be discarded and must not move the tip. + if err := p2p.Send(tp.app, NewBlockHashesMsg, newBlockHashesData{{Hash: common.Hash{0x01}, Number: ^uint64(0)}}); err != nil { + t.Fatalf("failed to send far-future announcement: %v", err) + } + // Observe the tip for a while to ensure the rejected announcement left it + // unchanged. + deadline = time.After(500 * time.Millisecond) + for { + if got := tp.peer.TipNumber(); got != block.NumberU64() { + t.Fatalf("far-future announcement moved the peer tip to %d, want %d", got, block.NumberU64()) + } + select { + case <-deadline: + return + case <-time.After(10 * time.Millisecond): + } + } +} + +// Tests that the NewBlockMsg handler records the peer's tip only for blocks +// within the fetcher's plausibility window, and ignores far-future blocks. +func TestNewBlockMsgUpdatesTip(t *testing.T) { + pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 8, nil, nil) + defer pm.Stop() + + tp, _ := newTestPeer("propagate", xdc165, pm, true) + defer tp.close() + + // A plausible propagated block records its number as the peer's tip. + plausible := pm.blockchain.GetBlockByNumber(5) + if plausible == nil { + t.Fatalf("block #5 not found") + } + if err := p2p.Send(tp.app, NewBlockMsg, []any{plausible, big.NewInt(131136)}); err != nil { + t.Fatalf("failed to send plausible NewBlockMsg: %v", err) + } + deadline := time.After(5 * time.Second) + for tp.peer.TipNumber() != plausible.NumberU64() { + select { + case <-deadline: + t.Fatalf("peer tip not updated by plausible NewBlockMsg, have %d", tp.peer.TipNumber()) + case <-time.After(10 * time.Millisecond): + } + } + // A far-future block (structurally valid but outside the plausibility + // window) must be ignored and must not move the tip. + future := types.NewBlockWithHeader(&types.Header{ + Number: new(big.Int).SetUint64(^uint64(0)), + Difficulty: big.NewInt(1), + UncleHash: types.EmptyUncleHash, + TxHash: types.EmptyRootHash, + }).WithBody(types.Body{}) + if err := p2p.Send(tp.app, NewBlockMsg, []any{future, big.NewInt(131136)}); err != nil { + t.Fatalf("failed to send far-future NewBlockMsg: %v", err) + } + // Observe the tip for a while to ensure the rejected block left it + // unchanged. + deadline = time.After(500 * time.Millisecond) + for { + if got := tp.peer.TipNumber(); got != plausible.NumberU64() { + t.Fatalf("far-future NewBlockMsg moved the peer tip to %d, want %d", got, plausible.NumberU64()) + } + select { + case <-deadline: + return + case <-time.After(10 * time.Millisecond): + } + } +} + +// Tests that the sync status heartbeat merges the downloader's discovered +// target with the live announced-tip high-water mark and computes the gap. +func TestComputeSyncStatus(t *testing.T) { + tests := []struct { + current, highest, announcedTip uint64 + wantHighest, wantBehind uint64 + }{ + // An active bulk sync target is reported even without announced tips. + {current: 0, highest: 100, announcedTip: 0, wantHighest: 100, wantBehind: 100}, + // The announced tip keeps the high-water mark live outside bulk syncs. + {current: 100, highest: 0, announcedTip: 150, wantHighest: 150, wantBehind: 50}, + // The higher of the two sources wins. + {current: 100, highest: 120, announcedTip: 130, wantHighest: 130, wantBehind: 30}, + // Fully in sync reports no gap. + {current: 120, highest: 120, announcedTip: 100, wantHighest: 120, wantBehind: 0}, + } + for i, tt := range tests { + got := computeSyncStatus(tt.current, tt.highest, tt.announcedTip) + if got.current != tt.current || got.highest != tt.wantHighest || got.behind != tt.wantBehind { + t.Fatalf("case %d: got %+v, want current=%d highest=%d behind=%d", i, got, tt.current, tt.wantHighest, tt.wantBehind) + } + } +} + +// Tests that the sync status heartbeat reports the downloader's discovered +// target while a bulk sync is actively running, instead of the announced-tip +// high-water mark (which peers may not populate during a bulk sync). The sync +// is paused deterministically at the downloader's bulk-download phase by a stub +// peer, so the test never relies on a real sync still being in flight. +func TestSyncStatusDuringSync(t *testing.T) { + pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil) + defer pm.Stop() + + // Capture the warn-level logs emitted by the sync status logger. + logBuf := new(lockedBuffer) + prevLog := log.Root() + glog := log.NewGlogHandler(log.NewTerminalHandlerWithLevel(logBuf, log.LevelTrace, false)) + glog.Verbosity(log.LevelTrace) + log.SetDefault(log.NewLogger(glog)) + defer log.SetDefault(prevLog) + + // Register a protocol peer so synchronise() has a best peer to pick, and + // advertise a head ahead of our own. + app, net := p2p.MsgPipe() + defer app.Close() + peer := pm.newPeer(xdc100, p2p.NewPeer(enode.ID{1}, "sync-status-peer", nil), net, pm.txpool.Get) + if err := pm.peers.Register(peer); err != nil { + t.Fatalf("failed to register peer: %v", err) + } + defer pm.peers.Unregister(peer.id) + + current := pm.blockchain.CurrentBlock() + localTD := pm.blockchain.GetTd(current.Hash(), current.Number.Uint64()) + peer.lock.Lock() + peer.head = current.Hash() + peer.td = new(big.Int).Add(localTD, big.NewInt(100)) + peer.lock.Unlock() + + // Register a stub downloader peer under the same id. It answers the height + // probe and the ancestor search, then blocks the bulk download, holding the + // downloader in a deterministic synchronising state with a known target. + stub := newStalledDownloaderPeer(pm.downloader, peer.id, pm.blockchain.Genesis().Header()) + if err := pm.downloader.RegisterPeer(peer.id, xdc100, stub); err != nil { + t.Fatalf("failed to register downloader peer: %v", err) + } + defer pm.downloader.UnregisterPeer(peer.id) + defer stub.release() + + // Kick off a sync in the background; the stub keeps it synchronising. + go pm.synchronise(pm.peers.BestPeer()) + + // Wait until the downloader is actively synchronising with a discovered + // target (the stub answers the height probe and the ancestor search, so the + // target is set while the bulk download is paused). + deadline := time.After(30 * time.Second) + ticker := time.NewTicker(time.Millisecond) + defer ticker.Stop() + for { + if pm.downloader.Synchronising() && pm.downloader.Progress().HighestBlock > 0 { + break + } + select { + case <-deadline: + t.Fatalf("downloader never reached an active sync with a discovered target") + case <-ticker.C: + } + } + // The sync is now deterministically paused with a known target: the + // heartbeat must report it rather than the (empty) announced-tip high-water + // mark. + target := pm.downloader.Progress().HighestBlock + logBuf.Reset() + pm.reportSyncStatus() + got := logBuf.String() + if !strings.Contains(got, fmt.Sprintf("highest=%d", target)) { + t.Fatalf("expected heartbeat to report downloader target %d during an active sync, got %q", target, got) + } +} + // Tests that fast sync gets disabled as soon as a real block is successfully // imported into the blockchain. func testFastSyncDisabling(t *testing.T, protocol int) {