blockchain: fix fast sync halt with initial height > 1 (#5249)

Blockchain reactors were not updated to handle arbitrary initial height after #5191.

blockchain/v0/reactor.go

@@ -75,11 +75,11 @@ func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *st
 	const capacity = 1000                      // must be bigger than peers count
 	errorsCh := make(chan peerError, capacity) // so we don't block in #Receive#pool.AddBlock
 
-	pool := NewBlockPool(
-		store.Height()+1,
-		requestsCh,
-		errorsCh,
-	)
+	startHeight := store.Height() + 1
+	if startHeight == 1 {
+		startHeight = state.InitialHeight
+	}
+	pool := NewBlockPool(startHeight, requestsCh, errorsCh)
 
 	bcR := &BlockchainReactor{
 		initialState: state,

blockchain/v1/reactor.go

@@ -84,6 +84,9 @@ func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *st
 	errorsForFSMCh := make(chan bcReactorMessage, capacity)
 
 	startHeight := store.Height() + 1
+	if startHeight == 1 {
+		startHeight = state.InitialHeight
+	}
 	bcR := &BlockchainReactor{
 		initialState:     state,
 		state:            state,

blockchain/v2/reactor.go

@@ -59,7 +59,11 @@ type blockApplier interface {
 // XXX: unify naming in this package around tmState
 func newReactor(state state.State, store blockStore, reporter behaviour.Reporter,
 	blockApplier blockApplier, fastSync bool) *BlockchainReactor {
-	scheduler := newScheduler(state.LastBlockHeight, time.Now())
+	initHeight := state.LastBlockHeight + 1
+	if initHeight == 1 {
+		initHeight = state.InitialHeight
+	}
+	scheduler := newScheduler(initHeight, time.Now())
 	pContext := newProcessorContext(store, blockApplier, state)
 	// TODO: Fix naming to just newProcesssor
 	// newPcState requires a processorContext

blockchain/v2/scheduler.go

@@ -181,7 +181,7 @@ func newScheduler(initHeight int64, startTime time.Time) *scheduler {
 		initHeight:     initHeight,
 		lastAdvance:    startTime,
 		syncTimeout:    60 * time.Second,
-		height:         initHeight + 1,
+		height:         initHeight,
 		blockStates:    make(map[int64]blockState),
 		peers:          make(map[p2p.ID]*scPeer),
 		pendingBlocks:  make(map[int64]p2p.ID),
@@ -636,8 +636,12 @@ func (sc *scheduler) handleTryPrunePeer(event rTryPrunePeer) (Event, error) {
 }
 
 func (sc *scheduler) handleResetState(event bcResetState) (Event, error) {
-	sc.initHeight = event.state.LastBlockHeight + 1
-	sc.height = event.state.LastBlockHeight + 1
+	initHeight := event.state.LastBlockHeight + 1
+	if initHeight == 1 {
+		initHeight = event.state.InitialHeight
+	}
+	sc.initHeight = initHeight
+	sc.height = initHeight
 	sc.lastAdvance = time.Now()
 	sc.addNewBlocks()
 	return noOp, nil

blockchain/v2/scheduler_test.go

@@ -44,7 +44,11 @@ func newTestScheduler(params scTestParams) *scheduler {
 	peers := make(map[p2p.ID]*scPeer)
 	var maxHeight int64
 
-	sc := newScheduler(params.initHeight, params.startTime)
+	initHeight := params.initHeight
+	if initHeight == 0 {
+		initHeight = 1
+	}
+	sc := newScheduler(initHeight, params.startTime)
 	if params.height != 0 {
 		sc.height = params.height
 	}
@@ -97,7 +101,8 @@ func TestScInit(t *testing.T) {
 		initHeight int64 = 5
 		sc               = newScheduler(initHeight, time.Now())
 	)
-	assert.Equal(t, blockStateProcessed, sc.getStateAtHeight(initHeight))
+	assert.Equal(t, blockStateProcessed, sc.getStateAtHeight(initHeight-1))
+	assert.Equal(t, blockStateUnknown, sc.getStateAtHeight(initHeight))
 	assert.Equal(t, blockStateUnknown, sc.getStateAtHeight(initHeight+1))
 }
 
@@ -116,9 +121,8 @@ func TestScMaxHeights(t *testing.T) {
 		{
 			name: "one ready peer",
 			sc: scheduler{
-				initHeight: 2,
-				height:     3,
-				peers:      map[p2p.ID]*scPeer{"P1": {height: 6, state: peerStateReady}},
+				height: 3,
+				peers:  map[p2p.ID]*scPeer{"P1": {height: 6, state: peerStateReady}},
 			},
 			wantMax: 6,
 		},
@@ -1161,14 +1165,13 @@ func TestScNextHeightToSchedule(t *testing.T) {
 	}{
 		{
 			name:       "no blocks",
-			fields:     scTestParams{initHeight: 10, height: 11},
+			fields:     scTestParams{initHeight: 11, height: 11},
 			wantHeight: -1,
 		},
 		{
 			name: "only New blocks",
 			fields: scTestParams{
-				initHeight: 2,
-				height:     3,
+				initHeight: 3,
 				peers:      map[string]*scPeer{"P1": {height: 6, state: peerStateReady}},
 				allB:       []int64{3, 4, 5, 6},
 			},
@@ -1177,7 +1180,7 @@ func TestScNextHeightToSchedule(t *testing.T) {
 		{
 			name: "only Pending blocks",
 			fields: scTestParams{
-				height:      1,
+				initHeight:  1,
 				peers:       map[string]*scPeer{"P1": {height: 4, state: peerStateReady}},
 				allB:        []int64{1, 2, 3, 4},
 				pending:     map[int64]p2p.ID{1: "P1", 2: "P1", 3: "P1", 4: "P1"},
@@ -1188,26 +1191,26 @@ func TestScNextHeightToSchedule(t *testing.T) {
 		{
 			name: "only Received blocks",
 			fields: scTestParams{
-				height:   1,
-				peers:    map[string]*scPeer{"P1": {height: 4, state: peerStateReady}},
-				allB:     []int64{1, 2, 3, 4},
-				received: map[int64]p2p.ID{1: "P1", 2: "P1", 3: "P1", 4: "P1"},
+				initHeight: 1,
+				peers:      map[string]*scPeer{"P1": {height: 4, state: peerStateReady}},
+				allB:       []int64{1, 2, 3, 4},
+				received:   map[int64]p2p.ID{1: "P1", 2: "P1", 3: "P1", 4: "P1"},
 			},
 			wantHeight: -1,
 		},
 		{
 			name: "only Processed blocks",
 			fields: scTestParams{
-				height: 1,
-				peers:  map[string]*scPeer{"P1": {height: 4, state: peerStateReady}},
-				allB:   []int64{1, 2, 3, 4},
+				initHeight: 1,
+				peers:      map[string]*scPeer{"P1": {height: 4, state: peerStateReady}},
+				allB:       []int64{1, 2, 3, 4},
 			},
 			wantHeight: 1,
 		},
 		{
 			name: "mixed block states",
 			fields: scTestParams{
-				height:      1,
+				initHeight:  1,
 				peers:       map[string]*scPeer{"P1": {height: 4, state: peerStateReady}},
 				allB:        []int64{1, 2, 3, 4},
 				pending:     map[int64]p2p.ID{2: "P1"},
@@ -1574,8 +1577,7 @@ func TestScHandleBlockProcessed(t *testing.T) {
 		{
 			name: "processed block we don't have",
 			fields: scTestParams{
-				initHeight:  5,
-				height:      6,
+				initHeight:  6,
 				peers:       map[string]*scPeer{"P1": {height: 8, state: peerStateReady}},
 				allB:        []int64{6, 7, 8},
 				pending:     map[int64]p2p.ID{6: "P1"},
@@ -1587,8 +1589,7 @@ func TestScHandleBlockProcessed(t *testing.T) {
 		{
 			name: "processed block ok, we processed all blocks",
 			fields: scTestParams{
-				initHeight: 5,
-				height:     6,
+				initHeight: 6,
 				peers:      map[string]*scPeer{"P1": {height: 7, state: peerStateReady}},
 				allB:       []int64{6, 7},
 				received:   map[int64]p2p.ID{6: "P1", 7: "P1"},
@@ -1599,8 +1600,7 @@ func TestScHandleBlockProcessed(t *testing.T) {
 		{
 			name: "processed block ok, we still have blocks to process",
 			fields: scTestParams{
-				initHeight: 5,
-				height:     6,
+				initHeight: 6,
 				peers:      map[string]*scPeer{"P1": {height: 8, state: peerStateReady}},
 				allB:       []int64{6, 7, 8},
 				pending:    map[int64]p2p.ID{7: "P1", 8: "P1"},
@@ -1644,8 +1644,7 @@ func TestScHandleBlockVerificationFailure(t *testing.T) {
 		{
 			name: "failed block we don't have, single peer is still removed",
 			fields: scTestParams{
-				initHeight:  5,
-				height:      6,
+				initHeight:  6,
 				peers:       map[string]*scPeer{"P1": {height: 8, state: peerStateReady}},
 				allB:        []int64{6, 7, 8},
 				pending:     map[int64]p2p.ID{6: "P1"},
@@ -1657,7 +1656,7 @@ func TestScHandleBlockVerificationFailure(t *testing.T) {
 		{
 			name: "failed block we don't have, one of two peers are removed",
 			fields: scTestParams{
-				initHeight:  5,
+				initHeight:  6,
 				peers:       map[string]*scPeer{"P1": {height: 8, state: peerStateReady}, "P2": {height: 8, state: peerStateReady}},
 				allB:        []int64{6, 7, 8},
 				pending:     map[int64]p2p.ID{6: "P1"},
@@ -1669,8 +1668,7 @@ func TestScHandleBlockVerificationFailure(t *testing.T) {
 		{
 			name: "failed block, all blocks are processed after removal",
 			fields: scTestParams{
-				initHeight: 5,
-				height:     6,
+				initHeight: 6,
 				peers:      map[string]*scPeer{"P1": {height: 7, state: peerStateReady}},
 				allB:       []int64{6, 7},
 				received:   map[int64]p2p.ID{6: "P1", 7: "P1"},
@@ -1681,7 +1679,7 @@ func TestScHandleBlockVerificationFailure(t *testing.T) {
 		{
 			name: "failed block, we still have blocks to process",
 			fields: scTestParams{
-				initHeight: 4,
+				initHeight: 5,
 				peers:      map[string]*scPeer{"P1": {height: 8, state: peerStateReady}, "P2": {height: 8, state: peerStateReady}},
 				allB:       []int64{5, 6, 7, 8},
 				pending:    map[int64]p2p.ID{7: "P1", 8: "P1"},
@@ -1693,7 +1691,7 @@ func TestScHandleBlockVerificationFailure(t *testing.T) {
 		{
 			name: "failed block, H+1 and H+2 delivered by different peers, we still have blocks to process",
 			fields: scTestParams{
-				initHeight: 4,
+				initHeight: 5,
 				peers: map[string]*scPeer{
 					"P1": {height: 8, state: peerStateReady},
 					"P2": {height: 8, state: peerStateReady},
@@ -1742,9 +1740,9 @@ func TestScHandleAddNewPeer(t *testing.T) {
 		{
 			name: "add duplicate peer",
 			fields: scTestParams{
-				height: 6,
-				peers:  map[string]*scPeer{"P1": {height: 8, state: peerStateReady}},
-				allB:   []int64{6, 7, 8},
+				initHeight: 6,
+				peers:      map[string]*scPeer{"P1": {height: 8, state: peerStateReady}},
+				allB:       []int64{6, 7, 8},
 			},
 			args:      args{event: addP1},
 			wantEvent: noOpEvent{},
@@ -1752,9 +1750,9 @@ func TestScHandleAddNewPeer(t *testing.T) {
 		{
 			name: "add P1 to non empty scheduler",
 			fields: scTestParams{
-				height: 6,
-				peers:  map[string]*scPeer{"P2": {height: 8, state: peerStateReady}},
-				allB:   []int64{6, 7, 8},
+				initHeight: 6,
+				peers:      map[string]*scPeer{"P2": {height: 8, state: peerStateReady}},
+				allB:       []int64{6, 7, 8},
 			},
 			args:      args{event: addP1},
 			wantEvent: noOpEvent{},