fix(p2p/pex): respect MaxNumOutboundPeers limit while dialing peers p…

…rovided by a seed node (backport cometbft#3360) (cometbft#3395)

Closes cometbft#486.

Commits:

1. Make `TestConnectionSpeedForPeerReceivedFromSeed` to fail, as it will
make the node to connect to more than the configured
`MaxNumOutboundPeers`
2. Solve the issue by introducing a channel to wake up
`ensurePeersRoutine()`. In this way, the node will not to wait until the
next `defaultEnsurePeersPeriod` (30s) before attempting again to connect
to new peers (`ensurePeers()` method) whose addresses were provide by
the seed node.

---

#### PR checklist

- [x] Tests written/updated
- [x] Changelog entry added in `.changelog` (we use
[unclog](https://github.com/informalsystems/unclog) to manage our
changelog)
- [x] Updated relevant documentation (`docs/` or `spec/`) and code
comments
- [x] Title follows the [Conventional
Commits](https://www.conventionalcommits.org/en/v1.0.0/) spec
<hr>This is an automatic backport of pull request cometbft#3360 done by
[Mergify](https://mergify.com).

---------

Co-authored-by: Daniel <[email protected]>

.changelog/unreleased/bug-fixes/486-p2p-max-outbound.md

@@ -0,0 +1,3 @@
+- `[p2p]` Node respects configured `max_num_outbound_peers` limit when dialing
+  peers provided by a seed node
+  ([\#486](https://github.com/cometbft/cometbft/issues/486))

p2p/pex/pex_reactor.go

@@ -83,6 +83,7 @@ type Reactor struct {
 
 	book              AddrBook
 	config            *ReactorConfig
+	ensurePeersCh     chan struct{} // Wakes up ensurePeersRoutine()
 	ensurePeersPeriod time.Duration // TODO: should go in the config
 	peersRoutineWg    sync.WaitGroup
 
@@ -132,6 +133,7 @@ func NewReactor(b AddrBook, config *ReactorConfig) *Reactor {
 	r := &Reactor{
 		book:                 b,
 		config:               config,
+		ensurePeersCh:        make(chan struct{}),
 		ensurePeersPeriod:    defaultEnsurePeersPeriod,
 		requestsSent:         cmap.NewCMap(),
 		lastReceivedRequests: cmap.NewCMap(),
@@ -367,14 +369,6 @@ func (r *Reactor) ReceiveAddrs(addrs []*p2p.NetAddress, src Peer) error {
 		return err
 	}
 
-	srcIsSeed := false
-	for _, seedAddr := range r.seedAddrs {
-		if seedAddr.Equals(srcAddr) {
-			srcIsSeed = true
-			break
-		}
-	}
-
 	for _, netAddr := range addrs {
 		// NOTE: we check netAddr validity and routability in book#AddAddress.
 		err = r.book.AddAddress(netAddr, srcAddr)
@@ -384,21 +378,16 @@ func (r *Reactor) ReceiveAddrs(addrs []*p2p.NetAddress, src Peer) error {
 			// peer here too?
 			continue
 		}
+	}
 
-		// If this address came from a seed node, try to connect to it without
-		// waiting (#2093)
-		if srcIsSeed {
-			go func(addr *p2p.NetAddress) {
-				err := r.dialPeer(addr)
-				if err != nil {
-					switch err.(type) {
-					case errMaxAttemptsToDial, errTooEarlyToDial, p2p.ErrCurrentlyDialingOrExistingAddress:
-						r.Logger.Debug(err.Error(), "addr", addr)
-					default:
-						r.Logger.Debug(err.Error(), "addr", addr)
-					}
-				}
-			}(netAddr)
+	// Try to connect to addresses coming from a seed node without waiting (#2093)
+	for _, seedAddr := range r.seedAddrs {
+		if seedAddr.Equals(srcAddr) {
+			select {
+			case r.ensurePeersCh <- struct{}{}:
+			default:
+			}
+			break
 		}
 	}
 
@@ -446,6 +435,8 @@ func (r *Reactor) ensurePeersRoutine() {
 		select {
 		case <-ticker.C:
 			r.ensurePeers()
+		case <-r.ensurePeersCh:
+			r.ensurePeers()
 		case <-r.book.Quit():
 			return
 		case <-r.Quit():

p2p/pex/pex_reactor_test.go

@@ -271,27 +271,48 @@ func TestConnectionSpeedForPeerReceivedFromSeed(t *testing.T) {
 	require.NoError(t, err)
 	defer os.RemoveAll(dir)
 
-	// 1. create peer
-	peerSwitch := testCreateDefaultPeer(dir, 1)
-	require.NoError(t, peerSwitch.Start())
-	defer peerSwitch.Stop() //nolint:errcheck // ignore for tests
+	// Default is 10, we need one connection for the seed node.
+	cfg.MaxNumOutboundPeers = 2
 
-	// 2. Create seed which knows about the peer
-	peerAddr := peerSwitch.NetAddress()
-	seed := testCreateSeed(dir, 2, []*p2p.NetAddress{peerAddr}, []*p2p.NetAddress{peerAddr})
-	require.NoError(t, seed.Start())
-	defer seed.Stop() //nolint:errcheck // ignore for tests
+	var id int
+	var knownAddrs []*p2p.NetAddress
 
-	// 3. create another peer with only seed configured.
-	secondPeer := testCreatePeerWithSeed(dir, 3, seed)
-	require.NoError(t, secondPeer.Start())
-	defer secondPeer.Stop() //nolint:errcheck // ignore for tests
+	// 1. Create some peers
+	for id = 0; id < cfg.MaxNumOutboundPeers+1; id++ {
+		peer := testCreateDefaultPeer(dir, id)
+		require.NoError(t, peer.Start())
+		addr := peer.NetAddress()
+		defer peer.Stop() //nolint:errcheck // ignore for tests
 
-	// 4. check that the second peer connects to seed immediately
-	assertPeersWithTimeout(t, []*p2p.Switch{secondPeer}, 3*time.Second, 1)
+		knownAddrs = append(knownAddrs, addr)
+		t.Log("Created peer", id, addr)
+	}
 
-	// 5. check that the second peer connects to the first peer immediately
-	assertPeersWithTimeout(t, []*p2p.Switch{secondPeer}, 1*time.Second, 2)
+	// 2. Create seed node which knows about the previous peers
+	seed := testCreateSeed(dir, id, knownAddrs, knownAddrs)
+	require.NoError(t, seed.Start())
+	defer seed.Stop() //nolint:errcheck // ignore for tests
+	t.Log("Created seed", id, seed.NetAddress())
+
+	// 3. Create a node with only seed configured.
+	id++
+	node := testCreatePeerWithSeed(dir, id, seed)
+	require.NoError(t, node.Start())
+	defer node.Stop() //nolint:errcheck // ignore for tests
+	t.Log("Created node", id, node.NetAddress())
+
+	// 4. Check that the node connects to seed immediately
+	assertPeersWithTimeout(t, []*p2p.Switch{node}, 3*time.Second, 1)
+
+	// 5. Check that the node connects to the peers reported by the seed node
+	assertPeersWithTimeout(t, []*p2p.Switch{node}, 1*time.Second, cfg.MaxNumOutboundPeers)
+
+	// 6. Assert that the configured maximum number of inbound/outbound peers
+	// are respected, see https://github.com/cometbft/cometbft/issues/486
+	outbound, inbound, dialing := node.NumPeers()
+	assert.LessOrEqual(t, inbound, cfg.MaxNumInboundPeers)
+	assert.LessOrEqual(t, outbound, cfg.MaxNumOutboundPeers)
+	assert.Zero(t, dialing)
 }
 
 func TestPEXReactorSeedMode(t *testing.T) {
@@ -590,16 +611,14 @@ func testCreatePeerWithConfig(dir string, id int, config *ReactorConfig) *p2p.Sw
 		id,
 		func(_ int, sw *p2p.Switch) *p2p.Switch {
 			book := NewAddrBook(filepath.Join(dir, fmt.Sprintf("addrbook%d.json", id)), false)
-			book.SetLogger(log.TestingLogger())
+			book.SetLogger(log.TestingLogger().With("book", id))
 			sw.SetAddrBook(book)
 
-			sw.SetLogger(log.TestingLogger())
-
 			r := NewReactor(
 				book,
 				config,
 			)
-			r.SetLogger(log.TestingLogger())
+			r.SetLogger(log.TestingLogger().With("pex", id))
 			sw.AddReactor("pex", r)
 			return sw
 		},

spec/p2p/implementation/peer_manager.md

@@ -85,16 +85,9 @@ To avoid this delay, which can be particularly relevant when the node has no
 peers, a node immediately attempts to dial peer addresses when they are
 received from a peer that is locally configured as a seed node.
 
-> FIXME: The current logic was introduced in [#3762](https://github.com/tendermint/tendermint/pull/3762).
-> Although it fix the issue, the delay between receiving an address and dialing
-> the peer, it does not impose and limit on how many addresses are dialed in this
-> scenario.
-> So, all addresses received from a seed node are dialed, regardless of the
-> current number of outbound peers, the number of dialing routines, or the
-> `MaxNumOutboundPeers` parameter.
->
-> Issue [#9548](https://github.com/tendermint/tendermint/issues/9548) was
-> created to handle this situation.
+> This was implemented in a rough way, leading to inconsistencies described in
+> this [issue](https://github.com/cometbft/cometbft/issues/486),
+> fixed by this [PR](https://github.com/cometbft/cometbft/pull/3360).
 
 ### First round