From c38a640f525d21ff7e54b6fbf35a2a12740122d3 Mon Sep 17 00:00:00 2001
From: Ethan Buchman <ethan@coinculture.info>
Date: Mon, 12 Sep 2022 15:01:23 +0200
Subject: [PATCH] stop peer on bad consensus msg: * statsMsgQueue ->
 peerInfoQueue * pass badPeerInfo on peerInfoQueue

---
 consensus/reactor.go | 55 ++++++++++++++++++++++++++++----------------
 consensus/state.go   | 31 +++++++++++++++++--------
 2 files changed, 56 insertions(+), 30 deletions(-)

diff --git a/consensus/reactor.go b/consensus/reactor.go
index b0d3e3675..0c0bc5444 100644
--- a/consensus/reactor.go
+++ b/consensus/reactor.go
@@ -877,28 +877,43 @@ func (conR *Reactor) peerStatsRoutine() {
 		}
 
 		select {
-		case msg := <-conR.conS.statsMsgQueue:
-			// Get peer
-			peer := conR.Switch.Peers().Get(msg.PeerID)
-			if peer == nil {
-				conR.Logger.Debug("Attempt to update stats for non-existent peer",
-					"peer", msg.PeerID)
-				continue
-			}
-			// Get peer state
-			ps, ok := peer.Get(types.PeerStateKey).(*PeerState)
-			if !ok {
-				panic(fmt.Sprintf("Peer %v has no state", peer))
-			}
-			switch msg.Msg.(type) {
-			case *VoteMessage:
-				if numVotes := ps.RecordVote(); numVotes%votesToContributeToBecomeGoodPeer == 0 {
-					conR.Switch.MarkPeerAsGood(peer)
+		case mInterface := <-conR.conS.peerInfoQueue:
+			switch msg := mInterface.(type) {
+			case msgInfo:
+				// Get peer
+				peer := conR.Switch.Peers().Get(msg.PeerID)
+				if peer == nil {
+					conR.Logger.Debug("Attempt to update stats for non-existent peer",
+						"peer", msg.PeerID)
+					continue
 				}
-			case *BlockPartMessage:
-				if numParts := ps.RecordBlockPart(); numParts%blocksToContributeToBecomeGoodPeer == 0 {
-					conR.Switch.MarkPeerAsGood(peer)
+				// Get peer state
+				ps, ok := peer.Get(types.PeerStateKey).(*PeerState)
+				if !ok {
+					panic(fmt.Sprintf("Peer %v has no state", peer))
 				}
+				switch msg.Msg.(type) {
+				case *VoteMessage:
+					if numVotes := ps.RecordVote(); numVotes%votesToContributeToBecomeGoodPeer == 0 {
+						conR.Switch.MarkPeerAsGood(peer)
+					}
+				case *BlockPartMessage:
+					if numParts := ps.RecordBlockPart(); numParts%blocksToContributeToBecomeGoodPeer == 0 {
+						conR.Switch.MarkPeerAsGood(peer)
+					}
+				}
+
+			case badPeerInfo:
+				err, peerID := msg.Error, msg.PeerID
+				peer := conR.Switch.Peers().Get(peerID)
+				if peer == nil {
+					conR.Logger.Debug("Attempt to remove non-existent peer",
+						"peer", peerID)
+					continue
+				}
+				// XXX: does this block? should we run this in its own go-routine
+				// to be safe?
+				conR.Switch.StopPeerForError(peer, err)
 			}
 		case <-conR.conS.Quit():
 			return
diff --git a/consensus/state.go b/consensus/state.go
index c147d7347..01eb4649d 100644
--- a/consensus/state.go
+++ b/consensus/state.go
@@ -48,6 +48,12 @@ type msgInfo struct {
 	PeerID p2p.ID  `json:"peer_key"`
 }
 
+// indicate to the reactor that a peer should be removed
+type badPeerInfo struct {
+	Error  error  `json:"error"`
+	PeerID p2p.ID `json:"peer_key"`
+}
+
 // internally generated messages which may update the state
 type timeoutInfo struct {
 	Duration time.Duration         `json:"duration"`
@@ -109,9 +115,10 @@ type State struct {
 	internalMsgQueue chan msgInfo
 	timeoutTicker    TimeoutTicker
 
-	// information about about added votes and block parts are written on this channel
+	// information about added votes and block parts, and bad peers, are written on this channel
 	// so statistics can be computed by reactor
-	statsMsgQueue chan msgInfo
+	// Takes a `msgInfo` or a `badPeerInfo`
+	peerInfoQueue chan interface{}
 
 	// we use eventBus to trigger msg broadcasts in the reactor,
 	// and to notify external subscribers, eg. through a websocket
@@ -163,7 +170,7 @@ func NewState(
 		peerMsgQueue:     make(chan msgInfo, msgQueueSize),
 		internalMsgQueue: make(chan msgInfo, msgQueueSize),
 		timeoutTicker:    NewTimeoutTicker(),
-		statsMsgQueue:    make(chan msgInfo, msgQueueSize),
+		peerInfoQueue:    make(chan interface{}, msgQueueSize),
 		done:             make(chan struct{}),
 		doWALCatchup:     true,
 		wal:              nilWAL{},
@@ -777,12 +784,16 @@ func (cs *State) receiveRoutine(maxSteps int) {
 					types.ErrVoteInvalidValidatorIndex,
 					types.ErrVoteInvalidValidatorAddress,
 					types.ErrVoteInvalidSignature:
-					//TODO: disconnect from peer. how best to do this ? options:
-					//  - give ConsensusState a reference to ConR or Switch or StopPeer func
-					//  - introduce new StopPeer event for reactor to listen for
-					//  - use the statsMsgQueue (or a new routine/channel)
+
+					// tell reactor to disconnect from peer
+					cs.peerInfoQueue <- badPeerInfo{
+						Error:  err,
+						PeerID: mi.PeerID,
+					}
 				default:
-					//nothing to do
+					// other errors don't necessarily mean the peer is bad,
+					// so there is nothing to do
+					// see https://github.com/tendermint/tendermint/issues/2871
 				}
 			}
 
@@ -861,7 +872,7 @@ func (cs *State) handleMsg(mi msgInfo) (err error) {
 			cs.handleCompleteProposal(msg.Height)
 		}
 		if added {
-			cs.statsMsgQueue <- mi
+			cs.peerInfoQueue <- mi
 		}
 
 		if err != nil && msg.Round != cs.Round {
@@ -880,7 +891,7 @@ func (cs *State) handleMsg(mi msgInfo) (err error) {
 		// if the vote gives us a 2/3-any or 2/3-one, we transition
 		added, err = cs.tryAddVote(msg.Vote, peerID)
 		if added {
-			cs.statsMsgQueue <- mi
+			cs.peerInfoQueue <- mi
 		}
 
 		// if err == ErrAddingVote {