netsync: Rework sync model to use hdr annoucements.

This overhauls the primary code that deals with synchronizing the
blockchain with other peers on the network to use a model based on
header announcements instead of inventory vectors.

Currently, all blocks are discovered via a combination of inventory
vector announcements and the getblocks protocol message, both of which
only include the hash of the blocks in question.

This is not ideal because it means blocks are blindly downloaded based
on those announcements without knowing if they're actually likely to be
something that is useful since there is no further information such as
what blocks they connect to.  It also means that extra logic is needed
to deal with orphan blocks (those whose parents are not yet known) such
as caching them, determining known orphan roots, and expiration.

In short, the current method generally ends up wasting bandwidth and
also makes it much more difficult to detect certain classes of malicious
behavior.

The recently added capability of blockchain to process headers
independently from blocks while the block data is added out of order
later opened the door for a much better model that addresses all of the
aforementioned issues as well as paves the way for many other related
enhancements.

The new model discovers and downloads all of the headers prior to any
block data via the getheaders protocol message and then uses those
headers to determine exactly which blocks need to be downloaded to reach
the tip of the best chain.  Notably, this means orphan blocks are now a
thing of the past as blocks that do not connect are no longer
downloaded under any circumstance.

Further, the new model also makes use of sendheaders so that all block
announcements are made via the associated block header.  This in turn is
used to better determine if an announced block is likely to be useful
prior to downloading it.

It should be noted that the changes herein are intentionally limited to
those necessary to use the new sync model based on headers in an
incremental fashion to help simplify review and make it easier to assert
correctness.  There are many more improvements that this model paves the
way to support planned for future commits.  For example, syncing from
multiple peers in parallel and improved DoS protection.

The following is a high-level overview of the key features:

- All headers are downloaded and examined prior to downloading any
  blocks
- The initial header sync process:
  - Detects and recovers from stalled/malicious peers
- The concept of orphan blocks no longer exists
  - This means blocks which are not already known to connect are not
    downloaded
- All block announcements are handled via header announcements
  - Detects and prevents malicious behavior related to orphan headers
- The chain sync process:
  - Starts once the headers are downloaded and entails both downloading
    blocks as well as verifying them
  - Uses the headers to determine the best blocks to download
  - Pipelines the requests to increase throughput
  - Actively avoids downloading duplicate blocks
- The sync height is dynamically updated as new headers are discovered

blockchain/chain.go

@@ -348,6 +348,19 @@ func (b *BlockChain) DisableVerify(disable bool) {
 	b.chainLock.Unlock()
 }
 
+// HaveHeader returns whether or not the chain instance has the block header
+// represented by the passed hash.  Note that this will return true for both the
+// main chain and any side chains.
+//
+// This function is safe for concurrent access.
+func (b *BlockChain) HaveHeader(hash *chainhash.Hash) bool {
+	b.index.RLock()
+	node := b.index.index[*hash]
+	headerKnown := node != nil
+	b.index.RUnlock()
+	return headerKnown
+}
+
 // HaveBlock returns whether or not the chain instance has the block represented
 // by the passed hash.  This includes checking the various places a block can
 // be like part of the main chain or on a side chain.
@@ -1436,7 +1449,7 @@ func (b *BlockChain) isOldTimestamp(node *blockNode) bool {
 }
 
 // maybeUpdateIsCurrent potentially updates whether or not the chain believes it
-// is current.
+// is current using the provided best chain tip.
 //
 // It makes use of a latching approach such that once the chain becomes current
 // it will only switch back to false in the case no new blocks have been seen
@@ -1474,7 +1487,20 @@ func (b *BlockChain) maybeUpdateIsCurrent(curBest *blockNode) {
 		log.Debugf("Chain latched to current at block %s (height %d)",
 			curBest.hash, curBest.height)
 	}
+}
 
+// MaybeUpdateIsCurrent potentially updates whether or not the chain believes it
+// is current.
+//
+// It makes use of a latching approach such that once the chain becomes current
+// it will only switch back to false in the case no new blocks have been seen
+// for an extended period of time.
+//
+// This function is safe for concurrent access.
+func (b *BlockChain) MaybeUpdateIsCurrent() {
+	b.chainLock.Lock()
+	b.maybeUpdateIsCurrent(b.bestChain.Tip())
+	b.chainLock.Unlock()
 }
 
 // isCurrent returns whether or not the chain believes it is current based on
