p2p: reduce ability of SendError to disconnect peers (backport #8597) (#8603)

internal/p2p/peermanager.go

@@ -452,6 +452,13 @@ func (m *PeerManager) PeerRatio() float64 {
 	return float64(m.store.Size()) / float64(m.options.MaxPeers)
 }
 
+func (m *PeerManager) HasMaxPeerCapacity() bool {
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
+
+	return len(m.connected) >= int(m.options.MaxConnected)
+}
+
 // DialNext finds an appropriate peer address to dial, and marks it as dialing.
 // If no peer is found, or all connection slots are full, it blocks until one
 // becomes available. The caller must call Dialed() or DialFailed() for the

internal/p2p/router.go

@@ -54,6 +54,7 @@ type Envelope struct {
 type PeerError struct {
 	NodeID types.NodeID
 	Err    error
+	Fatal  bool
 }
 
 // Channel is a bidirectional channel to exchange Protobuf messages with peers,
@@ -507,10 +508,20 @@ func (r *Router) routeChannel(
 				return
 			}
 
-			r.logger.Error("peer error, evicting", "peer", peerError.NodeID, "err", peerError.Err)
-
-			r.peerManager.Errored(peerError.NodeID, peerError.Err)
-
+			shouldEvict := peerError.Fatal || r.peerManager.HasMaxPeerCapacity()
+			r.logger.Error("peer error",
+				"peer", peerError.NodeID,
+				"err", peerError.Err,
+				"evicting", shouldEvict,
+			)
+			if shouldEvict {
+				r.peerManager.Errored(peerError.NodeID, peerError.Err)
+			} else {
+				r.peerManager.processPeerEvent(PeerUpdate{
+					NodeID: peerError.NodeID,
+					Status: PeerStatusBad,
+				})
+			}
 		case <-r.stopCh:
 			return
 		}