From 526985f1090289e299cb0eddc537eb66da286db7 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Sat, 26 Oct 2024 22:29:04 +0800 Subject: [PATCH 1/7] don't return video/rtx to client (#3142) --- pkg/rtc/participant.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index fcdcf68b3..04ba5eef0 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -2807,7 +2807,14 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod } func (p *ParticipantImpl) GetEnabledPublishCodecs() []*livekit.Codec { - return p.enabledPublishCodecs + codecs := make([]*livekit.Codec, 0, len(p.enabledPublishCodecs)) + for _, c := range p.enabledPublishCodecs { + if c.Mime == "video/rtx" { + continue + } + codecs = append(codecs, c) + } + return codecs } func (p *ParticipantImpl) UpdateAudioTrack(update *livekit.UpdateLocalAudioTrack) error { From 24f3c93204b14b335d57b8dd42382296d20c7feb Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Tue, 29 Oct 2024 12:16:21 -0700 Subject: [PATCH 2/7] ignore unexported fields in yaml lint (#3145) --- pkg/config/configtest/checkyamltag.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pkg/config/configtest/checkyamltag.go b/pkg/config/configtest/checkyamltag.go index a8ee58371..2458a5b9e 100644 --- a/pkg/config/configtest/checkyamltag.go +++ b/pkg/config/configtest/checkyamltag.go @@ -31,6 +31,11 @@ func checkYAMLTags(t reflect.Type, seen map[reflect.Type]struct{}) error { for i := 0; i < t.NumField(); i++ { field := t.Field(i) + if !field.IsExported() { + // ignore unexported fields + continue + } + if field.Type.Kind() == reflect.Bool { // ignore boolean fields continue From da9bd7f4264c1f3cf4f8366722c9c76852b17a61 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 30 Oct 2024 19:44:41 +0530 Subject: [PATCH 3/7] make a util of IP address truncation for logging. (#3148) * make a util of IP address truncation for logging. * exported method --- pkg/rtc/room.go | 25 ++++--------------------- pkg/rtc/transport.go | 4 ++-- pkg/rtc/utils.go | 16 ++++++++++++++++ 3 files changed, 22 insertions(+), 23 deletions(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 7da83559f..306d7b04a 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -18,7 +18,6 @@ import ( "context" "fmt" "math" - "net" "slices" "sort" "strings" @@ -55,8 +54,6 @@ const ( dataForwardLoadBalanceThreshold = 20 simulateDisconnectSignalTimeout = 5 * time.Second - - minIPTruncateLen = 8 ) var ( @@ -1840,25 +1837,11 @@ func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} { if c.Trickle { cStr += "[trickle]" } - remoteAddress := c.Remote.Address() - ipAddr := net.ParseIP(remoteAddress) - isPrivate := false - if ipAddr != nil { - isPrivate = ipAddr.IsPrivate() - } - if !isPrivate && len(remoteAddress) > minIPTruncateLen { - remoteAddress = remoteAddress[:len(remoteAddress)-3] + "..." - } - cStr += " " + fmt.Sprintf("%s %s %s:%d", c.Remote.NetworkType(), c.Remote.Type(), remoteAddress, c.Remote.Port()) + cStr += " " + fmt.Sprintf("%s %s %s:%d", c.Remote.NetworkType(), c.Remote.Type(), MaybeTruncateIP(c.Remote.Address()), c.Remote.Port()) if relatedAddress := c.Remote.RelatedAddress(); relatedAddress != nil { - ipAddr = net.ParseIP(relatedAddress.Address) - if ipAddr != nil { - isPrivate = ipAddr.IsPrivate() - relatedAddressAddress := relatedAddress.Address - if !isPrivate && len(relatedAddressAddress) > minIPTruncateLen { - relatedAddressAddress = relatedAddressAddress[:len(relatedAddressAddress)-3] + "..." - } - cStr += " " + fmt.Sprintf(" related %s:%d", relatedAddressAddress, relatedAddress.Port) + relatedAddr := MaybeTruncateIP(relatedAddress.Address) + if relatedAddr != "" { + cStr += " " + fmt.Sprintf(" related %s:%d", relatedAddr, relatedAddress.Port) } } candidates = append(candidates, cStr) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index a3665dd46..4fc32d13a 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -150,10 +150,10 @@ func (w wrappedICECandidatePairLogger) MarshalLogObject(e zapcore.ObjectEncoder) if w.pair.Remote != nil { e.AddString("remoteProtocol", w.pair.Remote.Protocol.String()) e.AddString("remoteCandidateType", w.pair.Remote.Typ.String()) - e.AddString("remoteAdddress", w.pair.Remote.Address[:len(w.pair.Remote.Address)-3]+"...") + e.AddString("remoteAdddress", MaybeTruncateIP(w.pair.Remote.Address)) e.AddUint16("remotePort", w.pair.Remote.Port) if w.pair.Remote.RelatedAddress != "" { - e.AddString("relatedAdddress", w.pair.Remote.RelatedAddress[:len(w.pair.Remote.RelatedAddress)-3]+"...") + e.AddString("relatedAdddress", MaybeTruncateIP(w.pair.Remote.RelatedAddress)) e.AddUint16("relatedPort", w.pair.Remote.RelatedPort) } } diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index c5a54d6ac..21390b362 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -18,6 +18,7 @@ import ( "encoding/json" "errors" "io" + "net" "strings" "github.com/pion/webrtc/v3" @@ -28,6 +29,8 @@ import ( const ( trackIdSeparator = "|" + + cMinIPTruncateLen = 8 ) func UnpackStreamID(packed string) (participantID livekit.ParticipantID, trackID livekit.TrackID) { @@ -198,3 +201,16 @@ func LoggerWithCodecMime(l logger.Logger, mime string) logger.Logger { } return l } + +func MaybeTruncateIP(addr string) string { + ipAddr := net.ParseIP(addr) + if ipAddr == nil { + return "" + } + + if ipAddr.IsPrivate() || len(addr) <= cMinIPTruncateLen { + return addr + } + + return addr[:len(addr)-3] + "..." +} From 1c80ce830870281ce71b2228e77b7f363fe81244 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 30 Oct 2024 21:20:34 +0530 Subject: [PATCH 4/7] Only drop srflx if configured. (#3149) --- pkg/rtc/transport.go | 21 ++++++++++----- pkg/rtc/transport_test.go | 54 +++++++++++++++++++-------------------- 2 files changed, 42 insertions(+), 33 deletions(-) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 4fc32d13a..df7ad0d61 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -234,6 +234,8 @@ type PCTransport struct { connectionDetails *types.ICEConnectionDetails selectedPair atomic.Pointer[webrtc.ICECandidatePair] + + dropRemoteICECandidates bool } type TransportParams struct { @@ -1415,7 +1417,7 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { c := e.data.(*webrtc.ICECandidateInit) filtered := false - if t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp") { + if t.preferTCP.Load() && !strings.Contains(strings.ToLower(c.Candidate), "tcp") { t.params.Logger.Debugw("filtering out remote candidate", "candidate", c.Candidate) filtered = true } @@ -1435,7 +1437,7 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { return nil } - if t.params.DropRemoteICECandidates { + if t.dropRemoteICECandidates && strings.Contains(strings.ToLower(c.Candidate), "srflx") { t.params.Logger.Debugw("dropping remote ICE candidate", "candidate", c.Candidate) t.connectionDetails.AddRemoteCandidate(*c, true, true, true) return nil @@ -1480,8 +1482,14 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, } if liteSet && t.localICEIsLite != nil && t.remoteICEIsLite != nil { // only drop remote candidates if local is lite and remote is not - t.params.DropRemoteICECandidates = t.params.DropRemoteICECandidates && (*t.localICEIsLite && !*t.remoteICEIsLite) - t.params.Logger.Debugw("setting DropRemoteICECandidates", "dropRemoteCandidate", t.params.DropRemoteICECandidates, "localICELite", *t.localICEIsLite, "remoteICELite", *t.remoteICEIsLite) + t.dropRemoteICECandidates = t.params.DropRemoteICECandidates && (*t.localICEIsLite && !*t.remoteICEIsLite) + t.params.Logger.Debugw( + "setting DropRemoteICECandidates", + "dropRemoteICECandidatesConfig", t.params.DropRemoteICECandidates, + "dropRemoteICECandidatesCalculated", t.dropRemoteICECandidates, + "localICELite", *t.localICEIsLite, + "remoteICELite", *t.remoteICEIsLite, + ) } filterAttributes := func(attrs []sdp.Attribute) []sdp.Attribute { @@ -1494,7 +1502,7 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, filteredAttrs = append(filteredAttrs, a) continue } - excluded := (!isLocal && t.params.DropRemoteICECandidates) || (preferTCP && !c.NetworkType().IsTCP()) + excluded := (!isLocal && t.dropRemoteICECandidates && c.Type() == ice.CandidateTypeServerReflexive) || (preferTCP && !c.NetworkType().IsTCP()) if !excluded { if !t.params.Config.UseMDNS && types.IsICECandidateMDNS(c) { excluded = true @@ -1712,7 +1720,8 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error { } for _, c := range t.pendingRemoteCandidates { - if t.params.DropRemoteICECandidates { + if t.dropRemoteICECandidates && strings.Contains(strings.ToLower(c.Candidate), "srflx") { + t.params.Logger.Debugw("dropping remote ICE candidate (pending)", "candidate", c.Candidate) t.connectionDetails.AddRemoteCandidate(*c, true, true, true) continue } diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index e475d32a3..4cda22e4f 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -506,39 +506,39 @@ func TestFilteringCandidates(t *testing.T) { } func TestDropRemoteICECandidates(t *testing.T) { cases := []struct { - name string - remoteLite bool - localLite bool - expecteLocalDrop bool - expecteRemoteDrop bool + name string + remoteLite bool + localLite bool + expectedLocalDrop bool + expectedRemoteDrop bool }{ { - name: "both not lite", - localLite: false, - remoteLite: false, - expecteLocalDrop: false, - expecteRemoteDrop: false, + name: "both not lite", + localLite: false, + remoteLite: false, + expectedLocalDrop: false, + expectedRemoteDrop: false, }, { - name: "remote lite", - localLite: false, - remoteLite: true, - expecteLocalDrop: false, - expecteRemoteDrop: true, + name: "remote lite", + localLite: false, + remoteLite: true, + expectedLocalDrop: false, + expectedRemoteDrop: true, }, { - name: "local lite", - localLite: true, - remoteLite: false, - expecteLocalDrop: true, - expecteRemoteDrop: false, + name: "local lite", + localLite: true, + remoteLite: false, + expectedLocalDrop: true, + expectedRemoteDrop: false, }, { - name: "both lite", - localLite: true, - remoteLite: true, - expecteLocalDrop: false, - expecteRemoteDrop: false, + name: "both lite", + localLite: true, + remoteLite: true, + expectedLocalDrop: false, + expectedRemoteDrop: false, }, } @@ -605,8 +605,8 @@ func TestDropRemoteICECandidates(t *testing.T) { return transportRemote.IsEstablished() }, 10*time.Second, time.Millisecond*10, "transportB is not established") - require.Equal(t, c.expecteLocalDrop, transportLocal.params.DropRemoteICECandidates) - require.Equal(t, c.expecteRemoteDrop, transportRemote.params.DropRemoteICECandidates) + require.Equal(t, c.expectedLocalDrop, transportLocal.dropRemoteICECandidates) + require.Equal(t, c.expectedRemoteDrop, transportRemote.dropRemoteICECandidates) transportLocal.Close() transportRemote.Close() From 150433ab6477126abdaad4582b1d340af670452e Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 31 Oct 2024 12:24:04 +0530 Subject: [PATCH 5/7] Update ICE to pick up accepting use-candidate unconditionally for ICE lite agents (#3150) --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index c3fce73ae..591c453a6 100644 --- a/go.mod +++ b/go.mod @@ -28,7 +28,7 @@ require ( github.com/olekukonko/tablewriter v0.0.5 github.com/ory/dockertest/v3 v3.11.0 github.com/pion/dtls/v2 v2.2.12 - github.com/pion/ice/v2 v2.3.36 + github.com/pion/ice/v2 v2.3.37 github.com/pion/interceptor v0.1.37 github.com/pion/rtcp v1.2.14 github.com/pion/rtp v1.8.9 diff --git a/go.sum b/go.sum index 6d0539605..b1ea92974 100644 --- a/go.sum +++ b/go.sum @@ -233,8 +233,8 @@ github.com/pion/datachannel v1.5.9/go.mod h1:kDUuk4CU4Uxp82NH4LQZbISULkX/HtzKa4P github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s= github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk= github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE= -github.com/pion/ice/v2 v2.3.36 h1:SopeXiVbbcooUg2EIR8sq4b13RQ8gzrkkldOVg+bBsc= -github.com/pion/ice/v2 v2.3.36/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ= +github.com/pion/ice/v2 v2.3.37 h1:ObIdaNDu1rCo7hObhs34YSBcO7fjslJMZV0ux+uZWh0= +github.com/pion/ice/v2 v2.3.37/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ= github.com/pion/interceptor v0.1.37 h1:aRA8Zpab/wE7/c0O3fh1PqY0AJI3fCSEM5lRWJVorwI= github.com/pion/interceptor v0.1.37/go.mod h1:JzxbJ4umVTlZAf+/utHzNesY8tmRkM2lVmkS82TTj8Y= github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY= From 35bef35d66bf9b03894d3debc363143a5c2cb60c Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 2 Nov 2024 10:50:55 +0530 Subject: [PATCH 6/7] Clean up drop ICE candidates. (#3153) * Clean up drop ICE candidates. With pion/ice v2.3.37, ICE Lite will accept use-candidate from peer. So, there is no need to drop candidates. Still leaving the FF change to not use Lite which was added as part of this effort initially due to how FF does nominations. Updated comment to explain why. * clean up test --- pkg/rtc/participant.go | 2 - pkg/rtc/transport.go | 50 +++------------- pkg/rtc/transport_test.go | 110 ------------------------------------ pkg/rtc/transportmanager.go | 3 - 4 files changed, 7 insertions(+), 158 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 04ba5eef0..f7e9937a1 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -158,7 +158,6 @@ type ParticipantParams struct { ForwardStats *sfu.ForwardStats DisableSenderReportPassThrough bool MetricConfig metric.MetricConfig - DropRemoteICECandidates bool } type ParticipantImpl struct { @@ -1449,7 +1448,6 @@ func (p *ParticipantImpl) setupTransportManager() error { PublisherHandler: pth, SubscriberHandler: sth, DataChannelStats: p.dataChannelStats, - DropRemoteICECandidates: p.params.DropRemoteICECandidates, } if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() { // we will disable playout delay for Firefox if the user is expecting diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index df7ad0d61..fba7f6c51 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -234,8 +234,6 @@ type PCTransport struct { connectionDetails *types.ICEConnectionDetails selectedPair atomic.Pointer[webrtc.ICECandidatePair] - - dropRemoteICECandidates bool } type TransportParams struct { @@ -256,7 +254,6 @@ type TransportParams struct { IsSendSide bool AllowPlayoutDelay bool DataChannelMaxBufferedAmount uint64 - DropRemoteICECandidates bool } func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimator cc.BandwidthEstimator)) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) { @@ -301,9 +298,12 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat se.DisableSRTCPReplayProtection(true) if !params.ProtocolVersion.SupportsICELite() || !params.ClientInfo.SupportPrflxOverRelay() { // if client don't support prflx over relay which is only Firefox, disable ICE Lite to ensure that - // dropping remote ICE candidates does not get enabled. Firefox does aggressive nomination and - // dropping remote ICE candidates means server would accept all switches and it could end up with - // the lower priority candidate. As Firefox does not support migration, ICE Lite can be disabled. + // aggressive nomination is handled properly. Firefox does aggressive nomination even if peer is + // ICE Lite (see comment as to historical reasons: https://github.com/pion/ice/pull/739#issuecomment-2452245066). + // pion/ice (as of v2.3.37) will accept all use-candidate switches when in ICE Lite mode. + // That combined with aggressive nomination from Firefox could potentially lead to the two ends + // ending up with different candidates. + // As Firefox does not support migration, ICE Lite can be disabled. se.SetLite(false) } se.SetDTLSRetransmissionInterval(dtlsRetransmissionInterval) @@ -1437,12 +1437,6 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { return nil } - if t.dropRemoteICECandidates && strings.Contains(strings.ToLower(c.Candidate), "srflx") { - t.params.Logger.Debugw("dropping remote ICE candidate", "candidate", c.Candidate) - t.connectionDetails.AddRemoteCandidate(*c, true, true, true) - return nil - } - if err := t.pc.AddICECandidate(*c); err != nil { t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c) return errors.Wrap(err, "add ice candidate failed") @@ -1467,31 +1461,6 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, return sd } - _, iceLite := parsed.Attribute("ice-lite") - var liteSet bool - if isLocal { - if t.localICEIsLite == nil { - t.localICEIsLite = &iceLite - liteSet = true - } - } else { - if t.remoteICEIsLite == nil { - t.remoteICEIsLite = &iceLite - liteSet = true - } - } - if liteSet && t.localICEIsLite != nil && t.remoteICEIsLite != nil { - // only drop remote candidates if local is lite and remote is not - t.dropRemoteICECandidates = t.params.DropRemoteICECandidates && (*t.localICEIsLite && !*t.remoteICEIsLite) - t.params.Logger.Debugw( - "setting DropRemoteICECandidates", - "dropRemoteICECandidatesConfig", t.params.DropRemoteICECandidates, - "dropRemoteICECandidatesCalculated", t.dropRemoteICECandidates, - "localICELite", *t.localICEIsLite, - "remoteICELite", *t.remoteICEIsLite, - ) - } - filterAttributes := func(attrs []sdp.Attribute) []sdp.Attribute { filteredAttrs := make([]sdp.Attribute, 0, len(attrs)) for _, a := range attrs { @@ -1502,7 +1471,7 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, filteredAttrs = append(filteredAttrs, a) continue } - excluded := (!isLocal && t.dropRemoteICECandidates && c.Type() == ice.CandidateTypeServerReflexive) || (preferTCP && !c.NetworkType().IsTCP()) + excluded := preferTCP && !c.NetworkType().IsTCP() if !excluded { if !t.params.Config.UseMDNS && types.IsICECandidateMDNS(c) { excluded = true @@ -1720,11 +1689,6 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error { } for _, c := range t.pendingRemoteCandidates { - if t.dropRemoteICECandidates && strings.Contains(strings.ToLower(c.Candidate), "srflx") { - t.params.Logger.Debugw("dropping remote ICE candidate (pending)", "candidate", c.Candidate) - t.connectionDetails.AddRemoteCandidate(*c, true, true, true) - continue - } if err := t.pc.AddICECandidate(*c); err != nil { t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c) return errors.Wrap(err, "add ice candidate failed") diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index 4cda22e4f..c80b206ef 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -28,7 +28,6 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/transport" "github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes" - "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/testutils" "github.com/livekit/protocol/livekit" ) @@ -504,115 +503,6 @@ func TestFilteringCandidates(t *testing.T) { transport.Close() } -func TestDropRemoteICECandidates(t *testing.T) { - cases := []struct { - name string - remoteLite bool - localLite bool - expectedLocalDrop bool - expectedRemoteDrop bool - }{ - { - name: "both not lite", - localLite: false, - remoteLite: false, - expectedLocalDrop: false, - expectedRemoteDrop: false, - }, - { - name: "remote lite", - localLite: false, - remoteLite: true, - expectedLocalDrop: false, - expectedRemoteDrop: true, - }, - { - name: "local lite", - localLite: true, - remoteLite: false, - expectedLocalDrop: true, - expectedRemoteDrop: false, - }, - { - name: "both lite", - localLite: true, - remoteLite: true, - expectedLocalDrop: false, - expectedRemoteDrop: false, - }, - } - - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - params := TransportParams{ - ParticipantID: "id", - ParticipantIdentity: "identity", - Config: &WebRTCConfig{}, - IsOfferer: true, - ProtocolVersion: types.CurrentProtocol, - DropRemoteICECandidates: true, - } - - paramsA := params - paramsA.Config.SettingEngine.SetLite(c.localLite) - handlerA := &transportfakes.FakeHandler{} - paramsA.Handler = handlerA - transportLocal, err := NewPCTransport(paramsA) - require.NoError(t, err) - _, err = transportLocal.pc.CreateDataChannel(LossyDataChannel, nil) - require.NoError(t, err) - - paramsB := params - paramsB.Config.SettingEngine.SetLite(c.remoteLite) - handlerB := &transportfakes.FakeHandler{} - paramsB.Handler = handlerB - paramsB.IsOfferer = false - transportRemote, err := NewPCTransport(paramsB) - require.NoError(t, err) - - require.False(t, transportLocal.IsEstablished()) - require.False(t, transportRemote.IsEstablished()) - - handleICEExchange(t, transportLocal, transportRemote, handlerA, handlerB) - var offer atomic.Pointer[webrtc.SessionDescription] - handlerA.OnOfferCalls(func(sd webrtc.SessionDescription) error { - parsed, err := sd.Unmarshal() - require.NoError(t, err) - _, lite := parsed.Attribute("ice-lite") - require.Equal(t, c.localLite, lite) - offer.Store(&sd) - return nil - }) - transportLocal.Negotiate(true) - require.Eventually(t, func() bool { - return offer.Load() != nil - }, 100*time.Millisecond, time.Millisecond*10, "offer not received") - - handlerB.OnAnswerCalls(func(sd webrtc.SessionDescription) error { - parsed, err := sd.Unmarshal() - require.NoError(t, err) - _, lite := parsed.Attribute("ice-lite") - require.Equal(t, c.remoteLite, lite, sd.SDP) - transportLocal.HandleRemoteDescription(sd) - return nil - }) - transportRemote.HandleRemoteDescription(*offer.Load()) - - require.Eventually(t, func() bool { - return transportLocal.IsEstablished() - }, 10*time.Second, time.Millisecond*10, "transportA is not established") - require.Eventually(t, func() bool { - return transportRemote.IsEstablished() - }, 10*time.Second, time.Millisecond*10, "transportB is not established") - - require.Equal(t, c.expectedLocalDrop, transportLocal.dropRemoteICECandidates) - require.Equal(t, c.expectedRemoteDrop, transportRemote.dropRemoteICECandidates) - - transportLocal.Close() - transportRemote.Close() - }) - } -} func handleICEExchange(t *testing.T, a, b *PCTransport, ah, bh *transportfakes.FakeHandler) { ah.OnICECandidateCalls(func(candidate *webrtc.ICECandidate, target livekit.SignalTarget) error { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index bee94e700..cc69e514c 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -103,7 +103,6 @@ type TransportManagerParams struct { PublisherHandler transport.Handler SubscriberHandler transport.Handler DataChannelStats *telemetry.BytesTrackStats - DropRemoteICECandidates bool } type TransportManager struct { @@ -158,7 +157,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro ClientInfo: params.ClientInfo, Transport: livekit.SignalTarget_PUBLISHER, Handler: TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t, lgr}}, - DropRemoteICECandidates: params.DropRemoteICECandidates, }) if err != nil { return nil, err @@ -182,7 +180,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, Transport: livekit.SignalTarget_SUBSCRIBER, Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, - DropRemoteICECandidates: params.DropRemoteICECandidates, }) if err != nil { return nil, err From 365e63230d88f264ebab3dc0e747bb6dcdd8811f Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 4 Nov 2024 11:26:41 +0530 Subject: [PATCH 7/7] Some misc clean up. (#3156) * Some misc clean up. - Have been seeing counterfeiter warnings about efficiency for a while with go:generate declaration multiple times in the same package. Address that: https://github.com/maxbrunsfeld/counterfeiter?tab=readme-ov-file#step-2b---add-counterfeitergenerate-directives - A bit more readability on parameters passed to `sendLeave` * spacing * revert some deletes as the complaint was in analytics service only * Declare in package only once. Although the warning is about go:generate multiple times when directly giving the interface to generate, have `go:generate` multiple times in a package even with `-generate` ends up generating once per invocation. Once per package is enough to run the generation just once. --- pkg/routing/roommanager.go | 2 -- pkg/routing/signal.go | 2 -- pkg/rtc/participant.go | 31 ++++++++++++++++++++++++++----- pkg/service/clients.go | 2 -- pkg/service/signal.go | 2 -- pkg/telemetry/analyticsservice.go | 2 +- pkg/telemetry/telemetryservice.go | 4 +++- 7 files changed, 30 insertions(+), 15 deletions(-) diff --git a/pkg/routing/roommanager.go b/pkg/routing/roommanager.go index 327c5806d..796ffe764 100644 --- a/pkg/routing/roommanager.go +++ b/pkg/routing/roommanager.go @@ -24,8 +24,6 @@ import ( "github.com/livekit/psrpc/pkg/middleware" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate - //counterfeiter:generate . RoomManagerClient type RoomManagerClient interface { rpc.TypedRoomManagerClient diff --git a/pkg/routing/signal.go b/pkg/routing/signal.go index 858fab7c1..ce7c5f42d 100644 --- a/pkg/routing/signal.go +++ b/pkg/routing/signal.go @@ -36,8 +36,6 @@ import ( var ErrSignalWriteFailed = errors.New("signal write failed") var ErrSignalMessageDropped = errors.New("signal message dropped") -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate - //counterfeiter:generate . SignalClient type SignalClient interface { ActiveCount() int diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index f7e9937a1..4a5a734d9 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -963,7 +963,12 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea p.clearMigrationTimer() if sendLeave { - p.sendLeaveRequest(reason, isExpectedToResume, false, false) + p.sendLeaveRequest( + reason, + isExpectedToResume, + false, // isExpectedToReconnect + false, // sendOnlyIfSupportingLeaveRequestWithAction + ) } if p.supervisor != nil { @@ -1070,7 +1075,12 @@ func (p *ParticipantImpl) MaybeStartMigration(force bool, onStart func()) bool { onStart() } - p.sendLeaveRequest(types.ParticipantCloseReasonMigrationRequested, true, false, true) + p.sendLeaveRequest( + types.ParticipantCloseReasonMigrationRequested, + true, // isExpectedToResume + false, // isExpectedToReconnect + true, // sendOnlyIfSupportingLeaveRequestWithAction + ) p.CloseSignalConnection(types.SignallingCloseReasonMigration) p.clearMigrationTimer() @@ -1818,8 +1828,14 @@ func (p *ParticipantImpl) setupDisconnectTimer() { } func (p *ParticipantImpl) onAnyTransportFailed() { - // clients support resuming of connections when websocket becomes disconnected - p.sendLeaveRequest(types.ParticipantCloseReasonPeerConnectionDisconnected, true, false, true) + p.sendLeaveRequest( + types.ParticipantCloseReasonPeerConnectionDisconnected, + true, // isExpectedToResume + false, // isExpectedToReconnect + true, // sendOnlyIfSupportingLeaveRequestWithAction + ) + + // clients support resuming of connections when signalling becomes disconnected p.CloseSignalConnection(types.SignallingCloseReasonTransportFailure) // detect when participant has actually left. @@ -2656,7 +2672,12 @@ func (p *ParticipantImpl) GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.R } func (p *ParticipantImpl) IssueFullReconnect(reason types.ParticipantCloseReason) { - p.sendLeaveRequest(reason, false, true, false) + p.sendLeaveRequest( + reason, + false, // isExpectedToResume + true, // isExpectedToReconnect + false, // sendOnlyIfSupportingLeaveRequestWithAction + ) scr := types.SignallingCloseReasonUnknown switch reason { diff --git a/pkg/service/clients.go b/pkg/service/clients.go index 9fd887878..8a0eb39c1 100644 --- a/pkg/service/clients.go +++ b/pkg/service/clients.go @@ -27,8 +27,6 @@ import ( "github.com/livekit/protocol/utils/guid" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate - //counterfeiter:generate . IOClient type IOClient interface { CreateEgress(ctx context.Context, info *livekit.EgressInfo) (*emptypb.Empty, error) diff --git a/pkg/service/signal.go b/pkg/service/signal.go index 742ce51d1..b636aa018 100644 --- a/pkg/service/signal.go +++ b/pkg/service/signal.go @@ -31,8 +31,6 @@ import ( "github.com/livekit/psrpc/pkg/middleware" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate - //counterfeiter:generate . SessionHandler type SessionHandler interface { Logger(ctx context.Context) logger.Logger diff --git a/pkg/telemetry/analyticsservice.go b/pkg/telemetry/analyticsservice.go index 365e7ed05..57873acb0 100644 --- a/pkg/telemetry/analyticsservice.go +++ b/pkg/telemetry/analyticsservice.go @@ -29,7 +29,7 @@ import ( "github.com/livekit/livekit-server/pkg/routing" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . AnalyticsService +//counterfeiter:generate . AnalyticsService type AnalyticsService interface { SendStats(ctx context.Context, stats []*livekit.AnalyticsStat) SendEvent(ctx context.Context, events *livekit.AnalyticsEvent) diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 0079c5d5a..ba8ca50df 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -26,7 +26,9 @@ import ( "github.com/livekit/protocol/webhook" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . TelemetryService +//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate + +//counterfeiter:generate . TelemetryService type TelemetryService interface { // TrackStats is called periodically for each track in both directions (published/subscribed) TrackStats(key StatsKey, stat *livekit.AnalyticsStat)