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()
@@ -264,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",
@@ -282,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 {
@@ -456,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
@@ -544,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 {
@@ -572,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 {
@@ -594,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()
@@ -611,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/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)
+	var blockID types.BlockID
+	var extCommitSigs []types.ExtendedCommitSig
+	extCommit := types.NewExtendedCommit(input.request.Height, 0, blockID, extCommitSigs)
+	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,23 @@ 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
 		}
 
 		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 +244,15 @@ 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())
+			r.pool.AddBlock(envelope.From, block, extCommit, block.Size())
 
 		case *bcproto.StatusRequest:
 			return blockSyncCh.Send(ctx, p2p.Envelope{
@@ -448,6 +463,19 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 			)
 
 			switch {
+			//case state.LastBlockHeight > 0 && r.store.LoadBlockExtCommit(state.LastBlockHeight) == nil:
+			case state.LastBlockHeight > 0 && blocksSynced == 0:
+				//If we have 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 +518,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 +548,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 +581,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
 

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.NewExtendedCommit(0, 0, types.BlockID{}, nil)
+
 	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.Copy()
 
-		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.NewExtendedCommit(
+			vote.Height,
+			vote.Round,
+			blockID,
+			[]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 commit *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)
+			commit = types.NewExtendedCommit(0, 0, types.BlockID{}, nil)
 		case lazyNodeState.LastCommit.HasTwoThirdsMajority():
 			// Make the commit from LastCommit
-			commit = lazyNodeState.LastCommit.MakeCommit()
+			commit = 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()
+		commit.ExtendedSignatures[len(commit.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, commit, proposerAddr)
 		require.NoError(t, err)
 		blockParts, err := block.MakePartSet(types.BlockPartSizeBytes)
 		require.NoError(t, err)

internal/consensus/reactor.go

@@ -794,10 +794,10 @@ func (r *Reactor) gossipVotesRoutine(ctx context.Context, ps *PeerState, voteCh
 		// catchup logic -- if peer is lagging by more than 1, send Commit
 		blockStoreBase := r.state.blockStore.Base()
 		if blockStoreBase > 0 && prs.Height != 0 && rs.Height >= prs.Height+2 && prs.Height >= blockStoreBase {
-			// Load the block commit for prs.Height, which contains precommit
+			// Load the block's extended commit for prs.Height, which contains precommit
 			// signatures for prs.Height.
-			if commit := r.state.blockStore.LoadBlockCommit(prs.Height); commit != nil {
-				if ok, err := r.pickSendVote(ctx, ps, commit, voteCh); err != nil {
+			if extCommit := r.state.blockStore.LoadBlockExtCommit(prs.Height); extCommit != nil {
+				if ok, err := r.pickSendVote(ctx, ps, extCommit, voteCh); err != nil {
 					return
 				} else if ok {
 					logger.Debug("picked Catchup commit to send", "height", prs.Height)