#2815 do not broadcast heartbeat proposal when we are non-validator (#…

…2819)

* #2815 do not broadcast heartbeat proposal when we are non-validator

* #2815 adding preliminary changelog entry

* #2815 cosmetics and added test

* #2815 missed a little detail

- it's enough to call getAddress() once here

* #2815 remove debug logging from tests

* #2815 OK. I seem to be doing something fundamentally wrong here

* #2815 next iteration of proposalHeartbeat tests

- try and use "ensure" pattern in common_test

* 2815 incorporate review comments

CHANGELOG_PENDING.md

@@ -39,5 +39,5 @@ program](https://hackerone.com/tendermint).
 - [p2p] \#2857 "Send failed" is logged at debug level instead of error.
 
 ### BUG FIXES:
-
+- [consensus] \#2819 Don't send proposalHearbeat if not a validator
 - [p2p] \#2869 Set connection config properly instead of always using default

consensus/common_test.go

@@ -425,6 +425,20 @@ func ensureNewRound(roundCh <-chan interface{}, height int64, round int) {
 	}
 }
 
+func ensureProposalHeartbeat(heartbeatCh <-chan interface{}) {
+	select {
+	case <-time.After(ensureTimeout):
+		panic("Timeout expired while waiting for ProposalHeartbeat event")
+	case ev := <-heartbeatCh:
+		heartbeat, ok := ev.(types.EventDataProposalHeartbeat)
+		if !ok {
+			panic(fmt.Sprintf("expected a *types.EventDataProposalHeartbeat, "+
+				"got %v. wrong subscription channel?",
+				reflect.TypeOf(heartbeat)))
+		}
+	}
+}
+
 func ensureNewTimeout(timeoutCh <-chan interface{}, height int64, round int, timeout int64) {
 	timeoutDuration := time.Duration(timeout*3) * time.Nanosecond
 	ensureNewEvent(timeoutCh, height, round, timeoutDuration,

consensus/reactor.go

@@ -402,7 +402,7 @@ func (conR *ConsensusReactor) unsubscribeFromBroadcastEvents() {
 
 func (conR *ConsensusReactor) broadcastProposalHeartbeatMessage(hb *types.Heartbeat) {
 	conR.Logger.Debug("Broadcasting proposal heartbeat message",
-		"height", hb.Height, "round", hb.Round, "sequence", hb.Sequence)
+		"height", hb.Height, "round", hb.Round, "sequence", hb.Sequence, "address", hb.ValidatorAddress)
 	msg := &ProposalHeartbeatMessage{hb}
 	conR.Switch.Broadcast(StateChannel, cdc.MustMarshalBinaryBare(msg))
 }

consensus/state.go

@@ -802,8 +802,14 @@ func (cs *ConsensusState) needProofBlock(height int64) bool {
 }
 
 func (cs *ConsensusState) proposalHeartbeat(height int64, round int) {
-	counter := 0
+	logger := cs.Logger.With("height", height, "round", round)
 	addr := cs.privValidator.GetAddress()
+
+	if !cs.Validators.HasAddress(addr) {
+		logger.Debug("Not sending proposalHearbeat. This node is not a validator", "addr", addr, "vals", cs.Validators)
+		return
+	}
+	counter := 0
 	valIndex, _ := cs.Validators.GetByAddress(addr)
 	chainID := cs.state.ChainID
 	for {

consensus/state_test.go

@@ -11,6 +11,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	cstypes "github.com/tendermint/tendermint/consensus/types"
+	tmevents "github.com/tendermint/tendermint/libs/events"
+
 	cmn "github.com/tendermint/tendermint/libs/common"
 	"github.com/tendermint/tendermint/libs/log"
 	tmpubsub "github.com/tendermint/tendermint/libs/pubsub"
@@ -1027,6 +1029,33 @@ func TestSetValidBlockOnDelayedPrevote(t *testing.T) {
 	assert.True(t, rs.ValidRound == round)
 }
 
+// regression for #2518
+func TestNoHearbeatWhenNotValidator(t *testing.T) {
+	cs, _ := randConsensusState(4)
+	cs.Validators = types.NewValidatorSet(nil) // make sure we are not in the validator set
+
+	cs.evsw.AddListenerForEvent("testing", types.EventProposalHeartbeat,
+		func(data tmevents.EventData) {
+			t.Errorf("Should not have broadcasted heartbeat")
+		})
+	go cs.proposalHeartbeat(10, 1)
+
+	cs.Stop()
+
+	// if a faulty implementation sends an event, we should wait here a little bit to make sure we don't miss it by prematurely leaving the test method
+	time.Sleep((proposalHeartbeatIntervalSeconds  + 1) * time.Second)
+}
+
+// regression for #2518
+func TestHearbeatWhenWeAreValidator(t *testing.T) {
+	cs, _ := randConsensusState(4)
+	heartbeatCh := subscribe(cs.eventBus, types.EventQueryProposalHeartbeat)
+
+	go cs.proposalHeartbeat(10, 1)
+	ensureProposalHeartbeat(heartbeatCh)
+
+}
+
 // What we want:
 // P0 miss to lock B as Proposal Block is missing, but set valid block to B after
 // receiving delayed Block Proposal.