diff --git a/netsync/manager_test.go b/netsync/manager_test.go index 177d0c57..0b900b96 100644 --- a/netsync/manager_test.go +++ b/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, ¶ms) + defer tearDown() + + blocks := generateTestBlocks(t, ¶ms, 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) +}