BlockStore holds extended commit

Cherry-pick 8d504d4b50ec6afbdffe2df7ababbef30e15053d and fix conflicts.

Signed-off-by: Thane Thomson <connect@thanethomson.com>

internal/blocksync/pool.go

@@ -208,12 +208,13 @@ func (pool *BlockPool) IsCaughtUp() bool {
 // We need to see the second block's Commit to validate the first block.
 // So we peek two blocks at a time.
 // The caller will verify the commit.
-func (pool *BlockPool) PeekTwoBlocks() (first *types.Block, second *types.Block) {
+func (pool *BlockPool) PeekTwoBlocks() (first *types.Block, second *types.Block, firstExtCommit *types.ExtendedCommit) {
 	pool.mtx.RLock()
 	defer pool.mtx.RUnlock()
 
 	if r := pool.requesters[pool.height]; r != nil {
 		first = r.getBlock()
+		firstExtCommit = r.getExtendedCommit()
 	}
 	if r := pool.requesters[pool.height+1]; r != nil {
 		second = r.getBlock()
@@ -222,7 +223,7 @@ func (pool *BlockPool) PeekTwoBlocks() (first *types.Block, second *types.Block)
 }
 
 // PopRequest pops the first block at pool.height.
-// It must have been validated by 'second'.Commit from PeekTwoBlocks().
+// It must have been validated by 'second'.Commit from PeekTwoBlocks(), TODO: (?) and its corresponding ExtendedCommit.
 func (pool *BlockPool) PopRequest() {
 	pool.mtx.Lock()
 	defer pool.mtx.Unlock()
@@ -268,10 +269,15 @@ func (pool *BlockPool) RedoRequest(height int64) types.NodeID {
 
 // AddBlock validates that the block comes from the peer it was expected from and calls the requester to store it.
 // TODO: ensure that blocks come in order for each peer.
-func (pool *BlockPool) AddBlock(peerID types.NodeID, block *types.Block, blockSize int) {
+func (pool *BlockPool) AddBlock(peerID types.NodeID, block *types.Block, extCommit *types.ExtendedCommit, blockSize int) {
 	pool.mtx.Lock()
 	defer pool.mtx.Unlock()
 
+	if block.Height != extCommit.Height {
+		pool.logger.Error("heights don't match, not adding block", "block_height", block.Height, "commit_height", extCommit.Height)
+		return
+	}
+
 	requester := pool.requesters[block.Height]
 	if requester == nil {
 		pool.logger.Error("peer sent us a block we didn't expect",
@@ -286,7 +292,7 @@ func (pool *BlockPool) AddBlock(peerID types.NodeID, block *types.Block, blockSi
 		return
 	}
 
-	if requester.setBlock(block, peerID) {
+	if requester.setBlock(block, extCommit, peerID) {
 		atomic.AddInt32(&pool.numPending, -1)
 		peer := pool.peers[peerID]
 		if peer != nil {
@@ -460,6 +466,7 @@ func (pool *BlockPool) debug() string {
 		} else {
 			str += fmt.Sprintf("H(%v):", h)
 			str += fmt.Sprintf("B?(%v) ", pool.requesters[h].block != nil)
+			str += fmt.Sprintf("C?(%v) ", pool.requesters[h].extCommit != nil)
 		}
 	}
 	return str
@@ -548,9 +555,10 @@ type bpRequester struct {
 	gotBlockCh chan struct{}
 	redoCh     chan types.NodeID // redo may send multitime, add peerId to identify repeat
 
-	mtx    sync.Mutex
-	peerID types.NodeID
-	block  *types.Block
+	mtx       sync.Mutex
+	peerID    types.NodeID
+	block     *types.Block
+	extCommit *types.ExtendedCommit
 }
 
 func newBPRequester(logger log.Logger, pool *BlockPool, height int64) *bpRequester {
@@ -576,13 +584,14 @@ func (bpr *bpRequester) OnStart(ctx context.Context) error {
 func (*bpRequester) OnStop() {}
 
 // Returns true if the peer matches and block doesn't already exist.
-func (bpr *bpRequester) setBlock(block *types.Block, peerID types.NodeID) bool {
+func (bpr *bpRequester) setBlock(block *types.Block, extCommit *types.ExtendedCommit, peerID types.NodeID) bool {
 	bpr.mtx.Lock()
 	if bpr.block != nil || bpr.peerID != peerID {
 		bpr.mtx.Unlock()
 		return false
 	}
 	bpr.block = block
+	bpr.extCommit = extCommit
 	bpr.mtx.Unlock()
 
 	select {
@@ -598,6 +607,12 @@ func (bpr *bpRequester) getBlock() *types.Block {
 	return bpr.block
 }
 
+func (bpr *bpRequester) getExtendedCommit() *types.ExtendedCommit {
+	bpr.mtx.Lock()
+	defer bpr.mtx.Unlock()
+	return bpr.extCommit
+}
+
 func (bpr *bpRequester) getPeerID() types.NodeID {
 	bpr.mtx.Lock()
 	defer bpr.mtx.Unlock()
@@ -615,6 +630,7 @@ func (bpr *bpRequester) reset() {
 
 	bpr.peerID = ""
 	bpr.block = nil
+	bpr.extCommit = nil
 }
 
 // Tells bpRequester to pick another peer and try again.

internal/blocksync/reactor.go

@@ -76,7 +76,7 @@ type Reactor struct {
 	stateStore sm.Store
 
 	blockExec   *sm.BlockExecutor
-	store       *store.BlockStore
+	store       sm.BlockStore
 	pool        *BlockPool
 	consReactor consensusReactor
 	blockSync   *atomicBool
@@ -186,15 +186,24 @@ func (r *Reactor) OnStop() {
 func (r *Reactor) respondToPeer(ctx context.Context, msg *bcproto.BlockRequest, peerID types.NodeID, blockSyncCh *p2p.Channel) error {
 	block := r.store.LoadBlock(msg.Height)
 	if block != nil {
+		extCommit := r.store.LoadBlockExtCommit(msg.Height)
+		if extCommit == nil {
+			r.logger.Error("peer requesting a block; we have the block but not its extended commit (%v)", block)
+			return fmt.Errorf("blockstore has block but not extended commit %v", block)
+		}
 		blockProto, err := block.ToProto()
 		if err != nil {
-			r.logger.Error("failed to convert msg to protobuf", "err", err)
+			r.logger.Error("failed to convert block to protobuf", "err", err)
 			return err
 		}
+		extCommitProto := extCommit.ToProto()
 
 		return blockSyncCh.Send(ctx, p2p.Envelope{
-			To:      peerID,
-			Message: &bcproto.BlockResponse{Block: blockProto},
+			To: peerID,
+			Message: &bcproto.BlockResponse{
+				Block:     blockProto,
+				ExtCommit: extCommitProto,
+			},
 		})
 	}
 
@@ -236,8 +245,15 @@ func (r *Reactor) handleMessage(ctx context.Context, chID p2p.ChannelID, envelop
 					"err", err)
 				return err
 			}
+			extCommit, err := types.ExtendedCommitFromProto(msg.ExtCommit)
+			if err != nil {
+				r.logger.Error("failed to convert extended commit from proto",
+					"peer", envelope.From,
+					"err", err)
+				return err
+			}
 
-			r.pool.AddBlock(envelope.From, block, block.Size())
+			r.pool.AddBlock(envelope.From, block, extCommit, block.Size())
 
 		case *bcproto.StatusRequest:
 			return blockSyncCh.Send(ctx, p2p.Envelope{
@@ -448,6 +464,10 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 			)
 
 			switch {
+			case r.pool.startHeight > state.InitialHeight && blocksSynced == 0:
+				//If we have state-synced, we need to blocksync at least one block
+				continue
+
 			case r.pool.IsCaughtUp():
 				r.logger.Info("switching to consensus reactor", "height", height)
 
@@ -490,9 +510,12 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 			// TODO: Uncouple from request routine.
 
 			// see if there are any blocks to sync
-			first, second := r.pool.PeekTwoBlocks()
-			if first == nil || second == nil {
-				// we need both to sync the first block
+			first, second, extCommit := r.pool.PeekTwoBlocks()
+			if first == nil || second == nil || extCommit == nil {
+				if first != nil && extCommit == nil {
+					r.logger.Error("peeked a block without extended commit", "height", first.Height)
+				}
+				// we need all to sync the first block
 				continue
 			} else {
 				// try again quickly next loop
@@ -517,6 +540,7 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 			// 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.
+			// TODO Should we also validate against the extended commit?
 			if err = state.Validators.VerifyCommitLight(chainID, firstID, first.Height, second.LastCommit); err != nil {
 				err = fmt.Errorf("invalid last commit: %w", err)
 				r.logger.Error(
@@ -549,7 +573,7 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 				r.pool.PopRequest()
 
 				// TODO: batch saves so we do not persist to disk every block
-				r.store.SaveBlock(first, firstParts, second.LastCommit)
+				r.store.SaveBlock(first, firstParts, extCommit)
 
 				var err error