webrtc: add a test for establishing many connections

go.mod

@@ -126,3 +126,5 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	lukechampine.com/blake3 v1.2.1 // indirect
 )
+
+// replace github.com/pion/webrtc/v3 => ../pion/webrtc

p2p/transport/webrtc/listener.go

@@ -330,20 +330,22 @@ func addOnConnectionStateChangeCallback(pc *webrtc.PeerConnection) <-chan error
 	errC := make(chan error, 1)
 	var once sync.Once
 	pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
-		switch state {
+		fmt.Println("state received: ", state, pc.ConnectionState())
+		switch pc.ConnectionState() {
 		case webrtc.PeerConnectionStateConnected:
 			once.Do(func() { close(errC) })
 		case webrtc.PeerConnectionStateFailed:
 			once.Do(func() {
 				errC <- errors.New("peerconnection failed")
 				close(errC)
 			})
+			log.Error("peer connection failed")
 		case webrtc.PeerConnectionStateDisconnected:
 			// the connection can move to a disconnected state and back to a connected state without ICE renegotiation.
 			// This could happen when underlying UDP packets are lost, and therefore the connection moves to the disconnected state.
 			// 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")
+			log.Error("peerconnection disconnected")
 		}
 	})
 	return errC

p2p/transport/webrtc/transport_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/rand"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -17,6 +18,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
+	tpt "github.com/libp2p/go-libp2p/core/transport"
 	ma "github.com/multiformats/go-multiaddr"
 	manet "github.com/multiformats/go-multiaddr/net"
 	"github.com/multiformats/go-multibase"
@@ -860,3 +862,93 @@ func TestMaxInFlightRequests(t *testing.T) {
 	require.Equal(t, count, int(success.Load()), "expected exactly 3 dial successes")
 	require.Equal(t, 1, int(fails.Load()), "expected exactly 1 dial failure")
 }
+
+func TestManyConnections(t *testing.T) {
+	const N = 20
+	errCh := make(chan error, 20)
+	successCh := make(chan struct{}, 1)
+
+	tr, lp := getTransport(t)
+	ln, err := tr.Listen(ma.StringCast("/ip4/127.0.0.1/udp/0/webrtc-direct"))
+	require.NoError(t, err)
+	defer ln.Close()
+
+	runListenConn := func(conn tpt.CapableConn) {
+		defer conn.Close()
+
+		s, err := conn.AcceptStream()
+		if err != nil {
+			t.Errorf("accept stream failed for listener: %s", err)
+			errCh <- err
+			return
+		}
+		var b [4]byte
+		if _, err := s.Read(b[:]); err != nil {
+			t.Errorf("read stream failed for listener: %s", err)
+			errCh <- err
+			return
+		}
+		s.Write(b[:])
+		_, err = s.Read(b[:]) // peer will close the connection after read
+		if !assert.Error(t, err) {
+			errCh <- errors.New("expected peer to close connection")
+			return
+		}
+	}
+
+	runDialConn := func(conn tpt.CapableConn) error {
+		defer conn.Close()
+
+		s, err := conn.OpenStream(context.Background())
+		if err != nil {
+			t.Errorf("accept stream failed for listener: %s", err)
+			return err
+		}
+		var b [4]byte
+		if _, err := s.Write(b[:]); err != nil {
+			t.Errorf("write stream failed for dialer: %s", err)
+			return err
+		}
+		if _, err := s.Read(b[:]); err != nil {
+			t.Errorf("read stream failed for dialer: %s", err)
+			return err
+		}
+		return nil
+	}
+
+	go func() {
+		for i := 0; i < N; i++ {
+			conn, err := ln.Accept()
+			if err != nil {
+				t.Errorf("listener failed to accept conneciton: %s", err)
+				return
+			}
+			runListenConn(conn)
+			successCh <- struct{}{}
+		}
+	}()
+
+	tp, _ := getTransport(t)
+	for i := 0; i < N; i++ {
+		// This test aims to check for deadlocks. So keep a high timeout
+		ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+		conn, err := tp.Dial(ctx, ln.Multiaddr(), lp)
+		if err != nil {
+			t.Errorf("dial failed: %s", err)
+			cancel()
+			return
+		}
+		err = runDialConn(conn)
+		require.NoError(t, err)
+		cancel()
+		select {
+		case <-time.After(120 * time.Second):
+			t.Fatalf("timed out")
+		case <-errCh:
+			t.Fatal("listener error:", err)
+		case <-successCh:
+		}
+		t.Log("completed conn:", i)
+	}
+
+}