netsync: change fetchHeaderBlocks to be based on the processed block headers

diff --git a/netsync/manager.go b/netsync/manager.go
index fe8464f..d8fd1fb 100644
--- a/netsync/manager.go
+++ b/netsync/manager.go
@@ -1006,25 +1006,40 @@ func (sm *SyncManager) handleBlockMsg(bmsg *blockMsg) {
 // fetchHeaderBlocks creates and sends a request to the syncPeer for the next
 // list of blocks to be downloaded based on the current list of headers.
 func (sm *SyncManager) fetchHeaderBlocks() {
-	// Nothing to do if there is no start header.
-	if sm.startHeader == nil {
-		log.Warnf("fetchHeaderBlocks called with no start header")
-		return
+	gdmsg := sm.buildBlockRequest()
+	if len(gdmsg.InvList) > 0 {
+		sm.syncPeer.QueueMessage(gdmsg, nil)
 	}
+}
 
-	// Build up a getdata request for the list of blocks the headers
-	// describe.  The size hint will be limited to wire.MaxInvPerMsg by
-	// the function, so no need to double check it here.
-	gdmsg := wire.NewMsgGetDataSizeHint(uint(sm.headerList.Len()))
+// buildBlockRequest builds a getdata message for blocks that need to be
+// downloaded based on the current list of headers.
+//
+// Start fetching from the fork point between the best chain and
+// the best header chain rather than from the best chain height.
+// When the best header chain has diverged (e.g. due to a reorg),
+// blocks between the fork point and the current height on the new
+// chain are different and must also be downloaded.
+func (sm *SyncManager) buildBlockRequest() *wire.MsgGetData {
+	_, bestHeaderHeight := sm.chain.BestHeader()
+	forkHeight := sm.chain.BestChainHeaderForkHeight()
+	if bestHeaderHeight < forkHeight {
+		// Should never happen but we're guarding against the uint cast
+		// that happens below.
+		return wire.NewMsgGetDataSizeHint(0)
+	}
+	length := bestHeaderHeight - forkHeight
+	gdmsg := wire.NewMsgGetDataSizeHint(uint(length))
 	numRequested := 0
-	for e := sm.startHeader; e != nil; e = e.Next() {
-		node, ok := e.Value.(*headerNode)
-		if !ok {
-			log.Warn("Header list node type is not a headerNode")
-			continue
+	for h := forkHeight + 1; h <= bestHeaderHeight; h++ {
+		hash, err := sm.chain.HeaderHashByHeight(h)
+		if err != nil {
+			log.Warnf("error while fetching the block hash for height %v -- %v",
+				h, err)
+			return gdmsg
 		}
 
-		iv := wire.NewInvVect(wire.InvTypeBlock, node.hash)
+		iv := wire.NewInvVect(wire.InvTypeBlock, hash)
 		haveInv, err := sm.haveInventory(iv)
 		if err != nil {
 			log.Warnf("Unexpected failure when checking for "+
@@ -1032,10 +1047,20 @@ func (sm *SyncManager) fetchHeaderBlocks() {
 				"fetch: %v", err)
 		}
 		if !haveInv {
+			// Skip blocks that are already in-flight to avoid
+			// sending duplicate getdata requests. Duplicates
+			// cause the peer to send the block twice; the second
+			// copy arrives after the first has been processed and
+			// removed from requestedBlocks, triggering an
+			// "unrequested block" disconnect.
+			if _, exists := sm.requestedBlocks[*hash]; exists {
+				continue
+			}
+
 			syncPeerState := sm.peerStates[sm.syncPeer]
 
-			sm.requestedBlocks[*node.hash] = struct{}{}
-			syncPeerState.requestedBlocks[*node.hash] = struct{}{}
+			sm.requestedBlocks[*hash] = struct{}{}
+			syncPeerState.requestedBlocks[*hash] = struct{}{}
 
 			// If we're fetching from a witness enabled peer
 			// post-fork, then ensure that we receive all the
@@ -1047,14 +1072,12 @@ func (sm *SyncManager) fetchHeaderBlocks() {
 			gdmsg.AddInvVect(iv)
 			numRequested++
 		}
-		sm.startHeader = e.Next()
+
 		if numRequested >= wire.MaxInvPerMsg {
 			break
 		}
 	}
-	if len(gdmsg.InvList) > 0 {
-		sm.syncPeer.QueueMessage(gdmsg, nil)
-	}
+	return gdmsg
 }
 
 // handleHeadersMsg handles block header messages from all peers.  Headers are
diff --git a/netsync/manager_test.go b/netsync/manager_test.go
index 8bf1d68..05c6cac 100644
--- a/netsync/manager_test.go
+++ b/netsync/manager_test.go
@@ -313,6 +313,107 @@ func (m *mockTimeSource) Offset() time.Duration {
 	return 0
 }
 
+// TestBuildBlockRequestSkipsInflightBlocks verifies that buildBlockRequest
+// does not re-request blocks that are already in sm.requestedBlocks.  When
+// the pipeline refill threshold triggers fetchHeaderBlocks while blocks are
+// still in-flight, re-requesting them causes the peer to send duplicates.
+// The first copy gets processed (removing the hash from requestedBlocks),
+// and the second copy then arrives as "unrequested", disconnecting the peer.
+func TestBuildBlockRequestSkipsInflightBlocks(t *testing.T) {
+	tests := []struct {
+		name string
+		// inflightHeights are the block heights already in
+		// requestedBlocks before calling buildBlockRequest.
+		inflightHeights []int32
+		// wantRequestedHeights are the block heights that should
+		// appear in the returned getdata message.
+		wantRequestedHeights []int32
+	}{
+		{
+			name:                 "no blocks in-flight requests all",
+			inflightHeights:      nil,
+			wantRequestedHeights: []int32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
+		},
+		{
+			name:                 "all blocks in-flight requests none",
+			inflightHeights:      []int32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
+			wantRequestedHeights: nil,
+		},
+		{
+			name:                 "first 5 in-flight requests remaining 6",
+			inflightHeights:      []int32{1, 2, 3, 4, 5},
+			wantRequestedHeights: []int32{6, 7, 8, 9, 10, 11},
+		},
+		{
+			name:                 "last 6 in-flight requests first 5",
+			inflightHeights:      []int32{6, 7, 8, 9, 10, 11},
+			wantRequestedHeights: []int32{1, 2, 3, 4, 5},
+		},
+		{
+			name:                 "scattered in-flight requests gaps",
+			inflightHeights:      []int32{2, 4, 6, 8, 10},
+			wantRequestedHeights: []int32{1, 3, 5, 7, 9, 11},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			params := chaincfg.MainNetParams
+			params.Checkpoints = nil
+			sm, tearDown := makeMockSyncManager(t, &params)
+			defer tearDown()
+
+			// Process headers 1-11 so the header chain is
+			// ahead of the block chain.
+			headers := loadHeaders(t)
+			for _, header := range headers {
+				_, err := sm.chain.ProcessBlockHeader(
+					header, blockchain.BFNone, false)
+				require.NoError(t, err)
+			}
+
+			// Set up a disconnected peer as syncPeer.
+			syncPeer := peer.NewInboundPeer(&peer.Config{})
+			sm.syncPeer = syncPeer
+			syncPeerState := &peerSyncState{
+				requestedTxns:   make(map[chainhash.Hash]struct{}),
+				requestedBlocks: make(map[chainhash.Hash]struct{}),
+			}
+			sm.peerStates[syncPeer] = syncPeerState
+
+			// Pre-populate in-flight blocks.
+			for _, h := range tc.inflightHeights {
+				hash, err := sm.chain.HeaderHashByHeight(h)
+				require.NoError(t, err)
+				sm.requestedBlocks[*hash] = struct{}{}
+				syncPeerState.requestedBlocks[*hash] = struct{}{}
+			}
+
+			gdmsg := sm.buildBlockRequest()
+
+			// Collect the hashes from the getdata message.
+			got := make(map[chainhash.Hash]struct{}, len(gdmsg.InvList))
+			for _, iv := range gdmsg.InvList {
+				got[iv.Hash] = struct{}{}
+			}
+
+			require.Equal(t, len(tc.wantRequestedHeights), len(gdmsg.InvList))
+			for _, h := range tc.wantRequestedHeights {
+				hash, err := sm.chain.HeaderHashByHeight(h)
+				require.NoError(t, err)
+				require.Contains(t, got, *hash,
+					"block at height %d should be requested", h)
+			}
+			for _, h := range tc.inflightHeights {
+				hash, err := sm.chain.HeaderHashByHeight(h)
+				require.NoError(t, err)
+				require.NotContains(t, got, *hash,
+					"in-flight block at height %d should not be re-requested", h)
+			}
+		})
+	}
+}
+
 func TestIsInIBDMode(t *testing.T) {
 	tests := []struct {
 		peerState  map[*peer.Peer]*peerSyncState
@@ -402,10 +503,8 @@ func TestIsInIBDMode(t *testing.T) {
 		},
 	}
 
-	for i, test := range tests {
-		db, tearDown, err := dbSetup(
-			fmt.Sprintf("TestIsInIBDMode-%v", i),
-			test.params)
+	for _, test := range tests {
+		db, tearDown, err := dbSetup(t, test.params)
 		if err != nil {
 			tearDown()
 			t.Fatal(err)