@@ -2189,9 +2215,8 @@ func New(ctx context.Context, config *Config) (*BlockChain, error) {
 		b.dbInfo.bidxVer)
 
 	tip := b.bestChain.Tip()
-	log.Infof("Chain state: height %d, hash %v, total transactions %d, "+
-		"work %v, stake version %v", tip.height, tip.hash,
-		b.stateSnapshot.TotalTxns, tip.workSum, 0)
+	log.Infof("Chain state: height %d, hash %v, total transactions %d, work %v",
+		tip.height, tip.hash, b.stateSnapshot.TotalTxns, tip.workSum)
 
 	return &b, nil
 }

blockchain/chainquery.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2020 The Decred developers
+// Copyright (c) 2018-2021 The Decred developers
 // Use of this source code is governed by an ISC
 // license that can be found in the LICENSE file.
 
@@ -154,3 +154,82 @@ func (b *BlockChain) BestInvalidHeader() chainhash.Hash {
 	b.index.RUnlock()
 	return hash
 }
+
+// NextNeededBlocks returns hashes for the next blocks after the current best
+// chain tip that are needed to make progress towards the current best known
+// header skipping any blocks that are already known or in the provided map of
+// blocks to exclude.  Typically the caller would want to exclude all blocks
+// that have outstanding requests.
+//
+// The maximum number of results is limited to the provided value or the maximum
+// allowed by the internal lookahead buffer in the case the requested number of
+// max results exceeds that value.
+//
+// This function is safe for concurrent access.
+func (b *BlockChain) NextNeededBlocks(maxResults uint8, exclude map[chainhash.Hash]struct{}) []*chainhash.Hash {
+	// Nothing to do when no results are requested.
+	if maxResults == 0 {
+		return nil
+	}
+
+	// Determine the common ancestor between the current best chain tip and the
+	// current best known header.  In practice this should never be nil because
+	// orphan headers are not allowed into the block index, but be paranoid and
+	// check anyway in case things change in the future.
+	b.index.RLock()
+	bestHeader := b.index.bestHeader
+	fork := b.bestChain.FindFork(bestHeader)
+	if fork == nil {
+		b.index.RUnlock()
+		return nil
+	}
+
+	// Determine the final block to consider for determining the next needed
+	// blocks by determining the descendants of the current best chain tip on
+	// the branch that leads to the best known header while clamping the number
+	// of descendants to consider to a lookahead buffer.
+	const lookaheadBuffer = 512
+	numBlocksToConsider := bestHeader.height - fork.height
+	if numBlocksToConsider == 0 {
+		b.index.RUnlock()
+		return nil
+	}
+	if numBlocksToConsider > lookaheadBuffer {
+		bestHeader = bestHeader.Ancestor(fork.height + lookaheadBuffer)
+		numBlocksToConsider = lookaheadBuffer
+	}
+
+	// Walk backwards from the final block to consider to the current best chain
+	// tip excluding any blocks that already have their data available or that
+	// the caller asked to be excluded (likely because they've already been
+	// requested).
+	neededBlocks := make([]*chainhash.Hash, 0, numBlocksToConsider)
+	for node := bestHeader; node != nil && node != fork; node = node.parent {
+		_, isExcluded := exclude[node.hash]
+		if isExcluded || node.status.HaveData() {
+			continue
+		}
+
+		neededBlocks = append(neededBlocks, &node.hash)
+	}
+	b.index.RUnlock()
+
+	// Reverse the needed blocks so they are in forwards order.
+	reverse := func(s []*chainhash.Hash) {
+		slen := len(s)
+		for i := 0; i < slen/2; i++ {
+			s[i], s[slen-1-i] = s[slen-1-i], s[i]
+		}
+	}
+	reverse(neededBlocks)
+
+	// Clamp the number of results to the lower of the requested max or number
+	// available.
+	if int64(maxResults) > numBlocksToConsider {
+		maxResults = uint8(numBlocksToConsider)
+	}
+	if uint16(len(neededBlocks)) > uint16(maxResults) {
+		neededBlocks = neededBlocks[:maxResults]
+	}
+	return neededBlocks
+}

peer/log.go

@@ -1,5 +1,5 @@
 // Copyright (c) 2015-2016 The btcsuite developers
