Expose detailed connection info with ICEConnectionDetails (#2287)

* Expose detailed connection info with ICEConnectionDetails

* clone to avoid data race

* lower transport

* simplify

* address feedback

pkg/rtc/participant.go

@@ -358,10 +358,6 @@ func (p *ParticipantImpl) GetClientConfiguration() *livekit.ClientConfiguration
 	return p.params.ClientConf
 }
 
-func (p *ParticipantImpl) GetICEConnectionType() types.ICEConnectionType {
-	return p.TransportManager.GetICEConnectionType()
-}
-
 func (p *ParticipantImpl) GetBufferFactory() *buffer.Factory {
 	return p.params.Config.BufferFactory
 }
@@ -582,7 +578,7 @@ func (p *ParticipantImpl) OnClaimsChanged(callback func(types.LocalParticipant))
 func (p *ParticipantImpl) HandleSignalSourceClose() {
 	p.TransportManager.SetSignalSourceValid(false)
 
-	if !p.TransportManager.HasPublisherEverConnected() && !p.TransportManager.HasSubscriberEverConnected() {
+	if !p.HasConnected() {
 		reason := types.ParticipantCloseReasonJoinFailed
 		_ = p.Close(false, reason, false)
 	}
@@ -1708,6 +1704,10 @@ func (p *ParticipantImpl) GetPendingTrack(trackID livekit.TrackID) *livekit.Trac
 	return nil
 }
 
+func (p *ParticipantImpl) HasConnected() bool {
+	return p.TransportManager.HasSubscriberEverConnected() || p.TransportManager.HasPublisherEverConnected()
+}
+
 func (p *ParticipantImpl) sendTrackPublished(cid string, ti *livekit.TrackInfo) {
 	p.pubLogger.Debugw("sending track published", "cid", cid, "trackInfo", logger.Proto(ti))
 	_ = p.writeMessage(&livekit.SignalResponse{

pkg/rtc/room.go

@@ -17,9 +17,11 @@ package rtc
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"math"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -325,11 +327,6 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me
 	// it's important to set this before connection, we don't want to miss out on any published tracks
 	participant.OnTrackPublished(r.onTrackPublished)
 	participant.OnStateChange(func(p types.LocalParticipant, oldState livekit.ParticipantInfo_State) {
-		r.Logger.Infow("participant state changed",
-			"state", p.State(),
-			"participant", p.Identity(),
-			"pID", p.ID(),
-			"oldState", oldState)
 		if r.onParticipantChanged != nil {
 			r.onParticipantChanged(participant)
 		}
@@ -343,15 +340,24 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me
 			// start the workers once connectivity is established
 			p.Start()
 
+			meta := &livekit.AnalyticsClientMeta{
+				ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds()),
+			}
+			cds := participant.GetICEConnectionDetails()
+			for _, cd := range cds {
+				if cd.Type != types.ICEConnectionTypeUnknown {
+					meta.ConnectionType = string(cd.Type)
+					break
+				}
+			}
 			r.telemetry.ParticipantActive(context.Background(),
 				r.ToProto(),
 				p.ToProto(),
-				&livekit.AnalyticsClientMeta{
-					ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds()),
-					ConnectionType:    string(p.GetICEConnectionType()),
-				},
+				meta,
 				false,
 			)
+
+			p.GetLogger().Infow("participant active", connectionDetailsFields(cds)...)
 		} else if state == livekit.ParticipantInfo_DISCONNECTED {
 			// remove participant from room
 			go r.RemoveParticipant(p.Identity(), p.ID(), types.ParticipantCloseReasonStateDisconnected)
@@ -495,24 +501,27 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, requestSource routing
 func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livekit.ParticipantID, reason types.ParticipantCloseReason) {
 	r.lock.Lock()
 	p, ok := r.participants[identity]
-	if ok {
-		if pID != "" && p.ID() != pID {
-			// participant session has been replaced
-			r.lock.Unlock()
-			return
-		}
+	if !ok {
+		r.lock.Unlock()
+		return
+	}
 
-		delete(r.participants, identity)
-		delete(r.participantOpts, identity)
-		delete(r.participantRequestSources, identity)
-		delete(r.hasPublished, identity)
-		if !p.Hidden() {
-			r.protoRoom.NumParticipants--
-		}
+	if pID != "" && p.ID() != pID {
+		// participant session has been replaced
+		r.lock.Unlock()
+		return
+	}
+
+	delete(r.participants, identity)
+	delete(r.participantOpts, identity)
+	delete(r.participantRequestSources, identity)
+	delete(r.hasPublished, identity)
+	if !p.Hidden() {
+		r.protoRoom.NumParticipants--
 	}
 
 	immediateChange := false
-	if (p != nil && p.IsRecorder()) || r.protoRoom.ActiveRecording {
+	if p.IsRecorder() {
 		activeRecording := false
 		for _, op := range r.participants {
 			if op.IsRecorder() {
@@ -524,14 +533,16 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek
 		if r.protoRoom.ActiveRecording != activeRecording {
 			r.protoRoom.ActiveRecording = activeRecording
 			immediateChange = true
-
 		}
 	}
 	r.lock.Unlock()
 	r.protoProxy.MarkDirty(immediateChange)
 
-	if !ok {
-		return
+	if !p.HasConnected() {
+		fields := append(connectionDetailsFields(p.GetICEConnectionDetails()),
+			"reason", reason.String(),
+		)
+		p.GetLogger().Infow("removing participant without connection", fields...)
 	}
 
 	// send broadcast only if it's not already closed
@@ -551,7 +562,6 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek
 	p.OnSubscribeStatusChanged(nil)
 
 	// close participant as well
-	r.Logger.Debugw("closing participant for removal", "pID", p.ID(), "participant", p.Identity())
 	_ = p.Close(true, reason, false)
 
 	r.leftAt.Store(time.Now().Unix())
@@ -1416,3 +1426,39 @@ func BroadcastDataPacketForRoom(r types.Room, source types.LocalParticipant, dp
 		}
 	})
 }
+
+func connectionDetailsFields(cds []*types.ICEConnectionDetails) []interface{} {
+	var fields []interface{}
+	connectionType := types.ICEConnectionTypeUnknown
+	for _, cd := range cds {
+		candidates := make([]string, 0, len(cd.Remote)+len(cd.Local))
+		for _, c := range cd.Local {
+			cStr := "[local]"
+			if c.Selected {
+				cStr += "[selected]"
+			} else if c.Filtered {
+				cStr += "[filtered]"
+			}
+			cStr += " " + c.Local.String()
+			candidates = append(candidates, cStr)
+		}
+		for _, c := range cd.Remote {
+			cStr := "[remote]"
+			if c.Selected {
+				cStr += "[selected]"
+			} else if c.Filtered {
+				cStr += "[filtered]"
+			}
+			cStr += " " + c.Remote.String()
+			candidates = append(candidates, cStr)
+		}
+		if len(candidates) > 0 {
+			fields = append(fields, fmt.Sprintf("%sCandidates", strings.ToLower(cd.Transport.String())), candidates)
+		}
+		if cd.Type != types.ICEConnectionTypeUnknown {
+			connectionType = cd.Type
+		}
+	}
+	fields = append(fields, "connectionType", connectionType)
+	return fields
+}

pkg/rtc/transport.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/bep/debounce"
 	"github.com/pion/dtls/v2/pkg/crypto/elliptic"
-	"github.com/pion/ice/v2"
 	"github.com/pion/interceptor"
 	"github.com/pion/interceptor/pkg/cc"
 	"github.com/pion/interceptor/pkg/gcc"
@@ -35,13 +34,6 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/atomic"
 
-	sutils "github.com/livekit/livekit-server/pkg/utils"
-	"github.com/livekit/protocol/livekit"
-	"github.com/livekit/protocol/logger"
-	"github.com/livekit/protocol/logger/pionlogger"
-	lksdp "github.com/livekit/protocol/sdp"
-	"github.com/livekit/protocol/utils"
-
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu/pacer"
@@ -49,6 +41,11 @@ import (
 	"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
+	sutils "github.com/livekit/livekit-server/pkg/utils"
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/logger/pionlogger"
+	lksdp "github.com/livekit/protocol/sdp"
 )
 
 const (
@@ -94,7 +91,6 @@ const (
 	signalICEGatheringComplete signal = iota
 	signalLocalICECandidate
 	signalRemoteICECandidate
-	signalLogICECandidates
 	signalSendOffer
 	signalRemoteDescriptionReceived
 	signalICERestart
@@ -108,8 +104,6 @@ func (s signal) String() string {
 		return "LOCAL_ICE_CANDIDATE"
 	case signalRemoteICECandidate:
 		return "REMOTE_ICE_CANDIDATE"
-	case signalLogICECandidates:
-		return "LOG_ICE_CANDIDATES"
 	case signalSendOffer:
 		return "SEND_OFFER"
 	case signalRemoteDescriptionReceived:
@@ -234,11 +228,7 @@ type PCTransport struct {
 	currentOfferIceCredential string // ice user:pwd, for publish side ice restart checking
 	pendingRestartIceOffer    *webrtc.SessionDescription
 
-	// for cleaner logging
-	allowedLocalCandidates   *utils.DedupedSlice[string]
-	allowedRemoteCandidates  *utils.DedupedSlice[string]
-	filteredLocalCandidates  *utils.DedupedSlice[string]
-	filteredRemoteCandidates *utils.DedupedSlice[string]
+	connectionDetails *types.ICEConnectionDetails
 }
 
 type TransportParams struct {
@@ -251,6 +241,7 @@ type TransportParams struct {
 	Telemetry                    telemetry.TelemetryService
 	EnabledCodecs                []*livekit.Codec
 	Logger                       logger.Logger
+	Transport                    livekit.SignalTarget
 	SimTracks                    map[uint32]SimulcastTrackInfo
 	ClientInfo                   ClientInfo
 	IsOfferer                    bool
@@ -393,10 +384,7 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
 		eventCh:                  make(chan event, 100),
 		previousTrackDescription: make(map[string]*trackDescription),
 		canReuseTransceiver:      true,
-		allowedLocalCandidates:   utils.NewDedupedSlice[string](maxICECandidates),
-		allowedRemoteCandidates:  utils.NewDedupedSlice[string](maxICECandidates),
-		filteredLocalCandidates:  utils.NewDedupedSlice[string](maxICECandidates),
-		filteredRemoteCandidates: utils.NewDedupedSlice[string](maxICECandidates),
+		connectionDetails:        types.NewICEConnectionDetails(params.Transport, params.Logger),
 	}
 	if params.IsSendSide {
 		t.streamAllocator = streamallocator.NewStreamAllocator(streamallocator.StreamAllocatorParams{
@@ -564,12 +552,6 @@ func (t *PCTransport) getSelectedPair() (*webrtc.ICECandidatePair, error) {
 	return iceTransport.GetSelectedCandidatePair()
 }
 
-func (t *PCTransport) logICECandidates() {
-	t.postEvent(event{
-		signal: signalLogICECandidates,
-	})
-}
-
 func (t *PCTransport) setConnectedAt(at time.Time) bool {
 	t.lock.Lock()
 	t.connectedAt = at
@@ -629,6 +611,14 @@ func (t *PCTransport) onICEConnectionStateChange(state webrtc.ICEConnectionState
 	switch state {
 	case webrtc.ICEConnectionStateConnected:
 		t.setICEConnectedAt(time.Now())
+		go func() {
+			pair, err := t.getSelectedPair()
+			if err != nil {
+				t.params.Logger.Warnw("failed to get selected candidate pair", err)
+				return
+			}
+			t.connectionDetails.SetSelectedPair(pair)
+		}()
 
 	case webrtc.ICEConnectionStateChecking:
 		t.setICEStartedAt(time.Now())
@@ -905,6 +895,10 @@ func (t *PCTransport) HasEverConnected() bool {
 	return !t.firstConnectedAt.IsZero()
 }
 
+func (t *PCTransport) GetICEConnectionDetails() *types.ICEConnectionDetails {
+	return t.connectionDetails
+}
+
 func (t *PCTransport) WriteRTCP(pkts []rtcp.Packet) error {
 	return t.pc.WriteRTCP(pkts)
 }
@@ -1198,44 +1192,6 @@ func (t *PCTransport) SetChannelCapacityOfStreamAllocator(channelCapacity int64)
 	t.streamAllocator.SetChannelCapacity(channelCapacity)
 }
 
-func (t *PCTransport) GetICEConnectionType() types.ICEConnectionType {
-	unknown := types.ICEConnectionTypeUnknown
-	if t.pc == nil {
-		return unknown
-	}
-	p, err := t.getSelectedPair()
-	if err != nil || p == nil {
-		return unknown
-	}
-
-	if p.Remote.Typ == webrtc.ICECandidateTypeRelay {
-		return types.ICEConnectionTypeTURN
-	} else if p.Remote.Typ == webrtc.ICECandidateTypePrflx {
-		// if the remote relay candidate pings us *before* we get a relay candidate,
-		// Pion would have created a prflx candidate with the same address as the relay candidate.
-		// to report an accurate connection type, we'll compare to see if existing relay candidates match
-		t.lock.RLock()
-		allowedRemoteCandidates := t.allowedRemoteCandidates.Get()
-		t.lock.RUnlock()
-
-		for _, ci := range allowedRemoteCandidates {
-			candidateValue := strings.TrimPrefix(ci, "candidate:")
-			candidate, err := ice.UnmarshalCandidate(candidateValue)
-			if err == nil && candidate.Type() == ice.CandidateTypeRelay {
-				if p.Remote.Address == candidate.Address() &&
-					p.Remote.Port == uint16(candidate.Port()) &&
-					p.Remote.Protocol.String() == candidate.NetworkType().NetworkShort() {
-					return types.ICEConnectionTypeTURN
-				}
-			}
-		}
-	}
-	if p.Remote.Protocol == webrtc.ICEProtocolTCP {
-		return types.ICEConnectionTypeTCP
-	}
-	return types.ICEConnectionTypeUDP
-}
-
 func (t *PCTransport) preparePC(previousAnswer webrtc.SessionDescription) error {
 	// sticky data channel to first m-lines, if someday we don't send sdp without media streams to
 	// client's subscribe pc after joining, should change this step
@@ -1453,7 +1409,6 @@ func (t *PCTransport) processEvents() {
 
 	t.clearSignalStateCheckTimer()
 	t.params.Logger.Debugw("leaving events processor")
-	t.handleLogICECandidates(nil)
 }
 
 func (t *PCTransport) handleEvent(e *event) error {
@@ -1464,8 +1419,6 @@ func (t *PCTransport) handleEvent(e *event) error {
 		return t.handleLocalICECandidate(e)
 	case signalRemoteICECandidate:
 		return t.handleRemoteICECandidate(e)
-	case signalLogICECandidates:
-		return t.handleLogICECandidates(e)
 	case signalSendOffer:
 		return t.handleSendOffer(e)
 	case signalRemoteDescriptionReceived:
@@ -1537,33 +1490,28 @@ func (t *PCTransport) localDescriptionSent() error {
 func (t *PCTransport) clearLocalDescriptionSent() {
 	t.cacheLocalCandidates = true
 	t.cachedLocalCandidates = nil
-
-	t.allowedLocalCandidates.Clear()
-	t.lock.Lock()
-	t.allowedRemoteCandidates.Clear()
-	t.lock.Unlock()
-	t.filteredLocalCandidates.Clear()
-	t.filteredRemoteCandidates.Clear()
+	t.connectionDetails.Clear()
 }
 
 func (t *PCTransport) handleLocalICECandidate(e *event) error {
 	c := e.data.(*webrtc.ICECandidate)
 
 	filtered := false
-	if t.preferTCP.Load() && c != nil && c.Protocol != webrtc.ICEProtocolTCP {
-		cstr := c.String()
-		t.params.Logger.Debugw("filtering out local candidate", "candidate", cstr)
-		t.filteredLocalCandidates.Add(cstr)
-		filtered = true
+	if c != nil {
+		if t.preferTCP.Load() && c.Protocol != webrtc.ICEProtocolTCP {
+			t.params.Logger.Debugw("filtering out local candidate",
+				"candidate", func() interface{} {
+					return c.String()
+				})
+			filtered = true
+		}
+		t.connectionDetails.AddLocalCandidate(c, filtered)
 	}
 
 	if filtered {
 		return nil
 	}
 
-	if c != nil {
-		t.allowedLocalCandidates.Add(c.String())
-	}
 	if t.cacheLocalCandidates {
 		t.cachedLocalCandidates = append(t.cachedLocalCandidates, c)
 		return nil
@@ -1582,18 +1530,14 @@ func (t *PCTransport) handleRemoteICECandidate(e *event) error {
 	filtered := false
 	if t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp") {
 		t.params.Logger.Debugw("filtering out remote candidate", "candidate", c.Candidate)
-		t.filteredRemoteCandidates.Add(c.Candidate)
 		filtered = true
 	}
 
+	t.connectionDetails.AddRemoteCandidate(*c, filtered)
 	if filtered {
 		return nil
 	}
 
-	t.lock.Lock()
-	t.allowedRemoteCandidates.Add(c.Candidate)
-	t.lock.Unlock()
-
 	if t.pc.RemoteDescription() == nil {
 		t.pendingRemoteCandidates = append(t.pendingRemoteCandidates, c)
 		return nil
@@ -1606,30 +1550,6 @@ func (t *PCTransport) handleRemoteICECandidate(e *event) error {
 	return nil
 }
 
-func (t *PCTransport) handleLogICECandidates(_ *event) error {
-	lc := t.allowedLocalCandidates.Get()
-	rc := t.allowedRemoteCandidates.Get()
-	var fields []interface{}
-	if len(lc) != 0 || len(rc) != 0 {
-		fields = append(fields,
-			"lc", lc,
-			"rc", rc,
-			"lc_filtered", t.filteredLocalCandidates.Get(),
-			"rc_filtered", t.filteredRemoteCandidates.Get(),
-		)
-
-	}
-	if pair, err := t.getSelectedPair(); err == nil {
-		fields = append(fields, "selected_pair", pair)
-	}
-
-	if len(fields) > 0 {
-		t.params.Logger.Infow("ice candidates", fields...)
-	}
-
-	return nil
-}
-
 func (t *PCTransport) setNegotiationState(state NegotiationState) {
 	t.negotiationState = state
 	if onNegotiationStateChanged := t.getOnNegotiationStateChanged(); onNegotiationStateChanged != nil {

pkg/rtc/transportmanager.go

@@ -124,6 +124,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
 		Logger:                  LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER),
 		SimTracks:               params.SimTracks,
 		ClientInfo:              params.ClientInfo,
+		Transport:               livekit.SignalTarget_PUBLISHER,
 	})
 	if err != nil {
 		return nil, err
@@ -159,6 +160,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
 		IsSendSide:                   true,
 		AllowPlayoutDelay:            params.AllowPlayoutDelay,
 		DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount,
+		Transport:                    livekit.SignalTarget_SUBSCRIBER,
 	})
 	if err != nil {
 		return nil, err
@@ -554,8 +556,15 @@ func (t *TransportManager) SubscriberAsPrimary() bool {
 	return t.params.SubscriberAsPrimary
 }
 
-func (t *TransportManager) GetICEConnectionType() types.ICEConnectionType {
-	return t.getTransport(true).GetICEConnectionType()
+func (t *TransportManager) GetICEConnectionDetails() []*types.ICEConnectionDetails {
+	details := make([]*types.ICEConnectionDetails, 0, 2)
+	for _, pc := range []*PCTransport{t.publisher, t.subscriber} {
+		cd := pc.GetICEConnectionDetails()
+		if cd.HasCandidates() {
+			details = append(details, cd.Clone())
+		}
+	}
+	return details
 }
 
 func (t *TransportManager) getTransport(isPrimary bool) *PCTransport {

pkg/rtc/types/ice.go

@@ -0,0 +1,254 @@
+/*
+ * Copyright 2023 LiveKit, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package types
+
+import (
+	"strings"
+	"sync"
+
+	"github.com/pion/ice/v2"
+	"github.com/pion/webrtc/v3"
+	"golang.org/x/exp/slices"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
+)
+
+type ICEConnectionType string
+
+const (
+	ICEConnectionTypeUDP     ICEConnectionType = "udp"
+	ICEConnectionTypeTCP     ICEConnectionType = "tcp"
+	ICEConnectionTypeTURN    ICEConnectionType = "turn"
+	ICEConnectionTypeUnknown ICEConnectionType = "unknown"
+)
+
+type ICECandidateExtended struct {
+	// only one of local or remote is set. This is due to type foo in Pion
+	Local    *webrtc.ICECandidate
+	Remote   ice.Candidate
+	Selected bool
+	Filtered bool
+}
+
+type ICEConnectionDetails struct {
+	Local     []*ICECandidateExtended
+	Remote    []*ICECandidateExtended
+	Transport livekit.SignalTarget
+	Type      ICEConnectionType
+	lock      sync.Mutex
+	logger    logger.Logger
+}
+
+func NewICEConnectionDetails(transport livekit.SignalTarget, l logger.Logger) *ICEConnectionDetails {
+	d := &ICEConnectionDetails{
+		Transport: transport,
+		Type:      ICEConnectionTypeUnknown,
+		logger:    l,
+	}
+	return d
+}
+
+func (d *ICEConnectionDetails) HasCandidates() bool {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	return len(d.Local) > 0 || len(d.Remote) > 0
+}
+
+// Clone returns a copy of the ICEConnectionDetails, where fields can be read without locking
+func (d *ICEConnectionDetails) Clone() *ICEConnectionDetails {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	clone := &ICEConnectionDetails{
+		Transport: d.Transport,
+		Type:      d.Type,
+		logger:    d.logger,
+		Local:     make([]*ICECandidateExtended, 0, len(d.Local)),
+		Remote:    make([]*ICECandidateExtended, 0, len(d.Remote)),
+	}
+	for _, c := range d.Local {
+		clone.Local = append(clone.Local, &ICECandidateExtended{
+			Local:    c.Local,
+			Filtered: c.Filtered,
+		})
+	}
+	for _, c := range d.Remote {
+		clone.Remote = append(clone.Remote, &ICECandidateExtended{
+			Remote:   c.Remote,
+			Filtered: c.Filtered,
+		})
+	}
+	return clone
+}
+
+func (d *ICEConnectionDetails) AddLocalCandidate(c *webrtc.ICECandidate, filtered bool) {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	compFn := func(e *ICECandidateExtended) bool {
+		return isCandidateEqualTo(e.Local, c)
+	}
+	if slices.ContainsFunc[[]*ICECandidateExtended, *ICECandidateExtended](d.Local, compFn) {
+		return
+	}
+	d.Local = append(d.Local, &ICECandidateExtended{
+		Local:    c,
+		Filtered: filtered,
+	})
+}
+
+func (d *ICEConnectionDetails) AddRemoteCandidate(c webrtc.ICECandidateInit, filtered bool) {
+	candidate, err := unmarshalICECandidate(c)
+	if err != nil {
+		d.logger.Errorw("could not unmarshal candidate", err, "candidate", c)
+		return
+	}
+
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	compFn := func(e *ICECandidateExtended) bool {
+		return isICECandidateEqualTo(e.Remote, candidate)
+	}
+	if slices.ContainsFunc[[]*ICECandidateExtended, *ICECandidateExtended](d.Remote, compFn) {
+		return
+	}
+	d.Remote = append(d.Remote, &ICECandidateExtended{
+		Remote:   candidate,
+		Filtered: filtered,
+	})
+}
+
+func (d *ICEConnectionDetails) Clear() {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	d.Local = nil
+	d.Remote = nil
+	d.Type = ICEConnectionTypeUnknown
+}
+
+func (d *ICEConnectionDetails) SetSelectedPair(pair *webrtc.ICECandidatePair) {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	remoteIdx := slices.IndexFunc[[]*ICECandidateExtended, *ICECandidateExtended](d.Remote, func(e *ICECandidateExtended) bool {
+		return isICECandidateEqualToCandidate(e.Remote, pair.Remote)
+	})
+	if remoteIdx < 0 {
+		// it's possible for prflx candidates to be generated by Pion, we'll add them
+		candidate, err := unmarshalICECandidate(pair.Remote.ToJSON())
+		if err != nil {
+			d.logger.Errorw("could not unmarshal remote candidate", err, "candidate", pair.Remote)
+			return
+		}
+		d.Remote = append(d.Remote, &ICECandidateExtended{
+			Remote:   candidate,
+			Filtered: false,
+		})
+		remoteIdx = len(d.Remote) - 1
+	}
+	remote := d.Remote[remoteIdx]
+	remote.Selected = true
+
+	localIdx := slices.IndexFunc[[]*ICECandidateExtended, *ICECandidateExtended](d.Local, func(e *ICECandidateExtended) bool {
+		return isCandidateEqualTo(e.Local, pair.Local)
+	})
+	if localIdx < 0 {
+		d.logger.Errorw("could not match local candidate", nil, "local", pair.Local)
+		// should not happen
+		return
+	}
+	local := d.Local[localIdx]
+	local.Selected = true
+
+	d.Type = ICEConnectionTypeUDP
+	if pair.Remote.Protocol == webrtc.ICEProtocolTCP {
+		d.Type = ICEConnectionTypeTCP
+	}
+	if pair.Remote.Typ == webrtc.ICECandidateTypeRelay {
+		d.Type = ICEConnectionTypeTURN
+	} else if pair.Remote.Typ == webrtc.ICECandidateTypePrflx {
+		// if the remote relay candidate pings us *before* we get a relay candidate,
+		// Pion would have created a prflx candidate with the same address as the relay candidate.
+		// to report an accurate connection type, we'll compare to see if existing relay candidates match
+		for _, other := range d.Remote {
+			or := other.Remote
+			if or.Type() == ice.CandidateTypeRelay &&
+				pair.Remote.Address == or.Address() &&
+				pair.Remote.Port == uint16(or.Port()) &&
+				pair.Remote.Protocol.String() == or.NetworkType().NetworkShort() {
+				d.Type = ICEConnectionTypeTURN
+			}
+		}
+	}
+}
+
+func isCandidateEqualTo(c1, c2 *webrtc.ICECandidate) bool {
+	if c1 == nil && c2 == nil {
+		return true
+	}
+	if (c1 == nil && c2 != nil) || (c1 != nil && c2 == nil) {
+		return false
+	}
+	return c1.Typ == c2.Typ &&
+		c1.Protocol == c2.Protocol &&
+		c1.Address == c2.Address &&
+		c1.Port == c2.Port &&
+		c1.Component == c2.Component &&
+		c1.Foundation == c2.Foundation &&
+		c1.Priority == c2.Priority &&
+		c1.RelatedAddress == c2.RelatedAddress &&
+		c1.RelatedPort == c2.RelatedPort &&
+		c1.TCPType == c2.TCPType
+}
+
+func isICECandidateEqualTo(c1, c2 ice.Candidate) bool {
+	if c1 == nil && c2 == nil {
+		return true
+	}
+	if (c1 == nil && c2 != nil) || (c1 != nil && c2 == nil) {
+		return false
+	}
+	return c1.Type() == c2.Type() &&
+		c1.NetworkType() == c2.NetworkType() &&
+		c1.Address() == c2.Address() &&
+		c1.Port() == c2.Port() &&
+		c1.Component() == c2.Component() &&
+		c1.Foundation() == c2.Foundation() &&
+		c1.Priority() == c2.Priority() &&
+		c1.RelatedAddress().Equal(c2.RelatedAddress()) &&
+		c1.TCPType() == c2.TCPType()
+}
+
+func isICECandidateEqualToCandidate(c1 ice.Candidate, c2 *webrtc.ICECandidate) bool {
+	if c1 == nil && c2 == nil {
+		return true
+	}
+	if (c1 == nil && c2 != nil) || (c1 != nil && c2 == nil) {
+		return false
+	}
+	return c1.Type().String() == c2.Typ.String() &&
+		c1.NetworkType().NetworkShort() == c2.Protocol.String() &&
+		c1.Address() == c2.Address &&
+		c1.Port() == int(c2.Port) &&
+		c1.Component() == c2.Component &&
+		c1.Foundation() == c2.Foundation &&
+		c1.Priority() == c2.Priority &&
+		c1.TCPType().String() == c2.TCPType
+}
+
+func unmarshalICECandidate(c webrtc.ICECandidateInit) (ice.Candidate, error) {
+	candidateValue := strings.TrimPrefix(c.Candidate, "candidate:")
+	return ice.UnmarshalCandidate(candidateValue)
+}

pkg/rtc/types/interfaces.go

@@ -285,15 +285,6 @@ type Participant interface {
 
 // -------------------------------------------------------
 
-type ICEConnectionType string
-
-const (
-	ICEConnectionTypeUDP     ICEConnectionType = "udp"
-	ICEConnectionTypeTCP     ICEConnectionType = "tcp"
-	ICEConnectionTypeTURN    ICEConnectionType = "turn"
-	ICEConnectionTypeUnknown ICEConnectionType = "unknown"
-)
-
 type AddTrackParams struct {
 	Stereo bool
 	Red    bool
@@ -320,10 +311,11 @@ type LocalParticipant interface {
 	SubscriberAsPrimary() bool
 	GetClientInfo() *livekit.ClientInfo
 	GetClientConfiguration() *livekit.ClientConfiguration
-	GetICEConnectionType() ICEConnectionType
 	GetBufferFactory() *buffer.Factory
 	GetPlayoutDelayConfig() *livekit.PlayoutDelay
 	GetPendingTrack(trackID livekit.TrackID) *livekit.TrackInfo
+	GetICEConnectionDetails() []*ICEConnectionDetails
+	HasConnected() bool
 
 	SetResponseSink(sink routing.MessageSink)
 	CloseSignalConnection(reason SignallingCloseReason)

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -233,15 +233,15 @@ type FakeLocalParticipant struct {
 	getConnectionQualityReturnsOnCall map[int]struct {
 		result1 *livekit.ConnectionQualityInfo
 	}
-	GetICEConnectionTypeStub        func() types.ICEConnectionType
-	getICEConnectionTypeMutex       sync.RWMutex
-	getICEConnectionTypeArgsForCall []struct {
+	GetICEConnectionDetailsStub        func() []*types.ICEConnectionDetails
+	getICEConnectionDetailsMutex       sync.RWMutex
+	getICEConnectionDetailsArgsForCall []struct {
 	}
-	getICEConnectionTypeReturns struct {
-		result1 types.ICEConnectionType
+	getICEConnectionDetailsReturns struct {
+		result1 []*types.ICEConnectionDetails
 	}
-	getICEConnectionTypeReturnsOnCall map[int]struct {
-		result1 types.ICEConnectionType
+	getICEConnectionDetailsReturnsOnCall map[int]struct {
+		result1 []*types.ICEConnectionDetails
 	}
 	GetLoggerStub        func() logger.Logger
 	getLoggerMutex       sync.RWMutex
@@ -371,6 +371,16 @@ type FakeLocalParticipant struct {
 	handleSignalSourceCloseMutex       sync.RWMutex
 	handleSignalSourceCloseArgsForCall []struct {
 	}
+	HasConnectedStub        func() bool
+	hasConnectedMutex       sync.RWMutex
+	hasConnectedArgsForCall []struct {
+	}
+	hasConnectedReturns struct {
+		result1 bool
+	}
+	hasConnectedReturnsOnCall map[int]struct {
+		result1 bool
+	}
 	HasPermissionStub        func(livekit.TrackID, livekit.ParticipantIdentity) bool
 	hasPermissionMutex       sync.RWMutex
 	hasPermissionArgsForCall []struct {
@@ -2062,15 +2072,15 @@ func (fake *FakeLocalParticipant) GetConnectionQualityReturnsOnCall(i int, resul
 	}{result1}
 }
 
-func (fake *FakeLocalParticipant) GetICEConnectionType() types.ICEConnectionType {
-	fake.getICEConnectionTypeMutex.Lock()
-	ret, specificReturn := fake.getICEConnectionTypeReturnsOnCall[len(fake.getICEConnectionTypeArgsForCall)]
-	fake.getICEConnectionTypeArgsForCall = append(fake.getICEConnectionTypeArgsForCall, struct {
+func (fake *FakeLocalParticipant) GetICEConnectionDetails() []*types.ICEConnectionDetails {
+	fake.getICEConnectionDetailsMutex.Lock()
+	ret, specificReturn := fake.getICEConnectionDetailsReturnsOnCall[len(fake.getICEConnectionDetailsArgsForCall)]
+	fake.getICEConnectionDetailsArgsForCall = append(fake.getICEConnectionDetailsArgsForCall, struct {
 	}{})
-	stub := fake.GetICEConnectionTypeStub
-	fakeReturns := fake.getICEConnectionTypeReturns
-	fake.recordInvocation("GetICEConnectionType", []interface{}{})
-	fake.getICEConnectionTypeMutex.Unlock()
+	stub := fake.GetICEConnectionDetailsStub
+	fakeReturns := fake.getICEConnectionDetailsReturns
+	fake.recordInvocation("GetICEConnectionDetails", []interface{}{})
+	fake.getICEConnectionDetailsMutex.Unlock()
 	if stub != nil {
 		return stub()
 	}
@@ -2080,38 +2090,38 @@ func (fake *FakeLocalParticipant) GetICEConnectionType() types.ICEConnectionType
 	return fakeReturns.result1
 }
 
-func (fake *FakeLocalParticipant) GetICEConnectionTypeCallCount() int {
-	fake.getICEConnectionTypeMutex.RLock()
-	defer fake.getICEConnectionTypeMutex.RUnlock()
-	return len(fake.getICEConnectionTypeArgsForCall)
+func (fake *FakeLocalParticipant) GetICEConnectionDetailsCallCount() int {
+	fake.getICEConnectionDetailsMutex.RLock()
+	defer fake.getICEConnectionDetailsMutex.RUnlock()
+	return len(fake.getICEConnectionDetailsArgsForCall)
 }
 
-func (fake *FakeLocalParticipant) GetICEConnectionTypeCalls(stub func() types.ICEConnectionType) {
-	fake.getICEConnectionTypeMutex.Lock()
-	defer fake.getICEConnectionTypeMutex.Unlock()
-	fake.GetICEConnectionTypeStub = stub
+func (fake *FakeLocalParticipant) GetICEConnectionDetailsCalls(stub func() []*types.ICEConnectionDetails) {
+	fake.getICEConnectionDetailsMutex.Lock()
+	defer fake.getICEConnectionDetailsMutex.Unlock()
+	fake.GetICEConnectionDetailsStub = stub
 }
 
-func (fake *FakeLocalParticipant) GetICEConnectionTypeReturns(result1 types.ICEConnectionType) {
-	fake.getICEConnectionTypeMutex.Lock()
-	defer fake.getICEConnectionTypeMutex.Unlock()
-	fake.GetICEConnectionTypeStub = nil
-	fake.getICEConnectionTypeReturns = struct {
-		result1 types.ICEConnectionType
+func (fake *FakeLocalParticipant) GetICEConnectionDetailsReturns(result1 []*types.ICEConnectionDetails) {
+	fake.getICEConnectionDetailsMutex.Lock()
+	defer fake.getICEConnectionDetailsMutex.Unlock()
+	fake.GetICEConnectionDetailsStub = nil
+	fake.getICEConnectionDetailsReturns = struct {
+		result1 []*types.ICEConnectionDetails
 	}{result1}
 }
 
-func (fake *FakeLocalParticipant) GetICEConnectionTypeReturnsOnCall(i int, result1 types.ICEConnectionType) {
-	fake.getICEConnectionTypeMutex.Lock()
-	defer fake.getICEConnectionTypeMutex.Unlock()
-	fake.GetICEConnectionTypeStub = nil
-	if fake.getICEConnectionTypeReturnsOnCall == nil {
-		fake.getICEConnectionTypeReturnsOnCall = make(map[int]struct {
-			result1 types.ICEConnectionType
+func (fake *FakeLocalParticipant) GetICEConnectionDetailsReturnsOnCall(i int, result1 []*types.ICEConnectionDetails) {
+	fake.getICEConnectionDetailsMutex.Lock()
+	defer fake.getICEConnectionDetailsMutex.Unlock()
+	fake.GetICEConnectionDetailsStub = nil
+	if fake.getICEConnectionDetailsReturnsOnCall == nil {
+		fake.getICEConnectionDetailsReturnsOnCall = make(map[int]struct {
+			result1 []*types.ICEConnectionDetails
 		})
 	}
-	fake.getICEConnectionTypeReturnsOnCall[i] = struct {
-		result1 types.ICEConnectionType
+	fake.getICEConnectionDetailsReturnsOnCall[i] = struct {
+		result1 []*types.ICEConnectionDetails
 	}{result1}
 }
 
@@ -2811,6 +2821,59 @@ func (fake *FakeLocalParticipant) HandleSignalSourceCloseCalls(stub func()) {
 	fake.HandleSignalSourceCloseStub = stub
 }
 
+func (fake *FakeLocalParticipant) HasConnected() bool {
+	fake.hasConnectedMutex.Lock()
+	ret, specificReturn := fake.hasConnectedReturnsOnCall[len(fake.hasConnectedArgsForCall)]
+	fake.hasConnectedArgsForCall = append(fake.hasConnectedArgsForCall, struct {
+	}{})
+	stub := fake.HasConnectedStub
+	fakeReturns := fake.hasConnectedReturns
+	fake.recordInvocation("HasConnected", []interface{}{})
+	fake.hasConnectedMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeLocalParticipant) HasConnectedCallCount() int {
+	fake.hasConnectedMutex.RLock()
+	defer fake.hasConnectedMutex.RUnlock()
+	return len(fake.hasConnectedArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) HasConnectedCalls(stub func() bool) {
+	fake.hasConnectedMutex.Lock()
+	defer fake.hasConnectedMutex.Unlock()
+	fake.HasConnectedStub = stub
+}
+
+func (fake *FakeLocalParticipant) HasConnectedReturns(result1 bool) {
+	fake.hasConnectedMutex.Lock()
+	defer fake.hasConnectedMutex.Unlock()
+	fake.HasConnectedStub = nil
+	fake.hasConnectedReturns = struct {
+		result1 bool
+	}{result1}
+}
+
+func (fake *FakeLocalParticipant) HasConnectedReturnsOnCall(i int, result1 bool) {
+	fake.hasConnectedMutex.Lock()
+	defer fake.hasConnectedMutex.Unlock()
+	fake.HasConnectedStub = nil
+	if fake.hasConnectedReturnsOnCall == nil {
+		fake.hasConnectedReturnsOnCall = make(map[int]struct {
+			result1 bool
+		})
+	}
+	fake.hasConnectedReturnsOnCall[i] = struct {
+		result1 bool
+	}{result1}
+}
+
 func (fake *FakeLocalParticipant) HasPermission(arg1 livekit.TrackID, arg2 livekit.ParticipantIdentity) bool {
 	fake.hasPermissionMutex.Lock()
 	ret, specificReturn := fake.hasPermissionReturnsOnCall[len(fake.hasPermissionArgsForCall)]
@@ -6156,8 +6219,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.getClientInfoMutex.RUnlock()
 	fake.getConnectionQualityMutex.RLock()
 	defer fake.getConnectionQualityMutex.RUnlock()
-	fake.getICEConnectionTypeMutex.RLock()
-	defer fake.getICEConnectionTypeMutex.RUnlock()
+	fake.getICEConnectionDetailsMutex.RLock()
+	defer fake.getICEConnectionDetailsMutex.RUnlock()
 	fake.getLoggerMutex.RLock()
 	defer fake.getLoggerMutex.RUnlock()
 	fake.getPacerMutex.RLock()
@@ -6186,6 +6249,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.handleReconnectAndSendResponseMutex.RUnlock()
 	fake.handleSignalSourceCloseMutex.RLock()
 	defer fake.handleSignalSourceCloseMutex.RUnlock()
+	fake.hasConnectedMutex.RLock()
+	defer fake.hasConnectedMutex.RUnlock()
 	fake.hasPermissionMutex.RLock()
 	defer fake.hasPermissionMutex.RUnlock()
 	fake.hiddenMutex.RLock()