[R4R] Add proposer to NewRound event and proposal info to CompleteProposal event

consensus/common_test.go

@@ -416,8 +416,24 @@ func ensureNewTimeout(timeoutCh <-chan interface{}, height int64, round int, tim
 }
 
 func ensureNewProposal(proposalCh <-chan interface{}, height int64, round int) {
-	ensureNewEvent(proposalCh, height, round, ensureTimeout,
-		"Timeout expired while waiting for NewProposal event")
+	select {
+	case <-time.After(ensureTimeout):
+		panic("Timeout expired while waiting for NewProposal event")
+	case ev := <-proposalCh:
+		rs, ok := ev.(types.EventDataCompleteProposal)
+		if !ok {
+			panic(
+				fmt.Sprintf(
+					"expected a EventDataCompleteProposal, got %v.Wrong subscription channel?",
+					reflect.TypeOf(rs)))
+		}
+		if rs.Height != height {
+			panic(fmt.Sprintf("expected height %v, got %v", height, rs.Height))
+		}
+		if rs.Round != round {
+			panic(fmt.Sprintf("expected round %v, got %v", round, rs.Round))
+		}
+	}
 }
 
 func ensureNewValidBlock(validBlockCh <-chan interface{}, height int64, round int) {

consensus/state.go

@@ -1408,7 +1408,7 @@ func (cs *ConsensusState) defaultSetProposal(proposal *types.Proposal) error {
 		return nil
 	}
 
-  // Verify POLRound, which must be -1 or in range [0, proposal.Round).
+	// Verify POLRound, which must be -1 or in range [0, proposal.Round).
 	if proposal.POLRound < -1 ||
 		(proposal.POLRound >= 0 && proposal.POLRound >= proposal.Round) {
 		return ErrInvalidProposalPOLRound
@@ -1466,7 +1466,7 @@ func (cs *ConsensusState) addProposalBlockPart(msg *BlockPartMessage, peerID p2p
 		}
 		// NOTE: it's possible to receive complete proposal blocks for future rounds without having the proposal
 		cs.Logger.Info("Received complete proposal block", "height", cs.ProposalBlock.Height, "hash", cs.ProposalBlock.Hash())
-		cs.eventBus.PublishEventCompleteProposal(cs.RoundStateEvent())
+		cs.eventBus.PublishEventCompleteProposal(cs.CompleteProposalEvent())
 
 		// Update Valid* if we can.
 		prevotes := cs.Votes.Prevotes(cs.Round)

consensus/types/round_state.go

@@ -112,6 +112,28 @@ func (rs *RoundState) RoundStateSimple() RoundStateSimple {
 	}
 }
 
+// CompleteProposalEvent returns information about a proposed block as an event.
+func (rs *RoundState) CompleteProposalEvent() types.EventDataCompleteProposal {
+	addr := rs.Validators.GetProposer().Address
+	idx, _ := rs.Validators.GetByAddress(addr)
+
+	// We must construct BlockID from ProposalBlock and ProposalBlockParts
+	// cs.Proposal is not guaranteed to be set when this function is called
+	blockId := types.BlockID{
+		Hash:        rs.ProposalBlock.Hash(),
+		PartsHeader: rs.ProposalBlockParts.Header(),
+	}
+
+	edcp := types.EventDataCompleteProposal{
+		Height:          rs.Height,
+		Round:           rs.Round,
+		ProposerAddress: addr,
+		ProposerIndex:   idx,
+		BlockID:         blockId,
+	}
+	return edcp
+}
+
 // RoundStateEvent returns the H/R/S of the RoundState as an event.
 func (rs *RoundState) RoundStateEvent() types.EventDataRoundState {
 	// copy the RoundState.

types/event_bus.go

@@ -140,7 +140,7 @@ func (b *EventBus) PublishEventNewRound(data EventDataRoundState) error {
 	return b.Publish(EventNewRound, data)
 }
 
-func (b *EventBus) PublishEventCompleteProposal(data EventDataRoundState) error {
+func (b *EventBus) PublishEventCompleteProposal(data EventDataCompleteProposal) error {
 	return b.Publish(EventCompleteProposal, data)
 }
 

types/event_bus_test.go

@@ -98,7 +98,7 @@ func TestEventBusPublish(t *testing.T) {
 	require.NoError(t, err)
 	err = eventBus.PublishEventNewRound(EventDataRoundState{})
 	require.NoError(t, err)
-	err = eventBus.PublishEventCompleteProposal(EventDataRoundState{})
+	err = eventBus.PublishEventCompleteProposal(EventDataCompleteProposal{})
 	require.NoError(t, err)
 	err = eventBus.PublishEventPolka(EventDataRoundState{})
 	require.NoError(t, err)

types/events.go

@@ -43,6 +43,7 @@ func RegisterEventDatas(cdc *amino.Codec) {
 	cdc.RegisterConcrete(EventDataNewBlockHeader{}, "tendermint/event/NewBlockHeader", nil)
 	cdc.RegisterConcrete(EventDataTx{}, "tendermint/event/Tx", nil)
 	cdc.RegisterConcrete(EventDataRoundState{}, "tendermint/event/RoundState", nil)
+	cdc.RegisterConcrete(EventDataCompleteProposal{}, "tendermint/event/CompleteProposal", nil)
 	cdc.RegisterConcrete(EventDataVote{}, "tendermint/event/Vote", nil)
 	cdc.RegisterConcrete(EventDataProposalHeartbeat{}, "tendermint/event/ProposalHeartbeat", nil)
 	cdc.RegisterConcrete(EventDataValidatorSetUpdates{}, "tendermint/event/ValidatorSetUpdates", nil)
@@ -80,6 +81,14 @@ type EventDataRoundState struct {
 	RoundState interface{} `json:"-"`
 }
 
+type EventDataCompleteProposal struct {
+	Height          int64   `json:"height"`
+	Round           int     `json:"round"`
+	ProposerAddress Address `json:"proposer_address"`
+	ProposerIndex   int     `json:"proposer_index"`
+	BlockID         BlockID `json:"block_id"`
+}
+
 type EventDataVote struct {
 	Vote *Vote
 }