feat(network): Add Conn.As

ConnAs works in a similar way to errors.As. It allows a user to cut
through the interface layers and extract a specific type of connection
if available.

This serves as a sort of escape hatch to allow users to leverage some
connection specific feature without having to support that feature for
all connections. Getting RTT information is one example.

It also allows us, within the library, to get specific types of
connections out of the interface box. This would have been useful in the
recent changes in tcpreuse. See
#3181 and
#3142.

Getting access to the underlying type can lead to hard to debug issues.
For example, if a user mutates connection state on the underlying type,
hooks that relied on only mutating that state from the wrapped
connection would never be called.

It is up to the user to ensure they are using this safely.

Author: MarcoPolo

core/network/conn.go

@@ -87,6 +87,20 @@ type Conn interface {
 	// IsClosed returns whether a connection is fully closed, so it can
 	// be garbage collected.
 	IsClosed() bool
+
+	// As finds the first conn in Conn's wrapped types that matches target, and
+	// if one is found, sets target to that conn value and returns true.
+	// Otherwise, it returns false. Similar to errors.As.
+	//
+	// target must be a pointer to the type you are matching against.
+	//
+	// This is an EXPERIMENTAL API. Getting access to the underlying type can
+	// lead to hard to debug issues. For example, if you mutate connection state
+	// on the underlying type, hooks that relied on only mutating that state
+	// from the wrapped connection would never be called.
+	//
+	// You very likely do not need to use this method.
+	As(target any) bool
 }
 
 // ConnectionState holds information about the connection.

core/network/mux.go

@@ -137,6 +137,20 @@ type MuxedConn interface {
 
 	// AcceptStream accepts a stream opened by the other side.
 	AcceptStream() (MuxedStream, error)
+
+	// As finds the first conn in MuxedConn's wrapped types that matches target,
+	// and if one is found, sets target to that conn value and returns true.
+	// Otherwise, it returns false. Similar to errors.As.
+	//
+	// target must be a pointer to the type you are matching against.
+	//
+	// This is an EXPERIMENTAL API. Getting access to the underlying type can
+	// lead to hard to debug issues. For example, if you mutate connection state
+	// on the underlying type, hooks that relied on only mutating that state
+	// from the wrapped connection would never be called.
+	//
+	// You very likely do not need to use this method.
+	As(target any) bool
 }
 
 // Multiplexer wraps a net.Conn with a stream multiplexing

libp2p_test.go

@@ -40,6 +40,10 @@ import (
 	libp2pwebrtc "github.com/libp2p/go-libp2p/p2p/transport/webrtc"
 	"github.com/libp2p/go-libp2p/p2p/transport/websocket"
 	webtransport "github.com/libp2p/go-libp2p/p2p/transport/webtransport"
+	"github.com/libp2p/go-yamux/v5"
+	"github.com/pion/webrtc/v4"
+	quicgo "github.com/quic-go/quic-go"
+	wtgo "github.com/quic-go/webtransport-go"
 	"go.uber.org/goleak"
 
 	ma "github.com/multiformats/go-multiaddr"
@@ -842,3 +846,120 @@ func BenchmarkAllAddrs(b *testing.B) {
 		addrsHost.AllAddrs()
 	}
 }
