abci++: Propagate vote extensions (RFC 017)

internal/blocksync/pool.go

@@ -200,16 +200,20 @@ func (pool *BlockPool) IsCaughtUp() bool {
 	return pool.height >= (pool.maxPeerHeight - 1)
 }
 
-// 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.
+// 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. We return an extended commit, containing vote extensions
+// and their associated signatures, as this is critical to consensus in ABCI++
+// as we switch from block sync to consensus mode.
+//
 // The caller will verify the commit.
-func (pool *BlockPool) PeekTwoBlocks() (first *types.Block, second *types.Block) {
+func (pool *BlockPool) PeekTwoBlocks() (first, 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()
@@ -218,7 +222,8 @@ 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 the second Commit from PeekTwoBlocks.
+// TODO(thane): (?) and its corresponding ExtendedCommit.
 func (pool *BlockPool) PopRequest() {
 	pool.mtx.Lock()
 	defer pool.mtx.Unlock()
@@ -262,37 +267,48 @@ func (pool *BlockPool) RedoRequest(height int64) types.NodeID {
 	return peerID
 }
 
-// AddBlock validates that the block comes from the peer it was expected from and calls the requester to store it.
+// AddBlock validates that the block comes from the peer it was expected from
+// and calls the requester to store it.
+//
+// This requires an extended commit at the same height as the supplied block -
+// the block contains the last commit, but we need the latest commit in case we
+// need to switch over from block sync to consensus at this height. If the
+// height of the extended commit and the height of the block do not match, we
+// do not add the block and return an error.
 // 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) error {
 	pool.mtx.Lock()
 	defer pool.mtx.Unlock()
 
+	if block.Height != extCommit.Height {
+		return fmt.Errorf("heights don't match, not adding block (block height: %d, commit height: %d)", block.Height, extCommit.Height)
+	}
+
 	requester := pool.requesters[block.Height]
 	if requester == nil {
-		pool.logger.Error("peer sent us a block we didn't expect",
-			"peer", peerID, "curHeight", pool.height, "blockHeight", block.Height)
 		diff := pool.height - block.Height
 		if diff < 0 {
 			diff *= -1
 		}
 		if diff > maxDiffBetweenCurrentAndReceivedBlockHeight {
 			pool.sendError(errors.New("peer sent us a block we didn't expect with a height too far ahead/behind"), peerID)
 		}
-		return
+		return fmt.Errorf("peer sent us a block we didn't expect (peer: %s, current height: %d, block height: %d)", peerID, pool.height, block.Height)
 	}
 
-	if requester.setBlock(block, peerID) {
+	if requester.setBlock(block, extCommit, peerID) {
 		atomic.AddInt32(&pool.numPending, -1)
 		peer := pool.peers[peerID]
 		if peer != nil {
 			peer.decrPending(blockSize)
 		}
 	} else {
 		err := errors.New("requester is different or block already exists")
-		pool.logger.Error(err.Error(), "peer", peerID, "requester", requester.getPeerID(), "blockHeight", block.Height)
 		pool.sendError(err, peerID)
+		return fmt.Errorf("%w (peer: %s, requester: %s, block height: %d)", err, peerID, requester.getPeerID(), block.Height)
 	}
+
+	return nil
 }
 
 // MaxPeerHeight returns the highest reported height.
@@ -456,6 +472,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
@@ -544,9 +561,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 {
@@ -572,13 +590,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 {
@@ -594,6 +613,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()
@@ -611,6 +636,7 @@ func (bpr *bpRequester) reset() {
 
 	bpr.peerID = ""
 	bpr.block = nil
+	bpr.extCommit = nil
 }
 
 // Tells bpRequester to pick another peer and try again.

internal/blocksync/pool_test.go

@@ -43,7 +43,10 @@ func (p testPeer) runInputRoutine() {
 // Request desired, pretend like we got the block immediately.
 func (p testPeer) simulateInput(input inputData) {
 	block := &types.Block{Header: types.Header{Height: input.request.Height}}
-	input.pool.AddBlock(input.request.PeerID, block, 123)
+	extCommit := &types.ExtendedCommit{
+		Height: input.request.Height,
+	}
+	_ = input.pool.AddBlock(input.request.PeerID, block, extCommit, 123)
 	// TODO: uncommenting this creates a race which is detected by:
 	// https://github.com/golang/go/blob/2bd767b1022dd3254bcec469f0ee164024726486/src/testing/testing.go#L854-L856
 	// see: https://github.com/tendermint/tendermint/issues/3390#issue-418379890
@@ -110,7 +113,7 @@ func TestBlockPoolBasic(t *testing.T) {
 			if !pool.IsRunning() {
 				return
 			}
-			first, second := pool.PeekTwoBlocks()
+			first, second, _ := pool.PeekTwoBlocks()
 			if first != nil && second != nil {
 				pool.PopRequest()
 			} else {
@@ -164,7 +167,7 @@ func TestBlockPoolTimeout(t *testing.T) {
 			if !pool.IsRunning() {
 				return
 			}
-			first, second := pool.PeekTwoBlocks()
+			first, second, _ := pool.PeekTwoBlocks()
 			if first != nil && second != nil {
 				pool.PopRequest()
 			} else {

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,21 @@ 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.LoadBlockExtendedCommit(msg.Height)
+		if extCommit == nil {
+			return fmt.Errorf("found block in store without extended commit: %v", block)
+		}
 		blockProto, err := block.ToProto()
 		if err != nil {
-			r.logger.Error("failed to convert msg to protobuf", "err", err)
-			return err
+			return fmt.Errorf("failed to convert block to protobuf: %w", err)
 		}
 
 		return blockSyncCh.Send(ctx, p2p.Envelope{
-			To:      peerID,
-			Message: &bcproto.BlockResponse{Block: blockProto},
+			To: peerID,
+			Message: &bcproto.BlockResponse{
+				Block:     blockProto,
+				ExtCommit: extCommit.ToProto(),
+			},
 		})
 	}
 
@@ -236,8 +242,17 @@ func (r *Reactor) handleMessage(ctx context.Context, envelope *p2p.Envelope, blo
 					"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())
+			if err := r.pool.AddBlock(envelope.From, block, extCommit, block.Size()); err != nil {
+				r.logger.Error("failed to add block", "err", err)
+			}
 
 		case *bcproto.StatusRequest:
 			return blockSyncCh.Send(ctx, p2p.Envelope{
@@ -448,6 +463,20 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 			)
 
 			switch {
+			// TODO(sergio) Might be needed for implementing the upgrading solution. Remove after that
+			//case state.LastBlockHeight > 0 && r.store.LoadBlockExtCommit(state.LastBlockHeight) == nil:
+			case state.LastBlockHeight > 0 && blocksSynced == 0:
+				// Having state-synced, we need to blocksync at least one block
+				r.logger.Info(
+					"no seen commit yet",
+					"height", height,
+					"last_block_height", state.LastBlockHeight,
+					"initial_height", state.InitialHeight,
+					"max_peer_height", r.pool.MaxPeerHeight(),
+					"timeout_in", syncTimeout-time.Since(lastAdvance),
+				)
+				continue
+
 			case r.pool.IsCaughtUp():
 				r.logger.Info("switching to consensus reactor", "height", height)
 
@@ -490,9 +519,13 @@ 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 {
+					// See https://github.com/tendermint/tendermint/pull/8433#discussion_r866790631
+					panic(fmt.Errorf("peeked first block without extended commit at height %d - possible node store corruption", first.Height))
+				}
+				// we need all to sync the first block
 				continue
 			} else {
 				// try again quickly next loop
@@ -517,6 +550,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(sergio): Should we also validate against the extended commit?
 			err = state.Validators.VerifyCommitLight(chainID, firstID, first.Height, second.LastCommit)
 
 			if err == nil {
@@ -559,7 +593,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)
 
 			// TODO: Same thing for app - but we would need a way to get the hash
 			// without persisting the state.

internal/blocksync/reactor_test.go

@@ -147,39 +147,43 @@ func (rts *reactorTestSuite) addNode(
 		sm.NopMetrics(),
 	)
 
+	var lastExtCommit *types.ExtendedCommit
+
+	// The commit we are building for the current height.
+	seenExtCommit := &types.ExtendedCommit{}
+
 	for blockHeight := int64(1); blockHeight <= maxBlockHeight; blockHeight++ {
-		lastCommit := types.NewCommit(blockHeight-1, 0, types.BlockID{}, nil)
-
-		if blockHeight > 1 {
-			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)
-			lastCommit = types.NewCommit(
-				vote.Height,
-				vote.Round,
-				lastBlockMeta.BlockID,
-				[]types.CommitSig{vote.CommitSig()},
-			)
-		}
+		lastExtCommit = seenExtCommit.Clone()
 
-		thisBlock := sf.MakeBlock(state, blockHeight, lastCommit)
+		thisBlock := sf.MakeBlock(state, blockHeight, lastExtCommit.StripExtensions())
 		thisParts, err := thisBlock.MakePartSet(types.BlockPartSizeBytes)
 		require.NoError(t, err)
 		blockID := types.BlockID{Hash: thisBlock.Hash(), PartSetHeader: thisParts.Header()}
 
+		// Simulate a commit for the current height
+		vote, err := factory.MakeVote(
+			ctx,
+			privVal,
+			thisBlock.Header.ChainID,
+			0,
+			thisBlock.Header.Height,
+			0,
+			2,
+			blockID,
+			time.Now(),
+		)
+		require.NoError(t, err)
+		seenExtCommit = &types.ExtendedCommit{
+			Height:             vote.Height,
+			Round:              vote.Round,
+			BlockID:            blockID,
+			ExtendedSignatures: []types.ExtendedCommitSig{vote.ExtendedCommitSig()},
+		}
+
 		state, err = blockExec.ApplyBlock(ctx, state, blockID, thisBlock)
 		require.NoError(t, err)
 
-		blockStore.SaveBlock(thisBlock, thisParts, lastCommit)
+		blockStore.SaveBlock(thisBlock, thisParts, seenExtCommit)
 	}
 
 	rts.peerChans[nodeID] = make(chan p2p.PeerUpdate)

internal/consensus/byzantine_test.go

@@ -178,22 +178,22 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 	lazyNodeState.decideProposal = func(ctx context.Context, height int64, round int32) {
 		require.NotNil(t, lazyNodeState.privValidator)
 
-		var commit *types.Commit
+		var extCommit *types.ExtendedCommit
 		switch {
 		case lazyNodeState.Height == lazyNodeState.state.InitialHeight:
 			// We're creating a proposal for the first block.
 			// The commit is empty, but not nil.
-			commit = types.NewCommit(0, 0, types.BlockID{}, nil)
+			extCommit = &types.ExtendedCommit{}
 		case lazyNodeState.LastCommit.HasTwoThirdsMajority():
 			// Make the commit from LastCommit
-			commit = lazyNodeState.LastCommit.MakeCommit()
+			extCommit = lazyNodeState.LastCommit.MakeExtendedCommit()
 		default: // This shouldn't happen.
 			lazyNodeState.logger.Error("enterPropose: Cannot propose anything: No commit for the previous block")
 			return
 		}
 
 		// omit the last signature in the commit
-		commit.Signatures[len(commit.Signatures)-1] = types.NewCommitSigAbsent()
+		extCommit.ExtendedSignatures[len(extCommit.ExtendedSignatures)-1] = types.NewExtendedCommitSigAbsent()
 
 		if lazyNodeState.privValidatorPubKey == nil {
 			// If this node is a validator & proposer in the current round, it will
@@ -204,7 +204,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 		proposerAddr := lazyNodeState.privValidatorPubKey.Address()
 
 		block, err := lazyNodeState.blockExec.CreateProposalBlock(
-			ctx, lazyNodeState.Height, lazyNodeState.state, commit, proposerAddr, lazyNodeState.LastCommit.GetVotes())
+			ctx, lazyNodeState.Height, lazyNodeState.state, extCommit, proposerAddr)
 		require.NoError(t, err)
 		blockParts, err := block.MakePartSet(types.BlockPartSizeBytes)
 		require.NoError(t, err)