webrtc: increase receive buffer size on listener (#2730)

p2p/transport/webrtc/listener.go

@@ -204,6 +204,10 @@ func (l *listener) setupConnection(
 		l.transport.peerConnectionTimeouts.Failed,
 		l.transport.peerConnectionTimeouts.Keepalive,
 	)
+	// This is higher than the path MTU due to a bug in the sctp chunking logic.
+	// Remove this after https://github.com/pion/sctp/pull/301 is included
+	// in a release.
+	settingEngine.SetReceiveMTU(udpmux.ReceiveBufSize)
 	settingEngine.DetachDataChannels()
 
 	w, err = newWebRTCConnection(settingEngine, l.config)
@@ -314,14 +318,13 @@ func (l *listener) Multiaddr() ma.Multiaddr {
 }
 
 // addOnConnectionStateChangeCallback adds the OnConnectionStateChange to the PeerConnection.
-// The channel returned here:
-// * is closed when the state changes to Connection
-// * receives an error when the state changes to Failed
-// * doesn't receive anything (nor is closed) when the state changes to Disconnected
+// If the connection establishment errors, an error is written to the channel before closing.
+// If the connection establishment is successful, the channel is closed without writing anything.
 func addOnConnectionStateChangeCallback(pc *webrtc.PeerConnection) <-chan error {
 	errC := make(chan error, 1)
 	var once sync.Once
 	pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
+		fmt.Println("connection state: ", state)
 		switch state {
 		case webrtc.PeerConnectionStateConnected:
 			once.Do(func() { close(errC) })
@@ -336,6 +339,13 @@ func addOnConnectionStateChangeCallback(pc *webrtc.PeerConnection) <-chan error
 			// If the connection then receives packets on the connection, it can move back to the connected state.
 			// If no packets are received until the failed timeout is triggered, the connection moves to the failed state.
 			log.Warn("peerconnection disconnected")
+		case webrtc.PeerConnectionStateClosed:
+			// ConnectionStateClosed is a terminal state. This happens when the peer closes the PeerConnection before
+			// connection establishment.
+			once.Do(func() {
+				errC <- errors.New("peerconnection closed")
+				close(errC)
+			})
 		}
 	})
 	return errC

p2p/transport/webrtc/transport_test.go

@@ -110,14 +110,16 @@ func TestTransportWebRTC_CanListenSingle(t *testing.T) {
 
 	done := make(chan struct{})
 	go func() {
-		_, err := tr1.Dial(context.Background(), listener.Multiaddr(), listeningPeer)
+		conn, err := tr1.Dial(context.Background(), listener.Multiaddr(), listeningPeer)
 		assert.NoError(t, err)
+		t.Cleanup(func() { conn.Close() })
 		close(done)
 	}()
 
 	conn, err := listener.Accept()
 	require.NoError(t, err)
 	require.NotNil(t, conn)
+	defer conn.Close()
 
 	require.Equal(t, connectingPeer, conn.RemotePeer())
 	select {
@@ -170,12 +172,14 @@ func TestTransportWebRTC_CanListenMultiple(t *testing.T) {
 			defer wg.Done()
 			ctr, _ := getTransport(t)
 			conn, err := ctr.Dial(ctx, listener.Multiaddr(), listeningPeer)
+			if conn != nil {
+				t.Cleanup(func() { conn.Close() })
+			}
 			select {
 			case <-ctx.Done():
 			default:
 				assert.NoError(t, err)
 				assert.NotNil(t, conn)
-				t.Cleanup(func() { conn.Close() })
 			}
 		}()
 	}