Repository: libp2p/go-libp2p
Pull request #1633 (2022-07-01 19:54:28)

Pass endpoint information to resource manager's OpenConnection
Updates callsites to support allowlists.

Before merge:
All of this is related to managing dependencies across repos.

- [x] https://github.com/libp2p/go-libp2p-resource-manager/pull/56
- [x] https://github.com/libp2p/go-libp2p-testing/pull/61
- [x] Release new go-libp2p-testing 


Patch:
diff --git a/go.mod b/go.mod
index 4f8d90f9..55781115 100644
--- a/go.mod
+++ b/go.mod
@@ -20,10 +20,10 @@ require (
 	github.com/libp2p/go-buffer-pool v0.0.2
 	github.com/libp2p/go-eventbus v0.2.1
 	github.com/libp2p/go-libp2p-asn-util v0.2.0
-	github.com/libp2p/go-libp2p-core v0.18.0
+	github.com/libp2p/go-libp2p-core v0.19.0
 	github.com/libp2p/go-libp2p-peerstore v0.7.1
-	github.com/libp2p/go-libp2p-resource-manager v0.3.0
-	github.com/libp2p/go-libp2p-testing v0.9.2
+	github.com/libp2p/go-libp2p-resource-manager v0.4.0
+	github.com/libp2p/go-libp2p-testing v0.10.0
 	github.com/libp2p/go-mplex v0.7.0
 	github.com/libp2p/go-msgio v0.2.0
 	github.com/libp2p/go-nat v0.1.0
@@ -68,7 +68,7 @@ require (
 	github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
 	github.com/godbus/dbus/v5 v5.0.3 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/google/uuid v1.1.1 // indirect
+	github.com/google/uuid v1.1.2 // indirect
 	github.com/huin/goupnp v1.0.0 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/goprocess v0.1.4 // indirect
@@ -110,7 +110,7 @@ require (
 	golang.org/x/text v0.3.6 // indirect
 	golang.org/x/tools v0.1.5 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
-	google.golang.org/grpc v1.31.1 // indirect
+	google.golang.org/grpc v1.33.2 // indirect
 	google.golang.org/protobuf v1.26.0 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
diff --git a/go.sum b/go.sum
index 61be1d13..1e16fa6e 100644
--- a/go.sum
+++ b/go.sum
@@ -201,6 +201,7 @@ github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
@@ -220,8 +221,9 @@ github.com/google/pprof v0.0.0-20200229191704-1ebb73c60ed3/go.mod h1:ZgVRPoUq/hf
 github.com/google/pprof v0.0.0-20200430221834-fc25d7d30c6d/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/pprof v0.0.0-20200708004538-1a94d8640e99/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
-github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.1.2 h1:EVhdT+1Kseyi1/pUmXKaFxYsDNy9RQYkMWRH68J/W7Y=
+github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 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/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
@@ -324,16 +326,15 @@ github.com/libp2p/go-libp2p-asn-util v0.2.0/go.mod h1:WoaWxbHKBymSN41hWSq/lGKJEc
 github.com/libp2p/go-libp2p-core v0.2.0/go.mod h1:X0eyB0Gy93v0DZtSYbEM7RnMChm9Uv3j7yRXjO77xSI=
 github.com/libp2p/go-libp2p-core v0.5.7/go.mod h1:txwbVEhHEXikXn9gfC7/UDDw7rkxuX0bJvM49Ykaswo=
 github.com/libp2p/go-libp2p-core v0.12.0/go.mod h1:ECdxehoYosLYHgDDFa2N4yE8Y7aQRAMf0sX9mf2sbGg=
-github.com/libp2p/go-libp2p-core v0.14.0/go.mod h1:tLasfcVdTXnixsLB0QYaT1syJOhsbrhG7q6pGrHtBg8=
-github.com/libp2p/go-libp2p-core v0.18.0 h1:giGYfMfEpHnRJdOiEH3+YLm8xch/wBTRxLBnHUqa5c0=
-github.com/libp2p/go-libp2p-core v0.18.0/go.mod h1:AkA+FUKQfYt1FLNef5fOPlo/naAWjKy/RCjkcPjqzYg=
+github.com/libp2p/go-libp2p-core v0.19.0 h1:KDw7hanmh0EuVdZqsHCAzmkdiYMk5uR5h0UGSCVTxSU=
+github.com/libp2p/go-libp2p-core v0.19.0/go.mod h1:AkA+FUKQfYt1FLNef5fOPlo/naAWjKy/RCjkcPjqzYg=
 github.com/libp2p/go-libp2p-peerstore v0.7.1 h1:7FpALlqR+3+oOBXdzm3AVt0vjMYLW1b7jM03E4iEHlw=
 github.com/libp2p/go-libp2p-peerstore v0.7.1/go.mod h1:cdUWTHro83vpg6unCpGUr8qJoX3e93Vy8o97u5ppIM0=
-github.com/libp2p/go-libp2p-resource-manager v0.3.0 h1:2+cYxUNi33tcydsVLt6K5Fv2E3OTiVeafltecAj15E0=
-github.com/libp2p/go-libp2p-resource-manager v0.3.0/go.mod h1:K+eCkiapf+ey/LADO4TaMpMTP9/Qde/uLlrnRqV4PLQ=
+github.com/libp2p/go-libp2p-resource-manager v0.4.0 h1:+/gSDLSJ+n8qHVdMoY7wfrk3EvvL9Ktw6sAyKKZPQRw=
+github.com/libp2p/go-libp2p-resource-manager v0.4.0/go.mod h1:+5QPxFLRXYlRDZ0P1bPKE7zyZDvex5TLVOqePwRmwfc=
 github.com/libp2p/go-libp2p-testing v0.1.1/go.mod h1:xaZWMJrPUM5GlDBxCeGUi7kI4eqnjVyavGroI2nxEM0=
-github.com/libp2p/go-libp2p-testing v0.9.2 h1:dCpODRtRaDZKF8HXT9qqqgON+OMEB423Knrgeod8j84=
-github.com/libp2p/go-libp2p-testing v0.9.2/go.mod h1:Td7kbdkWqYTJYQGTwzlgXwaqldraIanyjuRiAbK/XQU=
+github.com/libp2p/go-libp2p-testing v0.10.0 h1:LO7wuUPPNAe1D1s0HZ+9WoROaGIn/MEl1wtugXuTRzg=
+github.com/libp2p/go-libp2p-testing v0.10.0/go.mod h1:jJ4fiJwyZ3UlPTLcnz/sEmPPSviQ79Q0MVD/CykzrP0=
 github.com/libp2p/go-maddr-filter v0.1.0/go.mod h1:VzZhTXkMucEGGEOSKddrwGiOv0tUhgnKqNEmIAz/bPU=
 github.com/libp2p/go-mplex v0.7.0 h1:BDhFZdlk5tbr0oyFq/xv/NPGfjbnrsDam1EvutpBDbY=
 github.com/libp2p/go-mplex v0.7.0/go.mod h1:rW8ThnRcYWft/Jb2jeORBmPd6xuG3dGxWN/W168L9EU=
@@ -559,6 +560,7 @@ github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXf
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
@@ -580,6 +582,7 @@ go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
+go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -688,6 +691,7 @@ golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81R
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
@@ -928,8 +932,9 @@ google.golang.org/grpc v1.28.0/go.mod h1:rpkK4SK4GF4Ach/+MFLZUBavHOvF2JJB5uozKKa
 google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3IjizoKk=
 google.golang.org/grpc v1.30.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.31.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
-google.golang.org/grpc v1.31.1 h1:SfXqXS5hkufcdZ/mHtYCh53P2b+92WQq/DZcKLgsFRs=
 google.golang.org/grpc v1.31.1/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
+google.golang.org/grpc v1.33.2 h1:EQyQC3sa8M+p6Ulc8yy9SWSS2GVwyRc83gAbG8lrl4o=
+google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
diff --git a/p2p/net/upgrader/listener.go b/p2p/net/upgrader/listener.go
index fa94e8e6..cc462b53 100644
--- a/p2p/net/upgrader/listener.go
+++ b/p2p/net/upgrader/listener.go
@@ -93,7 +93,7 @@ func (l *listener) handleIncoming() {
 			continue
 		}
 
-		connScope, err := l.rcmgr.OpenConnection(network.DirInbound, true)
+		connScope, err := l.rcmgr.OpenConnection(network.DirInbound, true, maconn.RemoteMultiaddr())
 		if err != nil {
 			log.Debugw("resource manager blocked accept of new connection", "error", err)
 			if err := maconn.Close(); err != nil {
diff --git a/p2p/protocol/circuitv2/client/transport.go b/p2p/protocol/circuitv2/client/transport.go
index 9bdb53f6..cd1c302d 100644
--- a/p2p/protocol/circuitv2/client/transport.go
+++ b/p2p/protocol/circuitv2/client/transport.go
@@ -50,7 +50,7 @@ var _ transport.Transport = (*Client)(nil)
 var _ io.Closer = (*Client)(nil)
 
 func (c *Client) Dial(ctx context.Context, a ma.Multiaddr, p peer.ID) (transport.CapableConn, error) {
-	connScope, err := c.host.Network().ResourceManager().OpenConnection(network.DirOutbound, false)
+	connScope, err := c.host.Network().ResourceManager().OpenConnection(network.DirOutbound, false, a)
 	if err != nil {
 		return nil, err
 	}
diff --git a/p2p/protocol/internal/circuitv1-deprecated/dial.go b/p2p/protocol/internal/circuitv1-deprecated/dial.go
index 703521ef..f454bcc0 100644
--- a/p2p/protocol/internal/circuitv1-deprecated/dial.go
+++ b/p2p/protocol/internal/circuitv1-deprecated/dial.go
@@ -17,7 +17,7 @@ func (d *RelayTransport) Dial(ctx context.Context, a ma.Multiaddr, p peer.ID) (t
 		return nil, err
 	}
 	c.tagHop()
-	scope, _ := network.NullResourceManager.OpenConnection(network.DirOutbound, false)
+	scope, _ := network.NullResourceManager.OpenConnection(network.DirOutbound, false, a)
 	return d.upgrader.Upgrade(ctx, d, c, network.DirOutbound, p, scope)
 }
 
diff --git a/p2p/transport/quic/listener.go b/p2p/transport/quic/listener.go
index 293fa031..955c0f6c 100644
--- a/p2p/transport/quic/listener.go
+++ b/p2p/transport/quic/listener.go
@@ -98,7 +98,12 @@ func (l *listener) Accept() (tpt.CapableConn, error) {
 }
 
 func (l *listener) setupConn(qconn quic.Connection) (*conn, error) {
-	connScope, err := l.rcmgr.OpenConnection(network.DirInbound, false)
+	remoteMultiaddr, err := toQuicMultiaddr(qconn.RemoteAddr())
+	if err != nil {
+		return nil, err
+	}
+
+	connScope, err := l.rcmgr.OpenConnection(network.DirInbound, false, remoteMultiaddr)
 	if err != nil {
 		log.Debugw("resource manager blocked incoming connection", "addr", qconn.RemoteAddr(), "error", err)
 		return nil, err
@@ -122,11 +127,6 @@ func (l *listener) setupConn(qconn quic.Connection) (*conn, error) {
 		connScope.Done()
 		return nil, err
 	}
-	remoteMultiaddr, err := toQuicMultiaddr(qconn.RemoteAddr())
-	if err != nil {
-		connScope.Done()
-		return nil, err
-	}
 
 	l.conn.IncreaseCount()
 	return &conn{
diff --git a/p2p/transport/quic/transport.go b/p2p/transport/quic/transport.go
index a36b3e58..353f4213 100644
--- a/p2p/transport/quic/transport.go
+++ b/p2p/transport/quic/transport.go
@@ -201,7 +201,7 @@ func (t *transport) Dial(ctx context.Context, raddr ma.Multiaddr, p peer.ID) (tp
 		return t.holePunch(ctx, netw, addr, p)
 	}
 
-	scope, err := t.rcmgr.OpenConnection(network.DirOutbound, false)
+	scope, err := t.rcmgr.OpenConnection(network.DirOutbound, false, raddr)
 	if err != nil {
 		log.Debugw("resource manager blocked outgoing connection", "peer", p, "addr", raddr, "error", err)
 		return nil, err
diff --git a/p2p/transport/tcp/tcp.go b/p2p/transport/tcp/tcp.go
index 290717b2..120feee8 100644
--- a/p2p/transport/tcp/tcp.go
+++ b/p2p/transport/tcp/tcp.go
@@ -170,7 +170,7 @@ func (t *TcpTransport) maDial(ctx context.Context, raddr ma.Multiaddr) (manet.Co
 
 // Dial dials the peer at the remote address.
 func (t *TcpTransport) Dial(ctx context.Context, raddr ma.Multiaddr, p peer.ID) (transport.CapableConn, error) {
-	connScope, err := t.rcmgr.OpenConnection(network.DirOutbound, true)
+	connScope, err := t.rcmgr.OpenConnection(network.DirOutbound, true, raddr)
 	if err != nil {
 		log.Debugw("resource manager blocked outgoing connection", "peer", p, "addr", raddr, "error", err)
 		return nil, err
diff --git a/p2p/transport/websocket/websocket.go b/p2p/transport/websocket/websocket.go
index 851da9ed..af74ccfb 100644
--- a/p2p/transport/websocket/websocket.go
+++ b/p2p/transport/websocket/websocket.go
@@ -101,7 +101,7 @@ func (t *WebsocketTransport) Proxy() bool {
 }
 
 func (t *WebsocketTransport) Dial(ctx context.Context, raddr ma.Multiaddr, p peer.ID) (transport.CapableConn, error) {
-	connScope, err := t.rcmgr.OpenConnection(network.DirOutbound, true)
+	connScope, err := t.rcmgr.OpenConnection(network.DirOutbound, true, raddr)
 	if err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/p2p/muxer/mplex/transport_test.go b/p2p/muxer/mplex/transport_test.go
index d4474526..1226b5d7 100644
--- a/p2p/muxer/mplex/transport_test.go
+++ b/p2p/muxer/mplex/transport_test.go
@@ -6,7 +6,7 @@ import (
 	"testing"
 
 	"github.com/libp2p/go-libp2p-core/network"
-	test "github.com/libp2p/go-libp2p-testing/suites/mux"
+	test "github.com/libp2p/go-libp2p/p2p/muxer/testsuite"
 )
 
 func TestDefaultTransport(t *testing.T) {
diff --git a/p2p/muxer/testsuite/mux.go b/p2p/muxer/testsuite/mux.go
new file mode 100644
index 00000000..fbfea2c0
--- /dev/null
+++ b/p2p/muxer/testsuite/mux.go
@@ -0,0 +1,650 @@
+package mux
+
+import (
+	"bytes"
+	"context"
+	crand "crypto/rand"
+	"fmt"
+	"io"
+	mrand "math/rand"
+	"net"
+	"reflect"
+	"runtime"
+	"runtime/debug"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/network"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-testing/ci"
+
+	"github.com/stretchr/testify/require"
+)
+
+var randomness []byte
+var Subtests map[string]TransportTest
+
+func init() {
+	// read 1MB of randomness
+	randomness = make([]byte, 1<<20)
+	if _, err := crand.Read(randomness); err != nil {
+		panic(err)
+	}
+
+	Subtests = make(map[string]TransportTest)
+	for _, f := range subtests {
+		Subtests[getFunctionName(f)] = f
+	}
+}
+
+func getFunctionName(i interface{}) string {
+	return runtime.FuncForPC(reflect.ValueOf(i).Pointer()).Name()
+}
+
+type peerScope struct {
+	mx     sync.Mutex
+	memory int
+}
+
+func (p *peerScope) ReserveMemory(size int, _ uint8) error {
+	p.mx.Lock()
+	p.memory += size
+	p.mx.Unlock()
+	return nil
+}
+
+func (p *peerScope) ReleaseMemory(size int) {
+	p.mx.Lock()
+	defer p.mx.Unlock()
+	if p.memory < size {
+		panic(fmt.Sprintf("tried to release too much memory: %d (current: %d)", size, p.memory))
+	}
+	p.memory -= size
+}
+
+// Check checks that we don't have any more reserved memory.
+func (p *peerScope) Check(t *testing.T) {
+	p.mx.Lock()
+	defer p.mx.Unlock()
+	require.Zero(t, p.memory, "expected all reserved memory to have been released")
+}
+
+func (p *peerScope) Stat() network.ScopeStat                       { return network.ScopeStat{} }
+func (p *peerScope) BeginSpan() (network.ResourceScopeSpan, error) { return nil, nil }
+func (p *peerScope) Peer() peer.ID                                 { panic("implement me") }
+
+var _ network.PeerScope = &peerScope{}
+
+type Options struct {
+	tr        network.Multiplexer
+	connNum   int
+	streamNum int
+	msgNum    int
+	msgMin    int
+	msgMax    int
+}
+
+func randBuf(size int) []byte {
+	n := len(randomness) - size
+	if size < 1 {
+		panic(fmt.Errorf("requested too large buffer (%d). max is %d", size, len(randomness)))
+	}
+
+	start := mrand.Intn(n)
+	return randomness[start : start+size]
+}
+
+func checkErr(t *testing.T, err error) {
+	if err != nil {
+		debug.PrintStack()
+		t.Fatal(err)
+	}
+}
+
+func echoStream(s network.MuxedStream) {
+	defer s.Close()
+	io.Copy(s, s) // echo everything
+}
+
+func GoServe(t *testing.T, tr network.Multiplexer, l net.Listener) (done func()) {
+	closed := make(chan struct{}, 1)
+
+	go func() {
+		for {
+			c1, err := l.Accept()
+			if err != nil {
+				select {
+				case <-closed:
+					return // closed naturally.
+				default:
+					checkErr(t, err)
+				}
+			}
+
+			sc1, err := tr.NewConn(c1, true, nil)
+			checkErr(t, err)
+			go func() {
+				for {
+					str, err := sc1.AcceptStream()
+					if err != nil {
+						break
+					}
+					go echoStream(str)
+				}
+			}()
+		}
+	}()
+
+	return func() {
+		closed <- struct{}{}
+	}
+}
+
+func SubtestSimpleWrite(t *testing.T, tr network.Multiplexer) {
+	l, err := net.Listen("tcp", "localhost:0")
+	checkErr(t, err)
+	done := GoServe(t, tr, l)
+	defer done()
+
+	nc1, err := net.Dial("tcp", l.Addr().String())
+	checkErr(t, err)
+	defer nc1.Close()
+
+	scope := &peerScope{}
+	c1, err := tr.NewConn(nc1, false, scope)
+	checkErr(t, err)
+	defer func() {
+		c1.Close()
+		scope.Check(t)
+	}()
+
+	// serve the outgoing conn, because some muxers assume
+	// that we _always_ call serve. (this is an error?)
+	go c1.AcceptStream()
+
+	s1, err := c1.OpenStream(context.Background())
+	checkErr(t, err)
+	defer s1.Close()
+
+	buf1 := randBuf(4096)
+	_, err = s1.Write(buf1)
+	checkErr(t, err)
+
+	buf2 := make([]byte, len(buf1))
+	_, err = io.ReadFull(s1, buf2)
+	checkErr(t, err)
+
+	require.Equal(t, buf1, buf2)
+}
+
+func SubtestStress(t *testing.T, opt Options) {
+	msgsize := 1 << 11
+	errs := make(chan error) // dont block anything.
+
+	rateLimitN := 5000 // max of 5k funcs, because -race has 8k max.
+	rateLimitChan := make(chan struct{}, rateLimitN)
+	for i := 0; i < rateLimitN; i++ {
+		rateLimitChan <- struct{}{}
+	}
+
+	rateLimit := func(f func()) {
+		<-rateLimitChan
+		f()
+		rateLimitChan <- struct{}{}
+	}
+
+	writeStream := func(s network.MuxedStream, bufs chan<- []byte) {
+		for i := 0; i < opt.msgNum; i++ {
+			buf := randBuf(msgsize)
+			bufs <- buf
+			if _, err := s.Write(buf); err != nil {
+				errs <- fmt.Errorf("s.Write(buf): %s", err)
+				continue
+			}
+		}
+	}
+
+	readStream := func(s network.MuxedStream, bufs <-chan []byte) {
+		buf2 := make([]byte, msgsize)
+		for buf1 := range bufs {
+			if _, err := io.ReadFull(s, buf2); err != nil {
+				errs <- fmt.Errorf("io.ReadFull(s, buf2): %s", err)
+				continue
+			}
+			if !bytes.Equal(buf1, buf2) {
+				errs <- fmt.Errorf("buffers not equal (%x != %x)", buf1[:3], buf2[:3])
+			}
+		}
+	}
+
+	openStreamAndRW := func(c network.MuxedConn) {
+		s, err := c.OpenStream(context.Background())
+		if err != nil {
+			errs <- fmt.Errorf("failed to create NewStream: %s", err)
+			return
+		}
+
+		bufs := make(chan []byte, opt.msgNum)
+		go func() {
+			writeStream(s, bufs)
+			close(bufs)
+		}()
+
+		readStream(s, bufs)
+		s.Close()
+	}
+
+	openConnAndRW := func() {
+		l, err := net.Listen("tcp", "localhost:0")
+		checkErr(t, err)
+		done := GoServe(t, opt.tr, l)
+		defer done()
+
+		nla := l.Addr()
+		nc, err := net.Dial(nla.Network(), nla.String())
+		checkErr(t, err)
+		if err != nil {
+			t.Fatal(fmt.Errorf("net.Dial(%s, %s): %s", nla.Network(), nla.String(), err))
+			return
+		}
+
+		scope := &peerScope{}
+		c, err := opt.tr.NewConn(nc, false, scope)
+		if err != nil {
+			t.Fatal(fmt.Errorf("a.AddConn(%s <--> %s): %s", nc.LocalAddr(), nc.RemoteAddr(), err))
+			return
+		}
+
+		// serve the outgoing conn, because some muxers assume
+		// that we _always_ call serve. (this is an error?)
+		go func() {
+			for {
+				str, err := c.AcceptStream()
+				if err != nil {
+					break
+				}
+				go echoStream(str)
+			}
+		}()
+
+		var wg sync.WaitGroup
+		for i := 0; i < opt.streamNum; i++ {
+			wg.Add(1)
+			go rateLimit(func() {
+				defer wg.Done()
+				openStreamAndRW(c)
+			})
+		}
+		wg.Wait()
+		c.Close()
+		scope.Check(t)
+	}
+
+	openConnsAndRW := func() {
+		var wg sync.WaitGroup
+		for i := 0; i < opt.connNum; i++ {
+			wg.Add(1)
+			go rateLimit(func() {
+				defer wg.Done()
+				openConnAndRW()
+			})
+		}
+		wg.Wait()
+	}
+
+	go func() {
+		openConnsAndRW()
+		close(errs) // done
+	}()
+
+	for err := range errs {
+		t.Error(err)
+	}
+
+}
+
+func tcpPipe(t *testing.T) (net.Conn, net.Conn) {
+	list, err := net.Listen("tcp", "0.0.0.0:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	con1, err := net.Dial("tcp", list.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	con2, err := list.Accept()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return con1, con2
+}
+
+func SubtestStreamOpenStress(t *testing.T, tr network.Multiplexer) {
+	wg := new(sync.WaitGroup)
+
+	a, b := tcpPipe(t)
+	defer a.Close()
+	defer b.Close()
+
+	defer wg.Wait()
+
+	wg.Add(1)
+	count := 10000
+	workers := 5
+	go func() {
+		defer wg.Done()
+		muxa, err := tr.NewConn(a, true, nil)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		stress := func() {
+			defer wg.Done()
+			for i := 0; i < count; i++ {
+				s, err := muxa.OpenStream(context.Background())
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				err = s.CloseWrite()
+				if err != nil {
+					t.Error(err)
+				}
+				n, err := s.Read([]byte{0})
+				if n != 0 {
+					t.Error("expected to read no bytes")
+				}
+				if err != io.EOF {
+					t.Errorf("expected an EOF, got %s", err)
+				}
+			}
+		}
+
+		for i := 0; i < workers; i++ {
+			wg.Add(1)
+			go stress()
+		}
+	}()
+
+	scope := &peerScope{}
+	muxb, err := tr.NewConn(b, false, scope)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		muxb.Close()
+		scope.Check(t)
+	}()
+
+	time.Sleep(time.Millisecond * 50)
+
+	wg.Add(1)
+	recv := make(chan struct{}, count*workers)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < count*workers; i++ {
+			str, err := muxb.AcceptStream()
+			if err != nil {
+				break
+			}
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				str.Close()
+				select {
+				case recv <- struct{}{}:
+				default:
+					t.Error("too many stream")
+				}
+			}()
+		}
+	}()
+
+	timeout := time.Second * 10
+	if ci.IsRunning() {
+		timeout *= 10
+	}
+
+	limit := time.After(timeout)
+	for i := 0; i < count*workers; i++ {
+		select {
+		case <-recv:
+		case <-limit:
+			t.Fatal("timed out receiving streams")
+		}
+	}
+
+	wg.Wait()
+}
+
+func SubtestStreamReset(t *testing.T, tr network.Multiplexer) {
+	wg := new(sync.WaitGroup)
+	defer wg.Wait()
+
+	a, b := tcpPipe(t)
+	defer a.Close()
+	defer b.Close()
+
+	wg.Add(1)
+	scopea := &peerScope{}
+	muxa, err := tr.NewConn(a, true, scopea)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer func() {
+		muxa.Close()
+		scopea.Check(t)
+	}()
+
+	go func() {
+		defer wg.Done()
+		s, err := muxa.OpenStream(context.Background())
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		time.Sleep(time.Millisecond * 50)
+
+		_, err = s.Write([]byte("foo"))
+		if err != network.ErrReset {
+			t.Error("should have been stream reset")
+		}
+		s.Close()
+	}()
+
+	scopeb := &peerScope{}
+	muxb, err := tr.NewConn(b, false, scopeb)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		muxb.Close()
+		scopeb.Check(t)
+	}()
+
+	str, err := muxb.AcceptStream()
+	checkErr(t, err)
+	str.Reset()
+
+	wg.Wait()
+}
+
+// check that Close also closes the underlying net.Conn
+func SubtestWriteAfterClose(t *testing.T, tr network.Multiplexer) {
+	a, b := tcpPipe(t)
+
+	scopea := &peerScope{}
+	muxa, err := tr.NewConn(a, true, scopea)
+	checkErr(t, err)
+
+	scopeb := &peerScope{}
+	muxb, err := tr.NewConn(b, false, scopeb)
+	checkErr(t, err)
+
+	checkErr(t, muxa.Close())
+	scopea.Check(t)
+	checkErr(t, muxb.Close())
+	scopeb.Check(t)
+
+	// make sure the underlying net.Conn was closed
+	if _, err := a.Write([]byte("foobar")); err == nil || !strings.Contains(err.Error(), "use of closed network connection") {
+		t.Fatal("write should have failed")
+	}
+	if _, err := b.Write([]byte("foobar")); err == nil || !strings.Contains(err.Error(), "use of closed network connection") {
+		t.Fatal("write should have failed")
+	}
+}
+
+func SubtestStreamLeftOpen(t *testing.T, tr network.Multiplexer) {
+	a, b := tcpPipe(t)
+
+	const numStreams = 10
+	const dataLen = 50 * 1024
+
+	scopea := &peerScope{}
+	muxa, err := tr.NewConn(a, true, scopea)
+	checkErr(t, err)
+
+	scopeb := &peerScope{}
+	muxb, err := tr.NewConn(b, false, scopeb)
+	checkErr(t, err)
+
+	var wg sync.WaitGroup
+	wg.Add(1 + numStreams)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < numStreams; i++ {
+			stra, err := muxa.OpenStream(context.Background())
+			checkErr(t, err)
+			go func() {
+				defer wg.Done()
+				_, err = stra.Write(randBuf(dataLen))
+				checkErr(t, err)
+				// do NOT close or reset the stream
+			}()
+		}
+	}()
+
+	wg.Add(1 + numStreams)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < numStreams; i++ {
+			str, err := muxb.AcceptStream()
+			checkErr(t, err)
+			go func() {
+				defer wg.Done()
+				_, err = io.ReadFull(str, make([]byte, dataLen))
+				checkErr(t, err)
+			}()
+		}
+	}()
+
+	// Now we have a bunch of open streams.
+	// Make sure that their memory is returned when we close the connection.
+	wg.Wait()
+
+	muxa.Close()
+	scopea.Check(t)
+	muxb.Close()
+	scopeb.Check(t)
+}
+
+func SubtestStress1Conn1Stream1Msg(t *testing.T, tr network.Multiplexer) {
+	SubtestStress(t, Options{
+		tr:        tr,
+		connNum:   1,
+		streamNum: 1,
+		msgNum:    1,
+		msgMax:    100,
+		msgMin:    100,
+	})
+}
+
+func SubtestStress1Conn1Stream100Msg(t *testing.T, tr network.Multiplexer) {
+	SubtestStress(t, Options{
+		tr:        tr,
+		connNum:   1,
+		streamNum: 1,
+		msgNum:    100,
+		msgMax:    100,
+		msgMin:    100,
+	})
+}
+
+func SubtestStress1Conn100Stream100Msg(t *testing.T, tr network.Multiplexer) {
+	SubtestStress(t, Options{
+		tr:        tr,
+		connNum:   1,
+		streamNum: 100,
+		msgNum:    100,
+		msgMax:    100,
+		msgMin:    100,
+	})
+}
+
+func SubtestStress10Conn10Stream50Msg(t *testing.T, tr network.Multiplexer) {
+	SubtestStress(t, Options{
+		tr:        tr,
+		connNum:   10,
+		streamNum: 10,
+		msgNum:    50,
+		msgMax:    100,
+		msgMin:    100,
+	})
+}
+
+func SubtestStress1Conn1000Stream10Msg(t *testing.T, tr network.Multiplexer) {
+	SubtestStress(t, Options{
+		tr:        tr,
+		connNum:   1,
+		streamNum: 1000,
+		msgNum:    10,
+		msgMax:    100,
+		msgMin:    100,
+	})
+}
+
+func SubtestStress1Conn100Stream100Msg10MB(t *testing.T, tr network.Multiplexer) {
+	SubtestStress(t, Options{
+		tr:        tr,
+		connNum:   1,
+		streamNum: 100,
+		msgNum:    100,
+		msgMax:    10000,
+		msgMin:    1000,
+	})
+}
+
+// Subtests are all the subtests run by SubtestAll
+var subtests = []TransportTest{
+	SubtestSimpleWrite,
+	SubtestWriteAfterClose,
+	SubtestStress1Conn1Stream1Msg,
+	SubtestStress1Conn1Stream100Msg,
+	SubtestStress1Conn100Stream100Msg,
+	SubtestStress10Conn10Stream50Msg,
+	SubtestStress1Conn1000Stream10Msg,
+	SubtestStress1Conn100Stream100Msg10MB,
+	SubtestStreamOpenStress,
+	SubtestStreamReset,
+	SubtestStreamLeftOpen,
+}
+
+// SubtestAll runs all the stream multiplexer tests against the target
+// transport.
+func SubtestAll(t *testing.T, tr network.Multiplexer) {
+	for name, f := range Subtests {
+		t.Run(name, func(t *testing.T) {
+			f(t, tr)
+		})
+	}
+}
+
+// TransportTest is a stream multiplex transport test case
+type TransportTest func(t *testing.T, tr network.Multiplexer)
diff --git a/p2p/muxer/yamux/transport_test.go b/p2p/muxer/yamux/transport_test.go
index 35818966..4a68ac9c 100644
--- a/p2p/muxer/yamux/transport_test.go
+++ b/p2p/muxer/yamux/transport_test.go
@@ -3,7 +3,7 @@ package yamux
 import (
 	"testing"
 
-	tmux "github.com/libp2p/go-libp2p-testing/suites/mux"
+	tmux "github.com/libp2p/go-libp2p/p2p/muxer/testsuite"
 )
 
 func TestDefaultTransport(t *testing.T) {
diff --git a/p2p/net/upgrader/listener_test.go b/p2p/net/upgrader/listener_test.go
index 1ea80d68..929df0bb 100644
--- a/p2p/net/upgrader/listener_test.go
+++ b/p2p/net/upgrader/listener_test.go
@@ -362,7 +362,7 @@ func TestListenerResourceManagement(t *testing.T) {
 
 	connScope := mocknetwork.NewMockConnManagementScope(ctrl)
 	gomock.InOrder(
-		rcmgr.EXPECT().OpenConnection(network.DirInbound, true).Return(connScope, nil),
+		rcmgr.EXPECT().OpenConnection(network.DirInbound, true, gomock.Not(ln.Multiaddr())).Return(connScope, nil),
 		connScope.EXPECT().PeerScope(),
 		connScope.EXPECT().SetPeer(id),
 		connScope.EXPECT().PeerScope(),
@@ -385,7 +385,7 @@ func TestListenerResourceManagementDenied(t *testing.T) {
 	id, upgrader := createUpgrader(t, upgrader.WithResourceManager(rcmgr))
 	ln := createListener(t, upgrader)
 
-	rcmgr.EXPECT().OpenConnection(network.DirInbound, true).Return(nil, errors.New("nope"))
+	rcmgr.EXPECT().OpenConnection(network.DirInbound, true, gomock.Not(ln.Multiaddr())).Return(nil, errors.New("nope"))
 	_, err := dial(t, upgrader, ln.Multiaddr(), id, network.NullScope)
 	require.Error(t, err)
 
diff --git a/p2p/transport/quic/conn_test.go b/p2p/transport/quic/conn_test.go
index 628f4555..7d9e3750 100644
--- a/p2p/transport/quic/conn_test.go
+++ b/p2p/transport/quic/conn_test.go
@@ -122,7 +122,7 @@ func TestResourceManagerSuccess(t *testing.T) {
 	connChan := make(chan tpt.CapableConn)
 	serverConnScope := mocknetwork.NewMockConnManagementScope(ctrl)
 	go func() {
-		serverRcmgr.EXPECT().OpenConnection(network.DirInbound, false).Return(serverConnScope, nil)
+		serverRcmgr.EXPECT().OpenConnection(network.DirInbound, false, gomock.Not(ln.Multiaddr())).Return(serverConnScope, nil)
 		serverConnScope.EXPECT().SetPeer(clientID)
 		serverConn, err := ln.Accept()
 		require.NoError(t, err)
@@ -130,7 +130,7 @@ func TestResourceManagerSuccess(t *testing.T) {
 	}()
 
 	connScope := mocknetwork.NewMockConnManagementScope(ctrl)
-	clientRcmgr.EXPECT().OpenConnection(network.DirOutbound, false).Return(connScope, nil)
+	clientRcmgr.EXPECT().OpenConnection(network.DirOutbound, false, ln.Multiaddr()).Return(connScope, nil)
 	connScope.EXPECT().SetPeer(serverID)
 	conn, err := clientTransport.Dial(context.Background(), ln.Multiaddr(), serverID)
 	require.NoError(t, err)
@@ -153,13 +153,15 @@ func TestResourceManagerDialDenied(t *testing.T) {
 	defer clientTransport.(io.Closer).Close()
 
 	connScope := mocknetwork.NewMockConnManagementScope(ctrl)
-	rcmgr.EXPECT().OpenConnection(network.DirOutbound, false).Return(connScope, nil)
+	target := ma.StringCast("/ip4/127.0.0.1/udp/1234/quic")
+
+	rcmgr.EXPECT().OpenConnection(network.DirOutbound, false, target).Return(connScope, nil)
 	rerr := errors.New("nope")
 	p := peer.ID("server")
 	connScope.EXPECT().SetPeer(p).Return(rerr)
 	connScope.EXPECT().Done()
 
-	_, err = clientTransport.Dial(context.Background(), ma.StringCast("/ip4/127.0.0.1/udp/1234/quic"), p)
+	_, err = clientTransport.Dial(context.Background(), target, p)
 	require.ErrorIs(t, err, rerr)
 }
 
@@ -178,7 +180,7 @@ func TestResourceManagerAcceptDenied(t *testing.T) {
 	serverConnScope := mocknetwork.NewMockConnManagementScope(ctrl)
 	rerr := errors.New("denied")
 	gomock.InOrder(
-		serverRcmgr.EXPECT().OpenConnection(network.DirInbound, false).Return(serverConnScope, nil),
+		serverRcmgr.EXPECT().OpenConnection(network.DirInbound, false, gomock.Any()).Return(serverConnScope, nil),
 		serverConnScope.EXPECT().SetPeer(clientID).Return(rerr),
 		serverConnScope.EXPECT().Done(),
 	)
@@ -195,7 +197,7 @@ func TestResourceManagerAcceptDenied(t *testing.T) {
 	}()
 
 	clientConnScope := mocknetwork.NewMockConnManagementScope(ctrl)
-	clientRcmgr.EXPECT().OpenConnection(network.DirOutbound, false).Return(clientConnScope, nil)
+	clientRcmgr.EXPECT().OpenConnection(network.DirOutbound, false, ln.Multiaddr()).Return(clientConnScope, nil)
 	clientConnScope.EXPECT().SetPeer(serverID)
 	// In rare instances, the connection gating error will already occur on Dial.
 	// In that case, Done is called on the connection scope.
diff --git a/p2p/transport/tcp/tcp_test.go b/p2p/transport/tcp/tcp_test.go
index c0f02e2a..9874cade 100644
--- a/p2p/transport/tcp/tcp_test.go
+++ b/p2p/transport/tcp/tcp_test.go
@@ -70,7 +70,7 @@ func TestResourceManager(t *testing.T) {
 
 	t.Run("success", func(t *testing.T) {
 		scope := mocknetwork.NewMockConnManagementScope(ctrl)
-		rcmgr.EXPECT().OpenConnection(network.DirOutbound, true).Return(scope, nil)
+		rcmgr.EXPECT().OpenConnection(network.DirOutbound, true, ln.Multiaddr()).Return(scope, nil)
 		scope.EXPECT().SetPeer(peerA)
 		scope.EXPECT().PeerScope().Return(network.NullScope).AnyTimes() // called by the upgrader
 		conn, err := tb.Dial(context.Background(), ln.Multiaddr(), peerA)
@@ -81,14 +81,14 @@ func TestResourceManager(t *testing.T) {
 
 	t.Run("connection denied", func(t *testing.T) {
 		rerr := errors.New("nope")
-		rcmgr.EXPECT().OpenConnection(network.DirOutbound, true).Return(nil, rerr)
+		rcmgr.EXPECT().OpenConnection(network.DirOutbound, true, ln.Multiaddr()).Return(nil, rerr)
 		_, err = tb.Dial(context.Background(), ln.Multiaddr(), peerA)
 		require.ErrorIs(t, err, rerr)
 	})
 
 	t.Run("peer denied", func(t *testing.T) {
 		scope := mocknetwork.NewMockConnManagementScope(ctrl)
-		rcmgr.EXPECT().OpenConnection(network.DirOutbound, true).Return(scope, nil)
+		rcmgr.EXPECT().OpenConnection(network.DirOutbound, true, ln.Multiaddr()).Return(scope, nil)
 		rerr := errors.New("nope")
 		scope.EXPECT().SetPeer(peerA).Return(rerr)
 		scope.EXPECT().Done()