Merge remote-tracking branch 'origin/jasmina/4457-blocksync-verification_part1' into jasmina/4457_block_sync_verification

internal/blocksync/pool.go

@@ -28,7 +28,7 @@ eg, L = latency = 0.1s
 */
 
 const (
-	requestIntervalMS         = 2
+	requestInterval           = 2 * time.Millisecond
 	maxTotalRequesters        = 600
 	maxPeerErrBuffer          = 1000
 	maxPendingRequests        = maxTotalRequesters
@@ -121,7 +121,6 @@ func NewBlockPool(
 		peers:             make(map[types.NodeID]*bpPeer),
 		requesters:        make(map[int64]*bpRequester),
 		witnessRequesters: make(map[int64]*witnessRequester),
-		// verificationRequesters: make(map[int64]*bpRequester),
 		height:            start,
 		startHeight:       start,
 		numPending:        0,
@@ -148,27 +147,23 @@ func (*BlockPool) OnStop() {}
 
 // spawns requesters as needed
 func (pool *BlockPool) makeRequestersRoutine(ctx context.Context) {
-	for {
-		if !pool.IsRunning() {
-			break
+	for pool.IsRunning() {
+		if ctx.Err() != nil {
+			return
 		}
 
 		_, numPending, lenRequesters := pool.GetStatus()
-		switch {
-		case numPending >= maxPendingRequests:
-			// sleep for a bit.
-			time.Sleep(requestIntervalMS * time.Millisecond)
-			// check for timed out peers
+		if numPending >= maxPendingRequests || lenRequesters >= maxTotalRequesters {
+			// This is preferable to using a timer because the request interval
+			// is so small. Larger request intervals may necessitate using a
+			// timer/ticker.
+			time.Sleep(requestInterval)
 			pool.removeTimedoutPeers()
-		case lenRequesters >= maxTotalRequesters:
-			// sleep for a bit.
-			time.Sleep(requestIntervalMS * time.Millisecond)
-			// check for timed out peers
-			pool.removeTimedoutPeers()
-		default:
-			// request for more blocks.
-			pool.makeNextRequester(ctx)
+			continue
 		}
+
+		// request for more blocks.
+		pool.makeNextRequester(ctx)
 	}
 }
 
@@ -222,16 +217,6 @@ func (pool *BlockPool) IsCaughtUp() bool {
 	return pool.height >= (pool.maxPeerHeight - 1)
 }
 
-func (pool *BlockPool) PeekBlock() (first *types.Block) {
-	pool.mtx.RLock()
-	defer pool.mtx.RUnlock()
-
-	if r := pool.requesters[pool.height]; r != nil {
-		first = r.getBlock()
-	}
-	return
-}
-
 // PeekTwoBlocks returns blocks at pool.height and pool.height+1.
 // We need to see the second block's Commit to validate the first block.
 // So we peek two blocks at a time.
@@ -277,10 +262,6 @@ func (pool *BlockPool) PopRequest() {
 	} else {
 		panic(fmt.Sprintf("Expected requester to pop, got nothing at height %v", pool.height))
 	}
-	// if r := pool.verificationRequesters[pool.height]; r != nil {
-	// 	r.Stop()
-	// 	delete(pool.verificationRequesters, pool.height)
-	// }
 }
 
 // RedoRequest invalidates the block at pool.height,
@@ -856,9 +837,16 @@ OUTER_LOOP:
 			if !bpr.IsRunning() || !bpr.pool.IsRunning() {
 				return
 			}
+			if ctx.Err() != nil {
+				return
+			}
+
 			peer = bpr.pool.pickIncrAvailablePeer(bpr.height)
 			if peer == nil {
-				time.Sleep(requestIntervalMS * time.Millisecond)
+				// This is preferable to using a timer because the request
+				// interval is so small. Larger request intervals may
+				// necessitate using a timer/ticker.
+				time.Sleep(requestInterval)
 				continue PICK_PEER_LOOP
 			}
 			break PICK_PEER_LOOP

internal/blocksync/pool_test.go

@@ -111,8 +111,8 @@ func TestBlockPoolBasic(t *testing.T) {
 			if !pool.IsRunning() {
 				return
 			}
-			first := pool.PeekBlock()
-			if first != nil {
+			first, second := pool.PeekTwoBlocks()
+			if first != nil && second != nil {
 				pool.PopRequest()
 			} else {
 				time.Sleep(1 * time.Second)
@@ -166,8 +166,8 @@ func TestBlockPoolTimeout(t *testing.T) {
 			if !pool.IsRunning() {
 				return
 			}
-			first := pool.PeekBlock()
-			if first != nil {
+			first, second := pool.PeekTwoBlocks()
+			if first != nil && second != nil {
 				pool.PopRequest()
 			} else {
 				time.Sleep(1 * time.Second)

internal/blocksync/reactor.go

@@ -214,13 +214,10 @@ func (r *Reactor) sendHeaderToPeer(ctx context.Context, msg *bcproto.HeaderReque
 func (r *Reactor) respondToPeer(ctx context.Context, msg *bcproto.BlockRequest, peerID types.NodeID, blockSyncCh *p2p.Channel) error {
 	block := r.store.LoadBlockProto(msg.Height)
 	if block != nil {
-		blockCommit := r.store.LoadBlockCommitProto(msg.Height)
-		if blockCommit != nil {
-			return blockSyncCh.Send(ctx, p2p.Envelope{
-				To:      peerID,
-				Message: &bcproto.BlockResponse{Block: block},
-			})
-		}
+		return blockSyncCh.Send(ctx, p2p.Envelope{
+			To:      peerID,
+			Message: &bcproto.BlockResponse{Block: block},
+		})
 	}
 
 	r.logger.Info("peer requesting a block we do not have", "peer", peerID, "height", msg.Height)
@@ -234,7 +231,7 @@ func (r *Reactor) respondToPeer(ctx context.Context, msg *bcproto.BlockRequest,
 // handleMessage handles an Envelope sent from a peer on a specific p2p Channel.
 // It will handle errors and any possible panics gracefully. A caller can handle
 // any error returned by sending a PeerError on the respective channel.
-func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelope *p2p.Envelope, blockSyncCh *p2p.Channel) (err error) {
+func (r *Reactor) handleMessage(ctx context.Context, envelope *p2p.Envelope, blockSyncCh *p2p.Channel) (err error) {
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
@@ -248,7 +245,7 @@ func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelop
 
 	r.logger.Debug("received message", "message", envelope.Message, "peer", envelope.From)
 
-	switch chID {
+	switch envelope.ChannelID {
 	case BlockSyncChannel:
 		switch msg := envelope.Message.(type) {
 		case *bcproto.HeaderRequest:
@@ -292,7 +289,7 @@ func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelop
 		}
 
 	default:
-		err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
+		err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", envelope.ChannelID, envelope)
 	}
 
 	return err
@@ -307,12 +304,12 @@ func (r *Reactor) processBlockSyncCh(ctx context.Context, blockSyncCh *p2p.Chann
 	iter := blockSyncCh.Receive(ctx)
 	for iter.Next(ctx) {
 		envelope := iter.Envelope()
-		if err := r.handleMessage(ctx, blockSyncCh.ID, envelope, blockSyncCh); err != nil {
+		if err := r.handleMessage(ctx, envelope, blockSyncCh); err != nil {
 			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 				return
 			}
 
-			r.logger.Error("failed to process message", "ch_id", blockSyncCh.ID, "envelope", envelope, "err", err)
+			r.logger.Error("failed to process message", "ch_id", envelope.ChannelID, "envelope", envelope, "err", err)
 			if serr := blockSyncCh.SendError(ctx, p2p.PeerError{
 				NodeID: envelope.From,
 				Err:    err,
@@ -554,7 +551,7 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 
 			newBlockParts, err2 := newBlock.MakePartSet(types.BlockPartSizeBytes)
 			if err2 != nil {
-				r.logger.Error("failed to make ",
+				r.logger.Error("failed to make block at ",
 					"height", newBlock.Height,
 					"err", err2.Error())
 				return
@@ -564,11 +561,21 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 				newBlockPartSetHeader = newBlockParts.Header()
 				newBlockID            = types.BlockID{Hash: newBlock.Hash(), PartSetHeader: newBlockPartSetHeader}
 			)
-
+			// Finally, verify the first block using the second's commit.
+			//
+			// NOTE: We can probably make this more efficient, but note that calling
+			// first.Hash() doesn't verify the tx contents, so MakePartSet() is
+			// currently necessary.
 			if r.lastTrustedBlock != nil {
 				err := VerifyNextBlock(newBlock, newBlockID, verifyBlock, r.lastTrustedBlock.block, r.lastTrustedBlock.commit, state.NextValidators)
 
 				if err != nil {
+					r.logger.Error(
+						err.Error(),
+						"last_commit", verifyBlock.LastCommit,
+						"block_id", newBlock,
+						"height", newBlock.Height,
+					)
 					switch err.(type) {
 					case ErrBlockIDDiff:
 					case ErrValidationFailed:
@@ -579,12 +586,13 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 						}); serr != nil {
 							return
 						}
+
 					case ErrInvalidVerifyBlock:
 						r.logger.Error(
 							err.Error(),
 							"last_commit", verifyBlock.LastCommit,
-							"block_id", newBlockID,
-							"height", r.lastTrustedBlock.block.Height,
+							"verify_block_id", newBlockID,
+							"verify_block_height", newBlock.Height,
 						)
 						peerID := r.pool.RedoRequest(r.lastTrustedBlock.block.Height + 2)
 						if serr := blockSyncCh.SendError(ctx, p2p.PeerError{
@@ -593,10 +601,19 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 						}); serr != nil {
 							return
 						}
-					}
 
+					}
+					continue // was return previously
 				}
 
+				// // Finally, verify the first block using the second's commit.
+				// //
+				// // NOTE: We can probably make this more efficient, but note that calling
+				// // first.Hash() doesn't verify the tx contents, so MakePartSet() is
+				// // currently necessary.
+				// err = state.Validators.VerifyCommitLight(chainID, firstID, first.Height, second.LastCommit)
+				r.lastTrustedBlock.block = newBlock
+				r.lastTrustedBlock.commit = verifyBlock.LastCommit
 			} else {
 				// we need to load the last block we trusted
 				if r.initialState.LastBlockHeight != 0 {
@@ -607,36 +624,55 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 				}
 				oldHash := r.initialState.Validators.Hash()
 				if !bytes.Equal(oldHash, newBlock.ValidatorsHash) {
-
-					fmt.Println(
-
+					r.logger.Error("The validator set provided by the new block does not match the expected validator set",
 						"initial hash ", r.initialState.Validators.Hash(),
 						"new hash ", newBlock.ValidatorsHash,
 					)
-					return
+
+					peerID := r.pool.RedoRequest(r.lastTrustedBlock.block.Height + 1)
+					if serr := blockSyncCh.SendError(ctx, p2p.PeerError{
+						NodeID: peerID,
+						Err:    ErrValidationFailed{},
+					}); serr != nil {
+						return
+					}
+					continue // was return previously
 				}
 
 			}
 			if err := r.verifyWithWitnesses(newBlock); err != nil {
 				r.logger.Debug("Witness verificatio nfailed")
 			}
-			if r.lastTrustedBlock == nil {
-				r.lastTrustedBlock = &BlockResponse{block: newBlock, commit: verifyBlock.LastCommit}
-			} else {
-				r.lastTrustedBlock.block = newBlock
-				r.lastTrustedBlock.commit = verifyBlock.LastCommit
-			}
+			var err error
+			// validate the block before we persist it
+			err = r.blockExec.ValidateBlock(ctx, state, newBlock)
+			if err != nil {
+				r.logger.Error("The validator set provided by the new block does not match the expected validator set",
+					"initial hash ", r.initialState.Validators.Hash(),
+					"new hash ", newBlock.ValidatorsHash,
+				)
 
+				peerID := r.pool.RedoRequest(r.lastTrustedBlock.block.Height + 1)
+				if serr := blockSyncCh.SendError(ctx, p2p.PeerError{
+					NodeID: peerID,
+					Err:    ErrValidationFailed{},
+				}); serr != nil {
+					return
+				}
+				continue // was return previously
+			}
 			r.pool.PopRequest()
 
 			// TODO: batch saves so we do not persist to disk every block
 			r.store.SaveBlock(newBlock, newBlockParts, verifyBlock.LastCommit)
 
-			var err error
-
 			// TODO: Same thing for app - but we would need a way to get the hash
 			// without persisting the state.
 			state, err = r.blockExec.ApplyBlock(ctx, state, newBlockID, newBlock)
+			if err != nil {
+				panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", newBlock.Height, newBlock.Hash(), err))
+			}
+
 			if err != nil {
 				// TODO: This is bad, are we zombie?
 				panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", newBlock.Height, newBlock.Hash(), err))

internal/blocksync/reactor_test.go

@@ -48,7 +48,7 @@ func setup(
 	ctx context.Context,
 	t *testing.T,
 	genDoc *types.GenesisDoc,
-	privVal types.PrivValidator,
+	privValArray []types.PrivValidator,
 	maxBlockHeights []int64,
 ) *reactorTestSuite {
 	t.Helper()
@@ -75,10 +75,14 @@ func setup(
 	chDesc := &p2p.ChannelDescriptor{ID: BlockSyncChannel, MessageType: new(bcproto.Message)}
 	rts.blockSyncChannels = rts.network.MakeChannelsNoCleanup(ctx, t, chDesc)
 
-	i := 0
-	for nodeID := range rts.network.Nodes {
-		rts.addNode(ctx, t, nodeID, genDoc, privVal, maxBlockHeights[i])
-		i++
+	if maxBlockHeights[1] != 0 {
+		rts.addMultipleNodes(ctx, t, rts.network.NodeIDs(), genDoc, privValArray, maxBlockHeights, 0)
+	} else {
+		i := 0
+		for nodeID := range rts.network.Nodes {
+			rts.addNode(ctx, t, nodeID, genDoc, privValArray, maxBlockHeights[i])
+			i++
+		}
 	}
 
 	t.Cleanup(func() {
@@ -97,12 +101,151 @@ func setup(
 	return rts
 }
 
+// We add multiple nodes with varying initial heights
+// Allows us to test whether block sync works when a node
+// has previous state
+// maxBlockHeightPerNode - the heights for which the node already has state
+// maxBlockHeightIdx - the index of the node with maximum height
+func (rts *reactorTestSuite) addMultipleNodes(
+	ctx context.Context,
+	t *testing.T,
+	nodeIDs []types.NodeID,
+	genDoc *types.GenesisDoc,
+	privValArray []types.PrivValidator,
+	maxBlockHeightPerNode []int64,
+	maxBlockHeightIdx int64,
+
+) {
+	t.Helper()
+
+	logger := log.NewNopLogger()
+	blockDB := make([]*dbm.MemDB, len(nodeIDs))
+	stateDB := make([]*dbm.MemDB, len(nodeIDs))
+	blockExecutors := make([]*sm.BlockExecutor, len(nodeIDs))
+	blockStores := make([]*store.BlockStore, len(nodeIDs))
+	stateStores := make([]sm.Store, len(nodeIDs))
+
+	state, err := sm.MakeGenesisState(genDoc)
+	require.NoError(t, err)
+
+	for idx, nodeID := range nodeIDs {
+		rts.nodes = append(rts.nodes, nodeID)
+		rts.app[nodeID] = proxy.New(abciclient.NewLocalClient(logger, &abci.BaseApplication{}), logger, proxy.NopMetrics())
+		require.NoError(t, rts.app[nodeID].Start(ctx))
+		stateDB[idx] = dbm.NewMemDB()
+		stateStores[idx] = sm.NewStore(stateDB[idx])
+
+		blockDB[idx] = dbm.NewMemDB()
+		blockStores[idx] = store.NewBlockStore(blockDB[idx])
+
+		require.NoError(t, stateStores[idx].Save(state))
+		mp := &mpmocks.Mempool{}
+		mp.On("Lock").Return()
+		mp.On("Unlock").Return()
+		mp.On("FlushAppConn", mock.Anything).Return(nil)
+		mp.On("Update",
+			mock.Anything,
+			mock.Anything,
+			mock.Anything,
+			mock.Anything,
+			mock.Anything,
+			mock.Anything).Return(nil)
+
+		eventbus := eventbus.NewDefault(logger)
+		require.NoError(t, eventbus.Start(ctx))
+
+		blockExecutors[idx] = sm.NewBlockExecutor(stateStores[idx],
+			log.NewNopLogger(),
+			rts.app[nodeID],
+			mp,
+			sm.EmptyEvidencePool{},
+			blockStores[idx],
+			eventbus,
+			sm.NopMetrics(),
+		)
+	}
+
+	for blockHeight := int64(1); blockHeight <= maxBlockHeightPerNode[maxBlockHeightIdx]; blockHeight++ {
+		lastCommit := types.NewCommit(blockHeight-1, 0, types.BlockID{}, nil)
+
+		if blockHeight > 1 {
+			lastBlockMeta := blockStores[maxBlockHeightIdx].LoadBlockMeta(blockHeight - 1)
+			lastBlock := blockStores[maxBlockHeightIdx].LoadBlock(blockHeight - 1)
+
+			commitSigs := make([]types.CommitSig, len(privValArray))
+			votes := make([]types.Vote, len(privValArray))
+			for i, val := range privValArray {
+
+				vote, err := factory.MakeVote(
+					ctx,
+					val,
+					lastBlock.Header.ChainID, 0,
+					lastBlock.Header.Height, 0, 2,
+					lastBlockMeta.BlockID,
+					time.Now(),
+				)
+				require.NoError(t, err)
+				votes[i] = *vote
+				commitSigs[i] = vote.CommitSig()
+
+			}
+			lastCommit = types.NewCommit(
+				votes[0].Height,
+				votes[0].Round,
+				lastBlockMeta.BlockID,
+				commitSigs,
+			)
+
+		}
+		thisBlock := sf.MakeBlock(state, blockHeight, lastCommit)
+		thisParts, err := thisBlock.MakePartSet(types.BlockPartSizeBytes)
+		require.NoError(t, err)
+		blockID := types.BlockID{Hash: thisBlock.Hash(), PartSetHeader: thisParts.Header()}
+
+		for idx := range nodeIDs {
+
+			if blockHeight <= maxBlockHeightPerNode[idx] {
+				lastState, err := stateStores[idx].Load()
+				require.NoError(t, err)
+				state, err = blockExecutors[idx].ApplyBlock(ctx, lastState, blockID, thisBlock)
+				require.NoError(t, err)
+				blockStores[idx].SaveBlock(thisBlock, thisParts, lastCommit)
+			}
+		}
+	}
+
+	for idx, nodeID := range nodeIDs {
+		rts.peerChans[nodeID] = make(chan p2p.PeerUpdate)
+		rts.peerUpdates[nodeID] = p2p.NewPeerUpdates(rts.peerChans[nodeID], 1)
+		rts.network.Nodes[nodeID].PeerManager.Register(ctx, rts.peerUpdates[nodeID])
+
+		chCreator := func(ctx context.Context, chdesc *p2p.ChannelDescriptor) (*p2p.Channel, error) {
+			return rts.blockSyncChannels[nodeID], nil
+		}
+		rts.reactors[nodeID] = NewReactor(
+			rts.logger.With("nodeID", nodeID),
+			stateStores[idx],
+			blockExecutors[idx],
+			blockStores[idx],
+			nil,
+			chCreator,
+			func(ctx context.Context) *p2p.PeerUpdates { return rts.peerUpdates[nodeID] },
+			rts.blockSync,
+			consensus.NopMetrics(),
+			nil, // eventbus, can be nil
+		)
+
+		require.NoError(t, rts.reactors[nodeID].Start(ctx))
+		require.True(t, rts.reactors[nodeID].IsRunning())
+	}
+}
+
 func (rts *reactorTestSuite) addNode(
 	ctx context.Context,
 	t *testing.T,
 	nodeID types.NodeID,
 	genDoc *types.GenesisDoc,
-	privVal types.PrivValidator,
+	privValArray []types.PrivValidator,
 	maxBlockHeight int64,
 ) {
 	t.Helper()
@@ -154,21 +297,28 @@ func (rts *reactorTestSuite) addNode(
 			lastBlockMeta := blockStore.LoadBlockMeta(blockHeight - 1)
 			lastBlock := blockStore.LoadBlock(blockHeight - 1)
 
-			vote, err := factory.MakeVote(
-				ctx,
-				privVal,
-				lastBlock.Header.ChainID, 0,
-				lastBlock.Header.Height, 0, 2,
-				lastBlockMeta.BlockID,
-				time.Now(),
-			)
-			require.NoError(t, err)
+			commitSigs := make([]types.CommitSig, len(privValArray))
+			votes := make([]types.Vote, len(privValArray))
+			for i, val := range privValArray {
+				vote, err := factory.MakeVote(
+					ctx,
+					val,
+					lastBlock.Header.ChainID, 0,
+					lastBlock.Header.Height, 0, 2,
+					lastBlockMeta.BlockID,
+					time.Now(),
+				)
+				require.NoError(t, err)
+				votes[i] = *vote
+				commitSigs[i] = vote.CommitSig()
+			}
 			lastCommit = types.NewCommit(
-				vote.Height,
-				vote.Round,
+				votes[0].Height,
+				votes[0].Round,
 				lastBlockMeta.BlockID,
-				[]types.CommitSig{vote.CommitSig()},
+				commitSigs,
 			)
+
 		}
 
 		thisBlock := sf.MakeBlock(state, blockHeight, lastCommit)
@@ -227,7 +377,7 @@ func TestReactor_AbruptDisconnect(t *testing.T) {
 	genDoc := factory.GenesisDoc(cfg, time.Now(), valSet.Validators, factory.ConsensusParams())
 	maxBlockHeight := int64(64)
 
-	rts := setup(ctx, t, genDoc, privVals[0], []int64{maxBlockHeight, 0})
+	rts := setup(ctx, t, genDoc, privVals, []int64{maxBlockHeight, 0})
 
 	require.Equal(t, maxBlockHeight, rts.reactors[rts.nodes[0]].store.Height())
 
@@ -255,6 +405,51 @@ func TestReactor_AbruptDisconnect(t *testing.T) {
 	rts.network.Nodes[rts.nodes[1]].PeerManager.Disconnected(ctx, rts.nodes[0])
 }
 
+//@jmalicevic ToDO
+// a) Add tests that verify whether faulty peer is properly detected
+// 1. block at H + 1 is faulty
+// 2. block at H + 2 is faulty (the validator set does not match)
+// b) Add test to verify we replace a peer with a new one if we detect misbehavior
+func TestReactor_NonGenesisSync(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	cfg, err := config.ResetTestRoot(t.TempDir(), "block_sync_reactor_test")
+	require.NoError(t, err)
+	defer os.RemoveAll(cfg.RootDir)
+
+	valSet, privVals := factory.ValidatorSet(ctx, t, 4, 30)
+	genDoc := factory.GenesisDoc(cfg, time.Now(), valSet.Validators, factory.ConsensusParams())
+	maxBlockHeight := int64(101)
+
+	rts := setup(ctx, t, genDoc, privVals, []int64{maxBlockHeight, 2, 0}) //50, 4, 0})
+	require.Equal(t, maxBlockHeight, rts.reactors[rts.nodes[0]].store.Height())
+	rts.start(ctx, t)
+
+	require.Eventually(
+		t,
+		func() bool {
+			matching := true
+			for idx := range rts.nodes {
+				if idx == 0 {
+					continue
+				}
+				matching = matching && rts.reactors[rts.nodes[idx]].GetRemainingSyncTime() > time.Nanosecond &&
+					rts.reactors[rts.nodes[idx]].pool.getLastSyncRate() > 0.001
+
+				if !matching {
+					height, _, _ := rts.reactors[rts.nodes[idx]].pool.GetStatus()
+					t.Logf("%d %d %s %f", height, idx, rts.reactors[rts.nodes[idx]].GetRemainingSyncTime(), rts.reactors[rts.nodes[idx]].pool.getLastSyncRate())
+				}
+			}
+			return matching
+		},
+		10*time.Second,
+		10*time.Millisecond,
+		"expected node to be partially synced",
+	)
+}
+
 func TestReactor_SyncTime(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -263,18 +458,17 @@ func TestReactor_SyncTime(t *testing.T) {
 	require.NoError(t, err)
 	defer os.RemoveAll(cfg.RootDir)
 
-	valSet, privVals := factory.ValidatorSet(ctx, t, 1, 30)
+	valSet, privVals := factory.ValidatorSet(ctx, t, 4, 30)
 	genDoc := factory.GenesisDoc(cfg, time.Now(), valSet.Validators, factory.ConsensusParams())
 	maxBlockHeight := int64(101)
 
-	rts := setup(ctx, t, genDoc, privVals[0], []int64{maxBlockHeight, 0})
+	rts := setup(ctx, t, genDoc, privVals, []int64{maxBlockHeight, 0})
 	require.Equal(t, maxBlockHeight, rts.reactors[rts.nodes[0]].store.Height())
 	rts.start(ctx, t)
 
 	require.Eventually(
 		t,
 		func() bool {
-			//t.Logf("%d %d %s", rts.reactors[rts.nodes[1]].pool.height, rts.reactors[rts.nodes[0]].pool.height, rts.reactors[rts.nodes[1]].GetRemainingSyncTime())
 			return rts.reactors[rts.nodes[1]].GetRemainingSyncTime() > time.Nanosecond &&
 				rts.reactors[rts.nodes[1]].pool.getLastSyncRate() > 0.001
 		},
@@ -296,7 +490,7 @@ func TestReactor_NoBlockResponse(t *testing.T) {
 	genDoc := factory.GenesisDoc(cfg, time.Now(), valSet.Validators, factory.ConsensusParams())
 	maxBlockHeight := int64(65)
 
-	rts := setup(ctx, t, genDoc, privVals[0], []int64{maxBlockHeight, 0})
+	rts := setup(ctx, t, genDoc, privVals, []int64{maxBlockHeight, 0})
 
 	require.Equal(t, maxBlockHeight, rts.reactors[rts.nodes[0]].store.Height())
 
@@ -315,10 +509,7 @@ func TestReactor_NoBlockResponse(t *testing.T) {
 	secondaryPool := rts.reactors[rts.nodes[1]].pool
 	require.Eventually(
 		t,
-		func() bool {
-			t.Logf("%d %d", secondaryPool.MaxPeerHeight(), secondaryPool.height)
-			return secondaryPool.MaxPeerHeight() > 0 && secondaryPool.IsCaughtUp()
-		},
+		func() bool { return secondaryPool.MaxPeerHeight() > 0 && secondaryPool.IsCaughtUp() },
 		10*time.Second,
 		10*time.Millisecond,
 		"expected node to be fully synced",
@@ -351,7 +542,7 @@ func TestReactor_BadBlockStopsPeer(t *testing.T) {
 	valSet, privVals := factory.ValidatorSet(ctx, t, 1, 30)
 	genDoc := factory.GenesisDoc(cfg, time.Now(), valSet.Validators, factory.ConsensusParams())
 
-	rts := setup(ctx, t, genDoc, privVals[0], []int64{maxBlockHeight, 0, 0, 0, 0})
+	rts := setup(ctx, t, genDoc, privVals, []int64{maxBlockHeight, 0, 0, 0, 0})
 
 	require.Equal(t, maxBlockHeight, rts.reactors[rts.nodes[0]].store.Height())
 
@@ -389,7 +580,7 @@ func TestReactor_BadBlockStopsPeer(t *testing.T) {
 		MaxPeers:     uint16(len(rts.nodes) + 1),
 		MaxConnected: uint16(len(rts.nodes) + 1),
 	})
-	rts.addNode(ctx, t, newNode.NodeID, otherGenDoc, otherPrivVals[0], maxBlockHeight)
+	rts.addNode(ctx, t, newNode.NodeID, otherGenDoc, otherPrivVals, maxBlockHeight)
 
 	// add a fake peer just so we do not wait for the consensus ticker to timeout
 	rts.reactors[newNode.NodeID].pool.SetPeerRange("00ff", 10, 10)

internal/blocksync/verify.go

@@ -82,7 +82,8 @@ func (e ErrValidationFailed) Error() string {
 	return "failed to verify next block"
 }
 
-func VerifyNextBlock(newBlock *types.Block, newBlockID types.BlockID, verifyBlock *types.Block, trustedBlock *types.Block, trustedCommit *types.Commit, validators *types.ValidatorSet) error {
+func VerifyNextBlock(newBlock *types.Block, newBlockID types.BlockID, verifyBlock *types.Block, trustedBlock *types.Block,
+	trustedCommit *types.Commit, validators *types.ValidatorSet) error {
 
 	// If the blockID in LastCommit of NewBlock does not match the trusted block
 	// we can assume NewBlock is not correct
@@ -99,7 +100,8 @@ func VerifyNextBlock(newBlock *types.Block, newBlockID types.BlockID, verifyBloc
 	// Verify NewBlock usign the validator set obtained after applying the last block
 	// Note: VerifyAdjacent in the LightClient relies on a trusting period which is not applicable here
 	// ToDo: We need witness verification here as well and backwards verification from a state where we can trust validators
-	if err := VerifyAdjacent(&types.SignedHeader{Header: &trustedBlock.Header, Commit: trustedCommit}, &types.SignedHeader{Header: &newBlock.Header, Commit: verifyBlock.LastCommit}, validators); err != nil {
+	if err := VerifyAdjacent(&types.SignedHeader{Header: &trustedBlock.Header, Commit: trustedCommit},
+		&types.SignedHeader{Header: &newBlock.Header, Commit: verifyBlock.LastCommit}, validators); err != nil {
 		return ErrValidationFailed{Reason: err}
 	}