Merge branch 'master' into abci++

internal/blocksync/doc.go

@@ -1,19 +1,19 @@
 /*
-Package blockchain implements two versions of a reactor Service that are
-responsible for block propagation and gossip between peers. This mechanism is
-more formally known as fast-sync.
+Package blocksync implements two versions of a reactor Service that are
+responsible for block propagation and gossip between peers. This mechanism was
+formerly known as fast-sync.
 
 In order for a full node to successfully participate in consensus, it must have
-the latest view of state. The fast-sync protocol is a mechanism in which peers
+the latest view of state. The blocksync protocol is a mechanism in which peers
 may exchange and gossip entire blocks with one another, in a request/response
 type model, until they've successfully synced to the latest head block. Once
 succussfully synced, the full node can switch to an active role in consensus and
-will no longer fast-sync and thus no longer run the fast-sync process.
+will no longer blocksync and thus no longer run the blocksync process.
 
-Note, the blockchain reactor Service gossips entire block and relevant data such
-that each receiving peer may construct the entire view of the blockchain state.
+Note, the blocksync reactor Service gossips entire block and relevant data such
+that each receiving peer may construct the entire view of the blocksync state.
 
-There are two versions of the blockchain reactor Service, i.e. fast-sync:
+There are currently two versions of the blocksync reactor Service:
 
 - v0: The initial implementation that is battle-tested, but whose test coverage
 	is lacking and is not formally verifiable.
@@ -22,7 +22,7 @@ There are two versions of the blockchain reactor Service, i.e. fast-sync:
 	is known to have various bugs that could make it unreliable in production
 	environments.
 
-The v0 blockchain reactor Service has one p2p channel, BlockchainChannel. This
+The v0 blocksync reactor Service has one p2p channel, BlockchainChannel. This
 channel is responsible for handling messages that both request blocks and respond
 to block requests from peers. For every block request from a peer, the reactor
 will execute respondToPeer which will fetch the block from the node's state store
@@ -33,4 +33,4 @@ Internally, v0 runs a poolRoutine that constantly checks for what blocks it need
 and requests them. The poolRoutine is also responsible for taking blocks from the
 pool, saving and executing each block.
 */
-package blockchain
+package blocksync

internal/blocksync/msgs.go

@@ -1,7 +1,7 @@
-package blockchain
+package blocksync
 
 import (
-	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
+	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
 	"github.com/tendermint/tendermint/types"
 )
 

internal/blocksync/v0/pool.go

@@ -65,7 +65,7 @@ type BlockRequest struct {
 	PeerID types.NodeID
 }
 
