feat: error if the propagation reactor failed to propose the block (#2611)

Closes #2599

Author: rach-id

consensus/propagation/commitment.go

@@ -24,12 +24,12 @@ const (
 
 // ProposeBlock is called when the consensus routine has created a new proposal,
 // and it needs to be gossiped to the rest of the network.
-func (blockProp *Reactor) ProposeBlock(proposal *types.Proposal, block *types.PartSet, txs []proptypes.TxMetaData) {
+func (blockProp *Reactor) ProposeBlock(proposal *types.Proposal, block *types.PartSet, txs []proptypes.TxMetaData) error {
 	// create the parity data and the compact block
 	parityBlock, lastLen, err := types.Encode(block, types.BlockPartSizeBytes)
 	if err != nil {
 		blockProp.Logger.Error("failed to encode block", "err", err)
-		return
+		return err
 	}
 
 	partHashes := extractHashes(block, parityBlock)
@@ -48,7 +48,7 @@ func (blockProp *Reactor) ProposeBlock(proposal *types.Proposal, block *types.Pa
 	sbz, err := cb.SignBytes()
 	if err != nil {
 		blockProp.Logger.Error("failed to create signature for compact block", "err", err)
-		return
+		return err
 	}
 
 	// sign the hash of the compact block NOTE: p2p message sign bytes are
@@ -62,7 +62,7 @@ func (blockProp *Reactor) ProposeBlock(proposal *types.Proposal, block *types.Pa
 			"chain_id", blockProp.chainID,
 			"unique_id", CompactBlockUID,
 		)
-		return
+		return err
 	}
 
 	cb.Signature = sig
@@ -128,6 +128,7 @@ func (blockProp *Reactor) ProposeBlock(proposal *types.Proposal, block *types.Pa
 
 		schema.WriteBlockPartState(blockProp.traceClient, proposal.Height, proposal.Round, chunks[index].GetTrueIndices(), true, string(peer.peer.ID()), schema.Upload)
 	}
+	return nil
 }
 
 func extractHashes(blocks ...*types.PartSet) [][]byte {

consensus/propagation/commitment_test.go

@@ -35,7 +35,8 @@ func TestPropose(t *testing.T) {
 
 	prop, partSet, _, metaData := createTestProposal(t, sm, 1, 0, 100, 1000)
 
-	reactor1.ProposeBlock(prop, partSet, metaData)
+	err := reactor1.ProposeBlock(prop, partSet, metaData)
+	require.NoError(t, err)
 
 	time.Sleep(200 * time.Millisecond)
 
@@ -102,7 +103,8 @@ func TestPropose_OnlySendParityChunks(t *testing.T) {
 	// 128 mb block
 	prop, partSet, _, metaData := createTestProposal(t, sm, 1, 0, 30, 4_000_000)
 
-	reactor1.ProposeBlock(prop, partSet, metaData)
+	err := reactor1.ProposeBlock(prop, partSet, metaData)
+	require.NoError(t, err)
 
 	time.Sleep(200 * time.Millisecond)
 
@@ -217,7 +219,8 @@ func TestRecoverPartsLocally(t *testing.T) {
 		}
 	}
 
-	blockPropR.ProposeBlock(prop, partSet, metaData)
+	err = blockPropR.ProposeBlock(prop, partSet, metaData)
+	require.NoError(t, err)
 
 	_, actualParts, _ := blockPropR.GetProposal(prop.Height, prop.Round)
 

consensus/propagation/propagator.go

@@ -11,7 +11,7 @@ import (
 // consensus reactor and state.
 type Propagator interface {
 	GetProposal(height int64, round int32) (*types.Proposal, *types.PartSet, bool)
-	ProposeBlock(proposal *types.Proposal, parts *types.PartSet, txs []proptypes.TxMetaData)
+	ProposeBlock(proposal *types.Proposal, parts *types.PartSet, txs []proptypes.TxMetaData) error
 	AddCommitment(height int64, round int32, psh *types.PartSetHeader)
 	Prune(committedHeight int64)
 	SetHeightAndRound(height int64, round int32)
@@ -62,7 +62,8 @@ func (nop *NoOpPropagator) GetProposal(_ int64, _ int32) (*types.Proposal, *type
 	return nil, nil, false
 }
 
-func (nop *NoOpPropagator) ProposeBlock(_ *types.Proposal, _ *types.PartSet, _ []proptypes.TxMetaData) {
+func (nop *NoOpPropagator) ProposeBlock(_ *types.Proposal, _ *types.PartSet, _ []proptypes.TxMetaData) error {
+	return nil
 }
 
 func (nop *NoOpPropagator) AddCommitment(_ int64, _ int32, _ *types.PartSetHeader) {

consensus/propagation/reactor_test.go

@@ -457,7 +457,8 @@ func TestPropagationSmokeTest(t *testing.T) {
 			}
 		}
 
-		reactors[1].ProposeBlock(prop, ps, metaData)
+		err := reactors[1].ProposeBlock(prop, ps, metaData)
+		require.NoError(t, err)
 
 		distributing := true
 		for distributing {

consensus/state.go

@@ -1376,9 +1376,6 @@ func (cs *State) defaultDecideProposal(height int64, round int32) {
 	if err := cs.privValidator.SignProposal(cs.state.ChainID, p); err == nil {
 		proposal.Signature = p.Signature
 
-		// send proposal and block parts on internal msg queue
-		cs.sendInternalMessage(msgInfo{&ProposalMessage{proposal}, ""})
-
 		metaData := make([]proptypes.TxMetaData, len(block.Txs))
 		hashes := block.CachedHashes()
 		for i, pos := range blockParts.TxPos {
@@ -1389,6 +1386,15 @@ func (cs *State) defaultDecideProposal(height int64, round int32) {
 			}
 		}
 
+		err = cs.propagator.ProposeBlock(proposal, blockParts, metaData)
+		if err != nil {
+			cs.Logger.Error("propagation reactor failed to propose the block", "err", err)
+			return
+		}
+
+		// send proposal and block parts on internal msg queue
+		cs.sendInternalMessage(msgInfo{&ProposalMessage{proposal}, ""})
+
 		wg := sync.WaitGroup{}
 		wg.Add(1)
 		go func() {
@@ -1398,7 +1404,6 @@ func (cs *State) defaultDecideProposal(height int64, round int32) {
 				cs.sendInternalMessage(msgInfo{&BlockPartMessage{cs.rs.Height, cs.rs.Round, part}, ""})
 			}
 		}()
-		cs.propagator.ProposeBlock(proposal, blockParts, metaData)
 		wg.Wait()
 
 		cs.Logger.Debug("signed proposal", "height", height, "round", round, "proposal", proposal)