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

CHANGELOG_PENDING.md

@@ -22,6 +22,8 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
 
 ### FEATURES:
 
+- [eventbus] create new event types for NewRound and CompleteProposal. add proposed block info to CompleteProposal event and proposer info to NewRound event
+
 ### IMPROVEMENTS:
 
 ### BUG FIXES:

consensus/common_test.go

@@ -405,8 +405,24 @@ func ensureNewVote(voteCh <-chan interface{}, height int64, round int) {
 }
 
 func ensureNewRound(roundCh <-chan interface{}, height int64, round int) {
-	ensureNewEvent(roundCh, height, round, ensureTimeout,
-		"Timeout expired while waiting for NewRound event")
+	select {
+	case <-time.After(ensureTimeout):
+		panic("Timeout expired while waiting for NewRound event")
+	case ev := <-roundCh:
+		rs, ok := ev.(types.EventDataNewRound)
+		if !ok {
+			panic(
+				fmt.Sprintf(
+					"expected a EventDataNewRound, 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 ensureNewTimeout(timeoutCh <-chan interface{}, height int64, round int, timeout int64) {
@@ -416,8 +432,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) {
@@ -492,6 +524,31 @@ func ensureVote(voteCh <-chan interface{}, height int64, round int,
 	}
 }
 
+func ensureProposal(proposalCh <-chan interface{}, height int64, round int, propId types.BlockID) {
+	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))
+		}
+		if !rs.BlockID.Equals(propId) {
+			panic("Proposed block does not match expected block")
+		}
+	}
+}
+
+
 func ensurePrecommit(voteCh <-chan interface{}, height int64, round int) {
 	ensureVote(voteCh, height, round, types.PrecommitType)
 }

consensus/mempool_test.go

@@ -71,18 +71,18 @@ func TestMempoolProgressInHigherRound(t *testing.T) {
 	}
 	startTestRound(cs, height, round)
 
-	ensureNewRoundStep(newRoundCh, height, round) // first round at first height
+	ensureNewRound(newRoundCh, height, round) // first round at first height
 	ensureNewEventOnChannel(newBlockCh)           // first block gets committed
 
 	height = height + 1 // moving to the next height
 	round = 0
 
-	ensureNewRoundStep(newRoundCh, height, round) // first round at next height
+	ensureNewRound(newRoundCh, height, round) // first round at next height
 	deliverTxsRange(cs, 0, 1)                     // we deliver txs, but dont set a proposal so we get the next round
 	ensureNewTimeout(timeoutCh, height, round, cs.config.TimeoutPropose.Nanoseconds())
 
 	round = round + 1                             // moving to the next round
-	ensureNewRoundStep(newRoundCh, height, round) // wait for the next round
+	ensureNewRound(newRoundCh, height, round) // wait for the next round
 	ensureNewEventOnChannel(newBlockCh)           // now we can commit the block
 }
 

consensus/state.go