-// BlockPool keeps track of the fast sync peers, block requests and block responses.
+// BlockPool keeps track of the block sync peers, block requests and block responses.
 type BlockPool struct {
 	service.BaseService
 	lastAdvance time.Time

internal/blocksync/v0/reactor.go

@@ -6,13 +6,13 @@ import (
 	"sync"
 	"time"
 
-	bc "github.com/tendermint/tendermint/internal/blockchain"
+	bc "github.com/tendermint/tendermint/internal/blocksync"
 	cons "github.com/tendermint/tendermint/internal/consensus"
 	"github.com/tendermint/tendermint/internal/p2p"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/service"
 	tmSync "github.com/tendermint/tendermint/libs/sync"
-	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
+	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"
@@ -60,7 +60,7 @@ const (
 )
 
 type consensusReactor interface {
-	// For when we switch from blockchain reactor and fast sync to the consensus
+	// For when we switch from blockchain reactor and block sync to the consensus
 	// machine.
 	SwitchToConsensus(state sm.State, skipWAL bool)
 }
@@ -85,7 +85,7 @@ type Reactor struct {
 	store       *store.BlockStore
 	pool        *BlockPool
 	consReactor consensusReactor
-	fastSync    *tmSync.AtomicBool
+	blockSync   *tmSync.AtomicBool
 
 	blockchainCh *p2p.Channel
 	// blockchainOutBridgeCh defines a channel that acts as a bridge between sending Envelope
@@ -121,7 +121,7 @@ func NewReactor(
 	consReactor consensusReactor,
 	blockchainCh *p2p.Channel,
 	peerUpdates *p2p.PeerUpdates,
-	fastSync bool,
+	blockSync bool,
 	metrics *cons.Metrics,
 ) (*Reactor, error) {
 	if state.LastBlockHeight != store.Height() {
@@ -142,7 +142,7 @@ func NewReactor(
 		store:                 store,
 		pool:                  NewBlockPool(startHeight, requestsCh, errorsCh),
 		consReactor:           consReactor,
-		fastSync:              tmSync.NewBool(fastSync),
+		blockSync:             tmSync.NewBool(blockSync),
 		requestsCh:            requestsCh,
 		errorsCh:              errorsCh,
 		blockchainCh:          blockchainCh,
@@ -162,10 +162,10 @@ func NewReactor(
 // messages on that p2p channel accordingly. The caller must be sure to execute
 // OnStop to ensure the outbound p2p Channels are closed.
 //
-// If fastSync is enabled, we also start the pool and the pool processing
+// If blockSync is enabled, we also start the pool and the pool processing
 // goroutine. If the pool fails to start, an error is returned.
 func (r *Reactor) OnStart() error {
-	if r.fastSync.IsSet() {
+	if r.blockSync.IsSet() {
 		if err := r.pool.Start(); err != nil {
 			return err
 		}
@@ -183,7 +183,7 @@ func (r *Reactor) OnStart() error {
 // OnStop stops the reactor by signaling to all spawned goroutines to exit and
 // blocking until they all exit.
 func (r *Reactor) OnStop() {
-	if r.fastSync.IsSet() {
+	if r.blockSync.IsSet() {
 		if err := r.pool.Stop(); err != nil {
 			r.Logger.Error("failed to stop pool", "err", err)
 		}
@@ -371,10 +371,10 @@ func (r *Reactor) processPeerUpdates() {
 	}
 }
 
-// SwitchToFastSync is called by the state sync reactor when switching to fast
+// SwitchToBlockSync is called by the state sync reactor when switching to fast
 // sync.
-func (r *Reactor) SwitchToFastSync(state sm.State) error {
-	r.fastSync.Set()
+func (r *Reactor) SwitchToBlockSync(state sm.State) error {
+	r.blockSync.Set()
 	r.initialState = state
 	r.pool.height = state.LastBlockHeight + 1
 
@@ -496,7 +496,7 @@ FOR_LOOP:
 				r.Logger.Error("failed to stop pool", "err", err)
 			}
 
-			r.fastSync.UnSet()
+			r.blockSync.UnSet()
 
 			if r.consReactor != nil {
 				r.consReactor.SwitchToConsensus(state, blocksSynced > 0 || stateSynced)
@@ -591,7 +591,7 @@ FOR_LOOP:
 				if blocksSynced%100 == 0 {
 					lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds())
 					r.Logger.Info(
-						"fast sync rate",
+						"block sync rate",
 						"height", r.pool.height,
 						"max_peer_height", r.pool.MaxPeerHeight(),
 						"blocks/s", lastRate,
@@ -614,14 +614,14 @@ func (r *Reactor) GetMaxPeerBlockHeight() int64 {
 }
 
 func (r *Reactor) GetTotalSyncedTime() time.Duration {
-	if !r.fastSync.IsSet() || r.syncStartTime.IsZero() {
+	if !r.blockSync.IsSet() || r.syncStartTime.IsZero() {
 		return time.Duration(0)
 	}
 	return time.Since(r.syncStartTime)
 }
 
 func (r *Reactor) GetRemainingSyncTime() time.Duration {
-	if !r.fastSync.IsSet() {
+	if !r.blockSync.IsSet() {
 		return time.Duration(0)
 	}
 

internal/blocksync/v0/reactor_test.go

@@ -15,7 +15,7 @@ import (
 	"github.com/tendermint/tendermint/internal/p2p/p2ptest"
 	"github.com/tendermint/tendermint/internal/test/factory"
 	"github.com/tendermint/tendermint/libs/log"
-	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
+	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
 	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	sf "github.com/tendermint/tendermint/state/test/factory"
@@ -36,7 +36,7 @@ type reactorTestSuite struct {
 	peerChans          map[types.NodeID]chan p2p.PeerUpdate
 	peerUpdates        map[types.NodeID]*p2p.PeerUpdates
 
-	fastSync bool
+	blockSync bool
 }
 
 func setup(
@@ -61,7 +61,7 @@ func setup(
 		blockchainChannels: make(map[types.NodeID]*p2p.Channel, numNodes),
 		peerChans:          make(map[types.NodeID]chan p2p.PeerUpdate, numNodes),
 		peerUpdates:        make(map[types.NodeID]*p2p.PeerUpdates, numNodes),
-		fastSync:           true,
+		blockSync:          true,
 	}
 
 	chDesc := p2p.ChannelDescriptor{ID: byte(BlockchainChannel)}
@@ -163,7 +163,7 @@ func (rts *reactorTestSuite) addNode(t *testing.T,
 		nil,
 		rts.blockchainChannels[nodeID],
 		rts.peerUpdates[nodeID],
-		rts.fastSync,
+		rts.blockSync,
 		cons.NopMetrics())
 	require.NoError(t, err)
 

internal/blocksync/v2/internal/behavior/reporter_test.go

@@ -4,7 +4,7 @@ import (
 	"sync"
 	"testing"
 
-	bh "github.com/tendermint/tendermint/internal/blockchain/v2/internal/behavior"
+	bh "github.com/tendermint/tendermint/internal/blocksync/v2/internal/behavior"
 	"github.com/tendermint/tendermint/types"
 )
 

internal/blocksync/v2/io.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/tendermint/tendermint/internal/p2p"
-	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
+	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
 	"github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
 )
@@ -42,7 +42,7 @@ const (
 )
 
 type consensusReactor interface {
-	// for when we switch from blockchain reactor and fast sync to
+	// for when we switch from blockchain reactor and block sync to
 	// the consensus machine
 	SwitchToConsensus(state state.State, skipWAL bool)
 }

internal/blocksync/v2/reactor.go

@@ -7,14 +7,14 @@ import (
 
 	proto "github.com/gogo/protobuf/proto"
 
-	bc "github.com/tendermint/tendermint/internal/blockchain"
-	"github.com/tendermint/tendermint/internal/blockchain/v2/internal/behavior"
+	bc "github.com/tendermint/tendermint/internal/blocksync"
+	"github.com/tendermint/tendermint/internal/blocksync/v2/internal/behavior"
 	cons "github.com/tendermint/tendermint/internal/consensus"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
 	"github.com/tendermint/tendermint/internal/p2p"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/sync"
-	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
+	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
 	"github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
 )
@@ -31,12 +31,12 @@ type blockStore interface {
 	Height() int64
 }
 
-// BlockchainReactor handles fast sync protocol.
+// BlockchainReactor handles block sync protocol.
 type BlockchainReactor struct {
 	p2p.BaseReactor
 
-	fastSync    *sync.AtomicBool // enable fast sync on start when it's been Set
-	stateSynced bool             // set to true when SwitchToFastSync is called by state sync
+	blockSync   *sync.AtomicBool // enable block sync on start when it's been Set
+	stateSynced bool             // set to true when SwitchToBlockSync is called by state sync
 	scheduler   *Routine
 	processor   *Routine
 	logger      log.Logger
@@ -44,7 +44,7 @@ type BlockchainReactor struct {
 	mtx           tmsync.RWMutex
 	maxPeerHeight int64
 	syncHeight    int64
-	events        chan Event // non-nil during a fast sync
+	events        chan Event // non-nil during a block sync
 
 	reporter behavior.Reporter
 	io       iIO
@@ -61,7 +61,7 @@ type blockApplier interface {
 
 // XXX: unify naming in this package around tmState
 func newReactor(state state.State, store blockStore, reporter behavior.Reporter,
-	blockApplier blockApplier, fastSync bool, metrics *cons.Metrics) *BlockchainReactor {
+	blockApplier blockApplier, blockSync bool, metrics *cons.Metrics) *BlockchainReactor {
 	initHeight := state.LastBlockHeight + 1
 	if initHeight == 1 {
 		initHeight = state.InitialHeight
@@ -78,7 +78,7 @@ func newReactor(state state.State, store blockStore, reporter behavior.Reporter,
 		store:           store,
 		reporter:        reporter,
 		logger:          log.NewNopLogger(),
-		fastSync:        sync.NewBool(fastSync),
+		blockSync:       sync.NewBool(blockSync),
 		syncStartHeight: initHeight,
 		syncStartTime:   time.Time{},
 		lastSyncRate:    0,
@@ -90,10 +90,10 @@ func NewBlockchainReactor(
 	state state.State,
 	blockApplier blockApplier,
 	store blockStore,
-	fastSync bool,
+	blockSync bool,
 	metrics *cons.Metrics) *BlockchainReactor {
 	reporter := behavior.NewMockReporter()
-	return newReactor(state, store, reporter, blockApplier, fastSync, metrics)
+	return newReactor(state, store, reporter, blockApplier, blockSync, metrics)
 }
 
 // SetSwitch implements Reactor interface.
@@ -137,22 +137,22 @@ func (r *BlockchainReactor) SetLogger(logger log.Logger) {
 // Start implements cmn.Service interface
 func (r *BlockchainReactor) Start() error {
 	r.reporter = behavior.NewSwitchReporter(r.BaseReactor.Switch)
-	if r.fastSync.IsSet() {
+	if r.blockSync.IsSet() {
 		err := r.startSync(nil)
 		if err != nil {
-			return fmt.Errorf("failed to start fast sync: %w", err)
+			return fmt.Errorf("failed to start block sync: %w", err)
 		}
 	}
 	return nil
 }
 
-// startSync begins a fast sync, signaled by r.events being non-nil. If state is non-nil,
+// startSync begins a block sync, signaled by r.events being non-nil. If state is non-nil,
 // the scheduler and processor is updated with this state on startup.
 func (r *BlockchainReactor) startSync(state *state.State) error {
 	r.mtx.Lock()
 	defer r.mtx.Unlock()
 	if r.events != nil {
-		return errors.New("fast sync already in progress")
+		return errors.New("block sync already in progress")
 	}
 	r.events = make(chan Event, chBufferSize)
 	go r.scheduler.start()
@@ -167,7 +167,7 @@ func (r *BlockchainReactor) startSync(state *state.State) error {
 	return nil
 }
 
-// endSync ends a fast sync
+// endSync ends a block sync
 func (r *BlockchainReactor) endSync() {
 	r.mtx.Lock()
 	defer r.mtx.Unlock()
@@ -179,8 +179,8 @@ func (r *BlockchainReactor) endSync() {
 	r.processor.stop()
 }
 
-// SwitchToFastSync is called by the state sync reactor when switching to fast sync.
-func (r *BlockchainReactor) SwitchToFastSync(state state.State) error {
+// SwitchToBlockSync is called by the state sync reactor when switching to block sync.
+func (r *BlockchainReactor) SwitchToBlockSync(state state.State) error {
 	r.stateSynced = true
 	state = state.Copy()
 
@@ -434,7 +434,7 @@ func (r *BlockchainReactor) demux(events <-chan Event) {
 					} else {
 						r.lastSyncRate = 0.9*r.lastSyncRate + 0.1*newSyncRate
 					}
-					r.logger.Info("Fast Sync Rate", "height", r.syncHeight,
+					r.logger.Info("block sync Rate", "height", r.syncHeight,
 						"max_peer_height", r.maxPeerHeight, "blocks/s", r.lastSyncRate)
 					lastHundred = time.Now()
 				}
@@ -442,12 +442,12 @@ func (r *BlockchainReactor) demux(events <-chan Event) {
 			case pcBlockVerificationFailure:
 				r.scheduler.send(event)
 			case pcFinished:
-				r.logger.Info("Fast sync complete, switching to consensus")
+				r.logger.Info("block sync complete, switching to consensus")
 				if !r.io.trySwitchToConsensus(event.tmState, event.blocksSynced > 0 || r.stateSynced) {
 					r.logger.Error("Failed to switch to consensus reactor")
 				}
 				r.endSync()
-				r.fastSync.UnSet()
+				r.blockSync.UnSet()
 				return
 			case noOpEvent:
 			default:
@@ -617,14 +617,14 @@ func (r *BlockchainReactor) GetMaxPeerBlockHeight() int64 {
 }
 
 func (r *BlockchainReactor) GetTotalSyncedTime() time.Duration {
-	if !r.fastSync.IsSet() || r.syncStartTime.IsZero() {
+	if !r.blockSync.IsSet() || r.syncStartTime.IsZero() {
 		return time.Duration(0)
 	}
 	return time.Since(r.syncStartTime)
 }
 
 func (r *BlockchainReactor) GetRemainingSyncTime() time.Duration {
-	if !r.fastSync.IsSet() {
+	if !r.blockSync.IsSet() {
 		return time.Duration(0)
 	}
 

internal/blocksync/v2/reactor_test.go

@@ -15,7 +15,7 @@ import (
 
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
-	"github.com/tendermint/tendermint/internal/blockchain/v2/internal/behavior"
+	"github.com/tendermint/tendermint/internal/blocksync/v2/internal/behavior"
 	cons "github.com/tendermint/tendermint/internal/consensus"
 	"github.com/tendermint/tendermint/internal/mempool/mock"
 	"github.com/tendermint/tendermint/internal/p2p"
@@ -23,7 +23,7 @@ import (
 	"github.com/tendermint/tendermint/internal/test/factory"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/service"
-	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
+	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
 	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	sf "github.com/tendermint/tendermint/state/test/factory"

internal/blocksync/v2/scheduler.go

@@ -163,7 +163,7 @@ type scheduler struct {
 	height int64
 
 	// lastAdvance tracks the last time a block execution happened.
-	// syncTimeout is the maximum time the scheduler waits to advance in the fast sync process before finishing.
+	// syncTimeout is the maximum time the scheduler waits to advance in the block sync process before finishing.
 	// This covers the cases where there are no peers or all peers have a lower height.
 	lastAdvance time.Time
 	syncTimeout time.Duration

internal/consensus/common_test.go

@@ -88,6 +88,7 @@ type validatorStub struct {
 	Round  int32
 	types.PrivValidator
 	VotingPower int64
+	lastVote    *types.Vote
 }
 
 const testMinPower int64 = 10
@@ -121,8 +122,18 @@ func (vs *validatorStub) signVote(
 		BlockID:          types.BlockID{Hash: hash, PartSetHeader: header},
 	}
 	v := vote.ToProto()
-	err = vs.PrivValidator.SignVote(context.Background(), config.ChainID(), v)
+	if err := vs.PrivValidator.SignVote(context.Background(), config.ChainID(), v); err != nil {
+		return nil, fmt.Errorf("sign vote failed: %w", err)
+	}
+
+	// ref: signVote in FilePV, the vote should use the privious vote info when the sign data is the same.
+	if signDataIsEqual(vs.lastVote, v) {
+		v.Signature = vs.lastVote.Signature
+		v.Timestamp = vs.lastVote.Timestamp
+	}
+
 	vote.Signature = v.Signature
+	vote.Timestamp = v.Timestamp
 
 	return vote, err
 }
@@ -139,6 +150,9 @@ func signVote(
 	if err != nil {
 		panic(fmt.Errorf("failed to sign vote: %v", err))
 	}
+
+	vs.lastVote = v
+
 	return v
 }
 
@@ -876,3 +890,16 @@ func newKVStore() abci.Application {
 func newPersistentKVStoreWithPath(dbDir string) abci.Application {
 	return kvstore.NewPersistentKVStoreApplication(dbDir)
 }
+
+func signDataIsEqual(v1 *types.Vote, v2 *tmproto.Vote) bool {
+	if v1 == nil || v2 == nil {
+		return false
+	}
+
+	return v1.Type == v2.Type &&
+		bytes.Equal(v1.BlockID.Hash, v2.BlockID.GetHash()) &&
+		v1.Height == v2.GetHeight() &&
+		v1.Round == v2.Round &&
+		bytes.Equal(v1.ValidatorAddress.Bytes(), v2.GetValidatorAddress()) &&
+		v1.ValidatorIndex == v2.GetValidatorIndex()
+}

internal/consensus/metrics.go

@@ -54,8 +54,8 @@ type Metrics struct {
 	TotalTxs metrics.Gauge
 	// The latest block height.
 	CommittedHeight metrics.Gauge
-	// Whether or not a node is fast syncing. 1 if yes, 0 if no.
-	FastSyncing metrics.Gauge
+	// Whether or not a node is block syncing. 1 if yes, 0 if no.
+	BlockSyncing metrics.Gauge
 	// Whether or not a node is state syncing. 1 if yes, 0 if no.
 	StateSyncing metrics.Gauge
 
@@ -169,11 +169,11 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Name:      "latest_block_height",
 			Help:      "The latest block height.",
 		}, labels).With(labelsAndValues...),
-		FastSyncing: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+		BlockSyncing: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
 			Namespace: namespace,
 			Subsystem: MetricsSubsystem,
-			Name:      "fast_syncing",
-			Help:      "Whether or not a node is fast syncing. 1 if yes, 0 if no.",
+			Name:      "block_syncing",
+			Help:      "Whether or not a node is block syncing. 1 if yes, 0 if no.",
 		}, labels).With(labelsAndValues...),
 		StateSyncing: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
 			Namespace: namespace,
@@ -214,7 +214,7 @@ func NopMetrics() *Metrics {
 		BlockSizeBytes:  discard.NewHistogram(),
 		TotalTxs:        discard.NewGauge(),
 		CommittedHeight: discard.NewGauge(),
-		FastSyncing:     discard.NewGauge(),
+		BlockSyncing:    discard.NewGauge(),
 		StateSyncing:    discard.NewGauge(),
 		BlockParts:      discard.NewCounter(),
 	}

internal/consensus/mocks/cons_sync_reactor.go

@@ -1,4 +1,4 @@
-// Code generated by mockery 2.7.5. DO NOT EDIT.
+// Code generated by mockery. DO NOT EDIT.
 
 package mocks
 
@@ -12,8 +12,8 @@ type ConsSyncReactor struct {
 	mock.Mock
 }
 
-// SetFastSyncingMetrics provides a mock function with given fields: _a0
-func (_m *ConsSyncReactor) SetFastSyncingMetrics(_a0 float64) {
+// SetBlockSyncingMetrics provides a mock function with given fields: _a0
+func (_m *ConsSyncReactor) SetBlockSyncingMetrics(_a0 float64) {
 	_m.Called(_a0)
 }
 

internal/consensus/mocks/fast_sync_reactor.go

@@ -9,13 +9,13 @@ import (
 	time "time"
 )
 
-// FastSyncReactor is an autogenerated mock type for the FastSyncReactor type
-type FastSyncReactor struct {
+// BlockSyncReactor is an autogenerated mock type for the BlockSyncReactor type
+type BlockSyncReactor struct {
 	mock.Mock
 }
 
 // GetMaxPeerBlockHeight provides a mock function with given fields:
-func (_m *FastSyncReactor) GetMaxPeerBlockHeight() int64 {
+func (_m *BlockSyncReactor) GetMaxPeerBlockHeight() int64 {
 	ret := _m.Called()
 
 	var r0 int64
@@ -29,7 +29,7 @@ func (_m *FastSyncReactor) GetMaxPeerBlockHeight() int64 {
 }
 
 // GetRemainingSyncTime provides a mock function with given fields:
-func (_m *FastSyncReactor) GetRemainingSyncTime() time.Duration {
+func (_m *BlockSyncReactor) GetRemainingSyncTime() time.Duration {
 	ret := _m.Called()
 
 	var r0 time.Duration
@@ -43,7 +43,7 @@ func (_m *FastSyncReactor) GetRemainingSyncTime() time.Duration {
 }
 
 // GetTotalSyncedTime provides a mock function with given fields:
-func (_m *FastSyncReactor) GetTotalSyncedTime() time.Duration {
+func (_m *BlockSyncReactor) GetTotalSyncedTime() time.Duration {
 	ret := _m.Called()
 
 	var r0 time.Duration
@@ -56,8 +56,8 @@ func (_m *FastSyncReactor) GetTotalSyncedTime() time.Duration {
 	return r0
 }
 
-// SwitchToFastSync provides a mock function with given fields: _a0
-func (_m *FastSyncReactor) SwitchToFastSync(_a0 state.State) error {
+// SwitchToBlockSync provides a mock function with given fields: _a0
+func (_m *BlockSyncReactor) SwitchToBlockSync(_a0 state.State) error {
 	ret := _m.Called(_a0)
 
 	var r0 error

internal/consensus/reactor.go

@@ -96,28 +96,28 @@ const (
 
 type ReactorOption func(*Reactor)
 
-// Temporary interface for switching to fast sync, we should get rid of v0.
+// NOTE: Temporary interface for switching to block sync, we should get rid of v0.
 // See: https://github.com/tendermint/tendermint/issues/4595
-type FastSyncReactor interface {
-	SwitchToFastSync(sm.State) error
+type BlockSyncReactor interface {
+	SwitchToBlockSync(sm.State) error
 
 	GetMaxPeerBlockHeight() int64
 
-	// GetTotalSyncedTime returns the time duration since the fastsync starting.
+	// GetTotalSyncedTime returns the time duration since the blocksync starting.
 	GetTotalSyncedTime() time.Duration
 
 	// GetRemainingSyncTime returns the estimating time the node will be fully synced,
-	// if will return 0 if the fastsync does not perform or the number of block synced is
+	// if will return 0 if the blocksync does not perform or the number of block synced is
 	// too small (less than 100).
 	GetRemainingSyncTime() time.Duration
 }
 
-//go:generate mockery --case underscore --name ConsSyncReactor
+//go:generate ../../scripts/mockery_generate.sh ConsSyncReactor
 // ConsSyncReactor defines an interface used for testing abilities of node.startStateSync.
 type ConsSyncReactor interface {
 	SwitchToConsensus(sm.State, bool)
 	SetStateSyncingMetrics(float64)
-	SetFastSyncingMetrics(float64)
+	SetBlockSyncingMetrics(float64)
 }
 
 // Reactor defines a reactor for the consensus service.
@@ -265,7 +265,7 @@ func (r *Reactor) SetEventBus(b *types.EventBus) {
 	r.state.SetEventBus(b)
 }
 
-// WaitSync returns whether the consensus reactor is waiting for state/fast sync.
+// WaitSync returns whether the consensus reactor is waiting for state/block sync.
 func (r *Reactor) WaitSync() bool {
 	r.mtx.RLock()
 	defer r.mtx.RUnlock()
@@ -278,8 +278,8 @@ func ReactorMetrics(metrics *Metrics) ReactorOption {
 	return func(r *Reactor) { r.Metrics = metrics }
 }
 
-// SwitchToConsensus switches from fast-sync mode to consensus mode. It resets
-// the state, turns off fast-sync, and starts the consensus state-machine.
+// SwitchToConsensus switches from block-sync mode to consensus mode. It resets
+// the state, turns off block-sync, and starts the consensus state-machine.
 func (r *Reactor) SwitchToConsensus(state sm.State, skipWAL bool) {
 	r.Logger.Info("switching to consensus")
 
@@ -296,7 +296,7 @@ func (r *Reactor) SwitchToConsensus(state sm.State, skipWAL bool) {
 	r.waitSync = false
 	r.mtx.Unlock()
 
-	r.Metrics.FastSyncing.Set(0)
+	r.Metrics.BlockSyncing.Set(0)
 	r.Metrics.StateSyncing.Set(0)
 
 	if skipWAL {
@@ -313,9 +313,9 @@ conR:
 %+v`, err, r.state, r))
 	}
 
-	d := types.EventDataFastSyncStatus{Complete: true, Height: state.LastBlockHeight}
-	if err := r.eventBus.PublishEventFastSyncStatus(d); err != nil {
-		r.Logger.Error("failed to emit the fastsync complete event", "err", err)
+	d := types.EventDataBlockSyncStatus{Complete: true, Height: state.LastBlockHeight}
+	if err := r.eventBus.PublishEventBlockSyncStatus(d); err != nil {
+		r.Logger.Error("failed to emit the blocksync complete event", "err", err)
 	}
 }
 
@@ -969,7 +969,7 @@ func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
 			go r.gossipVotesRoutine(ps)
 			go r.queryMaj23Routine(ps)
 
-			// Send our state to the peer. If we're fast-syncing, broadcast a
+			// Send our state to the peer. If we're block-syncing, broadcast a
 			// RoundStepMessage later upon SwitchToConsensus().
 			if !r.waitSync {
 				go r.sendNewRoundStepMessage(ps.peerID)
@@ -1219,7 +1219,7 @@ func (r *Reactor) handleVoteSetBitsMessage(envelope p2p.Envelope, msgI Message)
 // It will handle errors and any possible panics gracefully. A caller can handle
 // any error returned by sending a PeerError on the respective channel.
 //
-// NOTE: We process these messages even when we're fast_syncing. Messages affect
+// NOTE: We process these messages even when we're block syncing. Messages affect
 // either a peer state or the consensus state. Peer state updates can happen in
 // parallel, but processing of proposals, block parts, and votes are ordered by
 // the p2p channel.
@@ -1442,6 +1442,6 @@ func (r *Reactor) SetStateSyncingMetrics(v float64) {
 	r.Metrics.StateSyncing.Set(v)
 }
 
-func (r *Reactor) SetFastSyncingMetrics(v float64) {
-	r.Metrics.FastSyncing.Set(v)
+func (r *Reactor) SetBlockSyncingMetrics(v float64) {
+	r.Metrics.BlockSyncing.Set(v)
 }

internal/consensus/reactor_test.go

@@ -43,7 +43,7 @@ type reactorTestSuite struct {
 	states              map[types.NodeID]*State
 	reactors            map[types.NodeID]*Reactor
 	subs                map[types.NodeID]types.Subscription
-	fastsyncSubs        map[types.NodeID]types.Subscription
+	blocksyncSubs       map[types.NodeID]types.Subscription
 	stateChannels       map[types.NodeID]*p2p.Channel
 	dataChannels        map[types.NodeID]*p2p.Channel
 	voteChannels        map[types.NodeID]*p2p.Channel
@@ -60,11 +60,11 @@ func setup(t *testing.T, numNodes int, states []*State, size int) *reactorTestSu
 	t.Helper()
 
 	rts := &reactorTestSuite{
-		network:      p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numNodes}),
-		states:       make(map[types.NodeID]*State),
-		reactors:     make(map[types.NodeID]*Reactor, numNodes),
-		subs:         make(map[types.NodeID]types.Subscription, numNodes),
-		fastsyncSubs: make(map[types.NodeID]types.Subscription, numNodes),
+		network:       p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numNodes}),
+		states:        make(map[types.NodeID]*State),
+		reactors:      make(map[types.NodeID]*Reactor, numNodes),
+		subs:          make(map[types.NodeID]types.Subscription, numNodes),
+		blocksyncSubs: make(map[types.NodeID]types.Subscription, numNodes),
 	}
 
 	rts.stateChannels = rts.network.MakeChannelsNoCleanup(t, chDesc(StateChannel), new(tmcons.Message), size)
@@ -94,13 +94,13 @@ func setup(t *testing.T, numNodes int, states []*State, size int) *reactorTestSu
 		blocksSub, err := state.eventBus.Subscribe(context.Background(), testSubscriber, types.EventQueryNewBlock, size)
 		require.NoError(t, err)
 
-		fsSub, err := state.eventBus.Subscribe(context.Background(), testSubscriber, types.EventQueryFastSyncStatus, size)
+		fsSub, err := state.eventBus.Subscribe(context.Background(), testSubscriber, types.EventQueryBlockSyncStatus, size)
 		require.NoError(t, err)
 
 		rts.states[nodeID] = state
 		rts.subs[nodeID] = blocksSub
 		rts.reactors[nodeID] = reactor
-		rts.fastsyncSubs[nodeID] = fsSub
+		rts.blocksyncSubs[nodeID] = fsSub
 
 		// simulate handle initChain in handshake
 		if state.state.LastBlockHeight == 0 {
@@ -263,9 +263,9 @@ func waitForBlockWithUpdatedValsAndValidateIt(
 	wg.Wait()
 }
 
-func ensureFastSyncStatus(t *testing.T, msg tmpubsub.Message, complete bool, height int64) {
+func ensureBlockSyncStatus(t *testing.T, msg tmpubsub.Message, complete bool, height int64) {
 	t.Helper()
-	status, ok := msg.Data().(types.EventDataFastSyncStatus)
+	status, ok := msg.Data().(types.EventDataBlockSyncStatus)
 
 	require.True(t, ok)
 	require.Equal(t, complete, status.Complete)
@@ -301,14 +301,14 @@ func TestReactorBasic(t *testing.T) {
 
 	wg.Wait()
 
-	for _, sub := range rts.fastsyncSubs {
+	for _, sub := range rts.blocksyncSubs {
 		wg.Add(1)
 
 		// wait till everyone makes the consensus switch
 		go func(s types.Subscription) {
 			defer wg.Done()
 			msg := <-s.Out()
-			ensureFastSyncStatus(t, msg, true, 0)
+			ensureBlockSyncStatus(t, msg, true, 0)
 		}(sub)
 	}
 

internal/consensus/replay_test.go

@@ -1203,8 +1203,8 @@ func (bs *mockBlockStore) SaveBlock(block *types.Block, blockParts *types.PartSe
 func (bs *mockBlockStore) LoadBlockCommit(height int64) *types.Commit {
 	return bs.commits[height-1]
 }
-func (bs *mockBlockStore) LoadSeenCommit(height int64) *types.Commit {
-	return bs.commits[height-1]
+func (bs *mockBlockStore) LoadSeenCommit() *types.Commit {
+	return bs.commits[len(bs.commits)-1]
 }
 
 func (bs *mockBlockStore) PruneBlocks(height int64) (uint64, error) {

internal/consensus/state.go

@@ -314,7 +314,14 @@ func (cs *State) LoadCommit(height int64) *types.Commit {
 	defer cs.mtx.RUnlock()
 
 	if height == cs.blockStore.Height() {
-		return cs.blockStore.LoadSeenCommit(height)
+		commit := cs.blockStore.LoadSeenCommit()
+		// NOTE: Retrieving the height of the most recent block and retrieving
+		// the most recent commit does not currently occur as an atomic
+		// operation. We check the height and commit here in case a more recent
+		// commit has arrived since retrieving the latest height.
+		if commit != nil && commit.Height == height {
+			return commit
+		}
 	}
 
 	return cs.blockStore.LoadBlockCommit(height)
@@ -594,15 +601,19 @@ func (cs *State) sendInternalMessage(mi msgInfo) {
 // Reconstruct LastCommit from SeenCommit, which we saved along with the block,
 // (which happens even before saving the state)
 func (cs *State) reconstructLastCommit(state sm.State) {
-	seenCommit := cs.blockStore.LoadSeenCommit(state.LastBlockHeight)
-	if seenCommit == nil {
+	commit := cs.blockStore.LoadSeenCommit()
+	if commit == nil || commit.Height != state.LastBlockHeight {
+		commit = cs.blockStore.LoadBlockCommit(state.LastBlockHeight)
+	}
+
+	if commit == nil {
 		panic(fmt.Sprintf(
-			"failed to reconstruct last commit; seen commit for height %v not found",
+			"failed to reconstruct last commit; commit for height %v not found",
 			state.LastBlockHeight,
 		))
 	}
 
-	lastPrecommits := types.CommitToVoteSet(state.ChainID, seenCommit, state.LastValidators)
+	lastPrecommits := types.CommitToVoteSet(state.ChainID, commit, state.LastValidators)
 	if !lastPrecommits.HasTwoThirdsMajority() {
 		panic("failed to reconstruct last commit; does not have +2/3 maj")
 	}
@@ -2218,6 +2229,7 @@ func (cs *State) signVote(
 
 	err := cs.privValidator.SignVote(ctx, cs.state.ChainID, v)
 	vote.Signature = v.Signature
+	vote.Timestamp = v.Timestamp
 
 	return vote, err
 }
@@ -2317,7 +2329,7 @@ func (cs *State) checkDoubleSigningRisk(height int64) error {
 		}
 
 		for i := int64(1); i < doubleSignCheckHeight; i++ {
-			lastCommit := cs.blockStore.LoadSeenCommit(height - i)
+			lastCommit := cs.LoadCommit(height - i)
 			if lastCommit != nil {
 				for sigIdx, s := range lastCommit.Signatures {
 					if s.BlockIDFlag == types.BlockIDFlagCommit && bytes.Equal(s.ValidatorAddress, valAddr) {

internal/consensus/state_test.go

@@ -1939,6 +1939,30 @@ func TestStateOutputVoteStats(t *testing.T) {
 
 }
 
+func TestSignSameVoteTwice(t *testing.T) {
+	config := configSetup(t)
+
+	_, vss := randState(config, 2)
+
+	randBytes := tmrand.Bytes(tmhash.Size)
+
+	vote := signVote(vss[1],
+		config,
+		tmproto.PrecommitType,
+		randBytes,
+		types.PartSetHeader{Total: 10, Hash: randBytes},
+	)
+
+	vote2 := signVote(vss[1],
+		config,
+		tmproto.PrecommitType,
+		randBytes,
+		types.PartSetHeader{Total: 10, Hash: randBytes},
+	)
+
+	require.Equal(t, vote, vote2)
+}
+
 // subscribe subscribes test client to the given query and returns a channel with cap = 1.
 func subscribe(eventBus *types.EventBus, q tmpubsub.Query) <-chan tmpubsub.Message {
 	sub, err := eventBus.Subscribe(context.Background(), testSubscriber, q)

internal/evidence/services.go

@@ -4,7 +4,7 @@ import (
 	"github.com/tendermint/tendermint/types"
 )
 
-//go:generate mockery --case underscore --name BlockStore
+//go:generate ../../scripts/mockery_generate.sh BlockStore
 
 type BlockStore interface {
 	LoadBlockMeta(height int64) *types.BlockMeta

internal/libs/clist/clist_property_test.go

@@ -0,0 +1,72 @@
+package clist_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"pgregory.net/rapid"
+
+	"github.com/tendermint/tendermint/internal/libs/clist"
+)
+
+func TestCListProperties(t *testing.T) {
+	rapid.Check(t, rapid.Run(&clistModel{}))
+}
+
+// clistModel is used by the rapid state machine testing framework.
+// clistModel contains both the clist that is being tested and a slice of *clist.CElements
+// that will be used to model the expected clist behavior.
+type clistModel struct {
+	clist *clist.CList
+
+	model []*clist.CElement
+}
+
+// Init is a method used by the rapid state machine testing library.
+// Init is called when the test starts to initialize the data that will be used
+// in the state machine test.
+func (m *clistModel) Init(t *rapid.T) {
+	m.clist = clist.New()
+	m.model = []*clist.CElement{}
+}
+
+// PushBack defines an action that will be randomly selected across by the rapid state
+// machines testing library. Every call to PushBack calls PushBack on the clist and
+// performs a similar action on the model data.
+func (m *clistModel) PushBack(t *rapid.T) {
+	value := rapid.String().Draw(t, "value").(string)
+	el := m.clist.PushBack(value)
+	m.model = append(m.model, el)
+}
+
+// Remove defines an action that will be randomly selected across by the rapid state
+// machine testing library. Every call to Remove selects an element from the model
+// and calls Remove on the CList with that element. The same element is removed from
+// the model to keep the objects in sync.
+func (m *clistModel) Remove(t *rapid.T) {
+	if len(m.model) == 0 {
+		return
+	}
+	ix := rapid.IntRange(0, len(m.model)-1).Draw(t, "index").(int)
+	value := m.model[ix]
+	m.model = append(m.model[:ix], m.model[ix+1:]...)
+	m.clist.Remove(value)
+}
+
+// Check is a method required by the rapid state machine testing library.
+// Check is run after each action and is used to verify that the state of the object,
+// in this case a clist.CList matches the state of the objec.
+func (m *clistModel) Check(t *rapid.T) {
+	require.Equal(t, len(m.model), m.clist.Len())
+	if len(m.model) == 0 {
+		return
+	}
+	require.Equal(t, m.model[0], m.clist.Front())
+	require.Equal(t, m.model[len(m.model)-1], m.clist.Back())
+
+	iter := m.clist.Front()
+	for _, val := range m.model {
+		require.Equal(t, val, iter)
+		iter = iter.Next()
+	}
+}

internal/libs/progressbar/progressbar.go

@@ -0,0 +1,41 @@
+package progressbar
+
+import "fmt"
+
+// the progressbar indicates the current status and progress would be desired.
+// ref: https://www.pixelstech.net/article/1596946473-A-simple-example-on-implementing-progress-bar-in-GoLang
+
+type Bar struct {
+	percent int64  // progress percentage
+	cur     int64  // current progress
+	start   int64  // the init starting value for progress
+	total   int64  // total value for progress
+	rate    string // the actual progress bar to be printed
+	graph   string // the fill value for progress bar
+}
+
+func (bar *Bar) NewOption(start, total int64) {
+	bar.cur = start
+	bar.start = start
+	bar.total = total
+	bar.graph = "█"
+	bar.percent = bar.getPercent()
+}
+
+func (bar *Bar) getPercent() int64 {
+	return int64(float32(bar.cur-bar.start) / float32(bar.total-bar.start) * 100)
+}
+
+func (bar *Bar) Play(cur int64) {
+	bar.cur = cur
+	last := bar.percent
+	bar.percent = bar.getPercent()
+	if bar.percent != last && bar.percent%2 == 0 {
+		bar.rate += bar.graph
+	}
+	fmt.Printf("\r[%-50s]%3d%% %8d/%d", bar.rate, bar.percent, bar.cur, bar.total)
+}
+
+func (bar *Bar) Finish() {
+	fmt.Println()
+}

internal/libs/progressbar/progressbar_test.go

@@ -0,0 +1,41 @@
+package progressbar
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestProgressBar(t *testing.T) {
+	zero := int64(0)
+	hundred := int64(100)
+
+	var bar Bar
+	bar.NewOption(zero, hundred)
+
+	require.Equal(t, zero, bar.start)
+	require.Equal(t, zero, bar.cur)
+	require.Equal(t, hundred, bar.total)
+	require.Equal(t, zero, bar.percent)
+	require.Equal(t, "█", bar.graph)
+	require.Equal(t, "", bar.rate)
+
+	defer bar.Finish()
+	for i := zero; i <= hundred; i++ {
+		time.Sleep(1 * time.Millisecond)
+		bar.Play(i)
+	}
+
+	require.Equal(t, zero, bar.start)
+	require.Equal(t, hundred, bar.cur)
+	require.Equal(t, hundred, bar.total)
+	require.Equal(t, hundred, bar.percent)
+
+	var rate string
+	for i := zero; i < hundred/2; i++ {
+		rate += "█"
+	}
+
+	require.Equal(t, rate, bar.rate)
+}

internal/mempool/v1/mempool.go

@@ -188,8 +188,8 @@ func (txmp *TxMempool) WaitForNextTx() <-chan struct{} {
 // NextGossipTx returns the next valid transaction to gossip. A caller must wait
 // for WaitForNextTx to signal a transaction is available to gossip first. It is
 // thread-safe.
-func (txmp *TxMempool) NextGossipTx() *WrappedTx {
-	return txmp.gossipIndex.Front().Value.(*WrappedTx)
+func (txmp *TxMempool) NextGossipTx() *clist.CElement {
+	return txmp.gossipIndex.Front()
 }
 
 // EnableTxsAvailable enables the mempool to trigger events when transactions

internal/mempool/v1/reactor.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	cfg "github.com/tendermint/tendermint/config"
+	"github.com/tendermint/tendermint/internal/libs/clist"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
 	"github.com/tendermint/tendermint/internal/mempool"
 	"github.com/tendermint/tendermint/internal/p2p"
@@ -53,6 +54,10 @@ type Reactor struct {
 	// goroutines.
 	peerWG sync.WaitGroup
 
+	// observePanic is a function for observing panics that were recovered in methods on
+	// Reactor. observePanic is called with the recovered value.
+	observePanic func(interface{})
+
 	mtx          tmsync.Mutex
 	peerRoutines map[types.NodeID]*tmsync.Closer
 }
@@ -76,12 +81,15 @@ func NewReactor(
 		peerUpdates:  peerUpdates,
 		closeCh:      make(chan struct{}),
 		peerRoutines: make(map[types.NodeID]*tmsync.Closer),
+		observePanic: defaultObservePanic,
 	}
 
 	r.BaseService = *service.NewBaseService(logger, "Mempool", r)
 	return r
 }
 
+func defaultObservePanic(r interface{}) {}
+
 // 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,6 +195,7 @@ 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 {
+			r.observePanic(e)
 			err = fmt.Errorf("panic in processing message: %v", e)
 			r.Logger.Error(
 				"recovering from processing message panic",
@@ -306,7 +315,7 @@ func (r *Reactor) processPeerUpdates() {
 
 func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer) {
 	peerMempoolID := r.ids.GetForPeer(peerID)
-	var memTx *WrappedTx
+	var nextGossipTx *clist.CElement
 
 	// remove the peer ID from the map of routines and mark the waitgroup as done
 	defer func() {
@@ -317,6 +326,7 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer)
 		r.peerWG.Done()
 
 		if e := recover(); e != nil {
+			r.observePanic(e)
 			r.Logger.Error(
 				"recovering from broadcasting mempool loop",
 				"err", e,
@@ -333,10 +343,10 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer)
 		// This happens because the CElement we were looking at got garbage
 		// collected (removed). That is, .NextWait() returned nil. Go ahead and
 		// start from the beginning.
-		if memTx == nil {
+		if nextGossipTx == nil {
 			select {
 			case <-r.mempool.WaitForNextTx(): // wait until a tx is available
-				if memTx = r.mempool.NextGossipTx(); memTx == nil {
+				if nextGossipTx = r.mempool.NextGossipTx(); nextGossipTx == nil {
 					continue
 				}
 
@@ -352,6 +362,8 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer)
 			}
 		}
 
+		memTx := nextGossipTx.Value.(*WrappedTx)
+
 		if r.peerMgr != nil {
 			height := r.peerMgr.GetHeight(peerID)
 			if height > 0 && height < memTx.height-1 {
@@ -380,16 +392,8 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer)
 		}
 
 		select {
-		case <-memTx.gossipEl.NextWaitChan():
-			// If there is a next element in gossip index, we point memTx to that node's
-			// value, otherwise we reset memTx to nil which will be checked at the
-			// parent for loop.
-			next := memTx.gossipEl.Next()
-			if next != nil {
-				memTx = next.Value.(*WrappedTx)
-			} else {
-				memTx = nil
-			}
+		case <-nextGossipTx.NextWaitChan():
+			nextGossipTx = nextGossipTx.Next()
 
 		case <-closer.Done():
 			// The peer is marked for removal via a PeerUpdate as the doneCh was

internal/mempool/v1/reactor_test.go

@@ -0,0 +1,147 @@
+package v1
+
+import (
+	"os"
+	"strings"
+	"sync"
+	"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)
+
+	observePanic := func(r interface{}) {
+		t.Fatal("panic detected in reactor")
+	}
+
+	primary := rts.nodes[0]
+	secondary := rts.nodes[1]
+	primaryReactor := rts.reactors[primary]
+	primaryMempool := primaryReactor.mempool
+	secondaryReactor := rts.reactors[secondary]
+
+	primaryReactor.observePanic = observePanic
+	secondaryReactor.observePanic = observePanic
+
+	firstTx := &WrappedTx{}
+	primaryMempool.insertTx(firstTx)
+
+	// run the router
+	rts.start(t)
+
+	closer := tmsync.NewCloser()
+	primaryReactor.peerWG.Add(1)
+	go primaryReactor.broadcastTxRoutine(secondary, closer)
+
+	wg := &sync.WaitGroup{}
+	for i := 0; i < 50; i++ {
+		next := &WrappedTx{}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			primaryMempool.insertTx(next)
+		}()
+	}
+
+	err := primaryReactor.Stop()
+	require.NoError(t, err)
+	primaryReactor.peerWG.Wait()
+	wg.Wait()
+}

internal/p2p/mocks/connection.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery. DO NOT EDIT.
 
 package mocks
 

internal/p2p/mocks/peer.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery. DO NOT EDIT.
 
 package mocks
 

internal/p2p/mocks/transport.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery. DO NOT EDIT.
 
 package mocks
 

internal/p2p/peer.go

@@ -14,7 +14,7 @@ import (
 	"github.com/tendermint/tendermint/types"
 )
 
-//go:generate mockery --case underscore --name Peer
+//go:generate ../../scripts/mockery_generate.sh Peer
 
 const metricsTickerDuration = 10 * time.Second
 

internal/p2p/peermanager.go

@@ -385,7 +385,7 @@ func (m *PeerManager) prunePeers() error {
 		peerID := ranked[i].ID
 		switch {
 		case m.store.Size() <= int(m.options.MaxPeers):
-			break
+			return nil
 		case m.dialing[peerID]:
 		case m.connected[peerID]:
 		default:

internal/p2p/pqueue.go

@@ -257,7 +257,11 @@ func (s *pqScheduler) process() {
 				s.metrics.PeerSendBytesTotal.With(
 					"chID", chIDStr,
 					"peer_id", string(pqEnv.envelope.To)).Add(float64(pqEnv.size))
-				s.dequeueCh <- pqEnv.envelope
+				select {
+				case s.dequeueCh <- pqEnv.envelope:
+				case <-s.closer.Done():
+					return
+				}
 			}
 
 		case <-s.closer.Done():

internal/p2p/pqueue_test.go

@@ -0,0 +1,39 @@
+package p2p
+
+import (
+	"testing"
+	"time"
+
+	"github.com/tendermint/tendermint/libs/log"
+)
+
+func TestCloseWhileDequeueFull(t *testing.T) {
+	enqueueLength := 5
+	chDescs := []ChannelDescriptor{
+		{ID: 0x01, Priority: 1, MaxSendBytes: 4},
+	}
+	pqueue := newPQScheduler(log.NewNopLogger(), NopMetrics(), chDescs, uint(enqueueLength), 1, 120)
+
+	for i := 0; i < enqueueLength; i++ {
+		pqueue.enqueue() <- Envelope{
+			channelID: 0x01,
+			Message:   &testMessage{Value: "foo"}, // 5 bytes
+		}
+	}
+
+	go pqueue.process()
+
+	// sleep to allow context switch for process() to run
+	time.Sleep(10 * time.Millisecond)
+	doneCh := make(chan struct{})
+	go func() {
+		pqueue.close()
+		close(doneCh)
+	}()
+
+	select {
+	case <-doneCh:
+	case <-time.After(2 * time.Second):
+		t.Fatal("pqueue failed to close")
+	}
+}

internal/p2p/transport.go

@@ -12,7 +12,7 @@ import (
 	"github.com/tendermint/tendermint/version"
 )
 
-//go:generate mockery --case underscore --name Transport|Connection
+//go:generate ../../scripts/mockery_generate.sh Transport|Connection
 
 const (
 	// defaultProtocol is the default protocol used for NodeAddress when

internal/statesync/mocks/state_provider.go

@@ -1,4 +1,4 @@
-// Code generated by mockery 2.7.5. DO NOT EDIT.
+// Code generated by mockery. DO NOT EDIT.
 
 package mocks
 

internal/statesync/stateprovider.go

@@ -20,7 +20,7 @@ import (
 	"github.com/tendermint/tendermint/types"
 )
 
-//go:generate mockery --case underscore --name StateProvider
+//go:generate ../../scripts/mockery_generate.sh StateProvider
 
 // StateProvider is a provider of trusted state data for bootstrapping a node. This refers
 // to the state.State object, not the state machine.