dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-15 12:29:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add option to exclude local IPv6 candidates. (#4657)

Browse Source 
Could be useful option to try in certain conditions where flakey IPv6
infrastructure is suspected for connection issues.
This commit is contained in:
Raja Subramanian
2026-07-10 15:56:58 +05:30
committed by GitHub
parent aff752a4bc
commit 19c3d00fc9
5 changed files with 75 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+2
View File
@@ -228,6 +228,7 @@ type ParticipantParams struct {
EnableParticipantDataBlob bool
EnableStartAtDesiredQuality bool
MigrationWaitDuration time.Duration
ExcludeIPv6LocalCandidates bool
}
type ParticipantImpl struct {
@@ -2057,6 +2058,7 @@ func (p *ParticipantImpl) setupTransportManager() error {
UseOneShotSignallingMode: p.params.UseOneShotSignallingMode,
FireOnTrackBySdp: p.params.FireOnTrackBySdp,
EnableDataTracks: p.params.EnableDataTracks,
ExcludeIPv6LocalCandidates: p.params.ExcludeIPv6LocalCandidates,
}
if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() {
// we will disable playout delay for Firefox if the user is expecting
pkg/rtc/transport.go
+56 -53
View File
@@ -319,6 +319,7 @@ type TransportParams struct {
IsSendSide bool
AllowPlayoutDelay bool
UseOneShotSignallingMode bool
ExcludeIPv6LocalCandidates bool
FireOnTrackBySdp bool
DataChannelMaxBufferedAmount uint64
DatachannelSlowThreshold int
@@ -1712,31 +1713,26 @@ func (t *PCTransport) GetAnswer() (webrtc.SessionDescription, uint32, error) {
cld := t.pc.CurrentLocalDescription()
// add local candidates to ICE connection details
parsed, err := cld.Unmarshal()
if err == nil {
addLocalICECandidates := func(attrs []sdp.Attribute) {
for _, a := range attrs {
if a.IsICECandidate() {
c, err := ice.UnmarshalCandidate(a.Value)
if err != nil {
continue
}
t.connectionDetails.AddLocalICECandidate(c, false, false)
}
}
}
preferTCP := t.preferTCP.Load()
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("local answer (unfiltered)", "sdp", cld.SDP)
}
addLocalICECandidates(parsed.Attributes)
for _, m := range parsed.MediaDescriptions {
addLocalICECandidates(m.Attributes)
}
//
// Filter after setting local description as pion expects the answer
// to match between CreateAnswer and SetLocalDescription.
// Filtered answer is sent to remote so that remote does not
// see filtered candidates.
//
filteredAnswer := t.filterCandidates(*cld, preferTCP, true)
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("local answer (filtered)", "sdp", filteredAnswer.SDP)
}
answerId := t.remoteOfferId.Load()
t.localAnswerId.Store(answerId)
return *cld, answerId, nil
return filteredAnswer, answerId, nil
}
func (t *PCTransport) GetICESessionUfrag() (string, error) {
@@ -1899,13 +1895,13 @@ func (t *PCTransport) HandleICERestartSDPFragment(sdpFragment string) (string, e
t.connectionDetails.AddRemoteICECandidate(c, false, false, false)
}
ans, err := t.pc.CreateAnswer(nil)
answer, err := t.pc.CreateAnswer(nil)
if err != nil {
t.params.Logger.Warnw("could not create answer", err)
return "", err
}
if err = t.pc.SetLocalDescription(ans); err != nil {
if err = t.pc.SetLocalDescription(answer); err != nil {
t.params.Logger.Warnw("could not set local description", err)
return "", err
}
@@ -1915,31 +1911,30 @@ func (t *PCTransport) HandleICERestartSDPFragment(sdpFragment string) (string, e
cld := t.pc.CurrentLocalDescription()
preferTCP := t.preferTCP.Load()
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("local answer (unfiltered)", "sdp", cld.SDP)
}
//
// Filter after setting local description as pion expects the answer
// to match between CreateAnswer and SetLocalDescription.
// Filtered answer is sent to remote so that remote does not
// see filtered candidates.
//
filteredAnswer := t.filterCandidates(*cld, preferTCP, true)
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("local answer (filtered)", "sdp", filteredAnswer.SDP)
}
// add local candidates to ICE connection details
parsedAnswer, err := cld.Unmarshal()
parsedFilteredAnswer, err := filteredAnswer.Unmarshal()
if err != nil {
t.params.Logger.Warnw("could not parse local description", err)
return "", err
}
addLocalICECandidates := func(attrs []sdp.Attribute) {
for _, a := range attrs {
if a.IsICECandidate() {
c, err := ice.UnmarshalCandidate(a.Value)
if err != nil {
continue
}
t.connectionDetails.AddLocalICECandidate(c, false, false)
}
}
}
addLocalICECandidates(parsedAnswer.Attributes)
for _, m := range parsedAnswer.MediaDescriptions {
addLocalICECandidates(m.Attributes)
}
parsedFragmentAnswer, err := lksdp.ExtractSDPFragment(parsedAnswer)
parsedFragmentAnswer, err := lksdp.ExtractSDPFragment(parsedFilteredAnswer)
if err != nil {
t.params.Logger.Warnw("could not extract SDP fragment", err)
return "", err
@@ -2402,9 +2397,15 @@ func (t *PCTransport) handleLocalICECandidate(e event) error {
filtered := false
if c != nil {
if t.preferTCP.Load() && c.Protocol != webrtc.ICEProtocolTCP {
t.params.Logger.Debugw("filtering out local candidate", "candidate", c.String())
t.params.Logger.Debugw("filtering out local candidate, TCP prefered", "candidate", c.String())
filtered = true
}
if !filtered && t.params.ExcludeIPv6LocalCandidates {
if IsIPv6(c.Address) {
t.params.Logger.Debugw("filtering out local candidate, IPv6 excluded", "candidate", c.String())
filtered = true
}
}
t.connectionDetails.AddLocalCandidate(c, filtered, true)
}
@@ -2469,6 +2470,10 @@ func (t *PCTransport) setNegotiationState(state transport.NegotiationState) {
}
}
func (t *PCTransport) isCandidateFilterActive(preferTCP bool) bool {
return preferTCP || t.params.ExcludeIPv6LocalCandidates
}
func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, isLocal bool) webrtc.SessionDescription {
parsed, err := sd.Unmarshal()
if err != nil {
@@ -2486,12 +2491,10 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP,
filteredAttrs = append(filteredAttrs, a)
continue
}
excluded := preferTCP && !c.NetworkType().IsTCP()
if !excluded {
if !t.params.Config.UseMDNS && types.IsICECandidateMDNS(c) {
excluded = true
}
}
excluded :=
(preferTCP && !c.NetworkType().IsTCP()) ||
(t.params.ExcludeIPv6LocalCandidates && isLocal && c.NetworkType().IsIPv6()) ||
(!t.params.Config.UseMDNS && types.IsICECandidateMDNS(c))
if !excluded {
filteredAttrs = append(filteredAttrs, a)
}
@@ -2634,7 +2637,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error {
}
preferTCP := t.preferTCP.Load()
if preferTCP {
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("local offer (unfiltered)", "sdp", offer.SDP)
}
@@ -2662,7 +2665,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error {
// see filtered candidates.
//
offer = t.filterCandidates(offer, preferTCP, true)
if preferTCP {
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("local offer (filtered)", "sdp", offer.SDP)
}
@@ -2728,11 +2731,11 @@ func (t *PCTransport) isRemoteOfferRestartICE(parsed *sdp.SessionDescription) (s
func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error {
// filter before setting remote description so that pion does not see filtered remote candidates
preferTCP := t.preferTCP.Load()
if preferTCP {
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("remote description (unfiltered)", "type", sd.Type, "sdp", sd.SDP)
}
sd = t.filterCandidates(sd, preferTCP, false)
if preferTCP {
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("remote description (filtered)", "type", sd.Type, "sdp", sd.SDP)
}
@@ -2792,7 +2795,7 @@ func (t *PCTransport) createAndSendAnswer() error {
}
preferTCP := t.preferTCP.Load()
if preferTCP {
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("local answer (unfiltered)", "sdp", answer.SDP)
}
@@ -2808,7 +2811,7 @@ func (t *PCTransport) createAndSendAnswer() error {
// see filtered candidates.
//
answer = t.filterCandidates(answer, preferTCP, true)
if preferTCP {
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("local answer (filtered)", "sdp", answer.SDP)
}
pkg/rtc/transportmanager.go
+3
View File
@@ -98,6 +98,7 @@ type TransportManagerParams struct {
UseOneShotSignallingMode bool
FireOnTrackBySdp bool
EnableDataTracks bool
ExcludeIPv6LocalCandidates bool
}
type TransportManager struct {
@@ -168,6 +169,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
DatachannelLossyTargetLatency: params.DatachannelLossyTargetLatency,
FireOnTrackBySdp: params.FireOnTrackBySdp,
EnableDataTracks: params.EnableDataTracks,
ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates,
})
if err != nil {
return nil, err
@@ -194,6 +196,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr},
FireOnTrackBySdp: params.FireOnTrackBySdp,
EnableDataTracks: params.EnableDataTracks,
ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates,
})
if err != nil {
return nil, err
pkg/rtc/utils.go
+5
View File
@@ -173,6 +173,11 @@ func MaybeTruncateIP(addr string) string {
return addr[:len(addr)-3] + "..."
}
func IsIPv6(addr string) bool {
ipAddr := net.ParseIP(addr)
return ipAddr != nil && ipAddr.To4() == nil
}
func ChunkProtoBatch[T proto.Message](batch []T, target int) [][]T {
var chunks [][]T
var start, size int
pkg/service/wire_gen.go
+9 -9
View File
@@ -86,23 +86,23 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
}
rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService, objectStore)
topicFormatter := rpc.NewTopicFormatter()
v, err := rpc.NewTypedRoomClient(clientParams)
roomClient, err := rpc.NewTypedRoomClient(clientParams)
if err != nil {
return nil, err
}
v2, err := rpc.NewTypedParticipantClient(clientParams)
participantClient, err := rpc.NewTypedParticipantClient(clientParams)
if err != nil {
return nil, err
}
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2)
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient)
if err != nil {
return nil, err
}
v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
if err != nil {
return nil, err
}
agentDispatchService := NewAgentDispatchService(limitConfig, v3, topicFormatter, roomAllocator, router)
agentDispatchService := NewAgentDispatchService(limitConfig, agentDispatchInternalClient, topicFormatter, roomAllocator, router)
egressService := NewEgressService(egressClient, rtcEgressLauncher, ioInfoService, roomService)
ingressConfig := getIngressConfig(conf)
ingressClient, err := rpc.NewIngressClient(clientParams)
@@ -117,11 +117,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
}
sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService)
rtcService := NewRTCService(conf, roomAllocator, router, telemetryService)
v4, err := rpc.NewTypedWHIPParticipantClient(clientParams)
whipParticipantClient, err := rpc.NewTypedWHIPParticipantClient(clientParams)
if err != nil {
return nil, err
}
serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, v4)
serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, whipParticipantClient)
if err != nil {
return nil, err
}
@@ -146,8 +146,8 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
if err != nil {
return nil, err
}
v5 := getTURNAuthHandlerFunc(turnAuthHandler)
server, err := newInProcessTurnServer(conf, v5)
authHandler := getTURNAuthHandlerFunc(turnAuthHandler)
server, err := newInProcessTurnServer(conf, authHandler)
if err != nil {
return nil, err
}