@@ -776,7 +776,7 @@ func (cs *ConsensusState) enterNewRound(height int64, round int) {
 	cs.Votes.SetRound(round + 1) // also track next round (round+1) to allow round-skipping
 	cs.triggeredTimeoutPrecommit = false
 
-	cs.eventBus.PublishEventNewRound(cs.RoundStateEvent())
+	cs.eventBus.PublishEventNewRound(cs.NewRoundEvent())
 	cs.metrics.Rounds.Set(float64(round))
 
 	// Wait for txs to be available in the mempool
@@ -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/state_test.go

@@ -197,9 +197,8 @@ func TestStateBadProposal(t *testing.T) {
 	stateHash[0] = byte((stateHash[0] + 1) % 255)
 	propBlock.AppHash = stateHash
 	propBlockParts := propBlock.MakePartSet(partSize)
-	proposal := types.NewProposal(
-		vs2.Height, round, -1,
-		types.BlockID{propBlock.Hash(), propBlockParts.Header()})
+	blockID := types.BlockID{propBlock.Hash(), propBlockParts.Header()}
+	proposal := types.NewProposal(vs2.Height, round, -1, blockID)
 	if err := vs2.SignProposal(config.ChainID(), proposal); err != nil {
 		t.Fatal("failed to sign bad proposal", err)
 	}
@@ -213,7 +212,7 @@ func TestStateBadProposal(t *testing.T) {
 	startTestRound(cs1, height, round)
 
 	// wait for proposal
-	ensureNewProposal(proposalCh, height, round)
+	ensureProposal(proposalCh, height, round, blockID)
 
 	// wait for prevote
 	ensurePrevote(voteCh, height, round)

consensus/types/round_state.go

@@ -112,18 +112,56 @@ func (rs *RoundState) RoundStateSimple() RoundStateSimple {
 	}
 }
 
+// NewRoundEvent returns the RoundState with proposer information as an event.
+func (rs *RoundState) NewRoundEvent() types.EventDataNewRound {
+	addr := rs.Validators.GetProposer().Address
+	idx, _ := rs.Validators.GetByAddress(addr)
+
+	ednr := types.EventDataNewRound{
+		HeightRoundStep: rs.heightRoundStep(),
+		Proposer: types.ValidatorInfo{
+			Address: addr,
+			Index:   idx,
+		},
+	}
+	return ednr
+}
+
+// CompleteProposalEvent returns information about a proposed block as an event.
+func (rs *RoundState) CompleteProposalEvent() types.EventDataCompleteProposal {
+	// 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{
+		HeightRoundStep: rs.heightRoundStep(),
+		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.
 	// TODO: if we want to avoid this, we may need synchronous events after all
 	rsCopy := *rs
 	edrs := types.EventDataRoundState{
+		HeightRoundStep: rs.heightRoundStep(),
+		RoundState: &rsCopy,
+	}
+	return edrs
+}
+
+func (rs *RoundState) heightRoundStep() types.HeightRoundStep {
+	hrs := types.HeightRoundStep{
 		Height:     rs.Height,
 		Round:      rs.Round,
 		Step:       rs.Step.String(),
-		RoundState: &rsCopy,
 	}
-	return edrs
+	return hrs
 }
 
 // String returns a string

types/event_bus.go

@@ -136,11 +136,11 @@ func (b *EventBus) PublishEventTimeoutWait(data EventDataRoundState) error {
 	return b.Publish(EventTimeoutWait, data)
 }
 
-func (b *EventBus) PublishEventNewRound(data EventDataRoundState) error {
+func (b *EventBus) PublishEventNewRound(data EventDataNewRound) 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

@@ -96,9 +96,9 @@ func TestEventBusPublish(t *testing.T) {
 	require.NoError(t, err)
 	err = eventBus.PublishEventTimeoutWait(EventDataRoundState{})
 	require.NoError(t, err)
-	err = eventBus.PublishEventNewRound(EventDataRoundState{})
+	err = eventBus.PublishEventNewRound(EventDataNewRound{})
 	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,8 @@ 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(EventDataNewRound{}, "tendermint/event/NewRound", 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)
@@ -70,16 +72,37 @@ type EventDataProposalHeartbeat struct {
 	Heartbeat *Heartbeat
 }
 
-// NOTE: This goes into the replay WAL
-type EventDataRoundState struct {
+type HeightRoundStep struct {
 	Height int64  `json:"height"`
 	Round  int    `json:"round"`
 	Step   string `json:"step"`
+}
+
+// NOTE: This goes into the replay WAL
+type EventDataRoundState struct {
+	HeightRoundStep
 
 	// private, not exposed to websockets
 	RoundState interface{} `json:"-"`
 }
 
+type ValidatorInfo struct {
+	Address Address `json:"address"`
+	Index   int     `json:"index"`
+}
+
+type EventDataNewRound struct {
+	HeightRoundStep
+
+	Proposer ValidatorInfo `json:"proposer"`
+}
+
+type EventDataCompleteProposal struct {
+	HeightRoundStep
+
+	BlockID         BlockID `json:"block_id"`
+}
+
 type EventDataVote struct {
 	Vote *Vote
 }