Merge branch 'v0.35.x' into callum/mempool-full-error

internal/consensus/mocks/cons_sync_reactor.go

@@ -26,3 +26,18 @@ func (_m *ConsSyncReactor) SetStateSyncingMetrics(_a0 float64) {
 func (_m *ConsSyncReactor) SwitchToConsensus(_a0 state.State, _a1 bool) {
 	_m.Called(_a0, _a1)
 }
+
+type NewConsSyncReactorT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewConsSyncReactor creates a new instance of ConsSyncReactor. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewConsSyncReactor(t NewConsSyncReactorT) *ConsSyncReactor {
+	mock := &ConsSyncReactor{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/evidence/mocks/block_store.go

@@ -57,3 +57,18 @@ func (_m *BlockStore) LoadBlockMeta(height int64) *types.BlockMeta {
 
 	return r0
 }
+
+type NewBlockStoreT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewBlockStore creates a new instance of BlockStore. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewBlockStore(t NewBlockStoreT) *BlockStore {
+	mock := &BlockStore{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/mempool/v0/clist_mempool.go

@@ -242,17 +242,13 @@ func (mem *CListMempool) CheckTx(
 		// so we only record the sender for txs still in the mempool.
 		if e, ok := mem.txsMap.Load(tx.Key()); ok {
 			memTx := e.(*clist.CElement).Value.(*mempoolTx)
-			_, loaded := memTx.senders.LoadOrStore(txInfo.SenderID, true)
+			memTx.senders.LoadOrStore(txInfo.SenderID, true)
 			// TODO: consider punishing peer for dups,
 			// its non-trivial since invalid txs can become valid,
 			// but they can spam the same tx with little cost to them atm.
-			if loaded {
-				return types.ErrTxInCache
-			}
 		}
 
-		mem.logger.Debug("tx exists already in cache", "tx_hash", tx.Hash())
-		return nil
+		return types.ErrTxInCache
 	}
 
 	if ctx == nil {

internal/mempool/v0/clist_mempool_test.go

@@ -200,7 +200,7 @@ func TestMempoolUpdate(t *testing.T) {
 		err := mp.Update(1, []types.Tx{[]byte{0x01}}, abciResponses(1, abci.CodeTypeOK), nil, nil)
 		require.NoError(t, err)
 		err = mp.CheckTx(context.Background(), []byte{0x01}, nil, mempool.TxInfo{})
-		require.NoError(t, err)
+		assert.Error(t, err)
 	}
 
 	// 2. Removes valid txs from the mempool
@@ -305,11 +305,15 @@ func TestMempool_KeepInvalidTxsInCache(t *testing.T) {
 
 		// a must be added to the cache
 		err = mp.CheckTx(context.Background(), a, nil, mempool.TxInfo{})
-		require.NoError(t, err)
+		if assert.Error(t, err) {
+			assert.Equal(t, types.ErrTxInCache, err)
+		}
 
 		// b must remain in the cache
 		err = mp.CheckTx(context.Background(), b, nil, mempool.TxInfo{})
-		require.NoError(t, err)
+		if assert.Error(t, err) {
+			assert.Equal(t, types.ErrTxInCache, err)
+		}
 	}
 
 	// 2. An invalid transaction must remain in the cache

internal/p2p/metrics.go

@@ -53,6 +53,9 @@ type Metrics struct {
 	// this node.
 	PeersConnectedIncoming metrics.Gauge
 
+	// Number of peers evicted by this node.
+	PeersEvicted metrics.Counter
+
 	// RouterPeerQueueRecv defines the time taken to read off of a peer's queue
 	// before sending on the connection.
 	RouterPeerQueueRecv metrics.Histogram
@@ -110,6 +113,12 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Name:      "peers_connected_success",
 			Help:      "Number of successful peer connection attempts",
 		}, labels).With(labelsAndValues...),
+		PeersEvicted: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "peers_evicted",
+			Help:      "Number of connected peers evicted",
+		}, labels).With(labelsAndValues...),
 		PeersConnectedFailure: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
 			Namespace: namespace,
 			Subsystem: MetricsSubsystem,
@@ -200,6 +209,7 @@ func NopMetrics() *Metrics {
 		PeersConnectedIncoming: discard.NewGauge(),
 		PeersConnectedOutgoing: discard.NewGauge(),
 		PeersInactivated:       discard.NewGauge(),
+		PeersEvicted:           discard.NewCounter(),
 		PeerReceiveBytesTotal:  discard.NewCounter(),
 		PeerSendBytesTotal:     discard.NewCounter(),
 		PeerPendingSendBytes:   discard.NewGauge(),

internal/p2p/mocks/connection.go

@@ -13,6 +13,8 @@ import (
 
 	p2p "github.com/tendermint/tendermint/internal/p2p"
 
+	time "time"
+
 	types "github.com/tendermint/tendermint/types"
 )
 
@@ -49,20 +51,20 @@ func (_m *Connection) FlushClose() error {
 	return r0
 }
 
-// Handshake provides a mock function with given fields: _a0, _a1, _a2
-func (_m *Connection) Handshake(_a0 context.Context, _a1 types.NodeInfo, _a2 crypto.PrivKey) (types.NodeInfo, crypto.PubKey, error) {
-	ret := _m.Called(_a0, _a1, _a2)
+// Handshake provides a mock function with given fields: _a0, _a1, _a2, _a3
+func (_m *Connection) Handshake(_a0 context.Context, _a1 time.Duration, _a2 types.NodeInfo, _a3 crypto.PrivKey) (types.NodeInfo, crypto.PubKey, error) {
+	ret := _m.Called(_a0, _a1, _a2, _a3)
 
 	var r0 types.NodeInfo
-	if rf, ok := ret.Get(0).(func(context.Context, types.NodeInfo, crypto.PrivKey) types.NodeInfo); ok {
-		r0 = rf(_a0, _a1, _a2)
+	if rf, ok := ret.Get(0).(func(context.Context, time.Duration, types.NodeInfo, crypto.PrivKey) types.NodeInfo); ok {
+		r0 = rf(_a0, _a1, _a2, _a3)
 	} else {
 		r0 = ret.Get(0).(types.NodeInfo)
 	}
 
 	var r1 crypto.PubKey
-	if rf, ok := ret.Get(1).(func(context.Context, types.NodeInfo, crypto.PrivKey) crypto.PubKey); ok {
-		r1 = rf(_a0, _a1, _a2)
+	if rf, ok := ret.Get(1).(func(context.Context, time.Duration, types.NodeInfo, crypto.PrivKey) crypto.PubKey); ok {
+		r1 = rf(_a0, _a1, _a2, _a3)
 	} else {
 		if ret.Get(1) != nil {
 			r1 = ret.Get(1).(crypto.PubKey)
@@ -70,8 +72,8 @@ func (_m *Connection) Handshake(_a0 context.Context, _a1 types.NodeInfo, _a2 cry
 	}
 
 	var r2 error
-	if rf, ok := ret.Get(2).(func(context.Context, types.NodeInfo, crypto.PrivKey) error); ok {
-		r2 = rf(_a0, _a1, _a2)
+	if rf, ok := ret.Get(2).(func(context.Context, time.Duration, types.NodeInfo, crypto.PrivKey) error); ok {
+		r2 = rf(_a0, _a1, _a2, _a3)
 	} else {
 		r2 = ret.Error(2)
 	}
@@ -206,3 +208,18 @@ func (_m *Connection) TrySendMessage(_a0 p2p.ChannelID, _a1 []byte) (bool, error
 
 	return r0, r1
 }
+
+type NewConnectionT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewConnection creates a new instance of Connection. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewConnection(t NewConnectionT) *Connection {
+	mock := &Connection{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/p2p/mocks/peer.go

@@ -332,3 +332,18 @@ func (_m *Peer) TrySend(_a0 byte, _a1 []byte) bool {
 func (_m *Peer) Wait() {
 	_m.Called()
 }
+
+type NewPeerT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewPeer creates a new instance of Peer. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewPeer(t NewPeerT) *Peer {
+	mock := &Peer{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/p2p/mocks/transport.go

@@ -119,3 +119,18 @@ func (_m *Transport) String() string {
 
 	return r0
 }
+
+type NewTransportT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewTransport creates a new instance of Transport. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewTransport(t NewTransportT) *Transport {
+	mock := &Transport{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/p2p/p2ptest/network.go

@@ -1,7 +1,6 @@
 package p2ptest
 
 import (
-	"context"
 	"math/rand"
 	"testing"
 	"time"
@@ -254,7 +253,7 @@ func (n *Network) MakeNode(t *testing.T, opts NodeOptions) *Node {
 		privKey,
 		peerManager,
 		[]p2p.Transport{transport},
-		p2p.RouterOptions{DialSleep: func(_ context.Context) {}},
+		p2p.RouterOptions{},
 	)
 	require.NoError(t, err)
 	require.NoError(t, router.Start())

internal/p2p/peer_test.go

@@ -90,7 +90,7 @@ func createOutboundPeerAndPerformHandshake(
 	if err != nil {
 		return nil, err
 	}
-	peerInfo, _, err := pc.conn.Handshake(context.Background(), ourNodeInfo, pk)
+	peerInfo, _, err := pc.conn.Handshake(context.Background(), 0, ourNodeInfo, pk)
 	if err != nil {
 		return nil, err
 	}
@@ -187,7 +187,7 @@ func (rp *remotePeer) Dial(addr *NetAddress) (net.Conn, error) {
 	if err != nil {
 		return nil, err
 	}
-	_, _, err = pc.conn.Handshake(context.Background(), rp.nodeInfo(), rp.PrivKey)
+	_, _, err = pc.conn.Handshake(context.Background(), 0, rp.nodeInfo(), rp.PrivKey)
 	if err != nil {
 		return nil, err
 	}
@@ -213,7 +213,7 @@ func (rp *remotePeer) accept() {
 		if err != nil {
 			golog.Printf("Failed to create a peer: %+v", err)
 		}
-		_, _, err = pc.conn.Handshake(context.Background(), rp.nodeInfo(), rp.PrivKey)
+		_, _, err = pc.conn.Handshake(context.Background(), 0, rp.nodeInfo(), rp.PrivKey)
 		if err != nil {
 			golog.Printf("Failed to handshake a peer: %+v", err)
 		}

internal/p2p/peermanager.go

@@ -162,6 +162,10 @@ type PeerManagerOptions struct {
 	// retry times, to avoid thundering herds. 0 disables jitter.
 	RetryTimeJitter time.Duration
 
+	// DisconnectCooldownPeriod is the amount of time after we
+	// disconnect from a peer before we'll consider dialing a new peer
+	DisconnectCooldownPeriod time.Duration
+
 	// PeerScores sets fixed scores for specific peers. It is mainly used
 	// for testing. A score of 0 is ignored.
 	PeerScores map[types.NodeID]PeerScore
@@ -534,12 +538,13 @@ func (m *PeerManager) HasDialedMaxPeers() bool {
 // returned peer.
 func (m *PeerManager) DialNext(ctx context.Context) (NodeAddress, error) {
 	for {
-		address, err := m.TryDialNext()
-		if err != nil || (address != NodeAddress{}) {
-			return address, err
+		if address := m.TryDialNext(); (address != NodeAddress{}) {
+			return address, nil
 		}
+
 		select {
 		case <-m.dialWaker.Sleep():
+			continue
 		case <-ctx.Done():
 			return NodeAddress{}, ctx.Err()
 		}
@@ -548,21 +553,20 @@ func (m *PeerManager) DialNext(ctx context.Context) (NodeAddress, error) {
 
 // TryDialNext is equivalent to DialNext(), but immediately returns an empty
 // address if no peers or connection slots are available.
-func (m *PeerManager) TryDialNext() (NodeAddress, error) {
+func (m *PeerManager) TryDialNext() NodeAddress {
 	m.mtx.Lock()
 	defer m.mtx.Unlock()
 
 	// We allow dialing MaxConnected+MaxConnectedUpgrade peers. Including
 	// MaxConnectedUpgrade allows us to probe additional peers that have a
 	// higher score than any other peers, and if successful evict it.
-	if m.options.MaxConnected > 0 && len(m.connected)+len(m.dialing) >=
-		int(m.options.MaxConnected)+int(m.options.MaxConnectedUpgrade) {
-		return NodeAddress{}, nil
+	if m.options.MaxConnected > 0 && len(m.connected)+len(m.dialing) >= int(m.options.MaxConnected)+int(m.options.MaxConnectedUpgrade) {
+		return NodeAddress{}
 	}
 
 	cinfo := m.getConnectedInfo()
 	if m.options.MaxOutgoingConnections > 0 && cinfo.outgoing >= m.options.MaxOutgoingConnections {
-		return NodeAddress{}, nil
+		return NodeAddress{}
 	}
 
 	for _, peer := range m.store.Ranked() {
@@ -570,11 +574,19 @@ func (m *PeerManager) TryDialNext() (NodeAddress, error) {
 			continue
 		}
 
+		if !peer.LastDisconnected.IsZero() && time.Since(peer.LastDisconnected) < m.options.DisconnectCooldownPeriod {
+			continue
+		}
+
 		for _, addressInfo := range peer.AddressInfo {
 			if time.Since(addressInfo.LastDialFailure) < m.retryDelay(addressInfo.DialFailures, peer.Persistent) {
 				continue
 			}
 
+			if id, ok := m.store.Resolve(addressInfo.Address); ok && (m.isConnected(id) || m.dialing[id]) {
+				continue
+			}
+
 			// We now have an eligible address to dial. If we're full but have
 			// upgrade capacity (as checked above), we find a lower-scored peer
 			// we can replace and mark it as upgrading so noone else claims it.
@@ -585,16 +597,16 @@ func (m *PeerManager) TryDialNext() (NodeAddress, error) {
 			if m.options.MaxConnected > 0 && len(m.connected) >= int(m.options.MaxConnected) {
 				upgradeFromPeer := m.findUpgradeCandidate(peer.ID, peer.Score())
 				if upgradeFromPeer == "" {
-					return NodeAddress{}, nil
+					return NodeAddress{}
 				}
 				m.upgrading[upgradeFromPeer] = peer.ID
 			}
 
 			m.dialing[peer.ID] = true
-			return addressInfo.Address, nil
+			return addressInfo.Address
 		}
 	}
-	return NodeAddress{}, nil
+	return NodeAddress{}
 }
 
 // DialFailed reports a failed dial attempt. This will make the peer available
@@ -702,8 +714,7 @@ func (m *PeerManager) Dialed(address NodeAddress) error {
 		return err
 	}
 
-	if upgradeFromPeer != "" && m.options.MaxConnected > 0 &&
-		len(m.connected) >= int(m.options.MaxConnected) {
+	if upgradeFromPeer != "" && m.options.MaxConnected > 0 && len(m.connected) >= int(m.options.MaxConnected) {
 		// Look for an even lower-scored peer that may have appeared since we
 		// started the upgrade.
 		if p, ok := m.store.Get(upgradeFromPeer); ok {
@@ -712,11 +723,11 @@ func (m *PeerManager) Dialed(address NodeAddress) error {
 			}
 		}
 		m.evict[upgradeFromPeer] = true
+		m.evictWaker.Wake()
 	}
 
 	m.metrics.PeersConnectedOutgoing.Add(1)
 	m.connected[peer.ID] = peerConnectionOutgoing
-	m.evictWaker.Wake()
 
 	return nil
 }
@@ -885,6 +896,22 @@ func (m *PeerManager) Disconnected(peerID types.NodeID) {
 	delete(m.evicting, peerID)
 	delete(m.ready, peerID)
 
+	if peer, ok := m.store.Get(peerID); ok {
+		peer.LastDisconnected = time.Now()
+		_ = m.store.Set(peer)
+		// launch a thread to ping the dialWaker when the
+		// disconnected peer can be dialed again.
+		go func() {
+			timer := time.NewTimer(m.options.DisconnectCooldownPeriod)
+			defer timer.Stop()
+			select {
+			case <-timer.C:
+				m.dialWaker.Wake()
+			case <-m.closeCh:
+			}
+		}()
+	}
+
 	if ready {
 		m.broadcast(PeerUpdate{
 			NodeID: peerID,
@@ -1263,6 +1290,7 @@ func (m *PeerManager) retryDelay(failures uint32, persistent bool) time.Duration
 type peerStore struct {
 	db     dbm.DB
 	peers  map[types.NodeID]*peerInfo
+	index  map[NodeAddress]types.NodeID
 	ranked []*peerInfo // cache for Ranked(), nil invalidates cache
 }
 
@@ -1282,6 +1310,7 @@ func newPeerStore(db dbm.DB) (*peerStore, error) {
 // loadPeers loads all peers from the database into memory.
 func (s *peerStore) loadPeers() error {
 	peers := map[types.NodeID]*peerInfo{}
+	addrs := map[NodeAddress]types.NodeID{}
 
 	start, end := keyPeerInfoRange()
 	iter, err := s.db.Iterator(start, end)
@@ -1301,11 +1330,18 @@ func (s *peerStore) loadPeers() error {
 			return fmt.Errorf("invalid peer data: %w", err)
 		}
 		peers[peer.ID] = peer
+		for addr := range peer.AddressInfo {
+			// TODO maybe check to see if we've seen this
+			// addr before for a different peer, there
+			// could be duplicates.
+			addrs[addr] = peer.ID
+		}
 	}
 	if iter.Error() != nil {
 		return iter.Error()
 	}
 	s.peers = peers
+	s.index = addrs
 	s.ranked = nil // invalidate cache if populated
 	return nil
 }
@@ -1317,6 +1353,12 @@ func (s *peerStore) Get(id types.NodeID) (peerInfo, bool) {
 	return peer.Copy(), ok
 }
 
+// Resolve returns the peer ID for a given node address if known.
+func (s *peerStore) Resolve(addr NodeAddress) (types.NodeID, bool) {
+	id, ok := s.index[addr]
+	return id, ok
+}
+
 // Set stores peer data. The input data will be copied, and can safely be reused
 // by the caller.
 func (s *peerStore) Set(peer peerInfo) error {
@@ -1345,20 +1387,29 @@ func (s *peerStore) Set(peer peerInfo) error {
 		// update the existing pointer address.
 		*current = peer
 	}
+	for addr := range peer.AddressInfo {
+		s.index[addr] = peer.ID
+	}
 
 	return nil
 }
 
 // Delete deletes a peer, or does nothing if it does not exist.
 func (s *peerStore) Delete(id types.NodeID) error {
-	if _, ok := s.peers[id]; !ok {
+	peer, ok := s.peers[id]
+	if !ok {
 		return nil
 	}
-	if err := s.db.Delete(keyPeerInfo(id)); err != nil {
-		return err
+	for _, addr := range peer.AddressInfo {
+		delete(s.index, addr.Address)
 	}
 	delete(s.peers, id)
 	s.ranked = nil
+
+	if err := s.db.Delete(keyPeerInfo(id)); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -1447,9 +1498,10 @@ func (s *peerStore) Size() int {
 
 // peerInfo contains peer information stored in a peerStore.
 type peerInfo struct {
-	ID            types.NodeID
-	AddressInfo   map[NodeAddress]*peerAddressInfo
-	LastConnected time.Time
+	ID               types.NodeID
+	AddressInfo      map[NodeAddress]*peerAddressInfo
+	LastConnected    time.Time
+	LastDisconnected time.Time
 
 	// These fields are ephemeral, i.e. not persisted to the database.
 	Persistent bool
@@ -1489,8 +1541,8 @@ func peerInfoFromProto(msg *p2pproto.PeerInfo) (*peerInfo, error) {
 func (p *peerInfo) ToProto() *p2pproto.PeerInfo {
 	msg := &p2pproto.PeerInfo{
 		ID:            string(p.ID),
-		LastConnected: &p.LastConnected,
 		Inactive:      p.Inactive,
+		LastConnected: &p.LastConnected,
 	}
 	for _, addressInfo := range p.AddressInfo {
 		msg.AddressInfo = append(msg.AddressInfo, addressInfo.ToProto())
@@ -1498,6 +1550,7 @@ func (p *peerInfo) ToProto() *p2pproto.PeerInfo {
 	if msg.LastConnected.IsZero() {
 		msg.LastConnected = nil
 	}
+
 	return msg
 }
 

internal/p2p/peermanager_test.go

@@ -378,16 +378,14 @@ func TestPeerManager_DialNext_WakeOnDialFailed(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 
 	// Add b. We shouldn't be able to dial it, due to MaxConnected.
 	added, err = peerManager.Add(b)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 
 	// Spawn a goroutine to fail a's dial attempt.
@@ -415,8 +413,7 @@ func TestPeerManager_DialNext_WakeOnDialFailedRetry(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 	require.NoError(t, peerManager.DialFailed(dial))
 	failed := time.Now()
@@ -443,8 +440,7 @@ func TestPeerManager_DialNext_WakeOnDisconnected(t *testing.T) {
 	err = peerManager.Accepted(a.NodeID)
 	require.NoError(t, err)
 
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Zero(t, dial)
 
 	go func() {
@@ -473,8 +469,7 @@ func TestPeerManager_TryDialNext_MaxConnected(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 	require.NoError(t, peerManager.Dialed(a))
 
@@ -482,16 +477,14 @@ func TestPeerManager_TryDialNext_MaxConnected(t *testing.T) {
 	added, err = peerManager.Add(b)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, b, dial)
 
 	// At this point, adding c will not allow dialing it.
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 }
 
@@ -520,7 +513,7 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
+	dial := peerManager.TryDialNext()
 	require.NoError(t, err)
 	require.Equal(t, a, dial)
 	require.NoError(t, peerManager.Dialed(a))
@@ -529,8 +522,7 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
 	added, err = peerManager.Add(b)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, b, dial)
 
 	// Even though we are at capacity, we should be allowed to dial c for an
@@ -538,8 +530,7 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, c, dial)
 
 	// However, since we're using all upgrade slots now, we can't add and dial
@@ -547,16 +538,14 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
 	added, err = peerManager.Add(d)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 
 	// We go through with c's upgrade.
 	require.NoError(t, peerManager.Dialed(c))
 
 	// Still can't dial d.
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 
 	// Now, if we disconnect a, we should be allowed to dial d because we have a
@@ -572,8 +561,7 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
 	added, err = peerManager.Add(e)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 }
 
@@ -593,8 +581,7 @@ func TestPeerManager_TryDialNext_UpgradeReservesPeer(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 	require.NoError(t, peerManager.Dialed(a))
 
@@ -602,8 +589,7 @@ func TestPeerManager_TryDialNext_UpgradeReservesPeer(t *testing.T) {
 	added, err = peerManager.Add(b)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, b, dial)
 
 	// Adding c and dialing it will fail, because a is the only connected
@@ -611,8 +597,7 @@ func TestPeerManager_TryDialNext_UpgradeReservesPeer(t *testing.T) {
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Empty(t, dial)
 }
 
@@ -633,22 +618,19 @@ func TestPeerManager_TryDialNext_DialingConnected(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 
 	// Adding a's TCP address will not dispense a, since it's already dialing.
 	added, err = peerManager.Add(aTCP)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 
 	// Marking a as dialed will still not dispense it.
 	require.NoError(t, peerManager.Dialed(a))
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 
 	// Adding b and accepting a connection from it will not dispense it either.
@@ -656,8 +638,7 @@ func TestPeerManager_TryDialNext_DialingConnected(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, added)
 	require.NoError(t, peerManager.Accepted(bID))
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 }
 
@@ -683,16 +664,14 @@ func TestPeerManager_TryDialNext_Multiple(t *testing.T) {
 	// All addresses should be dispensed as long as dialing them has failed.
 	dial := []p2p.NodeAddress{}
 	for range addresses {
-		address, err := peerManager.TryDialNext()
-		require.NoError(t, err)
+		address := peerManager.TryDialNext()
 		require.NotZero(t, address)
 		require.NoError(t, peerManager.DialFailed(address))
 		dial = append(dial, address)
 	}
 	require.ElementsMatch(t, dial, addresses)
 
-	address, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	address := peerManager.TryDialNext()
 	require.Zero(t, address)
 }
 
@@ -714,15 +693,14 @@ func TestPeerManager_DialFailed(t *testing.T) {
 	// Dialing and then calling DialFailed with a different address (same
 	// NodeID) should unmark as dialing and allow us to dial the other address
 	// again, but not register the failed address.
-	dial, err := peerManager.TryDialNext()
+	dial := peerManager.TryDialNext()
 	require.NoError(t, err)
 	require.Equal(t, a, dial)
 	require.NoError(t, peerManager.DialFailed(p2p.NodeAddress{
 		Protocol: "tcp", NodeID: aID, Hostname: "localhost"}))
 	require.Equal(t, []p2p.NodeAddress{a}, peerManager.Addresses(aID))
 
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 
 	// Calling DialFailed on same address twice should be fine.
@@ -753,8 +731,7 @@ func TestPeerManager_DialFailed_UnreservePeer(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 	require.NoError(t, peerManager.Dialed(a))
 
@@ -762,8 +739,7 @@ func TestPeerManager_DialFailed_UnreservePeer(t *testing.T) {
 	added, err = peerManager.Add(b)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, b, dial)
 
 	// Adding c and dialing it will fail, even though it could upgrade a and we
@@ -772,14 +748,12 @@ func TestPeerManager_DialFailed_UnreservePeer(t *testing.T) {
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Empty(t, dial)
 
 	// Failing b's dial will now make c available for dialing.
 	require.NoError(t, peerManager.DialFailed(b))
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, c, dial)
 }
 
@@ -794,8 +768,7 @@ func TestPeerManager_Dialed_Connected(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 
 	require.NoError(t, peerManager.Dialed(a))
@@ -805,8 +778,7 @@ func TestPeerManager_Dialed_Connected(t *testing.T) {
 	added, err = peerManager.Add(b)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, b, dial)
 
 	require.NoError(t, peerManager.Accepted(b.NodeID))
@@ -835,8 +807,7 @@ func TestPeerManager_Dialed_MaxConnected(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 
 	// Marking b as dialed in the meanwhile (even without TryDialNext)
@@ -878,8 +849,7 @@ func TestPeerManager_Dialed_MaxConnectedUpgrade(t *testing.T) {
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, c, dial)
 	require.NoError(t, peerManager.Dialed(c))
 
@@ -923,8 +893,7 @@ func TestPeerManager_Dialed_Upgrade(t *testing.T) {
 	added, err = peerManager.Add(b)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, b, dial)
 	require.NoError(t, peerManager.Dialed(b))
 
@@ -933,8 +902,7 @@ func TestPeerManager_Dialed_Upgrade(t *testing.T) {
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Empty(t, dial)
 
 	// a should now be evicted.
@@ -977,8 +945,7 @@ func TestPeerManager_Dialed_UpgradeEvenLower(t *testing.T) {
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, c, dial)
 
 	// In the meanwhile, a disconnects and d connects. d is even lower-scored
@@ -1028,7 +995,7 @@ func TestPeerManager_Dialed_UpgradeNoEvict(t *testing.T) {
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
+	dial := peerManager.TryDialNext()
 	require.NoError(t, err)
 	require.Equal(t, c, dial)
 
@@ -1074,8 +1041,7 @@ func TestPeerManager_Accepted(t *testing.T) {
 	added, err = peerManager.Add(c)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, c, dial)
 	require.NoError(t, peerManager.Accepted(c.NodeID))
 	require.Error(t, peerManager.Dialed(c))
@@ -1084,8 +1050,7 @@ func TestPeerManager_Accepted(t *testing.T) {
 	added, err = peerManager.Add(d)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, d, dial)
 	require.NoError(t, peerManager.Dialed(d))
 	require.Error(t, peerManager.Accepted(d.NodeID))
@@ -1233,8 +1198,7 @@ func TestPeerManager_Accepted_UpgradeDialing(t *testing.T) {
 	added, err = peerManager.Add(b)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, b, dial)
 
 	// a has already been claimed as an upgrade of a, so accepting
@@ -1394,8 +1358,7 @@ func TestPeerManager_EvictNext_WakeOnUpgradeDialed(t *testing.T) {
 		added, err := peerManager.Add(b)
 		require.NoError(t, err)
 		require.True(t, added)
-		dial, err := peerManager.TryDialNext()
-		require.NoError(t, err)
+		dial := peerManager.TryDialNext()
 		require.Equal(t, b, dial)
 		require.NoError(t, peerManager.Dialed(b))
 	}()
@@ -1521,13 +1484,11 @@ func TestPeerManager_Disconnected(t *testing.T) {
 
 	// Disconnecting a dialing peer does not unmark it as dialing, to avoid
 	// dialing it multiple times in parallel.
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 
 	peerManager.Disconnected(a.NodeID)
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Zero(t, dial)
 }
 
@@ -1595,8 +1556,7 @@ func TestPeerManager_Subscribe(t *testing.T) {
 	require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusDown}, <-sub.Updates())
 
 	// Outbound connection with peer error and eviction.
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 	require.Empty(t, sub.Updates())
 
@@ -1619,8 +1579,7 @@ func TestPeerManager_Subscribe(t *testing.T) {
 	require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusDown}, <-sub.Updates())
 
 	// Outbound connection with dial failure.
-	dial, err = peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial = peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 	require.Empty(t, sub.Updates())
 
@@ -1716,8 +1675,7 @@ func TestPeerManager_Close(t *testing.T) {
 	added, err := peerManager.Add(a)
 	require.NoError(t, err)
 	require.True(t, added)
-	dial, err := peerManager.TryDialNext()
-	require.NoError(t, err)
+	dial := peerManager.TryDialNext()
 	require.Equal(t, a, dial)
 	require.NoError(t, peerManager.DialFailed(a))
 

internal/p2p/router.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"math/rand"
 	"net"
 	"runtime"
 	"sync"
@@ -160,12 +159,6 @@ type RouterOptions struct {
 	// return an error to reject the peer.
 	FilterPeerByID func(context.Context, types.NodeID) error
 
-	// DialSleep controls the amount of time that the router
-	// sleeps between dialing peers. If not set, a default value
-	// is used that sleeps for a (random) amount of time up to 3
-	// seconds between submitting each peer to be dialed.
-	DialSleep func(context.Context)
-
 	// NumConcrruentDials controls how many parallel go routines
 	// are used to dial peers. This defaults to the value of
 	// runtime.NumCPU.
@@ -291,7 +284,7 @@ func NewRouter(
 
 	router := &Router{
 		logger:   logger,
-		metrics:  metrics,
+		metrics:  NopMetrics(),
 		nodeInfo: nodeInfo,
 		privKey:  privKey,
 		connTracker: newConnTracker(
@@ -312,6 +305,10 @@ func NewRouter(
 
 	router.BaseService = service.NewBaseService(logger, "router", router)
 
+	if metrics != nil {
+		router.metrics = metrics
+	}
+
 	qf, err := router.createQueueFactory()
 	if err != nil {
 		return nil, err
@@ -422,11 +419,7 @@ func (r *Router) routeChannel(
 ) {
 	for {
 		select {
-		case envelope, ok := <-outCh:
-			if !ok {
-				return
-			}
-
+		case envelope := <-outCh:
 			// Mark the envelope with the channel ID to allow sendPeer() to pass
 			// it on to Transport.SendMessage().
 			envelope.channelID = chID
@@ -503,20 +496,22 @@ func (r *Router) routeChannel(
 				}
 			}
 
-		case peerError, ok := <-errCh:
-			if !ok {
-				return
-			}
-
-			shouldEvict := peerError.Fatal || r.peerManager.HasMaxPeerCapacity()
+		case peerError := <-errCh:
+			maxPeerCapacity := r.peerManager.HasMaxPeerCapacity()
 			r.logger.Error("peer error",
 				"peer", peerError.NodeID,
 				"err", peerError.Err,
-				"evicting", shouldEvict,
+				"disconnecting", peerError.Fatal || maxPeerCapacity,
 			)
-			if shouldEvict {
+
+			if peerError.Fatal || maxPeerCapacity {
+				// if the error is fatal or all peer
+				// slots are in use, we can error
+				// (disconnect) from the peer.
 				r.peerManager.Errored(peerError.NodeID, peerError.Err)
 			} else {
+				// this just decrements the peer
+				// score.
 				r.peerManager.processPeerEvent(PeerUpdate{
 					NodeID: peerError.NodeID,
 					Status: PeerStatusBad,
@@ -530,7 +525,7 @@ func (r *Router) routeChannel(
 
 func (r *Router) numConcurrentDials() int {
 	if r.options.NumConcurrentDials == nil {
-		return runtime.NumCPU()
+		return runtime.NumCPU() * 32
 	}
 
 	return r.options.NumConcurrentDials()
@@ -552,35 +547,6 @@ func (r *Router) filterPeersID(ctx context.Context, id types.NodeID) error {
 	return r.options.FilterPeerByID(ctx, id)
 }
 
-func (r *Router) dialSleep(ctx context.Context) {
-	if r.options.DialSleep == nil {
-		const (
-			maxDialerInterval = 500
-			minDialerInterval = 100
-		)
-
-		// nolint:gosec // G404: Use of weak random number generator
-		dur := time.Duration(rand.Int63n(maxDialerInterval-minDialerInterval+1) + minDialerInterval)
-
-		timer := time.NewTimer(dur * time.Millisecond)
-		defer timer.Stop()
-
-		select {
-		case <-ctx.Done():
-		case <-timer.C:
-		}
-
-		return
-	}
-
-	r.options.DialSleep(ctx)
-
-	if !r.peerManager.HasDialedMaxPeers() {
-		r.peerManager.dialWaker.Wake()
-	}
-
-}
-
 // acceptPeers accepts inbound connections from peers on the given transport,
 // and spawns goroutines that route messages to/from them.
 func (r *Router) acceptPeers(transport Transport) {
@@ -702,19 +668,13 @@ LOOP:
 		case errors.Is(err, context.Canceled):
 			r.logger.Debug("stopping dial routine")
 			break LOOP
-		case err != nil:
-			r.logger.Error("failed to find next peer to dial", "err", err)
-			break LOOP
+		case address == NodeAddress{}:
+			continue LOOP
 		}
 
 		select {
 		case addresses <- address:
-			// this jitters the frequency that we call
-			// DialNext and prevents us from attempting to
-			// create connections too quickly.
-
-			r.dialSleep(ctx)
-			continue
+			continue LOOP
 		case <-ctx.Done():
 			close(addresses)
 			break LOOP
@@ -753,6 +713,7 @@ func (r *Router) connectPeer(ctx context.Context, address NodeAddress) {
 
 	if err := r.runWithPeerMutex(func() error { return r.peerManager.Dialed(address) }); err != nil {
 		r.logger.Error("failed to dial peer", "op", "outgoing/dialing", "peer", address.NodeID, "err", err)
+		r.peerManager.dialWaker.Wake()
 		conn.Close()
 		return
 	}
@@ -834,13 +795,7 @@ func (r *Router) handshakePeer(
 	expectID types.NodeID,
 ) (types.NodeInfo, crypto.PubKey, error) {
 
-	if r.options.HandshakeTimeout > 0 {
-		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(ctx, r.options.HandshakeTimeout)
-		defer cancel()
-	}
-
-	peerInfo, peerKey, err := conn.Handshake(ctx, r.nodeInfo, r.privKey)
+	peerInfo, peerKey, err := conn.Handshake(ctx, r.options.HandshakeTimeout, r.nodeInfo, r.privKey)
 	if err != nil {
 		return peerInfo, peerKey, err
 	}
@@ -1040,6 +995,8 @@ func (r *Router) evictPeers() {
 		queue, ok := r.peerQueues[peerID]
 		r.peerMtx.RUnlock()
 
+		r.metrics.PeersEvicted.Add(1)
+
 		if ok {
 			queue.close()
 		}

internal/p2p/router_test.go

@@ -1,7 +1,6 @@
 package p2p_test
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -133,13 +132,6 @@ func TestRouter_Channel_Basic(t *testing.T) {
 	require.NoError(t, err)
 	require.Contains(t, router.NodeInfo().Channels, chDesc2.ID)
 
-	// Closing the channel, then opening it again should be fine.
-	channel.Close()
-	time.Sleep(100 * time.Millisecond) // yes yes, but Close() is async...
-
-	channel, err = router.OpenChannel(chDesc, &p2ptest.Message{}, 0)
-	require.NoError(t, err)
-
 	// We should be able to send on the channel, even though there are no peers.
 	p2ptest.RequireSend(t, channel, p2p.Envelope{
 		To:      types.NodeID(strings.Repeat("a", 40)),
@@ -352,7 +344,7 @@ func TestRouter_AcceptPeers(t *testing.T) {
 			closer := tmsync.NewCloser()
 			mockConnection := &mocks.Connection{}
 			mockConnection.On("String").Maybe().Return("mock")
-			mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
+			mockConnection.On("Handshake", mock.Anything, mock.Anything, selfInfo, selfKey).
 				Return(tc.peerInfo, tc.peerKey, nil)
 			mockConnection.On("Close").Run(func(_ mock.Arguments) { closer.Close() }).Return(nil)
 			mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
@@ -462,7 +454,7 @@ func TestRouter_AcceptPeers_HeadOfLineBlocking(t *testing.T) {
 
 	mockConnection := &mocks.Connection{}
 	mockConnection.On("String").Maybe().Return("mock")
-	mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
+	mockConnection.On("Handshake", mock.Anything, mock.Anything, selfInfo, selfKey).
 		WaitUntil(closeCh).Return(types.NodeInfo{}, nil, io.EOF)
 	mockConnection.On("Close").Return(nil)
 	mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
@@ -543,7 +535,7 @@ func TestRouter_DialPeers(t *testing.T) {
 			mockConnection := &mocks.Connection{}
 			mockConnection.On("String").Maybe().Return("mock")
 			if tc.dialErr == nil {
-				mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
+				mockConnection.On("Handshake", mock.Anything, mock.Anything, selfInfo, selfKey).
 					Return(tc.peerInfo, tc.peerKey, nil)
 				mockConnection.On("Close").Run(func(_ mock.Arguments) { closer.Close() }).Return(nil)
 			}
@@ -630,7 +622,7 @@ func TestRouter_DialPeers_Parallel(t *testing.T) {
 
 	mockConnection := &mocks.Connection{}
 	mockConnection.On("String").Maybe().Return("mock")
-	mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
+	mockConnection.On("Handshake", mock.Anything, mock.Anything, selfInfo, selfKey).
 		WaitUntil(closeCh).Return(types.NodeInfo{}, nil, io.EOF)
 	mockConnection.On("Close").Return(nil)
 
@@ -671,7 +663,6 @@ func TestRouter_DialPeers_Parallel(t *testing.T) {
 		peerManager,
 		[]p2p.Transport{mockTransport},
 		p2p.RouterOptions{
-			DialSleep: func(_ context.Context) {},
 			NumConcurrentDials: func() int {
 				ncpu := runtime.NumCPU()
 				if ncpu <= 3 {
@@ -710,7 +701,7 @@ func TestRouter_EvictPeers(t *testing.T) {
 
 	mockConnection := &mocks.Connection{}
 	mockConnection.On("String").Maybe().Return("mock")
-	mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
+	mockConnection.On("Handshake", mock.Anything, mock.Anything, selfInfo, selfKey).
 		Return(peerInfo, peerKey.PubKey(), nil)
 	mockConnection.On("ReceiveMessage").WaitUntil(closeCh).Return(chID, nil, io.EOF)
 	mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
@@ -779,7 +770,7 @@ func TestRouter_ChannelCompatability(t *testing.T) {
 
 	mockConnection := &mocks.Connection{}
 	mockConnection.On("String").Maybe().Return("mock")
-	mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
+	mockConnection.On("Handshake", mock.Anything, mock.Anything, selfInfo, selfKey).
 		Return(incompatiblePeer, peerKey.PubKey(), nil)
 	mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
 	mockConnection.On("Close").Return(nil)
@@ -828,7 +819,7 @@ func TestRouter_DontSendOnInvalidChannel(t *testing.T) {
 
 	mockConnection := &mocks.Connection{}
 	mockConnection.On("String").Maybe().Return("mock")
-	mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
+	mockConnection.On("Handshake", mock.Anything, mock.Anything, selfInfo, selfKey).
 		Return(peer, peerKey.PubKey(), nil)
 	mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
 	mockConnection.On("Close").Return(nil)

internal/p2p/switch.go

@@ -865,11 +865,11 @@ func (sw *Switch) handshakePeer(
 	c Connection,
 	expectPeerID types.NodeID,
 ) (types.NodeInfo, crypto.PubKey, error) {
-	// Moved from transport and hardcoded until legacy P2P stack removal.
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	peerInfo, peerKey, err := c.Handshake(ctx, sw.nodeInfo, sw.nodeKey.PrivKey)
+	// Moved timeout from transport and hardcoded until legacy P2P stack removal.
+	peerInfo, peerKey, err := c.Handshake(ctx, 5*time.Second, sw.nodeInfo, sw.nodeKey.PrivKey)
 	if err != nil {
 		return peerInfo, peerKey, ErrRejected{
 			conn:          c.(*mConnConnection).conn,

internal/p2p/switch_test.go

@@ -267,7 +267,7 @@ func TestSwitchPeerFilter(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	peerInfo, _, err := c.Handshake(ctx, sw.nodeInfo, sw.nodeKey.PrivKey)
+	peerInfo, _, err := c.Handshake(ctx, 0, sw.nodeInfo, sw.nodeKey.PrivKey)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -324,7 +324,7 @@ func TestSwitchPeerFilterTimeout(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	peerInfo, _, err := c.Handshake(ctx, sw.nodeInfo, sw.nodeKey.PrivKey)
+	peerInfo, _, err := c.Handshake(ctx, 0, sw.nodeInfo, sw.nodeKey.PrivKey)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -360,7 +360,7 @@ func TestSwitchPeerFilterDuplicate(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	peerInfo, _, err := c.Handshake(ctx, sw.nodeInfo, sw.nodeKey.PrivKey)
+	peerInfo, _, err := c.Handshake(ctx, 0, sw.nodeInfo, sw.nodeKey.PrivKey)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -415,7 +415,7 @@ func TestSwitchStopsNonPersistentPeerOnError(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	peerInfo, _, err := c.Handshake(ctx, sw.nodeInfo, sw.nodeKey.PrivKey)
+	peerInfo, _, err := c.Handshake(ctx, 0, sw.nodeInfo, sw.nodeKey.PrivKey)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/p2p/test_util.go

@@ -126,7 +126,7 @@ func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
 		}
 		return err
 	}
-	peerNodeInfo, _, err := pc.conn.Handshake(context.Background(), sw.nodeInfo, sw.nodeKey.PrivKey)
+	peerNodeInfo, _, err := pc.conn.Handshake(context.Background(), 0, sw.nodeInfo, sw.nodeKey.PrivKey)
 	if err != nil {
 		if err := conn.Close(); err != nil {
 			sw.Logger.Error("Error closing connection", "err", err)

internal/p2p/transport.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"time"
 
 	"github.com/tendermint/tendermint/crypto"
 	"github.com/tendermint/tendermint/internal/p2p/conn"
@@ -84,7 +85,7 @@ type Connection interface {
 	// FIXME: The handshake should really be the Router's responsibility, but
 	// that requires the connection interface to be byte-oriented rather than
 	// message-oriented (see comment above).
-	Handshake(context.Context, types.NodeInfo, crypto.PrivKey) (types.NodeInfo, crypto.PubKey, error)
+	Handshake(context.Context, time.Duration, types.NodeInfo, crypto.PrivKey) (types.NodeInfo, crypto.PubKey, error)
 
 	// ReceiveMessage returns the next message received on the connection,
 	// blocking until one is available. Returns io.EOF if closed.

internal/p2p/transport_mconn.go

@@ -9,6 +9,7 @@ import (
 	"net"
 	"strconv"
 	"sync"
+	"time"
 
 	"golang.org/x/net/netutil"
 
@@ -255,6 +256,7 @@ func newMConnConnection(
 // Handshake implements Connection.
 func (c *mConnConnection) Handshake(
 	ctx context.Context,
+	timeout time.Duration,
 	nodeInfo types.NodeInfo,
 	privKey crypto.PrivKey,
 ) (types.NodeInfo, crypto.PubKey, error) {
@@ -264,6 +266,12 @@ func (c *mConnConnection) Handshake(
 		peerKey  crypto.PubKey
 		errCh    = make(chan error, 1)
 	)
+	handshakeCtx := ctx
+	if timeout > 0 {
+		var cancel context.CancelFunc
+		handshakeCtx, cancel = context.WithTimeout(ctx, timeout)
+		defer cancel()
+	}
 	// To handle context cancellation, we need to do the handshake in a
 	// goroutine and abort the blocking network calls by closing the connection
 	// when the context is canceled.
@@ -276,12 +284,17 @@ func (c *mConnConnection) Handshake(
 			}
 		}()
 		var err error
-		mconn, peerInfo, peerKey, err = c.handshake(ctx, nodeInfo, privKey)
-		errCh <- err
+		mconn, peerInfo, peerKey, err = c.handshake(handshakeCtx, nodeInfo, privKey)
+
+		select {
+		case errCh <- err:
+		case <-handshakeCtx.Done():
+		}
+
 	}()
 
 	select {
-	case <-ctx.Done():
+	case <-handshakeCtx.Done():
 		_ = c.Close()
 		return types.NodeInfo{}, nil, ctx.Err()
 

internal/p2p/transport_memory.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net"
 	"sync"
+	"time"
 
 	"github.com/tendermint/tendermint/crypto"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
@@ -270,9 +271,16 @@ func (c *MemoryConnection) Status() conn.ConnectionStatus {
 // Handshake implements Connection.
 func (c *MemoryConnection) Handshake(
 	ctx context.Context,
+	timeout time.Duration,
 	nodeInfo types.NodeInfo,
 	privKey crypto.PrivKey,
 ) (types.NodeInfo, crypto.PubKey, error) {
+	if timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, timeout)
+		defer cancel()
+	}
+
 	select {
 	case c.sendCh <- memoryMessage{nodeInfo: &nodeInfo, pubKey: privKey.PubKey()}:
 		c.logger.Debug("sent handshake", "nodeInfo", nodeInfo)

internal/p2p/transport_test.go

@@ -265,7 +265,7 @@ func TestConnection_Handshake(t *testing.T) {
 		errCh := make(chan error, 1)
 		go func() {
 			// Must use assert due to goroutine.
-			peerInfo, peerKey, err := ba.Handshake(ctx, bInfo, bKey)
+			peerInfo, peerKey, err := ba.Handshake(ctx, 0, bInfo, bKey)
 			if err == nil {
 				assert.Equal(t, aInfo, peerInfo)
 				assert.Equal(t, aKey.PubKey(), peerKey)
@@ -273,7 +273,7 @@ func TestConnection_Handshake(t *testing.T) {
 			errCh <- err
 		}()
 
-		peerInfo, peerKey, err := ab.Handshake(ctx, aInfo, aKey)
+		peerInfo, peerKey, err := ab.Handshake(ctx, 0, aInfo, aKey)
 		require.NoError(t, err)
 		require.Equal(t, bInfo, peerInfo)
 		require.Equal(t, bKey.PubKey(), peerKey)
@@ -291,7 +291,7 @@ func TestConnection_HandshakeCancel(t *testing.T) {
 		ab, ba := dialAccept(t, a, b)
 		timeoutCtx, cancel := context.WithTimeout(ctx, 1*time.Minute)
 		cancel()
-		_, _, err := ab.Handshake(timeoutCtx, types.NodeInfo{}, ed25519.GenPrivKey())
+		_, _, err := ab.Handshake(timeoutCtx, 0, types.NodeInfo{}, ed25519.GenPrivKey())
 		require.Error(t, err)
 		require.Equal(t, context.Canceled, err)
 		_ = ab.Close()
@@ -301,7 +301,7 @@ func TestConnection_HandshakeCancel(t *testing.T) {
 		ab, ba = dialAccept(t, a, b)
 		timeoutCtx, cancel = context.WithTimeout(ctx, 200*time.Millisecond)
 		defer cancel()
-		_, _, err = ab.Handshake(timeoutCtx, types.NodeInfo{}, ed25519.GenPrivKey())
+		_, _, err = ab.Handshake(timeoutCtx, 0, types.NodeInfo{}, ed25519.GenPrivKey())
 		require.Error(t, err)
 		require.Equal(t, context.DeadlineExceeded, err)
 		_ = ab.Close()
@@ -630,13 +630,13 @@ func dialAcceptHandshake(t *testing.T, a, b p2p.Transport) (p2p.Connection, p2p.
 	go func() {
 		privKey := ed25519.GenPrivKey()
 		nodeInfo := types.NodeInfo{NodeID: types.NodeIDFromPubKey(privKey.PubKey())}
-		_, _, err := ba.Handshake(ctx, nodeInfo, privKey)
+		_, _, err := ba.Handshake(ctx, 0, nodeInfo, privKey)
 		errCh <- err
 	}()
 
 	privKey := ed25519.GenPrivKey()
 	nodeInfo := types.NodeInfo{NodeID: types.NodeIDFromPubKey(privKey.PubKey())}
-	_, _, err := ab.Handshake(ctx, nodeInfo, privKey)
+	_, _, err := ab.Handshake(ctx, 0, nodeInfo, privKey)
 	require.NoError(t, err)
 
 	timer := time.NewTimer(2 * time.Second)

internal/proxy/mocks/app_conn_consensus.go

@@ -150,3 +150,18 @@ func (_m *AppConnConsensus) InitChainSync(_a0 context.Context, _a1 types.Request
 func (_m *AppConnConsensus) SetResponseCallback(_a0 abciclient.Callback) {
 	_m.Called(_a0)
 }
+
+type NewAppConnConsensusT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewAppConnConsensus creates a new instance of AppConnConsensus. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewAppConnConsensus(t NewAppConnConsensusT) *AppConnConsensus {
+	mock := &AppConnConsensus{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/proxy/mocks/app_conn_mempool.go

@@ -118,3 +118,18 @@ func (_m *AppConnMempool) FlushSync(_a0 context.Context) error {
 func (_m *AppConnMempool) SetResponseCallback(_a0 abciclient.Callback) {
 	_m.Called(_a0)
 }
+
+type NewAppConnMempoolT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewAppConnMempool creates a new instance of AppConnMempool. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewAppConnMempool(t NewAppConnMempoolT) *AppConnMempool {
+	mock := &AppConnMempool{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/proxy/mocks/app_conn_query.go

@@ -97,3 +97,18 @@ func (_m *AppConnQuery) QuerySync(_a0 context.Context, _a1 types.RequestQuery) (
 
 	return r0, r1
 }
+
+type NewAppConnQueryT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewAppConnQuery creates a new instance of AppConnQuery. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewAppConnQuery(t NewAppConnQueryT) *AppConnQuery {
+	mock := &AppConnQuery{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/proxy/mocks/app_conn_snapshot.go

@@ -120,3 +120,18 @@ func (_m *AppConnSnapshot) OfferSnapshotSync(_a0 context.Context, _a1 types.Requ
 
 	return r0, r1
 }
+
+type NewAppConnSnapshotT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewAppConnSnapshot creates a new instance of AppConnSnapshot. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewAppConnSnapshot(t NewAppConnSnapshotT) *AppConnSnapshot {
+	mock := &AppConnSnapshot{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/state/indexer/mocks/event_sink.go

@@ -165,3 +165,18 @@ func (_m *EventSink) Type() indexer.EventSinkType {
 
 	return r0
 }
+
+type NewEventSinkT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewEventSink creates a new instance of EventSink. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewEventSink(t NewEventSinkT) *EventSink {
+	mock := &EventSink{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/state/mocks/block_store.go

@@ -208,3 +208,18 @@ func (_m *BlockStore) Size() int64 {
 
 	return r0
 }
+
+type NewBlockStoreT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewBlockStore creates a new instance of BlockStore. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewBlockStore(t NewBlockStoreT) *BlockStore {
+	mock := &BlockStore{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/state/mocks/evidence_pool.go

@@ -68,3 +68,18 @@ func (_m *EvidencePool) PendingEvidence(maxBytes int64) ([]types.Evidence, int64
 func (_m *EvidencePool) Update(_a0 state.State, _a1 types.EvidenceList) {
 	_m.Called(_a0, _a1)
 }
+
+type NewEvidencePoolT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewEvidencePool creates a new instance of EvidencePool. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewEvidencePool(t NewEvidencePoolT) *EvidencePool {
+	mock := &EvidencePool{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/state/mocks/store.go

@@ -186,3 +186,18 @@ func (_m *Store) SaveValidatorSets(_a0 int64, _a1 int64, _a2 *types.ValidatorSet
 
 	return r0
 }
+
+type NewStoreT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewStore creates a new instance of Store. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewStore(t NewStoreT) *Store {
+	mock := &Store{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/statesync/mocks/state_provider.go

@@ -82,3 +82,18 @@ func (_m *StateProvider) State(ctx context.Context, height uint64) (state.State,
 
 	return r0, r1
 }
+
+type NewStateProviderT interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewStateProvider creates a new instance of StateProvider. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewStateProvider(t NewStateProviderT) *StateProvider {
+	mock := &StateProvider{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}