Release v0.35.2

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0
 	github.com/flynn/noise v1.1.0
 	github.com/google/gopacket v1.1.19
-	github.com/gorilla/websocket v1.5.1
+	github.com/gorilla/websocket v1.5.3
 	github.com/hashicorp/golang-lru/arc/v2 v2.0.7
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/ipfs/go-cid v0.4.1
@@ -55,7 +55,7 @@ require (
 	github.com/quic-go/webtransport-go v0.8.0
 	github.com/raulk/go-watchdog v1.3.0
 	github.com/stretchr/testify v1.9.0
-	go.uber.org/fx v1.21.1
+	go.uber.org/fx v1.22.1
 	go.uber.org/goleak v1.3.0
 	go.uber.org/mock v0.4.0
 	golang.org/x/crypto v0.23.0
@@ -126,9 +126,3 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	lukechampine.com/blake3 v1.2.1 // indirect
 )
-
-// Remove this once fx releases the next version.
-// We want to ship with a fix around SIGINT handling:
-// https://github.com/uber-go/fx/pull/1198.
-// Context: https://github.com/libp2p/go-libp2p/issues/2785
-replace go.uber.org/fx v1.21.1 => github.com/uber-go/fx v1.21.2-0.20240515133256-cb9cccf55845

go.sum

@@ -121,8 +121,8 @@ github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
 github.com/googleapis/gax-go/v2 v2.0.3/go.mod h1:LLvjysVCY1JZeum8Z6l8qUty8fiNwE08qbEPm1M08qg=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
-github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/grpc-gateway v1.5.0/go.mod h1:RSKVYQBd5MCa4OVpNdGskqpgL2+G+NZTnrVHpWWfpdw=
 github.com/hashicorp/golang-lru/arc/v2 v2.0.7 h1:QxkVTxwColcduO+LP7eJO56r2hFiG8zEbfAAzRv52KQ=
@@ -401,8 +401,6 @@ github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8
 github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE=
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
 github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07/go.mod h1:kDXzergiv9cbyO7IOYJZWg1U88JhDg3PB6klq9Hg2pA=
-github.com/uber-go/fx v1.21.2-0.20240515133256-cb9cccf55845 h1:1ZbnuG7aj1UxZnfsJmEpACmspZMkj5Fdvg7C1yWgQCE=
-github.com/uber-go/fx v1.21.2-0.20240515133256-cb9cccf55845/go.mod h1:HT2M7d7RHo+ebKGh9NRcrsrHHfpZ60nW3QRubMRfv48=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/urfave/cli v1.22.2/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/viant/assertly v0.4.8/go.mod h1:aGifi++jvCrUaklKEKT0BU95igDNaqkvz+49uaYMPRU=
@@ -417,6 +415,8 @@ go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/dig v1.17.1 h1:Tga8Lz8PcYNsWsyHMZ1Vm0OQOUaJNDyvPImgbAu9YSc=
 go.uber.org/dig v1.17.1/go.mod h1:Us0rSJiThwCv2GteUN0Q7OKvU7n5J4dxZ9JKUXozFdE=
+go.uber.org/fx v1.22.1 h1:nvvln7mwyT5s1q201YE29V/BFrGor6vMiDNpU/78Mys=
+go.uber.org/fx v1.22.1/go.mod h1:HT2M7d7RHo+ebKGh9NRcrsrHHfpZ60nW3QRubMRfv48=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=

p2p/host/peerstore/pstoreds/protobook.go

@@ -48,7 +48,7 @@ func NewProtoBook(meta pstore.PeerMetadata, opts ...ProtoBookOption) (*dsProtoBo
 			}
 			return ret
 		}(),
