consensus: tie peer threads to peer lifecylce context (#7792)

internal/consensus/peer_state.go

@@ -1,6 +1,7 @@
 package consensus
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -40,6 +41,7 @@ type PeerState struct {
 
 	// NOTE: Modify below using setters, never directly.
 	mtx     sync.RWMutex
+	cancel  context.CancelFunc
 	running bool
 	PRS     cstypes.PeerRoundState `json:"round_state"`
 	Stats   *peerStateStats        `json:"stats"`

internal/consensus/reactor.go

@@ -1009,6 +1009,7 @@ func (r *Reactor) processPeerUpdate(ctx context.Context, peerUpdate p2p.PeerUpda
 			// do not spawn multiple instances of the same goroutines and finally we
 			// set the waitgroup counter so we know when all goroutines have exited.
 			ps.SetRunning(true)
+			ctx, ps.cancel = context.WithCancel(ctx)
 
 			go func() {
 				select {
@@ -1045,6 +1046,7 @@ func (r *Reactor) processPeerUpdate(ctx context.Context, peerUpdate p2p.PeerUpda
 				r.mtx.Unlock()
 
 				ps.SetRunning(false)
+				ps.cancel()
 			}()
 		}
 	}