+
+func TestConnAs(t *testing.T) {
+	// t.Run("quic conns", func(t *testing.T) {
+	// 	h1, err := New(ListenAddrStrings(
+	// 		"/ip4/0.0.0.0/udp/0/quic-v1",
+	// 	))
+	// 	require.NoError(t, err)
+	// 	defer h1.Close()
+	// 	h2, err := New(ListenAddrStrings(
+	// 		"/ip4/0.0.0.0/udp/0/quic-v1",
+	// 	))
+	// 	require.NoError(t, err)
+	// 	defer h2.Close()
+	// 	err = h1.Connect(context.Background(), peer.AddrInfo{
+	// 		ID:    h2.ID(),
+	// 		Addrs: h2.Addrs(),
+	// 	})
+	// 	require.NoError(t, err)
+	// 	c := h1.Network().ConnsToPeer(h2.ID())[0]
+
+	// 	var quicConn *quicgo.Conn
+	// 	require.True(t, c.As(&quicConn))
+	// })
+
+	// t.Run("TCP conn with metrics", func(t *testing.T) {
+	// 	h1, err := New(ListenAddrStrings(
+	// 		"/ip4/0.0.0.0/tcp/0",
+	// 	), Transport(tcp.NewTCPTransport, tcp.WithMetrics()))
+	// 	require.NoError(t, err)
+	// 	defer h1.Close()
+	// 	h2, err := New(ListenAddrStrings(
+	// 		"/ip4/0.0.0.0/tcp/0",
+	// 	), Transport(tcp.NewTCPTransport, tcp.WithMetrics()))
+	// 	require.NoError(t, err)
+	// 	defer h2.Close()
+	// 	err = h1.Connect(context.Background(), peer.AddrInfo{
+	// 		ID:    h2.ID(),
+	// 		Addrs: h2.Addrs(),
+	// 	})
+	// 	require.NoError(t, err)
+	// 	c := h1.Network().ConnsToPeer(h2.ID())[0]
+
+	// 	var yamuxSession *yamux.Session
+	// 	require.True(t, c.As(&yamuxSession))
+	// })
+
+	type testCase struct {
+		name       string
+		listenAddr string
+		testAs     func(t *testing.T, c network.Conn)
+	}
+
+	testCases := []testCase{
+		{
+			"QUIC",
+			"/ip4/0.0.0.0/udp/0/quic-v1",
+			func(t *testing.T, c network.Conn) {
+				var quicConn *quicgo.Conn
+				require.True(t, c.As(&quicConn))
+			},
+		},
+		{
+			"TCP+Yamux",
+			"/ip4/0.0.0.0/tcp/0",
+			func(t *testing.T, c network.Conn) {
+				var yamuxSession *yamux.Session
+				require.True(t, c.As(&yamuxSession))
+			},
+		},
+		{
+			"WebRTC",
+			"/ip4/0.0.0.0/udp/0/webrtc-direct",
+			func(t *testing.T, c network.Conn) {
+				var webrtcPC *webrtc.PeerConnection
+				require.True(t, c.As(&webrtcPC))
+			},
+		},
+		{
+			"WebTransport Session",
+			"/ip4/0.0.0.0/udp/0/quic-v1/webtransport",
+			func(t *testing.T, c network.Conn) {
+				var s *wtgo.Session
+				require.True(t, c.As(&s))
+			},
+		},
+		{
+			"WebTransport QUIC Conn",
+			"/ip4/0.0.0.0/udp/0/quic-v1/webtransport",
+			func(t *testing.T, c network.Conn) {
+				var quicConn *quicgo.Conn
+				require.True(t, c.As(&quicConn))
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			h1, err := New(ListenAddrStrings(
+				tc.listenAddr,
+			))
+			require.NoError(t, err)
+			defer h1.Close()
+			h2, err := New(ListenAddrStrings(
+				tc.listenAddr,
+			))
+			require.NoError(t, err)
+			defer h2.Close()
+			err = h1.Connect(context.Background(), peer.AddrInfo{
+				ID:    h2.ID(),
+				Addrs: h2.Addrs(),
+			})
+			require.NoError(t, err)
+			c := h1.Network().ConnsToPeer(h2.ID())[0]
+			tc.testAs(t, c)
+		})
+	}
+}

p2p/muxer/yamux/conn.go

@@ -13,6 +13,14 @@ type conn yamux.Session
 
 var _ network.MuxedConn = &conn{}
 
+func (c *conn) As(target any) bool {
+	if t, ok := target.(**yamux.Session); ok {
+		*t = (*yamux.Session)(c)
+		return true
+	}
+	return false
+}
+
 // NewMuxedConn constructs a new MuxedConn from a yamux.Session.
 func NewMuxedConn(m *yamux.Session) network.MuxedConn {
 	return (*conn)(m)

p2p/net/connmgr/connmgr_test.go

@@ -818,6 +818,7 @@ func (m mockConn) NewStream(_ context.Context) (network.Stream, error) { panic("
 func (m mockConn) GetStreams() []network.Stream                        { panic("implement me") }
 func (m mockConn) Scope() network.ConnScope                            { panic("implement me") }
 func (m mockConn) ConnState() network.ConnectionState                  { return network.ConnectionState{} }
+func (m mockConn) As(target any) bool                                  { return false }
 
 func makeSegmentsWithPeerInfos(peerInfos peerInfos) *segments {
 	var s = func() *segments {

p2p/net/mock/mock_conn.go

@@ -86,6 +86,10 @@ func (c *conn) Close() error {
 	return nil
 }
 
+func (c *conn) As(target any) bool {
+	return false
+}
+
 func (c *conn) teardown() {
 	for _, s := range c.allStreams() {
 		s.Reset()

p2p/net/swarm/swarm.go

@@ -833,6 +833,10 @@ func wrapWithMetrics(capableConn transport.CapableConn, metricsTracer MetricsTra
 	return c
 }
 
+func (c *connWithMetrics) As(target any) bool {
+	return c.CapableConn.As(target)
+}
+
 func (c *connWithMetrics) completedHandshake() {
 	c.metricsTracer.CompletedHandshake(time.Since(c.opened), c.ConnState(), c.LocalMultiaddr())
 }

p2p/net/swarm/swarm_conn.go

@@ -42,6 +42,10 @@ type Conn struct {
 
 var _ network.Conn = &Conn{}
 
+func (c *Conn) As(target any) bool {
+	return c.conn.As(target)
+}
+
 func (c *Conn) IsClosed() bool {
 	return c.conn.IsClosed()
 }

p2p/net/upgrader/conn.go

@@ -23,6 +23,10 @@ type transportConn struct {
 
 var _ transport.CapableConn = &transportConn{}
 
+func (c *transportConn) As(target any) bool {
+	return c.MuxedConn.As(target)
+}
+
 func (t *transportConn) Transport() transport.Transport {
 	return t.transport
 }

p2p/transport/quic/conn.go

@@ -25,6 +25,15 @@ type conn struct {
 	remoteMultiaddr ma.Multiaddr
 }
 
+func (c *conn) As(target any) bool {
+	if t, ok := target.(**quic.Conn); ok {
+		*t = c.quicConn
+		return true
+	}
+
+	return false
+}
+
 var _ tpt.CapableConn = &conn{}
 
 // Close closes the connection.