evidence: terminate broadcastEvidenceRoutine when peer is stopped (#6068)

CHANGELOG_PENDING.md

@@ -26,4 +26,4 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
 
 - [light] \#6022 Fix a bug when the number of validators equals 100 (@melekes)
 - [light] \#6026 Fix a bug when height isn't provided for the rpc calls: `/commit` and `/validators` (@cmwaters)
-
+- [evidence] \#6068 Terminate broadcastEvidenceRoutine when peer is stopped (@melekes)

evidence/reactor.go

@@ -118,16 +118,18 @@ func (evR *Reactor) broadcastEvidenceRoutine(peer p2p.Peer) {
 			case <-evR.Quit():
 				return
 			}
+		} else if !peer.IsRunning() || !evR.IsRunning() {
+			return
 		}
 
 		ev := next.Value.(types.Evidence)
 		evis := evR.prepareEvidenceMessage(peer, ev)
 		if len(evis) > 0 {
+			evR.Logger.Debug("Gossiping evidence to peer", "ev", ev, "peer", peer)
 			msgBytes, err := encodeMsg(evis)
 			if err != nil {
 				panic(err)
 			}
-			evR.Logger.Debug("Gossiping evidence to peer", "ev", ev, "peer", peer.ID())
 			success := peer.Send(EvidenceChannel, msgBytes)
 			if !success {
 				time.Sleep(peerRetryMessageIntervalMS * time.Millisecond)

evidence/reactor_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/fortytw2/leaktest"
 	"github.com/go-kit/kit/log/term"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -21,6 +22,7 @@ import (
 	"github.com/tendermint/tendermint/evidence/mocks"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/p2p"
+	p2pmocks "github.com/tendermint/tendermint/p2p/mocks"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
@@ -185,6 +187,41 @@ func TestReactorsGossipNoCommittedEvidence(t *testing.T) {
 	assert.EqualValues(t, []types.Evidence{evList[0], evList[1]}, peerEv)
 }
 
+func TestReactorBroadcastEvidenceMemoryLeak(t *testing.T) {
+	evidenceTime := time.Date(2019, 1, 1, 0, 0, 0, 0, time.UTC)
+	evidenceDB := dbm.NewMemDB()
+	blockStore := &mocks.BlockStore{}
+	blockStore.On("LoadBlockMeta", mock.AnythingOfType("int64")).Return(
+		&types.BlockMeta{Header: types.Header{Time: evidenceTime}},
+	)
+	val := types.NewMockPV()
+	stateStore := initializeValidatorState(val, 1)
+	pool, err := evidence.NewPool(evidenceDB, stateStore, blockStore)
+	require.NoError(t, err)
+
+	p := &p2pmocks.Peer{}
+
+	p.On("IsRunning").Once().Return(true)
+	p.On("IsRunning").Return(false)
+	// check that we are not leaking any go-routines
+	// i.e. broadcastEvidenceRoutine finishes when peer is stopped
+	defer leaktest.CheckTimeout(t, 10*time.Second)()
+
+	p.On("Send", evidence.EvidenceChannel, mock.AnythingOfType("[]uint8")).Return(false)
+	quitChan := make(<-chan struct{})
+	p.On("Quit").Return(quitChan)
+	ps := peerState{2}
+	p.On("Get", types.PeerStateKey).Return(ps)
+	p.On("ID").Return("ABC")
+	p.On("String").Return("mock")
+
+	r := evidence.NewReactor(pool)
+	r.SetLogger(log.TestingLogger())
+	r.AddPeer(p)
+
+	_ = sendEvidence(t, pool, val, 2)
+}
+
 // evidenceLogger is a TestingLogger which uses a different
 // color for each validator ("validator" key must exist).
 func evidenceLogger() log.Logger {