mempool/v1: test reactor does not panic on broadcast

internal/mempool/v1/mempool.go

@@ -748,7 +748,7 @@ func (txmp *TxMempool) insertTx(wtx *WrappedTx) {
 	// Insert the transaction into the gossip index and mark the reference to the
 	// linked-list element, which will be needed at a later point when the
 	// transaction is removed.
-	gossipEl := txmp.gossipIndex.PushBack(wtx)
+	gossipEl := txmp.gotsipIndex.PushBack(wtx)
 	wtx.gossipEl = gossipEl
 
 	atomic.AddInt64(&txmp.sizeBytes, int64(wtx.Size()))

internal/mempool/v1/reactor.go

@@ -55,6 +55,8 @@ type Reactor struct {
 
 	mtx          tmsync.Mutex
 	peerRoutines map[types.NodeID]*tmsync.Closer
+
+	panicHandler func(interface{})
 }
 
 // NewReactor returns a reference to a new reactor.
@@ -76,12 +78,17 @@ func NewReactor(
 		peerUpdates:  peerUpdates,
 		closeCh:      make(chan struct{}),
 		peerRoutines: make(map[types.NodeID]*tmsync.Closer),
+		panicHandler: defaultPanicHandler,
 	}
 
 	r.BaseService = *service.NewBaseService(logger, "Mempool", r)
 	return r
 }
 
+func defaultPanicHandler(r interface{}) {
+	panic(r)
+}
+
 // GetChannelShims returns a map of ChannelDescriptorShim objects, where each
 // object wraps a reference to a legacy p2p ChannelDescriptor and the corresponding
 // p2p proto.Message the new p2p Channel is responsible for handling.
@@ -187,7 +194,8 @@ func (r *Reactor) handleMempoolMessage(envelope p2p.Envelope) error {
 func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err error) {
 	defer func() {
 		if e := recover(); e != nil {
-			err = fmt.Errorf("panic in processing message: %v", e)
+			r.panicHandler(e)
+			err = fmt.Errorf("panic in processing message: %v", r)
 			r.Logger.Error(
 				"recovering from processing message panic",
 				"err", err,
@@ -317,9 +325,10 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer)
 		r.peerWG.Done()
 
 		if e := recover(); e != nil {
+			r.panicHandler(e)
 			r.Logger.Error(
 				"recovering from broadcasting mempool loop",
-				"err", e,
+				"err", r,
 				"stack", string(debug.Stack()),
 			)
 		}

internal/mempool/v1/reactor_test.go

@@ -0,0 +1,137 @@
+package v1
+
+import (
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tendermint/tendermint/abci/example/kvstore"
+	"github.com/tendermint/tendermint/config"
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+	"github.com/tendermint/tendermint/internal/mempool"
+	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/internal/p2p/p2ptest"
+	"github.com/tendermint/tendermint/libs/log"
+	protomem "github.com/tendermint/tendermint/proto/tendermint/mempool"
+	"github.com/tendermint/tendermint/types"
+)
+
+type reactorTestSuite struct {
+	network *p2ptest.Network
+	logger  log.Logger
+
+	reactors        map[types.NodeID]*Reactor
+	mempoolChannels map[types.NodeID]*p2p.Channel
+	mempools        map[types.NodeID]*TxMempool
+	kvstores        map[types.NodeID]*kvstore.Application
+
+	peerChans   map[types.NodeID]chan p2p.PeerUpdate
+	peerUpdates map[types.NodeID]*p2p.PeerUpdates
+
+	nodes []types.NodeID
+}
+
+func setupReactors(t *testing.T, numNodes int, chBuf uint) *reactorTestSuite {
+	t.Helper()
+
+	cfg := config.ResetTestRoot(strings.ReplaceAll(t.Name(), "/", "|"))
+	t.Cleanup(func() {
+		os.RemoveAll(cfg.RootDir)
+	})
+
+	rts := &reactorTestSuite{
+		logger:          log.TestingLogger().With("testCase", t.Name()),
+		network:         p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numNodes}),
+		reactors:        make(map[types.NodeID]*Reactor, numNodes),
+		mempoolChannels: make(map[types.NodeID]*p2p.Channel, numNodes),
+		mempools:        make(map[types.NodeID]*TxMempool, numNodes),
+		kvstores:        make(map[types.NodeID]*kvstore.Application, numNodes),
+		peerChans:       make(map[types.NodeID]chan p2p.PeerUpdate, numNodes),
+		peerUpdates:     make(map[types.NodeID]*p2p.PeerUpdates, numNodes),
+	}
+
+	chDesc := p2p.ChannelDescriptor{ID: byte(mempool.MempoolChannel)}
+	rts.mempoolChannels = rts.network.MakeChannelsNoCleanup(t, chDesc, new(protomem.Message), int(chBuf))
+
+	for nodeID := range rts.network.Nodes {
+		rts.kvstores[nodeID] = kvstore.NewApplication()
+
+		mempool := setup(t, 0)
+		rts.mempools[nodeID] = mempool
+
+		rts.peerChans[nodeID] = make(chan p2p.PeerUpdate)
+		rts.peerUpdates[nodeID] = p2p.NewPeerUpdates(rts.peerChans[nodeID], 1)
+		rts.network.Nodes[nodeID].PeerManager.Register(rts.peerUpdates[nodeID])
+
+		rts.reactors[nodeID] = NewReactor(
+			rts.logger.With("nodeID", nodeID),
+			cfg.Mempool,
+			rts.network.Nodes[nodeID].PeerManager,
+			mempool,
+			rts.mempoolChannels[nodeID],
+			rts.peerUpdates[nodeID],
+		)
+
+		rts.nodes = append(rts.nodes, nodeID)
+
+		require.NoError(t, rts.reactors[nodeID].Start())
+		require.True(t, rts.reactors[nodeID].IsRunning())
+	}
+
+	require.Len(t, rts.reactors, numNodes)
+
+	t.Cleanup(func() {
+		for nodeID := range rts.reactors {
+			if rts.reactors[nodeID].IsRunning() {
+				require.NoError(t, rts.reactors[nodeID].Stop())
+				require.False(t, rts.reactors[nodeID].IsRunning())
+			}
+		}
+	})
+
+	return rts
+}
+
+func (rts *reactorTestSuite) start(t *testing.T) {
+	t.Helper()
+	rts.network.Start(t)
+	require.Len(t,
+		rts.network.RandomNode().PeerManager.Peers(),
+		len(rts.nodes)-1,
+		"network does not have expected number of nodes")
+}
+
+func TestReactorBroadcastDoesNotPanic(t *testing.T) {
+	numNodes := 2
+	rts := setupReactors(t, numNodes, 0)
+
+	panicHandler := func(r interface{}) {
+		t.Fatalf("panic detected in reactor")
+	}
+
+	primary := rts.nodes[0]
+	secondary := rts.nodes[1]
+	primaryReactor := rts.reactors[primary]
+	primaryMempool := primaryReactor.mempool
+	secondaryReactor := rts.reactors[secondary]
+
+	primaryReactor.panicHandler = panicHandler
+	secondaryReactor.panicHandler = panicHandler
+
+	firstTx := &WrappedTx{}
+	primaryMempool.insertTx(firstTx)
+
+	// run the router
+	rts.start(t)
+
+	closer := tmsync.NewCloser()
+	primaryReactor.peerWG.Add(1)
+	go primaryReactor.broadcastTxRoutine(secondary, closer)
+
+	second := &WrappedTx{}
+	primaryMempool.insertTx(second)
+
+	primaryReactor.Stop()
+	primaryReactor.peerWG.Wait()
+}