netsync: add TestSyncStateMachine for end-to-end IBD sync flow

Tests the full IBD state machine: startSync → fetchHeaders →
handleHeadersMsg → fetchHeaderBlocks → handleBlockMsg → IBD complete.

Three sub-cases exercise different paths:
- fresh IBD from genesis
- node restart with partially synced blocks
- stalled sync peer recovery via handleStallSample

netsync/manager_test.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"encoding/hex"
 	"fmt"
+	"math"
 	"os"
 	"path/filepath"
 	"testing"
@@ -23,6 +24,12 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// The package-level log variable is nil by default. Set it to the
+// disabled logger so that log calls in the sync manager don't panic.
+func init() {
+	DisableLog()
+}
+
 // noopPeerNotifier is a no-op implementation of PeerNotifier for tests.
 type noopPeerNotifier struct{}
 
@@ -543,3 +550,552 @@ func TestIsInIBDMode(t *testing.T) {
 		tearDown()
 	}
 }
+
+// createTestCoinbase creates a minimal coinbase transaction for the given
+// block height.  The signature script encodes the height to ensure unique
+// transaction hashes across blocks.
+func createTestCoinbase(height int32, params *chaincfg.Params) *wire.MsgTx {
+	tx := wire.NewMsgTx(wire.TxVersion)
+
+	// Push the height as data to guarantee unique txids per block.
+	sigScript := []byte{
+		0x04,
+		byte(height), byte(height >> 8),
+		byte(height >> 16), byte(height >> 24),
+	}
+
+	tx.AddTxIn(&wire.TxIn{
+		PreviousOutPoint: wire.OutPoint{
+			Hash:  chainhash.Hash{},
+			Index: wire.MaxPrevOutIndex,
+		},
+		SignatureScript: sigScript,
+		Sequence:        wire.MaxTxInSequenceNum,
+	})
+
+	tx.AddTxOut(&wire.TxOut{
+		Value:    blockchain.CalcBlockSubsidy(height, params),
+		PkScript: []byte{txscript.OP_TRUE},
+	})
+
+	return tx
+}
+
+// solveTestBlock finds a nonce that satisfies the proof of work for the given
+// header.  With regression test parameters the difficulty is minimal and a
+// solution is found almost immediately.
+func solveTestBlock(header *wire.BlockHeader, params *chaincfg.Params) bool {
+	target := blockchain.CompactToBig(params.PowLimitBits)
+	for nonce := uint32(0); nonce < math.MaxUint32; nonce++ {
+		header.Nonce = nonce
+		hash := header.BlockHash()
+		if blockchain.HashToBig(&hash).Cmp(target) <= 0 {
+			return true
+		}
+	}
+
+	return false
+}
+
+// generateTestBlocks creates count valid blocks chaining from the genesis
+// block of the given params.  Each block contains only a coinbase transaction.
+func generateTestBlocks(
+	t *testing.T, params *chaincfg.Params, count int) []*btcutil.Block {
+
+	t.Helper()
+
+	blocks := make([]*btcutil.Block, 0, count)
+	prevHash := params.GenesisHash
+	prevTime := params.GenesisBlock.Header.Timestamp
+
+	for h := int32(1); h <= int32(count); h++ {
+		cb := createTestCoinbase(h, params)
+		merkleRoot := cb.TxHash()
+
+		header := wire.BlockHeader{
+			Version:    1,
+			PrevBlock:  *prevHash,
+			MerkleRoot: merkleRoot,
+			Timestamp:  prevTime.Add(time.Minute),
+			Bits:       params.PowLimitBits,
+		}
+		require.True(t, solveTestBlock(&header, params),
+			"failed to solve block at height %d", h)
+
+		msgBlock := &wire.MsgBlock{
+			Header:       header,
+			Transactions: []*wire.MsgTx{cb},
+		}
+		block := btcutil.NewBlock(msgBlock)
+		blocks = append(blocks, block)
+
+		bh := block.Hash()
+		prevHash = bh
+		prevTime = header.Timestamp
+	}
+
+	return blocks
+}
+
+// TestSyncStateMachine exercises the end-to-end IBD sync flow:
+//
+//	┌→ startSync
+//	│      ↓
+//	│  fetchHeaders
+//	│      ↓
+//	│  handleHeadersMsg
+//	│      ↓
+//	│  fetchHeaderBlocks ←┐
+//	│      ↓              │ (refill)
+//	│  handleBlockMsg ────┘──→ IBD complete
+//	│
+//	│  (stall detected at any phase above)
+//	│      ↓
+//	│  handleStallSample
+//	│      ↓
+//	└── handleDonePeerMsg
+//
+// It verifies that header processing transitions to block download, that the
+// pipeline refill path in handleBlockMsg is exercised, and that IBD mode is
+// properly cleared once the chain catches up to the best header.
+//
+// The "fresh ibd" case tests a complete sync from genesis: headers are fetched
+// and then blocks are downloaded.
+//
+// The "stall before any headers" and "stall mid header download" cases test
+// recovery when the sync peer stalls during header download.  A replacement
+// peer delivers the remaining (or all) headers and then all blocks.
+//
+// The "headers complete peer stalls on blocks" case tests recovery when the
+// sync peer delivers all headers but stalls before sending any blocks; a
+// replacement peer downloads all blocks.
+//
+// The "stalled sync peer recovery" case tests recovery mid-block-download: a
+// sync peer stops responding after some blocks, handleStallSample detects the
+// inactivity, the stalled peer is disconnected, and a replacement peer
+// finishes IBD.
+//
+// The "stall mid headers then stall on blocks" case combines both failure
+// modes: one peer stalls during headers (peer 2 takes over and finishes
+// headers), then peer 2 stalls during block download (peer 3 finishes blocks).
+// This exercises recovery across three distinct peers.
+func TestSyncStateMachine(t *testing.T) {
+	t.Parallel()
+
+	const testTotalBlocks = 2 * minInFlightBlocks
+
+	tests := []struct {
+		name        string
+		totalBlocks int
+
+		// stallHeadersAfter, when >= 0, triggers a stall during
+		// header download: deliver this many headers, then stall
+		// the sync peer and verify a replacement finishes header
+		// download.  Set to -1 for no header stall.
+		stallHeadersAfter int
+
+		// stallAfter, when >= 0, triggers a stall during block
+		// download: deliver all headers, then process this many
+		// blocks before stalling.  Set to -1 for no block stall.
+		stallAfter int
+	}{
+		{
+			name:              "fresh ibd",
+			totalBlocks:       testTotalBlocks,
+			stallHeadersAfter: -1,
+			stallAfter:        -1,
+		},
+		{
+			name:              "stall before any headers",
+			totalBlocks:       testTotalBlocks,
+			stallHeadersAfter: 0,
+			stallAfter:        -1,
+		},
+		{
+			name:              "stall mid header download",
+			totalBlocks:       testTotalBlocks,
+			stallHeadersAfter: testTotalBlocks / 2,
+			stallAfter:        -1,
+		},
+		{
+			name:              "headers complete peer stalls on blocks",
+			totalBlocks:       testTotalBlocks,
+			stallHeadersAfter: -1,
+			stallAfter:        0,
+		},
+		{
+			name:              "stalled sync peer recovery",
+			totalBlocks:       testTotalBlocks,
+			stallHeadersAfter: -1,
+			stallAfter:        5,
+		},
+		{
+			name:              "stall mid headers then stall on blocks",
+			totalBlocks:       testTotalBlocks,
+			stallHeadersAfter: testTotalBlocks / 2,
+			stallAfter:        5,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			params := chaincfg.RegressionNetParams
+			params.Checkpoints = nil
+
+			sm, tearDown := makeMockSyncManager(t, &params)
+			defer tearDown()
+
+			blocks := generateTestBlocks(t, &params, tc.totalBlocks)
+
+			// Register a sync candidate and call startSync,
+			// which activates IBD mode and sends getheaders.
+			peer1 := startIBD(t, sm, tc.totalBlocks)
+
+			if tc.stallHeadersAfter >= 0 {
+				// Stall during header download;
+				// replacement sends remaining headers.
+				peer2 := newSyncCandidate(t, sm,
+					int32(tc.totalBlocks))
+				syncStalledHeaderRecovery(
+					t, sm, peer1, peer2,
+					blocks, tc.stallHeadersAfter,
+					tc.totalBlocks,
+				)
+
+				if tc.stallAfter >= 0 {
+					peer3 := newSyncCandidate(t, sm,
+						int32(tc.totalBlocks))
+					syncStalledPeerRecovery(
+						t, sm, peer2,
+						peer3, blocks,
+						tc.stallAfter,
+						tc.totalBlocks,
+					)
+				} else {
+					syncProcessBlocks(t, sm,
+						peer2, blocks,
+						tc.totalBlocks)
+				}
+			} else {
+				syncSendHeaders(t, sm, peer1,
+					blocks, tc.totalBlocks)
+
+				if tc.stallAfter >= 0 {
+					peer2 := newSyncCandidate(t, sm,
+						int32(tc.totalBlocks))
+					syncStalledPeerRecovery(
+						t, sm, peer1,
+						peer2, blocks,
+						tc.stallAfter,
+						tc.totalBlocks,
+					)
+				} else {
+					syncProcessBlocks(t, sm,
+						peer1, blocks,
+						tc.totalBlocks)
+				}
+			}
+		})
+	}
+}
+
+// newSyncCandidate creates and registers a sync-candidate peer at the
+// given height without triggering startSync.
+func newSyncCandidate(t *testing.T, sm *SyncManager,
+	height int32) *peer.Peer {
+
+	t.Helper()
+
+	p := peer.NewInboundPeer(&peer.Config{
+		ChainParams: sm.chainParams,
+	})
+	p.UpdateLastBlockHeight(height)
+	sm.peerStates[p] = &peerSyncState{
+		syncCandidate:   true,
+		requestedTxns:   make(map[chainhash.Hash]struct{}),
+		requestedBlocks: make(map[chainhash.Hash]struct{}),
+	}
+	return p
+}
+
+// assertIBDComplete verifies that IBD finished: chain height matches
+// totalBlocks, ibdMode is off, and no blocks remain in-flight.
+func assertIBDComplete(t *testing.T, sm *SyncManager,
+	peerState *peerSyncState, totalBlocks int) {
+
+	t.Helper()
+
+	best := sm.chain.BestSnapshot()
+	require.Equal(t, int32(totalBlocks), best.Height)
+	require.False(t, sm.ibdMode,
+		"ibdMode should be off after catching up")
+	require.Empty(t, sm.requestedBlocks,
+		"all requested blocks should be fulfilled")
+	require.Empty(t, peerState.requestedBlocks,
+		"peer should have no outstanding block requests")
+}
+
+// startIBD registers a sync peer and calls startSync, verifying that IBD
+// mode is activated and the peer is selected.
+func startIBD(t *testing.T, sm *SyncManager,
+	peerHeight int) *peer.Peer {
+
+	t.Helper()
+
+	syncPeer := newSyncCandidate(t, sm, int32(peerHeight))
+
+	sm.startSync()
+
+	require.True(t, sm.syncPeer == syncPeer,
+		"syncPeer should be set after startSync")
+	require.True(t, sm.ibdMode, "ibdMode should be on")
+	require.False(t, sm.lastProgressTime.IsZero(),
+		"lastProgressTime should be set")
+
+	return syncPeer
+}
+
+// syncSendHeaders delivers block headers to the sync manager and verifies
+// that block requests are generated.
+func syncSendHeaders(t *testing.T, sm *SyncManager,
+	syncPeer *peer.Peer, blocks []*btcutil.Block, totalBlocks int) {
+
+	t.Helper()
+
+	// Record the progress time set by startIBD so we can verify
+	// that handleHeadersMsg advances it.
+	progressBefore := sm.lastProgressTime
+
+	headers := wire.NewMsgHeaders()
+	for _, block := range blocks {
+		err := headers.AddBlockHeader(&block.MsgBlock().Header)
+		require.NoError(t, err)
+	}
+
+	sm.handleHeadersMsg(&headersMsg{
+		headers: headers,
+		peer:    syncPeer,
+	})
+
+	_, bestHeaderHeight := sm.chain.BestHeader()
+	require.Equal(t, int32(totalBlocks), bestHeaderHeight)
+
+	require.True(t, sm.lastProgressTime.After(progressBefore),
+		"handleHeadersMsg should update lastProgressTime")
+
+	wantRequested := make(map[chainhash.Hash]struct{}, len(blocks))
+	for _, block := range blocks {
+		wantRequested[*block.Hash()] = struct{}{}
+	}
+	require.Equal(t, wantRequested, sm.requestedBlocks)
+	require.Equal(t, wantRequested, sm.peerStates[syncPeer].requestedBlocks)
+}
+
+// syncProcessBlocks feeds all blocks to handleBlockMsg and verifies that IBD
+// mode remains active until the final block, at which point IBD completes.
+func syncProcessBlocks(t *testing.T, sm *SyncManager, syncPeer *peer.Peer,
+	blocks []*btcutil.Block, totalBlocks int) {
+
+	t.Helper()
+
+	peerState := sm.peerStates[syncPeer]
+
+	for i, block := range blocks {
+		sm.handleBlockMsg(&blockMsg{
+			block: block,
+			peer:  syncPeer,
+			reply: make(chan struct{}, 1),
+		})
+
+		if i < len(blocks)-1 {
+			require.True(t, sm.ibdMode,
+				"ibdMode should still be on at height %d", i+1)
+		}
+	}
+
+	assertIBDComplete(t, sm, peerState, totalBlocks)
+}
+
+// syncStalledPeerRecovery processes stallAfter blocks from stalledPeer,
+// triggers stall detection, verifies that stalledPeer is removed and
+// replacementPeer takes over, then feeds remaining blocks and verifies
+// IBD completes.
+func syncStalledPeerRecovery(t *testing.T, sm *SyncManager,
+	stalledPeer, replacementPeer *peer.Peer,
+	blocks []*btcutil.Block, stallAfter, totalBlocks int) {
+
+	t.Helper()
+
+	// Process the first stallAfter blocks from the stalled peer.
+	for _, block := range blocks[:stallAfter] {
+		sm.handleBlockMsg(&blockMsg{
+			block: block,
+			peer:  stalledPeer,
+			reply: make(chan struct{}, 1),
+		})
+	}
+
+	best := sm.chain.BestSnapshot()
+	require.Equal(t, int32(stallAfter), best.Height)
+	require.True(t, sm.ibdMode)
+
+	// Trigger stall detection.
+	sm.lastProgressTime = time.Now().Add(
+		-(maxStallDuration + time.Minute))
+	sm.handleStallSample()
+
+	// Verify that handleStallSample called Disconnect() on the
+	// stalled peer (which closes p.quit, making WaitForDisconnect
+	// return immediately).
+	disconnected := make(chan struct{})
+	go func() {
+		stalledPeer.WaitForDisconnect()
+		close(disconnected)
+	}()
+	select {
+	case <-disconnected:
+	case <-time.After(time.Second):
+		t.Fatal("Disconnect() was not called on stalled peer")
+	}
+
+	// Snapshot the stalled peer's outstanding requested blocks before
+	// disconnection so we can verify they are cleaned up.
+	stalledState := sm.peerStates[stalledPeer]
+	stalledRequested := make([]chainhash.Hash, 0, len(stalledState.requestedBlocks))
+	for hash := range stalledState.requestedBlocks {
+		stalledRequested = append(stalledRequested, hash)
+	}
+	require.NotEmpty(t, stalledRequested,
+		"stalled peer should have outstanding requested blocks")
+
+	// In production, Disconnect() triggers handleDonePeerMsg
+	// asynchronously via the peer goroutine.  Call it directly to
+	// complete the removal.  Note: handleDonePeerMsg first clears the
+	// stalled peer's requested blocks from the global map via
+	// clearRequestedState, then updateSyncPeer → startSync immediately
+	// re-requests them for the replacement peer.
+	sm.handleDonePeerMsg(stalledPeer)
+
+	_, stalledTracked := sm.peerStates[stalledPeer]
+	require.False(t, stalledTracked,
+		"stalled peer should be removed")
+	require.True(t, sm.syncPeer == replacementPeer,
+		"replacement peer should take over as sync peer")
+	require.True(t, sm.ibdMode)
+
+	// Verify that the replacement peer re-requested the exact same
+	// blocks that were outstanding from the stalled peer.
+	replacementState := sm.peerStates[replacementPeer]
+	require.Equal(t, len(stalledRequested),
+		len(replacementState.requestedBlocks),
+		"replacement peer should request same number of blocks")
+	for _, hash := range stalledRequested {
+		_, exists := replacementState.requestedBlocks[hash]
+		require.True(t, exists,
+			"block %v should be requested from replacement peer",
+			hash)
+	}
+
+	// Feed remaining blocks from the replacement peer.
+	for _, block := range blocks[stallAfter:] {
+		sm.handleBlockMsg(&blockMsg{
+			block: block,
+			peer:  replacementPeer,
+			reply: make(chan struct{}, 1),
+		})
+	}
+
+	assertIBDComplete(t, sm, replacementState, totalBlocks)
+}
+
+// syncStalledHeaderRecovery simulates a stall during header download.
+// It optionally delivers headersSent headers from stalledPeer, triggers stall
+// detection, verifies that stalledPeer is removed and replacementPeer takes
+// over, then delivers remaining headers and verifies block requests are
+// generated.  The caller is responsible for the block-download phase.
+func syncStalledHeaderRecovery(t *testing.T, sm *SyncManager,
+	stalledPeer, replacementPeer *peer.Peer,
+	blocks []*btcutil.Block, headersSent, totalBlocks int) {
+
+	t.Helper()
+
+	// Deliver partial headers from the stalled peer.  When
+	// headersSent is 0, this is a no-op (peer stalls immediately).
+	if headersSent > 0 {
+		headers := wire.NewMsgHeaders()
+		for _, block := range blocks[:headersSent] {
+			err := headers.AddBlockHeader(
+				&block.MsgBlock().Header)
+			require.NoError(t, err)
+		}
+
+		sm.handleHeadersMsg(&headersMsg{
+			headers: headers,
+			peer:    stalledPeer,
+		})
+
+		_, bestHeaderHeight := sm.chain.BestHeader()
+		require.Equal(t, int32(headersSent), bestHeaderHeight)
+	}
+
+	// No blocks should have been requested during header download
+	// since the headers haven't caught up to the peer's height yet.
+	require.Empty(t, sm.requestedBlocks,
+		"no blocks should be requested during header download")
+
+	// Trigger stall detection.
+	sm.lastProgressTime = time.Now().Add(
+		-(maxStallDuration + time.Minute))
+	sm.handleStallSample()
+
+	// Verify that handleStallSample called Disconnect() on the
+	// stalled peer.
+	disconnected := make(chan struct{})
+	go func() {
+		stalledPeer.WaitForDisconnect()
+		close(disconnected)
+	}()
+	select {
+	case <-disconnected:
+	case <-time.After(time.Second):
+		t.Fatal("Disconnect() was not called on stalled peer")
+	}
+
+	// Complete peer removal.  handleDonePeerMsg clears state and
+	// triggers startSync which selects the replacement peer.
+	sm.handleDonePeerMsg(stalledPeer)
+
+	_, stalledTracked := sm.peerStates[stalledPeer]
+	require.False(t, stalledTracked,
+		"stalled peer should be removed")
+	require.True(t, sm.syncPeer == replacementPeer,
+		"replacement peer should take over as sync peer")
+	require.True(t, sm.ibdMode)
+
+	// Deliver remaining headers from the replacement peer.  When
+	// headersSent is 0, this is all headers.
+	remainingHeaders := wire.NewMsgHeaders()
+	for _, block := range blocks[headersSent:] {
+		err := remainingHeaders.AddBlockHeader(
+			&block.MsgBlock().Header)
+		require.NoError(t, err)
+	}
+	sm.handleHeadersMsg(&headersMsg{
+		headers: remainingHeaders,
+		peer:    replacementPeer,
+	})
+
+	_, bestHeaderHeight := sm.chain.BestHeader()
+	require.Equal(t, int32(totalBlocks), bestHeaderHeight)
+
+	// Verify all blocks were requested from the replacement.
+	wantRequested := make(map[chainhash.Hash]struct{}, len(blocks))
+	for _, block := range blocks {
+		wantRequested[*block.Hash()] = struct{}{}
+	}
+	require.Equal(t, wantRequested, sm.requestedBlocks)
+	replacementState := sm.peerStates[replacementPeer]
+	require.Equal(t, wantRequested, replacementState.requestedBlocks)
+}