-// Copyright (c) 2016-2020 The Decred developers
+// Copyright (c) 2016-2021 The Decred developers
 // Use of this source code is governed by an ISC
 // license that can be found in the LICENSE file.
 
@@ -188,7 +188,13 @@ func messageSummary(msg wire.Message) string {
 		return locatorSummary(msg.BlockLocatorHashes, &msg.HashStop)
 
 	case *wire.MsgHeaders:
-		return fmt.Sprintf("num %d", len(msg.Headers))
+		summary := fmt.Sprintf("num %d", len(msg.Headers))
+		if len(msg.Headers) > 0 {
+			finalHeader := msg.Headers[len(msg.Headers)-1]
+			summary = fmt.Sprintf("%s, final hash %s, height %d", summary,
+				finalHeader.BlockHash(), finalHeader.Height)
+		}
+		return summary
 
 	case *wire.MsgReject:
 		// Ensure the variable length strings don't contain any

server.go

@@ -673,7 +673,7 @@ func (sp *serverPeer) OnVersion(p *peer.Peer, msg *wire.MsgVersion) *wire.MsgRej
 
 	// Ignore peers that have a protocol version that is too old.  The peer
 	// negotiation logic will disconnect it after this callback returns.
-	if msg.ProtocolVersion < int32(wire.InitialProcotolVersion) {
+	if msg.ProtocolVersion < int32(wire.SendHeadersVersion) {
 		return nil
 	}
 
@@ -737,6 +737,13 @@ func (sp *serverPeer) OnVersion(p *peer.Peer, msg *wire.MsgVersion) *wire.MsgRej
 	return nil
 }
 
+// OnVerAck is invoked when a peer receives a verack wire message.  It creates
+// and sends a sendheaders message to request all block annoucements are made
+// via full headers instead of the inv message.
+func (sp *serverPeer) OnVerAck(_ *peer.Peer, msg *wire.MsgVerAck) {
+	sp.QueueMessage(wire.NewMsgSendHeaders(), nil)
+}
+
 // OnMemPool is invoked when a peer receives a mempool wire message.  It creates
 // and sends an inventory message with the contents of the memory pool up to the
 // maximum inventory allowed per message.
@@ -1067,12 +1074,6 @@ func (sp *serverPeer) OnInv(p *peer.Peer, msg *wire.MsgInv) {
 // OnHeaders is invoked when a peer receives a headers wire message.  The
 // message is passed down to the net sync manager.
 func (sp *serverPeer) OnHeaders(_ *peer.Peer, msg *wire.MsgHeaders) {
-	// Ban peers sending empty headers requests.
-	if len(msg.Headers) == 0 {
-		sp.server.BanPeer(sp)
-		return
-	}
-
 	sp.server.syncManager.QueueHeaders(msg, sp.Peer)
 }
 
@@ -2068,6 +2069,7 @@ func newPeerConfig(sp *serverPeer) *peer.Config {
 	return &peer.Config{
 		Listeners: peer.MessageListeners{
 			OnVersion:        sp.OnVersion,
+			OnVerAck:         sp.OnVerAck,
 			OnMemPool:        sp.OnMemPool,
 			OnGetMiningState: sp.OnGetMiningState,
 			OnMiningState:    sp.OnMiningState,
@@ -3508,6 +3510,11 @@ func newServer(ctx context.Context, listenAddrs []string, db database.DB, chainP
 		},
 	}
 	s.txMemPool = mempool.New(&txC)
+
+	// Create a new sync manager instance with the appropriate configuration.
+	if cfg.DisableCheckpoints {
+		srvrLog.Info("Checkpoints are disabled")
+	}
 	s.syncManager = netsync.New(&netsync.Config{
 		PeerNotifier: &s,
 		Chain:        s.chain,
@@ -3516,7 +3523,6 @@ func newServer(ctx context.Context, listenAddrs []string, db database.DB, chainP
 		RpcServer: func() *rpcserver.Server {
 			return s.rpcServer
 		},
-		DisableCheckpoints:    cfg.DisableCheckpoints,
 		NoMiningStateSync:     cfg.NoMiningStateSync,
 		MaxPeers:              cfg.MaxPeers,
 		MaxOrphanTxs:          cfg.MaxOrphanTxs,