-		maxProtos: 1024,
+		maxProtos: 128,
 	}
 
 	for _, opt := range opts {

p2p/host/peerstore/pstoremem/protobook.go

@@ -26,9 +26,6 @@ type memoryProtoBook struct {
 	segments protoSegments
 
 	maxProtos int
-
-	lk       sync.RWMutex
-	interned map[protocol.ID]protocol.ID
 }
 
 var _ pstore.ProtoBook = (*memoryProtoBook)(nil)
@@ -44,7 +41,6 @@ func WithMaxProtocols(num int) ProtoBookOption {
 
 func NewProtoBook(opts ...ProtoBookOption) (*memoryProtoBook, error) {
 	pb := &memoryProtoBook{
-		interned: make(map[protocol.ID]protocol.ID, 256),
 		segments: func() (ret protoSegments) {
 			for i := range ret {
 				ret[i] = &protoSegment{
@@ -53,7 +49,7 @@ func NewProtoBook(opts ...ProtoBookOption) (*memoryProtoBook, error) {
 			}
 			return ret
 		}(),
-		maxProtos: 1024,
+		maxProtos: 128,
 	}
 
 	for _, opt := range opts {
@@ -64,38 +60,14 @@ func NewProtoBook(opts ...ProtoBookOption) (*memoryProtoBook, error) {
 	return pb, nil
 }
 
-func (pb *memoryProtoBook) internProtocol(proto protocol.ID) protocol.ID {
-	// check if it is interned with the read lock
-	pb.lk.RLock()
-	interned, ok := pb.interned[proto]
-	pb.lk.RUnlock()
-
-	if ok {
-		return interned
-	}
-
-	// intern with the write lock
-	pb.lk.Lock()
-	defer pb.lk.Unlock()
-
-	// check again in case it got interned in between locks
-	interned, ok = pb.interned[proto]
-	if ok {
-		return interned
-	}
-
-	pb.interned[proto] = proto
-	return proto
-}
-
 func (pb *memoryProtoBook) SetProtocols(p peer.ID, protos ...protocol.ID) error {
 	if len(protos) > pb.maxProtos {
 		return errTooManyProtocols
 	}
 
 	newprotos := make(map[protocol.ID]struct{}, len(protos))
 	for _, proto := range protos {
-		newprotos[pb.internProtocol(proto)] = struct{}{}
+		newprotos[proto] = struct{}{}
 	}
 
 	s := pb.segments.get(p)
@@ -121,7 +93,7 @@ func (pb *memoryProtoBook) AddProtocols(p peer.ID, protos ...protocol.ID) error
 	}
 
 	for _, proto := range protos {
-		protomap[pb.internProtocol(proto)] = struct{}{}
+		protomap[proto] = struct{}{}
 	}
 	return nil
 }
@@ -151,7 +123,10 @@ func (pb *memoryProtoBook) RemoveProtocols(p peer.ID, protos ...protocol.ID) err
 	}
 
 	for _, proto := range protos {
-		delete(protomap, pb.internProtocol(proto))
+		delete(protomap, proto)
+	}
+	if len(protomap) == 0 {
+		delete(s.protocols, p)
 	}
 	return nil
 }

p2p/host/pstoremanager/pstoremanager.go

@@ -121,10 +121,12 @@ func (m *PeerstoreManager) background(ctx context.Context, sub event.Subscriptio
 					// Check that the peer is actually not connected at this point.
 					// This avoids a race condition where the Connected notification
 					// is processed after this time has fired.
-					if m.network.Connectedness(p) != network.Connected {
+					switch m.network.Connectedness(p) {
+					case network.Connected, network.Limited:
+					default:
 						m.pstore.RemovePeer(p)
-						delete(disconnected, p)
 					}
+					delete(disconnected, p)
 				}
 			}
 		case <-ctx.Done():

p2p/protocol/identify/id_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/rand"
 	"slices"
 	"sync"
 	"testing"
@@ -730,6 +731,15 @@ func TestLargeIdentifyMessage(t *testing.T) {
 	}
 }
 
+func randString(n int) string {
+	chars := "abcdefghijklmnopqrstuvwxyz"
+	buf := make([]byte, n)
+	for i := 0; i < n; i++ {
+		buf[i] = chars[rand.Intn(len(chars))]
+	}
+	return string(buf)
+}
+
 func TestLargePushMessage(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -738,9 +748,9 @@ func TestLargePushMessage(t *testing.T) {
 	h2 := blhost.NewBlankHost(swarmt.GenSwarm(t))
 
 	// add protocol strings to make the message larger
-	// about 2K of protocol strings
-	for i := 0; i < 500; i++ {
-		r := protocol.ID(fmt.Sprintf("rand%d", i))
+	// about 3K of protocol strings
+	for i := 0; i < 100; i++ {
+		r := protocol.ID(fmt.Sprintf("%s-%d", randString(30), i))
 		h1.SetStreamHandler(r, func(network.Stream) {})
 		h2.SetStreamHandler(r, func(network.Stream) {})
 	}

p2p/test/transport/transport_test.go

@@ -20,7 +20,9 @@ import (
 	"github.com/libp2p/go-libp2p/core/connmgr"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
+	mocknetwork "github.com/libp2p/go-libp2p/core/network/mocks"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/peerstore"
 	"github.com/libp2p/go-libp2p/core/sec"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
 	"github.com/libp2p/go-libp2p/p2p/muxer/yamux"
@@ -29,8 +31,9 @@ import (
 	"github.com/libp2p/go-libp2p/p2p/security/noise"
 	tls "github.com/libp2p/go-libp2p/p2p/security/tls"
 	libp2pwebrtc "github.com/libp2p/go-libp2p/p2p/transport/webrtc"
+	"go.uber.org/mock/gomock"
 
-	"github.com/multiformats/go-multiaddr"
+	ma "github.com/multiformats/go-multiaddr"
 	"github.com/stretchr/testify/require"
 )
 
@@ -669,7 +672,7 @@ func TestDiscoverPeerIDFromSecurityNegotiation(t *testing.T) {
 
 			ai := &peer.AddrInfo{
 				ID:    bogusPeerId,
-				Addrs: []multiaddr.Multiaddr{h1.Addrs()[0]},
+				Addrs: []ma.Multiaddr{h1.Addrs()[0]},
 			}
 
 			// Try connecting with the bogus peer ID
@@ -688,3 +691,34 @@ func TestDiscoverPeerIDFromSecurityNegotiation(t *testing.T) {
 		})
 	}
 }
+
+// TestCloseConnWhenBlocked tests that the server closes the connection when the rcmgr blocks it.
+func TestCloseConnWhenBlocked(t *testing.T) {
+	for _, tc := range transportsToTest {
+		if tc.Name == "WebRTC" {
+			continue // WebRTC doesn't have a connection when we block so there's nothing to close
+		}
+		t.Run(tc.Name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			defer ctrl.Finish()
+			mockRcmgr := mocknetwork.NewMockResourceManager(ctrl)
+			mockRcmgr.EXPECT().OpenConnection(network.DirInbound, gomock.Any(), gomock.Any()).DoAndReturn(func(network.Direction, bool, ma.Multiaddr) (network.ConnManagementScope, error) {
+				// Block the connection
+				return nil, fmt.Errorf("connections blocked")
+			})
+			mockRcmgr.EXPECT().Close().AnyTimes()
+
+			server := tc.HostGenerator(t, TransportTestCaseOpts{ResourceManager: mockRcmgr})
+			client := tc.HostGenerator(t, TransportTestCaseOpts{NoListen: true})
+			defer server.Close()
+			defer client.Close()
+
+			client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			_, err := client.NewStream(ctx, server.ID(), ping.ID)
+			require.Error(t, err)
+			require.False(t, errors.Is(err, context.DeadlineExceeded), "expected error to be not be context deadline exceeded")
+		})
+	}
+}

p2p/transport/quic/listener.go

@@ -51,8 +51,10 @@ func (l *listener) Accept() (tpt.CapableConn, error) {
 		if err != nil {
 			return nil, err
 		}
-		c, err := l.setupConn(qconn)
+		c, err := l.wrapConn(qconn)
 		if err != nil {
+			log.Debugf("failed to setup connection: %s", err)
+			qconn.CloseWithError(1, "")
 			continue
 		}
 		l.transport.addConn(qconn, c)
@@ -79,7 +81,10 @@ func (l *listener) Accept() (tpt.CapableConn, error) {
 	}
 }
 
-func (l *listener) setupConn(qconn quic.Connection) (*conn, error) {
+// wrapConn wraps a QUIC connection into a libp2p [tpt.CapableConn].
+// If wrapping fails. The caller is responsible for cleaning up the
+// connection.
+func (l *listener) wrapConn(qconn quic.Connection) (*conn, error) {
 	remoteMultiaddr, err := quicreuse.ToQuicMultiaddr(qconn.RemoteAddr(), qconn.ConnectionState().Version)
 	if err != nil {
 		return nil, err
@@ -90,18 +95,16 @@ func (l *listener) setupConn(qconn quic.Connection) (*conn, error) {
 		log.Debugw("resource manager blocked incoming connection", "addr", qconn.RemoteAddr(), "error", err)
 		return nil, err
 	}
-	c, err := l.setupConnWithScope(qconn, connScope, remoteMultiaddr)
+	c, err := l.wrapConnWithScope(qconn, connScope, remoteMultiaddr)
 	if err != nil {
 		connScope.Done()
-		qconn.CloseWithError(1, "")
 		return nil, err
 	}
 
 	return c, nil
 }
 
-func (l *listener) setupConnWithScope(qconn quic.Connection, connScope network.ConnManagementScope, remoteMultiaddr ma.Multiaddr) (*conn, error) {
-
+func (l *listener) wrapConnWithScope(qconn quic.Connection, connScope network.ConnManagementScope, remoteMultiaddr ma.Multiaddr) (*conn, error) {
 	// The tls.Config used to establish this connection already verified the certificate chain.
 	// Since we don't have any way of knowing which tls.Config was used though,
 	// we have to re-determine the peer's identity here.

p2p/transport/quic/listener_test.go

@@ -1,9 +1,11 @@
 package libp2pquic
 
 import (
+	"context"
 	"crypto/rand"
 	"crypto/rsa"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -12,8 +14,11 @@ import (
 
 	ic "github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/network"
+	mocknetwork "github.com/libp2p/go-libp2p/core/network/mocks"
 	tpt "github.com/libp2p/go-libp2p/core/transport"
 	"github.com/libp2p/go-libp2p/p2p/transport/quicreuse"
+	"github.com/quic-go/quic-go"
+	"go.uber.org/mock/gomock"
 
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/stretchr/testify/require"
@@ -113,3 +118,51 @@ func TestCorrectNumberOfVirtualListeners(t *testing.T) {
 	ln.Close()
 	require.Empty(t, tpt.listeners[udpAddr.String()])
 }
+
+func TestCleanupConnWhenBlocked(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockRcmgr := mocknetwork.NewMockResourceManager(ctrl)
+	mockRcmgr.EXPECT().OpenConnection(network.DirInbound, false, gomock.Any()).DoAndReturn(func(network.Direction, bool, ma.Multiaddr) (network.ConnManagementScope, error) {
+		// Block the connection
+		return nil, fmt.Errorf("connections blocked")
+	})
+
+	server := newTransport(t, mockRcmgr)
+	serverTpt := server.(*transport)
+	defer server.(io.Closer).Close()
+
+	localAddrV1 := ma.StringCast("/ip4/127.0.0.1/udp/0/quic-v1")
+	ln, err := server.Listen(localAddrV1)
+	require.NoError(t, err)
+	defer ln.Close()
+	go ln.Accept()
+
+	client := newTransport(t, nil)
+	ctx := context.Background()
+
+	var quicErr *quic.ApplicationError = &quic.ApplicationError{}
+	conn, err := client.Dial(ctx, ln.Multiaddr(), serverTpt.localPeer)
+	if err != nil && errors.As(err, &quicErr) {
+		// We hit our expected application error
+		return
+	}
+
+	// No error yet, let's continue using the conn
+	s, err := conn.OpenStream(ctx)
+	if err != nil && errors.As(err, &quicErr) {
+		// We hit our expected application error
+		return
+	}
+
+	// No error yet, let's continue using the conn
+	s.SetReadDeadline(time.Now().Add(10 * time.Second))
+	b := [1]byte{}
+	_, err = s.Read(b[:])
+	if err != nil && errors.As(err, &quicErr) {
+		// We hit our expected application error
+		return
+	}
+
+	t.Fatalf("expected application error, got %v", err)
+}