Release v0.35.2 (#2863)

* pstoremanager: fix connectedness check

* Close quic conns when wrapping conn fails

* Add a transport level test to ensure we close conns after rejecting them by the rcmgr

* PR Comments

* chore: Bump fx to v1.22.1 (#2857)

* chore: Bump gorilla/websocket to 1.5.3

This change has some history. Originally there was v1.5.0, then the
project stalled and eventually the repo got archived. Some new
maintainers stepped up and released v1.5.1. That version had some
controversial changes including excessive logging (see
gorilla/websocket#880). This caused us to
downgrade this dep back to v1.5.0 (see #2762). The change was short
lived as I bumped this dep back up to v1.5.1 without remembering this
context.

Since then the maintainers of gorilla/websocket have released a new
version v1.5.3 that brings the project back to the state of when it got
archived (minus a README edit). Bumping to this version should solve our
issues with v1.5.1 without having to downgrade back down to v1.5.0.

* peerstore: don't intern protocols  (#2860)

* peerstore: don't intern protocols

* peerstore: reduce default protocol limit to 128

* Remove unused mutex

---------

Co-authored-by: Marco Munizaga <git@marcopolo.io>

* webtransport: close underlying h3 connection (#2862)

* release v0.35.2

---------

Co-authored-by: sukun <sukunrt@gmail.com>

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)
+}