network: handle stream creation on existing conns

network/p2p/p2p.go

@@ -195,6 +195,9 @@ type connLimitConfig struct {
 func deriveConnLimits(cfg config.Local) connLimitConfig {
 	var low, high, rcmgrConns, rcmgrConnsInbound, rcmgrConnsOutbound int
 	rcmgrConnsOutbound = cfg.GossipFanout * 3
+	if cfg.EnableDHTProviders {
+		rcmgrConnsOutbound += cfg.GossipFanout * 3
+	}
 	if cfg.IsListenServer() {
 		if cfg.IncomingConnectionsLimit < 0 {
 			rcmgrConns = math.MaxInt
@@ -340,8 +343,18 @@ func (s *serviceImpl) DialPeersUntilTargetCount(targetConnCount int) bool {
 		if numOutgoingConns >= targetConnCount {
 			return numOutgoingConns > preExistingConns
 		}
-		// if we are already connected to this peer, skip it
-		if len(s.host.Network().ConnsToPeer(peerInfo.ID)) > 0 {
+		// if we are already connected to this peer, ensure it's properly handled
+		if conns := s.host.Network().ConnsToPeer(peerInfo.ID); len(conns) > 0 {
+			if !s.host.ConnManager().IsProtected(peerInfo.ID, cnmgrTag) {
+				// connection was established by DHT/pubsub before the mesh thread
+				// could protect it, so handleConnected skipped stream creation.
+				// protect and re-trigger stream setup now.
+				s.host.ConnManager().Protect(peerInfo.ID, cnmgrTag)
+				go s.streams.handleConnected(conns[0])
+				if conns[0].Stat().Direction == network.DirOutbound {
+					numOutgoingConns++
+				}
+			}
 			continue
 		}
 		err := s.dialNode(context.Background(), peerInfo) // leaving the calls as blocking for now, to not over-connect beyond fanout

network/p2p/p2p_test.go

@@ -359,6 +359,23 @@ func TestDeriveConnLimits_UnboundedServer(t *testing.T) {
 	require.Equal(t, 0, limits.connMgrLow)
 }
 
+func TestDeriveConnLimits_DHTProviders(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	t.Parallel()
+
+	cfg := config.GetDefaultLocal()
+	cfg.NetAddress = ":4160"
+	cfg.IncomingConnectionsLimit = 2400
+	cfg.GossipFanout = 4
+	cfg.EnableDHTProviders = true
+	limits := deriveConnLimits(cfg)
+	require.Equal(t, 2400+12+12, limits.rcmgrConns)
+	require.Equal(t, 2400, limits.rcmgrConnsInbound)
+	require.Equal(t, 24, limits.rcmgrConnsOutbound)
+	require.Equal(t, 2424, limits.connMgrHigh)
+	require.Equal(t, 2327, limits.connMgrLow) // 2424 * 96 / 100
+}
+
 func TestDeriveConnLimits_Client(t *testing.T) {
 	partitiontest.PartitionTest(t)
 	t.Parallel()

network/p2p/streams.go

@@ -98,14 +98,16 @@ func (n *streamManager) streamHandler(stream network.Stream) {
 				// an error occurred while checking the old stream
 				n.log.Infof("Failed to check old stream with %s: %v", remotePeer, err)
 			}
-			n.streams[stream.Conn().RemotePeer()] = stream
+			// old stream is dead, remove
+			delete(n.streams, remotePeer)
 
 			incoming := stream.Conn().Stat().Direction == network.DirInbound
 			if err1 := n.dispatch(n.ctx, remotePeer, stream, incoming); err1 != nil {
 				n.log.Errorln(err1.Error())
 				_ = stream.Reset()
 				return
 			}
+			n.streams[stream.Conn().RemotePeer()] = stream
 			dispatched = true
 			return
 		}
@@ -115,14 +117,15 @@ func (n *streamManager) streamHandler(stream network.Stream) {
 		return
 	}
 	// no old stream
-	n.streams[stream.Conn().RemotePeer()] = stream
 	incoming := stream.Conn().Stat().Direction == network.DirInbound
 	if err := n.dispatch(n.ctx, remotePeer, stream, incoming); err != nil {
 		n.log.Errorln(err.Error())
 		_ = stream.Reset()
 		return
 	}
 
+	n.streams[stream.Conn().RemotePeer()] = stream
+
 	dispatched = true
 }
 
@@ -143,21 +146,12 @@ func (n *streamManager) dispatch(ctx context.Context, remotePeer peer.ID, stream
 // We do some read/write operations in this handler for metadata exchange that creates a race condition
 // with StopNotify on network shutdown. To avoid, run the handler as a goroutine.
 func (n *streamManager) Connected(net network.Network, conn network.Conn) {
-	go n.handleConnected(conn)
-}
-
-func (n *streamManager) handleConnected(conn network.Conn) {
-	dispatched := false
-	defer func() {
-		if !dispatched {
-			n.host.ConnManager().Unprotect(conn.RemotePeer(), cnmgrTag)
-		}
-	}()
 	remotePeer := conn.RemotePeer()
 	localPeer := n.host.ID()
 
 	if conn.Stat().Direction == network.DirInbound && !n.allowIncomingGossip {
 		n.log.Debugf("%s: ignoring incoming connection from %s", localPeer.String(), remotePeer.String())
+		n.host.ConnManager().Unprotect(conn.RemotePeer(), cnmgrTag)
 		return
 	}
 
@@ -166,7 +160,6 @@ func (n *streamManager) handleConnected(conn network.Conn) {
 	// so mark dispatched to preserve the cnmgr protection set by dialNode.
 	if localPeer > remotePeer {
 		n.log.Debugf("%s: ignoring a lesser peer ID %s", localPeer.String(), remotePeer.String())
-		dispatched = true
 		return
 	}
 
@@ -179,10 +172,23 @@ func (n *streamManager) handleConnected(conn network.Conn) {
 		}
 	}
 
+	go n.handleConnected(conn)
+}
+
+func (n *streamManager) handleConnected(conn network.Conn) {
+	dispatched := false
+	defer func() {
+		if !dispatched {
+			n.host.ConnManager().Unprotect(conn.RemotePeer(), cnmgrTag)
+		}
+	}()
+	remotePeer := conn.RemotePeer()
+	localPeer := n.host.ID()
+
 	n.streamsLock.Lock()
 	_, ok := n.streams[remotePeer]
+	n.streamsLock.Unlock()
 	if ok {
-		n.streamsLock.Unlock()
 		n.log.Debugf("%s: already have a stream to/from %s", localPeer.String(), remotePeer.String())
 		dispatched = true
 		return // there's already an active stream with this peer for our protocol
@@ -195,12 +201,8 @@ func (n *streamManager) handleConnected(conn network.Conn) {
 	stream, err := n.host.NewStream(n.ctx, remotePeer, protos...)
 	if err != nil {
 		n.log.Infof("%s: failed to open stream to %s (%s): %v", localPeer.String(), remotePeer, conn.RemoteMultiaddr().String(), err)
-		n.streamsLock.Unlock()
 		return
 	}
-	n.streams[remotePeer] = stream
-	n.streamsLock.Unlock()
-
 	n.log.Infof("%s: using protocol %s with peer %s", localPeer.String(), stream.Protocol(), remotePeer.String())
 
 	incoming := stream.Conn().Stat().Direction == network.DirInbound
@@ -210,6 +212,20 @@ func (n *streamManager) handleConnected(conn network.Conn) {
 		return
 	}
 
+	n.streamsLock.Lock()
+	defer n.streamsLock.Unlock()
+	if _, exists := n.streams[remotePeer]; exists {
+		// another stream was added in the meantime, close this one and keep the existing one
+		_ = stream.Reset()
+		dispatched = true
+		return
+	}
+	// don't add disconnected / died conns, so Disconnect won't need to clean up
+	if stream.Conn().IsClosed() {
+		_ = stream.Reset()
+		return // dispatched is still false
+	}
+	n.streams[remotePeer] = stream
 	dispatched = true
 }